diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 3673ccf..e028381 100644 --- a/backend/src/cyclone/api.py +++ b/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 ``) 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) diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index bf5131e..36f7058 100644 --- a/backend/src/cyclone/db.py +++ b/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: diff --git a/backend/src/cyclone/db_crypto.py b/backend/src/cyclone/db_crypto.py index e6c0bdc..e613ce5 100644 --- a/backend/src/cyclone/db_crypto.py +++ b/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 = ""`` (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 = ""`` 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) + diff --git a/backend/tests/test_api_rotate_key.py b/backend/tests/test_api_rotate_key.py new file mode 100644 index 0000000..5e094ae --- /dev/null +++ b/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() diff --git a/backend/tests/test_db_crypto.py b/backend/tests/test_db_crypto.py index 0bc9465..d9655d7 100644 --- a/backend/tests/test_db_crypto.py +++ b/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()