diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
5 files changed, 126 insertions, 157 deletions
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 [], + } |