""" `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)