cyclone/backend/tests/test_db_crypto.py

"""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()


# --------------------------------------------------------------------------- #
# 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()