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.

8 files changed, 313 insertions, 103 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 88af764..000f3f7 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -6,6 +6,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.api.deps import get_current_user, require_user_scope
+from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import (
     FederatedGroup, Group, GroupMember,
@@ -14,6 +15,10 @@ from meshbay_hub.db.models import (
 
 router = APIRouter(prefix="/v1/groups", tags=["groups"])
 
+# Swarm endpoints live at /v1/swarm/*. They were previously declared on the groups
+# router with a full path, which mounted them at /v1/groups/v1/swarm/* (H7).
+swarm_router = APIRouter(prefix="/v1/swarm", tags=["swarm"])
+
 
 @router.get("/mine")
 async def my_groups(
@@ -117,13 +122,21 @@ class SwarmRegisterRequest(BaseModel):
     endpoint:     str    # "ip:port"
 
 
-@router.post("/v1/swarm/register", status_code=201)
+@swarm_router.post("/register", status_code=201)
 async def swarm_register(
     body: SwarmRegisterRequest,
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
-    """Node registers itself as a source for a content hash (public swarm)."""
+    """
+    Node registers itself as a source for a PUBLIC content hash.
+
+    Finding H7: the node registered hashes for every group it hosted, private ones
+    included, and this route was mounted at /v1/groups/v1/swarm/register — so the
+    node's calls 404'd and the leak was masked by a routing bug rather than
+    prevented. Nodes now filter by group visibility before calling, and the path is
+    correct, so the filter has to be right.
+    """
     from meshbay_hub.csam import check_content_hash
     if check_content_hash(body.content_hash):
         raise HTTPException(status_code=451, detail="Content blocked")
@@ -144,12 +157,18 @@ async def swarm_register(
     return {"status": "registered", "hash": body.content_hash}
 
 
-@router.get("/v1/swarm/{content_hash}")
+@swarm_router.get("/{content_hash}")
 async def swarm_sources(
     content_hash: str,
+    current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
-    """Return list of nodes that can serve a content hash."""
+    """
+    Return nodes that can serve a content hash.
+
+    Authenticated (H7): an open endpoint lets anyone probe whether a given file
+    exists anywhere in the network and which node holds it.
+    """
     from datetime import datetime, timezone, timedelta
     cutoff = datetime.now(timezone.utc) - timedelta(minutes=30)
     result = await db.execute(
@@ -214,7 +233,7 @@ async def join_group(
 
     db.add(GroupMember(group_id=group_id, user_id=current_user.id))
     db.add(IPLog(user_id=current_user.id, event="group_join",
-                 ip_address=_ip(request), detail=group.name))
+                 ip_address=client_ip(request), detail=group.name))
     await db.commit()
     return {"status": "joined", "group_id": group_id, "name": group.name}
 
@@ -246,7 +265,7 @@ async def create_group(
 
     db.add(GroupMember(group_id=group.id, user_id=current_user.id))
     db.add(IPLog(user_id=current_user.id, event="group_create",
-                 ip_address=_ip(request), detail=body.name))
+                 ip_address=client_ip(request), detail=body.name))
     await db.commit()
     await db.refresh(group)
     return {"group_id": group.id, "name": group.name}
@@ -304,13 +323,9 @@ async def delete_group(
     from sqlalchemy import delete as sa_delete
     await db.execute(sa_delete(GroupMember).where(GroupMember.group_id == group_id))
     db.add(IPLog(user_id=current_user.id, event="group_delete",
-                 ip_address=_ip(request), detail=group.name))
+                 ip_address=client_ip(request), detail=group.name))
     await db.delete(group)
     await db.commit()
     return {"status": "deleted", "group_id": group_id}
 
 
-def _ip(request: Request) -> str:
-    fwd = request.headers.get("X-Forwarded-For")
-    return fwd.split(",")[0].strip() if fwd else (
-        request.client.host if request.client else "unknown")
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/moderation.py b/packages/meshbay-hub/src/meshbay_hub/api/moderation.py
index 6bb007b..853f255 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/moderation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/moderation.py
@@ -28,6 +28,7 @@ from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.api.deps import get_current_user, require_admin
+from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import ContentBlocklist, ContentReport, User
 
@@ -64,7 +65,7 @@ async def report_content(
     if len(body.content_hash) != 64 or not all(c in "0123456789abcdef" for c in body.content_hash):
         raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)")
 
-    ip = _ip(request)
+    ip = client_ip(request)
 
     # Count existing reports for this hash
     count_result = await db.execute(
@@ -193,8 +194,3 @@ async def admin_remove_blocklist(
     await db.commit()
     return {"status": "unblocked", "hash": content_hash}
 
-
-def _ip(request: Request) -> str:
-    fwd = request.headers.get("X-Forwarded-For")
-    return fwd.split(",")[0].strip() if fwd else (
-        request.client.host if request.client else "unknown")
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/netutil.py b/packages/meshbay-hub/src/meshbay_hub/api/netutil.py
new file mode 100644
index 0000000..aa8344a
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/api/netutil.py
@@ -0,0 +1,34 @@
+"""
+Client address resolution for the audit log and rate limiting.
+
+Finding M7: every call site did
+
+    fwd = request.headers.get("X-Forwarded-For")
+    return fwd.split(",")[0].strip() if fwd else request.client.host
+
+which trusts a header the client controls. Anyone could forge the IP written into
+the compliance log — the log that exists specifically to answer legal requests —
+and sidestep per-IP rate limiting at the same time.
+
+X-Forwarded-For is only consulted when the immediate peer is a trusted proxy, and
+then the *rightmost* entry is used: that is the one our own proxy appended, whereas
+the leftmost is whatever the client sent.
+"""
+
+from fastapi import Request
+
+# Caddy terminates TLS on the same host and proxies to 127.0.0.1:8000.
+TRUSTED_PROXIES = frozenset({"127.0.0.1", "::1", "localhost"})
+
+
+def client_ip(request: Request) -> str:
+    peer = request.client.host if request.client else ""
+
+    if peer in TRUSTED_PROXIES:
+        forwarded = request.headers.get("X-Forwarded-For")
+        if forwarded:
+            hops = [h.strip() for h in forwarded.split(",") if h.strip()]
+            if hops:
+                return hops[-1]
+
+    return peer or "unknown"
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/nodes.py b/packages/meshbay-hub/src/meshbay_hub/api/nodes.py
index 321e43c..0770148 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/nodes.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/nodes.py
@@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from meshbay_hub.auth import issue_access_token
 from meshbay_hub.api.deps import get_current_user
 from meshbay_hub.api.middleware import limiter
+from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import GroupMember, IPLog, Node, User
 
@@ -59,7 +60,7 @@ async def node_auth(
         sig = base64.b64decode(body.signature)
         pk.verify(sig, message)
     except (InvalidSignature, Exception):
-        db.add(IPLog(event="node_auth_fail", ip_address=_ip(request), detail=body.username))
+        db.add(IPLog(event="node_auth_fail", ip_address=client_ip(request), detail=body.username))
         await db.commit()
         raise HTTPException(status_code=401, detail="Invalid signature")
 
@@ -68,9 +69,9 @@ async def node_auth(
     group_ids = [gid for (gid,) in memberships.all()]
 
     access_token = issue_access_token(
-        user.id, user.pk_node_ed25519, ttl=3600, groups=group_ids, scope="node")
+        user.id, ttl=3600, groups=group_ids, scope="node")
 
-    db.add(IPLog(user_id=user.id, event="node_auth", ip_address=_ip(request)))
+    db.add(IPLog(user_id=user.id, event="node_auth", ip_address=client_ip(request)))
     await db.commit()
 
     return {
@@ -83,6 +84,8 @@ async def node_auth(
 class NodeAnnounceRequest(BaseModel):
     pk_node:       str
     endpoint_hint: str | None = None
+    timestamp:     int | None = None   # unix seconds
+    signature:     str | None = None   # base64 Ed25519 over the announce message
 
 
 @router.post("/announce", status_code=201)
@@ -92,6 +95,48 @@ async def announce_node(
     current_user: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ):
+    """
+    Register a node record.
+
+    Finding M8: this accepted any pk_node with no proof the announcer held the
+    matching private key, so a user could announce a record carrying someone
+    else's node key — useful for muddying node identity, and records accumulated
+    without limit. The announcer must now sign a domain-separated message binding
+    the key to their account, the same pattern already used by /v1/nodes/auth.
+    """
+    if body.timestamp is None or not body.signature:
+        raise HTTPException(
+            status_code=400,
+            detail="announce requires timestamp and signature (proof of possession)")
+
+    now = int(time.time())
+    if abs(now - body.timestamp) > NODE_AUTH_TIMESTAMP_WINDOW:
+        raise HTTPException(status_code=401, detail="Timestamp too old or too far ahead")
+
+    message = (f"meshbay:node_announce:{current_user.id}:"
+               f"{body.pk_node}:{body.timestamp}").encode()
+    try:
+        pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(body.pk_node))
+        pk.verify(base64.b64decode(body.signature), message)
+    except Exception:
+        db.add(IPLog(user_id=current_user.id, event="node_announce_fail",
+                     ip_address=client_ip(request), detail=body.pk_node[:16]))
+        await db.commit()
+        raise HTTPException(status_code=401, detail="Invalid node key proof of possession")
+
+    # One active record per key per account — announcing again updates in place
+    # instead of accumulating rows.
+    existing = await db.execute(
+        select(Node).where(Node.user_id == current_user.id,
+                           Node.pk_node == body.pk_node))
+    node = existing.scalar_one_or_none()
+    if node is not None:
+        node.endpoint_hint = body.endpoint_hint
+        db.add(IPLog(user_id=current_user.id, event="node_announce",
+                     ip_address=client_ip(request), detail=body.endpoint_hint))
+        await db.commit()
+        return {"node_id": node.id}
+
     node = Node(
         user_id=current_user.id,
         pk_node=body.pk_node,
@@ -101,7 +146,7 @@ async def announce_node(
     db.add(IPLog(
         user_id=current_user.id,
         event="node_announce",
-        ip_address=_ip(request),
+        ip_address=client_ip(request),
         detail=body.endpoint_hint,
     ))
     await db.commit()
@@ -128,8 +173,3 @@ async def get_node(
     }
 
 
-def _ip(request: Request) -> str:
-    fwd = request.headers.get("X-Forwarded-For")
-    if fwd:
-        return fwd.split(",")[0].strip()
-    return request.client.host if request.client else "unknown"
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")
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/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 53238de..af9141c 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -5,7 +5,7 @@ import uuid
 from datetime import datetime, timezone, timedelta
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
-from pydantic import BaseModel, EmailStr, field_validator
+from pydantic import BaseModel, field_validator
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -22,6 +22,7 @@ from meshbay_hub.auth import (
     verify_password,
 )
 from meshbay_hub.api.middleware import limiter
+from meshbay_hub.api.netutil import client_ip
 from meshbay_hub.config import HubConfig
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import GroupMember, IPLog, RefreshToken, User
@@ -49,8 +50,6 @@ class RegisterRequest(BaseModel):
     email:           str
     password:        str | None = None    # deprecated — legacy native clients
     auth_key:        str | None = None    # PBKDF2-derived, new clients
-    pk_user_ed25519: str        # base64 raw 32B
-    pk_user_x25519:  str        # base64 raw 32B
 
     @field_validator("username")
     @classmethod
@@ -62,6 +61,24 @@ class RegisterRequest(BaseModel):
             raise ValueError("username: only letters, digits, -, _, .")
         return v
 
+    @field_validator("email")
+    @classmethod
+    def email_valid(cls, v: str) -> str:
+        """
+        Sanity-check the address (L6): the field was plain `str`, so any junk was
+        accepted and stored encrypted forever. Deliberately not RFC 5322 — full
+        validation would pull in the email-validator dependency for little gain,
+        and the address is only ever used for recovery and legal contact.
+        """
+        v = v.strip()
+        local, sep, domain = v.partition("@")
+        if (not sep or not local or not domain
+                or "." not in domain
+                or len(v) > 254
+                or any(c.isspace() or ord(c) < 32 for c in v)):
+            raise ValueError("invalid email address")
+        return v
+
 
 class LoginRequest(BaseModel):
     username: str
@@ -101,26 +118,27 @@ async def register(
         pw_hash=pw_hash,
         pw_salt=pw_salt,
         pw_version=pw_ver,
-        pk_ed25519=body.pk_user_ed25519,
-        pk_x25519=body.pk_user_x25519,
         hub_id=hub_id,
     )
     db.add(user)
+    # flush assigns user.id so the log row can be attributed directly.
+    #
+    # Finding M6: this used to insert the row with a NULL user_id and then run
+    #   UPDATE ip_logs SET user_id = <new user> WHERE user_id IS NULL
+    # which claimed *every* unattributed row in the table — failed logins for other
+    # usernames, other registrations racing this one — and stamped them with the
+    # account just created. For logs retained a year to answer legal requests, that
+    # attributed other people's connections to the wrong person.
+    await db.flush()
     db.add(IPLog(
+        user_id=user.id,
         event="account_create",
-        ip_address=_client_ip(request),
+        ip_address=client_ip(request),
         detail=body.username,
     ))
     await db.commit()
     await db.refresh(user)
 
-    # Set user_id in IPLog after commit
-    await db.execute(
-        IPLog.__table__.update()
-        .where(IPLog.user_id == None)  # noqa: E711
-        .values(user_id=user.id))
-    await db.commit()
-
     return {"user_id": user.id}
 
 
@@ -135,7 +153,7 @@ async def login(
         select(User).where(User.username == body.username))
     user = result.scalar_one_or_none()
 
-    ip = _client_ip(request)
+    ip = client_ip(request)
 
     if not body.auth_key and not body.password:
         raise HTTPException(status_code=401, detail="No credentials provided")
@@ -189,8 +207,7 @@ async def login(
     memberships = await db.execute(
         select(GroupMember.group_id).where(GroupMember.user_id == user.id))
     group_ids = [gid for (gid,) in memberships.all()]
-    access_token = issue_access_token(
-        user.id, user.pk_ed25519, ttl=_ttl(), groups=group_ids)
+    access_token = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
     raw_rt, rt_hash = generate_refresh_token()
     family_id = str(uuid.uuid4())
 
@@ -255,8 +272,7 @@ async def token_refresh(
     memberships = await db.execute(
         select(GroupMember.group_id).where(GroupMember.user_id == user.id))
     group_ids = [gid for (gid,) in memberships.all()]
-    new_access = issue_access_token(
-        user.id, user.pk_ed25519, ttl=_ttl(), groups=group_ids)
+    new_access = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
     await db.commit()
 
     return {
@@ -302,39 +318,10 @@ async def register_node_key(
     return {"status": "stored", "pk_node_ed25519": body.pk_node_ed25519}
 
 
-class RotateKeysRequest(BaseModel):
-    pk_user_ed25519: str  # base64 raw 32B
-    pk_user_x25519: str   # base64 raw 32B
-
-
-@router.put("/me/keys")
-async def rotate_browser_keys(
-    body: RotateKeysRequest,
-    current_user: User = Depends(require_user_scope),
-    db: AsyncSession = Depends(get_db),
-):
-    for field, label in [
-        (body.pk_user_ed25519, "Ed25519"),
-        (body.pk_user_x25519, "X25519"),
-    ]:
-        try:
-            raw = base64.b64decode(field)
-            if len(raw) != 32:
-                raise ValueError
-        except Exception:
-            raise HTTPException(
-                status_code=400,
-                detail=f"Invalid {label} public key (need 32 bytes base64)",
-            )
-
-    current_user.pk_ed25519 = body.pk_user_ed25519
-    current_user.pk_x25519 = body.pk_user_x25519
-    await db.commit()
-    return {
-        "status": "updated",
-        "pk_ed25519": body.pk_user_ed25519,
-        "pk_x25519": body.pk_user_x25519,
-    }
+# Key rotation used to live here (`PUT /me/keys`). Identity keys are per node
+# now, so rotating means `meshbay-node member unpin <user>` and pairing again with
+# a fresh code — an operator decision on the machine that pinned it, not a hub
+# call that silently changes what every node believes about someone.
 
 
 @router.get("/{username}/pubkeys")
@@ -347,19 +334,16 @@ async def get_user_pubkeys(
     target = result.scalar_one_or_none()
     if not target:
         raise HTTPException(status_code=404, detail="User not found")
+    # Account lookup, not a key directory. `user_id` is how a username is resolved
+    # for an invitation, and `pk_node_ed25519` is a node's own linking key. The
+    # user identity keys this used to return were H3: whoever asked wrapped the
+    # group key for whatever came back.
     resp = {
-        "user_id":    target.id,
-        "username":   target.username,
-        "pk_ed25519": target.pk_ed25519,
-        "pk_x25519":  target.pk_x25519,
+        "user_id":  target.id,
+        "username": target.username,
     }
     if target.pk_node_ed25519:
         resp["pk_node_ed25519"] = target.pk_node_ed25519
     return resp
 
 
-def _client_ip(request: Request) -> str:
-    forwarded = request.headers.get("X-Forwarded-For")
-    if forwarded:
-        return forwarded.split(",")[0].strip()
-    return request.client.host if request.client else "unknown"
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/webapp.py b/packages/meshbay-hub/src/meshbay_hub/api/webapp.py
index 5ad7329..f804ec5 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/webapp.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/webapp.py
@@ -47,6 +47,10 @@ _HTML = """\
 </head>
 <body>
   <div id="app"></div>
+  <!-- Argon2id (WebAssembly, inlined) — WebCrypto has no memory-hard KDF, and the
+       keypair bundle needs one: it is protected by the passphrase alone and sits
+       on every node its owner joins (C4). Vendored, see static/vendor/PROVENANCE.md -->
+  <script src="/vendor/argon2.min.js"></script>
   <script src="/keyderive.js"></script>
   <script src="/crypto.js"></script>
   <script src="/transport.js"></script>