merge: Phase 11.5 security remediation, invite redesign, per-node identity

Brings in the security remediation branch. Three bodies of work, and what they
changed about what this project may claim.

Phase 11.5 closed the gap between the documents and the code: the unauthenticated
node HTTP API and the TCP transport deleted, one handshake shared by the
remaining two transports, mutual authentication, structured admin transcripts,
upload confinement, group isolation, revocation that reaches nodes. Six critical
and seven high findings closed, bounded, or deferred by decision.

The invite redesign closed H3 and M3 — the last open High. The hub was the key
directory: an inviter fetched the invitee's key from it and wrapped the group key
for whatever came back, so a hub answering with its own key was handed the group
key by an honest member following the protocol exactly. That lookup is gone. The
node holds the group key and wraps it itself, for a key its recipient proves
possession of, bound to an account by a one-time code the hub never sees. M3 fell
out of the same work: node authority comes from a local roster, never from the
hub.

Per-node identity cut what remains of C4 down to one operator. A single keypair
used to be copied to every node its owner joined; each node now gets its own, so
cracking the bundle on one machine yields a key that is a stranger everywhere
else — and on that machine, one that unlocks nothing its holder did not already
serve. The bundle KDF moved to Argon2id 128 MB, and the hub stopped storing or
publishing user keys at all.

What this project may now say: the hub cannot read your content unless it ships
you malicious client code. T3 remains, accepted (D1), and is what the native
client removes. C4 is reduced, not closed, until 13.3. Chat is still plaintext at
rest until Phase 15. Draft-v5 §2 states each claim against the adversary it holds
against, which is the convention this branch exists to keep.

Four defects were found by deploying it and using a browser, none by the test
suite: a node going deaf on its hub socket, a token that predated group
membership, a client reading values before they were assigned, and identity keys
a browser held but never re-read. The lessons are recorded in CLAUDE.md.

Tests: 343 across the three packages, plus QE/deploy/e2e.py — register, pair,
invite, join, download, stream, second browser, revoke — run against the live
deployment on a wiped hub and node.

1 files changed, 805 insertions, 189 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 13e90c8..fe4e3c2 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -28,7 +28,9 @@ import hashlib
 import hmac
 import logging
 import os
+import re
 import struct
+import time
 from pathlib import Path
 from typing import Any
 
@@ -41,16 +43,68 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import (
 )
 
 from meshbay_common import MNP_VERSION
-from meshbay_common.crypto import pk_to_b64
+from meshbay_common.handshake import (
+    NONCE_LEN,
+    ROLE_CLIENT,
+    ROLE_NODE,
+    HandshakeError,
+    authorize_token,
+    handshake_transcript,
+    make_proof,
+    verify_proof,
+    webrtc_binding,
+)
+from meshbay_common.adminop import (
+    ADMIN_CHALLENGE_TTL,
+    OP_FILE_DELETE,
+    OP_INVITE_CREATE,
+    admin_transcript,
+)
+from meshbay_common.crypto import pk_to_b64, wrap_gek_aes
+from meshbay_common.join import (
+    JOIN_TTL,
+    ROLE_MEMBER,
+    ROLE_OPERATOR,
+    join_transcript,
+)
 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes
 from meshbay_common.protocol import MNP
 from meshbay_node.indexer import GroupIndex
+from meshbay_node.roster import DEFAULT_INVITE_TTL
 
 log = logging.getLogger(__name__)
 
 CHUNK_SIZE = 1024 * 1024
 MAX_MSG = 64 * 1024 * 1024
 
+# Upload limits (finding C5a). Uploads used to land directly in the shared root under
+# a name the client chose, overwriting whatever was already there — which both violated
+# node sovereignty and defeated the delete authorization (overwrite a file, become its
+# recorded uploader, then delete it legitimately).
+MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024   # 4 GB per file
+
+# Budget for an unauthenticated peer: enough for a handshake and a bundle fetch,
+# nowhere near enough to be a memory-exhaustion primitive (H6).
+PRE_HANDSHAKE_MAX_MSG = 64 * 1024
+# ffmpeg is spawned per stream request; without a cap any member can fork-bomb
+# the node by requesting many streams at once (H6).
+MAX_CONCURRENT_TRANSCODES = 2
+# Bundle fetches are served in the pre-proof window (C4). Bounded and audited
+# until the native client removes remote keypair bundles entirely.
+MAX_PRE_PROOF_FETCHES = 4
+# Pairing codes carry 40 bits and are single-use, but a connection must not be
+# allowed to sit there guessing. Failures are audited, so a grind is visible.
+MAX_JOIN_ATTEMPTS = 5
+# Per-connection limits alone would not bind an attacker who can open connections
+# at will — and the adversary who can mint tokens for any account is the hub. So
+# failed pairings are also counted node-wide over a window.
+MAX_JOIN_FAILURES_WINDOW = 20
+JOIN_FAILURE_WINDOW = 600  # seconds
+UPLOAD_DIR_NAME = ".uploads"
+# Conservative allowlist: also what keeps markup out of filenames, which the node admin
+# UI used to render unescaped (finding H2).
+SAFE_UPLOAD_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._ -]{0,127}$")
+
 
 def _extract_dtls_fingerprint(sdp: str) -> bytes:
     """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes."""
@@ -119,10 +173,18 @@ def _pack(obj: dict) -> bytes:
 
 
 class _DataChannelBuffer:
-    """Accumulate DataChannel messages and extract length-prefixed msgpack."""
+    """
+    Accumulate DataChannel messages and extract length-prefixed msgpack.
+
+    Finding H6: the limit was a flat 64 MB applied even before the handshake, so an
+    unauthenticated peer could announce a 64 MB frame and dribble bytes into it,
+    holding that much memory per connection. Until a peer has proved GEK
+    possession it gets a small budget; the large one is for file uploads.
+    """
 
-    def __init__(self):
+    def __init__(self, max_message: int = MAX_MSG):
         self._buf = bytearray()
+        self.max_message = max_message
 
     def feed(self, data: bytes):
         self._buf.extend(data)
@@ -130,7 +192,7 @@ class _DataChannelBuffer:
     def messages(self):
         while len(self._buf) >= 4:
             length = struct.unpack(">I", self._buf[:4])[0]
-            if length > MAX_MSG:
+            if length > self.max_message:
                 raise ValueError(f"Message too large: {length}")
             if len(self._buf) < 4 + length:
                 break
@@ -162,15 +224,25 @@ class WebRTCPeerSession:
         self._pc = pc
         self._ctx = node_ctx
         self._channel: RTCDataChannel | None = None
-        self._buffer = _DataChannelBuffer()
+        self._buffer = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
+        self._pre_proof_fetches = 0
         self._user_id: str | None = None
         self._group_id: str | None = None
         self._peer_id: str = peer_id
         self._remote_ip: str = ""
         self._username: str = ""
-        self._pk_user: str = ""
+        # Set from the roster: the key this node pinned for this account. Never
+        # from the JWT — the hub picks what goes in there.
+        self._pinned_pk: str = ""
         self._gek_challenge: bytes | None = None
-        self._admin_challenges: dict[str, bytes] = {}
+        # Same value as the GEK challenge, but kept for the life of the connection:
+        # a join_request is signed over it, and it must stay verifiable after the
+        # handshake clears the challenge (an operator pairs while already connected).
+        self._nonce_node: bytes = b""
+        self._join_attempts = 0
+        self._nonce_client: bytes = b""
+        self._admin_ops: dict[str, dict] = {}   # op_id → pending admin operation
+        self._uploads: dict[str, dict] = {}   # filename → {next_index, bytes}
 
     def _setup_channel(self, channel: RTCDataChannel) -> None:
         self._channel = channel
@@ -191,10 +263,30 @@ class WebRTCPeerSession:
                 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 mtype in (MNP.GEK_BUNDLE_FETCH, MNP.KEYPAIR_BUNDLE_FETCH) \
+                    and self._gek_challenge is not None:
+                # Served before the GEK proof by necessity: the client needs its
+                # wrapped bundle in order to compute the proof. That window is a
+                # disclosure surface (C4) — a hub that forges a JWT reaches it — so
+                # it is bounded and audited here, and closed properly when clients
+                # stop storing keypair bundles on other people's nodes.
+                self._pre_proof_fetches += 1
+                if self._pre_proof_fetches > MAX_PRE_PROOF_FETCHES:
+                    self._audit_auth_failed(
+                        getattr(self, "_pending_group", ""), "pre-proof fetch flood")
+                    self._send({"type": "error", "detail": "Too many requests"})
+                    return
+                self._audit_pre_proof_fetch(mtype)
+                if mtype == MNP.GEK_BUNDLE_FETCH:
+                    asyncio.ensure_future(self._do_gek_bundle_fetch())
+                else:
+                    asyncio.ensure_future(self._do_keypair_bundle_fetch())
+            elif mtype == MNP.JOIN_REQUEST and self._nonce_node:
+                # Valid both before the GEK proof (a new member has no GEK to prove
+                # with) and after it (an operator pairing a browser is already
+                # connected). Authority comes from the pairing code and the
+                # signature, never from the session state.
+                asyncio.ensure_future(self._do_join_request(msg))
             elif self._user_id is None:
                 self._send({"type": "error", "detail": "Handshake required"})
             elif mtype == MNP.INDEX_SYNC:
@@ -213,17 +305,21 @@ class WebRTCPeerSession:
                 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.INVITE_CREATE:
+                self._do_invite_create(msg)
             elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
                 asyncio.ensure_future(self._do_keypair_bundle_store(msg))
+            elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
+                asyncio.ensure_future(self._do_keypair_bundle_delete())
             elif mtype == MNP.STREAM_REQUEST:
                 asyncio.ensure_future(self._stream_video(msg))
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
         except Exception as e:
-            log.error("Error handling %s on DataChannel: %s", mtype, e)
-            self._send({"type": "error", "detail": str(e)})
+            # Log the detail locally; send the peer a generic message. Exception
+            # text here carries filesystem paths and internal state (finding L3).
+            log.error("Error handling %s on DataChannel: %s", mtype, e, exc_info=True)
+            self._send({"type": "error", "detail": "Request failed"})
 
     def _audit(self, event: str, detail: str = "") -> None:
         audit = self._ctx.get("audit_store")
@@ -239,57 +335,73 @@ class WebRTCPeerSession:
                 detail=detail,
             ))
 
+    def _channel_binding(self) -> bytes:
+        """Both DTLS fingerprints, so a proof is valid on this connection only."""
+        offer_fp = b""
+        answer_fp = b""
+        if self._pc.remoteDescription:
+            offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp)
+        if self._pc.localDescription:
+            answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp)
+        if not offer_fp or not answer_fp:
+            return b""
+        return webrtc_binding(offer_fp, answer_fp)
+
     def _do_handshake(self, msg: dict) -> None:
-        token = msg.get("token", "")
         group_id = msg.get("group_id", "")
         try:
-            decoded = jwt.decode(token, self._ctx["hub_pk_pem"], algorithms=["EdDSA"])
-        except Exception as e:
-            self._send({"type": "error", "detail": f"Invalid JWT: {e}"})
-            self._audit_auth_failed(group_id, str(e))
-            return
-
-        denylist = self._ctx.get("denylist")
-        if denylist and denylist.is_denied(decoded.get("sub", ""), decoded.get("jti", "")):
-            self._send({"type": "error", "detail": "Token revoked"})
-            return
-
-        if group_id and group_id not in decoded.get("groups", []):
-            self._send({"type": "error", "detail": "Not a member of this group"})
+            peer = authorize_token(
+                msg.get("token", ""),
+                self._ctx["hub_pk_pem"],
+                group_id=group_id,
+                hosted_groups=self._ctx.get("groups"),
+                denylist=self._ctx.get("denylist"),
+            )
+        except HandshakeError as refusal:
+            # HandshakeError messages are authored to be peer-safe, unlike arbitrary
+            # exception text (L3) — the client needs to know *why* it was refused.
+            self._send({"type": "error", "detail": str(refusal),
+                        "code": getattr(refusal, "code", "")})
+            self._audit_auth_failed(group_id, str(refusal))
             return
 
-        if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]:
-            self._send({"type": "error", "detail": "Group not hosted on this node"})
+        try:
+            self._nonce_client = base64.b64decode(msg.get("nonce", ""))
+        except Exception:
+            self._nonce_client = b""
+        if len(self._nonce_client) < NONCE_LEN:
+            # The client nonce is what makes the NODE's proof fresh (C3). Without
+            # it a recorded ack could be replayed by an impersonating peer.
+            self._send({"type": "error", "detail": "Client nonce required"})
             return
 
-        # Store decoded JWT data but DO NOT set self._user_id yet —
-        # the user is not authenticated until they prove GEK possession.
-        self._pending_sub = decoded["sub"]
-        self._pending_group = group_id
-        self._pending_username = decoded.get("username", "")
-        self._pending_pk_user = decoded.get("pk_user", "")
+        # Decoded, but NOT authenticated: that happens on the GEK proof.
+        self._pending_sub = peer.user_id
+        self._pending_group = peer.group_id
+        self._pending_username = peer.username
 
-        ctx = self._ctx
-        if "groups" in ctx and group_id:
-            gctx = ctx["groups"].get(group_id, ctx)
-        else:
-            gctx = ctx
-        gek = gctx.get("gek")
-
-        nonce = os.urandom(32)
-        self._gek_challenge = nonce
-        challenge = {
-            "type": MNP.HANDSHAKE_CHALLENGE,
-            "v": MNP_VERSION,
-            "nonce": base64.b64encode(nonce).decode(),
-        }
-        if not gek:
+        gctx = self._ctx["groups"][peer.group_id] if "groups" in self._ctx else self._ctx
+        if not gctx.get("gek"):
             self._send({
                 "type": "error",
                 "detail": "Group encryption not initialized — contact node operator",
             })
             return
-        self._send(challenge)
+
+        self._gek_challenge = os.urandom(NONCE_LEN)
+        self._nonce_node = self._gek_challenge
+        self._send({
+            "type": MNP.HANDSHAKE_CHALLENGE,
+            "v": MNP_VERSION,
+            "nonce": base64.b64encode(self._gek_challenge).decode(),
+            # Announced here because a first-time joiner needs it *before* the
+            # ack: join_request signs a transcript naming this node, and someone
+            # who has never held the GEK cannot complete the handshake to learn
+            # it. Unverified at this point — the ack proves it, the client checks
+            # the two match, and a wrong value only makes our own verification
+            # fail. It is never a substitute for the ack's proof and signature.
+            "node_pk": self._node_pk_b64(),
+        })
 
     def _do_handshake_response(self, msg: dict) -> None:
         if not self._gek_challenge or not hasattr(self, "_pending_sub"):
@@ -297,61 +409,71 @@ class WebRTCPeerSession:
             return
 
         group_id = self._pending_group
-        ctx = self._ctx
-        if "groups" in ctx and group_id:
-            gctx = ctx["groups"].get(group_id, ctx)
-        else:
-            gctx = ctx
+        gctx = self._ctx["groups"][group_id] if "groups" in self._ctx else self._ctx
         gek = gctx.get("gek")
-
         if not gek:
             self._send({"type": "error", "detail": "Group encryption not initialized"})
             self._gek_challenge = None
             return
 
-        proof = msg.get("proof", "")
         try:
-            proof_bytes = base64.b64decode(proof)
+            proof_bytes = base64.b64decode(msg.get("proof", ""))
         except Exception:
             self._send({"type": "error", "detail": "Invalid proof encoding"})
             return
 
-        offer_fp = b""
-        answer_fp = b""
-        if self._pc.remoteDescription:
-            offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp)
-        if self._pc.localDescription:
-            answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp)
+        binding = self._channel_binding()
+        if not binding:
+            # Refuse rather than fall back to an unbound proof (L4).
+            self._send({"type": "error", "detail": "Channel binding unavailable"})
+            self._gek_challenge = None
+            self._audit_auth_failed(group_id, "no channel binding")
+            return
 
-        data = self._gek_challenge + offer_fp + answer_fp
-        expected = hmac.new(gek, data, hashlib.sha256).digest()
-        if not hmac.compare_digest(proof_bytes, expected):
+        if not verify_proof(gek, proof_bytes, ROLE_CLIENT, group_id,
+                            self._nonce_client, self._gek_challenge, binding):
             self._send({"type": "error", "detail": "GEK proof failed"})
             self._gek_challenge = None
             self._audit_auth_failed(group_id, "GEK HMAC mismatch")
             return
 
+        self._complete_handshake(gek, binding)
         self._gek_challenge = None
-        self._complete_handshake()
 
-    def _complete_handshake(self) -> None:
+    def _complete_handshake(self, gek: bytes, binding: bytes) -> None:
+        # Authenticated peers may send large frames (file uploads); unauthenticated
+        # ones may not (H6).
+        self._buffer.max_message = MAX_MSG
         self._user_id = self._pending_sub
         self._group_id = self._pending_group
         self._username = self._pending_username
-        self._pk_user = self._pending_pk_user
+        asyncio.ensure_future(self._load_pinned_pk())
 
-        peers = self._ctx.get("_peers")
-        if peers is not None:
-            peers[self._user_id] = self
+        self._peer_registry()[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],
                  self._group_id[:8] if self._group_id else "none")
+        # The node proves itself too (C3): possession of the GEK over the client's
+        # nonce, plus a signature over the same transcript with its long-term key.
+        # Previously the client received an unverifiable node_pk and trusted
+        # is_node_admin from whoever answered — so a peer that had hijacked
+        # signaling could serve a forged index, chat history and permissions.
+        node_transcript = handshake_transcript(
+            ROLE_NODE, self._group_id or "", self._nonce_client,
+            self._gek_challenge or b"", binding)
+        node_proof = make_proof(
+            gek, ROLE_NODE, self._group_id or "", self._nonce_client,
+            self._gek_challenge or b"", binding)
+
         ack = {
             "type": MNP.HANDSHAKE_ACK,
             "v": MNP_VERSION,
             "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()),
+            "proof": base64.b64encode(node_proof).decode(),
+            "sig": base64.b64encode(
+                self._ctx["sk_node"].sign(node_transcript)).decode(),
             "is_node_admin": bool(node_user_id and self._user_id == node_user_id),
         }
         if node_user_id:
@@ -388,65 +510,41 @@ class WebRTCPeerSession:
         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", "")
+    def _do_invite_create(self, msg: dict) -> None:
+        """
+        Issue a one-time pairing code for someone the operator wants to admit.
 
-        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"})
+        Replaces the old invite path, where the inviter fetched the invitee's
+        public key from the hub and wrapped the group key for whatever came back
+        (H3). The node now needs nothing but a name: it will wrap the key itself,
+        later, for a key the invitee proves they hold.
+        """
+        roster = self._ctx.get("roster")
+        if roster is None:
+            self._send({"type": "error", "detail": "Roster not available"})
             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:
+        invitee_id = msg.get("user_id", "")
+        group_id = msg.get("group_id") or self._group_id
+        if not invitee_id or not group_id:
+            self._send({"type": "error", "detail": "Missing user_id or group_id"})
             return
-
-        bundle = await bundle_store.fetch(group_id, user_id)
-        if not bundle:
+        if group_id != self._group_id:
+            self._send({"type": "error", "detail": "Wrong group for this session"})
             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)
+        if not self._has_admin_authority():
+            self._send({
+                "type": "error",
+                "detail": "No operator paired — run `meshbay-node operator pair`",
+            })
             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)")
+        self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, {
+            "group_id": group_id,
+            "user_id": invitee_id,
+            "username": str(msg.get("username", ""))[:64],
+        })
 
     async def _do_keypair_bundle_fetch(self) -> None:
         """Serve the caller's encrypted keypair bundle during the handshake window."""
@@ -491,6 +589,287 @@ class WebRTCPeerSession:
             "detail": "keypair_bundle_stored",
         })
 
+    # ── Pairing and join (H3, M3) ────────────────────────────────────────────
+
+    def _join_refuse(self, reason: str, audit_detail: str = "") -> None:
+        self._join_attempts += 1
+        # Node-wide window, shared across connections: reconnecting must not reset
+        # the budget.
+        now = time.time()
+        failures = [t for t in self._ctx.get("join_failures", [])
+                    if now - t < JOIN_FAILURE_WINDOW]
+        failures.append(now)
+        self._ctx["join_failures"] = failures
+        self._audit_join("join_refused", audit_detail or reason)
+        self._send({
+            "type": MNP.JOIN_RESULT,
+            "v": MNP_VERSION,
+            "ok": False,
+            "reason": reason,
+        })
+
+    def _audit_join(self, event: str, detail: str) -> None:
+        audit = self._ctx.get("audit_store")
+        if not audit:
+            return
+        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
+        asyncio.ensure_future(audit.log_event(
+            user_id=self._user_id or getattr(self, "_pending_sub", "unknown"),
+            event=event,
+            ip=self._remote_ip,
+            username=self._username or getattr(self, "_pending_username", ""),
+            group_id=self._group_id or getattr(self, "_pending_group", "") or "",
+            detail=detail,
+        ))
+
+    async def _do_join_request(self, msg: dict) -> None:
+        """
+        Pin an identity, or recognise one already pinned.
+
+        The client signs its own Ed25519 and X25519 keys together with the node's
+        nonce, so the identity key vouches for the encryption key — that is what
+        will make it safe for the node to wrap the GEK for a key that arrived over
+        the wire instead of one fetched from the hub's directory (H3).
+
+        A first pairing needs a one-time code, which the hub never sees. Afterwards
+        the pin is the credential and a changed key is refused outright, the same
+        rule the client applies to `pk_node` (11.5.8).
+        """
+        roster = self._ctx.get("roster")
+        if roster is None:
+            self._send({"type": "error", "detail": "Roster not available"})
+            return
+
+        if self._join_attempts >= MAX_JOIN_ATTEMPTS:
+            self._send({"type": "error", "detail": "Too many attempts"})
+            return
+
+        now = time.time()
+        recent = [t for t in self._ctx.get("join_failures", [])
+                  if now - t < JOIN_FAILURE_WINDOW]
+        if len(recent) >= MAX_JOIN_FAILURES_WINDOW:
+            self._audit_join("join_throttled", f"{len(recent)} failures in window")
+            self._send({"type": "error", "detail": "Pairing temporarily locked"})
+            return
+
+        user_id = self._user_id or getattr(self, "_pending_sub", "")
+        username = self._username or getattr(self, "_pending_username", "")
+        if not user_id:
+            self._send({"type": "error", "detail": "Handshake required"})
+            return
+
+        pk_ed_b64 = msg.get("pk_ed25519", "")
+        pk_x_b64 = msg.get("pk_x25519", "")
+        code = msg.get("code", "")
+        ts = msg.get("ts", 0)
+
+        try:
+            pk_ed_raw = base64.b64decode(pk_ed_b64)
+            pk_x_raw = base64.b64decode(pk_x_b64)
+            if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32:
+                raise ValueError
+            pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw)
+        except Exception:
+            self._join_refuse("invalid_keys")
+            return
+
+        if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL:
+            self._join_refuse("stale_request")
+            return
+
+        # An empty group_id means operator pairing, which is node-wide. Anything
+        # else must be the group this connection authenticated to — a signature
+        # obtained for one group must not name another.
+        group_id = msg.get("group_id", "") or ""
+        session_group = self._group_id or getattr(self, "_pending_group", "") or ""
+        if group_id and group_id != session_group:
+            self._join_refuse("group_mismatch")
+            return
+
+        transcript = join_transcript(
+            node_pk_b64=self._node_pk_b64(),
+            group_id=group_id,
+            user_id=user_id,
+            pk_ed25519_b64=pk_ed_b64,
+            pk_x25519_b64=pk_x_b64,
+            nonce_node=self._nonce_node,
+            ts=ts,
+        )
+        try:
+            sig = base64.b64decode(msg.get("sig", ""))
+        except Exception:
+            self._join_refuse("invalid_signature_encoding")
+            return
+        if not self._verify_sig(pk_ed, transcript, sig):
+            self._join_refuse("signature_invalid")
+            return
+
+        known = await roster.get_identity(user_id)
+        if known:
+            if known["pk_ed25519"] != pk_ed_b64 or known["pk_x25519"] != pk_x_b64:
+                # The blocking warning, raised where it matters: whoever this is
+                # holds a different key than the person the operator paired.
+                self._join_refuse(
+                    "key_changed",
+                    f"pinned={known['pk_ed25519'][:16]} presented={pk_ed_b64[:16]}")
+                return
+            # An operator's row is node-wide (empty group), so a lookup for the
+            # group they happen to be opening finds nothing. Fall back to it, or
+            # the client is told it has no role on a node it administers.
+            member = (await roster.get_member(group_id, user_id)
+                      or await roster.get_member("", user_id))
+            await self._join_ok(
+                user_id, pk_x_raw, session_group,
+                role=member["role"] if member else "",
+                recognised=True,
+            )
+            return
+
+        if not code:
+            if self._group_join_policy(session_group) == "open":
+                # An open-join group admits anyone the hub calls a member, so a
+                # code would protect nothing — the hub can walk in through the
+                # front door. Pin what turns up and say so in the audit log.
+                await self._pin_and_admit(
+                    roster, user_id, username, pk_ed_b64, pk_x_b64,
+                    group_id=session_group, role=ROLE_MEMBER,
+                    approved_by="open-join", via="tofu")
+                await self._join_ok(user_id, pk_x_raw, session_group,
+                                    role=ROLE_MEMBER, recognised=False)
+                return
+            self._join_refuse("code_required")
+            return
+
+        invite = await roster.consume_invite(code, user_id)
+        if not invite:
+            self._join_refuse("code_invalid")
+            return
+
+        await self._pin_and_admit(
+            # The name comes from the invitation, not from the token: the hub does
+            # not put a username claim in a JWT, so pinning from the session alone
+            # left the roster nameless and `member revoke <name>` unable to match.
+            roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64,
+            group_id=invite["group_id"], role=invite["role"],
+            approved_by=invite["created_by"], via="code")
+        # The roster row comes from the invitation; the key comes from the
+        # connection. An operator pairing is node-wide (empty group), but they
+        # redeemed the code while opening a group and expect to read it — and
+        # is_authorized() already grants an operator every group on this node.
+        await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"],
+                            role=invite["role"], recognised=False)
+
+    def _group_join_policy(self, group_id: str) -> str:
+        """
+        Admission policy for a group, read from the node's own configuration.
+
+        Never from the hub: a hub that could declare a group open would be handed
+        the key to it (§3.4 of docs/invite-pairing-v1.md).
+        """
+        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
+        return gctx.get("join_policy", "invite")
+
+    async def _pin_and_admit(
+        self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str,
+        *, group_id: str, role: str, approved_by: str, via: str,
+    ) -> None:
+        await roster.pin_identity(
+            user_id=user_id, username=username,
+            pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via,
+        )
+        await roster.set_member(
+            group_id=group_id, user_id=user_id, role=role,
+            status="active", approved_by=approved_by,
+        )
+        if role == ROLE_OPERATOR:
+            self._ctx["has_admin_authority"] = True
+
+        log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role)
+        self._audit_join("join_pinned", f"role={role} via={via}")
+
+    async def _join_ok(
+        self, user_id: str, pk_x_raw: bytes, group_id: str,
+        *, role: str, recognised: bool,
+    ) -> None:
+        """
+        Answer a join, wrapping the group key for the key the caller just proved.
+
+        This is the H3 fix. The inviter used to fetch the invitee's public key from
+        the hub and wrap the GEK for whatever came back, so a hub that answered
+        with its own key was handed the group key by an honest member following the
+        protocol exactly. The node now wraps for a key that arrived from its owner
+        over an authenticated channel, bound to a pinned identity.
+        """
+        reply = {
+            "type": MNP.JOIN_RESULT,
+            "v": MNP_VERSION,
+            "ok": True,
+            "recognised": recognised,
+            "role": role,
+        }
+
+        roster = self._ctx["roster"]
+        if group_id and not await roster.is_authorized(group_id, user_id):
+            # Pinned on this node, but not admitted to this group. Hub membership
+            # alone must not produce a key.
+            reply["gek"] = False
+            reply["reason"] = "not_authorized_for_group"
+            self._send(reply)
+            self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized")
+            return
+
+        gctx = (self._ctx.get("groups") or {}).get(group_id) or {}
+        gek = gctx.get("gek")
+        if not gek:
+            reply["gek"] = False
+            reply["reason"] = "no_gek"
+            self._send(reply)
+            return
+
+        bundle = wrap_gek_aes(gek, pk_x_raw)
+        reply["gek"] = True
+        reply["pk_eph_b64"] = bundle["pk_eph_b64"]
+        reply["nonce_b64"] = bundle["nonce_b64"]
+        reply["wrapped_b64"] = bundle["wrapped_b64"]
+        self._send(reply)
+        self._audit_join("gek_wrapped", f"group={group_id[:8]}")
+
+    async def _do_keypair_bundle_delete(self) -> None:
+        """
+        Withdraw our own key backup from this node.
+
+        Only ever our own: the user_id comes from the authenticated session, never
+        from the message. Someone who does not want a second browser should not be
+        leaving a PBKDF2-protected blob on every node they have ever joined (C4),
+        and turning the setting off has to remove what is already there — not just
+        stop adding to it.
+        """
+        bundle_store = self._ctx.get("bundle_store")
+        if not bundle_store:
+            self._send({"type": "error", "detail": "Bundle store not available"})
+            return
+
+        removed = await bundle_store.delete_keypair(self._user_id)
+        if removed:
+            log.info("Keypair bundle withdrawn by user=%s", self._user_id[:8])
+            self._audit("keypair_bundle_delete")
+        self._send({"type": "ack", "v": MNP_VERSION,
+                    "detail": "keypair_bundle_deleted", "removed": removed})
+
+    def _audit_pre_proof_fetch(self, mtype: str) -> None:
+        """Record bundle access made before the GEK proof (C4)."""
+        audit = self._ctx.get("audit_store")
+        if not audit:
+            return
+        self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
+        asyncio.ensure_future(audit.log_event(
+            user_id=getattr(self, "_pending_sub", "unknown"),
+            event="pre_proof_fetch",
+            ip=self._remote_ip,
+            group_id=getattr(self, "_pending_group", "") or "",
+            detail=mtype,
+        ))
+
     def _audit_auth_failed(self, group_id: str, reason: str) -> None:
         audit = self._ctx.get("audit_store")
         if audit:
@@ -508,6 +887,20 @@ class WebRTCPeerSession:
             return self._ctx["groups"][self._group_id]
         return self._ctx
 
+    def _peer_registry(self) -> dict:
+        """
+        Connected peers for THIS group only.
+
+        Finding H1: this used to live on the shared transport context, so a chat
+        message was broadcast to every peer on the node regardless of which group
+        they had authenticated to.
+        """
+        return self._group_ctx().setdefault("_peers", {})
+
+    def _user_names(self) -> dict:
+        """Display-name cache, per group — same leak as _peer_registry (H1)."""
+        return self._group_ctx().setdefault("_user_names", {})
+
     def _do_index_sync(self) -> None:
         ctx = self._group_ctx()
         idx = ctx["index"]
@@ -555,6 +948,17 @@ class WebRTCPeerSession:
             self._audit("file_download", entry.name)
 
     def _do_stream_segment(self, msg: dict) -> None:
+        asyncio.ensure_future(self._do_stream_segment_async(msg))
+
+    async def _do_stream_segment_async(self, msg: dict) -> None:
+        """
+        Legacy HLS segment extraction (superseded by stream_req/MSE).
+
+        Finding H6: this ran subprocess.run(..., timeout=30) directly inside the
+        event loop, so a single request stalled the whole daemon — every peer,
+        every group — for up to thirty seconds. Now async and under the same
+        transcode semaphore as _stream_video.
+        """
         ctx = self._group_ctx()
         file_id = msg["file_id"]
         segment_index = msg["segment_index"]
@@ -570,21 +974,34 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not on disk"})
             return
 
-        import subprocess
+        sem = self._ctx.get("_transcode_sem")
+        if sem is None:
+            sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES)
+            self._ctx["_transcode_sem"] = sem
+
         try:
-            result = subprocess.run(
-                ["ffmpeg", "-hide_banner", "-loglevel", "error",
-                 "-ss", str(segment_index * segment_duration),
-                 "-i", str(file_path),
-                 "-t", str(segment_duration),
-                 "-c:v", "copy", "-c:a", "copy",
-                 "-f", "mpegts", "pipe:1"],
-                capture_output=True, timeout=30,
-            )
-            if result.returncode != 0 or not result.stdout:
+            async with sem:
+                proc = await asyncio.create_subprocess_exec(
+                    "ffmpeg", "-hide_banner", "-loglevel", "error",
+                    "-ss", str(segment_index * segment_duration),
+                    "-i", str(file_path),
+                    "-t", str(segment_duration),
+                    "-c:v", "copy", "-c:a", "copy",
+                    "-f", "mpegts", "pipe:1",
+                    stdout=asyncio.subprocess.PIPE,
+                    stderr=asyncio.subprocess.DEVNULL,
+                )
+                try:
+                    stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=30)
+                except asyncio.TimeoutError:
+                    proc.kill()
+                    await proc.wait()
+                    self._send({"type": "error", "detail": "Segment extraction timed out"})
+                    return
+            if proc.returncode != 0 or not stdout:
                 self._send({"type": "error", "detail": "Segment extraction failed"})
                 return
-            segment_data = result.stdout
+            segment_data = stdout
         except Exception:
             self._send({"type": "error", "detail": "Segment extraction failed"})
             return
@@ -599,11 +1016,14 @@ class WebRTCPeerSession:
         })
 
     def _do_chat_message(self, msg: dict) -> None:
-        chat_store = self._ctx.get("chat_store")
+        # Per-group store — see _peer_registry() and finding H1. Reading chat_store
+        # off the shared transport context sent every group's messages to the first
+        # group's database, and served them back to anyone on the node.
+        chat_store = self._group_ctx().get("chat_store")
         payload = msg.get("payload", "")
         sender_name = msg.get("sender_name", "")
         if sender_name:
-            self._ctx.setdefault("_user_names", {})[self._user_id] = sender_name
+            self._user_names()[self._user_id] = sender_name
         if chat_store:
             raw = payload.encode() if isinstance(payload, str) else payload
             asyncio.ensure_future(chat_store.save_message(
@@ -614,7 +1034,7 @@ class WebRTCPeerSession:
                 sender_name=sender_name,
             ))
 
-        peers = self._ctx.get("_peers", {})
+        peers = self._peer_registry()
         broadcast = {
             "type": MNP.CHAT_MESSAGE,
             "v": MNP_VERSION,
@@ -647,7 +1067,7 @@ class WebRTCPeerSession:
         self._audit("chat_message")
 
     def _do_chat_history(self, msg: dict) -> None:
-        chat_store = self._ctx.get("chat_store")
+        chat_store = self._group_ctx().get("chat_store")
         if not chat_store:
             self._send({
                 "type": MNP.CHAT_HISTORY_RESPONSE,
@@ -662,7 +1082,7 @@ class WebRTCPeerSession:
 
     async def _send_chat_history(self, chat_store, since: float, limit: int) -> None:
         msgs = await chat_store.get_messages(since=since, limit=limit)
-        names = self._ctx.get("_user_names", {})
+        names = self._user_names()
         self._send({
             "type": MNP.CHAT_HISTORY_RESPONSE,
             "v": MNP_VERSION,
@@ -691,24 +1111,55 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "Missing filename or data"})
             return
 
+        if not SAFE_UPLOAD_NAME.match(filename):
+            self._send({"type": "error", "detail": "Invalid filename"})
+            return
+
         shared_root = ctx.get("shared_root")
         if not shared_root:
             self._send({"type": "error", "detail": "No shared directory"})
             return
 
-        upload_dir = shared_root / ".uploads"
-        upload_dir.mkdir(exist_ok=True)
-        safe_name = filename.replace("/", "_").replace("\\", "_").replace("..", "_")
-        tmp_path = upload_dir / f"{safe_name}.part"
+        # Per-user quarantine: a member can only ever write inside their own directory,
+        # so they cannot overwrite the operator's files or another member's (C5a).
+        rel_dir = f"{UPLOAD_DIR_NAME}/{self._user_id}"
+        user_dir = shared_root / UPLOAD_DIR_NAME / self._user_id
+        user_dir.mkdir(parents=True, exist_ok=True)
+        tmp_path = user_dir / f"{filename}.part"
+        final_path = user_dir / filename
+
+        state = self._uploads.get(filename)
+        if chunk_index == 0:
+            if final_path.exists():
+                self._send({"type": "error", "detail": "File already exists"})
+                return
+            state = {"next_index": 0, "bytes": 0}
+            self._uploads[filename] = state
+        elif state is None:
+            self._send({"type": "error", "detail": "Upload not started"})
+            return
+
+        # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
+        # blindly to whatever .part file is already on disk.
+        if chunk_index != state["next_index"]:
+            self._send({"type": "error", "detail": "Unexpected chunk index"})
+            return
 
         if isinstance(data, str):
             chunk_bytes = base64.b64decode(data)
         else:
             chunk_bytes = bytes(data)
 
-        mode = "ab" if chunk_index > 0 else "wb"
-        with open(tmp_path, mode) as f:
+        if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES:
+            self._uploads.pop(filename, None)
+            tmp_path.unlink(missing_ok=True)
+            self._send({"type": "error", "detail": "Upload exceeds size limit"})
+            return
+
+        with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f:
             f.write(chunk_bytes)
+        state["next_index"] = chunk_index + 1
+        state["bytes"] += len(chunk_bytes)
 
         self._send({
             "type": MNP.FILE_UPLOAD_ACK,
@@ -718,21 +1169,30 @@ class WebRTCPeerSession:
         })
 
         if chunk_index + 1 >= total_chunks:
-            final_path = shared_root / safe_name
+            self._uploads.pop(filename, None)
             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)
+            log.info("Upload complete: %s (%d chunks, %d bytes)",
+                     filename, total_chunks, state["bytes"])
+            self._audit("file_upload", f"{rel_dir}/{filename}")
+            self._register_uploader(ctx, rel_dir, filename)
+
+    def _register_uploader(self, ctx: dict, rel_dir: str, filename: str) -> None:
+        """
+        Tag the index entry with the uploader's identity after upload completes.
 
-    def _register_uploader(self, ctx: dict, filename: str) -> None:
-        """Tag the index entry with the uploader's user_id after upload completes."""
+        The key recorded here is the one this node pinned, not the one the token
+        carried. `pk_user` was a hub-chosen claim, and it decided who could later
+        delete the file: a hub issuing a token naming its own key could delete
+        anyone's uploads on any node. Deletion is supposed to be authorized by the
+        node, and this closes the last place where it was not.
+        """
         idx = ctx.get("index")
         if not idx:
             return
         for entry in idx.entries:
-            if entry.name == filename and entry.path == "":
+            if entry.name == filename and entry.path == rel_dir:
                 entry.uploader_id = self._user_id
-                entry.uploader_pk = self._pk_user
+                entry.uploader_pk = self._pinned_pk
                 return
 
     def _do_file_delete(self, msg: dict) -> None:
@@ -747,28 +1207,115 @@ class WebRTCPeerSession:
             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:
+        if not self._has_admin_authority() 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._issue_admin_challenge(OP_FILE_DELETE, file_id)
+
+    # ── Admin operation challenge/response (finding H5) ──────────────────────
+
+    def _node_pk_b64(self) -> str:
+        return pk_to_b64(self._ctx["sk_node"].public_key())
+
+    def _issue_admin_challenge(
+        self, op: str, subject: str, payload: dict | None = None,
+    ) -> None:
+        """
+        Ask the client to authorize `op` on `subject` with its Ed25519 identity key.
+
+        The client is sent the transcript *fields*, not opaque bytes, so it can
+        rebuild and inspect what it signs. The node keeps the authoritative copy and
+        rebuilds the transcript itself at verification time — nothing signed is ever
+        taken from the response message.
+        """
+        nonce = os.urandom(32)
+        ts = int(time.time())
+        op_id = base64.b64encode(os.urandom(16)).decode()
+        self._admin_ops[op_id] = {
+            "op": op, "subject": subject, "nonce": nonce, "ts": ts,
+            "payload": payload or {},
+        }
         self._send({
             "type": MNP.ADMIN_CHALLENGE,
             "v": MNP_VERSION,
-            "challenge": base64.b64encode(challenge).decode(),
-            "file_id": file_id,
+            "op_id": op_id,
+            "op": op,
+            "subject": subject,
+            "nonce": base64.b64encode(nonce).decode(),
+            "ts": ts,
+            "node_pk": self._node_pk_b64(),
+            "group_id": self._group_id or "",
         })
 
+    @staticmethod
+    def _verify_sig(pk: Ed25519PublicKey | None, transcript: bytes, sig: bytes) -> bool:
+        if pk is None:
+            return False
+        try:
+            pk.verify(sig, transcript)
+            return True
+        except Exception:
+            return False
+
+    async def _load_pinned_pk(self) -> None:
+        """Remember which key this node pinned for the peer we just authenticated."""
+        roster = self._ctx.get("roster")
+        if roster is None or not self._user_id:
+            return
+        ident = await roster.get_identity(self._user_id)
+        if ident:
+            self._pinned_pk = ident["pk_ed25519"]
+
+    def _has_admin_authority(self) -> bool:
+        """
+        Cheap synchronous pre-check: is there anyone who could authorize this?
+
+        Only decides whether to issue a challenge at all — the gate is
+        `_verify_admin_sig`. The flag is set at startup and refreshed in-process
+        when an operator pairs.
+        """
+        return bool(self._ctx.get("admin_pk_ed25519")
+                    or self._ctx.get("has_admin_authority"))
+
+    async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool:
+        """
+        Check a signature against every key holding node-operator authority.
+
+        Read from the roster on each call rather than cached: revoking a paired
+        browser must take effect immediately, and admin operations are rare enough
+        that a SQLite read costs nothing. `admin_pk_ed25519` in node.toml is still
+        honoured so an existing deployment keeps working until its operator pairs
+        (M3) — it is the legacy form of the same statement.
+        """
+        legacy = self._ctx.get("admin_pk_ed25519")
+        if self._verify_sig(legacy, transcript, sig):
+            return True
+
+        roster = self._ctx.get("roster")
+        if roster is None:
+            return False
+        for pk_b64 in await roster.operator_pks():
+            try:
+                pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64))
+            except Exception:
+                continue
+            if self._verify_sig(pk, transcript, sig):
+                return True
+        return False
+
     def _do_admin_response(self, msg: dict) -> None:
-        file_id = msg.get("file_id", "")
+        op_id = msg.get("op_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"})
+        pending = self._admin_ops.pop(op_id, None)
+        if not pending:
+            self._send({"type": "error", "detail": "No pending admin operation"})
+            return
+
+        if time.time() - pending["ts"] > ADMIN_CHALLENGE_TTL:
+            self._send({"type": "error", "detail": "Admin challenge expired"})
             return
 
         try:
@@ -777,40 +1324,96 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "Invalid signature encoding"})
             return
 
+        transcript = admin_transcript(
+            op=pending["op"],
+            node_pk_b64=self._node_pk_b64(),
+            group_id=self._group_id or "",
+            subject=pending["subject"],
+            nonce=pending["nonce"],
+            ts=pending["ts"],
+        )
+
+        if pending["op"] == OP_FILE_DELETE:
+            asyncio.ensure_future(
+                self._admin_exec_file_delete(pending, transcript, sig_bytes))
+        elif pending["op"] == OP_INVITE_CREATE:
+            asyncio.ensure_future(
+                self._admin_exec_invite_create(pending, transcript, sig_bytes))
+        else:
+            self._send({"type": "error", "detail": "Unknown admin operation"})
+
+    async def _admin_exec_file_delete(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        file_id = pending["subject"]
         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:
+        uploader_pk = None
+        if entry.uploader_pk:
             try:
-                uploader_key = Ed25519PublicKey.from_public_bytes(
+                uploader_pk = Ed25519PublicKey.from_public_bytes(
                     base64.b64decode(entry.uploader_pk))
-                uploader_key.verify(sig_bytes, challenge)
-                verified = True
             except Exception:
-                pass
+                uploader_pk = None
 
-        if not verified:
+        # Node operator, or the user who uploaded this file — verified by the key
+        # recorded at upload time, never by a JWT claim (the hub controls those).
+        if not (await self._verify_admin_sig(transcript, sig)
+                or self._verify_sig(uploader_pk, transcript, sig)):
             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)
 
+    async def _admin_exec_invite_create(
+        self, pending: dict, transcript: bytes, sig: bytes,
+    ) -> None:
+        # Node operator only. A group admin who does not run the node has no
+        # authority over who this node admits (deny by default). Delegation is
+        # designed but deferred — see §6.2 of docs/invite-pairing-v1.md.
+        if not await self._verify_admin_sig(transcript, sig):
+            self._send({"type": "error", "detail": "Signature verification failed"})
+            self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}")
+            return
+
+        roster = self._ctx.get("roster")
+        if roster is None:
+            self._send({"type": "error", "detail": "Roster not available"})
+            return
+
+        payload = pending["payload"]
+        code = await roster.create_invite(
+            group_id=payload["group_id"],
+            user_id=payload["user_id"],
+            role=ROLE_MEMBER,
+            created_by=self._user_id or "",
+            ttl=self._ctx.get("invite_ttl", DEFAULT_INVITE_TTL),
+            username=payload.get("username", ""),
+        )
+        invites = await roster.list_invites()
+        expires = next(
+            (i["expires_at"] for i in invites
+             if i["user_id"] == payload["user_id"]
+             and i["group_id"] == payload["group_id"]), "")
+
+        log.info("Invite created: group=%s user=%s",
+                 payload["group_id"][:8], payload["user_id"][:8])
+        self._audit("invite_create", f"target={payload['user_id'][:8]}")
+        # The code exists in the clear exactly here and in the operator's hands.
+        self._send({
+            "type": MNP.INVITE_RESULT,
+            "v": MNP_VERSION,
+            "code": code,
+            "expires_at": expires,
+            "user_id": payload["user_id"],
+            "username": payload.get("username", ""),
+        })
+
     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():
@@ -827,6 +1430,20 @@ class WebRTCPeerSession:
 
     async def _stream_video(self, msg: dict) -> None:
         """Stream a video file as fMP4 segments via MSE-compatible output."""
+        # One ffmpeg per request with no cap lets any member exhaust the node's
+        # CPU and process table (H6). The semaphore lives on the transport context
+        # so it is shared across all peers, not per-session.
+        sem = self._ctx.get("_transcode_sem")
+        if sem is None:
+            sem = asyncio.Semaphore(MAX_CONCURRENT_TRANSCODES)
+            self._ctx["_transcode_sem"] = sem
+        if sem.locked() and sem._value <= 0:
+            self._send({"type": "error", "detail": "Server busy, retry shortly"})
+            return
+        async with sem:
+            await self._stream_video_inner(msg)
+
+    async def _stream_video_inner(self, msg: dict) -> None:
         ctx = self._group_ctx()
         file_id = msg.get("file_id", "")
         entry = ctx["index"].get_entry(file_id)
@@ -914,9 +1531,8 @@ class WebRTCPeerSession:
 
     async def close(self) -> None:
         self._audit("disconnect")
-        peers = self._ctx.get("_peers")
-        if peers and self._user_id:
-            peers.pop(self._user_id, None)
+        if self._user_id:
+            self._peer_registry().pop(self._user_id, None)
         await self._pc.close()