feat(node): add QUIC transport (MNP v2) — quic_server + quic_client

QuicChunkServer/QuicChunkClient: same MNP protocol over QUIC/UDP.
Enables hole-punching (Spike 4 Cone NAT validated). Uses aioquic 1.3.0.

Bug found+fixed: asyncio.Event race condition in client recv loop
(quic_event_received overwrote _stream_events[0] after _recv
created it). Fixed with asyncio.Queue (no shared mutable state).

Server uses synchronous handlers in quic_event_received (avoids
ensure_future transmit timing issue). 3/3 tests. Full suite: 50/50.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

3 files changed, 482 insertions, 2 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/__init__.py b/packages/meshbay-node/src/meshbay_node/transport/__init__.py
index ddd11f5..8b49b22 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/__init__.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/__init__.py
@@ -1,6 +1,9 @@
-"""TCP+TLS transport layer (MNP v1). QUIC added in v2."""
+"""MeshBay Node transport layer — TCP+TLS (MNP v1) and QUIC (MNP v2)."""
 from .server import ChunkServer
 from .client import ChunkClient
 from .http_server import create_http_app
+from .quic_server import QuicChunkServer
+from .quic_client import QuicChunkClient
 
-__all__ = ["ChunkServer", "ChunkClient", "create_http_app"]
+__all__ = ["ChunkServer", "ChunkClient", "create_http_app",
+           "QuicChunkServer", "QuicChunkClient"]
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
new file mode 100644
index 0000000..6041fef
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
@@ -0,0 +1,197 @@
+"""
+MeshBay Node — QUIC chunk client (MNP v2).
+
+Drop-in replacement for ChunkClient with QUIC/UDP transport.
+Same application-layer protocol: MNP handshake → stream requests.
+
+Node identity verified via Ed25519 PK from hub (not TLS cert chain).
+TLS cert is self-signed; we use CERT_NONE equivalent in QUIC config.
+"""
+
+import asyncio
+import base64
+import logging
+import struct
+from pathlib import Path
+
+import blake3
+import jwt
+import msgpack
+from aioquic.asyncio import connect, QuicConnectionProtocol
+from aioquic.quic.configuration import QuicConfiguration
+from aioquic.quic.events import QuicEvent, StreamDataReceived
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+
+from meshbay_common import MNP_VERSION
+from meshbay_common.crypto import (
+    chunk_key as derive_chunk_key,
+    decrypt_chunk,
+    verify_chunk_signature,
+)
+from meshbay_common.protocol import MNP
+
+log = logging.getLogger(__name__)
+
+ALPN    = ["meshbay-mnp"]
+MAX_MSG = 64 * 1024 * 1024
+
+
+# ── Client-side protocol ──────────────────────────────────────────────────────
+
+class _MNPClientProtocol(QuicConnectionProtocol):
+    """
+    Client-side QUIC protocol.
+    Uses per-stream asyncio.Queue for received messages — no race conditions.
+    """
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._bufs:   dict[int, bytearray]    = {}
+        self._queues: dict[int, asyncio.Queue] = {}
+
+    def _queue_for(self, stream_id: int) -> asyncio.Queue:
+        if stream_id not in self._queues:
+            self._queues[stream_id] = asyncio.Queue()
+            self._bufs[stream_id]   = bytearray()
+        return self._queues[stream_id]
+
+    def quic_event_received(self, event: QuicEvent) -> None:
+        if isinstance(event, StreamDataReceived):
+            sid = event.stream_id
+            q   = self._queue_for(sid)
+            self._bufs[sid].extend(event.data)
+            buf = self._bufs[sid]
+            while len(buf) >= 4:
+                length = struct.unpack(">I", buf[:4])[0]
+                if length > MAX_MSG:
+                    raise ValueError(f"Message too large: {length}")
+                if len(buf) < 4 + length:
+                    break
+                msg_bytes = bytes(buf[4:4 + length])
+                del buf[:4 + length]
+                q.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    async def _recv(self, stream_id: int, timeout: float = 10.0) -> dict:
+        q = self._queue_for(stream_id)
+        return await asyncio.wait_for(q.get(), timeout=timeout)
+
+    def _send(self, stream_id: int, obj: dict) -> None:
+        data = msgpack.packb(obj, use_bin_type=True)
+        payload = struct.pack(">I", len(data)) + data
+        self._quic.send_stream_data(stream_id, payload)
+        self.transmit()
+
+
+# ── QuicChunkClient ───────────────────────────────────────────────────────────
+
+class QuicChunkClient:
+    """
+    QUIC-based MNP client. Drop-in replacement for ChunkClient.
+
+    Usage:
+        async with QuicChunkClient(host, port, token, gek, pk_node_b64) as client:
+            data = await client.fetch_chunk(file_id, chunk_index=0)
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int,
+        jwt_token: str,
+        gek: bytes,
+        pk_node_b64: str,
+    ):
+        self._host      = host
+        self._port      = port
+        self._jwt_token = jwt_token
+        self._gek       = gek
+        self._pk_node   = Ed25519PublicKey.from_public_bytes(
+            base64.b64decode(pk_node_b64))
+        self._proto: _MNPClientProtocol | None = None
+        self._cm        = None
+        self._ctrl_stream = 0   # stream 0 = control/handshake
+
+    async def __aenter__(self):
+        await self.connect()
+        return self
+
+    async def __aexit__(self, *_):
+        await self.close()
+
+    async def connect(self) -> None:
+        import ssl
+        config = QuicConfiguration(
+            is_client=True,
+            alpn_protocols=ALPN,
+            verify_mode=ssl.CERT_NONE,  # identity verified via Ed25519 at MNP layer
+        )
+        self._cm = connect(
+            self._host, self._port,
+            configuration=config,
+            create_protocol=_MNPClientProtocol,
+        )
+        self._proto = await self._cm.__aenter__()
+
+        # MNP handshake on stream 0
+        self._proto._send(self._ctrl_stream, {
+            "type":  MNP.HANDSHAKE,
+            "v":     MNP_VERSION,
+            "token": self._jwt_token,
+        })
+        ack = await self._proto._recv(self._ctrl_stream)
+        if ack.get("type") != MNP.HANDSHAKE_ACK:
+            raise ConnectionError(f"QUIC handshake rejected: {ack}")
+        log.debug("QUIC connected to %s:%d", self._host, self._port)
+
+    async def close(self) -> None:
+        if self._cm:
+            await self._cm.__aexit__(None, None, None)
+            self._cm = None
+            self._proto = None
+
+    def _new_stream(self) -> int:
+        """Open a new bidirectional QUIC stream for a request."""
+        stream_id = self._proto._quic.get_next_available_stream_id(is_unidirectional=False)
+        return stream_id
+
+    async def fetch_index(self) -> bytes:
+        """Request the Mesh Group Index."""
+        sid = self._new_stream()
+        self._proto._send(sid, {"type": MNP.INDEX_SYNC, "v": MNP_VERSION})
+        msg = await self._proto._recv(sid)
+        return base64.b64decode(msg["index_b64"])
+
+    async def fetch_chunk(self, file_id: str, chunk_index: int) -> bytes:
+        """Fetch, verify, and decrypt one chunk over QUIC."""
+        sid = self._new_stream()
+        self._proto._send(sid, {
+            "type":        MNP.FILE_REQUEST,
+            "v":           MNP_VERSION,
+            "file_id":     file_id,
+            "chunk_index": chunk_index,
+        })
+        msg = await self._proto._recv(sid, timeout=30.0)
+
+        if msg.get("type") == "error":
+            raise LookupError(msg.get("detail", "Unknown error"))
+
+        ct        = base64.b64decode(msg["ct_b64"])
+        nonce     = base64.b64decode(msg["nonce_b64"])
+        ct_hash   = base64.b64decode(msg["ct_hash_b64"])
+        pt_hash   = base64.b64decode(msg["pt_hash_b64"])
+        sig       = base64.b64decode(msg["sig_b64"])
+        file_hash = base64.b64decode(msg["file_hash_b64"])
+        ci        = msg["chunk_index"]
+
+        verify_chunk_signature(self._pk_node, ci, nonce, ct_hash, sig)
+
+        if blake3.blake3(ct).digest() != ct_hash:
+            raise ValueError("Ciphertext hash mismatch")
+
+        ckey      = derive_chunk_key(self._gek, file_hash, ci)
+        plaintext = decrypt_chunk(ckey, nonce, ct)
+
+        if blake3.blake3(plaintext).digest() != pt_hash:
+            raise ValueError("Plaintext hash mismatch after decryption")
+
+        return plaintext
diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
new file mode 100644
index 0000000..2d23da2
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -0,0 +1,280 @@
+"""
+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
+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 (
+    chunk_key as derive_chunk_key,
+    encrypt_chunk,
+    sign_chunk,
+    pk_to_b64,
+)
+from meshbay_common.protocol import MNP
+from meshbay_node.indexer import GroupIndex
+from meshbay_node.transport.tls_cert import server_ssl_context
+
+log = logging.getLogger(__name__)
+
+CHUNK_SIZE = 1024 * 1024
+MAX_MSG    = 64 * 1024 * 1024
+ALPN       = ["meshbay-mnp"]
+
+
+# ── 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._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)
+            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:
+        import time
+        token = msg.get("token", "")
+        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
+
+        if decoded.get("exp", 0) < int(time.time()):
+            self._send(stream_id, {"type": "error", "detail": "JWT expired"})
+            self._quic.close()
+            return
+
+        self._user_id = decoded["sub"]
+        log.info("QUIC handshake OK — user=%s", self._user_id[:8])
+        self._send(stream_id, {
+            "type":    MNP.HANDSHAKE_ACK,
+            "v":       MNP_VERSION,
+            "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()),
+        })
+
+    def _do_index_sync_sync(self, stream_id: int) -> None:
+        wire = self._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)."""
+        file_id     = msg["file_id"]
+        chunk_index = msg["chunk_index"]
+        entry = self._ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send(stream_id, {"type": "error", "detail": "File not found"})
+            return
+
+        file_path = self._ctx["shared_root"] / entry.path / entry.name
+        if not file_path.exists():
+            self._send(stream_id, {"type": "error", "detail": "File not on disk"})
+            return
+
+        chunk_data = _read_and_encrypt(
+            self._ctx["sk_node"],
+            self._ctx["gek"],
+            file_path,
+            chunk_index,
+        )
+        self._send(stream_id, chunk_data)
+
+    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,
+) -> 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)
+
+    file_hash = blake3.blake3(file_path.read_bytes()).digest()
+    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(),
+    }
+
+
+# ── 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 = "0.0.0.0",
+        port: int = 19000,
+        cert_path: Path | None = None,
+        key_path: Path | None = None,
+    ):
+        self._ctx = {
+            "sk_node":     sk_node,
+            "hub_pk_pem":  hub_pk_pem,
+            "gek":         gek,
+            "shared_root": shared_root,
+            "index":       index,
+        }
+        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
+
+    @property
+    def port(self) -> int:
+        return self._port
+
+    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
+
+    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,
+        )
+        log.info("QuicChunkServer listening on %s:%d (QUIC/UDP)", self._host, self._port)
+
+    async def stop(self) -> None:
+        if self._server:
+            self._server.close()
+            self._server = None
+            log.info("QuicChunkServer stopped")