diff options
Diffstat (limited to 'packages')
| -rw-r--r-- | packages/meshbay-node/tests/test_webrtc_transport.py | 179 |
1 files changed, 179 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index 284c461..3010fe9 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -9,6 +9,7 @@ Uses local loopback (no STUN/ICE needed for localhost). import asyncio import base64 +from contextlib import asynccontextmanager import hashlib import hmac import os @@ -1770,3 +1771,181 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_ await bundle_store.close() await browser_pc.close() await transport.close_all() + + +# ── Per-account blobs, MNP 3.1 (docs/playlists.md §15.2) ───────────────────── +# +# `test_user_blob_mnp.py` drives these six handlers directly, which proves what +# they decide but nothing about how their payloads travel: it never builds a +# frame and never crosses a channel. A playlist body is the largest `bin` value +# this protocol carries after a file chunk, and the failure it would hit is +# silent — a playlist that does not come back reports no error, it is simply +# absent, which is exactly what was seen on a phone once already. + + +@asynccontextmanager +async def _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path): + """ + A node with a bundle store, and peers, torn down whatever happens. + + The `finally` is not tidiness. `BundleStore` runs an aiosqlite thread, and a + test that fails before closing it leaves that thread alive — the process + then hangs in `threading._shutdown`, *after* pytest has printed the failure + and the summary. Found by breaking the node on purpose to check these tests + catch it: they did, in 0.66s, and then the run never ended. A red suite is a + result; a stuck one is an outage. + """ + indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", + sk_node=sk_node, gek=gek) + await indexer.initial_scan() + + bundle_store = BundleStore(db_path=tmp_path / "bundles.db") + await bundle_store.open() + + transport = WebRTCTransport( + sk_node=sk_node, hub_pk_pem=_hub_pk_pem(sk_hub), gek=gek, + roots=one_root(shared_dir), index=indexer.index, + stun_servers=[], + ) + transport._ctx["bundle_store"] = bundle_store + + peers: list = [] + try: + yield transport, peers + finally: + for pc in peers: + await pc.close() + await bundle_store.close() + await transport.close_all() + + +@pytest.mark.asyncio +async def test_user_blobs_round_trip_over_a_live_datachannel( + sk_node, sk_hub, gek, shared_dir, tmp_path): + """ + Sealed bytes out, the same bytes back, across two connections. + + The blobs are `os.urandom`, deliberately: sealed output is incompressible + and uses every byte value, so anything that treats this as text — a UTF-8 + decode, msgpack `str` instead of `bin` — corrupts it. Bytes the node could + have round-tripped by accident would prove nothing. + + The second connection is the case this feature exists for: a device that was + not the one that wrote. + """ + async with _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path) as ( + transport, peers): + manifest = os.urandom(512) + # Bodies are padded to a multiple of 4 KB before sealing, so this is the + # smallest one a real client ever sends. + body = os.urandom(4096) + + pc1, ch1, q1 = await _setup_peer(transport, sk_hub, gek, "peer-blob-write") + peers.append(pc1) + + ch1.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION, + "kind": "playlists", "rev": 7, "blob_enc": manifest})) + ack = await asyncio.wait_for(q1.get(), timeout=5.0) + assert ack["type"] == "ack" + assert ack["detail"] == "user_blob_stored" + + ch1.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION, + "kind": "playlist:favorites", "rev": 3, "blob_enc": body})) + ack = await asyncio.wait_for(q1.get(), timeout=5.0) + assert ack["detail"] == "user_blob_stored" + + await pc1.close() + + # A second device, same account, which has never seen either blob. + pc2, ch2, q2 = await _setup_peer(transport, sk_hub, gek, "peer-blob-read") + peers.append(pc2) + + # What is here, and at what revision. This is the call that lets a fresh + # device discover kinds it cannot guess — they carry client-made ids — + # and it must not carry payloads. + ch2.send(_pack({"type": MNP.USER_BLOB_LIST, "v": MNP_VERSION})) + listing = await asyncio.wait_for(q2.get(), timeout=5.0) + assert listing["type"] == MNP.USER_BLOB_LIST_RESP + assert listing["blobs"] == [{"kind": "playlist:favorites", "rev": 3}, + {"kind": "playlists", "rev": 7}] + + # `req_id` is what lets the client match a reply to its request rather + # than to whatever arrives next; over a channel carrying several + # requests at once, arrival order is not an answer. + ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION, + "kind": "playlists", "req_id": "r-91"})) + resp = await asyncio.wait_for(q2.get(), timeout=5.0) + assert resp["type"] == MNP.USER_BLOB_RESP + assert resp.get("req_id") == "r-91" + assert resp["kind"] == "playlists" + assert resp["rev"] == 7 + assert isinstance(resp["blob_enc"], bytes), "came back as something else than bin" + assert resp["blob_enc"] == manifest + + ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION, + "kind": "playlist:favorites"})) + resp = await asyncio.wait_for(q2.get(), timeout=5.0) + assert resp["rev"] == 3 + assert resp["blob_enc"] == body + + # A kind never written is an absence, not an error: the ordinary state + # of a fresh node. + ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION, + "kind": "playlist:never-written"})) + resp = await asyncio.wait_for(q2.get(), timeout=5.0) + assert resp["rev"] is None + assert resp["blob_enc"] is None + + # And a delete reclaims it, which is how a tombstoned playlist stops + # costing an account its quota. + ch2.send(_pack({"type": MNP.USER_BLOB_DELETE, "v": MNP_VERSION, + "kind": "playlist:favorites"})) + ack = await asyncio.wait_for(q2.get(), timeout=5.0) + assert ack["detail"] == "user_blob_deleted" + + ch2.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION, + "kind": "playlist:favorites"})) + resp = await asyncio.wait_for(q2.get(), timeout=5.0) + assert resp["blob_enc"] is None + + +@pytest.mark.asyncio +async def test_a_large_user_blob_round_trips_whole( + sk_node, sk_hub, gek, shared_dir, tmp_path): + """ + A quarter-megabyte blob, out and back, byte for byte. + + This is the risk §15.2 named, and measuring it corrected how it was + described: a 256 KB blob arrives as **one** application message of 262213 + bytes, not as fragments the four-byte length prefix reassembles. SCTP + fragments it and puts it back together underneath. What this pins is that + nothing in the node's own framing truncates or re-encodes a `bin` value of + that size. + + Note which direction is proved. A *browser* cannot send a frame this large + to this node: aiortc advertises `a=max-message-size:65536`, so Chrome + refuses anything above it — which is why uploads chunk at 48 KB + (`transport.js:41`). The store below is aiortc talking to aiortc and says + nothing about that ceiling. The fetch is the direction that matters here: a + node answers with a whole body, and a long playlist is the largest one. + """ + async with _user_blob_node(sk_node, sk_hub, gek, shared_dir, tmp_path) as ( + transport, peers): + # 256 KB: about nine hundred tracks at the 270 bytes a sealed track + # measures. + body = os.urandom(256 * 1024) + + pc, ch, q = await _setup_peer(transport, sk_hub, gek, "peer-blob-big") + peers.append(pc) + + ch.send(_pack({"type": MNP.USER_BLOB_STORE, "v": MNP_VERSION, + "kind": "playlist:long", "rev": 1, "blob_enc": body})) + ack = await asyncio.wait_for(q.get(), timeout=10.0) + assert ack["detail"] == "user_blob_stored", ack + + ch.send(_pack({"type": MNP.USER_BLOB_FETCH, "v": MNP_VERSION, + "kind": "playlist:long"})) + resp = await asyncio.wait_for(q.get(), timeout=10.0) + assert resp["type"] == MNP.USER_BLOB_RESP + assert len(resp["blob_enc"]) == len(body), "truncated" + assert resp["blob_enc"] == body |