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>

1 files changed, 197 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py
index 568d5d9..327158f 100644
--- a/packages/meshbay-hub/tests/test_hub_api.py
+++ b/packages/meshbay-hub/tests/test_hub_api.py
@@ -11,6 +11,7 @@ from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
 from cryptography.hazmat.primitives import serialization
 
 from meshbay_common.crypto import generate_gek, pk_to_b64, wrap_gek
+from meshbay_hub.api.deps import set_admin_usernames
 
 
 def _gen_user_keys():
@@ -35,6 +36,16 @@ async def test_hub_info(client):
 
 
 @pytest.mark.asyncio
+async def test_health(client):
+    r = await client.get("/v1/health")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["status"] == "ok"
+    assert "version" in data
+    assert "connected_nodes" in data
+
+
+@pytest.mark.asyncio
 async def test_hub_pubkey(client):
     r = await client.get("/v1/hub/pubkey")
     assert r.status_code == 200
@@ -123,6 +134,33 @@ async def test_token_refresh(client):
     r2 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt})
     assert r2.status_code == 200
     assert r2.json()["access_token"] != at   # new token (different jti)
+    assert "refresh_token" in r2.json()       # rotated refresh token returned
+
+
+@pytest.mark.asyncio
+async def test_refresh_token_rotation_old_rejected(client):
+    """After rotation, old refresh token is rejected."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "rot_user", "email": "rot@x.com", "password": "rotpass99",
+        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
+    r = await client.post("/v1/users/login", json={
+        "username": "rot_user", "password": "rotpass99"})
+    rt1 = r.json()["refresh_token"]
+
+    r2 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt1})
+    assert r2.status_code == 200
+    rt2 = r2.json()["refresh_token"]
+    assert rt2 != rt1
+
+    # Old token reuse → detected and family revoked
+    r3 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt1})
+    assert r3.status_code == 401
+    assert "reuse" in r3.json()["detail"].lower()
+
+    # New token also revoked (entire family)
+    r4 = await client.post("/v1/users/token/refresh", json={"refresh_token": rt2})
+    assert r4.status_code == 401
 
 
 @pytest.mark.asyncio
@@ -311,3 +349,162 @@ async def test_jwt_contains_groups_claim(client):
         "username": "grp_alice", "password": "alicepass99"})
     decoded_alice = pyjwt.decode(r.json()["access_token"], hub_pk, algorithms=["EdDSA"])
     assert group_id in decoded_alice["groups"]
+
+
+# ── Admin authz (8.1) ───────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_non_admin_cannot_revoke(client):
+    """Non-admin user gets 403 on admin endpoints."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "regular_user", "email": "ru@x.com", "password": "regularpass",
+        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
+    r = await client.post("/v1/users/login", json={
+        "username": "regular_user", "password": "regularpass"})
+    token = r.json()["access_token"]
+
+    set_admin_usernames(["someone_else"])
+    r = await client.post("/v1/admin/revoke", json={
+        "target": "user", "target_id": "fake-id", "reason": "test",
+    }, headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 403
+    assert "Admin" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_admin_can_revoke(client):
+    """Admin user (in config) can access admin endpoints."""
+    pk_ed_v, pk_x_v, _ = _gen_user_keys()
+    r = await client.post("/v1/users/register", json={
+        "username": "victim_a", "email": "va@x.com", "password": "victimpass9",
+        "pk_user_ed25519": pk_ed_v, "pk_user_x25519": pk_x_v})
+    victim_id = r.json()["user_id"]
+
+    pk_ed_a, pk_x_a, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "the_admin", "email": "ta@x.com", "password": "adminpass99",
+        "pk_user_ed25519": pk_ed_a, "pk_user_x25519": pk_x_a})
+    r = await client.post("/v1/users/login", json={
+        "username": "the_admin", "password": "adminpass99"})
+    admin_token = r.json()["access_token"]
+
+    set_admin_usernames(["the_admin"])
+    r = await client.post("/v1/admin/revoke", json={
+        "target": "user", "target_id": victim_id, "reason": "test",
+    }, headers={"Authorization": f"Bearer {admin_token}"})
+    assert r.status_code == 200
+    assert r.json()["status"] == "revoked"
+
+
+# ── Email encryption (8.2) ──────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_email_encrypted_at_rest(client):
+    """Email stored in DB must not contain plaintext address."""
+    from meshbay_hub.auth import encrypt_email, decrypt_email
+    encrypted = encrypt_email("test@example.com")
+    assert "@" not in encrypted
+    assert decrypt_email(encrypted) == "test@example.com"
+
+
+@pytest.mark.asyncio
+async def test_registered_email_not_plaintext(client, app):
+    """Registration stores encrypted email, not plaintext."""
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "email_test", "email": "secret@example.com",
+        "password": "emailpass9",
+        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
+
+    from meshbay_hub.db.engine import get_db
+    from meshbay_hub.db.models import User
+    from sqlalchemy import select
+
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "email_test"))
+        user = result.scalar_one()
+        assert "@" not in user.email
+        from meshbay_hub.auth import decrypt_email
+        assert decrypt_email(user.email) == "secret@example.com"
+        break
+
+
+# ── Argon2id rehash (8.10) ──────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_password_rehash_on_login(client, app):
+    """Users with pw_version=1 get rehashed to current version on login."""
+    from meshbay_hub.auth import current_pw_version
+    from meshbay_hub.db.engine import get_db
+    from meshbay_hub.db.models import User
+    from sqlalchemy import select
+
+    pk_ed, pk_x, _ = _gen_user_keys()
+    await client.post("/v1/users/register", json={
+        "username": "rehash_user", "email": "rh@x.com", "password": "rehashpass9",
+        "pk_user_ed25519": pk_ed, "pk_user_x25519": pk_x})
+
+    # Force pw_version to 1 (simulating pre-upgrade user)
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "rehash_user"))
+        user = result.scalar_one()
+        user.pw_version = 1
+        # Re-hash with v1 params so verify_password(version=1) succeeds
+        from meshbay_hub.auth import _ARGON2_VERSIONS, _ARGON2_KEY_LEN, _ARGON2_LANES
+        from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
+        import os
+        salt = os.urandom(16)
+        params = _ARGON2_VERSIONS[1]
+        pw_hash = Argon2id(
+            salt=salt, length=_ARGON2_KEY_LEN, iterations=params["iterations"],
+            lanes=_ARGON2_LANES, memory_cost=params["memory_cost"],
+        ).derive(b"rehashpass9")
+        user.pw_hash = pw_hash
+        user.pw_salt = salt
+        await db.commit()
+        break
+
+    # Login should succeed and trigger rehash
+    r = await client.post("/v1/users/login", json={
+        "username": "rehash_user", "password": "rehashpass9"})
+    assert r.status_code == 200
+
+    # Verify pw_version is now current
+    async for db in get_db():
+        result = await db.execute(select(User).where(User.username == "rehash_user"))
+        user = result.scalar_one()
+        assert user.pw_version == current_pw_version()
+        break
+
+    # Login still works after rehash
+    r = await client.post("/v1/users/login", json={
+        "username": "rehash_user", "password": "rehashpass9"})
+    assert r.status_code == 200
+
+
+# ── IP log cleanup (8.9) ────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_ip_log_cleanup(app):
+    """Old IP log entries are purged by cleanup task."""
+    from datetime import datetime, timezone, timedelta
+    from meshbay_hub.db.engine import get_db
+    from meshbay_hub.db.models import IPLog
+    from meshbay_hub.tasks.cleanup import purge_old_ip_logs
+
+    async for db in get_db():
+        old_ts = datetime.now(timezone.utc) - timedelta(days=400)
+        db.add(IPLog(event="test_old", ip_address="1.2.3.4", timestamp=old_ts))
+        db.add(IPLog(event="test_recent", ip_address="5.6.7.8"))
+        await db.commit()
+
+        deleted = await purge_old_ip_logs(db, retention_days=365)
+        assert deleted == 1
+
+        from sqlalchemy import select, func
+        count = (await db.execute(
+            select(func.count()).where(IPLog.event.in_(["test_old", "test_recent"]))
+        )).scalar_one()
+        assert count == 1
+        break