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>

2 files changed, 421 insertions, 0 deletions

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"