feat(sp15): SQLCipher key rotation via PRAGMA rekey
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).
This commit is contained in:
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user