""" MeshBay Node — QUIC chunk server (MNP v2). Replaces the TCP+TLS ChunkServer with QUIC transport. Advantages over TCP: - UDP-based → works with hole punching (Spike 4 confirmed Cone NAT on SFR) - Multiplexed streams — each request is an independent QUIC stream - 0-RTT reconnection (connection resumption) - Built-in TLS 1.3 Wire protocol: - Each bidirectional QUIC stream carries one request/response exchange - Messages: length-prefixed msgpack (4-byte big-endian, same as TCP+TLS) - MNP handshake on stream 0 (control stream); subsequent streams = requests Application protocol (MNP) is identical to TCP+TLS version. The transport is the only change — all crypto, auth, and message types stay the same. """ import asyncio import base64 import logging import os import struct import subprocess from pathlib import Path from typing import Any, Callable import jwt import msgpack from aioquic.asyncio import QuicConnectionProtocol, serve from aioquic.quic.configuration import QuicConfiguration from aioquic.quic.events import QuicEvent, StreamDataReceived, StreamReset from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from meshbay_common import MNP_VERSION from meshbay_node.roots import RootSet, entry_abs_path from meshbay_common.handshake import ( MNP_MIN_SUPPORTED, NONCE_LEN, ROLE_CLIENT, ROLE_NODE, HandshakeError, authorize_token, check_version, handshake_transcript, make_proof, quic_binding, verify_proof, ) from meshbay_common.crypto import pk_to_b64 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__) 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. Finding H4: revocations used to live only in memory, so a node restart silently un-revoked everyone, and group revocations were dropped entirely — the hub signed and broadcast them but the node's handler only understood "user" and "jti". Now persisted to disk and group targets are honoured. """ def __init__(self, path: Path | None = None): self.user_ids: set[str] = set() self.group_ids: set[str] = set() self.jtis: set[str] = set() self._path = path self._load() def is_denied(self, user_id: str, jti: str, group_id: str = "") -> bool: return (user_id in self.user_ids or jti in self.jtis or (bool(group_id) and group_id in self.group_ids)) def deny_user(self, user_id: str) -> None: self.user_ids.add(user_id) log.info("Denied user: %s", user_id[:8]) self._save() def deny_group(self, group_id: str) -> None: self.group_ids.add(group_id) log.info("Denied group: %s", group_id[:8]) self._save() def deny_jti(self, jti: str) -> None: self.jtis.add(jti) log.info("Denied jti: %s", jti[:8]) self._save() def entries(self) -> dict[str, list[str]]: """What is currently refused, for the operator to inspect (14.10).""" return { "users": sorted(self.user_ids), "groups": sorted(self.group_ids), "jtis": sorted(self.jtis), } def clear(self, subject: str = "") -> int: """ Drop everything, or one identifier. Returns how many entries went. Not silent by design: clearing re-admits whoever it was keeping out, and the count is what tells the operator whether they undid one revocation or all of them. """ before = len(self.user_ids) + len(self.group_ids) + len(self.jtis) if subject: self.user_ids.discard(subject) self.group_ids.discard(subject) self.jtis.discard(subject) else: self.user_ids.clear() self.group_ids.clear() self.jtis.clear() after = len(self.user_ids) + len(self.group_ids) + len(self.jtis) removed = before - after if removed: self._save() return removed def _load(self) -> None: if not self._path or not self._path.exists(): return try: import json data = json.loads(self._path.read_text()) self.user_ids = set(data.get("users", [])) self.group_ids = set(data.get("groups", [])) self.jtis = set(data.get("jtis", [])) log.info("Denylist loaded: %d users, %d groups, %d jtis", len(self.user_ids), len(self.group_ids), len(self.jtis)) except Exception as e: log.warning("Could not load denylist from %s: %s", self._path, e) def _save(self) -> None: if not self._path: return try: import json self._path.parent.mkdir(parents=True, exist_ok=True) self._path.write_text(json.dumps({ "users": sorted(self.user_ids), "groups": sorted(self.group_ids), "jtis": sorted(self.jtis), })) except Exception as e: log.warning("Could not persist denylist to %s: %s", self._path, e) # ── Wire helpers ────────────────────────────────────────────────────────────── def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) return struct.pack(">I", len(data)) + data class _StreamBuffer: """Accumulate incoming QUIC stream data and extract length-prefixed messages.""" def __init__(self): self._buf = bytearray() def feed(self, data: bytes): self._buf.extend(data) def messages(self): while len(self._buf) >= 4: length = struct.unpack(">I", self._buf[:4])[0] if length > MAX_MSG: raise ValueError(f"Message too large: {length}") if len(self._buf) < 4 + length: break msg_bytes = bytes(self._buf[4:4 + length]) del self._buf[:4 + length] yield msgpack.unpackb(msg_bytes, raw=False) # ── Per-connection server protocol ──────────────────────────────────────────── class _MNPServerProtocol(QuicConnectionProtocol): """ One instance per QUIC connection. Handles the MNP handshake and all subsequent streams. """ def __init__(self, *args, node_ctx: dict, **kwargs): super().__init__(*args, **kwargs) self._ctx = node_ctx # shared server context (keys, index, etc.) self._user_id: str | None = None self._group_id: str | None = None self._buffers: dict[int, _StreamBuffer] = {} 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): sid = event.stream_id if sid not in self._buffers: self._buffers[sid] = _StreamBuffer() self._buffers[sid].feed(event.data) for msg in self._buffers[sid].messages(): self._handle_message_sync(sid, msg) elif isinstance(event, StreamReset): self._buffers.pop(event.stream_id, None) def _handle_message_sync(self, stream_id: int, msg: dict) -> None: """Handle an MNP message synchronously (called from quic_event_received).""" mtype = msg.get("type") try: if mtype == MNP.HANDSHAKE: self._do_handshake_sync(stream_id, msg) elif mtype == MNP.HANDSHAKE_RESPONSE: self._do_handshake_response_sync(stream_id, msg) elif self._user_id is None: self._send(stream_id, {"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: 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: # Liveness is transport-agnostic, and a native client over QUIC # has the same half-open problem a DataChannel does. self._send(stream_id, {"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")}) else: log.warning("Unknown MNP message type: %s", mtype) except Exception as e: log.error("Error handling %s: %s", mtype, e) self._send(stream_id, {"type": "error", "detail": str(e)}) def _do_handshake_sync(self, stream_id: int, msg: dict) -> None: """ Authorization half of the unified handshake (11.5.4). This used to be a second, weaker copy of the WebRTC logic: group_id was optional (so omitting it skipped the membership check entirely — M1), node-scoped daemon tokens were accepted as client tokens (M9), and the checks could drift from the WebRTC path independently. All of that now comes from meshbay_common.handshake, shared with WebRTC. 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. """ # Same order as WebRTC: the range first, so a peer we cannot speak to is # told so, rather than served messages it will misread (L2). try: check_version(msg.get("v", ""), msg.get("v_min", "")) except HandshakeError as refusal: self._send(stream_id, {"type": "error", "detail": str(refusal), "code": refusal.code}) self._quic.close() return try: peer = authorize_token( msg.get("token", ""), self._ctx["hub_pk_pem"], group_id=msg.get("group_id", ""), hosted_groups=self._ctx.get("groups"), denylist=self._ctx.get("denylist"), ) except HandshakeError as refusal: self._send(stream_id, {"type": "error", "detail": str(refusal), "code": refusal.code}) self._quic.close() return try: self._nonce_client = base64.b64decode(msg.get("nonce", "")) except Exception: self._nonce_client = b"" if len(self._nonce_client) < NONCE_LEN: self._send(stream_id, {"type": "error", "detail": "Client nonce required"}) self._quic.close() return gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx if not gctx.get("gek"): self._send(stream_id, { "type": "error", "detail": "Group encryption not initialized — contact node operator", }) self._quic.close() return # Decoded but NOT authenticated: authentication is the GEK proof below. self._pending = peer self._gek_challenge = os.urandom(NONCE_LEN) self._send(stream_id, { "type": MNP.HANDSHAKE_CHALLENGE, "v": MNP_VERSION, "v_min": MNP_MIN_SUPPORTED, "nonce": base64.b64encode(self._gek_challenge).decode(), }) def _do_handshake_response_sync(self, stream_id: int, msg: dict) -> None: """Verify the client's GEK proof, then prove the node in return (C6, C3).""" if not self._gek_challenge or self._pending is None: self._send(stream_id, {"type": "error", "detail": "No pending handshake challenge"}) return peer = self._pending gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx gek = gctx.get("gek") if not gek: self._send(stream_id, {"type": "error", "detail": "Group encryption not initialized"}) self._quic.close() return binding = self._ctx.get("server_cert_der") if not binding: # Refuse rather than fall back to an unbound proof (L4). self._send(stream_id, {"type": "error", "detail": "Channel binding unavailable"}) self._quic.close() return binding = quic_binding(binding) try: proof = base64.b64decode(msg.get("proof", "")) except Exception: self._send(stream_id, {"type": "error", "detail": "Invalid proof encoding"}) return if not verify_proof(gek, proof, ROLE_CLIENT, peer.group_id, self._nonce_client, self._gek_challenge, binding): self._send(stream_id, {"type": "error", "detail": "GEK proof failed"}) self._quic.close() return self._user_id = peer.user_id self._group_id = peer.group_id self._peer_registry()[self._user_id] = self transcript = handshake_transcript( ROLE_NODE, peer.group_id, self._nonce_client, self._gek_challenge, binding) node_proof = make_proof( gek, ROLE_NODE, peer.group_id, self._nonce_client, self._gek_challenge, binding) log.info("QUIC handshake OK — user=%s group=%s", self._user_id[:8], self._group_id[:8]) # The config payload is empty here — QUIC serves no browser, so none of # the fields WebRTC carries has a consumer on this transport. It is sealed # anyway (decision D5): one shape per message on every transport, which is # the lesson of the two `file_chunk` encoders and the two `index_sync` # encodings. A field added later then has somewhere to go that is already # authenticated, instead of arriving in clear beside a sealed one. self._send(stream_id, { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), "proof": base64.b64encode(node_proof).decode(), "sig": base64.b64encode(self._ctx["sk_node"].sign(transcript)).decode(), **seal(gek, PURPOSE_ACK, MNP.HANDSHAKE_ACK, peer.group_id, {}), }) self._gek_challenge = None def _group_ctx(self) -> dict: """Resolve the active group context (multi-group or legacy single-group).""" if "groups" in self._ctx and self._group_id: 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() self._send(stream_id, index_sync_message(ctx["index"], ctx.get("roots"))) def _do_file_request_sync(self, stream_id: int, msg: dict) -> None: """Serve file chunk synchronously (blocking I/O — acceptable for test sizes).""" ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] 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_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( 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. `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: self._spawn(chat_store.save_message( sender_id=self._user_id, iteration=msg.get("iteration", 0), payload=payload, thread_id=msg.get("thread_id"), )) broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "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 list(self._peer_registry().items()): if uid != self._user_id and proto is not self: try: proto._send(0, broadcast) except Exception: pass self._send(stream_id, {"type": "ack", "v": MNP_VERSION}) def connection_lost(self, exc) -> 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: self._quic.send_stream_data(stream_id, _pack(obj)) self.transmit() def _read_and_encrypt( gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: """Read one chunk off disk and encrypt it, in the one shape every transport uses.""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) 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: """ QUIC-based MNP chunk server (MNP v2). Drop-in replacement for ChunkServer with UDP transport. """ def __init__( self, sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, roots: RootSet, index: GroupIndex, host: str = "::", # listen IPv4 + IPv6 (dual-stack Linux) port: int = 19000, cert_path: Path | None = None, key_path: Path | None = None, groups: dict[str, dict] | None = None, denylist: Denylist | None = None, ): self._ctx = { "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, "roots": roots, "index": index, } if groups: self._ctx["groups"] = groups self._denylist = denylist or Denylist() self._ctx["denylist"] = self._denylist # Peer sets are per group now — see _MNPServerProtocol._peer_registry(). self._host = host self._port = port from meshbay_node.platform import config_dir self._cert_path = cert_path or config_dir() / "node_tls.crt" self._key_path = key_path or config_dir() / "node_tls.key" self._server = None self._task = None self._session_tickets: dict[bytes, Any] = {} @property def port(self) -> int: return self._port @property def denylist(self) -> Denylist: return self._denylist def _make_config(self) -> QuicConfiguration: from meshbay_node.transport.tls_cert import generate_self_signed_cert if not self._cert_path.exists(): generate_self_signed_cert(self._cert_path, self._key_path) config = QuicConfiguration(is_client=False, alpn_protocols=ALPN) config.load_cert_chain(str(self._cert_path), str(self._key_path)) # Channel-binding anchor for the handshake proof (11.5.6). Read from our own # cert file — no aioquic internals needed on this side. from cryptography import x509 from cryptography.hazmat.primitives import serialization as _ser self._ctx["server_cert_der"] = x509.load_pem_x509_certificate( self._cert_path.read_bytes()).public_bytes(_ser.Encoding.DER) return config def _store_ticket(self, ticket: Any) -> None: self._session_tickets[ticket.ticket] = ticket def _fetch_ticket(self, label: bytes) -> Any | None: return self._session_tickets.pop(label, None) async def start(self) -> None: config = self._make_config() ctx = self._ctx def protocol_factory(*args, **kwargs): return _MNPServerProtocol(*args, node_ctx=ctx, **kwargs) self._server = await serve( self._host, self._port, configuration=config, create_protocol=protocol_factory, session_ticket_handler=self._store_ticket, session_ticket_fetcher=self._fetch_ticket, ) log.info("QuicChunkServer listening on %s:%d (QUIC/UDP)", self._host, self._port) def punch_nat(self, peer_ip: str, peer_port: int) -> None: """ Send a probe UDP packet FROM the QUIC server's own socket. Critical for Port-Restricted Cone NAT (SFR résidentiel) : the NAT only allows inbound from (peer_ip, peer_port) if the server previously sent a packet TO (peer_ip, peer_port) from this same socket. The QUIC client must connect FROM peer_port for the NAT entry to match. """ if self._server and hasattr(self._server, '_transport') and self._server._transport: self._server._transport.sendto(b'MESHBAY:NAT:PUNCH', (peer_ip, peer_port)) log.info("NAT probe sent → %s:%d", peer_ip, peer_port) else: log.warning("punch_nat: server transport not available") async def stop(self) -> None: if self._server: self._server.close() self._server = None log.info("QuicChunkServer stopped")