fix: swarm privacy, revocation persistence, keystore KDF, audit integrity

Phase 11.5 hardening batch — H7, H4, M2, M6, M7, L1, L3, L6.

H7 — private content hashes leaked to the hub. The daemon registered blake3
hashes for every group it hosted, private ones included, giving the hub a
content fingerprint of every private file and letting anyone confirm whether a
known file exists in the network. The leak was dormant only because the routes
were declared on the groups router with a full path and mounted at
/v1/groups/v1/swarm/* — the node's calls 404'd into a swallowed exception.
Fixing the path alone would have activated the leak, so both land together:
registration is gated on group visibility, the routes moved to a real
/v1/swarm router, and the lookup now requires authentication.

H4 — revocation was advisory. Group revocations were signed and broadcast by
the hub and then dropped by the node, whose handler understood only "user" and
"jti", so "suspend a group" enforced nothing. The denylist was also in-memory
only, so a restart silently un-revoked everyone. Now persisted to
data_dir/denylist.json, group targets honoured on both transports, and live
sessions for a revoked group are closed.

M2 — the node keystore, which protects the node's Ed25519 and X25519 private
keys, was still deriving at 64 MB long after the hub's password verifier moved
to 256 MB; the docs recorded the bump as done, true for the hub only. Raising
the constant alone would have made every existing keystore permanently
undecryptable, so envelopes now record the parameters they were written with
and pre-M2 files continue to open under the legacy profile.

M6 — registration inserted its audit row with a NULL user_id and then ran
UPDATE ip_logs SET user_id=<new> WHERE user_id IS NULL, claiming every
unattributed row in the table: failed logins for other usernames, concurrent
registrations. In logs retained a year for legal requests, that attributed
other people's connections to the wrong account.

M7 — X-Forwarded-For was trusted unconditionally at four call sites, so anyone
could forge the IP written to the compliance log and evade per-IP rate limits.
New netutil.client_ip honours the header only from a trusted proxy and takes
the rightmost hop (the one our proxy appended); no direct header reads remain.

L1 dead GEK_REQUEST/GEK_RESPONSE constants removed; L3 peer errors no longer
echo exception text (paths, internal state); L6 email sanity-checked instead of
accepting any string — deliberately not RFC 5322, to avoid a new dependency.

test_daemon_index_change_pushes_to_peers asserted that a PRIVATE group's hashes
are registered with the hub. Split: private asserts not-called (index push to
members still asserted), and a new test proves public groups still register.
That is the fourth pre-existing test found asserting a vulnerability as
intended behaviour, after gek auto-activation, the transport-wide chat_store
and the blind admin challenge.

Tests: 116 node, 132 hub+common. Regression suite now 43.

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

5 files changed, 98 insertions, 40 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..7738875 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")
 
@@ -70,7 +71,7 @@ async def node_auth(
     access_token = issue_access_token(
         user.id, user.pk_node_ed25519, 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 {
@@ -101,7 +102,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 +129,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/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 53238de..5a0f9dc 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
@@ -62,6 +63,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
@@ -106,21 +125,24 @@ async def register(
         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 +157,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")
@@ -358,8 +380,3 @@ async def get_user_pubkeys(
     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"