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>

1 files changed, 327 insertions, 26 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index 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()