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, 101 insertions, 14 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index c9c59c9..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
@@ -42,7 +42,7 @@ import jwt
 from meshbay_hub.auth import hub_public_key_pem, decode_access_token
 from meshbay_hub.api.deps import get_current_user, require_admin
 from meshbay_hub.db.engine import get_db
-from meshbay_hub.db.models import Group, IPLog, User
+from meshbay_hub.db.models import Group, GroupMember, IPLog, Node, User
 
 log = logging.getLogger(__name__)
 
@@ -128,38 +128,109 @@ async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: s
         log.warning("Chat notify failed: %s", e)
 
 
+async def _reject(ws: WebSocket, detail: str, code: int) -> None:
+    await ws.send_text(json.dumps({"type": "error", "detail": detail}))
+    await ws.close(code=code)
+
+
+async def _authorize_node_ws(token: str, claimed_id: str, claimed_groups) -> tuple:
+    """
+    Resolve a node WS registration against the database.
+
+    Returns (node_id, group_ids) on success, or (None, error_detail) on refusal.
+    Uses a short-lived session on purpose: a node WebSocket lives for hours, and a
+    request-scoped dependency would pin a PostgreSQL connection for its whole
+    lifetime, exhausting the pool once a handful of nodes connect.
+    """
+    from meshbay_hub.db.engine import get_session_factory
+
+    try:
+        decoded = decode_access_token(token)
+    except Exception as e:
+        return None, str(e)
+
+    if decoded.get("scope") != "node":
+        return None, "Node-scoped token required"
+
+    user_id = decoded.get("sub", "")
+    if not claimed_id:
+        return None, "node_id required"
+
+    async with get_session_factory()() as db:
+        node = await db.get(Node, claimed_id)
+        if node is None or node.user_id != user_id:
+            log.warning("Rejected WS registration for node %s by user %s",
+                        claimed_id[:8], (user_id or "?")[:8])
+            return None, "node_id does not belong to this account"
+
+        user = await db.get(User, user_id)
+        if user is None or user.status != "active":
+            return None, "Account not active"
+
+        # Groups come from the database. The node may narrow the set to what it
+        # actually hosts, but it cannot widen it to groups it is not a member of —
+        # otherwise it could advertise itself as a source for any group on the hub.
+        result = await db.execute(
+            select(GroupMember.group_id).where(GroupMember.user_id == user_id))
+        authorized = {gid for (gid,) in result.all()}
+
+    claimed = set(claimed_groups or authorized)
+    return claimed_id, sorted(authorized & claimed)
+
+
 @router.websocket("/v1/nodes/ws")
 async def node_websocket(ws: WebSocket):
     """
     Persistent WebSocket connection for nodes.
-    Nodes authenticate with a JWT bearer in the first message.
-    Hub sends revocation tokens as JSON messages.
+
+    Finding C2: this used to take `node_id` and `group_ids` straight from the
+    client's first message, with no check that the authenticated user owned that
+    node. Any registered user could connect with an ordinary browser token, claim a
+    victim node's id, and overwrite its entry in `_connected_nodes`. Every WebRTC
+    offer for that node was then relayed to the attacker, who answered with their
+    own SDP — a full node impersonation, and the DTLS channel binding does not help
+    because the attacker is the endpoint rather than a relay. The attacker received
+    the victim's encrypted keypair bundle, their chat, and their uploads.
+
+    Identity now comes from the token and the database, never from the message.
     """
     await ws.accept()
     node_id: str | None = None
 
     try:
-        # Auth: expect {"type": "auth", "token": "<jwt>"}
+        # Auth: expect {"type": "auth", "token": "<jwt>", "node_id": "..."}
         raw = await ws.receive_text()
         msg = json.loads(raw)
         if msg.get("type") != "auth" or "token" not in msg:
-            await ws.send_text(json.dumps({"type": "error", "detail": "Send auth first"}))
-            await ws.close(code=4001)
+            await _reject(ws, "Send auth first", 4001)
             return
 
         try:
             decoded = decode_access_token(msg["token"])
         except Exception as e:
-            await ws.send_text(json.dumps({"type": "error", "detail": str(e)}))
-            await ws.close(code=4001)
+            await _reject(ws, str(e), 4001)
             return
 
-        node_id = msg.get("node_id") or decoded.get("sub", "unknown")
+        claimed_id = msg.get("node_id") or ""
+
+        # Refuse to displace a live registration rather than silently overwriting it.
+        if claimed_id and claimed_id in _connected_nodes:
+            await _reject(ws, "Node already connected", 4009)
+            return
+
+        resolved_id, result = await _authorize_node_ws(
+            msg["token"], claimed_id, msg.get("group_ids"))
+        if resolved_id is None:
+            await _reject(ws, result, 4003)
+            return
+        group_ids = result
+
+        user_id = decoded.get("sub", "")
+        node_id = resolved_id
         _connected_nodes[node_id] = ws
-        group_ids = msg.get("group_ids", [])
-        if group_ids:
-            _node_groups[node_id] = group_ids
-        log.info("Node WS connected: %s (groups=%d)", node_id[:8], len(group_ids))
+        _node_groups[node_id] = group_ids
+        log.info("Node WS connected: %s (user=%s, groups=%d)",
+                 node_id[:8], user_id[:8], len(group_ids))
         await ws.send_text(json.dumps({"type": "auth_ok", "node_id": node_id}))
 
         # Message loop — handle ping, punch_ready, etc.
@@ -203,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")