tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: SP15 SQLCipher key rotation into main

Browse Source
This commit is contained in:
Tyler
2026-06-21 00:33:40 -06:00
parent 6233df1270 47902fd6b2
commit ab00909715
5 changed files with 756 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+155
View File
@@ -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
+38
View File
@@ -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
+229 -2
View File
@@ -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
+217
View File
@@ -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
+117
View File
@@ -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()