diff options
Diffstat (limited to 'packages/meshbay-node')
6 files changed, 45 insertions, 204 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 360b9ac..29dfef2 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 from pathlib import Path from typing import Any, Callable @@ -53,7 +52,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__) @@ -61,16 +59,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. @@ -243,8 +231,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: @@ -431,49 +417,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. @@ -542,25 +485,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 aaf3f81..f8b6c03 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -479,8 +479,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: @@ -3801,71 +3799,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: # Per-group store — see _peer_registry() and finding H1. Reading chat_store # off the shared transport context sent every group's messages to the first diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 1a318f7..fa821d9 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -743,20 +743,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_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py index dc74752..88b6f63 100644 --- a/packages/meshbay-node/tests/test_webrtc_transport.py +++ b/packages/meshbay-node/tests/test_webrtc_transport.py @@ -743,36 +743,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) |