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>

2 files changed, 70 insertions, 4 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: