summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-16 11:17:46 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-16 11:17:46 +0200
commitcde57423e04812fe2c939fdf983e3b45a77fd82d (patch)
tree7fe44bc9d83cf72382394ebf9b5177c207618be2 /packages/meshbay-node
parent20706fb9a4ec646816b44a10842aa8f58ea0fd75 (diff)
downloadmeshbay-cde57423e04812fe2c939fdf983e3b45a77fd82d.tar.gz
mnp 3.1: per-account blobs the node cannot read
One row per playlist plus a manifest, so starring a track rewrites that playlist rather than the whole collection. blob_enc is a BLOB, not base64 TEXT: these run to hundreds of kilobytes. user_id comes from the session and never from the message; kind is validated against a pattern; every cap refuses with a stated reason rather than truncating. Additive, so MNP_MIN_SUPPORTED does not move — a 3.0 node answers "unknown message type" and the client writes to the next one it reaches. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node')
-rw-r--r--packages/meshbay-node/src/meshbay_node/bundle_store.py115
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py142
-rw-r--r--packages/meshbay-node/tests/test_user_blob_mnp.py256
-rw-r--r--packages/meshbay-node/tests/test_user_blob_store.py165
4 files changed, 678 insertions, 0 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py
index 86b8de4..c65b94d 100644
--- a/packages/meshbay-node/src/meshbay_node/bundle_store.py
+++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py
@@ -63,6 +63,38 @@ CREATE TABLE IF NOT EXISTS keypair_bundles (
);
"""
+# Per-account blobs the node holds and cannot read — playlists today
+# (docs/playlists.md §3.3). The same shape as a keypair bundle, with a
+# different payload, so this adds no trust boundary the node was not already
+# on the wrong side of for this same account.
+#
+# Two deliberate differences from keypair_bundles, each a mistake avoided
+# rather than a preference:
+#
+# `blob_enc` is a BLOB, not base64 TEXT. A keypair bundle is a few hundred
+# bytes and nobody will ever notice the 33% tax; a playlist is hundreds of
+# kilobytes, where a third is not a rounding error.
+#
+# `kind` is a namespace, not an enum: "playlists" is the manifest and
+# "playlist:<uuid>" is one playlist's tracks. That is what lets one playlist
+# be rewritten without re-uploading the whole collection, and it costs no
+# second table. webrtc_server.py validates the shape.
+#
+# blob_enc_recovery is declared now and left NULL — the same column, on the
+# same kind of table, that keypair_bundles needed a PRAGMA migration for one
+# release late. CREATE TABLE IF NOT EXISTS never adds a column.
+_SCHEMA_USER_BLOBS = """\
+CREATE TABLE IF NOT EXISTS user_blobs (
+ user_id TEXT NOT NULL,
+ kind TEXT NOT NULL,
+ rev INTEGER NOT NULL,
+ blob_enc BLOB NOT NULL,
+ blob_enc_recovery BLOB,
+ stored_at TEXT NOT NULL DEFAULT (datetime('now')),
+ PRIMARY KEY (user_id, kind)
+);
+"""
+
class BundleStore:
def __init__(self, db_path: Path):
@@ -75,6 +107,7 @@ class BundleStore:
await self._db.execute(_SCHEMA_GEK)
await self._db.execute(_SCHEMA_KEYPAIR)
await self._db.execute(_SCHEMA_CHAT_EPOCHS)
+ await self._db.execute(_SCHEMA_USER_BLOBS)
await self._migrate_keypair_recovery()
await self._db.commit()
@@ -180,6 +213,88 @@ class BundleStore:
await self._db.commit()
return cur.rowcount > 0
+ # ── Per-account blobs (playlists) ────────────────────────────────────
+ #
+ # Opaque throughout: the node stores bytes it cannot read, returns them, and
+ # never looks inside. Every method takes `user_id` from the caller, which
+ # takes it from the authenticated session and never from a message body —
+ # a user_id off the wire would let any member read or overwrite any other
+ # member's blob.
+
+ async def store_user_blob(
+ self,
+ user_id: str,
+ kind: str,
+ rev: int,
+ blob_enc: bytes,
+ ) -> None:
+ """
+ Write one blob, replacing whatever was there.
+
+ Older revisions are not kept. The client is the authority on what the
+ merged state is (docs/playlists.md §6.4) and holds its own copy, so a
+ node keeping history would buy nothing and would mean the node deciding
+ which revision is current — which is exactly what it must not do.
+ """
+ assert self._db
+ await self._db.execute(
+ "INSERT INTO user_blobs (user_id, kind, rev, blob_enc, stored_at) "
+ "VALUES (?, ?, ?, ?, datetime('now')) "
+ "ON CONFLICT(user_id, kind) DO UPDATE SET "
+ " rev = excluded.rev, "
+ " blob_enc = excluded.blob_enc, "
+ " stored_at = excluded.stored_at",
+ (user_id, kind, rev, blob_enc),
+ )
+ await self._db.commit()
+
+ async def fetch_user_blob(self, user_id: str, kind: str) -> dict | None:
+ assert self._db
+ async with self._db.execute(
+ "SELECT rev, blob_enc FROM user_blobs WHERE user_id = ? AND kind = ?",
+ (user_id, kind),
+ ) as cursor:
+ row = await cursor.fetchone()
+ if not row:
+ return None
+ return {"rev": row[0], "blob_enc": row[1]}
+
+ async def list_user_blobs(self, user_id: str) -> list[dict]:
+ """
+ Which blobs exist and at what revision — never a payload.
+
+ A client that has lost its local state (a cache clear, a new device)
+ cannot otherwise discover which playlists exist: their kinds carry
+ client-generated UUIDs, and guessing is not a plan.
+ """
+ assert self._db
+ async with self._db.execute(
+ "SELECT kind, rev FROM user_blobs WHERE user_id = ? ORDER BY kind",
+ (user_id,),
+ ) as cursor:
+ return [{"kind": r[0], "rev": r[1]} for r in await cursor.fetchall()]
+
+ async def delete_user_blob(self, user_id: str, kind: str) -> bool:
+ assert self._db
+ cur = await self._db.execute(
+ "DELETE FROM user_blobs WHERE user_id = ? AND kind = ?",
+ (user_id, kind))
+ await self._db.commit()
+ return cur.rowcount > 0
+
+ async def user_blob_total_bytes(self, user_id: str) -> int:
+ """
+ What this account is already using, so a store can refuse to take it
+ over the per-account cap. An unbounded write primitive pointed at
+ somebody else's disk needs a number that is actually checked.
+ """
+ assert self._db
+ async with self._db.execute(
+ "SELECT COALESCE(SUM(LENGTH(blob_enc)), 0) FROM user_blobs "
+ "WHERE user_id = ?", (user_id,)) as cursor:
+ row = await cursor.fetchone()
+ return int(row[0]) if row else 0
+
# ── Chat epoch keys ──────────────────────────────────────────────────
async def store_chat_epoch(
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index af41061..298b3cd 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -29,6 +29,7 @@ import hashlib
import hmac
import logging
import os
+import re
import struct
import tempfile
import time
@@ -146,6 +147,32 @@ log = logging.getLogger(__name__)
CHUNK_SIZE = 1024 * 1024
MAX_MSG = 64 * 1024 * 1024
+# Per-account blobs (docs/playlists.md §4.3). These are an unbounded write
+# primitive pointed at somebody else's disk, so every one of them is checked —
+# and every check **refuses**, never truncates. A truncating cap silently loses
+# tracks, which is the one failure the whole playlist design exists to prevent.
+#
+# The numbers are sized against the measured shape: ~300 bytes per track before
+# compression, deflate worth about three on a payload this repetitive. A 1 MB
+# body is therefore roughly ten thousand tracks in one playlist, and the
+# manifest holds names and revisions only.
+USER_BLOB_MANIFEST_MAX = 64 * 1024
+USER_BLOB_BODY_MAX = 1024 * 1024
+USER_BLOB_ACCOUNT_MAX = 8 * 1024 * 1024
+
+# "playlists" is the manifest; "playlist:<id>" is one playlist's tracks. A
+# pattern rather than a set, because the ids are client-generated — but a
+# pattern, not anything at all, or the table becomes a key/value store for
+# whatever a client feels like writing.
+#
+# The character class is deliberately wider than a UUID: the reserved id is the
+# word "favorites" (docs/playlists.md §5.1), so a hex-only pattern refuses the
+# one playlist every account has. It stays narrow enough to carry no structure
+# of its own — no "/", no ".", no second ":" — so a kind can never be read as a
+# path or as anything but one name in one namespace.
+_USER_BLOB_KIND_RE = re.compile(
+ r"^(playlists|playlist:[A-Za-z0-9_-]{1,64})$")
+
# Chat link-preview results, kept in memory only (draft-v6 §2.7: the node
# produces enrichment on demand and keeps nothing durable — the asking device
# caches). Bounded and time-limited so a busy group cannot grow it without end
@@ -682,6 +709,14 @@ class WebRTCPeerSession:
self._spawn(self._do_keypair_bundle_store(msg))
elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
self._spawn(self._do_keypair_bundle_delete())
+ elif mtype == MNP.USER_BLOB_STORE:
+ self._spawn(self._do_user_blob_store(msg))
+ elif mtype == MNP.USER_BLOB_FETCH:
+ self._spawn(self._do_user_blob_fetch(msg))
+ elif mtype == MNP.USER_BLOB_LIST:
+ self._spawn(self._do_user_blob_list())
+ elif mtype == MNP.USER_BLOB_DELETE:
+ self._spawn(self._do_user_blob_delete(msg))
elif mtype == MNP.STREAM_REQUEST:
sem = self._ctx.get("_transcode_sem")
log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
@@ -1136,6 +1171,113 @@ class WebRTCPeerSession:
"detail": "keypair_bundle_stored",
})
+ # ── Per-account blobs (playlists, docs/playlists.md §8.2) ────────────────
+ #
+ # The node stores bytes it cannot read and hands them back. `self._user_id`
+ # comes from the authenticated session and never from the message: a
+ # `user_id` in the body would let any member of this group read or
+ # overwrite any other member's blob, which is finding C5 one size down.
+
+ def _user_blob_kind(self, msg: dict) -> str | None:
+ """The validated `kind`, or None having already refused."""
+ kind = msg.get("kind")
+ if not isinstance(kind, str) or not _USER_BLOB_KIND_RE.match(kind):
+ self._send({"type": "error", "detail": "Unknown blob kind"})
+ return None
+ return kind
+
+ def _user_blob_store_ok(self):
+ store = self._ctx.get("bundle_store")
+ if not store:
+ self._send({"type": "error", "detail": "Bundle store not available"})
+ return None
+ if not self._user_id:
+ self._send({"type": "error", "detail": "Not authenticated"})
+ return None
+ return store
+
+ async def _do_user_blob_store(self, msg: dict) -> None:
+ store = self._user_blob_store_ok()
+ if not store:
+ return
+ kind = self._user_blob_kind(msg)
+ if not kind:
+ return
+
+ blob = msg.get("blob_enc")
+ if not isinstance(blob, (bytes, bytearray)) or not blob:
+ self._send({"type": "error", "detail": "Missing blob_enc"})
+ return
+ blob = bytes(blob)
+
+ rev = msg.get("rev")
+ if not isinstance(rev, int) or rev < 0:
+ self._send({"type": "error", "detail": "Missing rev"})
+ return
+
+ limit = (USER_BLOB_MANIFEST_MAX if kind == "playlists"
+ else USER_BLOB_BODY_MAX)
+ if len(blob) > limit:
+ # A stated reason, not a bare error: the client turns this into a
+ # sentence the reader can act on ("this playlist is too large"),
+ # and a refusal nobody can read is a support case.
+ self._send({"type": "error",
+ "detail": f"Blob too large ({len(blob)} > {limit})"})
+ return
+
+ # What the account already uses, minus whatever this call replaces.
+ used = await store.user_blob_total_bytes(self._user_id)
+ existing = await store.fetch_user_blob(self._user_id, kind)
+ if existing:
+ used -= len(existing["blob_enc"])
+ if used + len(blob) > USER_BLOB_ACCOUNT_MAX:
+ self._send({"type": "error",
+ "detail": f"Account blob quota exceeded "
+ f"({used + len(blob)} > {USER_BLOB_ACCOUNT_MAX})"})
+ return
+
+ await store.store_user_blob(self._user_id, kind, rev, blob)
+ self._audit("user_blob_store", f"{kind} rev={rev} bytes={len(blob)}")
+ self._send({"type": "ack", "v": MNP_VERSION, "detail": "user_blob_stored"})
+
+ async def _do_user_blob_fetch(self, msg: dict) -> None:
+ store = self._user_blob_store_ok()
+ if not store:
+ return
+ kind = self._user_blob_kind(msg)
+ if not kind:
+ return
+ row = await store.fetch_user_blob(self._user_id, kind)
+ self._audit("user_blob_fetch", kind)
+ self._send({
+ "type": MNP.USER_BLOB_RESP, "v": MNP_VERSION, "kind": kind,
+ # A kind this account has never written is `null`, not an error:
+ # "no playlist here yet" is the ordinary state of a fresh node and
+ # the client must not read it as a failure.
+ "rev": row["rev"] if row else None,
+ "blob_enc": row["blob_enc"] if row else None,
+ })
+
+ async def _do_user_blob_list(self) -> None:
+ store = self._user_blob_store_ok()
+ if not store:
+ return
+ blobs = await store.list_user_blobs(self._user_id)
+ self._send({"type": MNP.USER_BLOB_LIST_RESP, "v": MNP_VERSION,
+ "blobs": blobs})
+
+ async def _do_user_blob_delete(self, msg: dict) -> None:
+ store = self._user_blob_store_ok()
+ if not store:
+ return
+ kind = self._user_blob_kind(msg)
+ if not kind:
+ return
+ removed = await store.delete_user_blob(self._user_id, kind)
+ self._audit("user_blob_delete", kind)
+ self._send({"type": "ack", "v": MNP_VERSION,
+ "detail": "user_blob_deleted" if removed else "user_blob_absent"})
+
# ── Pairing and join (H3, M3) ────────────────────────────────────────────
def _join_refuse(self, reason: str, audit_detail: str = "") -> None:
diff --git a/packages/meshbay-node/tests/test_user_blob_mnp.py b/packages/meshbay-node/tests/test_user_blob_mnp.py
new file mode 100644
index 0000000..296239f
--- /dev/null
+++ b/packages/meshbay-node/tests/test_user_blob_mnp.py
@@ -0,0 +1,256 @@
+"""
+`user_blob_*` over MNP (MNP 3.1, docs/playlists.md §8.1).
+
+The store itself is covered by `test_user_blob_store.py`. What is here is
+everything the handlers add on top, and each one is a refusal:
+
+ - `user_id` comes from the authenticated session and **never** from the
+ message. A `user_id` in the body would let any member of this group read or
+ overwrite any other member's blob — finding C5, one size down.
+ - `kind` is validated against a pattern, so the table does not become an
+ arbitrary key/value store for whatever a client feels like writing.
+ - every cap **refuses, with a stated reason**, and never truncates. A
+ truncating cap silently loses tracks; a bare `error` reaches the reader as
+ a dead button.
+
+The session is built the way `test_admin_ops_mnp.py` builds one — the real
+handler methods against a real store, with `_send` captured.
+"""
+
+import pytest
+from meshbay_common.protocol import MNP
+from meshbay_node.bundle_store import BundleStore
+from meshbay_node.transport.webrtc_server import (
+ USER_BLOB_ACCOUNT_MAX,
+ USER_BLOB_BODY_MAX,
+ USER_BLOB_MANIFEST_MAX,
+ WebRTCPeerSession,
+)
+
+
+@pytest.fixture
+async def store(tmp_path):
+ s = BundleStore(db_path=tmp_path / "bundles.db")
+ await s.open()
+ yield s
+ await s.close()
+
+
+def _session(store, user_id="alice"):
+ session = WebRTCPeerSession.__new__(WebRTCPeerSession)
+ session._ctx = {"bundle_store": store}
+ session._user_id = user_id
+ session._username = user_id
+ session._group_id = "g" * 32
+ session._remote_ip = ""
+ session.sent = []
+ session._send = session.sent.append
+ session.audited = []
+ session._audit = lambda event, detail="": session.audited.append((event, detail))
+ return session
+
+
+def _last(session):
+ return session.sent[-1] if session.sent else {}
+
+
+# ── the ordinary path ────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_store_then_fetch_returns_the_same_bytes(store):
+ s = _session(store)
+ sealed = bytes(range(256))
+ await s._do_user_blob_store(
+ {"kind": "playlists", "rev": 4, "blob_enc": sealed})
+ assert _last(s)["detail"] == "user_blob_stored"
+
+ await s._do_user_blob_fetch({"kind": "playlists"})
+ reply = _last(s)
+ assert reply["type"] == MNP.USER_BLOB_RESP
+ assert reply["kind"] == "playlists"
+ assert reply["rev"] == 4
+ assert reply["blob_enc"] == sealed
+
+
+@pytest.mark.asyncio
+async def test_fetching_something_never_written_is_null_not_an_error(store):
+ """A fresh node has no playlists, and that is not a failure."""
+ s = _session(store)
+ await s._do_user_blob_fetch({"kind": "playlist:abc"})
+ reply = _last(s)
+ assert reply["type"] == MNP.USER_BLOB_RESP
+ assert reply["rev"] is None and reply["blob_enc"] is None
+
+
+@pytest.mark.asyncio
+async def test_listing_carries_kinds_and_revisions_only(store):
+ s = _session(store)
+ await s._do_user_blob_store({"kind": "playlists", "rev": 1, "blob_enc": b"m"})
+ await s._do_user_blob_store(
+ {"kind": "playlist:abc", "rev": 9, "blob_enc": b"body-bytes"})
+
+ await s._do_user_blob_list()
+ reply = _last(s)
+ assert reply["type"] == MNP.USER_BLOB_LIST_RESP
+ assert reply["blobs"] == [{"kind": "playlist:abc", "rev": 9},
+ {"kind": "playlists", "rev": 1}]
+ assert "blob_enc" not in str(reply["blobs"])
+
+
+@pytest.mark.asyncio
+async def test_delete_removes_it_and_says_so_when_there_was_nothing(store):
+ s = _session(store)
+ await s._do_user_blob_store({"kind": "playlist:x", "rev": 1, "blob_enc": b"b"})
+ await s._do_user_blob_delete({"kind": "playlist:x"})
+ assert _last(s)["detail"] == "user_blob_deleted"
+ await s._do_user_blob_delete({"kind": "playlist:x"})
+ assert _last(s)["detail"] == "user_blob_absent"
+
+
+# ── the refusals ─────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_user_id_in_the_message_is_ignored(store):
+ """The one that matters. Alice writes; Mallory asks for Alice's blob by
+ naming her in the body and gets her own empty slot, not Alice's playlist."""
+ alice = _session(store, "alice")
+ await alice._do_user_blob_store(
+ {"kind": "playlists", "rev": 1, "blob_enc": b"alice-private"})
+
+ mallory = _session(store, "mallory")
+ await mallory._do_user_blob_fetch({"kind": "playlists", "user_id": "alice"})
+ assert _last(mallory)["blob_enc"] is None
+
+ await mallory._do_user_blob_store(
+ {"kind": "playlists", "rev": 99, "blob_enc": b"mallory", "user_id": "alice"})
+ await alice._do_user_blob_fetch({"kind": "playlists"})
+ assert _last(alice)["blob_enc"] == b"alice-private"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", [
+ "", "playlist", "playlists2", "../../etc/passwd", "resume_positions",
+ "playlist:", "playlist:../x", "playlist:" + "a" * 65, 42, None,
+])
+async def test_an_unknown_kind_is_refused(store, kind):
+ s = _session(store)
+ await s._do_user_blob_store({"kind": kind, "rev": 1, "blob_enc": b"x"})
+ assert _last(s)["type"] == "error"
+ assert await store.list_user_blobs("alice") == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", [
+ "playlists", "playlist:favorites", "playlist:abc-123",
+ "playlist:0f8fad5b-d9cb-469f-a165-70867728950e",
+])
+async def test_the_kinds_the_client_actually_writes_are_accepted(store, kind):
+ s = _session(store)
+ await s._do_user_blob_store({"kind": kind, "rev": 1, "blob_enc": b"x"})
+ assert _last(s).get("detail") == "user_blob_stored"
+
+
+@pytest.mark.asyncio
+async def test_an_oversized_body_is_refused_and_nothing_is_written(store):
+ """Refused, not truncated: a truncating cap loses tracks silently, which is
+ the failure playlists exist to prevent."""
+ s = _session(store)
+ await s._do_user_blob_store(
+ {"kind": "playlist:big", "rev": 1, "blob_enc": b"x" * (USER_BLOB_BODY_MAX + 1)})
+ reply = _last(s)
+ assert reply["type"] == "error"
+ assert "too large" in reply["detail"], "the refusal must say why"
+ assert str(USER_BLOB_BODY_MAX) in reply["detail"]
+ assert await store.fetch_user_blob("alice", "playlist:big") is None
+
+
+@pytest.mark.asyncio
+async def test_the_manifest_has_a_tighter_cap_than_a_body(store):
+ """It holds names and revisions; a manifest the size of a playlist means
+ something is writing tracks into the wrong blob."""
+ s = _session(store)
+ assert USER_BLOB_MANIFEST_MAX < USER_BLOB_BODY_MAX
+ await s._do_user_blob_store(
+ {"kind": "playlists", "rev": 1,
+ "blob_enc": b"x" * (USER_BLOB_MANIFEST_MAX + 1)})
+ assert _last(s)["type"] == "error"
+
+ await s._do_user_blob_store(
+ {"kind": "playlists", "rev": 1, "blob_enc": b"x" * USER_BLOB_MANIFEST_MAX})
+ assert _last(s).get("detail") == "user_blob_stored"
+
+
+@pytest.mark.asyncio
+async def test_the_account_quota_bounds_the_whole_collection(store):
+ """Per-blob caps bound one playlist; only this bounds what one account can
+ put on somebody else's disk."""
+ s = _session(store)
+ body = b"x" * USER_BLOB_BODY_MAX
+ fits = USER_BLOB_ACCOUNT_MAX // USER_BLOB_BODY_MAX
+ for i in range(fits):
+ await s._do_user_blob_store(
+ {"kind": f"playlist:p{i}", "rev": 1, "blob_enc": body})
+ assert _last(s).get("detail") == "user_blob_stored", f"blob {i} refused early"
+
+ await s._do_user_blob_store(
+ {"kind": "playlist:over", "rev": 1, "blob_enc": body})
+ assert _last(s)["type"] == "error"
+ assert "quota" in _last(s)["detail"]
+
+
+@pytest.mark.asyncio
+async def test_replacing_a_blob_at_the_quota_is_not_refused(store):
+ """The check subtracts what this write replaces. Without that, an account
+ at its limit could never edit a playlist again — only delete one."""
+ s = _session(store)
+ body = b"x" * USER_BLOB_BODY_MAX
+ for i in range(USER_BLOB_ACCOUNT_MAX // USER_BLOB_BODY_MAX):
+ await s._do_user_blob_store(
+ {"kind": f"playlist:p{i}", "rev": 1, "blob_enc": body})
+
+ await s._do_user_blob_store({"kind": "playlist:p0", "rev": 2, "blob_enc": body})
+ assert _last(s).get("detail") == "user_blob_stored"
+ assert (await store.fetch_user_blob("alice", "playlist:p0"))["rev"] == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("msg", [
+ {"kind": "playlists", "rev": 1}, # no blob
+ {"kind": "playlists", "rev": 1, "blob_enc": b""}, # empty blob
+ {"kind": "playlists", "rev": 1, "blob_enc": "a string"},
+ {"kind": "playlists", "blob_enc": b"x"}, # no rev
+ {"kind": "playlists", "rev": "4", "blob_enc": b"x"}, # rev not a number
+ {"kind": "playlists", "rev": -1, "blob_enc": b"x"},
+])
+async def test_a_malformed_store_is_refused(store, msg):
+ s = _session(store)
+ await s._do_user_blob_store(msg)
+ assert _last(s)["type"] == "error"
+ assert await store.list_user_blobs("alice") == []
+
+
+@pytest.mark.asyncio
+async def test_an_unauthenticated_session_reaches_nothing(store):
+ """`_user_id` is set by the handshake; before it, there is no account to
+ read or write and the handler must not invent one."""
+ s = _session(store, user_id=None)
+ await s._do_user_blob_fetch({"kind": "playlists"})
+ assert _last(s)["type"] == "error"
+ await s._do_user_blob_store({"kind": "playlists", "rev": 1, "blob_enc": b"x"})
+ assert _last(s)["type"] == "error"
+ assert await store.list_user_blobs(None) == []
+
+
+@pytest.mark.asyncio
+async def test_reads_and_writes_are_audited(store):
+ """The node logs that a blob moved, never what was in it — the same line
+ `keypair_bundle_store` already writes."""
+ s = _session(store)
+ await s._do_user_blob_store(
+ {"kind": "playlist:abc", "rev": 2, "blob_enc": b"sealed"})
+ await s._do_user_blob_fetch({"kind": "playlist:abc"})
+ await s._do_user_blob_delete({"kind": "playlist:abc"})
+
+ events = [e for e, _ in s.audited]
+ assert events == ["user_blob_store", "user_blob_fetch", "user_blob_delete"]
+ assert "sealed" not in " ".join(d for _, d in s.audited)
diff --git a/packages/meshbay-node/tests/test_user_blob_store.py b/packages/meshbay-node/tests/test_user_blob_store.py
new file mode 100644
index 0000000..d1bd41e
--- /dev/null
+++ b/packages/meshbay-node/tests/test_user_blob_store.py
@@ -0,0 +1,165 @@
+"""
+Per-account blobs on the node — the playlist store (docs/playlists.md §3.3, §8.2).
+
+The node holds bytes it cannot read, one row per playlist plus a manifest, and
+hands them back to the account that wrote them. Three things have to hold, and
+only the first is obvious:
+
+ - a blob round-trips through the process, byte for byte, as **bytes** — the
+ column is a BLOB rather than base64 TEXT because these run to hundreds of
+ kilobytes and base64 is a third of every write;
+ - the plaintext is never in the file, which is the whole claim; and
+ - every cap **refuses** rather than truncating. A truncating cap silently
+ loses tracks, which is the failure the whole design exists to prevent.
+"""
+
+import pytest
+from meshbay_node.bundle_store import BundleStore
+
+
+@pytest.mark.asyncio
+async def test_a_blob_round_trips_as_bytes(tmp_path):
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+
+ # Every byte value, so a text column or an encoding step anywhere in the
+ # path shows up as a difference rather than surviving by luck.
+ sealed = bytes(range(256)) * 8
+ await store.store_user_blob("u1", "playlists", 3, sealed)
+
+ row = await store.fetch_user_blob("u1", "playlists")
+ assert row == {"rev": 3, "blob_enc": sealed}
+ assert isinstance(row["blob_enc"], bytes)
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_a_blob_survives_the_process(tmp_path):
+ """Reopened from disk, not read back out of the same connection."""
+ db = tmp_path / "bundles.db"
+ sealed = b"\x00\x01sealed-body\xff"
+
+ store = BundleStore(db_path=db)
+ await store.open()
+ await store.store_user_blob("u1", "playlist:abc-123", 41, sealed)
+ await store.close()
+
+ again = BundleStore(db_path=db)
+ await again.open()
+ assert (await again.fetch_user_blob("u1", "playlist:abc-123"))["blob_enc"] == sealed
+ await again.close()
+
+
+@pytest.mark.asyncio
+async def test_one_playlist_is_one_row(tmp_path):
+ """The point of the split: rewriting Favourites must not touch the rest."""
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+
+ await store.store_user_blob("u1", "playlists", 1, b"manifest")
+ await store.store_user_blob("u1", "playlist:favorites", 1, b"fav-v1")
+ await store.store_user_blob("u1", "playlist:evening", 1, b"evening-v1")
+
+ await store.store_user_blob("u1", "playlist:favorites", 2, b"fav-v2")
+
+ assert (await store.fetch_user_blob("u1", "playlist:favorites"))["rev"] == 2
+ assert (await store.fetch_user_blob("u1", "playlist:evening"))["blob_enc"] == b"evening-v1"
+ assert (await store.fetch_user_blob("u1", "playlists"))["blob_enc"] == b"manifest"
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_an_unwritten_kind_is_absent_not_an_error(tmp_path):
+ """"No playlist here yet" is the ordinary state of a fresh node."""
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+ assert await store.fetch_user_blob("u1", "playlists") is None
+ assert await store.list_user_blobs("u1") == []
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_listing_reports_kinds_and_revisions_and_no_payload(tmp_path):
+ """What a client that lost its local state needs, and nothing more: the
+ kinds carry client-generated UUIDs and cannot be guessed."""
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+ await store.store_user_blob("u1", "playlists", 7, b"m")
+ await store.store_user_blob("u1", "playlist:aaa", 2, b"secret-body")
+
+ listing = await store.list_user_blobs("u1")
+ assert listing == [{"kind": "playlist:aaa", "rev": 2},
+ {"kind": "playlists", "rev": 7}]
+ assert "blob_enc" not in listing[0]
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_one_account_never_sees_another(tmp_path):
+ """`user_id` comes from the authenticated session; this is what that buys."""
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+ await store.store_user_blob("alice", "playlists", 1, b"alice")
+ await store.store_user_blob("bob", "playlists", 1, b"bob")
+
+ assert (await store.fetch_user_blob("alice", "playlists"))["blob_enc"] == b"alice"
+ assert await store.list_user_blobs("bob") == [{"kind": "playlists", "rev": 1}]
+
+ await store.delete_user_blob("alice", "playlists")
+ assert await store.fetch_user_blob("alice", "playlists") is None
+ assert await store.fetch_user_blob("bob", "playlists") is not None
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_deleting_something_absent_says_so_rather_than_raising(tmp_path):
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+ assert await store.delete_user_blob("u1", "playlist:gone") is False
+ await store.store_user_blob("u1", "playlist:gone", 1, b"x")
+ assert await store.delete_user_blob("u1", "playlist:gone") is True
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_the_account_total_is_what_the_cap_is_checked_against(tmp_path):
+ """The per-blob caps bound one playlist; only this bounds the account, and
+ an unbounded write primitive pointed at somebody else's disk needs it."""
+ store = BundleStore(db_path=tmp_path / "bundles.db")
+ await store.open()
+ assert await store.user_blob_total_bytes("u1") == 0
+
+ await store.store_user_blob("u1", "playlists", 1, b"x" * 100)
+ await store.store_user_blob("u1", "playlist:a", 1, b"y" * 250)
+ assert await store.user_blob_total_bytes("u1") == 350
+
+ # Replacing a blob replaces its contribution rather than adding to it.
+ await store.store_user_blob("u1", "playlist:a", 2, b"y" * 50)
+ assert await store.user_blob_total_bytes("u1") == 150
+
+ await store.delete_user_blob("u1", "playlist:a")
+ assert await store.user_blob_total_bytes("u1") == 100
+ await store.close()
+
+
+@pytest.mark.asyncio
+async def test_the_plaintext_is_not_in_the_file(tmp_path):
+ """The same check test_chat_key_storage.py makes for epoch keys, for the
+ same reason: a plaintext column beside the sealed one is the obvious thing
+ to write and would collapse the whole claim, silently."""
+ db = tmp_path / "bundles.db"
+ store = BundleStore(db_path=db)
+ await store.open()
+ await store.store_user_blob(
+ "u1", "playlist:abc", 1, b"SEALED-CIPHERTEXT-ONLY")
+ await store.close()
+
+ raw = db.read_bytes()
+ assert b"SEALED-CIPHERTEXT-ONLY" in raw, (
+ "the sealed bytes should be there — this test is only meaningful if it "
+ "is actually reading the right file")
+ # What must never be: a track title, an artist, a path. The store is handed
+ # ciphertext and stores exactly that; anything readable here would mean the
+ # client sealed nothing or the node unwrapped it.
+ for leak in (b"Un titre", b"tracks", b"artist", b"favorites"):
+ assert leak not in raw, f"{leak!r} is in bundles.db in clear"