Tyler
d54c44f04a
- 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.
176 lines
7.3 KiB
Python
176 lines
7.3 KiB
Python
"""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()
|