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>

7 files changed, 680 insertions, 123 deletions

diff --git a/packages/meshbay-common/src/meshbay_common/handshake.py b/packages/meshbay-common/src/meshbay_common/handshake.py
new file mode 100644
index 0000000..73a2858
--- /dev/null
+++ b/packages/meshbay-common/src/meshbay_common/handshake.py
@@ -0,0 +1,203 @@
+"""
+Unified MNP handshake — one implementation, every transport.
+
+Finding C6: the handshake existed three times over (WebRTC, QUIC, TCP), and only
+the newest copy enforced the GEK proof. QUIC and TCP accepted a bare JWT, so a
+forged or stolen token reached the node and could inject chat messages without
+ever holding the group key. TCP is gone (11.5.2); QUIC and WebRTC now share this
+module, and a parity test fails if either skips a step.
+
+The sequence:
+
+    client → node   handshake            {token, group_id, nonce_c}
+    node            authorize_token()    JWT, scope, denylist, membership, hosting
+    node → client   handshake_challenge  {nonce_s}
+    client → node   handshake_response   {proof}
+    node            verify client proof  HMAC(GEK, client transcript)
+    node → client   handshake_ack        {proof, sig, node_pk, is_node_admin}
+    client          verify node proof    HMAC(GEK, node transcript) + Ed25519
+
+Two properties this adds over the previous design:
+
+**Mutual authentication (C3).** Authentication used to run 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 node now proves GEK possession over a
+client-chosen nonce *and* signs the transcript with its long-term key, so the
+client can pin it.
+
+**Unambiguous transcripts (L4).** The old proof was `nonce ‖ offer_fp ‖ answer_fp`
+— bare concatenation, and a missing fingerprint silently degraded it to nonce-only.
+Every field is now length-prefixed and domain-separated, the role is bound so a
+client proof can never be replayed as a node proof, and an empty channel binding is
+refused rather than tolerated.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+import jwt
+
+HANDSHAKE_PREFIX = b"meshbay:mnp:handshake:v1"
+
+ROLE_CLIENT = "client"
+ROLE_NODE = "node"
+
+NONCE_LEN = 32
+
+
+class HandshakeError(Exception):
+    """Refusal, with a message safe to hand to the peer."""
+
+
+class DenylistLike(Protocol):
+    def is_denied(self, user_id: str, jti: str, group_id: str = "") -> bool: ...
+
+
+@dataclass
+class AuthorizedPeer:
+    user_id: str
+    group_id: str
+    username: str
+    pk_user: str
+    jti: str
+
+
+def handshake_transcript(
+    role: str,
+    group_id: str,
+    nonce_client: bytes,
+    nonce_node: bytes,
+    binding: bytes,
+) -> bytes:
+    """
+    Bytes covered by a handshake proof.
+
+    `binding` ties the proof to the concrete connection: the two DTLS fingerprints
+    for WebRTC, the TLS certificate hashes for QUIC. Without it a proof captured on
+    one connection is replayable on another (NS5).
+    """
+    fields = [
+        role.encode(),
+        group_id.encode(),
+        nonce_client,
+        nonce_node,
+        binding,
+    ]
+    out = bytearray(HANDSHAKE_PREFIX)
+    for field in fields:
+        out += len(field).to_bytes(4, "big")
+        out += field
+    return bytes(out)
+
+
+def make_proof(
+    gek: bytes,
+    role: str,
+    group_id: str,
+    nonce_client: bytes,
+    nonce_node: bytes,
+    binding: bytes,
+) -> bytes:
+    if not binding:
+        # An empty binding means the transport could not identify the channel.
+        # Proceeding would silently drop MitM detection (L4).
+        raise HandshakeError("Channel binding unavailable")
+    if not gek:
+        raise HandshakeError("Group encryption not initialized")
+    transcript = handshake_transcript(
+        role, group_id, nonce_client, nonce_node, binding)
+    return hmac.new(gek, transcript, hashlib.sha256).digest()
+
+
+def verify_proof(
+    gek: bytes,
+    proof: bytes,
+    role: str,
+    group_id: str,
+    nonce_client: bytes,
+    nonce_node: bytes,
+    binding: bytes,
+) -> bool:
+    try:
+        expected = make_proof(
+            gek, role, group_id, nonce_client, nonce_node, binding)
+    except HandshakeError:
+        return False
+    return hmac.compare_digest(proof, expected)
+
+
+def authorize_token(
+    token: str,
+    hub_pk_pem: bytes,
+    *,
+    group_id: str,
+    hosted_groups: Any | None = None,
+    denylist: DenylistLike | None = None,
+    require_scope: str | None = "user",
+) -> AuthorizedPeer:
+    """
+    Everything decided from the JWT, before any proof is exchanged.
+
+    Raises HandshakeError with a peer-safe message. Deliberately strict about
+    `group_id`: it used to be optional, and omitting it skipped the membership
+    check entirely and fell back to the node's first group (M1).
+    """
+    try:
+        decoded = jwt.decode(token, hub_pk_pem, algorithms=["EdDSA"])
+    except Exception as exc:
+        raise HandshakeError(f"Invalid JWT: {exc}") from exc
+
+    # A node-scoped daemon token must not be usable as a client token (M9).
+    if require_scope is not None and decoded.get("scope", "user") != require_scope:
+        raise HandshakeError("Wrong token scope")
+
+    user_id = decoded.get("sub", "")
+    jti = decoded.get("jti", "")
+    if not user_id:
+        raise HandshakeError("Token has no subject")
+
+    if not group_id:
+        raise HandshakeError("group_id is required")
+
+    if denylist is not None and denylist.is_denied(user_id, jti, group_id):
+        raise HandshakeError("Token revoked")
+
+    if group_id not in decoded.get("groups", []):
+        raise HandshakeError("Not a member of this group")
+
+    if hosted_groups is not None and group_id not in hosted_groups:
+        raise HandshakeError("Group not hosted on this node")
+
+    return AuthorizedPeer(
+        user_id=user_id,
+        group_id=group_id,
+        username=decoded.get("username", ""),
+        pk_user=decoded.get("pk_user", ""),
+        jti=jti,
+    )
+
+
+def webrtc_binding(offer_fp: bytes, answer_fp: bytes) -> bytes:
+    """Channel binding for WebRTC: both DTLS certificate fingerprints."""
+    return (len(offer_fp).to_bytes(4, "big") + offer_fp
+            + len(answer_fp).to_bytes(4, "big") + answer_fp)
+
+
+def quic_binding(server_cert_der: bytes) -> bytes:
+    """
+    Channel binding for QUIC.
+
+    QUIC has no DTLS fingerprint to reuse, so the anchor is a hash of the server's
+    self-signed certificate — the same value a client pins as the node identity.
+    An RFC 5705 exporter would be stronger; aioquic does not currently expose one
+    (11.5.6).
+    """
+    digest = hashlib.sha256(server_cert_der).digest()
+    return len(digest).to_bytes(4, "big") + digest
diff --git a/packages/meshbay-common/tests/test_handshake.py b/packages/meshbay-common/tests/test_handshake.py
new file mode 100644
index 0000000..ba8788a
--- /dev/null
+++ b/packages/meshbay-common/tests/test_handshake.py
@@ -0,0 +1,199 @@
+"""
+Unified handshake — properties every transport must inherit (11.5.4/5, C6, C3, L4).
+
+These test the shared module rather than any one transport. The point of the module
+is that WebRTC and QUIC cannot drift apart again: the handshake existed three times
+over and only the newest copy enforced the GEK proof.
+"""
+
+import time
+
+import jwt
+import pytest
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+from meshbay_common.handshake import (
+    HANDSHAKE_PREFIX,
+    NONCE_LEN,
+    ROLE_CLIENT,
+    ROLE_NODE,
+    AuthorizedPeer,
+    HandshakeError,
+    authorize_token,
+    handshake_transcript,
+    make_proof,
+    quic_binding,
+    verify_proof,
+    webrtc_binding,
+)
+
+GEK = b"\x11" * 32
+GROUP = "g" * 32
+NONCE_C = b"\x01" * NONCE_LEN
+NONCE_S = b"\x02" * NONCE_LEN
+BINDING = webrtc_binding(b"\xaa" * 32, b"\xbb" * 32)
+
+
+# ── Token authorization ───────────────────────────────────────────────────────
+
+@pytest.fixture
+def hub_key():
+    sk = Ed25519PrivateKey.generate()
+    pem = sk.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+    pub = sk.public_key().public_bytes(
+        serialization.Encoding.PEM,
+        serialization.PublicFormat.SubjectPublicKeyInfo,
+    )
+    return pem, pub
+
+
+def _token(sk_pem, **over):
+    now = int(time.time())
+    payload = {
+        "iss": "test-hub", "sub": "user-1", "jti": "jti-1",
+        "iat": now, "exp": now + 3600,
+        "groups": [GROUP], "scope": "user", "pk_user": "pk",
+    }
+    payload.update(over)
+    return jwt.encode(payload, sk_pem, algorithm="EdDSA")
+
+
+def test_valid_token_authorizes(hub_key):
+    sk_pem, pk_pem = hub_key
+    peer = authorize_token(_token(sk_pem), pk_pem, group_id=GROUP)
+    assert isinstance(peer, AuthorizedPeer)
+    assert peer.user_id == "user-1"
+
+
+def test_group_id_is_mandatory(hub_key):
+    """
+    M1: group_id used to be optional, and omitting it skipped the membership check
+    entirely while falling back to the node's first group.
+    """
+    sk_pem, pk_pem = hub_key
+    with pytest.raises(HandshakeError, match="group_id"):
+        authorize_token(_token(sk_pem), pk_pem, group_id="")
+
+
+def test_non_member_refused(hub_key):
+    sk_pem, pk_pem = hub_key
+    token = _token(sk_pem, groups=["other-group"])
+    with pytest.raises(HandshakeError, match="Not a member"):
+        authorize_token(token, pk_pem, group_id=GROUP)
+
+
+def test_node_scoped_token_refused_on_client_path(hub_key):
+    """M9: a daemon's node-scoped token must not be usable as a client token."""
+    sk_pem, pk_pem = hub_key
+    token = _token(sk_pem, scope="node")
+    with pytest.raises(HandshakeError, match="scope"):
+        authorize_token(token, pk_pem, group_id=GROUP)
+
+
+def test_unhosted_group_refused(hub_key):
+    sk_pem, pk_pem = hub_key
+    with pytest.raises(HandshakeError, match="not hosted"):
+        authorize_token(_token(sk_pem), pk_pem, group_id=GROUP,
+                        hosted_groups={"some-other-group"})
+
+
+def test_denylisted_token_refused(hub_key):
+    sk_pem, pk_pem = hub_key
+
+    class _Deny:
+        def is_denied(self, user_id, jti, group_id=""):
+            return group_id == GROUP
+
+    with pytest.raises(HandshakeError, match="revoked"):
+        authorize_token(_token(sk_pem), pk_pem, group_id=GROUP, denylist=_Deny())
+
+
+def test_forged_token_refused(hub_key):
+    _, pk_pem = hub_key
+    other = Ed25519PrivateKey.generate().private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+    with pytest.raises(HandshakeError, match="Invalid JWT"):
+        authorize_token(_token(other), pk_pem, group_id=GROUP)
+
+
+# ── Proof transcript ──────────────────────────────────────────────────────────
+
+def test_transcript_is_domain_separated():
+    assert handshake_transcript(
+        ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING
+    ).startswith(HANDSHAKE_PREFIX)
+
+
+def test_client_proof_is_not_a_node_proof():
+    """
+    C3: the node proves itself with the same key over the same connection. Without
+    the role bound in, a client's proof would satisfy the node check and vice
+    versa, so an impersonating peer could simply echo it back.
+    """
+    client = make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
+    assert not verify_proof(GEK, client, ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING)
+
+    node = make_proof(GEK, ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING)
+    assert not verify_proof(GEK, node, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
+    assert client != node
+
+
+@pytest.mark.parametrize("field,value", [
+    ("group_id", "other-group"),
+    ("nonce_client", b"\x09" * NONCE_LEN),
+    ("nonce_node", b"\x09" * NONCE_LEN),
+    ("binding", webrtc_binding(b"\xcc" * 32, b"\xdd" * 32)),
+])
+def test_proof_binds_every_field(field, value):
+    base = dict(role=ROLE_CLIENT, group_id=GROUP, nonce_client=NONCE_C,
+                nonce_node=NONCE_S, binding=BINDING)
+    proof = make_proof(GEK, **base)
+    altered = dict(base, **{field: value})
+    assert not verify_proof(GEK, proof, **altered), (
+        f"proof ignores {field} — replayable across connections")
+
+
+def test_proof_requires_channel_binding():
+    """
+    L4/NS5: the old transcript was nonce ‖ offer_fp ‖ answer_fp, and a missing
+    fingerprint silently degraded it to nonce-only, dropping MitM detection.
+    """
+    with pytest.raises(HandshakeError, match="binding"):
+        make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, b"")
+
+    assert not verify_proof(
+        GEK, b"\x00" * 32, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, b"")
+
+
+def test_proof_requires_gek():
+    with pytest.raises(HandshakeError):
+        make_proof(b"", ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
+
+
+def test_wrong_gek_fails():
+    proof = make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
+    assert not verify_proof(
+        b"\x22" * 32, proof, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING)
+
+
+def test_transcript_is_unambiguous():
+    """
+    L4: with bare concatenation, a crafted group id could impersonate the
+    following field and two different handshakes would produce identical bytes.
+    """
+    a = handshake_transcript(ROLE_CLIENT, "gg", NONCE_C, NONCE_S, BINDING)
+    b = handshake_transcript(ROLE_CLIENT, "g", b"g" + NONCE_C, NONCE_S, BINDING)
+    assert a != b
+
+
+def test_bindings_differ_by_transport():
+    """A WebRTC proof must not be replayable on a QUIC connection."""
+    assert webrtc_binding(b"\xaa" * 32, b"\xbb" * 32) != quic_binding(b"cert-der")
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/crypto.js b/packages/meshbay-hub/src/meshbay_hub/static/crypto.js
index 2346fa2..d18eeae 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/crypto.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/crypto.js
@@ -268,22 +268,70 @@ function adminTranscript(op, nodePkB64, groupId, subject, nonceB64, ts) {
 
 // ── GEK proof (HMAC-SHA256 for handshake challenge) ─────────────────────────
 
-async function hmacGEK(gekRaw, nonceB64, offerFp, answerFp) {
-  const nonce = b64decode(nonceB64);
-  const data = concatBuffers([
-    nonce,
-    offerFp || new Uint8Array(0),
-    answerFp || new Uint8Array(0),
+// Mirrors meshbay_common/handshake.py. Every field length-prefixed and the role
+// bound in, so a client proof can never be replayed as a node proof and a missing
+// fingerprint cannot silently degrade the proof to nonce-only (L4).
+const HANDSHAKE_PREFIX = new TextEncoder().encode('meshbay:mnp:handshake:v1');
+
+function _lenPrefixed(parts) {
+  let total = 0;
+  for (const p of parts) total += 4 + p.length;
+  const out = new Uint8Array(total);
+  const view = new DataView(out.buffer);
+  let off = 0;
+  for (const p of parts) {
+    view.setUint32(off, p.length, false);
+    off += 4;
+    out.set(p, off);
+    off += p.length;
+  }
+  return out;
+}
+
+function webrtcBinding(offerFp, answerFp) {
+  if (!offerFp || !offerFp.length || !answerFp || !answerFp.length) {
+    throw new Error('Channel binding unavailable — refusing to handshake');
+  }
+  return _lenPrefixed([offerFp, answerFp]);
+}
+
+function handshakeTranscript(role, groupId, nonceClient, nonceNode, binding) {
+  const enc = new TextEncoder();
+  const body = _lenPrefixed([
+    enc.encode(role), enc.encode(groupId), nonceClient, nonceNode, binding,
   ]);
+  const out = new Uint8Array(HANDSHAKE_PREFIX.length + body.length);
+  out.set(HANDSHAKE_PREFIX, 0);
+  out.set(body, HANDSHAKE_PREFIX.length);
+  return out;
+}
+
+async function handshakeProof(gekRaw, role, groupId, nonceClient, nonceNode, binding) {
+  const transcript = handshakeTranscript(role, groupId, nonceClient, nonceNode, binding);
   const key = await crypto.subtle.importKey(
     'raw', gekRaw, { name: 'HMAC', hash: 'SHA-256' }, false, ['sign']);
-  const sig = await crypto.subtle.sign('HMAC', key, data);
-  return b64encode(new Uint8Array(sig));
+  const sig = await crypto.subtle.sign('HMAC', key, transcript);
+  return new Uint8Array(sig);
+}
+
+function constantTimeEqual(a, b) {
+  if (a.length !== b.length) return false;
+  let diff = 0;
+  for (let i = 0; i < a.length; i++) diff |= a[i] ^ b[i];
+  return diff === 0;
+}
+
+/** Verify the node's Ed25519 signature over the handshake transcript (C3). */
+async function verifyNodeSignature(nodePkB64, sigB64, transcript) {
+  const raw = b64decode(nodePkB64);
+  const key = await crypto.subtle.importKey('raw', raw, { name: 'Ed25519' }, false, ['verify']);
+  return crypto.subtle.verify('Ed25519', key, b64decode(sigB64), transcript);
 }
 
 // Export for use in app.js
 window.MeshBayCrypto = {
   importGEK, deriveChunkKey, decryptChunk, decryptChunkBin, decryptFile,
   generateGEK, wrapGEK, unwrapGEK, encryptChunk, b64encode, b64decode,
-  hmacGEK, adminTranscript,
+  adminTranscript, handshakeTranscript, handshakeProof, webrtcBinding,
+  verifyNodeSignature, constantTimeEqual,
 };
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index d636085..18faea1 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -129,11 +129,16 @@ class MeshBayTransport {
 
     await channelReady;
 
+    // The client nonce is what makes the NODE's proof fresh (C3) — without it a
+    // recorded handshake_ack could be replayed by an impersonating peer.
+    this._nonceClient = crypto.getRandomValues(new Uint8Array(32));
+
     const reply = await this._sendAndWait({
       type: 'handshake',
       v: '0.1',
       token: jwtToken,
       group_id: groupId || '',
+      nonce: window.MeshBayCrypto.b64encode(this._nonceClient),
     });
 
     if (reply.type === 'handshake_challenge') {
@@ -190,28 +195,53 @@ class MeshBayTransport {
         throw new Error('Node requires GEK proof but no GEK available');
       }
 
-      let proof = '';
-      if (gekRaw) {
-        const offerFp = _extractDtlsFingerprint(this._pc.localDescription.sdp);
-        const answerFp = _extractDtlsFingerprint(this._rawAnswerSdp);
-        proof = await window.MeshBayCrypto.hmacGEK(gekRaw, reply.nonce, offerFp, answerFp);
-      }
+      const C = window.MeshBayCrypto;
+      // Node's answer SDP carries ITS fingerprint; our offer carries ours. Throws
+      // if either is missing rather than proceeding with an unbound proof (L4).
+      const binding = C.webrtcBinding(
+        _extractDtlsFingerprint(this._pc.localDescription.sdp),
+        _extractDtlsFingerprint(this._rawAnswerSdp),
+      );
+      const nonceNode = C.b64decode(reply.nonce);
+      const gid = groupId || '';
+
+      const proof = await C.handshakeProof(
+        gekRaw, 'client', gid, this._nonceClient, nonceNode, binding);
+
       const ack = await this._sendAndWait({
         type: 'handshake_response',
         v: '0.1',
-        proof,
+        proof: C.b64encode(proof),
       });
       if (ack.type !== 'handshake_ack') {
         throw new Error('GEK proof rejected: ' + (ack.detail || JSON.stringify(ack)));
       }
-      return ack;
-    }
 
-    if (reply.type !== 'handshake_ack') {
-      throw new Error('MNP handshake rejected: ' + (reply.detail || JSON.stringify(reply)));
+      // Authenticate the NODE before trusting anything it says (C3). Until this
+      // ran, node_pk was decorative: a peer that had hijacked signaling could
+      // accept our proof, ignore it, and serve a forged index, chat history and
+      // is_node_admin flag.
+      const expected = await C.handshakeProof(
+        gekRaw, 'node', gid, this._nonceClient, nonceNode, binding);
+      if (!ack.proof || !C.constantTimeEqual(C.b64decode(ack.proof), expected)) {
+        throw new Error('Node failed to prove GEK possession — refusing connection');
+      }
+      const transcript = C.handshakeTranscript(
+        'node', gid, this._nonceClient, nonceNode, binding);
+      if (!ack.node_pk || !ack.sig
+          || !await C.verifyNodeSignature(ack.node_pk, ack.sig, transcript)) {
+        throw new Error('Node signature invalid — refusing connection');
+      }
+      this.nodePk = ack.node_pk;
+
+      return ack;
     }
 
-    return reply;
+    // A node that answers a handshake with anything other than a challenge is not
+    // running the mutual protocol. Accepting a bare handshake_ack here would let a
+    // peer skip proving GEK possession entirely (C3/C6).
+    throw new Error(
+      'MNP handshake rejected: ' + (reply.detail || `unexpected ${reply.type}`));
   }
 
   async fetchIndex() {
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:
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index a6d69c6..9299bf4 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -460,10 +460,21 @@ def test_dead_gek_protocol_constants_removed():
 
 
 def test_peer_errors_do_not_leak_internals():
-    """L3: exception text carries filesystem paths and internal state."""
+    """
+    L3: arbitrary exception text carries filesystem paths and internal state, so
+    the catch-all handler must not relay it.
+
+    Deliberately narrow: HandshakeError messages ARE sent to the peer, because a
+    client needs to know why it was refused, and those strings are authored for
+    that purpose. The check targets the generic `except Exception as e` path.
+    """
     source = (Path(__file__).parent.parent / "src" / "meshbay_node"
               / "transport" / "webrtc_server.py").read_text()
-    assert '"detail": str(e)' not in source
+    assert '"detail": str(e)' not in source, (
+        "generic exception text relayed to peer — use a fixed message"
+    )
+    # And the catch-all must still exist, sending something opaque.
+    assert '"detail": "Request failed"' in source
 
 
 def test_pre_handshake_message_budget_is_small():
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index d57f4f5..59b48ac 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -19,7 +19,9 @@ import jwt
 import msgpack
 import pytest
 from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey, Ed25519PublicKey,
+)
 from aiortc import RTCPeerConnection, RTCSessionDescription
 
 from meshbay_common import MNP_VERSION
@@ -32,6 +34,12 @@ from meshbay_common.crypto import (
 )
 from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
 from meshbay_common.protocol import MNP
+TEST_GROUP = "g"
+
+from meshbay_common.handshake import (
+    NONCE_LEN, ROLE_CLIENT, ROLE_NODE, handshake_transcript,
+    make_proof, verify_proof, webrtc_binding,
+)
 from meshbay_common.adminop import (
     OP_FILE_DELETE,
     OP_GEK_BUNDLE_STORE,
@@ -77,6 +85,9 @@ def _hub_pk_pem(sk_hub):
 
 
 def _make_jwt(sk_hub, groups=None, pk_user="test"):
+    # group_id is mandatory now (M1), so the default token must be a member
+    # of the group the tests connect to. Tests that exercise refusal pass
+    # groups=[...] explicitly.
     sk_pem = sk_hub.private_bytes(
         serialization.Encoding.PEM,
         serialization.PrivateFormat.PKCS8,
@@ -87,7 +98,7 @@ def _make_jwt(sk_hub, groups=None, pk_user="test"):
         "iss": "test-hub", "sub": "user-001",
         "pk_user": pk_user, "hub_id": "test-hub",
         "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600,
-        "groups": groups or [],
+        "groups": groups if groups is not None else [TEST_GROUP],
     }, sk_pem, algorithm="EdDSA")
 
 
@@ -123,35 +134,58 @@ def _extract_dtls_fp(sdp: str) -> bytes:
     return b""
 
 
-async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None,
-                                    browser_pc=None):
-    """Send handshake, handle GEK challenge, return handshake_ack."""
-    token = _make_jwt(sk_hub, groups=groups)
+async def _do_mnp_handshake(channel, received, token, gek, pc, group_id):
+    """
+    Client half of the unified handshake (11.5.4): client nonce, length-prefixed
+    role-bound transcript, and verification of the node's own proof + signature.
+    """
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
-        "type": MNP.HANDSHAKE,
-        "v": MNP_VERSION,
-        "token": token,
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION,
+        "token": token, "group_id": group_id,
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
-    if msg["type"] == MNP.HANDSHAKE_CHALLENGE:
-        nonce = base64.b64decode(msg["nonce"])
-        offer_fp = b""
-        answer_fp = b""
-        if browser_pc:
-            offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp)
-            answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp)
-        proof = hmac.new(gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest()
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE_RESPONSE,
-            "v": MNP_VERSION,
-            "proof": base64.b64encode(proof).decode(),
-        }))
-        msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    if msg["type"] != MNP.HANDSHAKE_CHALLENGE:
+        return msg
+
+    nonce_s = base64.b64decode(msg["nonce"])
+    binding = webrtc_binding(
+        _extract_dtls_fp(pc.localDescription.sdp),
+        _extract_dtls_fp(pc.remoteDescription.sdp),
+    )
+    proof = make_proof(gek, ROLE_CLIENT, group_id, nonce_c, nonce_s, binding)
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
+        "proof": base64.b64encode(proof).decode(),
+    }))
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+
+    if ack.get("type") == MNP.HANDSHAKE_ACK:
+        # The client must authenticate the node too (C3).
+        assert verify_proof(
+            gek, base64.b64decode(ack["proof"]), ROLE_NODE,
+            group_id, nonce_c, nonce_s, binding), "node proof invalid"
+        transcript = handshake_transcript(
+            ROLE_NODE, group_id, nonce_c, nonce_s, binding)
+        Ed25519PublicKey.from_public_bytes(
+            base64.b64decode(ack["node_pk"])
+        ).verify(base64.b64decode(ack["sig"]), transcript)
+    return ack
+
+
+async def _handshake_with_gek_proof(channel, received, sk_hub, gek, groups=None,
+                                    browser_pc=None, group_id=TEST_GROUP):
+    """Send handshake, handle GEK challenge, return handshake_ack."""
+    token = _make_jwt(sk_hub, groups=groups or [group_id])
+    msg = await _do_mnp_handshake(
+        channel, received, token, gek, browser_pc, group_id)
     assert msg["type"] == MNP.HANDSHAKE_ACK
     return msg
 
 
-async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None):
+async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_user=None,
+                      group_id=TEST_GROUP):
     """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue)."""
     pc = RTCPeerConnection()
     q = asyncio.Queue()
@@ -199,21 +233,10 @@ async def _setup_peer(transport, sk_hub, gek, peer_id, jwt_sub="user-001", sk_us
         "iss": "test-hub", "sub": jwt_sub,
         "pk_user": pk_user, "hub_id": "test-hub",
         "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
-        "groups": [],
+        "groups": [group_id], "scope": "user",
     }, sk_h_pem, algorithm="EdDSA")
 
-    ch.send(_pack({"type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token}))
-    msg = await asyncio.wait_for(q.get(), timeout=5.0)
-    if msg["type"] == MNP.HANDSHAKE_CHALLENGE:
-        nonce = base64.b64decode(msg["nonce"])
-        offer_fp = _extract_dtls_fp(pc.localDescription.sdp)
-        answer_fp = _extract_dtls_fp(pc.remoteDescription.sdp)
-        proof = hmac.new(gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest()
-        ch.send(_pack({
-            "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
-            "proof": base64.b64encode(proof).decode(),
-        }))
-        msg = await asyncio.wait_for(q.get(), timeout=5.0)
+    msg = await _do_mnp_handshake(ch, q, token, gek, pc, group_id)
     assert msg["type"] == MNP.HANDSHAKE_ACK
     return pc, ch, q
 
@@ -696,6 +719,8 @@ async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir)
     token = _make_jwt(sk_hub)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
     }))
 
     challenge = await asyncio.wait_for(received.get(), timeout=5.0)
@@ -754,6 +779,8 @@ async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, sh
     token = _make_jwt(sk_hub)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
     }))
 
     challenge = await asyncio.wait_for(received.get(), timeout=5.0)
@@ -1159,8 +1186,10 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
 
     # Step 1: Send handshake with group_id so _pending_group is set
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
@@ -1175,11 +1204,12 @@ async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_di
     recovered_gek = unwrap_gek(bundle_resp, sk_x_raw, pk_x_raw)
     assert recovered_gek == gek
 
-    nonce = base64.b64decode(msg["nonce"])
-    offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp)
-    answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp)
-    proof = hmac.new(recovered_gek, nonce + offer_fp + answer_fp,
-                     hashlib.sha256).digest()
+    nonce_s = base64.b64decode(msg["nonce"])
+    binding = webrtc_binding(
+        _extract_dtls_fp(browser_pc.localDescription.sdp),
+        _extract_dtls_fp(browser_pc.remoteDescription.sdp),
+    )
+    proof = make_proof(recovered_gek, ROLE_CLIENT, "g", nonce_c, nonce_s, binding)
 
     # Step 4: Complete handshake
     channel.send(_pack({
@@ -1264,8 +1294,10 @@ async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
     await asyncio.wait_for(ready.wait(), timeout=5.0)
 
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
@@ -1331,8 +1363,10 @@ async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir,
     await asyncio.wait_for(ready.wait(), timeout=5.0)
 
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
@@ -1448,6 +1482,8 @@ async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir):
     token = _make_jwt(sk_hub)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+        "group_id": TEST_GROUP,
+        "nonce": base64.b64encode(os.urandom(NONCE_LEN)).decode(),
     }))
 
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
@@ -1507,8 +1543,10 @@ async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_
     await asyncio.wait_for(ready.wait(), timeout=5.0)
 
     token = _make_jwt(sk_hub, groups=["g"])
+    nonce_c = os.urandom(NONCE_LEN)
     channel.send(_pack({
         "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+        "nonce": base64.b64encode(nonce_c).decode(),
     }))
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == MNP.HANDSHAKE_CHALLENGE