cyclone/backend/tests/test_api_rotate_key.py

"""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.
        # SP36 Task 3: this endpoint moved from cyclone.api to
        # cyclone.api_routers.admin; patch the live import surface.
        monkeypatch.setattr("cyclone.api_routers.admin._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_routers.admin._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.
        # SP36 Task 3: endpoint moved to cyclone.api_routers.admin.
        monkeypatch.setattr("cyclone.api_routers.admin._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_routers.admin._secrets.set_secret", lambda n, v: True,
        )
        # SP36 Task 3: lock moved with the endpoint into admin router.
        from cyclone.api_routers import admin as admin_mod
        admin_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:
            admin_mod._db_rotate_lock.release()