diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/quic_server.py | 144 |
1 files changed, 95 insertions, 49 deletions
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 ce6fe17..30c7daf 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -61,6 +61,15 @@ 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: """ @@ -204,6 +213,9 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._nonce_client: bytes = b"" self._gek_challenge: bytes | None = None self._pending = None + # asyncio holds only a weak reference to a bare task, so a spawned + # handler still running can be collected mid-flight. Hold them. + self._tasks: set[asyncio.Task] = set() def quic_event_received(self, event: QuicEvent) -> None: if isinstance(event, StreamDataReceived): @@ -232,7 +244,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): elif mtype == MNP.FILE_REQUEST: self._do_file_request_sync(stream_id, msg) elif mtype == MNP.STREAM_SEGMENT: - self._do_stream_segment_sync(stream_id, msg) + 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: @@ -256,11 +268,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): checks could drift from the WebRTC path independently. All of that now comes from meshbay_common.handshake, shared with WebRTC. - NOT YET DONE — finding C6 remains open on this transport: there is still no - GEK proof here, so a forged or stolen token reaches the node and can inject - chat without holding the group key. The challenge/response and mutual node - proof (quic_binding() is written and unit-tested for exactly this) are the - remaining work in 11.5.4/5/6. + The GEK proof is enforced here too: `_do_handshake_response_sync` runs + the same challenge/response and mutual node proof, from the same shared + module, bound to the QUIC certificate hash. Finding C6 is closed on this + transport. """ try: peer = authorize_token( @@ -339,9 +350,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._user_id = peer.user_id self._group_id = peer.group_id - peers = self._ctx.get("_peers") - if peers is not None: - peers[self._user_id] = self + self._peer_registry()[self._user_id] = self transcript = handshake_transcript( ROLE_NODE, peer.group_id, self._nonce_client, self._gek_challenge, binding) @@ -365,6 +374,20 @@ class _MNPServerProtocol(QuicConnectionProtocol): return self._ctx["groups"][self._group_id] return self._ctx + def _spawn(self, coro) -> None: + task = asyncio.ensure_future(coro) + self._tasks.add(task) + task.add_done_callback(self._tasks.discard) + + def _peer_registry(self) -> dict: + """QUIC peers for THIS connection's group, keyed per group so a message + never crosses into another group on a multi-group node (findings M2b / + H1). Deliberately separate from the WebRTC registry that also lives in + the group context: the two transports' session objects have different + `_send` signatures, and cross-transport chat fan-out is not wired (no + QUIC client ships yet).""" + return self._group_ctx().setdefault("_quic_peers", {}) + def _do_index_sync_sync(self, stream_id: int) -> None: ctx = self._group_ctx() wire = ctx["index"].serialize() @@ -399,61 +422,83 @@ class _MNPServerProtocol(QuicConnectionProtocol): ) self._send(stream_id, chunk_data) - def _do_stream_segment_sync(self, stream_id: int, msg: dict) -> None: - """Extract and serve one HLS segment via ffmpeg.""" - ctx = self._group_ctx() - file_id = msg["file_id"] - segment_index = msg["segment_index"] - segment_duration = msg.get("segment_duration", 4) + 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 + 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 + 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 - segment_data = _extract_segment(file_path, start_time, segment_duration) - if segment_data is None: - self._send(stream_id, {"type": "error", "detail": "Segment extraction failed"}) - 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), - }) + 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: - """Receive a chat message, store it, and broadcast to other connected peers.""" - chat_store = self._ctx.get("chat_store") + """ + Store a chat message and broadcast it to the rest of THIS group. + + `sender_id` is the authenticated session's, never the wire's — a peer + must not be able to post as someone else (NS6 / finding M2a). The store + and the peer set come from the group context, not a connection-global + one, so a message never crosses into another group on a multi-group node + (findings M2b / H1). The WebRTC path has done both since Phase 11.5. + """ + gctx = self._group_ctx() + payload = msg.get("payload", b"") + if isinstance(payload, str): + payload = payload.encode() + + chat_store = gctx.get("chat_store") if chat_store: - import asyncio - asyncio.ensure_future(chat_store.save_message( - sender_id=msg.get("sender_id", self._user_id), + self._spawn(chat_store.save_message( + sender_id=self._user_id, iteration=msg.get("iteration", 0), - payload=msg.get("payload", b"").encode() if isinstance(msg.get("payload"), str) else msg.get("payload", b""), + payload=payload, thread_id=msg.get("thread_id"), )) - peers = self._ctx.get("_peers", {}) broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, - "sender_id": msg.get("sender_id", self._user_id), + "sender_id": self._user_id, "iteration": msg.get("iteration", 0), "payload": msg.get("payload", ""), "thread_id": msg.get("thread_id"), "group_id": self._group_id or "", } - for uid, proto in peers.items(): + for uid, proto in list(self._peer_registry().items()): if uid != self._user_id and proto is not self: try: proto._send(0, broadcast) @@ -463,9 +508,10 @@ class _MNPServerProtocol(QuicConnectionProtocol): self._send(stream_id, {"type": "ack", "v": MNP_VERSION}) def connection_lost(self, exc) -> None: - peers = self._ctx.get("_peers") - if peers and self._user_id: - peers.pop(self._user_id, None) + if self._user_id: + self._peer_registry().pop(self._user_id, None) + for task in list(self._tasks): + task.cancel() super().connection_lost(exc) def _send(self, stream_id: int, obj: dict) -> None: @@ -558,7 +604,7 @@ class QuicChunkServer: self._ctx["groups"] = groups self._denylist = denylist or Denylist() self._ctx["denylist"] = self._denylist - self._ctx["_peers"] = {} + # Peer sets are per group now — see _MNPServerProtocol._peer_registry(). self._host = host self._port = port self._cert_path = cert_path or Path.home() / ".config/meshbay/node_tls.crt" |