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>

10 files changed, 1687 insertions, 421 deletions

diff --git a/packages/meshbay-node/pyproject.toml b/packages/meshbay-node/pyproject.toml
index 592de54..864fd04 100644
--- a/packages/meshbay-node/pyproject.toml
+++ b/packages/meshbay-node/pyproject.toml
@@ -17,6 +17,7 @@ dependencies = [
     "aioquic>=1.0",             # QUIC transport (MNP v2) — implemented in Phase 5
     "websockets>=12.0",         # hub→node revocation push
     "aiortc>=1.9",               # WebRTC DataChannel for browser P2P (Phase 9)
+    "aiosqlite>=0.20",           # async SQLite for chat, audit, bundle stores
 ]
 
 [project.optional-dependencies]
diff --git a/packages/meshbay-node/src/meshbay_node/bundle_store.py b/packages/meshbay-node/src/meshbay_node/bundle_store.py
new file mode 100644
index 0000000..e7c6981
--- /dev/null
+++ b/packages/meshbay-node/src/meshbay_node/bundle_store.py
@@ -0,0 +1,105 @@
+"""
+Bundle store — SQLite-backed storage for GEK bundles and keypair bundles.
+
+GEK bundles: ECIES-wrapped GEK targeted at a specific user's X25519 key.
+Keypair bundles: AES-GCM encrypted (Ed25519 + X25519) private keys, encrypted
+with the user's password-derived bundle_key. Opaque to the node.
+
+Both are stored and served over the P2P DataChannel during MNP handshake.
+"""
+
+import logging
+from pathlib import Path
+
+import aiosqlite
+
+log = logging.getLogger(__name__)
+
+_SCHEMA_GEK = """\
+CREATE TABLE IF NOT EXISTS gek_bundles (
+    group_id    TEXT NOT NULL,
+    user_id     TEXT NOT NULL,
+    pk_eph_b64  TEXT NOT NULL,
+    nonce_b64   TEXT NOT NULL,
+    wrapped_b64 TEXT NOT NULL,
+    stored_at   TEXT NOT NULL DEFAULT (datetime('now')),
+    PRIMARY KEY (group_id, user_id)
+);
+"""
+
+_SCHEMA_KEYPAIR = """\
+CREATE TABLE IF NOT EXISTS keypair_bundles (
+    user_id     TEXT PRIMARY KEY,
+    bundle_enc  TEXT NOT NULL,
+    stored_at   TEXT NOT NULL DEFAULT (datetime('now'))
+);
+"""
+
+
+class BundleStore:
+    def __init__(self, db_path: Path):
+        self._db_path = db_path
+        self._db: aiosqlite.Connection | None = None
+
+    async def open(self) -> None:
+        self._db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._db = await aiosqlite.connect(str(self._db_path))
+        await self._db.execute(_SCHEMA_GEK)
+        await self._db.execute(_SCHEMA_KEYPAIR)
+        await self._db.commit()
+
+    async def store(
+        self,
+        group_id: str,
+        user_id: str,
+        pk_eph_b64: str,
+        nonce_b64: str,
+        wrapped_b64: str,
+    ) -> None:
+        assert self._db
+        await self._db.execute(
+            "INSERT OR REPLACE INTO gek_bundles "
+            "(group_id, user_id, pk_eph_b64, nonce_b64, wrapped_b64, stored_at) "
+            "VALUES (?, ?, ?, ?, ?, datetime('now'))",
+            (group_id, user_id, pk_eph_b64, nonce_b64, wrapped_b64),
+        )
+        await self._db.commit()
+
+    async def fetch(self, group_id: str, user_id: str) -> dict | None:
+        assert self._db
+        async with self._db.execute(
+            "SELECT pk_eph_b64, nonce_b64, wrapped_b64 FROM gek_bundles "
+            "WHERE group_id = ? AND user_id = ?",
+            (group_id, user_id),
+        ) as cursor:
+            row = await cursor.fetchone()
+            if not row:
+                return None
+            return {
+                "pk_eph_b64": row[0],
+                "nonce_b64": row[1],
+                "wrapped_b64": row[2],
+            }
+
+    async def store_keypair(self, user_id: str, bundle_enc: str) -> None:
+        assert self._db
+        await self._db.execute(
+            "INSERT OR REPLACE INTO keypair_bundles "
+            "(user_id, bundle_enc, stored_at) VALUES (?, ?, datetime('now'))",
+            (user_id, bundle_enc),
+        )
+        await self._db.commit()
+
+    async def fetch_keypair(self, user_id: str) -> str | None:
+        assert self._db
+        async with self._db.execute(
+            "SELECT bundle_enc FROM keypair_bundles WHERE user_id = ?",
+            (user_id,),
+        ) as cursor:
+            row = await cursor.fetchone()
+            return row[0] if row else None
+
+    async def close(self) -> None:
+        if self._db:
+            await self._db.close()
+            self._db = None
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py
index 9e6a391..a7a0785 100644
--- a/packages/meshbay-node/src/meshbay_node/config.py
+++ b/packages/meshbay-node/src/meshbay_node/config.py
@@ -52,6 +52,11 @@ visibility  = "public"
 [keystore]
 # unlock_file = "~/.config/meshbay/unlock.key"
 # or set MESHBAY_UNLOCK_KEY env var
+
+# Node sovereignty: pin the operator's Ed25519 public key (base64, 32 bytes raw).
+# Admin operations (file delete) require cryptographic proof of this key.
+# Auto-pinned on first startup from the node operator's keystore.
+# admin_pk_ed25519 = "base64-encoded-32-bytes"
 """
 
 
@@ -59,7 +64,6 @@ visibility  = "public"
 class HubConfig:
     url:      str = "https://meshbay.org"
     username: str = ""
-    password: str = ""   # loaded from keystore or env; never written to TOML
 
 
 @dataclass
@@ -94,6 +98,7 @@ class Config:
     groups:   list[GroupConfig] = field(default_factory=list)
     keystore: KeystoreConfig = field(default_factory=KeystoreConfig)
     data_dir: Path = field(default_factory=lambda: Path.home() / ".local" / "share" / "meshbay")
+    admin_pk_ed25519: str = ""  # base64 raw Ed25519 public key pinned locally
 
     # Back-compat: single-group access
     @property
@@ -144,6 +149,9 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
         if "data_dir" in raw:
             cfg.data_dir = Path(raw["data_dir"]).expanduser().resolve()
 
+        if "admin_pk_ed25519" in raw:
+            cfg.admin_pk_ed25519 = raw["admin_pk_ed25519"]
+
         ks = raw.get("keystore", {})
         if "path" in ks:
             cfg.keystore.path = Path(ks["path"]).expanduser()
@@ -155,8 +163,6 @@ def load_config(path: Path = DEFAULT_CONFIG_PATH) -> Config:
         cfg.hub.url = url
     if user := os.environ.get("MESHBAY_USERNAME"):
         cfg.hub.username = user
-    if pwd  := os.environ.get("MESHBAY_PASSWORD"):
-        cfg.hub.password = pwd
     if port := os.environ.get("MESHBAY_PORT"):
         cfg.node.port = int(port)
 
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 5851b34..fe12909 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -23,6 +23,7 @@ Usage:
 """
 
 import asyncio
+import base64
 import json
 import logging
 import signal
@@ -30,12 +31,14 @@ import sys
 from pathlib import Path
 
 import uvicorn
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 
 from meshbay_common import MNP_VERSION
 from meshbay_common.protocol import MNP
 from meshbay_node.audit import AuditStore
+from meshbay_node.bundle_store import BundleStore
 from meshbay_node.chat.store import ChatStore
-from meshbay_node.config import Config, load_config, write_example_config
+from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config
 from meshbay_node.hub_client import HubClient, HubConfig
 from meshbay_node.indexer import DirectoryIndexer
 from meshbay_node.keystore import NodeKeys, load_or_create_keystore
@@ -111,6 +114,7 @@ class NodeDaemon:
         self._denylist = Denylist() if Denylist else None
         self._chat_stores: dict[str, ChatStore] = {}
         self._audit_store: AuditStore | None = None
+        self._bundle_store: BundleStore | None = None
         self._indexers: list[DirectoryIndexer] = []
         self._tasks: list[asyncio.Task] = []
         self._hub: HubClient | None = None
@@ -126,18 +130,46 @@ class NodeDaemon:
         )
         log.info("Keys loaded: %s", keys.pk_ed25519_b64[:16])
 
-        # 2. Hub connection
+        # 2. Start admin UI early (so operator can copy node key before hub login)
+        self._state["pk_node_ed25519"] = keys.pk_ed25519_b64
+        self._state["config"] = self._config
+        from meshbay_node.ui import create_ui_app
+        ui_app = create_ui_app(self._state)
+        ui_cfg = uvicorn.Config(
+            ui_app,
+            host="127.0.0.1",
+            port=self._config.node.ui_port,
+            log_level="warning",
+        )
+        ui_server = uvicorn.Server(ui_cfg)
+        self._tasks.append(asyncio.create_task(ui_server.serve()))
+        log.info("Admin UI at http://localhost:%d", self._config.node.ui_port)
+
+        # 3. Hub connection (Ed25519 auth — retries until node key is linked)
         hub_cfg = HubConfig(
             hub_url=self._config.hub.url,
             username=self._config.hub.username,
-            password=self._config.hub.password,
         )
         async with HubClient(hub_cfg, keys) as hub:
             self._hub = hub
-            session = await hub.startup(endpoint_hint=None)
+            session = await self._login_with_retry(hub)
             self._state["endpoint_hint"] = session.node_id
 
-            # 3. Build per-group contexts
+            # 4. Bundle store (P2P GEK bundles)
+            data_dir = self._config.data_dir
+            data_dir.mkdir(parents=True, exist_ok=True)
+            self._bundle_store = BundleStore(db_path=data_dir / "bundles.db")
+            await self._bundle_store.open()
+            log.info("Bundle store opened: %s", data_dir / "bundles.db")
+
+            # X25519 key material for GEK unwrapping
+            from cryptography.hazmat.primitives import serialization
+            sk_x_raw = keys.sk_x25519.private_bytes(
+                serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
+                serialization.NoEncryption())
+            pk_x_raw = base64.b64decode(keys.pk_x25519_b64)
+
+            # 4. Build per-group contexts
             groups_ctx: dict[str, dict] = {}
             for group_cfg in self._config.groups:
                 if not group_cfg.id or not group_cfg.shared_dir:
@@ -153,12 +185,13 @@ class NodeDaemon:
 
                 gek = None
                 if group_cfg.visibility == "private":
-                    try:
-                        gek = await hub.fetch_gek(group_cfg.id)
+                    gek = await self._load_gek(
+                        group_cfg.id, session.user_id, sk_x_raw, pk_x_raw)
+                    if gek:
                         log.info("GEK loaded for group %s", group_cfg.id[:8])
-                    except LookupError:
-                        log.warning("No GEK for group %s — skipping", group_cfg.name)
-                        continue
+                    else:
+                        log.info("No GEK yet for group %s — will accept first setup",
+                                 group_cfg.name)
 
                 indexer = DirectoryIndexer(
                     root=shared_root,
@@ -183,9 +216,7 @@ class NodeDaemon:
                 log.error("No valid groups configured — exiting")
                 return
 
-            # 4. Chat stores (one SQLite DB per group)
-            data_dir = self._config.data_dir
-            data_dir.mkdir(parents=True, exist_ok=True)
+            # 5. Chat stores (one SQLite DB per group)
             for gid in groups_ctx:
                 chat_db = data_dir / gid[:16] / "chat.db"
                 store = ChatStore(db_path=chat_db)
@@ -194,7 +225,7 @@ class NodeDaemon:
                 groups_ctx[gid]["chat_store"] = store
             log.info("Chat stores opened: %d groups", len(self._chat_stores))
 
-            # 4b. Audit store (legal compliance — IP + action logging)
+            # 6. Audit store (legal compliance — IP + action logging)
             audit_db = data_dir / "audit.db"
             self._audit_store = AuditStore(db_path=audit_db)
             await self._audit_store.open()
@@ -219,6 +250,17 @@ class NodeDaemon:
                 self._webrtc._ctx["hub_ws"] = _WsSender(hub)
                 self._webrtc._ctx["node_user_id"] = session.user_id
                 self._webrtc._ctx["audit_store"] = self._audit_store
+                self._webrtc._ctx["bundle_store"] = self._bundle_store
+                self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw
+                self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw
+                self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64
+
+                admin_pk = self._resolve_admin_pk(keys)
+                if admin_pk:
+                    self._webrtc._ctx["admin_pk_ed25519"] = admin_pk
+                    log.info("Admin Ed25519 key pinned for node sovereignty")
+                else:
+                    log.warning("No admin_pk_ed25519 — admin operations disabled")
                 log.info("WebRTC transport ready")
             else:
                 log.warning("WebRTC not available (aiortc not installed)")
@@ -323,23 +365,13 @@ class NodeDaemon:
                 log.info("HTTP API on port %d for group %s",
                          group_cfg.http_port, group_cfg.name)
 
-            # 10. Local web UI
+            # 10. Update admin UI state (UI already running from step 2)
             self._state["groups_ctx"] = groups_ctx
-            self._state["config"] = self._config
             self._state["audit_store"] = self._audit_store
+            self._state["bundle_store"] = self._bundle_store
             self._state["webrtc"] = self._webrtc
             self._state["hub"] = hub
-            from meshbay_node.ui import create_ui_app
-            ui_app = create_ui_app(self._state)
-            ui_cfg = uvicorn.Config(
-                ui_app,
-                host="127.0.0.1",
-                port=self._config.node.ui_port,
-                log_level="warning",
-            )
-            ui_server = uvicorn.Server(ui_cfg)
-            self._tasks.append(asyncio.create_task(ui_server.serve()))
-            log.info("Local UI at http://localhost:%d", self._config.node.ui_port)
+            self._state["pk_x25519_raw"] = pk_x_raw
 
             self._state["status"] = "running"
             log.info("Node ready — %d groups, WebRTC=%s, QUIC=%s",
@@ -363,6 +395,75 @@ class NodeDaemon:
 
         await self._shutdown()
 
+    async def _login_with_retry(self, hub: HubClient):
+        """Login to hub, retrying if the node key hasn't been linked yet."""
+        import httpx as _httpx
+        while True:
+            try:
+                return await hub.startup(endpoint_hint=None)
+            except _httpx.HTTPStatusError as e:
+                body = e.response.text if hasattr(e.response, 'text') else ''
+                if e.response.status_code == 401 and "No node key" in body:
+                    self._state["status"] = "waiting_for_node_key"
+                    log.warning(
+                        "Node key not linked — open admin UI at "
+                        "http://localhost:%d, copy the key, and paste it in "
+                        "Settings > Link Node on the hub. Retrying in 30s...",
+                        self._config.node.ui_port,
+                    )
+                    await asyncio.sleep(30)
+                else:
+                    raise
+            except Exception as e:
+                log.warning("Hub login failed: %s — retrying in 10s", e)
+                await asyncio.sleep(10)
+
+    async def _load_gek(
+        self,
+        group_id: str,
+        node_user_id: str,
+        sk_x_raw: bytes,
+        pk_x_raw: bytes,
+    ) -> bytes | None:
+        """Load GEK from local bundle store (node-only, hub never touches crypto)."""
+        from meshbay_common.crypto import unwrap_gek_aes
+
+        if not self._bundle_store:
+            return None
+
+        # Try node-specific bundle first (stored by init_gek for daemon reload),
+        # then fall back to operator's user bundle (legacy / pre-dual-key)
+        for user_key in [f"_node_{node_user_id}", node_user_id]:
+            bundle = await self._bundle_store.fetch(group_id, user_key)
+            if not bundle:
+                continue
+            try:
+                gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw)
+                log.info("GEK loaded from local bundle store for group %s (key=%s)",
+                         group_id[:8], user_key[:16])
+                return gek
+            except Exception as e:
+                log.debug("Failed to unwrap GEK bundle (key=%s): %s", user_key[:16], e)
+
+        log.warning("No unwrappable GEK bundle found for group %s", group_id[:8])
+        return None
+
+    def _resolve_admin_pk(self, keys: NodeKeys) -> Ed25519PublicKey | None:
+        """Resolve the admin Ed25519 public key: config → auto-pin from node keystore."""
+        if self._config.admin_pk_ed25519:
+            try:
+                raw = base64.b64decode(self._config.admin_pk_ed25519)
+                return Ed25519PublicKey.from_public_bytes(raw)
+            except Exception as e:
+                log.error("Invalid admin_pk_ed25519 in config: %s", e)
+                return None
+
+        pk = keys.sk_ed25519.public_key()
+        from meshbay_common.crypto import pk_to_b64
+        pk_b64 = pk_to_b64(pk)
+        log.info("Auto-pinning admin key from node keystore: %s", pk_b64[:16])
+        return pk
+
     async def _on_index_change(self, indexer: DirectoryIndexer) -> None:
         """Called when a DirectoryIndexer detects file changes."""
         group_id = indexer.group_id
@@ -429,6 +530,9 @@ class NodeDaemon:
         if self._audit_store:
             await self._audit_store.close()
 
+        if self._bundle_store:
+            await self._bundle_store.close()
+
         for store in self._chat_stores.values():
             await store.close()
 
@@ -475,7 +579,7 @@ def main() -> None:
         calibrate_argon2()
         return
 
-    cfg = load_config(args.config)
+    cfg = load_config(args.config or DEFAULT_CONFIG_PATH)
     if not cfg.hub.username:
         print("Error: hub.username not set in config. Run: meshbay-node init")
         sys.exit(1)
diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py
index ba9d3ff..432af0a 100644
--- a/packages/meshbay-node/src/meshbay_node/hub_client.py
+++ b/packages/meshbay-node/src/meshbay_node/hub_client.py
@@ -2,15 +2,15 @@
 MeshBay Node — Hub client.
 
 Handles all communication from the node to a Mesh Hub:
-  - User registration (first run)
-  - Login → JWT (access token + refresh token)
+  - Ed25519 authentication (node-scoped JWT, no password material on node)
   - JWT offline verification and auto-refresh
   - Node announcement (endpoint_hint)
-  - GEK bundle retrieval for a group
   - User public key lookup (for GEK wrapping)
+  - Swarm hash registration
 
-JWT verification is done locally using the hub's cached Ed25519 public key.
-The hub is only contacted for login and refresh — not for every request.
+The node authenticates via Ed25519 challenge-response (/v1/nodes/auth).
+No auth_key or password is ever stored on or transmitted from the node.
+The hub issues a node-scoped JWT that cannot manage group membership.
 """
 
 import base64
@@ -23,10 +23,7 @@ from typing import Any, Callable
 
 import httpx
 import jwt
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
-from cryptography.hazmat.primitives import serialization
 
-from meshbay_common.crypto import pk_to_b64, unwrap_gek
 from meshbay_node.keystore import NodeKeys
 
 log = logging.getLogger(__name__)
@@ -62,7 +59,6 @@ class HubSession:
 class HubConfig:
     hub_url:  str
     username: str
-    password: str
     cache_dir: Path = field(default_factory=lambda: Path.home() / ".config" / "meshbay")
 
     @property
@@ -110,80 +106,54 @@ class HubClient:
         log.info("Hub PK fetched and cached: %s", cache)
         return pem
 
-    # ── Registration ──────────────────────────────────────────────────────────
-
-    async def register(self) -> str:
-        """Register this node's user on the hub. Returns user_id. Idempotent (409 ok)."""
-        r = await self._http.post("/v1/users/register", json={
-            "username":        self._config.username,
-            "password":        self._config.password,
-            "pk_user_ed25519": self._keys.pk_ed25519_b64,
-            "pk_user_x25519":  self._keys.pk_x25519_b64,
-        })
-        if r.status_code == 201:
-            log.info("Registered user '%s' on hub", self._config.username)
-            return r.json()["user_id"]
-        if r.status_code == 409:
-            log.debug("User '%s' already registered", self._config.username)
-            return ""
-        r.raise_for_status()
-        return ""
-
-    # ── Login ─────────────────────────────────────────────────────────────────
+    # ── Ed25519 authentication ───────────────────────────────────────────────
 
     async def login(self) -> HubSession:
-        """Login, verify JWT offline, return HubSession."""
+        """Authenticate via Ed25519 challenge-response. Returns node-scoped HubSession."""
         hub_pk_pem = await self._fetch_hub_pk()
 
-        r = await self._http.post("/v1/users/login", json={
-            "username": self._config.username,
-            "password": self._config.password,
+        timestamp = int(time.time())
+        message = f"meshbay:node_auth:{self._config.username}:{timestamp}".encode()
+        signature = self._keys.sk_ed25519.sign(message)
+
+        r = await self._http.post("/v1/nodes/auth", json={
+            "username":  self._config.username,
+            "timestamp": timestamp,
+            "signature": base64.b64encode(signature).decode(),
         })
         r.raise_for_status()
         data = r.json()
 
-        access_token  = data["access_token"]
-        refresh_token = data["refresh_token"]
+        access_token = data["access_token"]
 
-        # Verify offline — if this passes, the hub's identity is confirmed
         decoded = jwt.decode(access_token, hub_pk_pem, algorithms=["EdDSA"])
         assert decoded["pk_user"] == self._keys.pk_ed25519_b64, \
             "Hub returned token for wrong public key"
         assert "jti" in decoded, "Hub token missing jti — hub is outdated"
+        assert decoded.get("scope") == "node", \
+            "Expected node-scoped token"
 
-        self._session = HubSession(
-            hub_url=self._config.hub_url,
-            username=self._config.username,
-            user_id=decoded["sub"],
-            access_token=access_token,
-            refresh_token=refresh_token,
-            hub_pk_pem=hub_pk_pem,
-            _token_exp=decoded["exp"],
-        )
+        if self._session:
+            self._session.access_token = access_token
+            self._session._token_exp = decoded["exp"]
+        else:
+            self._session = HubSession(
+                hub_url=self._config.hub_url,
+                username=self._config.username,
+                user_id=decoded["sub"],
+                access_token=access_token,
+                refresh_token="",
+                hub_pk_pem=hub_pk_pem,
+                _token_exp=decoded["exp"],
+            )
         log.info("Logged in as '%s' (exp in %ds)", self._config.username,
                  self._session.token_expires_in)
         return self._session
 
-    async def refresh_token(self) -> None:
-        """Refresh the access token using the refresh token."""
-        if self._session is None:
-            raise RuntimeError("Not logged in")
-
-        r = await self._http.post("/v1/users/token/refresh", json={
-            "refresh_token": self._session.refresh_token,
-        })
-        r.raise_for_status()
-        new_token = r.json()["access_token"]
-
-        decoded = jwt.decode(new_token, self._session.hub_pk_pem, algorithms=["EdDSA"])
-        self._session.access_token = new_token
-        self._session._token_exp   = decoded["exp"]
-        log.debug("Access token refreshed (exp in %ds)", self._session.token_expires_in)
-
     async def ensure_fresh_token(self) -> None:
-        """Auto-refresh token if close to expiry."""
+        """Re-authenticate with Ed25519 if token is close to expiry."""
         if self._session and self._session.token_needs_refresh:
-            await self.refresh_token()
+            await self.login()
 
     # ── Node announcement ─────────────────────────────────────────────────────
 
@@ -203,33 +173,6 @@ class HubClient:
         log.info("Node announced: %s (hint=%s)", node_id[:8], endpoint_hint)
         return node_id
 
-    # ── GEK retrieval ─────────────────────────────────────────────────────────
-
-    async def fetch_gek(self, group_id: str) -> bytes:
-        """
-        Fetch and unwrap the GEK bundle for a group.
-        Returns the raw GEK bytes.
-        """
-        if self._session is None:
-            raise RuntimeError("Not logged in")
-        await self.ensure_fresh_token()
-
-        r = await self._http.get(f"/v1/groups/{group_id}/gek",
-                                 headers=self._session.auth_headers)
-        if r.status_code == 404:
-            raise LookupError(f"No GEK bundle found for group {group_id!r}")
-        r.raise_for_status()
-
-        bundle     = r.json()
-        sk_x_raw   = self._keys.sk_x25519.private_bytes(
-            serialization.Encoding.Raw, serialization.PrivateFormat.Raw,
-            serialization.NoEncryption())
-        pk_x_raw   = base64.b64decode(self._keys.pk_x25519_b64)
-
-        gek = unwrap_gek(bundle, sk_x_raw, pk_x_raw)
-        log.info("GEK unwrapped for group %s", group_id[:8])
-        return gek
-
     # ── User pubkey lookup ────────────────────────────────────────────────────
 
     async def get_user_pubkeys(self, username: str) -> dict:
@@ -354,10 +297,9 @@ class HubClient:
 
     async def startup(self, endpoint_hint: str | None = None) -> HubSession:
         """
-        Full startup sequence: register (idempotent) → login → announce node.
-        Returns an active HubSession.
+        Full startup sequence: Ed25519 login → announce node.
+        The operator must register separately (browser or setup script).
         """
-        await self.register()
         session = await self.login()
         await self.announce_node(endpoint_hint)
         return session
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 e692c80..13e90c8 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -24,7 +24,10 @@ Signaling flow (handled externally by the hub):
 
 import asyncio
 import base64
+import hashlib
+import hmac
 import logging
+import os
 import struct
 from pathlib import Path
 from typing import Any
@@ -32,7 +35,10 @@ from typing import Any
 import jwt
 import msgpack
 from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
 
 from meshbay_common import MNP_VERSION
 from meshbay_common.crypto import pk_to_b64
@@ -46,6 +52,15 @@ CHUNK_SIZE = 1024 * 1024
 MAX_MSG = 64 * 1024 * 1024
 
 
+def _extract_dtls_fingerprint(sdp: str) -> bytes:
+    """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
+    for line in sdp.splitlines():
+        if line.startswith("a=fingerprint:sha-256 "):
+            hex_str = line.split(" ", 1)[1].replace(":", "")
+            return bytes.fromhex(hex_str)
+    return b""
+
+
 STREAM_SEGMENT_SIZE = 256 * 1024
 
 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}
@@ -153,6 +168,9 @@ class WebRTCPeerSession:
         self._peer_id: str = peer_id
         self._remote_ip: str = ""
         self._username: str = ""
+        self._pk_user: str = ""
+        self._gek_challenge: bytes | None = None
+        self._admin_challenges: dict[str, bytes] = {}
 
     def _setup_channel(self, channel: RTCDataChannel) -> None:
         self._channel = channel
@@ -171,6 +189,12 @@ class WebRTCPeerSession:
         try:
             if mtype == MNP.HANDSHAKE:
                 self._do_handshake(msg)
+            elif mtype == MNP.HANDSHAKE_RESPONSE:
+                self._do_handshake_response(msg)
+            elif mtype == MNP.GEK_BUNDLE_FETCH and self._gek_challenge is not None:
+                asyncio.ensure_future(self._do_gek_bundle_fetch())
+            elif mtype == MNP.KEYPAIR_BUNDLE_FETCH and self._gek_challenge is not None:
+                asyncio.ensure_future(self._do_keypair_bundle_fetch())
             elif self._user_id is None:
                 self._send({"type": "error", "detail": "Handshake required"})
             elif mtype == MNP.INDEX_SYNC:
@@ -179,8 +203,6 @@ class WebRTCPeerSession:
                 self._do_file_request(msg)
             elif mtype == MNP.STREAM_SEGMENT:
                 self._do_stream_segment(msg)
-            elif mtype == MNP.GEK_REQUEST:
-                self._do_gek_request()
             elif mtype == MNP.CHAT_MESSAGE:
                 self._do_chat_message(msg)
             elif mtype == MNP.CHAT_HISTORY:
@@ -189,6 +211,12 @@ class WebRTCPeerSession:
                 self._do_file_upload(msg)
             elif mtype == MNP.FILE_DELETE:
                 self._do_file_delete(msg)
+            elif mtype == MNP.ADMIN_RESPONSE:
+                self._do_admin_response(msg)
+            elif mtype == MNP.GEK_BUNDLE_STORE:
+                asyncio.ensure_future(self._do_gek_bundle_store(msg))
+            elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
+                asyncio.ensure_future(self._do_keypair_bundle_store(msg))
             elif mtype == MNP.STREAM_REQUEST:
                 asyncio.ensure_future(self._stream_video(msg))
             else:
@@ -234,23 +262,235 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "Group not hosted on this node"})
             return
 
-        self._user_id = decoded["sub"]
-        self._group_id = group_id
-        self._username = decoded.get("username", "")
+        # 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", "")
+
+        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:
+            self._send({
+                "type": "error",
+                "detail": "Group encryption not initialized — contact node operator",
+            })
+            return
+        self._send(challenge)
+
+    def _do_handshake_response(self, msg: dict) -> None:
+        if not self._gek_challenge or not hasattr(self, "_pending_sub"):
+            self._send({"type": "error", "detail": "No pending handshake challenge"})
+            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
+        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)
+        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)
+
+        data = self._gek_challenge + offer_fp + answer_fp
+        expected = hmac.new(gek, data, hashlib.sha256).digest()
+        if not hmac.compare_digest(proof_bytes, expected):
+            self._send({"type": "error", "detail": "GEK proof failed"})
+            self._gek_challenge = None
+            self._audit_auth_failed(group_id, "GEK HMAC mismatch")
+            return
+
+        self._gek_challenge = None
+        self._complete_handshake()
+
+    def _complete_handshake(self) -> None:
+        self._user_id = self._pending_sub
+        self._group_id = self._pending_group
+        self._username = self._pending_username
+        self._pk_user = self._pending_pk_user
 
         peers = self._ctx.get("_peers")
         if peers is not None:
             peers[self._user_id] = self
 
+        node_user_id = self._ctx.get("node_user_id")
         log.info("WebRTC handshake OK — user=%s group=%s",
-                 self._user_id[:8], group_id[:8] if group_id else "none")
-        self._send({
+                 self._user_id[:8],
+                 self._group_id[:8] if self._group_id else "none")
+        ack = {
             "type": MNP.HANDSHAKE_ACK,
             "v": MNP_VERSION,
             "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()),
-        })
+            "is_node_admin": bool(node_user_id and self._user_id == node_user_id),
+        }
+        if node_user_id:
+            ack["node_user_id"] = node_user_id
+        pk_x_b64 = self._ctx.get("pk_x25519_b64")
+        if pk_x_b64:
+            ack["node_pk_x25519"] = pk_x_b64
+        self._send(ack)
         self._audit("handshake")
 
+    async def _do_gek_bundle_fetch(self) -> None:
+        """Serve the caller's wrapped GEK bundle during the handshake window."""
+        bundle_store = self._ctx.get("bundle_store")
+        if not bundle_store:
+            self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
+            return
+
+        group_id = getattr(self, "_pending_group", "")
+        user_id = getattr(self, "_pending_sub", "")
+        if not group_id or not user_id:
+            self._send({"type": "error", "detail": "No pending handshake"})
+            return
+
+        bundle = await bundle_store.fetch(group_id, user_id)
+        if bundle:
+            self._send({
+                "type": MNP.GEK_BUNDLE_RESP,
+                "v": MNP_VERSION,
+                "found": True,
+                "pk_eph_b64": bundle["pk_eph_b64"],
+                "nonce_b64": bundle["nonce_b64"],
+                "wrapped_b64": bundle["wrapped_b64"],
+            })
+        else:
+            self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
+
+    async def _do_gek_bundle_store(self, msg: dict) -> None:
+        """Store a wrapped GEK bundle for a target user (admin operation)."""
+        bundle_store = self._ctx.get("bundle_store")
+        if not bundle_store:
+            self._send({"type": "error", "detail": "Bundle store not available"})
+            return
+
+        target_user_id = msg.get("user_id", "")
+        group_id = msg.get("group_id") or self._group_id
+        pk_eph = msg.get("pk_eph_b64", "")
+        nonce = msg.get("nonce_b64", "")
+        wrapped = msg.get("wrapped_b64", "")
+
+        if not target_user_id or not pk_eph or not nonce or not wrapped or not group_id:
+            self._send({"type": "error", "detail": "Missing bundle fields"})
+            return
+
+        await bundle_store.store(group_id, target_user_id, pk_eph, nonce, wrapped)
+        log.info("GEK bundle stored: group=%s user=%s", group_id[:8], target_user_id[:8])
+        self._audit("gek_bundle_store", f"target={target_user_id[:8]}")
+
+        self._send({
+            "type": "ack", "v": MNP_VERSION,
+            "detail": "gek_bundle_stored",
+            "user_id": target_user_id,
+        })
+
+        # Auto-activate GEK if the bundle is for the node operator
+        node_user_id = self._ctx.get("node_user_id")
+        if node_user_id and target_user_id == node_user_id and group_id:
+            await self._try_activate_gek(group_id, target_user_id)
+
+    async def _try_activate_gek(self, group_id: str, user_id: str) -> None:
+        """Unwrap and activate GEK for the node when the operator's bundle arrives."""
+        from meshbay_common.crypto import unwrap_gek_aes
+
+        bundle_store = self._ctx.get("bundle_store")
+        sk_x_raw = self._ctx.get("sk_x25519_raw")
+        pk_x_raw = self._ctx.get("pk_x25519_raw")
+        if not bundle_store or not sk_x_raw or not pk_x_raw:
+            return
+
+        bundle = await bundle_store.fetch(group_id, user_id)
+        if not bundle:
+            return
+
+        try:
+            gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw)
+        except Exception as e:
+            log.warning("Failed to unwrap GEK for auto-activation: %s", e)
+            return
+
+        groups = self._ctx.get("groups")
+        if groups and group_id in groups:
+            groups[group_id]["gek"] = gek
+            log.info("GEK auto-activated for group %s", group_id[:8])
+        elif "gek" in self._ctx:
+            self._ctx["gek"] = gek
+            log.info("GEK auto-activated (single-group mode)")
+
+    async def _do_keypair_bundle_fetch(self) -> None:
+        """Serve the caller's encrypted keypair bundle during the handshake window."""
+        bundle_store = self._ctx.get("bundle_store")
+        if not bundle_store:
+            self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
+            return
+
+        user_id = getattr(self, "_pending_sub", "")
+        if not user_id:
+            self._send({"type": "error", "detail": "No pending handshake"})
+            return
+
+        bundle_enc = await bundle_store.fetch_keypair(user_id)
+        if bundle_enc:
+            self._send({
+                "type": MNP.KEYPAIR_BUNDLE_RESP,
+                "v": MNP_VERSION,
+                "found": True,
+                "bundle_enc": bundle_enc,
+            })
+        else:
+            self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False})
+
+    async def _do_keypair_bundle_store(self, msg: dict) -> None:
+        """Store an encrypted keypair bundle (user backs up their own keys on node)."""
+        bundle_store = self._ctx.get("bundle_store")
+        if not bundle_store:
+            self._send({"type": "error", "detail": "Bundle store not available"})
+            return
+
+        bundle_enc = msg.get("bundle_enc", "")
+        if not bundle_enc:
+            self._send({"type": "error", "detail": "Missing bundle_enc"})
+            return
+
+        await bundle_store.store_keypair(self._user_id, bundle_enc)
+        log.info("Keypair bundle stored for user=%s", self._user_id[:8])
+        self._audit("keypair_bundle_store")
+        self._send({
+            "type": "ack", "v": MNP_VERSION,
+            "detail": "keypair_bundle_stored",
+        })
+
     def _audit_auth_failed(self, group_id: str, reason: str) -> None:
         audit = self._ctx.get("audit_store")
         if audit:
@@ -275,6 +515,7 @@ class WebRTCPeerSession:
             {
                 "id": e.id, "name": e.name, "path": e.path,
                 "size": e.size, "type": e.type, "added_at": e.added_at,
+                "uploader_id": e.uploader_id,
             }
             for e in idx.entries
         ]
@@ -286,18 +527,6 @@ class WebRTCPeerSession:
             "entries": entries,
         })
 
-    def _do_gek_request(self) -> None:
-        ctx = self._group_ctx()
-        gek = ctx.get("gek")
-        if not gek:
-            self._send({"type": "error", "detail": "No GEK available"})
-            return
-        self._send({
-            "type": MNP.GEK_RESPONSE,
-            "v": MNP_VERSION,
-            "gek_b64": base64.b64encode(gek).decode(),
-        })
-
     def _do_file_request(self, msg: dict) -> None:
         ctx = self._group_ctx()
         file_id = msg["file_id"]
@@ -378,7 +607,7 @@ class WebRTCPeerSession:
         if chat_store:
             raw = payload.encode() if isinstance(payload, str) else payload
             asyncio.ensure_future(chat_store.save_message(
-                sender_id=msg.get("sender_id", self._user_id),
+                sender_id=self._user_id,
                 iteration=msg.get("iteration", 0),
                 payload=raw,
                 thread_id=msg.get("thread_id"),
@@ -389,7 +618,7 @@ class WebRTCPeerSession:
         broadcast = {
             "type": MNP.CHAT_MESSAGE,
             "v": MNP_VERSION,
-            "sender_id": msg.get("sender_id", self._user_id),
+            "sender_id": self._user_id,
             "sender_name": sender_name,
             "payload": payload,
             "thread_id": msg.get("thread_id"),
@@ -493,6 +722,18 @@ class WebRTCPeerSession:
             tmp_path.rename(final_path)
             log.info("Upload complete: %s (%d chunks)", safe_name, total_chunks)
             self._audit("file_upload", safe_name)
+            self._register_uploader(ctx, safe_name)
+
+    def _register_uploader(self, ctx: dict, filename: str) -> None:
+        """Tag the index entry with the uploader's user_id after upload completes."""
+        idx = ctx.get("index")
+        if not idx:
+            return
+        for entry in idx.entries:
+            if entry.name == filename and entry.path == "":
+                entry.uploader_id = self._user_id
+                entry.uploader_pk = self._pk_user
+                return
 
     def _do_file_delete(self, msg: dict) -> None:
         ctx = self._group_ctx()
@@ -501,16 +742,76 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "Missing file_id"})
             return
 
-        node_user_id = self._ctx.get("node_user_id")
-        if node_user_id and self._user_id != node_user_id:
-            self._send({"type": "error", "detail": "Only node admin can delete files"})
+        entry = ctx["index"].get_entry(file_id)
+        if not entry:
+            self._send({"type": "error", "detail": "File not found"})
+            return
+
+        admin_pk = self._ctx.get("admin_pk_ed25519")
+        has_uploader_pk = bool(entry.uploader_pk)
+        if not admin_pk and not has_uploader_pk:
+            self._send({"type": "error", "detail": "No authorized key for deletion"})
+            return
+
+        challenge = os.urandom(32)
+        self._admin_challenges[file_id] = challenge
+        self._send({
+            "type": MNP.ADMIN_CHALLENGE,
+            "v": MNP_VERSION,
+            "challenge": base64.b64encode(challenge).decode(),
+            "file_id": file_id,
+        })
+
+    def _do_admin_response(self, msg: dict) -> None:
+        file_id = msg.get("file_id", "")
+        sig_b64 = msg.get("signature", "")
+
+        challenge = self._admin_challenges.pop(file_id, None)
+        if not challenge:
+            self._send({"type": "error", "detail": "No pending admin challenge"})
+            return
+
+        try:
+            sig_bytes = base64.b64decode(sig_b64)
+        except Exception:
+            self._send({"type": "error", "detail": "Invalid signature encoding"})
             return
 
+        ctx = self._group_ctx()
         entry = ctx["index"].get_entry(file_id)
         if not entry:
             self._send({"type": "error", "detail": "File not found"})
             return
 
+        verified = False
+
+        # Try admin key (locally pinned)
+        admin_pk = self._ctx.get("admin_pk_ed25519")
+        if admin_pk:
+            try:
+                admin_pk.verify(sig_bytes, challenge)
+                verified = True
+            except Exception:
+                pass
+
+        # Try uploader key (stored at upload time)
+        if not verified and entry.uploader_pk:
+            try:
+                uploader_key = Ed25519PublicKey.from_public_bytes(
+                    base64.b64decode(entry.uploader_pk))
+                uploader_key.verify(sig_bytes, challenge)
+                verified = True
+            except Exception:
+                pass
+
+        if not verified:
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}")
+            return
+
+        self._exec_file_delete(ctx, file_id, entry)
+
+    def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None:
         file_path = ctx["shared_root"] / entry.path / entry.name
         if file_path.exists():
             file_path.unlink()
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index 5e77ed8..b4885af 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -12,15 +12,17 @@ Served only on 127.0.0.1 — not exposed to the network.
 No authentication required (localhost only).
 """
 
+import base64
 import json
 import logging
 import time
 from pathlib import Path
 
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
-from fastapi.responses import HTMLResponse
+from fastapi.responses import HTMLResponse, JSONResponse
 
 from meshbay_node import __version__
+from meshbay_common.crypto import generate_gek, wrap_gek_aes
 
 log = logging.getLogger(__name__)
 
@@ -52,6 +54,7 @@ def create_ui_app(state: dict) -> FastAPI:
             "group_count": len(groups_ctx),
             "total_files": total_files,
             "webrtc_peers": webrtc.active_peers if webrtc else 0,
+            "pk_node_ed25519": state.get("pk_node_ed25519", ""),
         }
 
     @app.get("/api/groups")
@@ -167,6 +170,95 @@ def create_ui_app(state: dict) -> FastAPI:
             ],
         }
 
+    # ── GEK initialization (operator only, localhost) ──────────────────────
+
+    @app.post("/api/groups/{group_id}/gek")
+    async def init_gek(group_id: str):
+        """Generate GEK, wrap for all group members, store, and activate."""
+        groups_ctx = state.get("groups_ctx", {})
+        if group_id not in groups_ctx:
+            return JSONResponse({"error": "Group not hosted on this node"}, 404)
+
+        hub = state.get("hub")
+        if not hub or not hub._session:
+            return JSONResponse({"error": "Hub not connected"}, 503)
+
+        bundle_store = state.get("bundle_store")
+        if not bundle_store:
+            return JSONResponse({"error": "Bundle store not available"}, 503)
+
+        await hub.ensure_fresh_token()
+        session = hub._session
+        members_resp = await hub._http.get(
+            f"/v1/groups/{group_id}/members",
+            headers=session.auth_headers,
+        )
+        if not members_resp.is_success:
+            return JSONResponse(
+                {"error": f"Failed to fetch members: {members_resp.status_code}"}, 502)
+        members = members_resp.json().get("members", [])
+        if not members:
+            return JSONResponse({"error": "No members in group"}, 400)
+
+        existing_gek = groups_ctx[group_id].get("gek")
+        gek = existing_gek or generate_gek()
+
+        wrapped_count = 0
+        errors = []
+        for member in members:
+            username = member["username"]
+            user_id = member["user_id"]
+            try:
+                pk_data = await hub.get_user_pubkeys(username)
+                pk_x_raw = base64.b64decode(pk_data["pk_x25519"])
+                bundle = wrap_gek_aes(gek, pk_x_raw)
+                await bundle_store.store(
+                    group_id, user_id,
+                    bundle["pk_eph_b64"], bundle["nonce_b64"], bundle["wrapped_b64"],
+                )
+                wrapped_count += 1
+                log.info("GEK wrapped for %s (%s)", username, user_id[:8])
+            except Exception as e:
+                errors.append(f"{username}: {e}")
+                log.warning("Failed to wrap GEK for %s: %s", username, e)
+
+        if wrapped_count == 0:
+            return JSONResponse(
+                {"error": "Failed to wrap GEK for any member", "details": errors}, 500)
+
+        # Also store a copy wrapped for the node keystore X25519 key
+        # so the daemon can reload GEK on restart without the operator's browser keys
+        config = state.get("config")
+        node_user_id = hub._session.user_id if hub._session else None
+        pk_x_node_raw = state.get("pk_x25519_raw")
+        if pk_x_node_raw and node_user_id:
+            try:
+                node_bundle = wrap_gek_aes(gek, pk_x_node_raw)
+                await bundle_store.store(
+                    group_id, f"_node_{node_user_id}",
+                    node_bundle["pk_eph_b64"], node_bundle["nonce_b64"],
+                    node_bundle["wrapped_b64"],
+                )
+                log.info("GEK also wrapped for node keystore (daemon reload)")
+            except Exception as e:
+                log.warning("Failed to wrap GEK for node keystore: %s", e)
+
+        groups_ctx[group_id]["gek"] = gek
+        log.info("GEK initialized for group %s — wrapped for %d/%d members",
+                 group_id[:8], wrapped_count, len(members))
+
+        webrtc = state.get("webrtc")
+        if webrtc and "groups" in webrtc._ctx and group_id in webrtc._ctx["groups"]:
+            webrtc._ctx["groups"][group_id]["gek"] = gek
+
+        return {
+            "status": "ok",
+            "group_id": group_id,
+            "wrapped_count": wrapped_count,
+            "total_members": len(members),
+            "errors": errors,
+        }
+
     # ── Chat endpoints ───────────────────────────────────────────────────────
 
     _chat_subscribers: list[WebSocket] = []
@@ -246,7 +338,10 @@ def _render_page(state: dict) -> str:
     webrtc = state.get("webrtc")
     total_files = sum(idx.count for idx in indexes.values())
     peer_count = webrtc.active_peers if webrtc else 0
-    status_color = {"running": "#22c55e", "error": "#ef4444"}.get(status, "#f59e0b")
+    status_color = {
+        "running": "#22c55e", "error": "#ef4444",
+        "waiting_for_node_key": "#f97316",
+    }.get(status, "#f59e0b")
 
     # Groups section
     groups_html = ""
@@ -269,13 +364,34 @@ def _render_page(state: dict) -> str:
                     f"<td>{_fmt_size(e.size)}</td><td>{e.path or '/'}</td></tr>"
                 )
 
+        has_gek = bool(ctx.get("gek"))
+        gek_badge = (
+            '<span class="badge" style="background:#22c55e">GEK active</span>'
+            if has_gek
+            else '<span class="badge" style="background:#ef4444">No GEK</span>'
+        )
+        gek_label = "Re-wrap GEK for all members" if has_gek else "Initialize GEK"
+        gek_color = "#3b82f6" if has_gek else "#22c55e"
+        gek_action = f"""
+            <div style="margin:10px 0">
+              <button onclick="initGEK('{gid}')"
+                id="gek-btn-{gid[:8]}"
+                style="padding:8px 16px;background:{gek_color};color:#fff;border:none;
+                border-radius:6px;cursor:pointer;font-size:0.85em">
+                {gek_label}
+              </button>
+              <span id="gek-status-{gid[:8]}" class="muted" style="margin-left:8px"></span>
+            </div>"""
+
         groups_html += f"""
         <div class="card">
           <h3>{name}
             <span class="badge" style="background:#6366f1">{vis}</span>
+            {gek_badge}
           </h3>
           <p><b>Directory:</b> <code>{shared}</code></p>
           <p><b>Files:</b> {fcount} &mdash; <b>Total:</b> {_fmt_size(total_size)}</p>
+          {gek_action}
           <p class="muted">ID: {gid}</p>
           <details><summary>File list</summary>
           <table>
@@ -382,6 +498,21 @@ def _render_page(state: dict) -> str:
     <p><b>Node ID:</b> <code>{state.get("endpoint_hint") or "—"}</code></p>
   </div>
 
+  <h2>Link Node to Hub Account</h2>
+  <div class="card">
+    <p>To connect to your group from a browser, link this node to your hub account.
+       Copy the key below and paste it in <b>Settings &gt; Link Node</b> on the hub.</p>
+    <div style="margin:12px 0;display:flex;align-items:center;gap:8px">
+      <code id="nodeKey" style="flex:1;padding:8px;word-break:break-all;background:var(--border);
+        border-radius:4px;font-size:0.9em;user-select:all">{state.get("pk_node_ed25519", "—")}</code>
+      <button onclick="navigator.clipboard.writeText(document.getElementById('nodeKey').textContent).then(()=>{{this.textContent='Copied!';setTimeout(()=>this.textContent='Copy',2000)}})"
+        style="padding:8px 16px;background:var(--accent);color:#fff;border:none;border-radius:6px;
+        cursor:pointer;font-size:0.85em;white-space:nowrap">Copy</button>
+    </div>
+    <p class="muted">This is the node's Ed25519 public key. It's safe to share — it identifies
+       this node but cannot be used to impersonate it.</p>
+  </div>
+
   <div class="footer">
     MeshBay Node v{__version__} &mdash; localhost only &mdash;
     <a href="/api/status">status</a> &middot;
@@ -392,7 +523,33 @@ def _render_page(state: dict) -> str:
     &mdash; auto-refresh 10s
   </div>
 </div>
-<script>setTimeout(()=>location.reload(), 10000);</script>
+<script>
+async function initGEK(groupId) {{
+  const btn = document.getElementById('gek-btn-' + groupId.slice(0,8));
+  const status = document.getElementById('gek-status-' + groupId.slice(0,8));
+  if (btn) btn.disabled = true;
+  if (status) status.textContent = 'Initializing...';
+  try {{
+    const resp = await fetch('/api/groups/' + groupId + '/gek', {{ method: 'POST' }});
+    const data = await resp.json();
+    if (resp.ok) {{
+      if (status) status.textContent = 'GEK initialized — wrapped for '
+        + data.wrapped_count + '/' + data.total_members + ' members';
+      if (status) status.style.color = '#22c55e';
+      setTimeout(() => location.reload(), 2000);
+    }} else {{
+      if (status) status.textContent = data.error || 'Failed';
+      if (status) status.style.color = '#ef4444';
+      if (btn) btn.disabled = false;
+    }}
+  }} catch (e) {{
+    if (status) status.textContent = 'Error: ' + e.message;
+    if (status) status.style.color = '#ef4444';
+    if (btn) btn.disabled = false;
+  }}
+}}
+setTimeout(()=>location.reload(), 10000);
+</script>
 </body>
 </html>"""
 
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()