""" 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 struct import subprocess from pathlib import Path from typing import Any, Callable import blake3 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_common.crypto import ( sign_chunk, pk_to_b64, ) from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, encrypt_chunk_aes as encrypt_chunk from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex log = logging.getLogger(__name__) CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 ALPN = ["meshbay-mnp"] 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 _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] = {} 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 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._do_stream_segment_sync(stream_id, msg) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message_sync(stream_id, msg) 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: token = msg.get("token", "") group_id = msg.get("group_id", "") try: decoded = jwt.decode(token, self._ctx["hub_pk_pem"], algorithms=["EdDSA"]) except Exception as e: self._send(stream_id, {"type": "error", "detail": f"Invalid JWT: {e}"}) self._quic.close() return denylist = self._ctx.get("denylist") if denylist and denylist.is_denied( decoded.get("sub", ""), decoded.get("jti", ""), group_id): self._send(stream_id, {"type": "error", "detail": "Token revoked"}) self._quic.close() return if group_id and group_id not in decoded.get("groups", []): self._send(stream_id, {"type": "error", "detail": "Not a member of this group"}) self._quic.close() return if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]: self._send(stream_id, {"type": "error", "detail": "Group not hosted on this node"}) self._quic.close() return self._user_id = decoded["sub"] self._group_id = group_id peers = self._ctx.get("_peers") if peers is not None: peers[self._user_id] = self log.info("QUIC handshake OK — user=%s group=%s", self._user_id[:8], group_id[:8] if group_id else "none") self._send(stream_id, { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), }) 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 _do_index_sync_sync(self, stream_id: int) -> None: ctx = self._group_ctx() wire = ctx["index"].serialize() self._send(stream_id, { "type": MNP.INDEX_SYNC, "v": MNP_VERSION, "index_b64": base64.b64encode(wire).decode(), }) 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 = ctx["shared_root"] / entry.path / entry.name 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( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, file_hash, ) 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) entry = ctx["index"].get_entry(file_id) if not entry: self._send(stream_id, {"type": "error", "detail": "File not found"}) return file_path = ctx["shared_root"] / entry.path / entry.name 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 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), }) 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") if chat_store: import asyncio asyncio.ensure_future(chat_store.save_message( sender_id=msg.get("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""), 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), "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(): 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: peers = self._ctx.get("_peers") if peers and self._user_id: peers.pop(self._user_id, None) 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( sk_node: Ed25519PrivateKey, gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, ) -> dict: """Read and encrypt one chunk (blocking — runs in executor).""" with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) pt_hash = blake3.blake3(plaintext).digest() ckey = derive_chunk_key(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk(ckey, plaintext) ct_hash = blake3.blake3(ct).digest() sig = sign_chunk(sk_node, chunk_index, nonce, ct_hash) return { "type": MNP.FILE_CHUNK, "v": MNP_VERSION, "chunk_index": chunk_index, "plaintext_size": len(plaintext), "nonce_b64": base64.b64encode(nonce).decode(), "ct_b64": base64.b64encode(ct).decode(), "ct_hash_b64": base64.b64encode(ct_hash).decode(), "pt_hash_b64": base64.b64encode(pt_hash).decode(), "sig_b64": base64.b64encode(sig).decode(), "pk_node_b64": pk_to_b64(sk_node.public_key()), "file_hash_b64": base64.b64encode(file_hash).decode(), } 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( ["ffmpeg", "-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, shared_root: Path, 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, "shared_root": shared_root, "index": index, } if groups: self._ctx["groups"] = groups self._denylist = denylist or Denylist() self._ctx["denylist"] = self._denylist self._ctx["_peers"] = {} self._host = host self._port = port self._cert_path = cert_path or Path.home() / ".config/meshbay/node_tls.crt" self._key_path = key_path or Path.home() / ".config/meshbay/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)) 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")