feat: Phase 12 — P2P crypto material, password split, node Ed25519 auth

Baseline commit capturing in-progress Phase 12 work that was already present
in the working tree (uncommitted) before the Phase 11.5 security remediation
begins. Committed as-is, without review or modification, so that remediation
changes arrive as a separable diff.

Contents: BundleStore (P2P GEK + keypair bundles), password split
(auth_key / bundle_key), node Ed25519 auth (POST /v1/nodes/auth, node-scoped
JWT), GEK-HMAC handshake proof with DTLS channel binding, Ed25519 admin
challenge-response, node local admin UI rewrite, browser key persistence.

Not authored in this session — captured to establish a baseline.

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

3 files changed, 918 insertions, 268 deletions

diff --git a/packages/meshbay-node/tests/test_daemon.py b/packages/meshbay-node/tests/test_daemon.py
index faf12e3..1c5a07e 100644
--- a/packages/meshbay-node/tests/test_daemon.py
+++ b/packages/meshbay-node/tests/test_daemon.py
@@ -7,11 +7,13 @@ Hub interaction is mocked.
 """
 
 import asyncio
+import base64
 import os
 
 import pytest
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from meshbay_common.crypto import generate_gek
@@ -20,6 +22,20 @@ from meshbay_node.daemon import NodeDaemon
 from meshbay_node.indexer import DirectoryIndexer
 
 
+def _mock_keystore_keys(sk_ed):
+    """Create a mock keystore with real Ed25519 + X25519 key material."""
+    sk_x = X25519PrivateKey.generate()
+    pk_x_raw = sk_x.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+
+    mock_keys = MagicMock()
+    mock_keys.sk_ed25519 = sk_ed
+    mock_keys.pk_ed25519_b64 = "test"
+    mock_keys.sk_x25519 = sk_x
+    mock_keys.pk_x25519_b64 = base64.b64encode(pk_x_raw).decode()
+    return mock_keys
+
+
 @pytest.fixture
 def sk_hub():
     return Ed25519PrivateKey.generate()
@@ -48,7 +64,7 @@ def shared_dir(tmp_path):
 @pytest.fixture
 def node_config(tmp_path, shared_dir):
     return Config(
-        hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"),
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
         node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000),
         groups=[GroupConfig(
             id="g" * 32,
@@ -70,10 +86,7 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem)
     daemon = NodeDaemon(node_config)
 
     sk_node = Ed25519PrivateKey.generate()
-    mock_keys = MagicMock()
-    mock_keys.sk_ed25519 = sk_node
-    mock_keys.pk_ed25519_b64 = "test"
-    mock_keys.pk_x25519_b64 = "test"
+    mock_keys = _mock_keystore_keys(sk_node)
 
     mock_session = MagicMock()
     mock_session.node_id = "node123"
@@ -85,7 +98,6 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem)
 
         hub_instance = AsyncMock()
         hub_instance.startup = AsyncMock(return_value=mock_session)
-        hub_instance.fetch_gek = AsyncMock(return_value=gek)
         hub_instance.maintain_ws = AsyncMock()
         hub_instance.send_ws = AsyncMock()
         hub_instance._ws = None
@@ -139,7 +151,7 @@ async def test_daemon_creates_chat_store(tmp_path, node_config, gek, hub_pk_pem)
 async def test_daemon_no_groups_exits(tmp_path):
     """Daemon with no valid groups exits cleanly."""
     config = Config(
-        hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"),
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
         node=NodeConfig(),
         groups=[GroupConfig(id="", name="empty", shared_dir="")],
         keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
@@ -148,17 +160,22 @@ async def test_daemon_no_groups_exits(tmp_path):
     daemon = NodeDaemon(config)
 
     sk_node = Ed25519PrivateKey.generate()
-    mock_keys = MagicMock()
-    mock_keys.sk_ed25519 = sk_node
-    mock_keys.pk_ed25519_b64 = "test"
+    mock_keys = _mock_keystore_keys(sk_node)
 
     mock_session = MagicMock()
     mock_session.node_id = "node123"
     mock_session.user_id = "user123"
     mock_session.hub_pk_pem = b"pem"
 
+    mock_server = AsyncMock()
+    mock_server.serve = AsyncMock()
+
     with patch("meshbay_node.daemon.load_or_create_keystore", return_value=mock_keys), \
-         patch("meshbay_node.daemon.HubClient") as MockHub:
+         patch("meshbay_node.daemon.HubClient") as MockHub, \
+         patch("meshbay_node.daemon.uvicorn") as mock_uvicorn:
+
+        mock_uvicorn.Config = MagicMock()
+        mock_uvicorn.Server = MagicMock(return_value=mock_server)
 
         hub_instance = AsyncMock()
         hub_instance.startup = AsyncMock(return_value=mock_session)
@@ -169,7 +186,6 @@ async def test_daemon_no_groups_exits(tmp_path):
 
         await daemon.run()
 
-    assert daemon._state["status"] == "starting"
     assert len(daemon._chat_stores) == 0
 
 
@@ -177,7 +193,7 @@ async def test_daemon_no_groups_exits(tmp_path):
 async def test_daemon_index_change_pushes_to_peers(tmp_path, shared_dir, gek, hub_pk_pem):
     """Index change callback pushes updated index to WebRTC peers."""
     config = Config(
-        hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"),
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
         node=NodeConfig(port=29000, quic_port=29010, http_port=29001, ui_port=28000),
         groups=[GroupConfig(
             id="a" * 32,
@@ -228,7 +244,7 @@ async def test_daemon_index_change_skips_other_group_peers(
 ):
     """Index change only pushes to peers in the same group."""
     config = Config(
-        hub=HubConfig(url="http://localhost:9999", username="testuser", password="testpass"),
+        hub=HubConfig(url="http://localhost:9999", username="testuser"),
         node=NodeConfig(),
         groups=[],
         keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
diff --git a/packages/meshbay-node/tests/test_hub_client.py b/packages/meshbay-node/tests/test_hub_client.py
index fe8a2af..2975ee8 100644
--- a/packages/meshbay-node/tests/test_hub_client.py
+++ b/packages/meshbay-node/tests/test_hub_client.py
@@ -2,7 +2,6 @@
 Tests for meshbay_node.hub_client — uses httpx.MockTransport to avoid network.
 """
 
-import base64
 import json
 import os
 import time
@@ -16,7 +15,7 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from cryptography.hazmat.primitives import serialization
 
-from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek
+
 from meshbay_node.hub_client import HubClient, HubConfig, HubSession
 from meshbay_node.keystore import NodeKeys
 
@@ -51,16 +50,15 @@ def hub_config(tmp_path):
     return HubConfig(
         hub_url="http://fake-hub",
         username="testuser",
-        password="testpass99",
         cache_dir=tmp_path,
     )
 
 
-def make_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600):
+def make_node_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600):
     now = int(time.time())
     return jwt.encode({
         "iss": hub_id, "sub": user_id, "pk_user": pk_user_b64,
-        "hub_id": hub_id, "jti": "test-jti",
+        "hub_id": hub_id, "jti": "test-jti", "scope": "node",
         "iat": now, "exp": now + ttl,
     }, sk_pem, algorithm="EdDSA")
 
@@ -71,14 +69,14 @@ def make_token(sk_pem, user_id, pk_user_b64, hub_id="fake-hub", ttl=3600):
 async def test_login_verifies_jwt_offline(hub_keys, node_keys, hub_config):
     sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
     user_id = "user-uuid-001"
-    token = make_token(sk_hub_pem, user_id, node_keys.pk_ed25519_b64)
+    token = make_node_token(sk_hub_pem, user_id, node_keys.pk_ed25519_b64)
 
     def handler(request):
         if request.url.path == "/v1/hub/pubkey":
             return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
-        if request.url.path == "/v1/users/login":
+        if request.url.path == "/v1/nodes/auth":
             return httpx.Response(200, json={
-                "access_token": token, "refresh_token": "rt-abc", "expires_in": 3600})
+                "access_token": token, "token_type": "bearer", "expires_in": 3600})
         return httpx.Response(404)
 
     transport = httpx.MockTransport(handler)
@@ -96,18 +94,18 @@ async def test_login_verifies_jwt_offline(hub_keys, node_keys, hub_config):
 @pytest.mark.asyncio
 async def test_login_rejects_missing_jti(hub_keys, node_keys, hub_config):
     sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
-    # Token without jti
     bad_token = jwt.encode({
         "iss": "fake-hub", "sub": "uid", "pk_user": node_keys.pk_ed25519_b64,
-        "hub_id": "fake-hub", "iat": int(time.time()), "exp": int(time.time()) + 3600,
+        "hub_id": "fake-hub", "scope": "node",
+        "iat": int(time.time()), "exp": int(time.time()) + 3600,
     }, sk_hub_pem, algorithm="EdDSA")
 
     def handler(request):
         if request.url.path == "/v1/hub/pubkey":
             return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
-        if request.url.path == "/v1/users/login":
+        if request.url.path == "/v1/nodes/auth":
             return httpx.Response(200, json={
-                "access_token": bad_token, "refresh_token": "rt", "expires_in": 3600})
+                "access_token": bad_token, "token_type": "bearer", "expires_in": 3600})
         return httpx.Response(404)
 
     transport = httpx.MockTransport(handler)
@@ -119,33 +117,16 @@ async def test_login_rejects_missing_jti(hub_keys, node_keys, hub_config):
 
 
 @pytest.mark.asyncio
-async def test_register_idempotent(hub_keys, node_keys, hub_config):
-    def handler(request):
-        if request.url.path == "/v1/users/register":
-            return httpx.Response(409, json={"detail": "Username already taken"})
-        return httpx.Response(404)
-
-    transport = httpx.MockTransport(handler)
-    client = HubClient(hub_config, node_keys)
-    client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
-
-    # Should not raise on 409
-    result = await client.register()
-    assert result == ""
-
-
-@pytest.mark.asyncio
 async def test_token_needs_refresh(hub_keys, node_keys, hub_config):
     sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
-    # Token expiring in 60s (< TOKEN_REFRESH_MARGIN of 300s)
-    short_token = make_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64, ttl=60)
+    short_token = make_node_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64, ttl=60)
 
     def handler(request):
         if request.url.path == "/v1/hub/pubkey":
             return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
-        if request.url.path == "/v1/users/login":
+        if request.url.path == "/v1/nodes/auth":
             return httpx.Response(200, json={
-                "access_token": short_token, "refresh_token": "rt", "expires_in": 60})
+                "access_token": short_token, "token_type": "bearer", "expires_in": 60})
         return httpx.Response(404)
 
     transport = httpx.MockTransport(handler)
@@ -157,37 +138,6 @@ async def test_token_needs_refresh(hub_keys, node_keys, hub_config):
 
 
 @pytest.mark.asyncio
-async def test_fetch_gek(hub_keys, node_keys, hub_config):
-    """Admin wraps GEK for this node; client fetches and unwraps."""
-    sk_hub, sk_hub_pem, pk_hub_pem = hub_keys
-    gek = generate_gek()
-
-    # Simulate admin wrapping GEK for this node
-    pk_x_raw = base64.b64decode(node_keys.pk_x25519_b64)
-    bundle = wrap_gek(gek, pk_x_raw)
-
-    token = make_token(sk_hub_pem, "uid", node_keys.pk_ed25519_b64)
-
-    def handler(request):
-        if request.url.path == "/v1/hub/pubkey":
-            return httpx.Response(200, json={"pk_hub_pem": pk_hub_pem.decode()})
-        if request.url.path == "/v1/users/login":
-            return httpx.Response(200, json={
-                "access_token": token, "refresh_token": "rt", "expires_in": 3600})
-        if "/v1/groups/" in request.url.path and request.url.path.endswith("/gek"):
-            return httpx.Response(200, json=bundle)
-        return httpx.Response(404)
-
-    transport = httpx.MockTransport(handler)
-    client = HubClient(hub_config, node_keys)
-    client._http = httpx.AsyncClient(transport=transport, base_url="http://fake-hub")
-
-    await client.login()
-    recovered = await client.fetch_gek("group-abc")
-    assert recovered == gek
-
-
-@pytest.mark.asyncio
 async def test_hub_pk_cached(hub_keys, node_keys, hub_config, tmp_path):
     _, _, pk_hub_pem = hub_keys
     call_count = {"n": 0}
diff --git a/packages/meshbay-node/tests/test_webrtc_transport.py b/packages/meshbay-node/tests/test_webrtc_transport.py
index b6664e8..693a68b 100644
--- a/packages/meshbay-node/tests/test_webrtc_transport.py
+++ b/packages/meshbay-node/tests/test_webrtc_transport.py
@@ -9,6 +9,8 @@ Uses local loopback (no STUN/ICE needed for localhost).
 
 import asyncio
 import base64
+import hashlib
+import hmac
 import os
 import struct
 import time
@@ -24,9 +26,13 @@ from meshbay_common import MNP_VERSION
 from meshbay_common.crypto import (
     generate_gek,
     pk_to_b64,
+    wrap_gek,
+    wrap_gek_aes,
+    unwrap_gek,
 )
 from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes
 from meshbay_common.protocol import MNP
+from meshbay_node.bundle_store import BundleStore
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.transport.webrtc_server import WebRTCTransport
 
@@ -42,6 +48,11 @@ def sk_hub():
 
 
 @pytest.fixture
+def sk_user():
+    return Ed25519PrivateKey.generate()
+
+
+@pytest.fixture
 def gek():
     return generate_gek()
 
@@ -60,7 +71,7 @@ def _hub_pk_pem(sk_hub):
         serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo)
 
 
-def _make_jwt(sk_hub, groups=None):
+def _make_jwt(sk_hub, groups=None, pk_user="test"):
     sk_pem = sk_hub.private_bytes(
         serialization.Encoding.PEM,
         serialization.PrivateFormat.PKCS8,
@@ -69,7 +80,7 @@ def _make_jwt(sk_hub, groups=None):
     now = int(time.time())
     return jwt.encode({
         "iss": "test-hub", "sub": "user-001",
-        "pk_user": "test", "hub_id": "test-hub",
+        "pk_user": pk_user, "hub_id": "test-hub",
         "jti": "test-jti-webrtc", "iat": now, "exp": now + 3600,
         "groups": groups or [],
     }, sk_pem, algorithm="EdDSA")
@@ -85,6 +96,108 @@ def _unpack(raw: bytes) -> dict:
     return msgpack.unpackb(raw[4:4 + length], raw=False)
 
 
+def _extract_dtls_fp(sdp: str) -> bytes:
+    for line in sdp.splitlines():
+        if line.startswith("a=fingerprint:sha-256 "):
+            return bytes.fromhex(line.split(" ", 1)[1].replace(":", ""))
+    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)
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE,
+        "v": MNP_VERSION,
+        "token": token,
+    }))
+    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)
+    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):
+    """Create a peer connection, perform handshake with GEK proof, return (pc, channel, queue)."""
+    pc = RTCPeerConnection()
+    q = asyncio.Queue()
+    buf = bytearray()
+    ch = pc.createDataChannel("mnp")
+    ready = asyncio.Event()
+
+    @ch.on("open")
+    def on_open():
+        ready.set()
+
+    @ch.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            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))
+
+    offer = await pc.createOffer()
+    await pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(pc.localDescription.sdp, peer_id)
+    await pc.setRemoteDescription(RTCSessionDescription(sdp=answer_sdp, type="answer"))
+    await asyncio.wait_for(ready.wait(), timeout=5.0)
+
+    pk_user = "test"
+    if sk_user:
+        pk_user = base64.b64encode(
+            sk_user.public_key().public_bytes(
+                serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+        ).decode()
+
+    sk_h_pem = sk_hub.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    )
+    now = int(time.time())
+    token = jwt.encode({
+        "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": [],
+    }, 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)
+    assert msg["type"] == MNP.HANDSHAKE_ACK
+    return pc, ch, q
+
+
 @pytest.mark.asyncio
 async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir):
     """WebRTC DataChannel: browser sends MNP handshake, node responds with handshake_ack."""
@@ -120,15 +233,8 @@ async def test_webrtc_datachannel_handshake(sk_node, sk_hub, gek, shared_dir):
 
     await asyncio.sleep(0.5)
 
-    token = _make_jwt(sk_hub)
-    channel.send(_pack({
-        "type": MNP.HANDSHAKE,
-        "v": MNP_VERSION,
-        "token": token,
-    }))
-
-    msg = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert msg["type"] == MNP.HANDSHAKE_ACK
+    msg = await _handshake_with_gek_proof(channel, received, sk_hub, gek,
+                                          browser_pc=browser_pc)
     assert msg["v"] == MNP_VERSION
     assert "node_pk" in msg
 
@@ -153,15 +259,11 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir
     received = asyncio.Queue()
 
     channel = browser_pc.createDataChannel("mnp")
+    channel_ready = asyncio.Event()
 
     @channel.on("open")
     def on_open():
-        token = _make_jwt(sk_hub)
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE,
-            "v": MNP_VERSION,
-            "token": token,
-        }))
+        channel_ready.set()
 
     buf = bytearray()
 
@@ -186,8 +288,11 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir
     await browser_pc.setRemoteDescription(
         RTCSessionDescription(sdp=answer_sdp, type="answer"))
 
-    # 1) Handshake ack
-    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    await asyncio.wait_for(channel_ready.wait(), timeout=5.0)
+
+    # 1) Handshake with GEK proof
+    ack = await _handshake_with_gek_proof(channel, received, sk_hub, gek,
+                                          browser_pc=browser_pc)
     assert ack["type"] == MNP.HANDSHAKE_ACK
 
     # 2) Request index
@@ -220,13 +325,6 @@ async def test_webrtc_datachannel_file_transfer(sk_node, sk_hub, gek, shared_dir
     original = (shared_dir / "test.bin").read_bytes()
     assert plaintext == original
 
-    # 5) Request GEK over DataChannel
-    channel.send(_pack({"type": MNP.GEK_REQUEST, "v": MNP_VERSION}))
-    gek_msg = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert gek_msg["type"] == MNP.GEK_RESPONSE
-    received_gek = base64.b64decode(gek_msg["gek_b64"])
-    assert received_gek == gek
-
     await browser_pc.close()
     await transport.close_all()
 
@@ -342,44 +440,8 @@ async def test_webrtc_chat_send_and_history(sk_node, sk_hub, gek, shared_dir, tm
     )
     transport._ctx["chat_store"] = chat_store
 
-    browser_pc = RTCPeerConnection()
-    received = asyncio.Queue()
-    buf = bytearray()
-
-    channel = browser_pc.createDataChannel("mnp")
-
-    @channel.on("open")
-    def on_open():
-        token = _make_jwt(sk_hub)
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE,
-            "v": MNP_VERSION,
-            "token": token,
-        }))
-
-    @channel.on("message")
-    def on_msg(message):
-        if isinstance(message, str):
-            message = message.encode()
-        buf.extend(message)
-        while len(buf) >= 4:
-            length = struct.unpack(">I", buf[:4])[0]
-            if len(buf) < 4 + length:
-                break
-            msg_bytes = bytes(buf[4:4 + length])
-            del buf[:4 + length]
-            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
-
-    offer = await browser_pc.createOffer()
-    await browser_pc.setLocalDescription(offer)
-
-    answer_sdp, _ = await transport.handle_offer(
-        browser_pc.localDescription.sdp, "peer-chat")
-    await browser_pc.setRemoteDescription(
-        RTCSessionDescription(sdp=answer_sdp, type="answer"))
-
-    ack = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert ack["type"] == MNP.HANDSHAKE_ACK
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-chat")
 
     channel.send(_pack({
         "type": MNP.CHAT_MESSAGE,
@@ -421,41 +483,8 @@ async def test_webrtc_chat_history_no_store(sk_node, sk_hub, gek, shared_dir):
         stun_servers=[],
     )
 
-    browser_pc = RTCPeerConnection()
-    received = asyncio.Queue()
-    buf = bytearray()
-
-    channel = browser_pc.createDataChannel("mnp")
-
-    @channel.on("open")
-    def on_open():
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
-            "token": _make_jwt(sk_hub),
-        }))
-
-    @channel.on("message")
-    def on_msg(message):
-        if isinstance(message, str):
-            message = message.encode()
-        buf.extend(message)
-        while len(buf) >= 4:
-            length = struct.unpack(">I", buf[:4])[0]
-            if len(buf) < 4 + length:
-                break
-            msg_bytes = bytes(buf[4:4 + length])
-            del buf[:4 + length]
-            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
-
-    offer = await browser_pc.createOffer()
-    await browser_pc.setLocalDescription(offer)
-    answer_sdp, _ = await transport.handle_offer(
-        browser_pc.localDescription.sdp, "peer-no-store")
-    await browser_pc.setRemoteDescription(
-        RTCSessionDescription(sdp=answer_sdp, type="answer"))
-
-    ack = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert ack["type"] == MNP.HANDSHAKE_ACK
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-no-store")
 
     channel.send(_pack({
         "type": MNP.CHAT_HISTORY, "v": MNP_VERSION, "since": 0, "limit": 50,
@@ -487,57 +516,8 @@ async def test_webrtc_chat_broadcast(sk_node, sk_hub, gek, shared_dir, tmp_path)
     )
     transport._ctx["chat_store"] = chat_store
 
-    async def _connect_peer(peer_id, jwt_sub, groups=None):
-        pc = RTCPeerConnection()
-        q = asyncio.Queue()
-        b = bytearray()
-        ch = pc.createDataChannel("mnp")
-
-        sk_h_pem = sk_hub.private_bytes(
-            serialization.Encoding.PEM,
-            serialization.PrivateFormat.PKCS8,
-            serialization.NoEncryption(),
-        )
-        now = int(time.time())
-        token = jwt.encode({
-            "iss": "test-hub", "sub": jwt_sub,
-            "pk_user": "test", "hub_id": "test-hub",
-            "jti": f"jti-{peer_id}", "iat": now, "exp": now + 3600,
-            "groups": groups or [],
-        }, sk_h_pem, algorithm="EdDSA")
-
-        @ch.on("open")
-        def on_open():
-            ch.send(_pack({
-                "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
-            }))
-
-        @ch.on("message")
-        def on_msg(message):
-            if isinstance(message, str):
-                message = message.encode()
-            b.extend(message)
-            while len(b) >= 4:
-                length = struct.unpack(">I", b[:4])[0]
-                if len(b) < 4 + length:
-                    break
-                msg_bytes = bytes(b[4:4 + length])
-                del b[:4 + length]
-                q.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
-
-        offer = await pc.createOffer()
-        await pc.setLocalDescription(offer)
-        answer_sdp, _ = await transport.handle_offer(
-            pc.localDescription.sdp, peer_id)
-        await pc.setRemoteDescription(
-            RTCSessionDescription(sdp=answer_sdp, type="answer"))
-
-        ack = await asyncio.wait_for(q.get(), timeout=5.0)
-        assert ack["type"] == MNP.HANDSHAKE_ACK
-        return pc, ch, q
-
-    pc_a, ch_a, q_a = await _connect_peer("peer-A", "user-A")
-    pc_b, ch_b, q_b = await _connect_peer("peer-B", "user-B")
+    pc_a, ch_a, q_a = await _setup_peer(transport, sk_hub, gek, "peer-A", "user-A")
+    pc_b, ch_b, q_b = await _setup_peer(transport, sk_hub, gek, "peer-B", "user-B")
 
     ch_a.send(_pack({
         "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "payload": "hi from A",
@@ -617,18 +597,509 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
         stun_servers=[],
     )
 
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-cleanup")
+
+    assert "user-001" in transport._ctx["_peers"]
+    assert transport.active_peers == 1
+
+    await transport.close_peer("peer-cleanup")
+
+    assert "user-001" not in transport._ctx["_peers"]
+    assert transport.active_peers == 0
+
+    await browser_pc.close()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: stream_segment for non-existent file returns error."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-stream")
+
+    channel.send(_pack({
+        "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION,
+        "file_id": "nonexistent-file-id",
+        "segment_index": 0, "segment_duration": 4,
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "not found" in msg["detail"].lower()
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_wrong_gek_proof_rejected(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: wrong GEK proof is rejected — hub admin can't fake membership."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
     browser_pc = RTCPeerConnection()
     received = asyncio.Queue()
-    buf = bytearray()
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        received.put_nowait(_unpack(message))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-fake")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    await asyncio.sleep(0.5)
+
+    token = _make_jwt(sk_hub)
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+    }))
+
+    challenge = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE
+
+    fake_gek = os.urandom(32)
+    nonce = base64.b64decode(challenge["nonce"])
+    offer_fp = _extract_dtls_fp(browser_pc.localDescription.sdp)
+    answer_fp = _extract_dtls_fp(browser_pc.remoteDescription.sdp)
+    bad_proof = hmac.new(fake_gek, nonce + offer_fp + answer_fp, hashlib.sha256).digest()
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE_RESPONSE, "v": MNP_VERSION,
+        "proof": base64.b64encode(bad_proof).decode(),
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "GEK proof failed" in msg["detail"]
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_dtls_channel_binding_detects_mitm(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC: DTLS channel binding detects fingerprint substitution (simulated MitM)."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        received.put_nowait(_unpack(message))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-mitm")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    await asyncio.sleep(0.5)
+
+    token = _make_jwt(sk_hub)
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
+    }))
+
+    challenge = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert challenge["type"] == MNP.HANDSHAKE_CHALLENGE
+
+    nonce = base64.b64decode(challenge["nonce"])
+    # Correct GEK but fake fingerprints — simulates MitM substituting DTLS certs
+    fake_fp = os.urandom(32)
+    proof = hmac.new(gek, nonce + fake_fp + fake_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)
+    assert msg["type"] == "error"
+    assert "GEK proof failed" in msg["detail"]
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_admin_challenge_response(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: admin file delete requires Ed25519 challenge-response."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    sk_admin = Ed25519PrivateKey.generate()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["admin_pk_ed25519"] = sk_admin.public_key()
+    transport._ctx["node_user_id"] = "user-001"
+
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-admin")
+
+    entry = indexer.index.entries[0]
+    channel.send(_pack({
+        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
+    }))
+
+    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
+    assert challenge_msg["file_id"] == entry.id
+
+    challenge = base64.b64decode(challenge_msg["challenge"])
+    signature = sk_admin.sign(challenge)
+    channel.send(_pack({
+        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
+        "file_id": entry.id,
+        "signature": base64.b64encode(signature).decode(),
+    }))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.FILE_DELETE_ACK
+    assert ack["file_id"] == entry.id
+
+    assert indexer.index.get_entry(entry.id) is None
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_admin_bad_signature_rejected(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: wrong Ed25519 signature is rejected — hub can't fake admin."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    sk_admin = Ed25519PrivateKey.generate()
+    sk_attacker = Ed25519PrivateKey.generate()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["admin_pk_ed25519"] = sk_admin.public_key()
+    transport._ctx["node_user_id"] = "user-001"
+
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-attacker")
+
+    entry = indexer.index.entries[0]
+    channel.send(_pack({
+        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
+    }))
+
+    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
+
+    challenge = base64.b64decode(challenge_msg["challenge"])
+    bad_sig = sk_attacker.sign(challenge)
+    channel.send(_pack({
+        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
+        "file_id": entry.id,
+        "signature": base64.b64encode(bad_sig).decode(),
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "signature" in msg["detail"].lower() or "verification" in msg["detail"].lower()
+
+    assert indexer.index.get_entry(entry.id) is not None
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir):
+    """WebRTC DataChannel: stream_request for non-existent file returns error."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-mse")
+
+    channel.send(_pack({
+        "type": MNP.STREAM_REQUEST, "v": MNP_VERSION,
+        "file_id": "nonexistent-file-id",
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "not found" in msg["detail"].lower()
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_uploader_delete_requires_challenge(sk_node, sk_hub, gek, shared_dir):
+    """Uploader must prove Ed25519 key ownership to delete — no uploader shortcut."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    sk_uploader = Ed25519PrivateKey.generate()
+    pk_uploader_b64 = base64.b64encode(
+        sk_uploader.public_key().public_bytes(
+            serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    ).decode()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    # No admin_pk configured — only uploader_pk should authorize deletion
+
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-uploader-del", sk_user=sk_uploader)
+
+    # Tag an existing entry with the uploader's public key
+    entry = indexer.index.entries[0]
+    entry.uploader_id = "user-001"
+    entry.uploader_pk = pk_uploader_b64
+
+    # Request deletion — should get a challenge (no shortcut)
+    channel.send(_pack({
+        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
+    }))
+
+    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
+    assert challenge_msg["file_id"] == entry.id
+
+    # Sign with uploader's Ed25519 key
+    challenge = base64.b64decode(challenge_msg["challenge"])
+    signature = sk_uploader.sign(challenge)
+    channel.send(_pack({
+        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
+        "file_id": entry.id,
+        "signature": base64.b64encode(signature).decode(),
+    }))
+
+    ack = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert ack["type"] == MNP.FILE_DELETE_ACK
+    assert ack["file_id"] == entry.id
+
+    # Verify file was removed from index
+    assert indexer.index.get_entry(entry.id) is None
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_uploader_impersonation_blocked(sk_node, sk_hub, gek, shared_dir):
+    """Hub-forged JWT with same sub cannot delete — wrong Ed25519 key is rejected."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    # User A uploaded the file
+    sk_user_a = Ed25519PrivateKey.generate()
+    pk_a_b64 = base64.b64encode(
+        sk_user_a.public_key().public_bytes(
+            serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    ).decode()
+
+    # User B is the attacker (different Ed25519 key, but hub forges JWT with same sub)
+    sk_user_b = Ed25519PrivateKey.generate()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    # No admin_pk — only uploader_pk matters
+
+    # Tag entry with user A's public key
+    entry = indexer.index.entries[0]
+    entry.uploader_id = "user-001"
+    entry.uploader_pk = pk_a_b64
+
+    # Connect as user B (same jwt_sub "user-001" via hub forgery, but B's Ed25519 key)
+    browser_pc, channel, received = await _setup_peer(
+        transport, sk_hub, gek, "peer-impersonator",
+        jwt_sub="user-001", sk_user=sk_user_b)
+
+    # Request deletion — should get a challenge
+    channel.send(_pack({
+        "type": MNP.FILE_DELETE, "v": MNP_VERSION, "file_id": entry.id,
+    }))
+
+    challenge_msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert challenge_msg["type"] == MNP.ADMIN_CHALLENGE
+
+    # Sign with user B's key (wrong key)
+    challenge = base64.b64decode(challenge_msg["challenge"])
+    bad_sig = sk_user_b.sign(challenge)
+    channel.send(_pack({
+        "type": MNP.ADMIN_RESPONSE, "v": MNP_VERSION,
+        "file_id": entry.id,
+        "signature": base64.b64encode(bad_sig).decode(),
+    }))
+
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == "error"
+    assert "verification" in msg["detail"].lower() or "signature" in msg["detail"].lower()
+
+    # File must still exist in the index
+    assert indexer.index.get_entry(entry.id) is not None
+
+    await browser_pc.close()
+    await transport.close_all()
+
+
+# ── GEK bundle P2P exchange tests ──────────────────────────────────────────
+
+
+@pytest.fixture
+def x25519_keypair():
+    from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
+    sk = X25519PrivateKey.generate()
+    sk_raw = sk.private_bytes(
+        serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
+        serialization.NoEncryption())
+    pk_raw = sk.public_key().public_bytes(
+        serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    return sk_raw, pk_raw
+
+
+@pytest.mark.asyncio
+async def test_gek_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir,
+                                          tmp_path, x25519_keypair):
+    """GEK bundle stored on node via DataChannel, then fetched during handshake."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
 
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["bundle_store"] = bundle_store
+
+    # Connect as admin and store a GEK bundle for user-002
+    pc_admin, ch_admin, q_admin = await _setup_peer(
+        transport, sk_hub, gek, "peer-admin")
+
+    sk_x_raw, pk_x_raw = x25519_keypair
+    bundle = wrap_gek(gek, pk_x_raw)
+
+    ch_admin.send(_pack({
+        "type": MNP.GEK_BUNDLE_STORE,
+        "v": MNP_VERSION,
+        "user_id": "user-002",
+        "group_id": "g",
+        "pk_eph_b64": bundle["pk_eph_b64"],
+        "nonce_b64": bundle["nonce_b64"],
+        "wrapped_b64": bundle["wrapped_b64"],
+    }))
+    ack = await asyncio.wait_for(q_admin.get(), timeout=5.0)
+    assert ack["type"] == "ack"
+    assert ack["detail"] == "gek_bundle_stored"
+
+    # Verify bundle was persisted
+    stored = await bundle_store.fetch("g", "user-002")
+    assert stored is not None
+    assert stored["pk_eph_b64"] == bundle["pk_eph_b64"]
+
+    # Unwrap to verify it's correct
+    recovered = unwrap_gek(stored, sk_x_raw, pk_x_raw)
+    assert recovered == gek
+
+    await bundle_store.close()
+    await pc_admin.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_gek_bundle_fetch_during_handshake(sk_node, sk_hub, gek, shared_dir,
+                                                  tmp_path, x25519_keypair):
+    """Browser fetches GEK bundle from node during the handshake challenge window."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    sk_x_raw, pk_x_raw = x25519_keypair
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
+
+    # Pre-populate a bundle for user-001 in group "g"
+    bundle = wrap_gek(gek, pk_x_raw)
+    await bundle_store.store("g", "user-001",
+                             bundle["pk_eph_b64"], bundle["nonce_b64"],
+                             bundle["wrapped_b64"])
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["bundle_store"] = bundle_store
+
+    # Connect manually: handshake → challenge → gek_bundle_fetch → response
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
     channel = browser_pc.createDataChannel("mnp")
+    ready = asyncio.Event()
 
     @channel.on("open")
     def on_open():
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
-            "token": _make_jwt(sk_hub),
-        }))
+        ready.set()
 
     @channel.on("message")
     def on_msg(message):
@@ -646,49 +1117,162 @@ async def test_webrtc_peer_cleanup_on_close(sk_node, sk_hub, gek, shared_dir):
     offer = await browser_pc.createOffer()
     await browser_pc.setLocalDescription(offer)
     answer_sdp, _ = await transport.handle_offer(
-        browser_pc.localDescription.sdp, "peer-cleanup")
+        browser_pc.localDescription.sdp, "peer-fetch")
     await browser_pc.setRemoteDescription(
         RTCSessionDescription(sdp=answer_sdp, type="answer"))
+    await asyncio.wait_for(ready.wait(), timeout=5.0)
+
+    # Step 1: Send handshake with group_id so _pending_group is set
+    token = _make_jwt(sk_hub, groups=["g"])
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+    }))
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
+
+    # Step 2: Fetch GEK bundle from node (during challenge window)
+    channel.send(_pack({"type": MNP.GEK_BUNDLE_FETCH, "v": MNP_VERSION}))
+    bundle_resp = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert bundle_resp["type"] == MNP.GEK_BUNDLE_RESP
+    assert bundle_resp["found"] is True
+
+    # Step 3: Unwrap GEK and compute HMAC proof
+    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()
+
+    # Step 4: Complete handshake
+    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)
     assert ack["type"] == MNP.HANDSHAKE_ACK
 
-    assert "user-001" in transport._ctx["_peers"]
-    assert transport.active_peers == 1
+    await bundle_store.close()
+    await browser_pc.close()
+    await transport.close_all()
 
-    await transport.close_peer("peer-cleanup")
 
-    assert "user-001" not in transport._ctx["_peers"]
-    assert transport.active_peers == 0
+# ── Keypair bundle P2P tests ─────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_keypair_bundle_store_and_fetch(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """Keypair bundle stored on node, then fetched during handshake window."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
 
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["bundle_store"] = bundle_store
+
+    # Connect and store a keypair bundle
+    pc1, ch1, q1 = await _setup_peer(transport, sk_hub, gek, "peer-kp-store")
+    ch1.send(_pack({
+        "type": MNP.KEYPAIR_BUNDLE_STORE,
+        "v": MNP_VERSION,
+        "bundle_enc": "encrypted-keypair-data-base64",
+    }))
+    ack = await asyncio.wait_for(q1.get(), timeout=5.0)
+    assert ack["type"] == "ack"
+    assert ack["detail"] == "keypair_bundle_stored"
+
+    # Verify in DB
+    stored = await bundle_store.fetch_keypair("user-001")
+    assert stored == "encrypted-keypair-data-base64"
+
+    await pc1.close()
+
+    # New connection: fetch during handshake window
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    buf = bytearray()
+    channel = browser_pc.createDataChannel("mnp")
+    ready = asyncio.Event()
+
+    @channel.on("open")
+    def on_open():
+        ready.set()
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        buf.extend(message)
+        while len(buf) >= 4:
+            length = struct.unpack(">I", buf[:4])[0]
+            if len(buf) < 4 + length:
+                break
+            msg_bytes = bytes(buf[4:4 + length])
+            del buf[:4 + length]
+            received.put_nowait(msgpack.unpackb(msg_bytes, raw=False))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-kp-fetch")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+    await asyncio.wait_for(ready.wait(), timeout=5.0)
+
+    token = _make_jwt(sk_hub, groups=["g"])
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+    }))
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
+
+    # Fetch keypair bundle during challenge window
+    channel.send(_pack({"type": MNP.KEYPAIR_BUNDLE_FETCH, "v": MNP_VERSION}))
+    kp_resp = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert kp_resp["type"] == MNP.KEYPAIR_BUNDLE_RESP
+    assert kp_resp["found"] is True
+    assert kp_resp["bundle_enc"] == "encrypted-keypair-data-base64"
+
+    await bundle_store.close()
     await browser_pc.close()
+    await transport.close_all()
 
 
 @pytest.mark.asyncio
-async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_dir):
-    """WebRTC DataChannel: stream_segment for non-existent file returns error."""
+async def test_keypair_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """Keypair bundle fetch returns found=false when no bundle exists."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
     indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
+
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
         shared_root=shared_dir, index=indexer.index,
         stun_servers=[],
     )
+    transport._ctx["bundle_store"] = bundle_store
 
     browser_pc = RTCPeerConnection()
     received = asyncio.Queue()
     buf = bytearray()
-
     channel = browser_pc.createDataChannel("mnp")
+    ready = asyncio.Event()
 
     @channel.on("open")
     def on_open():
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
-            "token": _make_jwt(sk_hub),
-        }))
+        ready.set()
 
     @channel.on("message")
     def on_msg(message):
@@ -706,52 +1290,150 @@ async def test_webrtc_stream_segment_missing_file(sk_node, sk_hub, gek, shared_d
     offer = await browser_pc.createOffer()
     await browser_pc.setLocalDescription(offer)
     answer_sdp, _ = await transport.handle_offer(
-        browser_pc.localDescription.sdp, "peer-stream")
+        browser_pc.localDescription.sdp, "peer-kp-none")
     await browser_pc.setRemoteDescription(
         RTCSessionDescription(sdp=answer_sdp, type="answer"))
+    await asyncio.wait_for(ready.wait(), timeout=5.0)
 
-    ack = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert ack["type"] == MNP.HANDSHAKE_ACK
+    token = _make_jwt(sk_hub, groups=["g"])
+    channel.send(_pack({
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
+    }))
+    msg = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
 
+    channel.send(_pack({"type": MNP.KEYPAIR_BUNDLE_FETCH, "v": MNP_VERSION}))
+    resp = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert resp["type"] == MNP.KEYPAIR_BUNDLE_RESP
+    assert resp["found"] is False
+
+    await bundle_store.close()
+    await browser_pc.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_gek_auto_activate_on_node_bundle_store(sk_node, sk_hub, gek, shared_dir,
+                                                       tmp_path, x25519_keypair):
+    """Storing the node operator's GEK bundle auto-activates GEK (AES variant)."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
+    await indexer.initial_scan()
+
+    sk_x_raw, pk_x_raw = x25519_keypair
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
+
+    new_gek = generate_gek()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+    transport._ctx["bundle_store"] = bundle_store
+    transport._ctx["node_user_id"] = "node-operator"
+    transport._ctx["sk_x25519_raw"] = sk_x_raw
+    transport._ctx["pk_x25519_raw"] = pk_x_raw
+    transport._ctx["pk_x25519_b64"] = base64.b64encode(pk_x_raw).decode()
+
+    pc_admin, ch_admin, q_admin = await _setup_peer(
+        transport, sk_hub, gek, "peer-setup-admin")
+
+    # Store GEK bundle wrapped with AES-GCM (browser-compatible)
+    node_bundle = wrap_gek_aes(new_gek, pk_x_raw)
+    ch_admin.send(_pack({
+        "type": MNP.GEK_BUNDLE_STORE,
+        "v": MNP_VERSION,
+        "user_id": "node-operator",
+        "group_id": "g",
+        "pk_eph_b64": node_bundle["pk_eph_b64"],
+        "nonce_b64": node_bundle["nonce_b64"],
+        "wrapped_b64": node_bundle["wrapped_b64"],
+    }))
+    ack = await asyncio.wait_for(q_admin.get(), timeout=5.0)
+    assert ack["type"] == "ack"
+
+    await asyncio.sleep(0.2)
+
+    assert transport._ctx.get("gek") == new_gek
+
+    await bundle_store.close()
+    await pc_admin.close()
+    await transport.close_all()
+
+
+@pytest.mark.asyncio
+async def test_webrtc_no_gek_connection_refused(sk_node, sk_hub, shared_dir):
+    """WebRTC DataChannel: connection refused when GEK is not initialized."""
+    hub_pk_pem = _hub_pk_pem(sk_hub)
+    indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=None)
+    await indexer.initial_scan()
+
+    transport = WebRTCTransport(
+        sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=None,
+        shared_root=shared_dir, index=indexer.index,
+        stun_servers=[],
+    )
+
+    browser_pc = RTCPeerConnection()
+    received = asyncio.Queue()
+    channel = browser_pc.createDataChannel("mnp")
+
+    @channel.on("message")
+    def on_msg(message):
+        if isinstance(message, str):
+            message = message.encode()
+        received.put_nowait(_unpack(message))
+
+    offer = await browser_pc.createOffer()
+    await browser_pc.setLocalDescription(offer)
+    answer_sdp, _ = await transport.handle_offer(
+        browser_pc.localDescription.sdp, "peer-no-gek")
+    await browser_pc.setRemoteDescription(
+        RTCSessionDescription(sdp=answer_sdp, type="answer"))
+
+    await asyncio.sleep(0.5)
+
+    token = _make_jwt(sk_hub)
     channel.send(_pack({
-        "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION,
-        "file_id": "nonexistent-file-id",
-        "segment_index": 0, "segment_duration": 4,
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token,
     }))
 
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
     assert msg["type"] == "error"
-    assert "not found" in msg["detail"].lower()
+    assert "not initialized" in msg["detail"].lower()
 
     await browser_pc.close()
     await transport.close_all()
 
 
 @pytest.mark.asyncio
-async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_dir):
-    """WebRTC DataChannel: stream_request for non-existent file returns error."""
+async def test_gek_bundle_fetch_not_found(sk_node, sk_hub, gek, shared_dir, tmp_path):
+    """GEK bundle fetch returns found=false when no bundle exists."""
     hub_pk_pem = _hub_pk_pem(sk_hub)
     indexer = DirectoryIndexer(root=shared_dir, group_id="g", sk_node=sk_node, gek=gek)
     await indexer.initial_scan()
 
+    bundle_store = BundleStore(db_path=tmp_path / "bundles.db")
+    await bundle_store.open()
+
     transport = WebRTCTransport(
         sk_node=sk_node, hub_pk_pem=hub_pk_pem, gek=gek,
         shared_root=shared_dir, index=indexer.index,
         stun_servers=[],
     )
+    transport._ctx["bundle_store"] = bundle_store
 
     browser_pc = RTCPeerConnection()
     received = asyncio.Queue()
     buf = bytearray()
-
     channel = browser_pc.createDataChannel("mnp")
+    ready = asyncio.Event()
 
     @channel.on("open")
     def on_open():
-        channel.send(_pack({
-            "type": MNP.HANDSHAKE, "v": MNP_VERSION,
-            "token": _make_jwt(sk_hub),
-        }))
+        ready.set()
 
     @channel.on("message")
     def on_msg(message):
@@ -769,21 +1451,23 @@ async def test_webrtc_stream_request_missing_file(sk_node, sk_hub, gek, shared_d
     offer = await browser_pc.createOffer()
     await browser_pc.setLocalDescription(offer)
     answer_sdp, _ = await transport.handle_offer(
-        browser_pc.localDescription.sdp, "peer-mse")
+        browser_pc.localDescription.sdp, "peer-nofound")
     await browser_pc.setRemoteDescription(
         RTCSessionDescription(sdp=answer_sdp, type="answer"))
+    await asyncio.wait_for(ready.wait(), timeout=5.0)
 
-    ack = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert ack["type"] == MNP.HANDSHAKE_ACK
-
+    token = _make_jwt(sk_hub, groups=["g"])
     channel.send(_pack({
-        "type": MNP.STREAM_REQUEST, "v": MNP_VERSION,
-        "file_id": "nonexistent-file-id",
+        "type": MNP.HANDSHAKE, "v": MNP_VERSION, "token": token, "group_id": "g",
     }))
-
     msg = await asyncio.wait_for(received.get(), timeout=5.0)
-    assert msg["type"] == "error"
-    assert "not found" in msg["detail"].lower()
+    assert msg["type"] == MNP.HANDSHAKE_CHALLENGE
+
+    channel.send(_pack({"type": MNP.GEK_BUNDLE_FETCH, "v": MNP_VERSION}))
+    resp = await asyncio.wait_for(received.get(), timeout=5.0)
+    assert resp["type"] == MNP.GEK_BUNDLE_RESP
+    assert resp["found"] is False
 
+    await bundle_store.close()
     await browser_pc.close()
     await transport.close_all()