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>

15 files changed, 291 insertions, 85 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")
diff --git a/packages/meshbay-hub/src/meshbay_hub/app.py b/packages/meshbay-hub/src/meshbay_hub/app.py
index 1c494fa..c80da28 100644
--- a/packages/meshbay-hub/src/meshbay_hub/app.py
+++ b/packages/meshbay-hub/src/meshbay_hub/app.py
@@ -9,6 +9,7 @@ Usage:
     app = create_app(cfg)
 """
 
+import asyncio
 from contextlib import asynccontextmanager
 from pathlib import Path
 
@@ -22,12 +23,14 @@ from meshbay_hub.config import HubConfig
 from meshbay_hub.db.engine import close_db, init_db
 from meshbay_hub.api.hub        import router as hub_router
 from meshbay_hub.api.users      import router as users_router, set_config as users_set_config
+from meshbay_hub.api.deps       import set_admin_usernames
 from meshbay_hub.api.nodes      import router as nodes_router
 from meshbay_hub.api.groups     import router as groups_router
 from meshbay_hub.api.revocation  import router as revocation_router
 from meshbay_hub.api.moderation  import router as moderation_router
 from meshbay_hub.api.federation  import router as federation_router
 from meshbay_hub.csam            import csam_router
+from meshbay_hub.api.health      import router as health_router
 from meshbay_hub.api.relay       import router as relay_router
 from meshbay_hub.api.webapp     import router as webapp_router
 from meshbay_hub.api.middleware import limiter
@@ -48,9 +51,23 @@ def create_app(cfg: HubConfig | None = None) -> FastAPI:
             generate_hub_keypair(kp)
         load_hub_keypair(kp, cfg.identity.id)
         users_set_config(cfg)
+        set_admin_usernames(cfg.identity.admin_usernames)
+
+        from meshbay_hub.csam import get_csam_checker
+        get_csam_checker().load()
+
+        from meshbay_hub.tasks.cleanup import cleanup_loop
+        from meshbay_hub.db.engine import get_session_factory
+        cleanup_task = asyncio.create_task(cleanup_loop(get_session_factory()))
 
         yield
 
+        cleanup_task.cancel()
+        try:
+            await cleanup_task
+        except asyncio.CancelledError:
+            pass
+
         # Shutdown
         await close_db()
 
@@ -74,6 +91,7 @@ def create_app(cfg: HubConfig | None = None) -> FastAPI:
     app.include_router(moderation_router)
     app.include_router(federation_router)
     app.include_router(csam_router)
+    app.include_router(health_router)
     app.include_router(relay_router)
     app.include_router(webapp_router)
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/auth.py b/packages/meshbay-hub/src/meshbay_hub/auth.py
index 2b2c61e..11ad112 100644
--- a/packages/meshbay-hub/src/meshbay_hub/auth.py
+++ b/packages/meshbay-hub/src/meshbay_hub/auth.py
@@ -18,25 +18,33 @@ import blake3
 import jwt
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+from cryptography.hazmat.primitives.hashes import SHA256
 
-# Argon2id parameters — see CLAUDE.md for production calibration guidance
-_ARGON2_ITERATIONS  = 3
-_ARGON2_MEMORY_COST = 65536   # 64 MB — increase to 262144 in production
-_ARGON2_LANES       = 4
-_ARGON2_KEY_LEN     = 32
+# Argon2id parameters — versioned for gradual migration
+_ARGON2_LANES   = 4
+_ARGON2_KEY_LEN = 32
+
+_ARGON2_VERSIONS = {
+    1: {"iterations": 3, "memory_cost": 65536},     # 64 MB  — initial
+    2: {"iterations": 3, "memory_cost": 262144},    # 256 MB — production target
+}
+_ARGON2_CURRENT_VERSION = 2
 
 # Module-level hub keypair (loaded once at startup)
 _hub_sk_pem: bytes | None = None
 _hub_pk_pem: bytes | None = None
 _hub_id:     str          = "meshbay.org"
+_email_key:  bytes | None = None
 
 
 # ── Hub keypair ───────────────────────────────────────────────────────────────
 
 def load_hub_keypair(private_key_path: Path, hub_id: str) -> None:
     """Load hub Ed25519 keypair from PEM file. Call once at startup."""
-    global _hub_sk_pem, _hub_pk_pem, _hub_id
+    global _hub_sk_pem, _hub_pk_pem, _hub_id, _email_key
     _hub_sk_pem = private_key_path.read_bytes()
     sk = serialization.load_pem_private_key(_hub_sk_pem, password=None)
     _hub_pk_pem = sk.public_key().public_bytes(
@@ -45,6 +53,15 @@ def load_hub_keypair(private_key_path: Path, hub_id: str) -> None:
     )
     _hub_id = hub_id
 
+    sk_raw = sk.private_bytes(
+        serialization.Encoding.Raw,
+        serialization.PrivateFormat.Raw,
+        serialization.NoEncryption(),
+    )
+    _email_key = HKDF(
+        algorithm=SHA256(), length=32, salt=None, info=b"meshbay:email:v1",
+    ).derive(sk_raw)
+
 
 def generate_hub_keypair(private_key_path: Path) -> None:
     """Generate a new hub Ed25519 keypair and save PEM files. Run once."""
@@ -73,32 +90,42 @@ def hub_public_key_pem() -> bytes:
 # ── Password ──────────────────────────────────────────────────────────────────
 
 def hash_password(password: str) -> tuple[bytes, bytes]:
-    """Hash a password with Argon2id. Returns (hash, salt)."""
+    """Hash a password with Argon2id (current version). Returns (hash, salt)."""
     salt = os.urandom(16)
+    params = _ARGON2_VERSIONS[_ARGON2_CURRENT_VERSION]
     pw_hash = Argon2id(
         salt=salt,
         length=_ARGON2_KEY_LEN,
-        iterations=_ARGON2_ITERATIONS,
+        iterations=params["iterations"],
         lanes=_ARGON2_LANES,
-        memory_cost=_ARGON2_MEMORY_COST,
+        memory_cost=params["memory_cost"],
     ).derive(password.encode())
     return pw_hash, salt
 
 
-def verify_password(password: str, pw_hash: bytes, salt: bytes) -> bool:
+def verify_password(password: str, pw_hash: bytes, salt: bytes, version: int = 2) -> bool:
+    params = _ARGON2_VERSIONS.get(version, _ARGON2_VERSIONS[_ARGON2_CURRENT_VERSION])
     try:
         Argon2id(
             salt=salt,
             length=_ARGON2_KEY_LEN,
-            iterations=_ARGON2_ITERATIONS,
+            iterations=params["iterations"],
             lanes=_ARGON2_LANES,
-            memory_cost=_ARGON2_MEMORY_COST,
+            memory_cost=params["memory_cost"],
         ).verify(password.encode(), pw_hash)
         return True
     except Exception:
         return False
 
 
+def pw_needs_rehash(version: int) -> bool:
+    return version < _ARGON2_CURRENT_VERSION
+
+
+def current_pw_version() -> int:
+    return _ARGON2_CURRENT_VERSION
+
+
 # ── JWT ───────────────────────────────────────────────────────────────────────
 
 def issue_access_token(
@@ -135,6 +162,26 @@ def decode_access_token(token: str) -> dict:
     return jwt.decode(token, _hub_pk_pem, algorithms=["EdDSA"])
 
 
+# ── Email encryption at rest ──────────────────────────────────────────────────
+
+def encrypt_email(plaintext: str) -> str:
+    """Encrypt an email address for storage. Returns base64(nonce + ciphertext)."""
+    if _email_key is None:
+        raise RuntimeError("Hub keypair not loaded")
+    nonce = os.urandom(12)
+    ct = AESGCM(_email_key).encrypt(nonce, plaintext.encode(), None)
+    return base64.b64encode(nonce + ct).decode()
+
+
+def decrypt_email(stored: str) -> str:
+    """Decrypt an email address from storage."""
+    if _email_key is None:
+        raise RuntimeError("Hub keypair not loaded")
+    raw = base64.b64decode(stored)
+    nonce, ct = raw[:12], raw[12:]
+    return AESGCM(_email_key).decrypt(nonce, ct, None).decode()
+
+
 # ── Refresh tokens ────────────────────────────────────────────────────────────
 
 def generate_refresh_token() -> tuple[str, str]:
diff --git a/packages/meshbay-hub/src/meshbay_hub/config.py b/packages/meshbay-hub/src/meshbay_hub/config.py
index 297ace7..31912d1 100644
--- a/packages/meshbay-hub/src/meshbay_hub/config.py
+++ b/packages/meshbay-hub/src/meshbay_hub/config.py
@@ -41,6 +41,7 @@ class HubIdentityConfig:
     id:               str  = "meshbay.org"
     private_key_path: Path = field(default_factory=lambda:
         Path.home() / ".config" / "meshbay" / "hub_private.pem")
+    admin_usernames: list[str] = field(default_factory=list)
 
 
 @dataclass
@@ -75,6 +76,8 @@ def load_config(path: Path | None = None) -> HubConfig:
                 cfg.identity.id = idn.get("id", cfg.identity.id)
                 if kp := idn.get("private_key_path"):
                     cfg.identity.private_key_path = Path(kp).expanduser()
+                if admins := idn.get("admin_usernames"):
+                    cfg.identity.admin_usernames = list(admins)
             if jwt := raw.get("jwt", {}):
                 cfg.jwt.access_token_ttl  = jwt.get("access_token_ttl",  cfg.jwt.access_token_ttl)
                 cfg.jwt.refresh_token_ttl = jwt.get("refresh_token_ttl", cfg.jwt.refresh_token_ttl)
@@ -91,5 +94,7 @@ def load_config(path: Path | None = None) -> HubConfig:
         cfg.identity.id = hub_id
     if kp := os.environ.get("MESHBAY_HUB_KEY"):
         cfg.identity.private_key_path = Path(kp).expanduser()
+    if admin_users := os.environ.get("MESHBAY_ADMIN_USERS"):
+        cfg.identity.admin_usernames = [u.strip() for u in admin_users.split(",") if u.strip()]
 
     return cfg
diff --git a/packages/meshbay-hub/src/meshbay_hub/csam.py b/packages/meshbay-hub/src/meshbay_hub/csam.py
index 4dd0aed..2a0ce25 100644
--- a/packages/meshbay-hub/src/meshbay_hub/csam.py
+++ b/packages/meshbay-hub/src/meshbay_hub/csam.py
@@ -124,14 +124,14 @@ def check_content_hash(blake3_hex: str) -> bool:
 # ── Hub API integration ───────────────────────────────────────────────────────
 
 from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
-from meshbay_hub.api.deps import get_current_user
+from meshbay_hub.api.deps import require_admin
 from meshbay_hub.db.models import User
 
 csam_router = APIRouter(prefix="/v1/admin/csam", tags=["csam"])
 
 
 @csam_router.get("/status")
-async def csam_status(current_user: User = Depends(get_current_user)):
+async def csam_status(current_user: User = Depends(require_admin)):
     """Return CSAM checker status (hash count, database path)."""
     return {
         "hash_count": _checker.hash_count,
@@ -144,7 +144,7 @@ async def csam_status(current_user: User = Depends(get_current_user)):
 @csam_router.post("/check")
 async def check_hash(
     body: dict,
-    current_user: User = Depends(get_current_user),
+    current_user: User = Depends(require_admin),
 ):
     """
     Check a single hash. Admin use only.
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py
index 2aeae41..c0b0491 100644
--- a/packages/meshbay-hub/src/meshbay_hub/db/models.py
+++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py
@@ -42,6 +42,7 @@ class User(Base):
     email:       Mapped[str] = mapped_column(String(256), nullable=False)   # kept for recovery
     pw_hash:     Mapped[bytes] = mapped_column(nullable=False)
     pw_salt:     Mapped[bytes] = mapped_column(nullable=False)
+    pw_version:  Mapped[int]  = mapped_column(Integer, default=1)
     pk_ed25519:  Mapped[str] = mapped_column(String(64), nullable=False)    # base64 raw 32B
     pk_x25519:   Mapped[str] = mapped_column(String(64), nullable=False)    # base64 raw 32B
     hub_id:           Mapped[str] = mapped_column(String(128), nullable=False)
@@ -132,17 +133,31 @@ class RefreshToken(Base):
     id:         Mapped[str] = mapped_column(String(36), primary_key=True, default=_uuid)
     user_id:    Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=False)
     token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)  # blake3 hex
+    family_id:  Mapped[str] = mapped_column(String(36), nullable=False, default=_uuid)
     expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
     revoked:    Mapped[bool] = mapped_column(Boolean, default=False)
 
     user: Mapped["User"] = relationship(back_populates="refresh_tokens")
 
-    __table_args__ = (Index("ix_refresh_tokens_hash", "token_hash"),)
+    __table_args__ = (
+        Index("ix_refresh_tokens_hash", "token_hash"),
+        Index("ix_refresh_tokens_family", "family_id"),
+    )
 
 
 # ── IP logs (legal compliance) ────────────────────────────────────────────────
 
+class HubPeer(Base):
+    """Trusted peer hub for MHP federation."""
+    __tablename__ = "hub_peers"
+
+    hub_id:        Mapped[str] = mapped_column(String(128), primary_key=True)
+    hub_url:       Mapped[str] = mapped_column(String(256), nullable=False)
+    pk_hub_pem:    Mapped[str] = mapped_column(Text, nullable=False)
+    trusted_since: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+
+
 class FederatedGroup(Base):
     """
     Groups received from peer hubs via MHP federation.
diff --git a/packages/meshbay-hub/src/meshbay_hub/tasks/__init__.py b/packages/meshbay-hub/src/meshbay_hub/tasks/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/tasks/__init__.py
diff --git a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
new file mode 100644
index 0000000..7674c6e
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
@@ -0,0 +1,40 @@
+"""Scheduled cleanup tasks — IP log purge (1-year retention)."""
+
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from meshbay_hub.db.models import IPLog
+
+log = logging.getLogger(__name__)
+
+RETENTION_DAYS = 365
+CLEANUP_INTERVAL_HOURS = 24
+
+
+async def purge_old_ip_logs(db: AsyncSession, retention_days: int = RETENTION_DAYS) -> int:
+    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
+    result = await db.execute(delete(IPLog).where(IPLog.timestamp < cutoff))
+    await db.commit()
+    return result.rowcount
+
+
+async def cleanup_loop(get_session):
+    """Run cleanup once at startup, then every 24 hours."""
+    try:
+        while True:
+            try:
+                async with get_session() as db:
+                    deleted = await purge_old_ip_logs(db)
+                    if deleted:
+                        log.info("Purged %d IP log entries older than %d days", deleted, RETENTION_DAYS)
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                log.error("IP log cleanup failed: %s", e)
+            await asyncio.sleep(CLEANUP_INTERVAL_HOURS * 3600)
+    except asyncio.CancelledError:
+        return