""" The transports must produce the same wire, message for message. `file_chunk` and `index_sync` were each built twice — once in `webrtc_server.py`, once in `quic_server.py` — and the two copies disagreed. WebRTC sent binary, unsigned chunks carrying a `file_id`; QUIC sent base64 fields, two hashes and a per-chunk Ed25519 signature, and no `file_id` at all. `index_sync` was plain entries on one transport and a `GroupIndex.serialize()` envelope on the other. One type, two shapes, and nothing that failed when they drifted. This is the same guard the unified handshake has (`meshbay_common/handshake.py`): the encoders now live in one place, and these tests fail if a transport grows its own copy again. """ import inspect import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common.crypto import generate_gek from meshbay_common.groupbox import PURPOSE_INDEX, unseal from meshbay_common.protocol import MNP, file_chunk_plaintext, file_chunk_wire from meshbay_node.indexer import DirectoryIndexer from meshbay_node.transport import quic_server, webrtc_server from meshbay_node.transport.wire import index_sync_message from conftest import one_root @pytest.fixture def gek(): return generate_gek() @pytest.fixture def shared_dir(tmp_path): d = tmp_path / "shared" d.mkdir() (d / "film.mkv").write_bytes(b"payload " * 500) (d / "sub").mkdir() return d def test_both_transports_use_the_one_chunk_encoder(): """Neither server may encrypt a chunk itself.""" for module in (webrtc_server, quic_server): source = inspect.getsource(module) assert "file_chunk_wire" in source, f"{module.__name__} bypasses the shared encoder" assert "encrypt_chunk_aes(" not in source, ( f"{module.__name__} encrypts a chunk on its own — that is how the two " f"copies diverged the first time") assert "chunk_key_aes(" not in source, ( f"{module.__name__} derives a chunk key on its own") def test_both_transports_use_the_one_index_builder(): for module in (webrtc_server, quic_server): source = inspect.getsource(module) assert "index_sync_message" in source, ( f"{module.__name__} builds index_sync itself") assert "index_b64" not in inspect.getsource(quic_server), ( "QUIC is serializing the index again — that was the fork") def test_neither_transport_seals_by_hand(): """ `groupbox` is the only sealer, the same rule `file_chunk_wire` already has. A server reaching for AESGCM or HKDF directly is a second envelope waiting to disagree with the first about a nonce length, an info string or an AAD. """ for module in (webrtc_server, quic_server): source = inspect.getsource(module) assert "seal(" in source, f"{module.__name__} sends an unsealed ack" assert "AESGCM(" not in source, ( f"{module.__name__} builds its own AEAD instead of using groupbox") assert "HKDF(" not in source, ( f"{module.__name__} derives its own subkey instead of using groupbox") def test_the_daemon_does_not_build_an_index_message_itself(): """ The delta was hand-built in `_broadcast_index_change` — the third construction site for an index message, and the one that would have kept sending cleartext while the other two were sealed. """ from meshbay_node import daemon source = inspect.getsource(daemon) assert "index_delta_message" in source and "index_sync_message" in source assert '"type": MNP.INDEX_DELTA' not in source, ( "the daemon builds index_delta by hand again") assert '"type": MNP.INDEX_SYNC' not in source, ( "the daemon builds index_sync by hand again") def test_chunk_wire_shape_is_identical_across_transports(gek, shared_dir): """The two servers' read-and-encrypt helpers agree on every field but the nonce.""" path = shared_dir / "film.mkv" file_hash = bytes.fromhex("ab" * 32) from_webrtc = webrtc_server._read_and_encrypt(gek, path, 0, file_hash, "ab" * 32) from_quic = quic_server._read_and_encrypt(gek, path, 0, file_hash, "ab" * 32) assert from_webrtc.keys() == from_quic.keys() assert set(from_webrtc) == { "type", "v", "file_id", "chunk_index", "plaintext_size", "nonce", "ct"} assert from_webrtc["type"] == from_quic["type"] == MNP.FILE_CHUNK for field in ("v", "file_id", "chunk_index", "plaintext_size"): assert from_webrtc[field] == from_quic[field] # Binary, not base64 — the 33% Phase 9.15 removed, and the QUIC copy kept. assert isinstance(from_webrtc["nonce"], bytes) assert isinstance(from_webrtc["ct"], bytes) # A fresh nonce per encryption, so the ciphertexts differ while the plaintext # both sides recover does not. assert from_webrtc["ct"] != from_quic["ct"] plaintext = path.read_bytes() for msg in (from_webrtc, from_quic): assert file_chunk_plaintext(gek, msg) == plaintext def test_chunk_round_trip_rejects_a_tampered_ciphertext(gek): msg = file_chunk_wire(gek, b"the payload", 3, bytes.fromhex("cd" * 32), "cd" * 32) msg["ct"] = bytes([msg["ct"][0] ^ 1]) + msg["ct"][1:] with pytest.raises(Exception): file_chunk_plaintext(gek, msg) def test_chunk_key_is_bound_to_file_and_index(gek): """A chunk cannot be replayed as another chunk, or as one of another file.""" file_hash = bytes.fromhex("ef" * 32) msg = file_chunk_wire(gek, b"the payload", 7, file_hash, "ef" * 32) moved = dict(msg, chunk_index=8) with pytest.raises(Exception): file_chunk_plaintext(gek, moved) with pytest.raises(Exception): file_chunk_plaintext(gek, msg, file_hash=bytes.fromhex("11" * 32)) @pytest.mark.asyncio async def test_index_sync_shape(gek, shared_dir): sk_node = Ed25519PrivateKey.generate() roots = one_root(shared_dir) indexer = DirectoryIndexer(roots=roots, group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() msg = index_sync_message(indexer.index, roots) # In clear: what a receiver needs to route and version-check before it can # decrypt, and nothing else. assert msg["type"] == MNP.INDEX_SYNC assert msg["group_id"] == "g" assert set(msg) == {"type", "v", "group_id", "nonce", "ct"} payload = unseal(gek, PURPOSE_INDEX, MNP.INDEX_SYNC, "g", msg) assert [e["name"] for e in payload["entries"]] == ["film.mkv"] # Directories are not index entries, so they travel separately — including the # empty one, which no entry's path would have revealed. assert any(d.endswith("sub") for d in payload["dirs"]) assert payload["roots"] @pytest.mark.asyncio async def test_a_wrong_key_raises_rather_than_reporting_an_empty_group(gek, shared_dir): """ §3.4, at the level a client would hit it. An index that fails to open must not become an empty index: "the group has no files" is a legitimate state, so a fallback there is indistinguishable from the truth — which is exactly what makes it worse than a stop. """ sk_node = Ed25519PrivateKey.generate() roots = one_root(shared_dir) indexer = DirectoryIndexer(roots=roots, group_id="g", sk_node=sk_node, gek=gek) await indexer.initial_scan() msg = index_sync_message(indexer.index, roots) with pytest.raises(Exception) as caught: unseal(generate_gek(), PURPOSE_INDEX, MNP.INDEX_SYNC, "g", msg) # Assert on the refusal, not on a degraded result. assert not isinstance(caught.value, dict)