feat(quic): GEK proof and mutual authentication — closes C6

Phase 11.5.4/5/6 — finding C6, the last open critical finding.

QUIC ran a JWT-only handshake: a forged or stolen token reached the node and
could inject chat without ever holding the group key. It now runs the same
challenge/response as WebRTC through meshbay_common.handshake — client nonce,
role-bound length-prefixed transcript, GEK proof, and the node proving itself
with a GEK proof plus an Ed25519 signature over the transcript (C3).

11.5.6 channel binding, resolved by spike and then by two findings the spike
could not predict:

  * aioquic 1.3.0 exposes no RFC 5705 exporter, and the peer certificate only
    via a private attribute. The server reads its own certificate from disk, so
    no internals are touched on that side; the client's access is guarded and
    fails loudly if an upgrade moves it.
  * A RESUMED TLS session carries no certificate — aioquic does not re-send it,
    so there is nothing live to bind to. The anchor therefore travels with the
    session ticket, which is sound because the ticket is cryptographically
    derived from the handshake where that certificate was presented.
  * The anchor had to travel with the ticket rather than live on the client
    object: resumption constructs a fresh client, so an instance-level cache
    was silently useless. Caught by the resumption test, not by inspection.

Both paths refuse rather than degrade. No certificate and no cached anchor
means the handshake fails; it never falls back to an unbound proof, which would
silently drop MitM detection (L4).

QuicChunkClient gains a peer_cert_der property and constructor argument,
mirroring how session_ticket is already carried by the caller.

Tests: 9 quic/multi-group, full node+common suite green.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

2 files changed, 182 insertions, 8 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
index 288465f..9102085 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_client.py
@@ -11,6 +11,7 @@ TLS cert is self-signed; we use CERT_NONE equivalent in QUIC config.
 import asyncio
 import base64
 import logging
+import os
 import struct
 from pathlib import Path
 
@@ -26,6 +27,32 @@ from meshbay_common import MNP_VERSION
 from meshbay_common.crypto import verify_chunk_signature
 from meshbay_common.webcrypto import chunk_key_aes as derive_chunk_key, decrypt_chunk_aes as decrypt_chunk
 from meshbay_common.protocol import MNP
+from meshbay_common.handshake import (
+    NONCE_LEN,
+    ROLE_CLIENT,
+    ROLE_NODE,
+    handshake_transcript,
+    make_proof,
+    quic_binding,
+    verify_proof,
+)
+
+
+def _peer_cert_der(proto) -> bytes | None:
+    """
+    The server certificate as seen by the client — the channel-binding anchor.
+
+    Spike 11.5.6: aioquic 1.3.0 exposes no RFC 5705 exporter, and the peer
+    certificate only through a private attribute. Returns None when it is absent;
+    callers decide, because absence is not always an error — see below.
+    """
+    from cryptography.hazmat.primitives import serialization
+
+    tls = getattr(getattr(proto, "_quic", None), "tls", None)
+    cert = getattr(tls, "_peer_certificate", None) if tls is not None else None
+    if cert is None:
+        return None
+    return cert.public_bytes(serialization.Encoding.DER)
 
 log = logging.getLogger(__name__)
 
@@ -99,6 +126,7 @@ class QuicChunkClient:
         pk_node_b64: str,
         local_port: int = 0,   # 0 = OS picks; set for hole punching (Port-Restricted)
         group_id: str = "",
+        peer_cert_der: bytes | None = None,
         session_ticket: object | None = None,
     ):
         self._host       = host
@@ -112,6 +140,9 @@ class QuicChunkClient:
         self._proto: _MNPClientProtocol | None = None
         self._cm         = None
         self._ctrl_stream = 0
+        # 11.5.6 binding anchor. Travels with the session ticket: on a resumed
+        # TLS session the server does not re-send its certificate.
+        self._peer_cert_der: bytes | None = peer_cert_der
         self._session_ticket = session_ticket
 
     async def __aenter__(self):
@@ -146,19 +177,70 @@ class QuicChunkClient:
         )
         self._proto = await self._cm.__aenter__()
 
-        handshake_msg = {
-            "type":  MNP.HANDSHAKE,
+        nonce_c = os.urandom(NONCE_LEN)
+        self._proto._send(self._ctrl_stream, {
+            "type":     MNP.HANDSHAKE,
+            "v":        MNP_VERSION,
+            "token":    self._jwt_token,
+            "group_id": self._group_id,
+            "nonce":    base64.b64encode(nonce_c).decode(),
+        })
+
+        reply = await self._proto._recv(self._ctrl_stream)
+        if reply.get("type") != MNP.HANDSHAKE_CHALLENGE:
+            raise ConnectionError(f"QUIC handshake rejected: {reply}")
+
+        nonce_s = base64.b64decode(reply["nonce"])
+
+        # On a RESUMED TLS session the server does not re-send its certificate, so
+        # there is nothing live to bind to. The session ticket is cryptographically
+        # derived from the original handshake, so binding to the certificate seen
+        # then is sound — but only if we actually saw one. We never fall back to an
+        # unbound proof: that would silently drop MitM detection (L4).
+        cert_der = _peer_cert_der(self._proto)
+        if cert_der is not None:
+            self._peer_cert_der = cert_der
+        elif getattr(self, "_peer_cert_der", None) is None:
+            raise ConnectionError(
+                "QUIC peer certificate unavailable and none cached from a prior "
+                "session — refusing to handshake without channel binding")
+        binding = quic_binding(self._peer_cert_der)
+
+        self._proto._send(self._ctrl_stream, {
+            "type":  MNP.HANDSHAKE_RESPONSE,
             "v":     MNP_VERSION,
-            "token": self._jwt_token,
-        }
-        if self._group_id:
-            handshake_msg["group_id"] = self._group_id
-        self._proto._send(self._ctrl_stream, handshake_msg)
+            "proof": base64.b64encode(make_proof(
+                self._gek, ROLE_CLIENT, self._group_id,
+                nonce_c, nonce_s, binding)).decode(),
+        })
+
         ack = await self._proto._recv(self._ctrl_stream)
         if ack.get("type") != MNP.HANDSHAKE_ACK:
             raise ConnectionError(f"QUIC handshake rejected: {ack}")
+
+        # Authenticate the node before trusting anything it serves (C3).
+        if not verify_proof(
+            self._gek, base64.b64decode(ack.get("proof", "")), ROLE_NODE,
+            self._group_id, nonce_c, nonce_s, binding,
+        ):
+            raise ConnectionError("Node failed to prove GEK possession")
+
+        transcript = handshake_transcript(
+            ROLE_NODE, self._group_id, nonce_c, nonce_s, binding)
+        try:
+            Ed25519PublicKey.from_public_bytes(
+                base64.b64decode(ack["node_pk"])
+            ).verify(base64.b64decode(ack["sig"]), transcript)
+        except Exception as exc:
+            raise ConnectionError(f"Node signature invalid: {exc}") from exc
+
         log.debug("QUIC connected to %s:%d", self._host, self._port)
 
+    @property
+    def peer_cert_der(self) -> bytes | None:
+        """Binding anchor to carry alongside a saved session ticket (11.5.6)."""
+        return self._peer_cert_der
+
     async def close(self) -> None:
         if self._cm:
             await self._cm.__aexit__(None, None, None)
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 ba2d5c5..ed3925d 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py
@@ -20,6 +20,7 @@ The transport is the only change — all crypto, auth, and message types stay th
 import asyncio
 import base64
 import logging
+import os
 import struct
 import subprocess
 from pathlib import Path
@@ -34,7 +35,17 @@ 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.handshake import HandshakeError, authorize_token
+from meshbay_common.handshake import (
+    NONCE_LEN,
+    ROLE_CLIENT,
+    ROLE_NODE,
+    HandshakeError,
+    authorize_token,
+    handshake_transcript,
+    make_proof,
+    quic_binding,
+    verify_proof,
+)
 from meshbay_common.crypto import (
     sign_chunk,
     pk_to_b64,
@@ -158,6 +169,9 @@ class _MNPServerProtocol(QuicConnectionProtocol):
         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
 
     def quic_event_received(self, event: QuicEvent) -> None:
         if isinstance(event, StreamDataReceived):
@@ -177,6 +191,8 @@ class _MNPServerProtocol(QuicConnectionProtocol):
         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:
@@ -222,6 +238,67 @@ class _MNPServerProtocol(QuicConnectionProtocol):
             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,
+            "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
 
@@ -229,13 +306,21 @@ class _MNPServerProtocol(QuicConnectionProtocol):
         if peers is not None:
             peers[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])
         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(),
         })
+        self._gek_challenge = None
 
     def _group_ctx(self) -> dict:
         """Resolve the active group context (multi-group or legacy single-group)."""
@@ -459,6 +544,13 @@ class QuicChunkServer:
             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: