feat(hub): Phase 8 — Hub v2 security hardening + production readiness

8.1  Config-based admin authz (require_admin on all admin endpoints)
8.2  Email encrypted at rest (AES-256-GCM, HKDF from hub Ed25519 key)
8.3  Refresh token rotation with family-based reuse detection
8.4  Federation persistence (HubPeer model replaces in-memory dict)
8.5  Federation token verification now async (DB-backed)
8.6  CSAM hash check wired into swarm registration flow
8.7  Rate limiting on auth endpoints (5/10/20 per minute)
8.8  Healthcheck endpoint (GET /v1/health, no auth)
8.9  IP log cleanup background task (365-day retention)
8.10 Argon2id params bumped to 256 MB (pw_version, rehash on login)

Deployed to meshbay.org — schema migrated, existing emails encrypted.
117 tests pass (29 hub, 88 common+node).

Resolves security review items S1, S2, S5.

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

8 files changed, 150 insertions, 69 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/deps.py b/packages/meshbay-hub/src/meshbay_hub/api/deps.py
index cb637f3..7a580ad 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/deps.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/deps.py
@@ -2,8 +2,6 @@
 FastAPI shared dependencies — injected via Depends().
 """
 
-from collections.abc import AsyncGenerator
-
 from fastapi import Depends, Header, HTTPException, status
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select
@@ -12,6 +10,13 @@ from meshbay_hub.auth import decode_access_token
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import User
 
+_admin_usernames: set[str] = set()
+
+
+def set_admin_usernames(usernames: list[str]) -> None:
+    global _admin_usernames
+    _admin_usernames = set(usernames)
+
 
 async def get_current_user(
     authorization: str = Header(...),
@@ -45,3 +50,12 @@ async def get_current_user(
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
                             detail=f"Account {user.status}")
     return user
+
+
+async def require_admin(
+    current_user: User = Depends(get_current_user),
+) -> User:
+    if current_user.username not in _admin_usernames:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,
+                            detail="Admin access required")
+    return current_user
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/federation.py b/packages/meshbay-hub/src/meshbay_hub/api/federation.py
index 88bb645..b6b4acf 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/federation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/federation.py
@@ -32,21 +32,15 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_common import MHP_VERSION
 from meshbay_hub import __version__
-from meshbay_hub.api.deps import get_current_user
+from meshbay_hub.api.deps import require_admin
 from meshbay_hub.auth import _hub_id, _hub_sk_pem, hub_public_key_pem
 from meshbay_hub.db.engine import get_db
-from meshbay_hub.db.models import Group, User
-from sqlalchemy.ext.asyncio import AsyncSession
+from meshbay_hub.db.models import FederatedGroup, Group, HubPeer, User
 
 log = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/mhp", tags=["federation"])
 
-# ── In-memory peer registry ───────────────────────────────────────────────────
-# Production: move to DB table (peers: hub_id, hub_url, pk_hub_pem, trusted_since)
-
-_peers: dict[str, dict] = {}   # hub_id → {url, pk_pem, trusted_since}
-
 
 def _issue_mhp_token(target_hub_id: str) -> str:
     """Issue a short-lived JWT for authenticating to a peer hub."""
@@ -57,18 +51,19 @@ def _issue_mhp_token(target_hub_id: str) -> str:
         "aud": target_hub_id,
         "jti": str(uuid.uuid4()),
         "iat": now,
-        "exp": now + 300,   # 5 minute window
+        "exp": now + 300,
         "mhp": MHP_VERSION,
     }, _hub_sk_pem, algorithm="EdDSA")
 
 
-def _verify_mhp_token(token: str, expected_aud: str | None = None) -> dict:
-    """Verify a JWT from a peer hub."""
-    # First decode without verification to get iss (sender hub_id)
+async def _verify_mhp_token(
+    token: str, db: AsyncSession, expected_aud: str | None = None,
+) -> dict:
+    """Verify a JWT from a peer hub using DB-stored public key."""
     unverified = jwt.decode(token, options={"verify_signature": False})
     sender_id = unverified.get("iss")
 
-    peer = _peers.get(sender_id)
+    peer = await db.get(HubPeer, sender_id)
     if not peer:
         raise PermissionError(f"Unknown hub: {sender_id!r}. Register as peer first.")
 
@@ -77,7 +72,7 @@ def _verify_mhp_token(token: str, expected_aud: str | None = None) -> dict:
         options["audience"] = expected_aud
 
     decoded = jwt.decode(
-        token, peer["pk_pem"].encode(),
+        token, peer.pk_hub_pem.encode(),
         algorithms=["EdDSA"],
         options=options,
     )
@@ -104,12 +99,8 @@ async def export_directory(
     db: AsyncSession = Depends(get_db),
     authorization: str = Header(...),
 ):
-    """
-    Export our public Mesh Directory to a peer hub.
-    Auth: Bearer JWT signed by peer hub's key.
-    """
     try:
-        _verify_mhp_token(authorization.removeprefix("Bearer "))
+        await _verify_mhp_token(authorization.removeprefix("Bearer "), db)
     except Exception as e:
         raise HTTPException(status_code=401, detail=str(e))
 
@@ -144,16 +135,11 @@ async def receive_directory(
     authorization: str = Header(...),
     db: AsyncSession = Depends(get_db),
 ):
-    """
-    Receive a Mesh Directory update from a peer hub.
-    Persists groups to federated_groups table for cross-hub search.
-    """
     try:
-        _verify_mhp_token(authorization.removeprefix("Bearer "))
+        await _verify_mhp_token(authorization.removeprefix("Bearer "), db)
     except Exception as e:
         raise HTTPException(status_code=401, detail=str(e))
 
-    from meshbay_hub.db.models import FederatedGroup
     from datetime import datetime, timezone
     now = datetime.now(timezone.utc)
     count = 0
@@ -179,24 +165,19 @@ async def receive_directory(
 # ── Revocation propagation ────────────────────────────────────────────────────
 
 class RevocationPayload(BaseModel):
-    token: str   # signed revocation JWT from originating hub
-
+    token: str
 
 @router.post("/revoke", status_code=202)
 async def receive_revocation(
     body: RevocationPayload,
     authorization: str = Header(...),
+    db: AsyncSession = Depends(get_db),
 ):
-    """
-    Receive a revocation from a peer hub. Verify and propagate to our nodes.
-    """
     try:
-        _verify_mhp_token(authorization.removeprefix("Bearer "))
+        await _verify_mhp_token(authorization.removeprefix("Bearer "), db)
     except Exception as e:
         raise HTTPException(status_code=401, detail=str(e))
 
-    # The revocation token is signed by the ORIGINATING hub's key (not the relaying hub)
-    # For now: re-broadcast to our connected nodes
     from meshbay_hub.api.revocation import broadcast_revocation
     sent = await broadcast_revocation(body.token)
     log.info("Propagated revocation to %d local nodes", sent)
@@ -208,30 +189,45 @@ async def receive_revocation(
 class PeerRegisterRequest(BaseModel):
     hub_id:    str
     hub_url:   str
-    pk_hub_pem: str   # peer hub's Ed25519 public key PEM
-
+    pk_hub_pem: str
 
 @router.post("/peers", status_code=201)
 async def register_peer(
     body: PeerRegisterRequest,
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
+    db: AsyncSession = Depends(get_db),
 ):
-    """Admin: register a trusted peer hub. Manual step — no auto-discovery."""
-    _peers[body.hub_id] = {
-        "url":          body.hub_url,
-        "pk_pem":       body.pk_hub_pem,
-        "trusted_since": int(time.time()),
-    }
+    """Admin: register a trusted peer hub."""
+    existing = await db.get(HubPeer, body.hub_id)
+    if existing:
+        existing.hub_url = body.hub_url
+        existing.pk_hub_pem = body.pk_hub_pem
+    else:
+        db.add(HubPeer(
+            hub_id=body.hub_id,
+            hub_url=body.hub_url,
+            pk_hub_pem=body.pk_hub_pem,
+        ))
+    await db.commit()
     log.info("Peer registered: %s (%s)", body.hub_id, body.hub_url)
     return {"status": "registered", "hub_id": body.hub_id}
 
 
 @router.get("/peers")
-async def list_peers(current_user: User = Depends(get_current_user)):
+async def list_peers(
+    current_user: User = Depends(require_admin),
+    db: AsyncSession = Depends(get_db),
+):
     """Admin: list registered peer hubs."""
+    result = await db.execute(select(HubPeer))
+    peers = result.scalars().all()
     return {
         "peers": [
-            {"hub_id": hid, "url": p["url"], "trusted_since": p["trusted_since"]}
-            for hid, p in _peers.items()
+            {
+                "hub_id": p.hub_id,
+                "url": p.hub_url,
+                "trusted_since": p.trusted_since.isoformat(),
+            }
+            for p in peers
         ]
     }
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 2ae2c93..5bccf71 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -67,10 +67,11 @@ async def swarm_register(
     db: AsyncSession = Depends(get_db),
 ):
     """Node registers itself as a source for a content hash (public swarm)."""
+    from meshbay_hub.csam import check_content_hash
+    if check_content_hash(body.content_hash):
+        raise HTTPException(status_code=451, detail="Content blocked")
+
     from datetime import datetime, timezone
-    node_result = await db.execute(
-        select(User).where(User.id == current_user.id))
-    # Use current_user.id as node_id for simplicity
     existing = await db.get(SwarmSource, (body.content_hash, current_user.id))
     now = datetime.now(timezone.utc)
     if existing:
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/health.py b/packages/meshbay-hub/src/meshbay_hub/api/health.py
new file mode 100644
index 0000000..516856b
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/api/health.py
@@ -0,0 +1,21 @@
+"""Healthcheck endpoint — no auth required (monitoring probes)."""
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from meshbay_hub import __version__
+from meshbay_hub.api.revocation import get_connected_node_count
+from meshbay_hub.db.engine import get_db
+
+router = APIRouter(tags=["health"])
+
+
+@router.get("/v1/health")
+async def health(db: AsyncSession = Depends(get_db)):
+    await db.execute(text("SELECT 1"))
+    return {
+        "status": "ok",
+        "version": __version__,
+        "connected_nodes": get_connected_node_count(),
+    }
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/moderation.py b/packages/meshbay-hub/src/meshbay_hub/api/moderation.py
index a8bf84f..6bb007b 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/moderation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/moderation.py
@@ -27,7 +27,7 @@ from pydantic import BaseModel
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from meshbay_hub.api.deps import get_current_user
+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 ContentBlocklist, ContentReport, User
 
@@ -138,7 +138,7 @@ async def get_blocklist(
 
 @router.get("/v1/admin/blocklist")
 async def admin_list_blocklist(
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
     db: AsyncSession = Depends(get_db),
     limit: int = 500,
 ):
@@ -164,7 +164,7 @@ async def admin_list_blocklist(
 @router.post("/v1/admin/blocklist", status_code=201)
 async def admin_add_blocklist(
     body: BlocklistAddRequest,
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
     db: AsyncSession = Depends(get_db),
 ):
     existing = await db.get(ContentBlocklist, body.content_hash)
@@ -183,7 +183,7 @@ async def admin_add_blocklist(
 @router.delete("/v1/admin/blocklist/{content_hash}", status_code=200)
 async def admin_remove_blocklist(
     content_hash: str,
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
     db: AsyncSession = Depends(get_db),
 ):
     entry = await db.get(ContentBlocklist, content_hash)
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/relay.py b/packages/meshbay-hub/src/meshbay_hub/api/relay.py
index d2a4b81..f82495b 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/relay.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/relay.py
@@ -24,7 +24,7 @@ from fastapi import APIRouter, Depends, Header, HTTPException
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from meshbay_hub.api.deps import get_current_user
+from meshbay_hub.api.deps import get_current_user, require_admin
 from meshbay_hub.auth import _hub_id, hub_public_key_pem
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import User
@@ -100,7 +100,7 @@ async def list_relays():
 @router.post("/approve", status_code=201)
 async def admin_approve_relay(
     body: RelayAdminApproveRequest,
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
 ):
     """Admin: pre-approve a relay by registering its public key."""
     _relays[body.relay_id] = {
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index bb88283..bbd1bc2 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -40,7 +40,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 import jwt
 from meshbay_hub.auth import hub_public_key_pem, decode_access_token
-from meshbay_hub.api.deps import get_current_user
+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
 
@@ -193,7 +193,7 @@ class RevokeRequest(BaseModel):
 @router.post("/v1/admin/revoke", status_code=200)
 async def admin_revoke(
     body: RevokeRequest,
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
     db: AsyncSession = Depends(get_db),
 ):
     """
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/users.py b/packages/meshbay-hub/src/meshbay_hub/api/users.py
index 5a2c7be..f91b381 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/users.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/users.py
@@ -1,5 +1,6 @@
 """User endpoints — /v1/users/*"""
 
+import uuid
 from datetime import datetime, timezone, timedelta
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
@@ -8,14 +9,18 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.auth import (
+    current_pw_version,
     decode_access_token,
+    encrypt_email,
     generate_refresh_token,
     hash_password,
     hash_refresh_token,
     hub_public_key_pem,
     issue_access_token,
+    pw_needs_rehash,
     verify_password,
 )
+from meshbay_hub.api.middleware import limiter
 from meshbay_hub.config import HubConfig
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import GroupMember, IPLog, RefreshToken, User
@@ -76,6 +81,7 @@ class RefreshRequest(BaseModel):
 # ── Endpoints ─────────────────────────────────────────────────────────────────
 
 @router.post("/register", status_code=201)
+@limiter.limit("5/minute")
 async def register(
     body: RegisterRequest,
     request: Request,
@@ -90,9 +96,10 @@ async def register(
     hub_id = _cfg.identity.id if _cfg else "meshbay.org"
     user = User(
         username=body.username,
-        email=body.email,
+        email=encrypt_email(body.email),
         pw_hash=pw_hash,
         pw_salt=pw_salt,
+        pw_version=current_pw_version(),
         pk_ed25519=body.pk_user_ed25519,
         pk_x25519=body.pk_user_x25519,
         hub_id=hub_id,
@@ -118,6 +125,7 @@ async def register(
 
 
 @router.post("/login")
+@limiter.limit("10/minute")
 async def login(
     body: LoginRequest,
     request: Request,
@@ -128,7 +136,9 @@ async def login(
     user = result.scalar_one_or_none()
 
     ip = _client_ip(request)
-    if not user or not verify_password(body.password, user.pw_hash, user.pw_salt):
+    if not user or not verify_password(
+        body.password, user.pw_hash, user.pw_salt, version=user.pw_version
+    ):
         db.add(IPLog(event="login_fail", ip_address=ip, detail=body.username))
         await db.commit()
         raise HTTPException(status_code=401, detail="Invalid credentials")
@@ -136,15 +146,25 @@ async def login(
     if user.status != "active":
         raise HTTPException(status_code=403, detail=f"Account {user.status}")
 
+    if pw_needs_rehash(user.pw_version):
+        new_hash, new_salt = hash_password(body.password)
+        user.pw_hash = new_hash
+        user.pw_salt = new_salt
+        user.pw_version = current_pw_version()
+
     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)
     raw_rt, rt_hash = generate_refresh_token()
+    family_id = str(uuid.uuid4())
 
     expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
-    db.add(RefreshToken(user_id=user.id, token_hash=rt_hash, expires_at=expires_at))
+    db.add(RefreshToken(
+        user_id=user.id, token_hash=rt_hash,
+        family_id=family_id, expires_at=expires_at,
+    ))
     db.add(IPLog(user_id=user.id, event="login", ip_address=ip))
     await db.commit()
 
@@ -160,31 +180,60 @@ async def login(
 
 
 @router.post("/token/refresh")
+@limiter.limit("20/minute")
 async def token_refresh(
     body: RefreshRequest,
+    request: Request,
     db: AsyncSession = Depends(get_db),
 ):
     rt_hash = hash_refresh_token(body.refresh_token)
     result = await db.execute(
-        select(RefreshToken).where(
-            RefreshToken.token_hash == rt_hash,
-            RefreshToken.revoked == False,  # noqa: E712
-        ))
+        select(RefreshToken).where(RefreshToken.token_hash == rt_hash))
     rt = result.scalar_one_or_none()
 
-    if not rt or rt.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc):
-        raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
+    if not rt:
+        raise HTTPException(status_code=401, detail="Invalid refresh token")
+
+    if rt.revoked:
+        # Reuse detected — revoke entire token family
+        await db.execute(
+            RefreshToken.__table__.update()
+            .where(RefreshToken.family_id == rt.family_id)
+            .values(revoked=True))
+        await db.commit()
+        raise HTTPException(status_code=401, detail="Token reuse detected — family revoked")
+
+    if rt.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc):
+        raise HTTPException(status_code=401, detail="Expired refresh token")
 
     user = await db.get(User, rt.user_id)
     if not user or user.status != "active":
         raise HTTPException(status_code=401, detail="User not found or suspended")
 
+    # Revoke old token
+    rt.revoked = True
+
+    # Issue new refresh token in the same family
+    new_raw_rt, new_rt_hash = generate_refresh_token()
+    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
+    db.add(RefreshToken(
+        user_id=user.id, token_hash=new_rt_hash,
+        family_id=rt.family_id, expires_at=expires_at,
+    ))
+
     memberships = await db.execute(
         select(GroupMember.group_id).where(GroupMember.user_id == user.id))
     group_ids = [gid for (gid,) in memberships.all()]
-    new_token = issue_access_token(
+    new_access = issue_access_token(
         user.id, user.pk_ed25519, ttl=_ttl(), groups=group_ids)
-    return {"access_token": new_token, "token_type": "bearer", "expires_in": _ttl()}
+    await db.commit()
+
+    return {
+        "access_token": new_access,
+        "refresh_token": new_raw_rt,
+        "token_type": "bearer",
+        "expires_in": _ttl(),
+    }
 
 
 @router.get("/{username}/pubkeys")