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

feat(sp12): SQLCipher encryption at rest (optional)

Browse Source 
- New cyclone.db_crypto module:
  * is_sqlcipher_available() — capability check
  * is_encryption_enabled() — Keychain key + sqlcipher3 present
  * get_db_key() — reads 'cyclone.db.key' from Keychain
  * make_sqlcipher_connect_creator(url, key) — SQLAlchemy creator
- db._make_engine() now switches to SQLCipher when key is present
- pyproject.toml: optional 'sqlcipher' extra (sqlcipher3>=0.6,<1)
- Fallback: without Keychain key, DB stays plain SQLite (no surprise
  behavior for operators who haven't set up encryption yet)
- Verified: encrypted file is unreadable as plain SQLite, wrong key
  raises on first query, migrations + ORM work transparently
- HIPAA §164.312(a)(2)(iv) compliance note in docs

Tests: 705 -> 717 (12 new for SQLCipher). All 717 backend tests pass.
This commit is contained in:
Tyler
2026-06-20 23:52:41 -06:00
parent 84d2f39760
commit d54c44f04a
5 changed files with 406 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/db.py
+20 -1
View File
@@ -57,7 +57,26 @@ def _resolve_url() -> str:
def _make_engine(url: str) -> sa.Engine:
"""Build an Engine with sensible defaults for SQLite + FastAPI."""
"""Build an Engine with sensible defaults for SQLite + FastAPI.
SP12: when ``cyclone.db_crypto.is_encryption_enabled()`` returns
True, swap the underlying driver to ``sqlcipher3`` and apply the
Keychain-stored key via a connect-time PRAGMA. Otherwise the
plain sqlite3 driver is used (current behavior, no surprises for
operators who haven't set up Keychain yet).
"""
from cyclone import db_crypto # late import to avoid cycles
if url.startswith("sqlite") and db_crypto.is_encryption_enabled():
key = db_crypto.get_db_key()
if key:
creator = db_crypto.make_sqlcipher_connect_creator(url, key)
return sa.create_engine(
url,
creator=creator,
future=True,
)
connect_args: dict[str, object] = {}
if url.startswith("sqlite"):
connect_args = {"check_same_thread": False}
backend/src/cyclone/db_crypto.py
+162
View File
@@ -0,0 +1,162 @@
"""SQLCipher integration — encryption at rest for the SQLite DB.
SP12.
When ``cyclone.db.key`` is present in the macOS Keychain and the
``sqlcipher3`` Python package is installed, the database file is
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.
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``.
Operators create the entry one-time via ``security add-generic-password``
(see docs/reference/co-medicaid.md §"Keychain setup").
- We don't *require* SQLCipher at import time. ``sqlcipher3`` is an
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
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).
"""
from __future__ import annotations
import logging
import sqlite3
from pathlib import Path
import sqlalchemy as sa
import sqlalchemy.event
from cyclone.secrets import STUB_SECRET, get_secret
log = logging.getLogger(__name__)
# Keychain account name for the DB encryption key.
KEYCHAIN_ACCOUNT = "cyclone.db.key"
# --------------------------------------------------------------------------- #
# Capability checks
# --------------------------------------------------------------------------- #
def is_sqlcipher_available() -> bool:
"""Return True if the ``sqlcipher3`` package is importable.
We import lazily so the check doesn't fail at module import time
on systems that don't have SQLCipher built.
"""
try:
import sqlcipher3 # noqa: F401
return True
except ImportError:
return False
def is_encryption_enabled() -> bool:
"""Return True when SQLCipher is available AND a DB key exists in Keychain.
Both conditions must hold. SQLCipher without a key is useless (we'd
just be running encrypted with a stub secret), and a key without
SQLCipher means we silently degrade to plain SQLite (we'd warn).
"""
if not is_sqlcipher_available():
return False
key = get_secret(KEYCHAIN_ACCOUNT)
return bool(key) and key != STUB_SECRET
# --------------------------------------------------------------------------- #
# Key retrieval
# --------------------------------------------------------------------------- #
def get_db_key() -> str | None:
"""Return the SQLCipher DB key from Keychain, or ``None`` if not set.
``None`` means "fall back to plain SQLite". This is the only
function that reads the key — the engine builder passes the
result directly to the connect creator without storing it.
"""
key = get_secret(KEYCHAIN_ACCOUNT)
if not key or key == STUB_SECRET:
return None
return key
# --------------------------------------------------------------------------- #
# Engine wiring
# --------------------------------------------------------------------------- #
def make_sqlcipher_connect_creator(url: str, key: str):
"""Return a SQLAlchemy connect creator that opens via ``sqlcipher3``.
SQLAlchemy's ``creator`` hook expects a zero-arg callable. We
capture the SQLite URL (extracted from the SQLAlchemy URL) in the
closure and pass it to ``sqlcipher3.connect()`` at every new
pool connection.
Why a creator and not a pool event: SQLAlchemy's creator is the
canonical hook for swapping out the DB-API module. The connect
event would require us to first open a plain connection and then
upgrade it, which doesn't work for SQLCipher because the
encryption happens at the driver level.
"""
import sqlcipher3 # late import — only needed when encryption is on
# Strip the ``sqlite:///`` prefix; SQLCipher takes a plain path.
if url.startswith("sqlite:///"):
db_path = url[len("sqlite:///"):]
elif url.startswith("sqlite://"):
db_path = url[len("sqlite://"):]
else:
# In-memory or other — leave the URL alone.
db_path = url
def _creator() -> sqlite3.Connection:
# SQLCipher's PRAGMA key must be the FIRST statement issued
# on a connection — before any other read or write.
conn = sqlcipher3.connect(db_path)
# SQLCipher accepts hex-encoded keys with ``PRAGMA key = "x'..'"``
# but the simpler ``PRAGMA key = "..."`` form uses PBKDF2 with
# an empty salt — adequate for a key generated by the operator
# (random 32 bytes from /dev/urandom is what we recommend).
conn.execute(f'PRAGMA key = "{key}"')
return conn
return _creator
def configure_engine_for_encryption(engine: sa.Engine, key: str) -> None:
"""Attach the SQLCipher PRAGMA hook to a SQLAlchemy engine.
After this call, every new connection opens via the
``sqlcipher3`` driver with the given key applied. Idempotent —
safe to call once per engine. We use ``connect`` (not ``pool_connect``)
so the key is applied at connection open time, before any other
statement.
"""
creator = make_sqlcipher_connect_creator(key)
# Swap the underlying driver. SQLAlchemy calls ``creator(dbapi_connection_url)``
# for each new pool connection. The ``url`` argument is the path
# string after ``sqlite:///`` (e.g. ``/path/to/cyclone.db``).
@sa.event.listens_for(engine, "connect")
def _on_connect(dbapi_connection, connection_record): # noqa: ANN001
# The engine already routed this connection through the
# creator, which applied PRAGMA key. We could re-issue here
# for paranoia, but it's not needed.
pass
# Replace the pool's creator. SQLAlchemy 2.0 exposes this on the
# pool; setting ``creator`` directly is supported but deprecated.
# Instead we use the dialect-level hook.
engine.pool._creator = creator # type: ignore[attr-defined]
log.info("SQLCipher encryption enabled (db key in Keychain)")