diff options
Diffstat (limited to 'packages/meshbay-node')
10 files changed, 740 insertions, 333 deletions
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 b22b8df..af87b70 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py @@ -317,22 +317,3 @@ class QuicChunkClient: # 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, - ) -> bytes: - """Fetch one HLS segment (MPEG-TS bytes) over QUIC.""" - sid = self._new_stream() - self._proto._send(sid, { - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "segment_duration": segment_duration, - }) - msg = await self._proto._recv(sid, timeout=30.0) - - if msg.get("type") == "error": - raise LookupError(msg.get("detail", "Unknown error")) - - return base64.b64decode(msg["data_b64"]) 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 284b488..e34153c 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -22,7 +22,6 @@ import base64 import logging import os import struct -import subprocess import uuid from pathlib import Path from typing import Any, Callable @@ -54,7 +53,6 @@ from meshbay_common.groupbox import PURPOSE_ACK, seal from meshbay_common.protocol import MNP, file_chunk_wire from meshbay_node.indexer import GroupIndex from meshbay_node.transport.wire import index_sync_message -from meshbay_node import platform log = logging.getLogger(__name__) @@ -62,16 +60,6 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] -# ffmpeg is spawned per STREAM_SEGMENT request, and `_extract_segment` runs -# `subprocess.run` synchronously — so without a bound, an authenticated peer can -# both fork-bomb the node and block its event loop for up to 30 s per request -# (finding M2c). Extraction now runs in a thread and passes through this -# semaphore. Small on purpose: the QUIC path has no shipping client yet, this is -# parity work with the WebRTC transcode cap. -_MAX_CONCURRENT_SEGMENTS = 4 -_segment_sem = asyncio.Semaphore(_MAX_CONCURRENT_SEGMENTS) - - class Denylist: """ Denylist for revoked users, groups and invalidated JWTs. @@ -247,8 +235,6 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._do_index_sync_sync(stream_id) elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) - elif mtype == MNP.STREAM_SEGMENT: - self._spawn(self._do_stream_segment(stream_id, msg)) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message_sync(stream_id, msg) elif mtype == MNP.PING: @@ -435,49 +421,6 @@ class _MNPServerProtocol(QuicConnectionProtocol): 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: - """ - Extract and serve one segment via ffmpeg — off the event loop and behind - a concurrency bound, so one request can neither stall the whole node nor - fork-bomb it (finding M2c). The WebRTC path has had both since Phase 11.5. - """ - try: - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) - - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send(stream_id, {"type": "error", "detail": "File not found"}) - return - - file_path = entry_abs_path(ctx["roots"], entry) - if not file_path.exists(): - self._send(stream_id, {"type": "error", "detail": "File not on disk"}) - return - - start_time = segment_index * segment_duration - loop = asyncio.get_event_loop() - async with _segment_sem: - segment_data = await loop.run_in_executor( - None, _extract_segment, file_path, start_time, segment_duration) - if segment_data is None: - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - return - - self._send(stream_id, { - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "data_b64": base64.b64encode(segment_data).decode(), - "size": len(segment_data), - }) - except Exception as e: - log.error("stream_segment: %s", e) - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - def _do_chat_message_sync(self, stream_id: int, msg: dict) -> None: """ Store a chat message and broadcast it to the rest of THIS group. @@ -548,25 +491,6 @@ def _read_and_encrypt( 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: - """Extract one HLS segment via ffmpeg. Returns MPEG-TS bytes or None on failure.""" - try: - result = subprocess.run( - [platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", - "-ss", str(start_time), - "-i", str(file_path), - "-t", str(duration), - "-c:v", "copy", "-c:a", "copy", - "-f", "mpegts", "pipe:1"], - capture_output=True, timeout=30, - ) - if result.returncode == 0 and result.stdout: - return result.stdout - return None - except Exception: - return None - - # ── QuicChunkServer ──────────────────────────────────────────────────────────── class QuicChunkServer: 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 9774831..4e4a23f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -24,6 +24,7 @@ Signaling flow (handled externally by the hub): import asyncio import base64 +import contextvars import hashlib import hmac import logging @@ -105,9 +106,18 @@ from meshbay_common.join import ( ROLE_OPERATOR, join_transcript, ) -from meshbay_common.protocol import MNP, chunk_ciphertext, file_chunk_wire -from meshbay_common.chatbox import NONCE_LEN as CHAT_NONCE_LEN, SIG_LEN as CHAT_SIG_LEN -from meshbay_node.chat import FORMAT_PLAIN, FORMAT_SEALED_V1, ReplayedMessage +from meshbay_common.chatbox import ( + NONCE_LEN as CHAT_NONCE_LEN, + SIG_LEN as CHAT_SIG_LEN, +) +from meshbay_common.protocol import ( + MNP, + chunk_ciphertext, + file_chunk_wire, + file_upload_ack_wire, + file_upload_payload, +) +from meshbay_node.chat import FORMAT_SEALED_V1, ReplayedMessage from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer @@ -263,6 +273,32 @@ def _pack(obj: dict) -> bytes: _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 +# The request this session is currently answering, as (session, req_id). +# +# MNP has never carried a correlation id: a reply named its own type and +# nothing else, so a client with more than one request outstanding had to guess +# which one a message answered — by arrival order, for every reply the client +# could not key off a field of its own. The guess is wrong whenever two replies +# reorder, and catastrophically wrong for the replies that name *nothing*: this +# module sends `{"type": "error"}` from 240 places and two of them name what +# they are about. A refusal therefore reached no caller at all, and the request +# it belonged to waited out the client's 30s timeout while some unrelated +# request was resolved with the refusal instead. Live symptom, found 2026-09-06: +# the Chat composer is disabled while a send is in flight, so a chat message +# whose reply went astray froze the tab for 30 seconds. +# +# `req_id` closes it: whatever the caller put on the request is stamped on the +# reply. A ContextVar rather than a parameter because the alternative is +# threading an argument through all 240 send sites — and asyncio copies the +# current context into a task, so a handler that `_spawn`s its real work still +# answers under the id of the request that started it. +# +# The session is held alongside the id because a handler may send to *other* +# sessions as well as its own (a chat broadcast, an index push): those are not +# replies to anything and must not be stamped. _send checks the owner. +_REPLY_TO: contextvars.ContextVar[tuple] = contextvars.ContextVar( + "meshbay_reply_to", default=(None, None)) + class _DataChannelBuffer: """ @@ -416,6 +452,22 @@ class WebRTCPeerSession: ) def _handle_message(self, msg: dict) -> None: + """Answer one MNP message, under the correlation id it carries. + + The id is published for the whole handler — see _REPLY_TO — so that + every reply _send puts on the wire, including the ones a spawned task + sends much later and the generic refusal below, names the request it + answers. Resetting on the way out only clears it for *this* call: a + task spawned in between captured its own copy of the context when it + was created and keeps answering under the right id. + """ + token = _REPLY_TO.set((self, msg.get("req_id"))) + try: + self._dispatch_message(msg) + finally: + _REPLY_TO.reset(token) + + def _dispatch_message(self, msg: dict) -> None: mtype = msg.get("type") log.debug("WebRTC recv: %s", mtype) try: @@ -459,8 +511,6 @@ class WebRTCPeerSession: # Chunks are matched by file and index on the client, so # answering out of order is safe. self._spawn(self._do_file_request(msg)) - elif mtype == MNP.STREAM_SEGMENT: - self._do_stream_segment(msg) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: @@ -3172,7 +3222,14 @@ class WebRTCPeerSession: def _group_ctx(self) -> dict: if "groups" in self._ctx and self._group_id: - return self._ctx["groups"][self._group_id] + # `.get`, not a bare subscript. A config reload removes a group + # from this map (daemon.py's reload does `groups_ctx.pop`) while + # sessions connected to it are still open, and the next request + # any of them made raised KeyError into _dispatch_message's + # catch-all. An absent group now reads the way an unconfigured + # one already does — the handlers all test for what they need — + # instead of failing every request the session has left. + return self._ctx["groups"].get(self._group_id) or {} return self._ctx def _indexing_status(self) -> dict: @@ -4005,71 +4062,6 @@ class WebRTCPeerSession: "director": director, } - def _do_stream_segment(self, msg: dict) -> None: - self._spawn(self._do_stream_segment_async(msg)) - - async def _do_stream_segment_async(self, msg: dict) -> None: - """ - Legacy HLS segment extraction (superseded by stream_req/MSE). - - Finding H6: this ran subprocess.run(..., timeout=30) directly inside the - event loop, so a single request stalled the whole daemon — every peer, - every group — for up to thirty seconds. Now async and under the same - transcode semaphore as _stream_video. - """ - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) - - entry = ctx["index"].get_entry(file_id) - if not entry: - self._send({"type": "error", "detail": "File not found"}) - return - - file_path = entry_abs_path(ctx["roots"], entry) - if not file_path.exists(): - self._send({"type": "error", "detail": "File not on disk"}) - return - - sem = self._transcode_semaphore() - - try: - async with sem: - proc = await asyncio.create_subprocess_exec( - platform.ffmpeg_cmd(), "-hide_banner", "-loglevel", "error", - "-ss", str(segment_index * segment_duration), - "-i", str(file_path), - "-t", str(segment_duration), - "-c:v", "copy", "-c:a", "copy", - "-f", "mpegts", "pipe:1", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, - ) - try: - stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30) - except asyncio.TimeoutError: - proc.kill() - await proc.wait() - self._send({"type": "error", "detail": "Segment extraction timed out"}) - return - if proc.returncode != 0 or not stdout: - self._send({"type": "error", "detail": "Segment extraction failed"}) - return - segment_data = stdout - except Exception: - self._send({"type": "error", "detail": "Segment extraction failed"}) - return - - self._send({ - "type": MNP.STREAM_SEGMENT, - "v": MNP_VERSION, - "file_id": file_id, - "segment_index": segment_index, - "data_b64": base64.b64encode(segment_data).decode(), - "size": len(segment_data), - }) - def _do_chat_message(self, msg: dict) -> None: """ Store one message and hand it to everyone else in this group. @@ -4417,27 +4409,88 @@ class WebRTCPeerSession: self._send(resp) def _do_file_upload(self, msg: dict) -> None: + """ + One chunk of an upload, sealed under the group key (MNP 2.0). + + Sealing this direction is not symmetry for its own sake. Downloads have + been under a GEK-derived key since the beginning; uploads carried the + filename and the raw bytes in plain msgpack, so the same file was + ciphertext leaving a node and plaintext arriving at one. The node holds + the GEK for its own group, so it opens the payload here — before it + decides a destination, before it touches the disk — and refuses a chunk + that does not open. + + `upload_id` is the correlation key and stays in clear; `filename`, `dir` + and `root` moved inside the seal, which is why every refusal below names + the upload rather than the file. A `code` says which refusal it is, and + the client already knows what it sent. + """ ctx = self._group_ctx() - filename = msg.get("filename", "") - chunk_index = msg.get("chunk_index", 0) - total_chunks = msg.get("total_chunks", 1) - data = msg.get("data") + upload_id = str(msg.get("upload_id") or "")[:64] + + gek = ctx.get("gek") + if not gek: + self._send({"type": "error", "detail": "Group encryption not initialized", + "code": "no_group_key", "upload_id": upload_id}) + return - if not filename or data is None: - self._send({"type": "error", "detail": "Missing filename or data", - "filename": filename}) + try: + payload = file_upload_payload(gek, self._group_id or "", msg) + except Exception: + # Deliberately one answer for "not sealed at all" and "sealed wrong": + # distinguishing them tells a peer which of the two it got right. + # An MNP 1.x client lands here, which is the whole of the upgrade + # story — everything else it does still works. + self._audit("upload_refused", "unsealed") + self._send({ + "type": "error", + "detail": "This upload did not open under the group key — the " + "client may be running an older version", + "code": "upload_not_sealed", + "upload_id": upload_id, + }) return + filename = payload.get("filename") or "" + data = payload.get("data") + # From the clear part of the message, so peer-controlled and unchecked + # by the AEAD. Everything below compares and adds to them. + try: + chunk_index = int(msg.get("chunk_index", 0)) + total_chunks = int(msg.get("total_chunks", 1)) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid chunk index", + "code": "bad_chunk_index", "upload_id": upload_id}) + return + + def _refuse(detail: str, code: str = "") -> None: + """A refusal names the upload, never the file: the name is sealed.""" + out = {"type": "error", "detail": detail, "upload_id": upload_id} + if code: + out["code"] = code + self._send(out) + + # Types first, and before any state is created. What comes out of a + # sealed payload is authenticated, not validated: it is msgpack a + # member wrote, and `SAFE_UPLOAD_NAME.match(123)` raises where a + # refusal was meant. + if not isinstance(filename, str) or not filename: + _refuse("Missing filename or data", "upload_incomplete") + return + # Bytes, always: base64 was the shape of the old plaintext `data` field + # and there is no sealed message that can carry a string here. + if not isinstance(data, (bytes, bytearray)): + _refuse("Invalid chunk encoding", "bad_chunk_encoding") + return + chunk_bytes = bytes(data) + if not SAFE_UPLOAD_NAME.match(filename): - self._send({"type": "error", "detail": "Invalid filename", - "filename": filename}) + _refuse("Invalid filename", "invalid_filename") return roots: RootSet | None = ctx.get("roots") if not roots: - self._send({"type": "error", - "detail": "No directories configured for this group", - "filename": filename}) + _refuse("No directories configured for this group", "no_roots") return # The client names the root it is uploading into — it is browsing one, @@ -4448,47 +4501,32 @@ class WebRTCPeerSession: # An unknown name is refused rather than falling back to a writable # root, because "the file went somewhere else" is discovered weeks # later — the same reason the old single upload root was never guessed. - # A client that names nothing is an MNP 1.0 one, and there was exactly - # one destination in its world: the first writable root. # `dir` is the folder being browsed, as a virtual path # (`Media/Films/1999`); `root` is the older, coarser form and is what - # its first segment means on its own. - target_rel = str(msg.get("dir") or "").strip().strip("/") + # its first segment means on its own. Both are sealed now, so a refusal + # below can no longer quote them back. + target_rel = str(payload.get("dir") or "").strip().strip("/") target_root_name = (target_rel.split("/")[0] if target_rel - else str(msg.get("root") or "").strip()) + else str(payload.get("root") or "").strip()) upload_root = None if target_root_name: upload_root = roots.by_name(target_root_name) if upload_root is None: - self._send({"type": "error", - "detail": f"No directory named " - f"{target_root_name!r} in this group", - "code": "no_such_root", - "filename": filename}) + _refuse("No such directory in this group", "no_such_root") return else: writable = roots.writable_roots upload_root = writable[0] if writable else None if upload_root is None: - self._send({"type": "error", - "detail": "No writable directory in this group", - "code": "no_writable_root", - "filename": filename}) + _refuse("No writable directory in this group", "no_writable_root") return if not upload_root.writable: - self._send({"type": "error", - "detail": f"Directory '{upload_root.name}' is read-only", - "code": "root_read_only", - "filename": filename}) + _refuse("That directory is read-only", "root_read_only") self._audit("upload_refused", filename[:64]) return if not upload_root.available: - self._send({"type": "error", - "detail": f"Directory '{upload_root.name}' is " - f"currently unavailable", - "code": "root_unavailable", - "filename": filename}) + _refuse("That directory is currently unavailable", "root_unavailable") return # The folder the sender is looking at, and no subdirectory of the node's @@ -4512,23 +4550,17 @@ class WebRTCPeerSession: if target_rel: target_dir = roots.resolve(target_rel) if target_dir is None or not target_dir.is_dir(): - self._send({"type": "error", - "detail": "Not a directory in this group", - "code": "no_such_directory", - "filename": filename}) + _refuse("Not a directory in this group", "no_such_directory") return rel_dir = target_rel else: - # An MNP 1.0 client names nothing; the root itself is where its one - # destination now is. + # A client that names nothing: the first writable root is where its + # one destination is. target_dir = upload_root.path rel_dir = upload_root.name if not target_dir.is_dir(): - self._send({"type": "error", - "detail": f"Directory '{upload_root.name}' is " - f"currently unavailable", - "code": "root_unavailable", - "filename": filename}) + _refuse("That directory is currently unavailable", + "root_unavailable") return upload_key = f"{rel_dir}/{filename}" @@ -4544,33 +4576,24 @@ class WebRTCPeerSession: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. if final_path.exists(): - self._send({"type": "error", "detail": "File already exists", - "filename": filename}) + _refuse("File already exists", "already_exists") return state = {"next_index": 0, "bytes": 0, "stored_name": stored_name} self._uploads[upload_key] = state elif state is None: - self._send({"type": "error", "detail": "Upload not started", - "filename": filename}) + _refuse("Upload not started", "not_started") return # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends # blindly to whatever .part file is already on disk. if chunk_index != state["next_index"]: - self._send({"type": "error", "detail": "Unexpected chunk index", - "filename": filename}) + _refuse("Unexpected chunk index", "bad_chunk_index") return - if isinstance(data, str): - chunk_bytes = base64.b64decode(data) - else: - chunk_bytes = bytes(data) - if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES: self._uploads.pop(upload_key, None) tmp_path.unlink(missing_ok=True) - self._send({"type": "error", "detail": "Upload exceeds size limit", - "filename": filename}) + _refuse("Upload exceeds size limit", "too_large") return with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f: @@ -4578,16 +4601,16 @@ class WebRTCPeerSession: state["next_index"] = chunk_index + 1 state["bytes"] += len(chunk_bytes) - self._send({ - "type": MNP.FILE_UPLOAD_ACK, - "v": MNP_VERSION, - "chunk_index": chunk_index, - "filename": filename, + self._send(file_upload_ack_wire( + gek, self._group_id or "", + upload_id=upload_id, + chunk_index=chunk_index, + filename=filename, # What it is actually called on disk, which a chat attachment has to # reference and the uploader deserves to be told. - "stored_as": stored_name, - "dir": rel_dir, - }) + stored_as=stored_name, + dir=rel_dir, + )) if chunk_index + 1 >= total_chunks: self._uploads.pop(upload_key, None) @@ -5384,6 +5407,16 @@ class WebRTCPeerSession: self._audit("stream_video", entry.name) def _send(self, obj: dict) -> None: + # Stamp the reply with the id of the request being answered, so the + # caller never has to guess. Only for this session's own replies: a + # handler that also pushes to other peers (a chat broadcast, an index + # delta) reaches them through *their* _send, where the owner no longer + # matches and nothing is stamped — those messages answer no request. + # An explicit req_id already on the object wins, and an unsolicited + # push (no request in scope) carries none, exactly as before. + owner, req_id = _REPLY_TO.get() + if req_id is not None and owner is self and "req_id" not in obj: + obj = {**obj, "req_id": req_id} if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) else: diff --git a/packages/meshbay-node/tests/conftest.py b/packages/meshbay-node/tests/conftest.py index 3dc9cd9..ba86c13 100644 --- a/packages/meshbay-node/tests/conftest.py +++ b/packages/meshbay-node/tests/conftest.py @@ -39,3 +39,34 @@ def one_root(path: Path, *, name: str = "", kind: str = "generic", """ return RootSet.build([{"path": str(path), "name": name, "kind": kind, "writable": writable}]) + + +def sealed_upload(session, *, filename: str, data: bytes, + chunk_index: int = 0, total_chunks: int = 1, + dir: str = "", root: str = "", + upload_id: str = "up-test") -> dict: + """ + A `file_upload` message as the shipping client builds one (MNP 2.0). + + Built through `file_upload_wire`, not by hand: a test that assembles the + wire shape itself is a second encoder, and a second encoder is how + `file_chunk` and `index_sync` forked between the transports (finding C6) + with nobody noticing. The key and the AAD are taken off the session, so + these agree with the handler by construction rather than by copying. + """ + from meshbay_common.protocol import file_upload_wire + + ctx = session._group_ctx() + return file_upload_wire( + ctx["gek"], session._group_id or "", + upload_id=upload_id, chunk_index=chunk_index, total_chunks=total_chunks, + filename=filename, data=data, dir=dir, root=root, + ) + + +def opened_ack(session, msg: dict) -> dict: + """The payload of a `file_upload_ack` the node sent, opened as a client would.""" + from meshbay_common.protocol import file_upload_ack_payload + + ctx = session._group_ctx() + return file_upload_ack_payload(ctx["gek"], session._group_id or "", msg) diff --git a/packages/meshbay-node/tests/test_reply_correlation.py b/packages/meshbay-node/tests/test_reply_correlation.py new file mode 100644 index 0000000..bfb336e --- /dev/null +++ b/packages/meshbay-node/tests/test_reply_correlation.py @@ -0,0 +1,165 @@ +""" +A reply names the request it answers. + +MNP carried no correlation id until 2026-09-07. A reply named its own type and +nothing else, so a client with more than one request outstanding had to work out +which one a message answered from the message itself — and for the replies that +name nothing, it could not. This module sends `{"type": "error"}` from 240 +places and two of them say what they are about; `_dispatch_message`'s catch-all +is one of the 238. Such a refusal reached no caller at all: the browser handed +it to whichever request happened to be waiting, and the request it belonged to +sat until its own 30s timeout. Live symptom (2026-09-06): the Chat composer is +disabled while a send is in flight, so a chat message whose refusal went astray +froze the tab for thirty seconds. + +`req_id` is the client's own pending-map key, put on the wire and stamped back +onto the reply by `_send`. What matters here, and what the browser cannot check +for itself: + + * a reply carries it, including the refusals that name nothing else; + * a *broadcast* does not — it answers no request, and stamping it would hand + another peer's client a reply to a request it never made; + * work handed to a background task still answers under the right id, which is + why this is a ContextVar and not an attribute on the session. +""" +import asyncio +import base64 + +import msgpack +import pytest +from meshbay_common.protocol import MNP +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +pytestmark = pytest.mark.asyncio + +# A device key's raw bytes. Only its length and its identity with the +# connection's own pin matter here; nothing verifies a signature over it. +_DEVICE = b"\x07" * 32 + + +class _Channel: + readyState = "open" + + def __init__(self): + self.sent = [] + + def send(self, framed): + # Skip the 4-byte length prefix _pack writes. + self.sent.append(msgpack.unpackb(framed[4:], raw=False)) + + +def _session(peer_id="p"): + s = WebRTCPeerSession.__new__(WebRTCPeerSession) + s._ctx = {} + s._peer_id = peer_id + s._user_id = None + s._group_id = "" + s._channel = _Channel() + s._tasks = set() + return s + + +async def test_a_refusal_that_names_nothing_else_names_the_request(): + """The reply at the root of the defect: no type of its own to match on.""" + s = _session() + # No handshake yet, so any other message is refused — a bare `error`, the + # same shape the catch-all sends and the same shape a browser could not + # route. + s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 41}) + + (reply,) = s._channel.sent + assert reply["type"] == "error" + assert reply["req_id"] == 41, ( + "a refusal that names neither the request nor a type of its own is a " + "reply no caller can claim") + + +async def test_the_catch_all_refusal_names_the_request_too(): + """Every failure in the dispatch loop funnels into one generic reply.""" + s = _session() + s._user_id = "u" + + def _boom(msg): + raise RuntimeError("filesystem path that must not reach the peer") + s._do_chat_message = _boom + + s._handle_message({"type": MNP.CHAT_MESSAGE, "req_id": 7}) + + (reply,) = s._channel.sent + assert reply == {"type": "error", "detail": "Request failed", "req_id": 7}, ( + "the catch-all is where an unforeseen failure ends up, so it is exactly " + "the reply that must still be routable") + + +async def test_a_request_without_an_id_is_answered_without_one(): + """An older client sends none; nothing may be invented for it.""" + s = _session() + s._handle_message({"type": MNP.CHAT_MESSAGE}) + + (reply,) = s._channel.sent + assert "req_id" not in reply + + +async def test_a_broadcast_to_another_peer_is_not_stamped(): + """The reply goes to the asker; the broadcast goes to everyone else. + + They travel out of the same handler, and only the first answers anything. + Stamping the second would hand another browser a reply keyed to a pending + request of its own that it never sent — the very confusion this fixes. + """ + asker, other = _session("asker"), _session("other") + asker._user_id, other._user_id = "a", "b" + registry = {"ka": asker, "kb": other} + asker._peer_registry = lambda: registry + asker._user_names = lambda: {} + asker._audit = lambda *a, **k: None + asker._group_ctx = lambda: {} + asker._spawn = lambda coro: coro.close() + + asker._registry_key, other._registry_key = "ka", "kb" + asker._pinned_pk = base64.b64encode(_DEVICE).decode() + asker._device_confirmed = True + + # A sealed message, because MNP 2.0 has no plaintext chat and the node + # refuses one — the bytes need not decrypt, since nothing here opens them. + # What this test is about is unchanged: which of the two messages leaving + # this handler carries the id. + asker._handle_message({ + "type": MNP.CHAT_MESSAGE, "req_id": 3, + "format": 1, "epoch": 1, "device": _DEVICE, "ct": b"ciphertext", + "nonce": b"\x02" * 12, "sig": b"\x03" * 64, + }) + + (ack,) = asker._channel.sent + assert ack["type"] == "ack" and ack["req_id"] == 3 + (broadcast,) = other._channel.sent + assert broadcast["type"] == MNP.CHAT_MESSAGE + assert "req_id" not in broadcast, ( + "a broadcast answers no request and must not look like a reply") + + +async def test_work_handed_to_a_task_still_answers_under_the_right_id(): + """Most handlers `_spawn` their real work, and the reply leaves long after + the dispatch call that started it has returned. + + This is the reason the id lives in a ContextVar: asyncio copies the current + context into a task, so the answer keeps the id even though nothing passed + it along. An attribute on the session would have been overwritten by the + next message to arrive in the meantime. + """ + s = _session() + s._user_id = "u" + + async def _late(reply): + await asyncio.sleep(0.01) + s._send({"type": "roster_read_resp", "detail": reply}) + s._do_chat_message = lambda msg: s._spawn(_late(msg["payload"])) + + s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "first", "req_id": 11}) + # A second request arrives while the first one's task is still asleep. + s._handle_message({"type": MNP.CHAT_MESSAGE, "payload": "second", "req_id": 12}) + await asyncio.gather(*list(s._tasks)) + + by_id = {m["req_id"]: m["detail"] for m in s._channel.sent} + assert by_id == {11: "first", 12: "second"}, ( + "a late reply answered under whichever request arrived most recently") diff --git a/packages/meshbay-node/tests/test_root_writable_policy.py b/packages/meshbay-node/tests/test_root_writable_policy.py index 7eb75fd..8345880 100644 --- a/packages/meshbay-node/tests/test_root_writable_policy.py +++ b/packages/meshbay-node/tests/test_root_writable_policy.py @@ -21,13 +21,14 @@ anything. A deprecated instruction that still works is not deprecated, and this one would reopen uploads group-wide. """ -import base64 from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from conftest import sealed_upload from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG +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 @@ -47,6 +48,8 @@ def _session(tmp_path: Path, user_id: str, *, "index": index, "sk_node": index.sk_node, "node_user_id": operator, + # Uploads are sealed under the group key since MNP 2.0. + "gek": generate_gek(), } session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = ctx @@ -61,11 +64,8 @@ def _session(tmp_path: Path, user_id: str, *, def _upload(session, filename="clip.mp4", body=b"bytes"): - session._do_file_upload({ - "filename": filename, "dir": "shared", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(body).decode(), - }) + session._do_file_upload(sealed_upload( + session, filename=filename, data=body, dir="shared")) def _uploads_dir(session) -> Path: diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 1a318f7..988aa46 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -16,10 +16,11 @@ from pathlib import Path import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek from meshbay_common.protocol import IndexEntry from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roots import RootSet -from conftest import one_root +from conftest import one_root, opened_ack, sealed_upload from meshbay_node.transport.webrtc_server import WebRTCPeerSession @@ -156,7 +157,10 @@ def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: shared_root = tmp_path / "shared" shared_root.mkdir(exist_ok=True) index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) - ctx = {"roots": one_root(shared_root), "index": index, "sk_node": index.sk_node} + # A group key, because uploads are sealed under it since MNP 2.0 — the + # handler opens the payload before it has a filename to refuse. + ctx = {"roots": one_root(shared_root), "index": index, + "sk_node": index.sk_node, "gek": generate_gek()} session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = ctx @@ -188,12 +192,8 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): original.write_bytes(b"operator's original content") attacker = _session(tmp_path, "attacker-user") - attacker._do_file_upload({ - "filename": "important.mp4", - "chunk_index": 0, - "total_chunks": 1, - "data": base64.b64encode(b"attacker content").decode(), - }) + attacker._do_file_upload(sealed_upload( + attacker, filename="important.mp4", data=b"attacker content")) assert original.read_bytes() == b"operator's original content", ( "an upload replaced an existing file (C5a)") @@ -203,12 +203,15 @@ def test_upload_cannot_overwrite_another_members_file(tmp_path): def test_upload_second_attempt_cannot_replace_own_completed_file(tmp_path): """C5a: even the original uploader does not get to overwrite.""" session = _session(tmp_path, "user-1") - payload = {"filename": "movie.mp4", "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"first").decode()} - session._do_file_upload(dict(payload)) - session.sent.clear() + def _send_it(): + # Sealed afresh each time: a nonce is drawn per message, so re-sending + # the same dict would be a replay rather than a second upload. + session._do_file_upload(sealed_upload( + session, filename="movie.mp4", data=b"first")) - session._do_file_upload(dict(payload)) + _send_it() + session.sent.clear() + _send_it() uploads = _uploads_dir(session) assert (uploads / "movie.mp4").read_bytes() == b"first", ( "the first upload was replaced") @@ -252,11 +255,8 @@ def test_the_client_names_a_folder_and_never_a_filesystem_path(tmp_path): for bad in ("../../etc", "/etc", "shared/../..", "shared/../../etc", "nope", "shared/missing"): session.sent.clear() - session._do_file_upload({ - "filename": "note.txt", "dir": bad, - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir=bad)) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal, f"{bad!r} was accepted" assert refusal[0].get("code") in ("no_such_root", "no_such_directory"), bad @@ -274,11 +274,8 @@ def test_an_upload_lands_in_the_folder_it_names(tmp_path): root = session._ctx["roots"].roots[0] (root.path / "Albums").mkdir() - session._do_file_upload({ - "filename": "note.txt", "dir": f"{root.name}/Albums", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir=f"{root.name}/Albums")) assert (root.path / "Albums" / "note.txt").read_bytes() == b"x" assert not (root.path / "Albums" / "uploads").exists(), ( @@ -304,11 +301,8 @@ def test_an_upload_goes_to_the_root_it_names(tmp_path): {"path": str(incoming), "writable": True}, ]) - session._do_file_upload({ - "filename": "note.txt", "dir": "Incoming", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir="Incoming")) assert (incoming / "note.txt").read_bytes() == b"x" assert not (media / "note.txt").exists(), "it went to the first root instead" @@ -327,11 +321,8 @@ def test_a_read_only_root_refuses_an_upload(tmp_path): session._ctx["roots"] = RootSet.build([{"path": str(published)}]) session._is_node_admin = lambda: True - session._do_file_upload({ - "filename": "note.txt", "dir": "Published", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir="Published")) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_read_only" @@ -349,11 +340,8 @@ def test_a_fully_read_only_group_refuses_an_unaddressed_upload(tmp_path): session = _session(tmp_path, "user-1") session._ctx["roots"] = RootSet.build([{"path": str(published)}]) - session._do_file_upload({ - "filename": "note.txt", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x")) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "no_writable_root" @@ -375,11 +363,8 @@ def test_an_ejected_root_refuses_an_upload(tmp_path): roots.roots[0].available = False session._ctx["roots"] = roots - session._do_file_upload({ - "filename": "note.txt", "dir": "USB", - "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"x").decode(), - }) + session._do_file_upload(sealed_upload( + session, filename="note.txt", data=b"x", dir="USB")) refusal = [m for m in session.sent if m.get("type") == "error"] assert refusal and refusal[0].get("code") == "root_unavailable" @@ -392,22 +377,22 @@ def test_two_members_can_send_the_same_filename(tmp_path): IMG_1234.jpg. The second gets a free name; neither replaces the other. """ first = _session(tmp_path, "user-1") - first._do_file_upload({ - "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"first").decode(), - }) + first._do_file_upload(sealed_upload( + first, filename="IMG_1234.jpg", data=b"first")) second = _session(tmp_path, "user-2") - second._do_file_upload({ - "filename": "IMG_1234.jpg", "chunk_index": 0, "total_chunks": 1, - "data": base64.b64encode(b"second").decode(), - }) + # Same group, so the same key: `_session` builds one per call, and two + # members of one group do not have two. + second._ctx["gek"] = first._ctx["gek"] + second._do_file_upload(sealed_upload( + second, filename="IMG_1234.jpg", data=b"second")) uploads = _uploads_dir(first) assert (uploads / "IMG_1234.jpg").read_bytes() == b"first" assert (uploads / "IMG_1234 (2).jpg").read_bytes() == b"second" ack = [m for m in second.sent if m.get("type") == "file_upload_ack"][-1] - assert ack["stored_as"] == "IMG_1234 (2).jpg", ( + assert "stored_as" not in ack, "the name the node chose must be sealed" + assert opened_ack(second, ack)["stored_as"] == "IMG_1234 (2).jpg", ( "the sender must be told the name that was used, or a chat attachment " "points at someone else's file") @@ -743,20 +728,48 @@ def test_pre_handshake_message_budget_is_small(): list(buf.messages()) -def test_stream_segment_is_not_synchronous(): +def test_no_transport_ships_media_outside_the_aead(): """ - H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop, - stalling every peer on the node for up to thirty seconds per request. + `stream_seg` served an MPEG-TS segment as base64 with no encryption at all + — the one content-plane message that never went through a GEK-derived key, + on both transports, answering any authenticated member. Its browser caller + was defined and never invoked. Removed in MNP 2.0 rather than repaired: + `stream_data` already does the job under `chunk_ciphertext`. - Asserts the property (the worker is a coroutine, ffmpeg is spawned through - asyncio) rather than grepping for "subprocess.run" — which also matches the - comment that documents the old behaviour. + Asserted as the property, not as "the function is gone": what matters is + that no transport has a field carrying media bytes past the AEAD. The old + H6 test lived here — it pinned `_do_stream_segment_async` to a coroutine so + ffmpeg could not block the event loop — and the handler outliving that + concern is exactly what this replaces. """ - import ast - import inspect - from meshbay_node.transport.webrtc_server import WebRTCPeerSession + import re + + from meshbay_common.protocol import MNP + + assert not hasattr(MNP, "STREAM_SEGMENT"), ( + "the constant outliving the handlers is how a deleted endpoint keeps " + "looking like part of the wire contract") + + root = Path(__file__).parent.parent / "src" / "meshbay_node" / "transport" + for name in ("webrtc_server.py", "quic_server.py", "quic_client.py"): + source = (root / name).read_text(encoding="utf-8") + # Word boundaries: `_stream_segments` and `STREAM_SEGMENT_SIZE` belong + # to the live `stream_data` path, which is encrypted and stays. + assert not re.search(r"\bstream_seg\b", source), ( + f"{name} still speaks stream_seg") + assert not re.search(r"\bSTREAM_SEGMENT\b", source), ( + f"{name} still names the removed type") + assert "data_b64" not in source, ( + f"{name} carries a base64 media field — media leaves this node " + "encrypted or not at all") - assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async) + +def test_ffmpeg_never_blocks_the_event_loop(): + """ + H6, the half that survives `stream_seg`: the live streaming path still + spawns ffmpeg, and a synchronous spawn stalls every peer on the node. + """ + import ast source = (Path(__file__).parent.parent / "src" / "meshbay_node" / "transport" / "webrtc_server.py").read_text(encoding="utf-8") diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py index 3a5b8a5..9dffedb 100644 --- a/packages/meshbay-node/tests/test_task_lifetime.py +++ b/packages/meshbay-node/tests/test_task_lifetime.py @@ -245,8 +245,13 @@ def test_chunks_wait_for_room_on_the_channel(session): work through the rest — which is what "stuck at 1 MB" looks like, one chunk being exactly one megabyte. """ - fn = session[session.index("async def _do_file_request"):] - fn = fn[:fn.index("\n def _do_stream_segment")] + start = session.index("async def _do_file_request") + # Up to whatever the next member is. This used to end at + # "\n def _do_stream_segment" — a neighbour removed in MNP 2.0 — and an + # `index()` on a name that no longer exists fails the test for a reason + # that has nothing to do with what it is about. + nxt = re.search(r"\n (?:@|(?:async )?def )", session[start:]) + fn = session[start:start + nxt.start()] if nxt else session[start:] assert "DOWNLOAD_BUFFER_HIGH" in fn, "the send buffer has to be watched" assert "await asyncio.sleep" in fn, "waiting for room is the point" assert 'readyState != "open"' in fn, ( diff --git a/packages/meshbay-node/tests/test_upload_sealed.py b/packages/meshbay-node/tests/test_upload_sealed.py new file mode 100644 index 0000000..7c1be96 --- /dev/null +++ b/packages/meshbay-node/tests/test_upload_sealed.py @@ -0,0 +1,285 @@ +""" +The write path, sealed under the group key (MNP 2.0). + +Downloads have been encrypted under a GEK-derived key since the beginning: +`file_chunk` and `stream_data` both go through `chunk_ciphertext`. Uploads did +not. `file_upload` carried the filename and the raw bytes in plain msgpack and +`file_upload_ack` carried the name the node stored them under, so the same file +was ciphertext leaving a node and plaintext arriving at one — an asymmetry with +no threat model behind it. + +What sealing buys is what `groupbox.py` says and no more: nothing against a +network observer (DTLS covers that), nothing against the hub (never on this +channel), nothing against a member (they hold the GEK). It buys defence in +depth against our own next handshake bug, of a class already shipped twice — +C1, the unauthenticated node HTTP API, and C6, the transport that took a bare +JWT. Both were "a peer that had not finished the handshake was served data". +Sealed, the equivalent bug on this path leaks ciphertext instead of the +operator's filenames. +""" + +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_common.protocol import MNP, file_upload_wire +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root, opened_ack, sealed_upload + +GROUP = "g" * 32 + + +def _session(tmp_path: Path, *, gek: bytes | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = {"roots": one_root(shared_root), "index": index, + "sk_node": index.sk_node, "gek": gek or generate_gek()} + session._group_id = GROUP + session._user_id = "user-1" + session._pk_user = "" + session._uploads = {} + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _root(session): + return session._ctx["roots"].roots[0] + + +def _errors(session): + return [m for m in session.sent if m.get("type") == "error"] + + +def _wrote_anything(tmp_path) -> bool: + return any(p.is_file() for p in tmp_path.rglob("*")) + + +# ── The message itself ─────────────────────────────────────────────────────── + +def test_the_wire_message_carries_no_filename_and_no_plaintext(tmp_path): + """ + The point of the exercise. `upload_id` and `chunk_index` are outside the + seal because the node routes and orders on them before it can decrypt; + everything that names or is the operator's content is inside it. + """ + session = _session(tmp_path) + msg = sealed_upload(session, filename="holiday.jpg", data=b"JPEGDATA", + dir=f"{_root(session).name}") + + assert set(msg) == {"type", "v", "upload_id", "chunk_index", + "total_chunks", "nonce", "ct"} + blob = repr(msg).encode() + msg["ct"] + assert b"holiday.jpg" not in blob, "the filename is on the wire in clear" + assert b"JPEGDATA" not in blob, "the file content is on the wire in clear" + + +def test_the_ack_carries_no_stored_name(tmp_path): + """ + `stored_as` is the name the node settled on — it finds a free one rather + than replacing anything — and naming it in clear would hand back exactly + what the request took the trouble to hide. + """ + session = _session(tmp_path) + session._do_file_upload(sealed_upload( + session, filename="holiday.jpg", data=b"x", dir=_root(session).name)) + + ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1] + assert set(ack) == {"type", "v", "upload_id", "chunk_index", "nonce", "ct"} + assert b"holiday.jpg" not in repr(ack).encode() + ack["ct"] + assert opened_ack(session, ack) == { + "filename": "holiday.jpg", "stored_as": "holiday.jpg", + "dir": _root(session).name} + + +def test_the_ack_names_the_upload_so_one_refusal_fails_one_upload(tmp_path): + """ + `filename` used to be the correlation key on both sides. It cannot be one + any more, and `upload_id` replaces it — a client-chosen label, opaque to + the node, never an authorization input. Without it a client running several + uploads could only match replies by arrival order, which is how a refusal + for one file used to fail every upload in flight. + """ + session = _session(tmp_path) + session._do_file_upload(sealed_upload( + session, filename="a.txt", data=b"x", dir=_root(session).name, + upload_id="upload-A")) + session._do_file_upload(sealed_upload( + session, filename="../evil", data=b"x", dir=_root(session).name, + upload_id="upload-B")) + + ack = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK][-1] + assert ack["upload_id"] == "upload-A" + assert _errors(session)[0]["upload_id"] == "upload-B" + + +# ── What is refused ────────────────────────────────────────────────────────── + +def test_a_plaintext_upload_is_refused(tmp_path): + """ + The MNP 1.x shape, which is what an un-updated client sends. Refused with a + code and a message saying which side is old — never accepted "just this + once", because a path that still takes plaintext is not a sealed path. + """ + session = _session(tmp_path) + session._do_file_upload({ + "filename": "note.txt", "dir": _root(session).name, + "chunk_index": 0, "total_chunks": 1, "data": b"x", + }) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_a_tampered_chunk_is_refused(tmp_path): + """ + AES-GCM's tag, asserted where it matters: a flipped bit in the ciphertext + must stop the upload, not produce a corrupt file with a plausible name. + """ + session = _session(tmp_path) + msg = sealed_upload(session, filename="note.txt", data=b"x" * 64, + dir=_root(session).name) + msg["ct"] = bytes([msg["ct"][0] ^ 0x01]) + msg["ct"][1:] + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_an_upload_sealed_for_another_group_is_refused(tmp_path): + """ + The group is the AAD, so a node hosting two groups cannot have a chunk + moved between them — and a member of one cannot write into the other by + reaching a session that is on it (finding H1's shape, on the write path). + """ + session = _session(tmp_path) + msg = file_upload_wire( + session._ctx["gek"], "some-other-group", + upload_id="u1", chunk_index=0, total_chunks=1, + filename="note.txt", data=b"x", dir=_root(session).name) + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_an_upload_under_another_key_is_refused(tmp_path): + """A peer past the handshake with the wrong GEK still writes nothing.""" + session = _session(tmp_path) + msg = file_upload_wire( + generate_gek(), GROUP, + upload_id="u1", chunk_index=0, total_chunks=1, + filename="note.txt", data=b"x", dir=_root(session).name) + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "upload_not_sealed" + assert not _wrote_anything(tmp_path) + + +def test_an_ack_replayed_as_a_request_does_not_open(tmp_path): + """ + The message type is in the AAD, so the two halves of an upload cannot be + confused for each other. Cheap, and it closes a class that is tedious to + reason about after the fact. + """ + from cryptography.exceptions import InvalidTag + from meshbay_common.protocol import file_upload_ack_wire, file_upload_payload + + session = _session(tmp_path) + ack = file_upload_ack_wire( + session._ctx["gek"], GROUP, upload_id="u1", chunk_index=0, + filename="note.txt", stored_as="note.txt", dir="shared") + with pytest.raises(InvalidTag): + file_upload_payload(session._ctx["gek"], GROUP, ack) + + +def test_a_group_with_no_key_refuses_rather_than_falling_back(tmp_path): + """ + A node whose group has no GEK yet cannot open anything. It must say so, not + read the message as though it were the old plaintext shape. + """ + session = _session(tmp_path) + msg = sealed_upload(session, filename="note.txt", data=b"x", + dir=_root(session).name) + session._ctx["gek"] = b"" + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "no_group_key" + assert not _wrote_anything(tmp_path) + + +# ── What must still work ───────────────────────────────────────────────────── + +def test_a_multi_chunk_upload_reassembles(tmp_path): + """ + Every chunk is sealed under its own nonce, and the node appends in order. + Nothing about the seal may change what lands on disk. + """ + session = _session(tmp_path) + body = bytes(range(256)) * 40 + parts = [body[i:i + 1024] for i in range(0, len(body), 1024)] + for i, part in enumerate(parts): + session._do_file_upload(sealed_upload( + session, filename="blob.bin", data=part, + chunk_index=i, total_chunks=len(parts), dir=_root(session).name)) + + assert (_root(session).path / "blob.bin").read_bytes() == body + assert not list(_root(session).path.glob("*.part")), "a temp file was left" + acks = [m for m in session.sent if m.get("type") == MNP.FILE_UPLOAD_ACK] + assert [m["chunk_index"] for m in acks] == list(range(len(parts))) + + +def test_two_identical_chunks_do_not_reuse_a_nonce(tmp_path): + """ + A file of repeated bytes is ordinary, so the nonce must come from the RNG + and never from the payload. Cheap to assert and expensive to discover. + """ + session = _session(tmp_path) + a = sealed_upload(session, filename="f", data=b"same", chunk_index=0) + b = sealed_upload(session, filename="f", data=b"same", chunk_index=0) + assert a["nonce"] != b["nonce"] + assert a["ct"] != b["ct"] + + +def test_a_sealed_payload_is_authenticated_not_validated(tmp_path): + """ + Opening a payload proves a member wrote it, not that they wrote something + sensible. A member can seal anything, so the fields still need their types + checked — `SAFE_UPLOAD_NAME.match(123)` raises where a refusal was meant, + and the dispatcher's catch-all would turn that into "Request failed". + """ + from meshbay_common.groupbox import PURPOSE_UPLOAD, seal + + session = _session(tmp_path) + for payload in ({"filename": 123, "data": b"x"}, + {"filename": "note.txt", "data": "not bytes"}, + {"filename": "note.txt"}): + session.sent.clear() + session._do_file_upload({ + "type": MNP.FILE_UPLOAD, "v": "2.0", "upload_id": "u1", + "chunk_index": 0, "total_chunks": 1, + **seal(session._ctx["gek"], PURPOSE_UPLOAD, MNP.FILE_UPLOAD, + GROUP, payload), + }) + errs = _errors(session) + assert errs, f"{payload!r} was accepted" + assert errs[0]["code"] in ("upload_incomplete", "bad_chunk_encoding") + assert not _wrote_anything(tmp_path) + + +def test_a_peer_controlled_chunk_index_cannot_crash_the_handler(tmp_path): + """`chunk_index` is outside the seal by necessity, so it is unchecked input.""" + session = _session(tmp_path) + msg = sealed_upload(session, filename="note.txt", data=b"x", + dir=_root(session).name) + msg["chunk_index"] = "zero" + session._do_file_upload(msg) + + assert _errors(session)[0]["code"] == "bad_chunk_index" + assert not _wrote_anything(tmp_path) diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index ea13d96..c1a5287 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -783,36 +783,6 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir): @pytest.mark.asyncio -async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir): - """WebRTC DataChannel: stream_segment for non-existent file returns error.""" - hub_pk_pem = _hub_pk_pem(sk_hub) - indexer = DirectoryIndexer(roots=one_root(shared_dir), group_id="g", sk_node=sk_node, gek=gek) - await indexer.initial_scan() - - transport = WebRTCTransport( - sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek, - roots=one_root(shared_dir), index=indexer.index, - stun_servers=[], - ) - - browser_pc, channel, received = await _setup_peer( - transport, sk_hub, gek, "peer-stream") - - channel.send(_pack({ - "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION, - "file_id": "nonexistent-file-id", - "segment_index": 0, "segment_duration": 4, - })) - - msg = await asyncio.wait_for(received.get(), timeout=5.0) - assert msg["type"] == "error" - assert "not found" in msg["detail"].lower() - - await browser_pc.close() - await transport.close_all() - - -@pytest.mark.asyncio async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir): """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership.""" hub_pk_pem = _hub_pk_pem(sk_hub) |