""" The browser half of MNP 2.0's sealed upload, measured rather than read. `test_upload_sealed.py` (node side) proves the node opens what the shared encoder produces and refuses everything else. This proves the *shipped browser code* produces it — and, the part that matters more, that a caller of `uploadFile` is still told the name the node stored the file under, which now arrives sealed and would otherwise be `undefined` with nothing on screen to say so: a chat attachment would point at a file that is not there. Driven through `harness/upload_seal_probe.mjs`, which runs the shipped `transport.js` over the shipped `crypto.js`. The node half in between is the real `_do_file_upload`, writing to a real directory. A source-reading test can see that `sealGroup` is called. Only this can see whether what comes out of it opens. """ import json import shutil import subprocess import tempfile from pathlib import Path import msgpack import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.protocol import MNP from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import RootSet from meshbay_node.transport.webrtc_server import WebRTCPeerSession STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" PROBE = Path(__file__).resolve().parent / "harness" / "upload_seal_probe.mjs" pytestmark = pytest.mark.skipif( shutil.which("node") is None or not PROBE.exists(), reason="node unavailable — the client half cannot be measured", ) GROUP = "g-upload-probe" CHUNK = 32 BODY = bytes(range(256)) * 3 # 768 bytes → 24 chunks of 32 def _run_probe(payload: dict) -> dict: with tempfile.TemporaryDirectory() as d: f = Path(d) / "input.json" f.write_text(json.dumps(payload)) proc = subprocess.run( ["node", str(PROBE), str(STATIC), str(f)], capture_output=True, text=True, timeout=60, ) if proc.returncode != 0 or not proc.stdout: pytest.fail(f"upload probe failed:\n{proc.stderr}") return json.loads(proc.stdout) def _node_session(tmp_path: Path, gek: bytes) -> WebRTCPeerSession: root = tmp_path / "library" root.mkdir() index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = { "roots": RootSet.build([{"path": str(root), "name": "library", "writable": True}]), "index": index, "sk_node": index.sk_node, "gek": gek, } session._group_id = GROUP session._user_id = "prober" session._pk_user = "" session._uploads = {} session.sent = [] session._send = session.sent.append session._audit = lambda *a, **k: None return session def _probe_input(gek: bytes, mode: str, **extra) -> dict: return { "mode": mode, "gek": gek.hex(), "group_id": GROUP, "node_version": "2.0", "chunk_size": CHUNK, "dir": "library", "root": "library", "file": {"name": "holiday.jpg", "data": BODY.hex()}, **extra, } @pytest.fixture(scope="module") def _gek(): return generate_gek() @pytest.fixture(scope="module") def _sent(_gek): """Every frame the shipped `uploadFile` puts on the wire, unanswered.""" return _run_probe(_probe_input(_gek, "send")) def test_the_browser_puts_no_filename_and_no_content_on_the_wire(_sent): """The whole point, measured on the bytes rather than read off the source.""" assert _sent["frames"], "the client sent nothing" for hexframe in _sent["frames"]: raw = bytes.fromhex(hexframe) assert b"holiday.jpg" not in raw, "the filename is on the wire in clear" msg = msgpack.unpackb(raw, raw=False) assert set(msg) == {"type", "v", "upload_id", "chunk_index", "total_chunks", "nonce", "ct"} assert msg["type"] == MNP.FILE_UPLOAD def test_a_real_node_opens_what_the_real_browser_sealed(tmp_path, _gek, _sent): """ End to end: the shipped browser encoder into the shipped node handler, with the file that lands on disk as the assertion. A mismatch in the HKDF salt, the AAD encoding or the payload shape shows up here as "did not open" — and nowhere else until somebody tries to upload something. """ session = _node_session(tmp_path, _gek) for hexframe in _sent["frames"]: session._do_file_upload(msgpack.unpackb(bytes.fromhex(hexframe), raw=False)) errors = [m for m in session.sent if m.get("type") == "error"] assert not errors, f"the node refused a frame the browser built: {errors[:1]}" root = session._ctx["roots"].roots[0].path assert (root / "holiday.jpg").read_bytes() == BODY assert not list(root.glob("*.part")), "a temp file was left behind" def test_the_caller_is_told_the_name_the_node_chose(tmp_path, _gek, _sent): """ `stored_as` is sealed now, so reading it takes a decrypt that can fail silently. It must not: the node finds a free name rather than replacing anything, and a chat attachment that never learns which name points at nothing. The acks below are the ones the node really produced — only their `upload_id`, which is outside the seal, is retargeted to the second probe run's own upload. """ session = _node_session(tmp_path, _gek) # A file of that name is already there, so the node has to choose another. (session._ctx["roots"].roots[0].path / "holiday.jpg").write_bytes(b"someone else's") for hexframe in _sent["frames"]: session._do_file_upload(msgpack.unpackb(bytes.fromhex(hexframe), raw=False)) acks = [msgpack.packb(m, use_bin_type=True).hex() for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK] assert len(acks) == len(_sent["frames"]) result = _run_probe(_probe_input(_gek, "receive", acks=acks)) assert result["state"] == "resolved", result.get("message") assert result["stored"]["stored_as"] == "holiday (2).jpg" assert result["stored"]["dir"] == "library" def test_the_client_refuses_an_older_node_before_sending_a_chunk(_gek): """ A 1.x node would answer "Missing filename or data" — an error about the wrong thing, naming no upload, which fails every upload in flight. Asked first instead, and nothing goes on the wire. """ result = _run_probe(_probe_input(_gek, "receive", acks=[], node_version="1.1")) assert result["state"] == "rejected" assert "older MeshBay" in result["message"] assert result["frames"] == [], "a chunk was sent to a node that cannot open it"