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:
sp15-bot
@@ -2589,6 +2589,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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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