feat(mnp): unified handshake with mutual authentication

Phase 11.5.4/5/7/8 — findings C6 (WebRTC half), C3, L4, M1, M9.

New meshbay_common/handshake.py is the single implementation of authorization
and proof: JWT verify, scope, denylist, mandatory group_id, membership, hosting.
The handshake previously existed three times over and only the newest copy
enforced the GEK proof.

C3 — mutual authentication. Authentication ran one way: the client proved
itself, the node proved nothing. handshake_ack.node_pk was never verified
against anything and per-chunk signatures had been dropped in Phase 9.15, so a
peer that had hijacked signaling (C2) or been substituted by the hub could
accept the client's proof, ignore it, and serve a forged index, forged chat
history and a forged is_node_admin flag. The client now sends a nonce; the node
answers with its own GEK proof over that nonce AND an Ed25519 signature over
the transcript; the browser verifies both and refuses otherwise. It also
refuses an unchallenged handshake_ack, which previously let a peer skip proving
anything at all.

L4 — the proof was nonce ‖ offer_fp ‖ answer_fp: bare concatenation, and a
missing fingerprint silently degraded it to nonce-only, dropping MitM detection
(NS5). Every field is now length-prefixed and domain-separated, the role is
bound so a client proof cannot be replayed as a node proof, and an absent
channel binding is refused rather than tolerated.

M1 — group_id was optional; omitting it skipped the membership check entirely
and fell back to the node's first group. Now mandatory.
M9 — node-scoped daemon tokens are refused on the client path.

NOT DONE: quic_server.py still runs its own JWT-only handshake, so C6 remains
open — a forged or stolen token reaches a node over QUIC and can inject chat
without holding the GEK. quic_binding() is written and unit-tested but unwired.
11.5.6 (whether the certificate-hash anchor works with aioquic, or an RFC 5705
exporter is reachable) is unproven. 11.5.8 TOFU pinning of pk_node is not done:
the client verifies the node's signature but does not yet remember which key it
saw last.

Adds packages/meshbay-common/tests/test_handshake.py (18 tests) covering the
properties every transport must inherit. WebRTC test helpers rewritten around
the shared module; _make_jwt now defaults to the test group, since group_id is
mandatory.

Tests: 24 webrtc, 176+ node+common.

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

1 files changed, 86 insertions, 58 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 190d580..34a96bd 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -43,6 +43,17 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import (
 )
 
 from meshbay_common import MNP_VERSION
+from meshbay_common.handshake import (
+    NONCE_LEN,
+    ROLE_CLIENT,
+    ROLE_NODE,
+    HandshakeError,
+    authorize_token,
+    handshake_transcript,
+    make_proof,
+    verify_proof,
+    webrtc_binding,
+)
 from meshbay_common.adminop import (
     ADMIN_CHALLENGE_TTL,
     OP_FILE_DELETE,
@@ -207,6 +218,7 @@ class WebRTCPeerSession:
         self._username: str = ""
         self._pk_user: str = ""
         self._gek_challenge: bytes | None = None
+        self._nonce_client: bytes = b""
         self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
         self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}
 
@@ -293,58 +305,65 @@ class WebRTCPeerSession:
                 detail=detail,
             ))
 
+    def _channel_binding(self) -> bytes:
+        """Both DTLS fingerprints, so a proof is valid on this connection only."""
+        offer_fp = b""
+        answer_fp = b""
+        if self._pc.remoteDescription:
+            offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp)
+        if self._pc.localDescription:
+            answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp)
+        if not offer_fp or not answer_fp:
+            return b""
+        return webrtc_binding(offer_fp, answer_fp)
+
     def _do_handshake(self, 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({"type": "error", "detail": f"Invalid JWT: {e}"})
-            self._audit_auth_failed(group_id, str(e))
-            return
-
-        denylist = self._ctx.get("denylist")
-        if denylist and denylist.is_denied(
-                decoded.get("sub", ""), decoded.get("jti", ""), group_id):
-            self._send({"type": "error", "detail": "Token revoked"})
-            return
-
-        if group_id and group_id not in decoded.get("groups", []):
-            self._send({"type": "error", "detail": "Not a member of this group"})
+            peer = authorize_token(
+                msg.get("token", ""),
+                self._ctx["hub_pk_pem"],
+                group_id=group_id,
+                hosted_groups=self._ctx.get("groups"),
+                denylist=self._ctx.get("denylist"),
+            )
+        except HandshakeError as refusal:
+            # HandshakeError messages are authored to be peer-safe, unlike arbitrary
+            # exception text (L3) — the client needs to know *why* it was refused.
+            self._send({"type": "error", "detail": str(refusal)})
+            self._audit_auth_failed(group_id, str(refusal))
             return
 
-        if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]:
-            self._send({"type": "error", "detail": "Group not hosted on this node"})
+        try:
+            self._nonce_client = base64.b64decode(msg.get("nonce", ""))
+        except Exception:
+            self._nonce_client = b""
+        if len(self._nonce_client) < NONCE_LEN:
+            # The client nonce is what makes the NODE's proof fresh (C3). Without
+            # it a recorded ack could be replayed by an impersonating peer.
+            self._send({"type": "error", "detail": "Client nonce required"})
             return
 
-        # Store decoded JWT data but DO NOT set self._user_id yet —
-        # the user is not authenticated until they prove GEK possession.
-        self._pending_sub = decoded["sub"]
-        self._pending_group = group_id
-        self._pending_username = decoded.get("username", "")
-        self._pending_pk_user = decoded.get("pk_user", "")
+        # Decoded, but NOT authenticated: that happens on the GEK proof.
+        self._pending_sub = peer.user_id
+        self._pending_group = peer.group_id
+        self._pending_username = peer.username
+        self._pending_pk_user = peer.pk_user
 
-        ctx = self._ctx
-        if "groups" in ctx and group_id:
-            gctx = ctx["groups"].get(group_id, ctx)
-        else:
-            gctx = ctx
-        gek = gctx.get("gek")
-
-        nonce = os.urandom(32)
-        self._gek_challenge = nonce
-        challenge = {
-            "type": MNP.HANDSHAKE_CHALLENGE,
-            "v": MNP_VERSION,
-            "nonce": base64.b64encode(nonce).decode(),
-        }
-        if not gek:
+        gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx
+        if not gctx.get("gek"):
             self._send({
                 "type": "error",
                 "detail": "Group encryption not initialized — contact node operator",
             })
             return
-        self._send(challenge)
+
+        self._gek_challenge = os.urandom(NONCE_LEN)
+        self._send({
+            "type": MNP.HANDSHAKE_CHALLENGE,
+            "v": MNP_VERSION,
+            "nonce": base64.b64encode(self._gek_challenge).decode(),
+        })
 
     def _do_handshake_response(self, msg: dict) -> None:
         if not self._gek_challenge or not hasattr(self, "_pending_sub"):
@@ -352,44 +371,38 @@ class WebRTCPeerSession:
             return
 
         group_id = self._pending_group
-        ctx = self._ctx
-        if "groups" in ctx and group_id:
-            gctx = ctx["groups"].get(group_id, ctx)
-        else:
-            gctx = ctx
+        gctx = self._ctx["groups"][group_id] if "groups" in self._ctx else self._ctx
         gek = gctx.get("gek")
-
         if not gek:
             self._send({"type": "error", "detail": "Group encryption not initialized"})
             self._gek_challenge = None
             return
 
-        proof = msg.get("proof", "")
         try:
-            proof_bytes = base64.b64decode(proof)
+            proof_bytes = base64.b64decode(msg.get("proof", ""))
         except Exception:
             self._send({"type": "error", "detail": "Invalid proof encoding"})
             return
 
-        offer_fp = b""
-        answer_fp = b""
-        if self._pc.remoteDescription:
-            offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp)
-        if self._pc.localDescription:
-            answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp)
+        binding = self._channel_binding()
+        if not binding:
+            # Refuse rather than fall back to an unbound proof (L4).
+            self._send({"type": "error", "detail": "Channel binding unavailable"})
+            self._gek_challenge = None
+            self._audit_auth_failed(group_id, "no channel binding")
+            return
 
-        data = self._gek_challenge + offer_fp + answer_fp
-        expected = hmac.new(gek, data, hashlib.sha256).digest()
-        if not hmac.compare_digest(proof_bytes, expected):
+        if not verify_proof(gek, proof_bytes, ROLE_CLIENT, group_id,
+                            self._nonce_client, self._gek_challenge, binding):
             self._send({"type": "error", "detail": "GEK proof failed"})
             self._gek_challenge = None
             self._audit_auth_failed(group_id, "GEK HMAC mismatch")
             return
 
+        self._complete_handshake(gek, binding)
         self._gek_challenge = None
-        self._complete_handshake()
 
-    def _complete_handshake(self) -> None:
+    def _complete_handshake(self, gek: bytes, binding: bytes) -> None:
         # Authenticated peers may send large frames (file uploads); unauthenticated
         # ones may not (H6).
         self._buffer.max_message = MAX_MSG
@@ -404,10 +417,25 @@ class WebRTCPeerSession:
         log.info("WebRTC handshake OK — user=%s group=%s",
                  self._user_id[:8],
                  self._group_id[:8] if self._group_id else "none")
+        # The node proves itself too (C3): possession of the GEK over the client's
+        # nonce, plus a signature over the same transcript with its long-term key.
+        # Previously the client received an unverifiable node_pk and trusted
+        # is_node_admin from whoever answered — so a peer that had hijacked
+        # signaling could serve a forged index, chat history and permissions.
+        node_transcript = handshake_transcript(
+            ROLE_NODE, self._group_id or "", self._nonce_client,
+            self._gek_challenge or b"", binding)
+        node_proof = make_proof(
+            gek, ROLE_NODE, self._group_id or "", self._nonce_client,
+            self._gek_challenge or b"", binding)
+
         ack = {
             "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(node_transcript)).decode(),
             "is_node_admin": bool(node_user_id and self._user_id == node_user_id),
         }
         if node_user_id: