fix(node): bring the QUIC chat and stream handlers to WebRTC parity

The unified handshake reached QUIC in Phase 11.5, but the chat and
stream handlers did not get the authorization rules the WebRTC path
gained at the same time:

- _do_chat_message_sync took `sender_id` from the wire, so an
  authenticated peer could post as anyone (NS6 / M2a). It is now the
  authenticated session's id, always.
- chat used a connection-global peer registry and read chat_store from
  the top-level context, so on a multi-group node a message from one
  group fanned out to peers of the others (M2b / H1). Both are now
  resolved per group via _peer_registry() / _group_ctx(), mirroring
  the WebRTC path. The QUIC peer set is kept separate from the WebRTC
  one in the same group context — the two session types have
  different _send signatures and no cross-transport fan-out is wired.
- _do_stream_segment_sync ran `subprocess.run(timeout=30)` on the
  event loop with no concurrency cap, so one request stalled the whole
  node and any member could fork-bomb it with ffmpeg (M2c). Extraction
  now runs in a thread behind a small semaphore, spawned as a tracked
  task (cancelled on connection_lost).

Also corrects the stale docstring claiming C6 is still open here — the
GEK proof has been enforced on this transport since 11.5.

No behaviour change for shipping clients: none speak QUIC, and the
listener is off by default (previous commit).

Third security review, finding M2.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011pG75yGK3NthNfyjH74omG

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"