fix: resource limits, signaling authz, node admin UI token

Phase 11.5 — findings H6, C4 (partial), and milestone 11.5.3.

H6 — resource exhaustion. Several paths let one peer degrade or stall a node:

  * the DataChannel frame limit was a flat 64 MB applied BEFORE authentication,
    so an unauthenticated peer could announce a huge frame and dribble bytes
    into it. Unauthenticated peers now get 64 KB; the large budget is granted
    only after the GEK proof, where it is needed for uploads.
  * _do_stream_segment ran subprocess.run(..., timeout=30) directly in the event
    loop, stalling the entire daemon — every peer, every group — for up to
    thirty seconds per request. Now async, with a timeout and process kill.
  * ffmpeg was spawned per stream request with no cap. Both streaming paths now
    share a transport-wide semaphore.
  * POST /v1/nodes/{id}/webrtc/offer was reachable by any authenticated user for
    any node, with no membership check and no rate limit, making the target node
    allocate an aiortc PeerConnection and gather ICE on demand — remote resource
    exhaustion against a third party's machine. Now rate limited, capped per
    user, SDP size bounded, and the caller must share an active group with the
    node. That also closes the H4 gap where signaling ignored group status.
  * POST /v1/nodes/{id}/incoming took peer_ip verbatim, so any user could make an
    arbitrary node emit UDP packets to an address of their choosing. The probe
    target must now match the caller's own source address.

C4 (partial) — the pre-proof bundle window. GEK and keypair bundle fetches are
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 a hub can reach
by forging a JWT. Bounded to 4 fetches per session and audited as
"pre_proof_fetch". The real fix is removing remote keypair bundles entirely,
which belongs to the native client (Phase 13.3).

11.5.3 — the node admin UI was unauthenticated because it binds loopback. But
any local process can reach it, and so can a page in the operator's browser via
DNS rebinding — and this API re-initialises group keys and reads the audit log.
H2 showed script execution there equals full control. Now gated by a per-run
token, printed at startup, accepted as ?t= or X-MeshBay-Token.

One test needed rewriting rather than adding: the first version asserted
"subprocess.run(" was absent from the source, which also matched the comment
documenting the old behaviour. It now parses the AST and checks the property.

Tests: 121 node, 142 hub+common. Regression suite 47 node + 10 hub.

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

7 files changed, 380 insertions, 32 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 0f1ddba..2e3323c 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -33,7 +33,7 @@ import time
 import uuid
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect
+from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect
 from pydantic import BaseModel
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -274,12 +274,28 @@ class IncomingRequest(BaseModel):
 async def notify_incoming(
     node_id: str,
     body: IncomingRequest,
+    request: Request,
     current_user: User = Depends(get_current_user),
 ):
     """
     Signal a node that a client wants to connect (NAT punch coordination).
     Hub forwards the request via WebSocket; node punches NAT and replies punch_ready.
+
+    Finding H6: peer_ip was taken verbatim, so any authenticated user could make an
+    arbitrary node emit UDP packets to an address of their choosing — a small
+    reflection primitive using someone else's machine. The probe target must now be
+    the caller's own source address.
     """
+    from meshbay_hub.api.netutil import client_ip
+
+    caller_ip = client_ip(request)
+    if body.peer_ip != caller_ip:
+        raise HTTPException(
+            status_code=403,
+            detail="peer_ip must match the requesting address")
+    if not (1 <= body.peer_port <= 65535):
+        raise HTTPException(status_code=422, detail="Invalid peer_port")
+
     ws = _connected_nodes.get(node_id)
     if not ws:
         raise HTTPException(status_code=404, detail="Node not connected")
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
index bd343c9..8f84163 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
@@ -17,11 +17,15 @@ import json
 import logging
 import uuid
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.api.deps import get_current_user
-from meshbay_hub.db.models import User
+from meshbay_hub.api.middleware import limiter
+from meshbay_hub.db.engine import get_db
+from meshbay_hub.db.models import Group, GroupMember, User
 
 log = logging.getLogger(__name__)
 
@@ -41,25 +45,66 @@ class WebRTCOfferResponse(BaseModel):
     peer_id: str
 
 
+MAX_SDP_BYTES = 16 * 1024          # an SDP offer is ~2 KB
+MAX_PENDING_PER_USER = 3           # concurrent in-flight offers per account
+
+_pending_per_user: dict[str, int] = {}
+
+
 @router.post("/{node_id}/webrtc/offer", response_model=WebRTCOfferResponse)
+@limiter.limit("30/minute")
 async def webrtc_offer(
     node_id: str,
     body: WebRTCOfferRequest,
+    request: Request,
     current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
 ):
     """
     Browser sends WebRTC SDP offer for a node. Hub relays via WebSocket.
     Returns the node's SDP answer once received.
+
+    Finding H6: this was reachable by any authenticated user, for any node, with no
+    rate limit and no membership check. Each call makes the node allocate an
+    aiortc RTCPeerConnection and gather ICE, so it was a remote resource-exhaustion
+    primitive against an arbitrary third party's machine.
+
+    Finding H4: it also ignored group status, so "suspend a group" did not stop new
+    connections from being brokered to nodes hosting it.
     """
-    from meshbay_hub.api.revocation import _connected_nodes
+    from meshbay_hub.api.revocation import _connected_nodes, _node_groups
+
+    if len(body.sdp) > MAX_SDP_BYTES:
+        raise HTTPException(status_code=413, detail="SDP too large")
 
     ws = _connected_nodes.get(node_id)
     if not ws:
         raise HTTPException(status_code=404, detail="Node not connected")
 
+    # The caller must share at least one active group with the target node.
+    node_group_ids = set(_node_groups.get(node_id, []))
+    if node_group_ids:
+        result = await db.execute(
+            select(GroupMember.group_id).where(
+                GroupMember.user_id == current_user.id,
+                GroupMember.group_id.in_(node_group_ids),
+            ))
+        shared = [gid for (gid,) in result.all()]
+        if not shared:
+            raise HTTPException(status_code=403, detail="Not a member of any group on this node")
+
+        active = await db.execute(
+            select(Group.id).where(Group.id.in_(shared), Group.status == "active"))
+        if not active.first():
+            raise HTTPException(status_code=403, detail="Group is not active")
+
+    if _pending_per_user.get(current_user.id, 0) >= MAX_PENDING_PER_USER:
+        raise HTTPException(status_code=429, detail="Too many pending connections")
+
     peer_id = str(uuid.uuid4())
     answer_future: asyncio.Future = asyncio.get_event_loop().create_future()
     _webrtc_answers[peer_id] = answer_future
+    _pending_per_user[current_user.id] = _pending_per_user.get(current_user.id, 0) + 1
 
     try:
         await ws.send_text(json.dumps({
@@ -83,6 +128,11 @@ async def webrtc_offer(
         )
     finally:
         _webrtc_answers.pop(peer_id, None)
+        remaining = _pending_per_user.get(current_user.id, 1) - 1
+        if remaining > 0:
+            _pending_per_user[current_user.id] = remaining
+        else:
+            _pending_per_user.pop(current_user.id, None)
 
 
 def handle_webrtc_answer(msg: dict) -> None:
diff --git a/packages/meshbay-hub/tests/test_node_ws_auth.py b/packages/meshbay-hub/tests/test_node_ws_auth.py
index 6db95d7..9ec251d 100644
--- a/packages/meshbay-hub/tests/test_node_ws_auth.py
+++ b/packages/meshbay-hub/tests/test_node_ws_auth.py
@@ -150,6 +150,86 @@ async def test_ws_group_claims_cannot_widen_beyond_membership(client):
 
 
 @pytest.mark.asyncio
+async def test_signaling_rejects_non_member(client):
+    """
+    H6/H4: POST /v1/nodes/{id}/webrtc/offer was reachable by any authenticated
+    user for any node, with no membership check and no rate limit. Each call makes
+    the target node allocate an aiortc PeerConnection and gather ICE, so it was a
+    remote resource-exhaustion primitive against a third party's machine.
+    """
+    from meshbay_hub.api import revocation as rev
+
+    owner = await _make_user(client, "owner8")
+    outsider = await _make_user(client, "outsider8")
+    node_id = await _announce_node(client, owner)
+
+    r = await client.post(
+        "/v1/groups",
+        json={"name": "private-g", "visibility": "private", "join_policy": "invite"},
+        headers={"Authorization": f"Bearer {owner['token']}"},
+    )
+    group_id = r.json()["group_id"]
+
+    # Pretend the node is connected and hosting that group.
+    class _FakeWS:
+        async def send_text(self, _):
+            raise AssertionError("offer relayed to node despite non-membership")
+
+    rev._connected_nodes[node_id] = _FakeWS()
+    rev._node_groups[node_id] = [group_id]
+    try:
+        resp = await client.post(
+            f"/v1/nodes/{node_id}/webrtc/offer",
+            json={"sdp": "v=0", "ice_candidates": []},
+            headers={"Authorization": f"Bearer {outsider['token']}"},
+        )
+        assert resp.status_code == 403, resp.text
+    finally:
+        rev._connected_nodes.pop(node_id, None)
+        rev._node_groups.pop(node_id, None)
+
+
+@pytest.mark.asyncio
+async def test_signaling_rejects_oversized_sdp(client):
+    """H6: an SDP offer is ~2 KB; unbounded input is a memory amplifier."""
+    user = await _make_user(client, "user9")
+    resp = await client.post(
+        "/v1/nodes/whatever/webrtc/offer",
+        json={"sdp": "v=0" + ("x" * 200_000), "ice_candidates": []},
+        headers={"Authorization": f"Bearer {user['token']}"},
+    )
+    assert resp.status_code == 413
+
+
+@pytest.mark.asyncio
+async def test_incoming_rejects_foreign_peer_ip(client):
+    """
+    H6: peer_ip was taken verbatim, letting any user make an arbitrary node emit
+    UDP packets to an address of their choosing — reflection via someone else's
+    machine. The probe target must be the caller's own address.
+    """
+    from meshbay_hub.api import revocation as rev
+
+    owner = await _make_user(client, "owner10")
+    node_id = await _announce_node(client, owner)
+
+    class _FakeWS:
+        async def send_text(self, _):
+            raise AssertionError("punch relayed with attacker-chosen peer_ip")
+
+    rev._connected_nodes[node_id] = _FakeWS()
+    try:
+        resp = await client.post(
+            f"/v1/nodes/{node_id}/incoming",
+            json={"peer_ip": "198.51.100.7", "peer_port": 9999},
+            headers={"Authorization": f"Bearer {owner['token']}"},
+        )
+        assert resp.status_code == 403, resp.text
+    finally:
+        rev._connected_nodes.pop(node_id, None)
+
+
+@pytest.mark.asyncio
 async def test_ws_node_may_narrow_its_group_set(client):
     """A node hosting a subset of the operator's groups may say so."""
     from meshbay_hub.api.revocation import _authorize_node_ws
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py
index 7dedad8..fa8c0ee 100644
--- a/packages/meshbay-node/src/meshbay_node/daemon.py
+++ b/packages/meshbay-node/src/meshbay_node/daemon.py
@@ -26,6 +26,7 @@ import asyncio
 import base64
 import json
 import logging
+import os
 import signal
 import sys
 from pathlib import Path
@@ -130,6 +131,11 @@ class NodeDaemon:
         # 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
+        # Per-run token for the local admin UI (11.5.3). Not a password: it keeps
+        # other local processes and rebound browser pages out of an API that can
+        # re-initialise group keys.
+        ui_token = base64.urlsafe_b64encode(os.urandom(18)).decode().rstrip("=")
+        self._state["ui_token"] = ui_token
         from meshbay_node.ui import create_ui_app
         ui_app = create_ui_app(self._state)
         ui_cfg = uvicorn.Config(
@@ -140,7 +146,8 @@ class NodeDaemon:
         )
         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)
+        log.info("Admin UI at http://127.0.0.1:%d/?t=%s",
+                 self._config.node.ui_port, ui_token)
 
         # 3. Hub connection (Ed25519 auth — retries until node key is linked)
         hub_cfg = HubConfig(
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 43e6fdc..190d580 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -64,6 +64,16 @@ MAX_MSG = 64 * 1024 * 1024
 # 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
 UPLOAD_DIR_NAME = ".uploads"
 # Conservative allowlist: also what keeps markup out of filenames, which the node admin
 # UI used to render unescaped (finding H2).
@@ -137,10 +147,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)
@@ -148,7 +166,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
@@ -180,7 +198,8 @@ 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
@@ -210,10 +229,24 @@ 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 self._user_id is None:
                 self._send({"type": "error", "detail": "Handshake required"})
             elif mtype == MNP.INDEX_SYNC:
@@ -357,6 +390,9 @@ class WebRTCPeerSession:
         self._complete_handshake()
 
     def _complete_handshake(self) -> 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
@@ -496,6 +532,20 @@ class WebRTCPeerSession:
             "detail": "keypair_bundle_stored",
         })
 
+    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:
@@ -574,6 +624,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"]
@@ -589,21 +650,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
@@ -963,6 +1037,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)
diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py
index d2c3429..21e4445 100644
--- a/packages/meshbay-node/src/meshbay_node/ui/app.py
+++ b/packages/meshbay-node/src/meshbay_node/ui/app.py
@@ -9,7 +9,7 @@ FastAPI app providing:
   - API endpoints for all data (JSON)
 
 Served only on 127.0.0.1 — not exposed to the network.
-No authentication required (localhost only).
+Gated by a per-run session token (11.5.3) — printed at daemon startup.
 """
 
 import base64
@@ -37,6 +37,28 @@ def create_ui_app(state: dict) -> FastAPI:
     )
 
     @app.middleware("http")
+    async def _require_session_token(request, call_next):
+        """
+        Gate the admin UI behind a per-run token (11.5.3).
+
+        "localhost only" is weaker than it sounds: any process on the machine can
+        reach it, and a page in the operator's browser can reach it too via DNS
+        rebinding. Since this API can re-initialise a group's GEK and read the
+        audit log, an unauthenticated loopback service is a privilege boundary
+        waiting to be crossed. The token is printed at startup and accepted as
+        ?t= or the X-MeshBay-Token header.
+        """
+        from fastapi.responses import PlainTextResponse
+
+        token = state.get("ui_token")
+        if token:
+            supplied = (request.query_params.get("t")
+                        or request.headers.get("X-MeshBay-Token"))
+            if supplied != token:
+                return PlainTextResponse("Forbidden", status_code=403)
+        return await call_next(request)
+
+    @app.middleware("http")
     async def _security_headers(request, call_next):
         """
         Defence in depth behind the escaping fixes for H2. This UI is unauthenticated
@@ -340,7 +362,7 @@ def create_ui_app(state: dict) -> FastAPI:
 
     @app.get("/audit", response_class=HTMLResponse)
     async def audit_page():
-        return _render_audit_page()
+        return _render_audit_page(state.get("ui_token", ""))
 
     return app
 
@@ -356,6 +378,7 @@ def _fmt_size(n: int) -> str:
 
 
 def _render_page(state: dict) -> str:
+    token_js = json.dumps(state.get("ui_token", ""))
     status = state.get("status", "starting")
     indexes = state.get("indexes", {})
     groups_ctx = state.get("groups_ctx", {})
@@ -552,13 +575,14 @@ def _render_page(state: dict) -> str:
   </div>
 </div>
 <script>
+const TOKEN = {token_js};
 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 resp = await fetch('/api/groups/' + groupId + '/gek?t=' + TOKEN, {{ method: 'POST' }});
     const data = await resp.json();
     if (resp.ok) {{
       if (status) status.textContent = 'GEK initialized — wrapped for '
@@ -582,8 +606,11 @@ setTimeout(()=>location.reload(), 10000);
 </html>"""
 
 
-def _render_audit_page() -> str:
-    return """<!DOCTYPE html>
+def _render_audit_page(token: str = "") -> str:
+    return _AUDIT_HTML.replace("__TOKEN__", json.dumps(token))
+
+
+_AUDIT_HTML = """<!DOCTYPE html>
 <html lang="en">
 <head>
 <meta charset="utf-8">
@@ -619,7 +646,7 @@ def _render_audit_page() -> str:
 <body>
 <div class="container">
   <h1>Audit Log</h1>
-  <nav><a href="/">Dashboard</a><a href="/audit">Audit Log</a></nav>
+  <nav><a id="navHome" href="/">Dashboard</a><a id="navAudit" href="/audit">Audit Log</a></nav>
 
   <div class="filters">
     <select id="eventFilter">
@@ -648,10 +675,11 @@ def _render_audit_page() -> str:
   </table>
 </div>
 <script>
+const TOKEN = __TOKEN__;
 async function load() {
   const ev = document.getElementById('eventFilter').value;
   const limit = document.getElementById('limitSelect').value;
-  let url = '/api/audit?limit=' + limit;
+  let url = '/api/audit?limit=' + limit + (TOKEN ? '&t=' + TOKEN : '');
   if (ev) url += '&event=' + ev;
   const r = await fetch(url);
   const data = await r.json();
@@ -677,6 +705,10 @@ async function load() {
     return tr;
   }));
 }
+for (const [id, href] of [['navHome','/'],['navAudit','/audit']]) {
+  const el = document.getElementById(id);
+  if (el && TOKEN) el.href = href + '?t=' + TOKEN;
+}
 document.getElementById('eventFilter').onchange = load;
 document.getElementById('limitSelect').onchange = load;
 let debounceTimer;
diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py
index 9552848..a6d69c6 100644
--- a/packages/meshbay-node/tests/test_security_regressions.py
+++ b/packages/meshbay-node/tests/test_security_regressions.py
@@ -10,6 +10,7 @@ If one of these starts failing, a fix has been reverted. Do not "fix" the test.
 """
 
 import base64
+import struct
 from pathlib import Path
 
 import pytest
@@ -465,6 +466,80 @@ def test_peer_errors_do_not_leak_internals():
     assert '"detail": str(e)' not in source
 
 
+def test_pre_handshake_message_budget_is_small():
+    """
+    H6: the frame limit was a flat 64 MB applied before authentication, so an
+    unauthenticated peer could announce a huge frame and dribble bytes into it.
+    """
+    from meshbay_node.transport.webrtc_server import (
+        MAX_MSG, PRE_HANDSHAKE_MAX_MSG, _DataChannelBuffer,
+    )
+    assert PRE_HANDSHAKE_MAX_MSG <= 1024 * 1024
+    assert PRE_HANDSHAKE_MAX_MSG < MAX_MSG
+
+    buf = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
+    buf.feed(struct.pack(">I", PRE_HANDSHAKE_MAX_MSG + 1) + b"x")
+    with pytest.raises(ValueError):
+        list(buf.messages())
+
+
+def test_stream_segment_is_not_synchronous():
+    """
+    H6: _do_stream_segment ran subprocess.run(timeout=30) inside the event loop,
+    stalling every peer on the node for up to thirty seconds per request.
+
+    Asserts the property (the worker is a coroutine, ffmpeg is spawned through
+    asyncio) rather than grepping for "subprocess.run" — which also matches the
+    comment that documents the old behaviour.
+    """
+    import ast
+    import inspect
+    from meshbay_node.transport.webrtc_server import WebRTCPeerSession
+
+    assert inspect.iscoroutinefunction(WebRTCPeerSession._do_stream_segment_async)
+
+    source = (Path(__file__).parent.parent / "src" / "meshbay_node"
+              / "transport" / "webrtc_server.py").read_text()
+    tree = ast.parse(source)
+    blocking = [
+        node for node in ast.walk(tree)
+        if isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Attribute)
+        and node.func.attr == "run"
+        and isinstance(node.func.value, ast.Name)
+        and node.func.value.id == "subprocess"
+    ]
+    assert not blocking, "blocking subprocess.run() in the event loop"
+    assert "_transcode_sem" in source, "ffmpeg spawns must be capped"
+
+
+def test_pre_proof_fetches_are_bounded():
+    """C4: the pre-proof bundle window is a disclosure surface; bound it."""
+    from meshbay_node.transport.webrtc_server import MAX_PRE_PROOF_FETCHES
+    assert 0 < MAX_PRE_PROOF_FETCHES <= 10
+
+
+def test_node_admin_ui_requires_token():
+    """
+    11.5.3: "localhost only" is not authentication. Any local process — or a
+    rebound browser page — could re-initialise a group's GEK and read the audit log.
+    """
+    from fastapi.testclient import TestClient
+    from meshbay_node.ui.app import create_ui_app
+
+    app = create_ui_app({"status": "running", "groups_ctx": {},
+                         "indexes": {}, "ui_token": "secret-token"})
+    client = TestClient(app)
+
+    assert client.get("/api/status").status_code == 403
+    assert client.get("/api/status?t=wrong").status_code == 403
+    assert client.get("/api/config?t=wrong").status_code == 403
+    assert client.get("/api/status?t=secret-token").status_code == 200
+    assert client.get(
+        "/api/status", headers={"X-MeshBay-Token": "secret-token"}
+    ).status_code == 200
+
+
 def test_admin_ui_escapes_filenames(tmp_path):
     """
     H2: filenames are chosen by any group member and were rendered into the