diff options
Diffstat (limited to 'packages')
11 files changed, 379 insertions, 201 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/__init__.py b/packages/meshbay-common/src/meshbay_common/__init__.py index fdf6d86..61502c9 100644 --- a/packages/meshbay-common/src/meshbay_common/__init__.py +++ b/packages/meshbay-common/src/meshbay_common/__init__.py @@ -61,5 +61,14 @@ __version__ = "0.10.0" # under the account's recovery key, so a forgotten passphrase does not strand # the identity (docs/auth-confirm.md §4.3). Additive: an older node ignores the # field on store and never returns one; an older client never sends it. -MNP_VERSION = "0.14" +# 0.15: `file_chunk` and `index_sync` had forked between the transports — WebRTC +# sent binary, unsigned chunks and plain index entries, QUIC sent base64 chunks +# with a per-chunk Ed25519 signature and a `GroupIndex.serialize()` envelope. One +# type, two shapes, a single consumer each and no test that they agreed. Both now +# come from one encoder (`protocol.file_chunk_wire`, `transport/wire.py`), which is +# QUIC adopting what WebRTC already sent. **Breaking on the QUIC wire**, and only +# there: the WebRTC shape — the one every deployed client speaks — is byte for byte +# what it was, and no QUIC client ships. Recorded as a MINOR bump for that reason; +# a deployed QUIC peer would have made it a MAJOR one. +MNP_VERSION = "0.15" MHP_VERSION = "0.1" diff --git a/packages/meshbay-common/src/meshbay_common/crypto.py b/packages/meshbay-common/src/meshbay_common/crypto.py index b2ff3c0..eadb42b 100644 --- a/packages/meshbay-common/src/meshbay_common/crypto.py +++ b/packages/meshbay-common/src/meshbay_common/crypto.py @@ -225,7 +225,14 @@ def decrypt_keystore(iv: bytes, ciphertext: bytes, tag: bytes, key: bytes) -> by def sign_chunk(sk_node: Ed25519PrivateKey, chunk_index: int, nonce: bytes, ct_hash: bytes) -> bytes: - """Sign chunk metadata. Payload: chunk_index || nonce || ct_hash.""" + """ + Sign chunk metadata. Payload: chunk_index || nonce || ct_hash. + + Despite the name, this no longer signs file chunks — Phase 9.15 dropped per-chunk + signatures on the WebRTC path and 2026-09-03 dropped the QUIC copy that had been + left behind. Its one caller is `GroupIndex.serialize()`, which signs a whole index + envelope under the pseudo-index `INDEX_CHUNK`. + """ payload = chunk_index.to_bytes(4, "big") + nonce + ct_hash return sk_node.sign(payload) diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py index c789287..4d2dd9d 100644 --- a/packages/meshbay-common/src/meshbay_common/protocol.py +++ b/packages/meshbay-common/src/meshbay_common/protocol.py @@ -15,6 +15,11 @@ from typing import Any # second copy here said "0.1" while every message on the wire carried "0.2". # Nothing imported it, which is the only reason it was harmless. from meshbay_common import MNP_VERSION, MHP_VERSION # noqa: F401 (re-export) +from meshbay_common.webcrypto import ( + chunk_key_aes, + decrypt_chunk_aes, + encrypt_chunk_aes, +) # ── MNP message types ───────────────────────────────────────────────────────── @@ -236,21 +241,84 @@ class IndexDelta: updates: list[IndexEntry] = field(default_factory=list) -# ── Chunk request/response ──────────────────────────────────────────────────── +# ── File chunk ──────────────────────────────────────────────────────────────── +# +# One encoder and one decoder, for every transport. +# +# There used to be two. WebRTC moved to a binary wire format in Phase 9.15 (base64 +# costs 33%) and dropped the per-chunk Ed25519 signature with it; the QUIC encoder +# was not brought along, so `file_chunk` meant two different messages depending on +# which transport carried it — base64 fields, hashes and a signature on one, raw +# bytes and a `file_id` on the other. Nothing in the type name said which, and the +# dataclass that used to sit here described only the QUIC half while reading like +# the contract for both. That is finding C6 one size down: two implementations of +# one message, free to drift, with a test for neither. +# +# Why no per-chunk signature: the AES-GCM tag already authenticates the ciphertext +# under a key derived from the GEK, which only group members hold, and since C3 the +# node authenticates itself in the handshake and is pinned by the client. A +# signature per chunk re-proved, once per megabyte, what the session established +# once. (`sign_chunk` still exists in `crypto.py` — it signs the serialized index +# envelope, which is a different artifact; see `indexer/group_index.py`.) -@dataclass -class ChunkRequest: - file_id: str # blake3 hash of file (hex) - chunk_index: int -@dataclass -class ChunkResponse: - chunk_index: int - plaintext_size: int - nonce_b64: str - ct_b64: str - ct_hash_b64: str - pt_hash_b64: str - sig_b64: str - pk_node_b64: str - file_hash_b64: str +def chunk_ciphertext( + gek: bytes, + plaintext: bytes, + chunk_index: int, + file_hash: bytes, +) -> tuple[bytes, bytes]: + """ + `(nonce, ciphertext)` for one chunk. + + Split out because `stream_data` encrypts exactly like `file_chunk` — same key + derivation, indexed by segment instead of by chunk — and differs only in the + message it lands in. It used to do so through its own copy of these two lines. + """ + ckey = chunk_key_aes(gek, file_hash, chunk_index) + return encrypt_chunk_aes(ckey, plaintext) + + +def file_chunk_wire( + gek: bytes, + plaintext: bytes, + chunk_index: int, + file_hash: bytes, + file_id: str = "", +) -> dict: + """ + Encrypt one chunk and build the `file_chunk` message. + + `file_hash` is the raw content hash the chunk key is derived from; `file_id` is + the same value hex-encoded, echoed so a client running several downloads at once + can tell whose reply arrived. Serving a thumbnail or a cached transcode passes + the cache blob's own hash for both. + """ + nonce, ct = chunk_ciphertext(gek, plaintext, chunk_index, file_hash) + return { + "type": MNP.FILE_CHUNK, + "v": MNP_VERSION, + "file_id": file_id, + "chunk_index": chunk_index, + "plaintext_size": len(plaintext), + "nonce": nonce, + "ct": ct, + } + + +def file_chunk_plaintext( + gek: bytes, + msg: dict, + file_hash: bytes | None = None, +) -> bytes: + """ + Decrypt a `file_chunk`. Raises `InvalidTag` if the ciphertext was tampered with. + + `file_hash` defaults to the message's own `file_id`, which is what a client that + asked for a whole file already has. Pass it explicitly only when the caller knows + better than the peer does. + """ + if file_hash is None: + file_hash = bytes.fromhex(msg["file_id"]) + ckey = chunk_key_aes(gek, file_hash, msg["chunk_index"]) + return decrypt_chunk_aes(ckey, msg["nonce"], msg["ct"]) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index 25081ef..2340c78 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -107,8 +107,15 @@ class GroupIndex: def serialize(self) -> bytes: """ - Produce a wire-ready byte string: + Produce a signed index envelope: msgpack → zstd → [GEK encrypt if private] → sign → length-prefixed envelope + + **This is not an MNP message.** It was the payload of `index_sync` on the QUIC + transport, while WebRTC sent plain entries under the same type — one message + type, two encodings (2026-09-03). Both transports now build `index_sync` from + `transport/wire.py`. This stays as a correct at-rest/interchange format, and + as the only thing that signs and encrypts a whole index; read it as that, not + as a wire contract. """ payload = msgpack.packb({ "group_id": self.group_id, diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py index 9102085..4debd27 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -15,7 +15,6 @@ import os import struct from pathlib import Path -import blake3 import jwt import msgpack from aioquic.asyncio import connect, QuicConnectionProtocol @@ -24,9 +23,7 @@ from aioquic.quic.events import QuicEvent, StreamDataReceived from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from meshbay_common import MNP_VERSION -from meshbay_common.crypto import verify_chunk_signature -from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk -from meshbay_common.protocol import MNP +from meshbay_common.protocol import MNP, file_chunk_plaintext from meshbay_common.handshake import ( NONCE_LEN, ROLE_CLIENT, @@ -252,15 +249,28 @@ class QuicChunkClient: stream_id = self._proto._quic.get_next_available_stream_id(is_unidirectional=False) return stream_id - async def fetch_index(self) -> bytes: - """Request the Mesh Group Index.""" + async def fetch_index(self) -> dict: + """ + Request the Mesh Group Index. + + Returns the message itself — `{group_id, version, entries, dirs, roots}` — + which is what the WebRTC client has always received. It used to return the + bytes of a `GroupIndex.serialize()` envelope for the caller to deserialize: + the same message type carrying a different encoding on this transport alone. + """ sid = self._new_stream() self._proto._send(sid, {"type": MNP.INDEX_SYNC, "v": MNP_VERSION}) - msg = await self._proto._recv(sid) - return base64.b64decode(msg["index_b64"]) + return await self._proto._recv(sid) async def fetch_chunk(self, file_id: str, chunk_index: int) -> bytes: - """Fetch, verify, and decrypt one chunk over QUIC.""" + """ + Fetch and decrypt one chunk over QUIC. + + The per-chunk Ed25519 signature this used to verify is gone (see + `meshbay_common.protocol`): the AEAD tag authenticates the ciphertext under a + GEK-derived key, and the node proved its identity in the handshake — which + `connect()` verified and pinned — rather than once per megabyte. + """ sid = self._new_stream() self._proto._send(sid, { "type": MNP.FILE_REQUEST, @@ -273,26 +283,10 @@ class QuicChunkClient: if msg.get("type") == "error": raise LookupError(msg.get("detail", "Unknown error")) - ct = base64.b64decode(msg["ct_b64"]) - nonce = base64.b64decode(msg["nonce_b64"]) - ct_hash = base64.b64decode(msg["ct_hash_b64"]) - pt_hash = base64.b64decode(msg["pt_hash_b64"]) - sig = base64.b64decode(msg["sig_b64"]) - file_hash = base64.b64decode(msg["file_hash_b64"]) - ci = msg["chunk_index"] - - verify_chunk_signature(self._pk_node, ci, nonce, ct_hash, sig) - - if blake3.blake3(ct).digest() != ct_hash: - raise ValueError("Ciphertext hash mismatch") - - ckey = derive_chunk_key(self._gek, file_hash, ci) - plaintext = decrypt_chunk(ckey, nonce, ct) - - if blake3.blake3(plaintext).digest() != pt_hash: - raise ValueError("Plaintext hash mismatch after decryption") - - return plaintext + # From the id we asked for, not the one the answer claims: a peer that + # substitutes it would otherwise choose which key we decrypt with. + return file_chunk_plaintext( + self._gek, msg, file_hash=bytes.fromhex(file_id)) async def fetch_stream_segment( self, file_id: str, segment_index: int, segment_duration: int = 4, diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 30c7daf..8a32538 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -26,7 +26,6 @@ import subprocess from pathlib import Path from typing import Any, Callable -import blake3 import jwt import msgpack from aioquic.asyncio import QuicConnectionProtocol, serve @@ -47,13 +46,10 @@ from meshbay_common.handshake import ( quic_binding, verify_proof, ) -from meshbay_common.crypto import ( - sign_chunk, - pk_to_b64, -) -from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk -from meshbay_common.protocol import MNP +from meshbay_common.crypto import pk_to_b64 +from meshbay_common.protocol import MNP, file_chunk_wire from meshbay_node.indexer import GroupIndex +from meshbay_node.transport.wire import index_sync_message log = logging.getLogger(__name__) @@ -390,12 +386,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): def _do_index_sync_sync(self, stream_id: int) -> None: ctx = self._group_ctx() - wire = ctx["index"].serialize() - self._send(stream_id, { - "type": MNP.INDEX_SYNC, - "v": MNP_VERSION, - "index_b64": base64.b64encode(wire).decode(), - }) + self._send(stream_id, index_sync_message(ctx["index"], ctx.get("roots"))) def _do_file_request_sync(self, stream_id: int, msg: dict) -> None: """Serve file chunk synchronously (blocking I/O — acceptable for test sizes).""" @@ -414,12 +405,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( - self._ctx["sk_node"], - ctx["gek"], - file_path, - chunk_index, - file_hash, - ) + ctx["gek"], file_path, chunk_index, file_hash, entry.id) self._send(stream_id, chunk_data) async def _do_stream_segment(self, stream_id: int, msg: dict) -> None: @@ -520,36 +506,17 @@ class _MNPServerProtocol(QuicConnectionProtocol): def _read_and_encrypt( - sk_node: Ed25519PrivateKey, gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, + file_id: str = "", ) -> dict: - """Read and encrypt one chunk (blocking — runs in executor).""" + """Read one chunk off disk and encrypt it, in the one shape every transport uses.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - - pt_hash = blake3.blake3(plaintext).digest() - ckey = derive_chunk_key(gek, file_hash, chunk_index) - nonce, ct = encrypt_chunk(ckey, plaintext) - ct_hash = blake3.blake3(ct).digest() - sig = sign_chunk(sk_node, chunk_index, nonce, ct_hash) - - return { - "type": MNP.FILE_CHUNK, - "v": MNP_VERSION, - "chunk_index": chunk_index, - "plaintext_size": len(plaintext), - "nonce_b64": base64.b64encode(nonce).decode(), - "ct_b64": base64.b64encode(ct).decode(), - "ct_hash_b64": base64.b64encode(ct_hash).decode(), - "pt_hash_b64": base64.b64encode(pt_hash).decode(), - "sig_b64": base64.b64encode(sig).decode(), - "pk_node_b64": pk_to_b64(sk_node.public_key()), - "file_hash_b64": base64.b64encode(file_hash).decode(), - } + return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) def _extract_segment(file_path: Path, start_time: float, duration: float) -> bytes | None: diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 1f069e7..370a8b4 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -93,8 +93,8 @@ from meshbay_common.join import ( ROLE_OPERATOR, join_transcript, ) -from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes -from meshbay_common.protocol import MNP, index_entry_wire +from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire +from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import linkpreview, ops @@ -2608,50 +2608,7 @@ class WebRTCPeerSession: def _do_index_sync(self) -> None: ctx = self._group_ctx() - idx = ctx["index"] - entries = [index_entry_wire(e) for e in idx.entries] - self._send({ - "type": MNP.INDEX_SYNC, - "v": MNP_VERSION, - "group_id": idx.group_id, - "version": idx.version, - "entries": entries, - # Directories are not index entries, so the client used to infer them - # from file paths — which means a folder someone just created, or one - # they emptied, simply did not exist as far as the UI was concerned. - "dirs": self._list_dirs(ctx.get("roots")), - # Which top-level folders are roots, and whether each is readable. - # A frozen root's files stay listed, so without this a member cannot - # tell "the drive is unplugged" from "it is all still there". - "roots": ctx["roots"].describe() if ctx.get("roots") else [], - }) - - @staticmethod - def _list_dirs(roots: RootSet | None) -> list[str]: - """ - Every directory in the group, as members address them, sorted. - - Each root appears as a directory in its own right, so a root holding no - files yet is still somewhere a member can navigate to and upload into. - An unavailable root is listed too — its content is frozen, not gone, and - hiding it would look exactly like deletion. - """ - if not roots: - return [] - out: list[str] = [] - for root in roots: - out.append(root.name) - if not root.available: - continue - try: - for path in sorted(root.path.rglob("*")): - if path.is_dir() and not path.name.startswith("."): - rel = path.relative_to(root.path) - if not any(part.startswith(".") for part in rel.parts): - out.append(f"{root.name}/{rel.as_posix()}") - except OSError: - continue - return sorted(out)[:2000] + self._send(index_sync_message(ctx["index"], ctx.get("roots"))) async def _try_serve_thumbnail( self, thumb_hash: str, chunk_index: int, gek: bytes | None, @@ -2677,10 +2634,8 @@ class WebRTCPeerSession: if start > len(blob) or (start == len(blob) and chunk_index != 0): return None piece = blob[start:start + CHUNK_SIZE] - return _encrypt_chunk_bytes( - self._ctx["sk_node"], gek, piece, chunk_index, - bytes.fromhex(thumb_hash), thumb_hash, - ) + return file_chunk_wire( + gek, piece, chunk_index, bytes.fromhex(thumb_hash), thumb_hash) async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() @@ -2707,13 +2662,7 @@ class WebRTCPeerSession: getattr(self._channel, "bufferedAmount", "?")) file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( - self._ctx["sk_node"], - ctx["gek"], - file_path, - chunk_index, - file_hash, - entry.id, - ) + ctx["gek"], file_path, chunk_index, file_hash, entry.id) # Backpressure. Without it the node hands the whole window to the # channel at once and the reader sees the first chunk, then nothing for # as long as the link takes to drain the rest. @@ -4435,8 +4384,9 @@ class WebRTCPeerSession: data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break - ckey = chunk_key_aes(gek, file_hash, index) - nonce, ct = encrypt_chunk_aes(ckey, data) + # Same derivation as a file chunk, indexed by segment: one + # implementation, in `meshbay_common.protocol`. + nonce, ct = chunk_ciphertext(gek, data, index, file_hash) self._send({ "type": MNP.STREAM_DATA, "v": MNP_VERSION, @@ -4546,43 +4496,18 @@ class WebRTCPeerSession: await self._pc.close() -def _encrypt_chunk_bytes( - sk_node: Ed25519PrivateKey, - gek: bytes, - plaintext: bytes, - chunk_index: int, - file_hash: bytes, - file_id: str = "", -) -> dict: - ckey = chunk_key_aes(gek, file_hash, chunk_index) - nonce, ct = encrypt_chunk_aes(ckey, plaintext) - - return { - "type": MNP.FILE_CHUNK, - "v": MNP_VERSION, - # Named so a client running several downloads at once can tell whose - # reply this is. It used to carry only the index, which made matching a - # reply to its request a question of arrival order. - "file_id": file_id, - "chunk_index": chunk_index, - "plaintext_size": len(plaintext), - "nonce": nonce, - "ct": ct, - } - - def _read_and_encrypt( - sk_node: Ed25519PrivateKey, gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: + """Read one chunk off disk and encrypt it. Blocking; the caller keeps it short.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) - return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id) + return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) async def _transcode_audio_to_aac(file_path: Path) -> bytes: diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py new file mode 100644 index 0000000..4e09167 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py @@ -0,0 +1,76 @@ +""" +Wire shapes shared by every node transport. + +`file_chunk` lives in `meshbay_common.protocol` — it is pure crypto and shape, so a +client can use the same encoder. This module is for the messages that also need the +node's own view of its disk, which `meshbay-common` cannot see. + +Why it exists at all: `index_sync` was built twice, and the two copies did not agree. +WebRTC sent `{group_id, version, entries, dirs, roots}` — the shape the shipping +client reads — while QUIC sent `{index_b64}`, a signed, compressed, GEK-encrypted +envelope produced by `GroupIndex.serialize()`. Same message type, two encodings, one +consumer each and nothing asserting they matched. Same failure mode as the two +`file_chunk` encoders, and the same fix: one builder, used by both. + +`GroupIndex.serialize()`/`deserialize()` are unchanged and still tested — they remain +a correct signed index envelope — but they no longer describe any MNP message. Read +them as an at-rest/interchange format, not as a wire contract. +""" + +from __future__ import annotations + +from meshbay_common import MNP_VERSION +from meshbay_common.protocol import MNP, index_entry_wire + +from meshbay_node.roots import RootSet + +# A group with a deep tree can hold more directories than anyone will navigate in one +# sitting, and the whole list rides on one message. +MAX_DIRS = 2000 + + +def list_dirs(roots: RootSet | None) -> list[str]: + """ + Every directory in the group, as members address them, sorted. + + Each root appears as a directory in its own right, so a root holding no files yet + is still somewhere a member can navigate to and upload into. An unavailable root + is listed too — its content is frozen, not gone, and hiding it would look exactly + like deletion. + """ + if not roots: + return [] + out: list[str] = [] + for root in roots: + out.append(root.name) + if not root.available: + continue + try: + for path in sorted(root.path.rglob("*")): + if path.is_dir() and not path.name.startswith("."): + rel = path.relative_to(root.path) + if not any(part.startswith(".") for part in rel.parts): + out.append(f"{root.name}/{rel.as_posix()}") + except OSError: + continue + return sorted(out)[:MAX_DIRS] + + +def index_sync_message(index, roots: RootSet | None) -> dict: + """ + The full `index_sync` message for one group. + + `dirs` and `roots` are here because directories are not index entries: without + them a folder someone just created, or one they emptied, does not exist as far as + a client is concerned, and a member cannot tell "the drive is unplugged" from "it + is all still there". + """ + return { + "type": MNP.INDEX_SYNC, + "v": MNP_VERSION, + "group_id": index.group_id, + "version": index.version, + "entries": [index_entry_wire(e) for e in index.entries], + "dirs": list_dirs(roots), + "roots": roots.describe() if roots else [], + } diff --git a/packages/meshbay-node/tests/test_multi_group.py b/packages/meshbay-node/tests/test_multi_group.py index 2828d08..3dc4778 100644 --- a/packages/meshbay-node/tests/test_multi_group.py +++ b/packages/meshbay-node/tests/test_multi_group.py @@ -16,7 +16,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives import serialization from meshbay_common.crypto import generate_gek, pk_to_b64 -from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.indexer import DirectoryIndexer from conftest import one_root from meshbay_node.transport.quic_server import QuicChunkServer from meshbay_node.transport.quic_client import QuicChunkClient @@ -79,9 +79,14 @@ async def multi_group_server(sk_node, sk_hub, gek_a, gek_b, dir_a, dir_b, tmp_pa indexer_b = DirectoryIndexer(roots=one_root(dir_b), group_id="group-b", sk_node=sk_node, gek=gek_b) await indexer_b.initial_scan() + # RootSet, not a bare Path — what the daemon actually puts in a group context. + # This held a Path until 2026-09-03 and nothing noticed: the QUIC index handler + # only called `index.serialize()`, and `entry_abs_path` fell through + # `Path.resolve(strict=...)`, reading the virtual path as a truthy flag and + # returning the right file by accident. groups = { - "group-a": {"gek": gek_a, "roots": dir_a, "index": indexer_a.index}, - "group-b": {"gek": gek_b, "roots": dir_b, "index": indexer_b.index}, + "group-a": {"gek": gek_a, "roots": one_root(dir_a), "index": indexer_a.index}, + "group-b": {"gek": gek_b, "roots": one_root(dir_b), "index": indexer_b.index}, } cert_path = tmp_path / "node.crt" @@ -113,13 +118,12 @@ async def test_user_can_access_own_group( pk_node_b64=pk_to_b64(sk_node.public_key()), group_id="group-a", ) as client: - wire = await client.fetch_index() - recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek_a) - assert recovered.count == 1 + msg = await client.fetch_index() + assert len(msg["entries"]) == 1 - entry = recovered.entries[0] - assert entry.name == "file_a.txt" - chunk = await client.fetch_chunk(entry.id, chunk_index=0) + entry = msg["entries"][0] + assert entry["name"] == "file_a.txt" + chunk = await client.fetch_chunk(entry["id"], chunk_index=0) assert chunk == b"content from group A " * 100 @@ -155,9 +159,8 @@ async def test_dual_group_user_accesses_both( pk_node_b64=pk_to_b64(sk_node.public_key()), group_id="group-a", ) as client_a: - wire_a = await client_a.fetch_index() - idx_a = GroupIndex.deserialize(wire_a, sk_node=sk_node, gek=gek_a) - assert idx_a.entries[0].name == "file_a.txt" + msg_a = await client_a.fetch_index() + assert msg_a["entries"][0]["name"] == "file_a.txt" async with QuicChunkClient( host="127.0.0.1", port=19200, @@ -165,6 +168,5 @@ async def test_dual_group_user_accesses_both( pk_node_b64=pk_to_b64(sk_node.public_key()), group_id="group-b", ) as client_b: - wire_b = await client_b.fetch_index() - idx_b = GroupIndex.deserialize(wire_b, sk_node=sk_node, gek=gek_b) - assert idx_b.entries[0].name == "file_b.txt" + msg_b = await client_b.fetch_index() + assert msg_b["entries"][0]["name"] == "file_b.txt" diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py index 5720043..fb28295 100644 --- a/packages/meshbay-node/tests/test_quic_transport.py +++ b/packages/meshbay-node/tests/test_quic_transport.py @@ -13,7 +13,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives import serialization from meshbay_common.crypto import generate_gek, pk_to_b64 -from meshbay_node.indexer import DirectoryIndexer, GroupIndex +from meshbay_node.indexer import DirectoryIndexer from conftest import one_root from meshbay_node.transport.quic_server import QuicChunkServer, Denylist from meshbay_node.transport.quic_client import QuicChunkClient @@ -95,7 +95,7 @@ async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path): @pytest.mark.asyncio async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): - """QUIC index sync returns deserializable GroupIndex.""" + """QUIC index sync returns the same message shape WebRTC sends.""" hub_pk_pem = sk_hub.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) @@ -121,9 +121,10 @@ async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path): pk_node_b64=pk_to_b64(sk_node.public_key()), group_id="g", ) as client: - wire = await client.fetch_index() - recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek) - assert recovered.count == 2 + msg = await client.fetch_index() + assert msg["group_id"] == "g" + assert len(msg["entries"]) == 2 + assert "dirs" in msg and "roots" in msg await server.stop() @@ -226,8 +227,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat jwt_token=token, gek=gek, pk_node_b64=pk_b64, group_id="g", ) as client: - wire = await client.fetch_index() - assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + assert len((await client.fetch_index())["entries"]) == 2 saved_ticket = client.session_ticket saved_cert = client.peer_cert_der @@ -244,8 +244,7 @@ async def test_quic_session_resumption(sk_node, sk_hub, gek, shared_dir, tmp_pat peer_cert_der=saved_cert, group_id="g", ) as client: - wire = await client.fetch_index() - assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + assert len((await client.fetch_index())["entries"]) == 2 await server.stop() @@ -281,8 +280,7 @@ async def test_quic_denylist_blocks_user(sk_node, sk_hub, gek, shared_dir, tmp_p jwt_token=token, gek=gek, pk_node_b64=pk_b64, group_id="g", ) as client: - wire = await client.fetch_index() - assert GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek).count == 2 + assert len((await client.fetch_index())["entries"]) == 2 # Add user to denylist denylist.deny_user("user-001") diff --git a/packages/meshbay-node/tests/test_transport_wire_parity.py b/packages/meshbay-node/tests/test_transport_wire_parity.py new file mode 100644 index 0000000..5e04525 --- /dev/null +++ b/packages/meshbay-node/tests/test_transport_wire_parity.py @@ -0,0 +1,125 @@ +""" +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 conftest import one_root +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +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 + + +@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_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) + + assert msg["type"] == MNP.INDEX_SYNC + assert msg["group_id"] == "g" + assert [e["name"] for e in msg["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 msg["dirs"]) + assert msg["roots"] |