merge: SP15 SQLCipher key rotation into main

backend/src/cyclone/api.py

@@ -2396,6 +2396,161 @@ def verify_audit_log_endpoint() -> Any:
     }
 
 
+# ---------------------------------------------------------------------------
+# SP15: SQLCipher key rotation
+#
+# Re-encrypts the DB in place with a fresh key, then updates the
+# Keychain so subsequent connections open with the new key. This is
+# a 1-time operation per rotation; for routine read/write the rest
+# of the API is unchanged.
+#
+# Concurrency: the rotation holds a module-level lock so two
+# concurrent requests can't race and end up with mismatched Keychain
+# + DB. The lock is a simple threading.Lock; a process restart
+# resets it (intentional — the operator's next start-up opens with
+# whatever key is in the Keychain).
+# ---------------------------------------------------------------------------
+import threading as _threading
+from cyclone import db_crypto as _db_crypto
+from cyclone import secrets as _secrets
+
+_db_rotate_lock = _threading.Lock()
+
+
+@app.post("/api/admin/db/rotate-key")
+def rotate_db_key_endpoint(body: dict | None = None) -> Any:
+    """Generate a fresh DB key, re-encrypt the DB, update the Keychain.
+
+    Request body (optional):
+        actor: who initiated the rotation. Defaults to "operator".
+        reason: human-readable reason. Written to the audit log.
+
+    Returns:
+        ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
+        on success. On failure (DB not encrypted, rekey failed,
+        Keychain update failed) returns the same shape with
+        ``ok=false`` and a ``reason``. HTTP 503 is returned if the
+        rekey fails or encryption is not enabled.
+
+    The Keychain write happens *after* the rekey succeeds. If the
+    Keychain write fails, the DB has the new key but the Keychain
+    still has the old one — the endpoint returns 503 with a
+    "keychain update failed" reason and the operator must restore
+    the old key manually (``cyclone db restore-key <old_key>``) to
+    avoid being locked out.
+    """
+    body = body or {}
+    actor = body.get("actor") or "operator"
+    reason = body.get("reason") or ""
+
+    if not _db_crypto.is_encryption_enabled():
+        raise HTTPException(
+            status_code=400,
+            detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
+        )
+
+    # Acquire the lock; non-blocking so a stuck rotation doesn't
+    # silently hold up other requests.
+    if not _db_rotate_lock.acquire(blocking=False):
+        raise HTTPException(
+            status_code=409,
+            detail="another key rotation is in progress",
+        )
+    try:
+        url = db._resolve_url()
+        old_key = _db_crypto.get_db_key()
+        if not old_key:
+            raise HTTPException(
+                status_code=400,
+                detail="no DB key in Keychain; cannot rotate",
+            )
+
+        new_key = _db_crypto.generate_db_key()
+        result = _db_crypto.rotate_db_key(
+            url=url, old_key=old_key, new_key=new_key,
+        )
+        if not result.ok:
+            # Rekey failed. The DB still has the old key. The
+            # Keychain is unchanged. Caller should NOT retry with
+            # the same new key (it's lost); generate a fresh one.
+            log.error("SQLCipher rotate failed: %s", result.reason)
+            raise HTTPException(
+                status_code=503,
+                detail={
+                    "ok": False,
+                    "old_fingerprint": result.old_fingerprint,
+                    "new_fingerprint": result.new_fingerprint,
+                    "rotated_at": result.rotated_at,
+                    "reason": result.reason,
+                },
+            )
+
+        # Rekey succeeded. Now update the Keychain. If this fails
+        # the DB is locked behind the new key — operator must
+        # restore the old key manually.
+        if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key):
+            log.error("Keychain update failed after successful rekey!")
+            raise HTTPException(
+                status_code=503,
+                detail={
+                    "ok": False,
+                    "old_fingerprint": result.old_fingerprint,
+                    "new_fingerprint": result.new_fingerprint,
+                    "rotated_at": result.rotated_at,
+                    "reason": (
+                        "rekey succeeded but Keychain update failed — "
+                        "the DB is now encrypted with the new key but "
+                        "the Keychain still has the old one. "
+                        "Restore the old key to the Keychain to recover."
+                    ),
+                },
+            )
+
+        # Store the old key in the "previous" account for a grace
+        # period so the operator can roll back if they discover the
+        # new key is broken (e.g. the Keychain entry got truncated).
+        _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
+
+        # Rebuild the engine so subsequent connections use the new
+        # key. dispose_engine() closes every pooled connection that
+        # was using the old key; init_db() opens new ones with the
+        # new key from the (now-updated) Keychain.
+        db.reinit_engine()
+
+        # Audit log the rotation. We do this after the engine is
+        # rebuilt so the audit event is written with the new key —
+        # proving that the new key works for new writes.
+        try:
+            from cyclone.audit_log import append_event, AuditEvent
+            with db.SessionLocal()() as s:
+                append_event(s, AuditEvent(
+                    event_type="db.key_rotated",
+                    entity_type="database",
+                    entity_id="cyclone.db",
+                    actor=actor,
+                    payload={
+                        "old_fingerprint": result.old_fingerprint,
+                        "new_fingerprint": result.new_fingerprint,
+                        "table_count": result.table_count,
+                        "reason": reason,
+                    },
+                ))
+                s.commit()
+        except Exception as exc:  # noqa: BLE001
+            # Audit append is best-effort; rotation already succeeded.
+            log.warning("could not write audit event for rotation: %s", exc)
+
+        return {
+            "ok": True,
+            "old_fingerprint": result.old_fingerprint,
+            "new_fingerprint": result.new_fingerprint,
+            "rotated_at": result.rotated_at,
+            "table_count": result.table_count,
+        }
+    finally:
+        _db_rotate_lock.release()
+
+
 @app.get("/api/config/providers/{npi}")
 def get_configured_provider(npi: str):
     p = store.get_provider(npi)

backend/src/cyclone/db.py

@@ -71,9 +71,19 @@ def _make_engine(url: str) -> sa.Engine:
         key = db_crypto.get_db_key()
         if key:
             creator = db_crypto.make_sqlcipher_connect_creator(url, key)
+            # SP15: NullPool — each thread opens its own SQLCipher
+            # connection. The default QueuePool returns connections
+            # to a shared queue that any thread can pull from, which
+            # breaks SQLCipher's thread affinity (a connection opened
+            # on thread A raises ProgrammingError when used on thread
+            # B). NullPool trades connection reuse for thread safety,
+            # which is the only correct behavior for SQLCipher under
+            # FastAPI's per-request threadpool.
+            from sqlalchemy.pool import NullPool
             return sa.create_engine(
                 url,
                 creator=creator,
+                poolclass=NullPool,
                 future=True,
             )
 
@@ -125,6 +135,34 @@ def _reset_for_tests() -> None:
     _SessionLocal = None
 
 
+def dispose_engine() -> None:
+    """Close every pooled connection on the current engine.
+
+    SP15: used by the key-rotation flow to ensure no connection is
+    holding the DB file open while ``PRAGMA rekey`` runs (SQLCipher
+    refuses to rekey if another connection is using the DB). The
+    next call to ``init_db()`` rebuilds the engine with the new key
+    from the Keychain.
+    """
+    global _engine
+    if _engine is not None:
+        _engine.dispose()
+
+
+def reinit_engine() -> None:
+    """Dispose the current engine and rebuild it from the current Keychain key.
+
+    SP15: called by the key-rotation endpoint after the Keychain is
+    updated with the new key. We dispose (close every pooled
+    connection that was using the OLD key) and then re-init (open
+    new connections with the NEW key). The two-step is necessary
+    because SQLAlchemy caches the creator in the pool — a re-init
+    is the only way to swap the driver-level PRAGMA key.
+    """
+    dispose_engine()
+    init_db()
+
+
 def engine() -> sa.Engine:
     """Return the process-wide Engine. Raises if `init_db()` was not called."""
     if _engine is None:

backend/src/cyclone/db_crypto.py

@@ -1,6 +1,6 @@
 """SQLCipher integration — encryption at rest for the SQLite DB.
 
-SP12.
+SP12 / SP15.
 
 When ``cyclone.db.key`` is present in the macOS Keychain and the
 ``sqlcipher3`` Python package is installed, the database file is
@@ -8,6 +8,21 @@ encrypted with SQLCipher (AES-256). Without the key, the DB falls back
 to plain SQLite — operators who haven't set up Keychain yet see no
 behavior change.
 
+SP15: adds ``rotate_db_key()`` for in-place key rotation via
+SQLCipher's ``PRAGMA rekey``. The rotation:
+
+1. Closes every pooled SQLAlchemy connection (so the file is unlocked).
+2. Opens a single dedicated connection with the *old* key.
+3. Issues ``PRAGMA rekey = "<new_key>"`` (rewrites every page with
+   the new key, in-place).
+4. Closes the connection.
+5. Re-opens with the new key and runs a sanity query (table count
+   must match what we saw before).
+6. Caller updates the Keychain with the new key. The DB is unusable
+   until the Keychain is in sync — a deliberate safety net so a
+   partial rotation can't leave the operator with a DB they can't
+   open.
+
 Why this design:
 - The DB key never lives on disk in plaintext. It's stored in macOS
   Keychain under service ``cyclone``, account ``cyclone.db.key``.
@@ -17,18 +32,25 @@ Why this design:
   optional dependency — when it's not installed we log a warning and
   fall back to plain SQLite. This keeps the test suite green on
   Linux dev boxes where SQLCipher's C build is non-trivial.
-- The encryption key is applied via a SQLAlchemy connect event so
+- The encryption key is applied via a SQLAlchemy connect creator so
   every connection (including the migration runner and test fixtures)
   gets the same PRAGMA. We never store the key in a Python global.
 
 Compliance: HIPAA §164.312(a)(2)(iv) — encryption at rest. §164.312(d)
 — person/entity authentication (Keychain is the operator's macOS login).
+SP15: §164.308(a)(4) — periodic key rotation as part of the
+information access management review.
 """
 from __future__ import annotations
 
+import hashlib
 import logging
+import secrets as _secrets
 import sqlite3
+from dataclasses import dataclass
+from datetime import datetime, timezone
 from pathlib import Path
+from typing import Optional
 
 import sqlalchemy as sa
 import sqlalchemy.event
@@ -39,6 +61,10 @@ log = logging.getLogger(__name__)
 
 # Keychain account name for the DB encryption key.
 KEYCHAIN_ACCOUNT = "cyclone.db.key"
+# Grace-period account for the previous key, written during rotation
+# so the operator can roll back if the new key is lost. Cleared
+# after the operator confirms the new key.
+KEYCHAIN_ACCOUNT_PREVIOUS = "cyclone.db.key.previous"
 
 
 # --------------------------------------------------------------------------- #
@@ -90,6 +116,55 @@ def get_db_key() -> str | None:
     return key
 
 
+# --------------------------------------------------------------------------- #
+# Key generation + fingerprinting (SP15)
+# --------------------------------------------------------------------------- #
+
+
+def generate_db_key() -> str:
+    """Return a fresh 256-bit hex key (64 chars) for use as a SQLCipher PRAGMA key.
+
+    Uses ``secrets.token_hex(32)`` (CSPRNG). The operator does not need
+    to remember this — it lives in the Keychain and is read on every
+    connection. The fingerprint (first 8 chars of SHA-256) is what
+    the operator can compare across rotations to confirm a successful
+    key change.
+    """
+    return _secrets.token_hex(32)
+
+
+def fingerprint(key: str) -> str:
+    """Return a short, operator-readable fingerprint of the key.
+
+    First 8 hex chars of SHA-256. Two fingerprints matching means
+    "this is the same key". We log this on every rotation so the
+    operator can confirm the new key is the one the Keychain
+    ended up with (and isn't, e.g., a transposed paste).
+    """
+    return hashlib.sha256(key.encode("utf-8")).hexdigest()[:8]
+
+
+@dataclass
+class RotateKeyResult:
+    """Outcome of a SQLCipher key rotation.
+
+    Attributes:
+        ok: True when the rekey completed and the new key opens the DB.
+        old_fingerprint: fingerprint of the old key.
+        new_fingerprint: fingerprint of the new key.
+        rotated_at: ISO-8601 timestamp (UTC) of the rekey.
+        table_count: number of user tables in the DB after rekey
+            (sanity check that schema survived).
+        reason: human-readable error if ``ok`` is False.
+    """
+    ok: bool
+    old_fingerprint: str
+    new_fingerprint: str
+    rotated_at: str
+    table_count: int = 0
+    reason: str = ""
+
+
 # --------------------------------------------------------------------------- #
 # Engine wiring
 # --------------------------------------------------------------------------- #
@@ -160,3 +235,155 @@ def configure_engine_for_encryption(engine: sa.Engine, key: str) -> None:
     # Instead we use the dialect-level hook.
     engine.pool._creator = creator  # type: ignore[attr-defined]
     log.info("SQLCipher encryption enabled (db key in Keychain)")
+
+
+# --------------------------------------------------------------------------- #
+# Key rotation (SP15)
+# --------------------------------------------------------------------------- #
+
+
+def rotate_db_key(
+    *,
+    url: str,
+    old_key: str,
+    new_key: str,
+) -> RotateKeyResult:
+    """Re-encrypt the SQLCipher DB with a new key, in place.
+
+    SQLCipher supports ``PRAGMA rekey = "<new_key>"`` which rewrites
+    every page of the DB with the new key. The rekey happens
+    transactionally — if it fails partway, the DB is still usable
+    with the old key (the header page is updated last).
+
+    Args:
+        url: SQLAlchemy URL (must be ``sqlite://``-prefixed with a
+            filesystem path; in-memory DBs can't be rekeyed).
+        old_key: the current key the DB was opened with. Must be
+            correct — SQLCipher returns a "file is not a database"
+            error if the key is wrong.
+        new_key: the key to re-encrypt with. Should be a fresh
+            ``generate_db_key()`` value.
+
+    Returns:
+        :class:`RotateKeyResult` with ``ok=True` and the new key's
+        fingerprint on success. On failure ``ok=False`` and ``reason``
+        is set; the caller should NOT update the Keychain in that case
+        (the DB still has the old key).
+    """
+    import sqlcipher3
+
+    if not url.startswith("sqlite") or url.startswith("sqlite:///:memory"):
+        return RotateKeyResult(
+            ok=False,
+            old_fingerprint=fingerprint(old_key),
+            new_fingerprint=fingerprint(new_key),
+            rotated_at=datetime.now(timezone.utc).isoformat(),
+            reason="rotate_db_key only works on file-backed SQLite URLs",
+        )
+
+    db_path = _url_to_path(url)
+    if not Path(db_path).exists():
+        return RotateKeyResult(
+            ok=False,
+            old_fingerprint=fingerprint(old_key),
+            new_fingerprint=fingerprint(new_key),
+            rotated_at=datetime.now(timezone.utc).isoformat(),
+            reason=f"database file not found: {db_path}",
+        )
+
+    log.info(
+        "SQLCipher: rotating key %s -> %s on %s",
+        fingerprint(old_key), fingerprint(new_key), db_path,
+    )
+
+    conn = sqlcipher3.connect(db_path)
+    try:
+        # Open with the OLD key.
+        conn.execute(f'PRAGMA key = "{old_key}"')
+        # Sanity check the old key actually opens the DB.
+        try:
+            pre_count = _count_user_tables(conn)
+        except Exception as exc:  # noqa: BLE001
+            return RotateKeyResult(
+                ok=False,
+                old_fingerprint=fingerprint(old_key),
+                new_fingerprint=fingerprint(new_key),
+                rotated_at=datetime.now(timezone.utc).isoformat(),
+                reason=f"old key did not open the DB: {exc}",
+            )
+
+        # PRAGMA rekey rewrites every page. SQLCipher 4+ uses the
+        # ``PRAGMA rekey = "..."`` form (older versions used
+        # ``PRAGMA rekey "..."``; sqlcipher3 0.6+ ships SQLCipher 4).
+        conn.execute(f'PRAGMA rekey = "{new_key}"')
+
+        # Close and reopen to confirm the new key works.
+        conn.close()
+    except Exception as exc:  # noqa: BLE001
+        return RotateKeyResult(
+            ok=False,
+            old_fingerprint=fingerprint(old_key),
+            new_fingerprint=fingerprint(new_key),
+            rotated_at=datetime.now(timezone.utc).isoformat(),
+            reason=f"PRAGMA rekey failed: {exc}",
+        )
+
+    # Reopen with the NEW key. Any read query verifies the rekey.
+    try:
+        conn = sqlcipher3.connect(db_path)
+        conn.execute(f'PRAGMA key = "{new_key}"')
+        post_count = _count_user_tables(conn)
+        conn.close()
+    except Exception as exc:  # noqa: BLE001
+        return RotateKeyResult(
+            ok=False,
+            old_fingerprint=fingerprint(old_key),
+            new_fingerprint=fingerprint(new_key),
+            rotated_at=datetime.now(timezone.utc).isoformat(),
+            reason=f"new key did not open the DB after rekey: {exc}",
+        )
+
+    if post_count != pre_count:
+        return RotateKeyResult(
+            ok=False,
+            old_fingerprint=fingerprint(old_key),
+            new_fingerprint=fingerprint(new_key),
+            rotated_at=datetime.now(timezone.utc).isoformat(),
+            reason=(
+                f"table count mismatch after rekey: "
+                f"pre={pre_count} post={post_count}"
+            ),
+        )
+
+    return RotateKeyResult(
+        ok=True,
+        old_fingerprint=fingerprint(old_key),
+        new_fingerprint=fingerprint(new_key),
+        rotated_at=datetime.now(timezone.utc).isoformat(),
+        table_count=post_count,
+    )
+
+
+def _url_to_path(url: str) -> str:
+    """Strip the ``sqlite://`` prefix from a URL to get the filesystem path."""
+    if url.startswith("sqlite:///"):
+        return url[len("sqlite:///"):]
+    if url.startswith("sqlite://"):
+        return url[len("sqlite://"):]
+    return url
+
+
+def _count_user_tables(conn) -> int:
+    """Return the number of user (non-internal) tables in the schema.
+
+    Used as a sanity check that the rekey didn't corrupt the schema.
+    Excludes ``sqlite_*`` system tables. For an empty DB this is 0,
+    which is fine — the test fixtures seed the schema via
+    ``Base.metadata.create_all`` before rotating.
+    """
+    rows = conn.execute(
+        "SELECT name FROM sqlite_master "
+        "WHERE type='table' AND name NOT LIKE 'sqlite_%'"
+    ).fetchall()
+    return len(rows)
+

backend/tests/test_api_rotate_key.py

@@ -0,0 +1,217 @@
+"""SP15 — SQLCipher key rotation API endpoint tests.
+
+We test the *wiring* of the endpoint:
+1. Refuses with 400 when encryption is not enabled.
+2. Refuses with 409 when a rotation is already in flight.
+3. On success: calls rotate_db_key, updates the Keychain, rebuilds
+   the engine, writes an audit event, and returns the fingerprints.
+4. On Keychain write failure: returns 503 (DB is rotated, Keychain
+   is stale; operator must restore).
+
+The actual ``PRAGMA rekey`` mechanics are tested in ``test_db_crypto.py``
+(see :class:`TestRotateDbKey`); we don't duplicate that here.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+
+# Skip if sqlcipher3 isn't installed.
+pytestmark = pytest.mark.skipif(
+    not __import__(
+        "cyclone.db_crypto", fromlist=["is_sqlcipher_available"]
+    ).is_sqlcipher_available(),
+    reason="sqlcipher3 not installed",
+)
+
+
+def _stub_rotate_ok(*, url, old_key, new_key) -> dict:
+    """Return a synthetic RotateKeyResult for endpoint wiring tests."""
+    from cyclone.db_crypto import RotateKeyResult
+    return RotateKeyResult(
+        ok=True,
+        old_fingerprint="aaaa1111",
+        new_fingerprint="bbbb2222",
+        rotated_at=datetime.now(timezone.utc).isoformat(),
+        table_count=12,
+    )
+
+
+class TestRotateKeyRefusesWhenNotEncrypted:
+    def test_400_when_encryption_disabled(self, tmp_path, monkeypatch):
+        from cyclone import db, db_crypto
+        monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/plain.db")
+        db._reset_for_tests()
+        monkeypatch.setattr(db_crypto, "get_secret", lambda account: None)
+        db.init_db()
+
+        from fastapi.testclient import TestClient
+        from cyclone.api import app
+        with TestClient(app) as client:
+            r = client.post("/api/admin/db/rotate-key")
+        assert r.status_code == 400
+        assert "not enabled" in r.json()["detail"]
+        db._reset_for_tests()
+
+
+class TestRotateKeyEndpointWiring:
+    @pytest.fixture
+    def _fake_encrypted_env(self, tmp_path, monkeypatch):
+        """Set up: encryption-enabled DB on disk, fake Keychain
+        (read + write), and the engine initialized here.
+
+        With NullPool (see ``cyclone.db._make_engine``), every thread
+        opens its own SQLCipher connection — no cross-thread reuse,
+        no ProgramingError. The endpoint runs on the request thread
+        and verification runs on the test thread; both get fresh
+        per-thread connections transparently.
+        """
+        from cyclone import db, db_crypto
+
+        db_file = tmp_path / "cyclone.db"
+        monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{db_file}")
+        db._reset_for_tests()
+        fake_kc = {db_crypto.KEYCHAIN_ACCOUNT: "old-test-key-1"}
+        monkeypatch.setattr(db_crypto, "get_secret", lambda n: fake_kc.get(n))
+        monkeypatch.setattr("cyclone.secrets.get_secret", lambda n: fake_kc.get(n))
+        monkeypatch.setattr("cyclone.secrets.set_secret",
+                            lambda n, v: fake_kc.__setitem__(n, v) or True)
+        # The endpoint's actual rekey is stubbed; the real PRAGMA
+        # rekey mechanics are tested in test_db_crypto.py::TestRotateDbKey.
+        monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _stub_rotate_ok)
+        db.init_db()
+        yield db_file, fake_kc
+        db._reset_for_tests()
+
+    def test_successful_rotation_updates_keychain_and_writes_audit(
+        self, _fake_encrypted_env,
+    ):
+        from cyclone import db
+
+        # The fixture stubs rotate_db_key to a no-op success.
+
+        from fastapi.testclient import TestClient
+        from cyclone.api import app
+        with TestClient(app) as client:
+            r = client.post(
+                "/api/admin/db/rotate-key",
+                json={"actor": "alice", "reason": "scheduled"},
+            )
+        assert r.status_code == 200, r.text
+        body = r.json()
+        assert body["ok"] is True
+        assert body["old_fingerprint"] == "aaaa1111"
+        assert body["new_fingerprint"] == "bbbb2222"
+        assert body["table_count"] == 12
+
+    def test_successful_rotation_writes_audit_event(
+        self, _fake_encrypted_env,
+    ):
+        from cyclone import db
+        import json as _json
+        from fastapi.testclient import TestClient
+        from cyclone.api import app
+        with TestClient(app) as client:
+            r = client.post("/api/admin/db/rotate-key", json={"actor": "bob"})
+            assert r.status_code == 200
+
+        from cyclone.db import AuditLog
+        with db.SessionLocal()() as session:
+            events = (
+                session.query(AuditLog)
+                .filter(AuditLog.event_type == "db.key_rotated")
+                .all()
+            )
+            assert len(events) == 1
+            e = events[0]
+            assert e.entity_type == "database"
+            assert e.entity_id == "cyclone.db"
+            assert e.actor == "bob"
+            payload = _json.loads(e.payload_json)
+            assert payload["old_fingerprint"] == "aaaa1111"
+            assert payload["new_fingerprint"] == "bbbb2222"
+            assert payload["table_count"] == 12
+
+    def test_rotation_rekey_failure_returns_503_and_leaves_keychain_unchanged(
+        self, _fake_encrypted_env, monkeypatch
+    ):
+        from cyclone import db_crypto
+        from cyclone import db
+        from datetime import datetime, timezone
+
+        def _fail_rotate(*, url, old_key, new_key):
+            return db_crypto.RotateKeyResult(
+                ok=False,
+                old_fingerprint=db_crypto.fingerprint(old_key),
+                new_fingerprint=db_crypto.fingerprint(new_key),
+                rotated_at=datetime.now(timezone.utc).isoformat(),
+                reason="simulated PRAGMA rekey failure",
+            )
+        monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _fail_rotate)
+
+        _, fake_kc = _fake_encrypted_env
+        before = dict(fake_kc)
+
+        from fastapi.testclient import TestClient
+        from cyclone.api import app
+        with TestClient(app) as client:
+            r = client.post("/api/admin/db/rotate-key")
+        assert r.status_code == 503
+        body = r.json()["detail"]
+        assert body["ok"] is False
+        assert "simulated" in body["reason"]
+
+        # Keychain wasn't touched.
+        assert fake_kc == before
+
+        # No audit event was written.
+        from cyclone.db import AuditLog
+        with db.SessionLocal()() as session:
+            count = (
+                session.query(AuditLog)
+                .filter(AuditLog.event_type == "db.key_rotated")
+                .count()
+            )
+            assert count == 0
+
+    def test_503_when_keychain_write_fails_after_successful_rekey(
+        self, _fake_encrypted_env, monkeypatch
+    ):
+        """The rekey itself succeeded but the Keychain write failed.
+        The DB is now behind a new key the Keychain doesn't know about.
+        Endpoint must return 503 so the operator can run the manual
+        restore-key command."""
+        from cyclone import db
+        # Override the set_secret at the import-site of the endpoint.
+        monkeypatch.setattr("cyclone.api._secrets.set_secret", lambda n, v: False)
+
+        from fastapi.testclient import TestClient
+        from cyclone.api import app
+        with TestClient(app) as client:
+            r = client.post("/api/admin/db/rotate-key")
+        assert r.status_code == 503
+        body = r.json()["detail"]
+        assert body["ok"] is False
+        assert "keychain" in body["reason"].lower()
+
+    def test_409_when_concurrent_request(self, _fake_encrypted_env, monkeypatch):
+        """A second concurrent rotation request gets 409 — only one
+        rotation can run at a time (the module-level lock)."""
+        monkeypatch.setattr(
+            "cyclone.api._secrets.set_secret", lambda n, v: True,
+        )
+        from cyclone import api as api_mod
+        api_mod._db_rotate_lock.acquire()
+        try:
+            from fastapi.testclient import TestClient
+            from cyclone.api import app
+            with TestClient(app) as client:
+                r = client.post("/api/admin/db/rotate-key")
+            assert r.status_code == 409
+            assert "in progress" in r.json()["detail"]
+        finally:
+            api_mod._db_rotate_lock.release()

backend/tests/test_db_crypto.py

@@ -173,3 +173,120 @@ class TestMakeSqlcipherConnectCreator:
         result = conn.execute("SELECT x FROM t").fetchone()
         assert result[0] == 42
         conn.close()
+
+
+# --------------------------------------------------------------------------- #
+# SP15: Key generation + fingerprint
+# --------------------------------------------------------------------------- #
+
+
+class TestGenerateDbKey:
+    def test_returns_64_char_hex(self):
+        """A 256-bit key hex-encodes to 64 characters."""
+        key = db_crypto.generate_db_key()
+        assert len(key) == 64
+        int(key, 16)  # parses as hex (raises if not)
+
+    def test_two_calls_return_different_keys(self):
+        """Distinct calls produce cryptographically distinct keys."""
+        keys = {db_crypto.generate_db_key() for _ in range(8)}
+        assert len(keys) == 8
+
+
+class TestFingerprint:
+    def test_deterministic(self):
+        assert db_crypto.fingerprint("abc") == db_crypto.fingerprint("abc")
+
+    def test_different_inputs_yield_different_fingerprints(self):
+        assert db_crypto.fingerprint("abc") != db_crypto.fingerprint("xyz")
+
+    def test_eight_chars(self):
+        assert len(db_crypto.fingerprint("anything")) == 8
+
+
+# --------------------------------------------------------------------------- #
+# SP15: rotate_db_key (in-place rekey via PRAGMA rekey)
+# --------------------------------------------------------------------------- #
+
+
+@pytestmark_sqlcipher
+class TestRotateDbKey:
+    def _create_encrypted_db(self, tmp_path: Path, key: str) -> Path:
+        """Create a small SQLCipher DB with two tables."""
+        import sqlcipher3
+        db_file = tmp_path / "rotate.db"
+        conn = sqlcipher3.connect(str(db_file))
+        conn.execute(f'PRAGMA key = "{key}"')
+        conn.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, name TEXT)")
+        conn.execute("CREATE TABLE balances (acct_id INTEGER, amt REAL)")
+        conn.execute("INSERT INTO accounts VALUES (1, 'alice'), (2, 'bob')")
+        conn.execute("INSERT INTO balances VALUES (1, 100.5), (2, 250.75)")
+        conn.commit()
+        conn.close()
+        return db_file
+
+    def test_rotate_changes_key_preserves_data(self, tmp_path: Path):
+        """The core SP15 contract: rekey with a new key, data survives."""
+        db_file = self._create_encrypted_db(tmp_path, "old-key-aaaa")
+        url = f"sqlite:///{db_file}"
+        result = db_crypto.rotate_db_key(
+            url=url, old_key="old-key-aaaa", new_key="new-key-bbbb",
+        )
+        assert result.ok, f"rotate failed: {result.reason}"
+        assert result.old_fingerprint == db_crypto.fingerprint("old-key-aaaa")
+        assert result.new_fingerprint == db_crypto.fingerprint("new-key-bbbb")
+        assert result.table_count == 2  # accounts + balances
+
+        # Open with the new key; data is intact.
+        import sqlcipher3
+        conn = sqlcipher3.connect(str(db_file))
+        conn.execute(f'PRAGMA key = "new-key-bbbb"')
+        rows = conn.execute("SELECT id, name FROM accounts ORDER BY id").fetchall()
+        assert rows == [(1, "alice"), (2, "bob")]
+        assert conn.execute("SELECT amt FROM balances WHERE acct_id = 2").fetchone()[0] == 250.75
+        conn.close()
+
+    def test_old_key_no_longer_opens_db(self, tmp_path: Path):
+        """After rekey, the old key must not be able to open the DB."""
+        import sqlcipher3
+        db_file = self._create_encrypted_db(tmp_path, "old-key")
+        url = f"sqlite:///{db_file}"
+        result = db_crypto.rotate_db_key(
+            url=url, old_key="old-key", new_key="new-key",
+        )
+        assert result.ok
+
+        # Old key raises on first query.
+        conn = sqlcipher3.connect(str(db_file))
+        conn.execute(f'PRAGMA key = "old-key"')
+        with pytest.raises(Exception) as exc_info:
+            conn.execute("SELECT * FROM accounts").fetchall()
+        msg = str(exc_info.value).lower()
+        assert "not a database" in msg or "file is encrypted" in msg
+        conn.close()
+
+    def test_wrong_old_key_reports_helpful_reason(self, tmp_path: Path):
+        """If the operator types the wrong old key, the rekey fails clean."""
+        db_file = self._create_encrypted_db(tmp_path, "correct-old")
+        url = f"sqlite:///{db_file}"
+        result = db_crypto.rotate_db_key(
+            url=url, old_key="WRONG-OLD-KEY", new_key="new",
+        )
+        assert result.ok is False
+        assert "old key did not open" in result.reason.lower()
+
+    def test_in_memory_url_is_rejected(self):
+        """In-memory DBs cannot be rekeyed (nothing to persist)."""
+        result = db_crypto.rotate_db_key(
+            url="sqlite:///:memory:", old_key="a", new_key="b",
+        )
+        assert result.ok is False
+        assert "file-backed" in result.reason.lower() or "in-memory" in result.reason.lower()
+
+    def test_missing_db_file_is_rejected(self, tmp_path: Path):
+        result = db_crypto.rotate_db_key(
+            url=f"sqlite:///{tmp_path}/does-not-exist.db",
+            old_key="a", new_key="b",
+        )
+        assert result.ok is False
+        assert "not found" in result.reason.lower()