sp15-bot
47902fd6b2
Adds in-place key rotation for the encrypted DB at rest (HIPAA sec.164.308(a)(4) - periodic key rotation). - db_crypto.rotate_db_key(): opens with old key, issues PRAGMA rekey, reopens with new key, verifies schema survived (table-count sanity). - db_crypto.generate_db_key(): fresh 256-bit CSPRNG hex key. - db_crypto.fingerprint(): SHA-256[:8] of a key, for the operator to compare across rotations. - db.dispose_engine() + db.reinit_engine(): SP15 plumbing. The rotation endpoint disposes the pooled connections (SQLCipher refuses to rekey while another connection holds the file), runs the rekey, then rebuilds the engine with the new key from the Keychain. - API: POST /api/admin/db/rotate-key with module-level threading.Lock to serialize rotations. 400 when encryption not enabled, 409 when a rotation is already in flight, 503 on rekey or Keychain failure with a reason that tells the operator what to do next. - Engine uses NullPool when SQLCipher is enabled: the default QueuePool returns connections to a shared queue that any thread can pull from, which breaks SQLCipher's thread affinity. NullPool trades connection reuse for thread safety, the only correct behavior under FastAPI's per-request threadpool. - Audit event db.key_rotated with old/new fingerprints and table_count, written after the engine is rebuilt so the new key proves it can take new writes. - previous key is stashed to a second Keychain account so the operator can roll back if the new key turns out to be broken. Tests: - test_db_crypto.py: 12 new tests for generate/fingerprint/rekey mechanics (5 require SQLCipher at runtime; skipped otherwise). - test_api_rotate_key.py: 6 new tests for endpoint wiring (encryption-required, Keychain update, audit event, rekey-failure rollback, Keychain-write-failure 503, concurrent-rotation 409).
218 lines
8.2 KiB
Python
218 lines
8.2 KiB
Python
"""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()
|