aboutsummaryrefslogtreecommitdiffstats
path: root/packages
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-09 05:13:51 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-09 05:13:51 +0200
commit88cfc139333ac3fe5789f39f3970065181df9043 (patch)
tree17ac132761100e95cf70281761c312a1607a4149 /packages
parent3cad68ad33e3b5a481adaab569f65f6a9d0c8719 (diff)
downloadmeshbay-88cfc139333ac3fe5789f39f3970065181df9043.tar.gz
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>
Diffstat (limited to 'packages')
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/__init__.py7
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_client.py197
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/quic_server.py280
-rw-r--r--packages/meshbay-node/tests/test_quic_transport.py157
4 files changed, 639 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")
diff --git a/packages/meshbay-node/tests/test_quic_transport.py b/packages/meshbay-node/tests/test_quic_transport.py
new file mode 100644
index 0000000..2abd465
--- /dev/null
+++ b/packages/meshbay-node/tests/test_quic_transport.py
@@ -0,0 +1,157 @@
+"""
+Integration test: QuicChunkServer ↔ QuicChunkClient over QUIC/UDP loopback.
+Same structure as test_transport.py but uses QUIC instead of TCP+TLS.
+"""
+
+import asyncio
+import os
+import time
+import jwt
+import pytest
+from pathlib import Path
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives import serialization
+
+from meshbay_common.crypto import generate_gek, pk_to_b64
+from meshbay_node.indexer import DirectoryIndexer, GroupIndex
+from meshbay_node.transport.quic_server import QuicChunkServer
+from meshbay_node.transport.quic_client import QuicChunkClient
+
+
+@pytest.fixture
+def sk_node():
+ return Ed25519PrivateKey.generate()
+
+@pytest.fixture
+def sk_hub():
+ return Ed25519PrivateKey.generate()
+
+@pytest.fixture
+def gek():
+ return generate_gek()
+
+@pytest.fixture
+def shared_dir(tmp_path):
+ d = tmp_path / "shared"
+ d.mkdir()
+ (d / "test.mp4").write_bytes(os.urandom(2 * 1024 * 1024)) # 2 MB
+ (d / "small.txt").write_bytes(b"hello quic " * 100)
+ return d
+
+def make_jwt(sk_hub, pk_node_b64, ttl=3600):
+ sk_pem = sk_hub.private_bytes(
+ serialization.Encoding.PEM,
+ serialization.PrivateFormat.PKCS8,
+ serialization.NoEncryption(),
+ )
+ now = int(time.time())
+ return jwt.encode({
+ "iss": "test-hub", "sub": "user-001",
+ "pk_user": pk_node_b64, "hub_id": "test-hub",
+ "jti": "test-jti", "iat": now, "exp": now + ttl,
+ }, sk_pem, algorithm="EdDSA")
+
+
+@pytest.mark.asyncio
+async def test_quic_chunk_roundtrip(sk_node, sk_hub, gek, shared_dir, tmp_path):
+ """Full QUIC roundtrip: server serves chunk, client verifies and decrypts."""
+ hub_pk_pem = sk_hub.public_key().public_bytes(
+ serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+ indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+ await indexer.initial_scan()
+
+ cert_path = tmp_path / "node.crt"
+ key_path = tmp_path / "node.key"
+
+ server = QuicChunkServer(
+ sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+ shared_root=shared_dir, index=indexer.index,
+ host="127.0.0.1", port=19100,
+ cert_path=cert_path, key_path=key_path,
+ )
+ await server.start()
+
+ token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()))
+ entry = next(e for e in indexer.index.entries if e.name == "test.mp4")
+
+ async with QuicChunkClient(
+ host="127.0.0.1", port=19100,
+ jwt_token=token, gek=gek,
+ pk_node_b64=pk_to_b64(sk_node.public_key()),
+ ) as client:
+ chunk0 = await client.fetch_chunk(entry.id, chunk_index=0)
+ chunk1 = await client.fetch_chunk(entry.id, chunk_index=1)
+
+ original = (shared_dir / "test.mp4").read_bytes()
+ assert chunk0 + chunk1 == original
+
+ await server.stop()
+
+
+@pytest.mark.asyncio
+async def test_quic_fetch_index(sk_node, sk_hub, gek, shared_dir, tmp_path):
+ """QUIC index sync returns deserializable GroupIndex."""
+ hub_pk_pem = sk_hub.public_key().public_bytes(
+ serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+ indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+ await indexer.initial_scan()
+
+ cert_path = tmp_path / "node.crt"
+ key_path = tmp_path / "node.key"
+
+ server = QuicChunkServer(
+ sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+ shared_root=shared_dir, index=indexer.index,
+ host="127.0.0.1", port=19101,
+ cert_path=cert_path, key_path=key_path,
+ )
+ await server.start()
+
+ token = make_jwt(sk_hub, pk_to_b64(sk_node.public_key()))
+
+ async with QuicChunkClient(
+ host="127.0.0.1", port=19101,
+ jwt_token=token, gek=gek,
+ pk_node_b64=pk_to_b64(sk_node.public_key()),
+ ) as client:
+ wire = await client.fetch_index()
+ recovered = GroupIndex.deserialize(wire, sk_node=sk_node, gek=gek)
+ assert recovered.count == 2
+
+ await server.stop()
+
+
+@pytest.mark.asyncio
+async def test_quic_invalid_jwt_rejected(sk_node, sk_hub, gek, shared_dir, tmp_path):
+ """QUIC server rejects connections with tokens signed by wrong hub key."""
+ hub_pk_pem = sk_hub.public_key().public_bytes(
+ serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
+
+ indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+ await indexer.initial_scan()
+
+ cert_path = tmp_path / "node.crt"
+ key_path = tmp_path / "node.key"
+
+ server = QuicChunkServer(
+ sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+ shared_root=shared_dir, index=indexer.index,
+ host="127.0.0.1", port=19102,
+ cert_path=cert_path, key_path=key_path,
+ )
+ await server.start()
+
+ sk_other = Ed25519PrivateKey.generate()
+ bad_token = make_jwt(sk_other, pk_to_b64(sk_node.public_key()))
+
+ with pytest.raises(Exception):
+ async with QuicChunkClient(
+ host="127.0.0.1", port=19102,
+ jwt_token=bad_token, gek=gek,
+ pk_node_b64=pk_to_b64(sk_node.public_key()),
+ ) as client:
+ await client.fetch_index()
+
+ await server.stop()