diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 89b9718..348fba8 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -25,6 +25,11 @@ dev = [ "pytest-asyncio>=0.23,<1", "httpx>=0.27,<1", ] +sqlcipher = [ + # SP12: encryption at rest. Optional — without it the DB is plain SQLite. + # Install via: pip install -e .[sqlcipher] (after brew install sqlcipher). + "sqlcipher3>=0.6,<1", +] [project.scripts] cyclone = "cyclone.cli:main" diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py index 33ccf31..1f57a00 100644 --- a/backend/src/cyclone/db.py +++ b/backend/src/cyclone/db.py @@ -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} diff --git a/backend/src/cyclone/db_crypto.py b/backend/src/cyclone/db_crypto.py new file mode 100644 index 0000000..e6c0bdc --- /dev/null +++ b/backend/src/cyclone/db_crypto.py @@ -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)") diff --git a/backend/tests/test_db_crypto.py b/backend/tests/test_db_crypto.py new file mode 100644 index 0000000..0bc9465 --- /dev/null +++ b/backend/tests/test_db_crypto.py @@ -0,0 +1,175 @@ +"""Tests for SQLCipher encryption at rest. SP12. + +We exercise the encryption path end-to-end: +1. With a Keychain key + sqlcipher3 installed: the DB file is encrypted + on disk and decryptable only with the same key. +2. Without a Keychain key: the DB falls back to plain SQLite. +3. With the wrong key: opening the DB raises on the first query. + +``sqlcipher3`` is an optional dep — these tests skip when it isn't +installed, so the suite still runs on Linux dev boxes without SQLCipher. +""" +from __future__ import annotations + +import os +import sqlite3 +import tempfile +from pathlib import Path +from unittest.mock import patch + +import pytest +import sqlalchemy as sa + +from cyclone import db, db_crypto + + +# --------------------------------------------------------------------------- # +# Skip-if-no-sqlcipher gate +# --------------------------------------------------------------------------- # + + +pytestmark_sqlcipher = pytest.mark.skipif( + not db_crypto.is_sqlcipher_available(), + reason="sqlcipher3 not installed (pip install -e .[sqlcipher])", +) + + +# --------------------------------------------------------------------------- # +# Capability checks +# --------------------------------------------------------------------------- # + + +class TestIsSqlcipherAvailable: + def test_returns_true_when_package_installed(self): + """This test only runs when sqlcipher3 is importable.""" + assert db_crypto.is_sqlcipher_available() is True + + +class TestIsEncryptionEnabled: + def test_no_key_disables_encryption(self, monkeypatch): + """Without a Keychain key, encryption is off even with sqlcipher3.""" + monkeypatch.setattr(db_crypto, "get_secret", lambda account: None) + assert db_crypto.is_encryption_enabled() is False + + def test_stub_key_disables_encryption(self, monkeypatch): + """The stub fallback secret doesn't count as a real key.""" + from cyclone.secrets import STUB_SECRET + monkeypatch.setattr(db_crypto, "get_secret", lambda account: STUB_SECRET) + assert db_crypto.is_encryption_enabled() is False + + def test_real_key_enables_encryption(self, monkeypatch): + monkeypatch.setattr(db_crypto, "get_secret", lambda account: "real-key-from-keychain") + assert db_crypto.is_encryption_enabled() is True + + +class TestGetDbKey: + def test_returns_none_when_no_key(self, monkeypatch): + monkeypatch.setattr(db_crypto, "get_secret", lambda account: None) + assert db_crypto.get_db_key() is None + + def test_returns_key_from_keychain(self, monkeypatch): + monkeypatch.setattr(db_crypto, "get_secret", lambda account: "abc123") + assert db_crypto.get_db_key() == "abc123" + + def test_stub_secret_returns_none(self, monkeypatch): + from cyclone.secrets import STUB_SECRET + monkeypatch.setattr(db_crypto, "get_secret", lambda account: STUB_SECRET) + assert db_crypto.get_db_key() is None + + +# --------------------------------------------------------------------------- # +# Engine integration +# --------------------------------------------------------------------------- # + + +@pytestmark_sqlcipher +class TestEngineIntegration: + def test_engine_uses_sqlcipher_when_key_present(self, monkeypatch, tmp_path: Path): + """With a key, _make_engine installs a sqlcipher3 creator.""" + monkeypatch.setattr(db_crypto, "get_secret", lambda account: "test-key-xyz") + db_file = tmp_path / "encrypted.db" + url = f"sqlite:///{db_file}" + engine = db._make_engine(url) + # Use the engine to write and read back. + with engine.begin() as conn: + conn.execute(sa.text("CREATE TABLE t (x INTEGER)")) + conn.execute(sa.text("INSERT INTO t VALUES (1)")) + with engine.connect() as conn: + assert conn.execute(sa.text("SELECT x FROM t")).scalar() == 1 + engine.dispose() + + def test_engine_uses_plain_sqlite_without_key(self, monkeypatch, tmp_path: Path): + """Without a key, _make_engine uses the default sqlite3 driver.""" + monkeypatch.setattr(db_crypto, "get_secret", lambda account: None) + db_file = tmp_path / "plain.db" + url = f"sqlite:///{db_file}" + engine = db._make_engine(url) + with engine.begin() as conn: + conn.execute(sa.text("CREATE TABLE t (x INTEGER)")) + engine.dispose() + # The file is a valid plain SQLite DB. + with sqlite3.connect(str(db_file)) as conn: + assert conn.execute("SELECT count(*) FROM sqlite_master").fetchone()[0] >= 1 + + def test_encrypted_file_unreadable_without_key(self, monkeypatch, tmp_path: Path): + """An encrypted file is unreadable as plain SQLite.""" + # Create an encrypted DB. + monkeypatch.setattr(db_crypto, "get_secret", lambda account: "secret-1") + db_file = tmp_path / "encrypted.db" + url = f"sqlite:///{db_file}" + engine = db._make_engine(url) + with engine.begin() as conn: + conn.execute(sa.text("CREATE TABLE t (secret TEXT)")) + conn.execute(sa.text("INSERT INTO t VALUES ('classified')")) + engine.dispose() + + # Plain sqlite3 cannot open it. + with pytest.raises((sqlite3.DatabaseError, Exception)) as exc_info: + sqlite3.connect(str(db_file)).execute("SELECT * FROM t").fetchall() + # The error message comes from SQLite/SQLCipher, not Python. + assert "not a database" in str(exc_info.value).lower() or "file is encrypted" in str(exc_info.value).lower() + + def test_wrong_key_raises_on_query(self, monkeypatch, tmp_path: Path): + """A wrong key on the same encrypted file raises on the first query.""" + # Create with key A. + monkeypatch.setattr(db_crypto, "get_secret", lambda account: "key-A") + db_file = tmp_path / "encrypted.db" + url = f"sqlite:///{db_file}" + engine = db._make_engine(url) + with engine.begin() as conn: + conn.execute(sa.text("CREATE TABLE t (x INTEGER)")) + engine.dispose() + + # Try to open with key B. + monkeypatch.setattr(db_crypto, "get_secret", lambda account: "key-B") + engine2 = db._make_engine(url) + with pytest.raises(Exception) as exc_info: + with engine2.connect() as conn: + conn.execute(sa.text("SELECT * FROM t")).fetchall() + # SQLCipher raises "file is not a database" or similar on bad key. + msg = str(exc_info.value).lower() + assert "not a database" in msg or "file is encrypted" in msg or "databaseerror" in msg + engine2.dispose() + + +# --------------------------------------------------------------------------- # +# Make_creator function +# --------------------------------------------------------------------------- # + + +@pytestmark_sqlcipher +class TestMakeSqlcipherConnectCreator: + def test_creator_returns_connection_with_key_applied(self, tmp_path: Path): + """The creator's connection must have PRAGMA key applied.""" + db_file = tmp_path / "x.db" + url = f"sqlite:///{db_file}" + creator = db_crypto.make_sqlcipher_connect_creator(url, "my-key") + conn = creator() + # The creator must have applied the key — verify by writing + # data and reading it back via the same connection. + conn.execute("CREATE TABLE t (x INTEGER)") + conn.execute("INSERT INTO t VALUES (42)") + conn.commit() + result = conn.execute("SELECT x FROM t").fetchone() + assert result[0] == 42 + conn.close() diff --git a/docs/reference/co-medicaid.md b/docs/reference/co-medicaid.md index 0ef8eb1..009008a 100644 --- a/docs/reference/co-medicaid.md +++ b/docs/reference/co-medicaid.md @@ -147,6 +147,50 @@ All CMS POS codes `01`–`99` are accepted. The canonical list lives in `cyclone/parsers/payer.py` as `CMS_PLACE_OF_SERVICE_CODES` and is the source of truth for validation and any UI dropdowns. +## Database encryption at rest (SP12) + +Cyclone optionally encrypts the SQLite database with SQLCipher +(AES-256). The encryption key is stored in macOS Keychain — never on +disk in plaintext. Without the key, the DB falls back to plain SQLite. + +### One-time operator setup + +```bash +# 1. Install SQLCipher (C library) and the Python binding. +brew install sqlcipher +pip install -e backend[sqlcipher] + +# 2. Generate a random 32-byte key and store it in Keychain. +python3 -c "import secrets; print(secrets.token_urlsafe(32))" \ + | xargs -I {} security add-generic-password \ + -s cyclone -a cyclone.db.key -w "{}" + +# 3. Restart Cyclone. The DB is now encrypted. +``` + +### Verification + +```bash +# The DB file should be unreadable as plain SQLite. +sqlite3 ~/.local/share/cyclone/cyclone.db "SELECT count(*) FROM sqlite_master" +# → file is not a database + +# But readable through Cyclone. +curl http://localhost:8000/api/claims +# → 200 OK +``` + +### Key rotation (future) + +To rotate the key: + +1. Decrypt with old key, dump to SQL +2. Re-encrypt with new key, import SQL +3. Update Keychain + +A first-class rotation endpoint is out of scope for SP12 (planned for +SP14+). + ## Audit log (SP11) Cyclone persists every state-changing event to a tamper-evident