stackchain-dashboard/tests/test_security_event_store.py

import sqlite3

import pytest

from src.security_event_store import SecurityEventStore, SecurityEventStoreError


PRIVATE_KEY = b"e" * 32


def test_security_event_schema_exposes_no_plaintext_metadata_columns(tmp_path):
    store = SecurityEventStore(
        tmp_path / "security.sqlite3", clock=lambda: 1_000, encryption_key=PRIVATE_KEY
    )
    store.record("issue_closed", method="passkey", target="private/repo#42")

    with sqlite3.connect(store.path) as connection:
        columns = {
            row[1]
            for row in connection.execute("PRAGMA table_info(security_events)")
        }
    assert columns == {
        "id",
        "payload",
        "created_at",
        "status",
        "operation_id",
    }


def test_security_event_payload_is_encrypted_at_rest_and_survives_restart(tmp_path):
    path = tmp_path / "security.sqlite3"
    canaries = {
        "kind": "private_issue_closed_canary",
        "method": "private_passkey_canary",
        "device_label": "Private Operator Phone Canary",
        "target": "private-org/roadmap#4242",
    }
    store = SecurityEventStore(path, clock=lambda: 1_000, encryption_key=PRIVATE_KEY)

    store.record(**canaries)

    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT payload FROM security_events"
        ).fetchone()[0]
    assert payload.startswith("v1:")
    database_bytes = path.read_bytes()
    assert all(value.encode() not in database_bytes for value in canaries.values())
    reopened = SecurityEventStore(path, clock=lambda: 1_001, encryption_key=PRIVATE_KEY)
    event = reopened.list(limit=10).events[0]
    assert (event.kind, event.method, event.device_label, event.target) == tuple(
        canaries.values()
    )


def test_legacy_security_events_migrate_without_changing_journal_semantics(tmp_path):
    path = tmp_path / "security.sqlite3"
    with sqlite3.connect(path) as connection:
        connection.executescript(
            """
            CREATE TABLE security_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                kind TEXT NOT NULL,
                method TEXT,
                device_label TEXT,
                target TEXT,
                created_at INTEGER NOT NULL,
                status TEXT NOT NULL DEFAULT 'completed',
                operation_id TEXT
            );
            INSERT INTO security_events
                (id, kind, method, device_label, target, created_at, status, operation_id)
            VALUES
                (7, 'legacy_sign_in_canary', 'legacy_token_canary',
                 'Legacy Phone Canary', 'dashboard', 900, 'completed', NULL),
                (9, 'legacy_issue_closed_canary', 'passkey',
                 'Legacy Tablet Canary', 'private/repo#9', 901, 'pending', 'operation-9');
            """
        )

    store = SecurityEventStore(path, clock=lambda: 1_000, encryption_key=PRIVATE_KEY)
    page = store.list(limit=1)
    older = store.list(limit=10, cursor=page.next_cursor)

    assert [(event.id, event.kind, event.created_at, event.status) for event in page.events] == [
        (9, "legacy_issue_closed_canary", 901, "pending")
    ]
    assert [(event.id, event.kind, event.created_at, event.status) for event in older.events] == [
        (7, "legacy_sign_in_canary", 900, "completed")
    ]
    store.finalize("operation-9")
    assert store.list(limit=1).events[0].status == "completed"
    with sqlite3.connect(path) as connection:
        rows = connection.execute(
            "SELECT id, payload FROM security_events ORDER BY id"
        ).fetchall()
    assert [row[0] for row in rows] == [7, 9]
    assert all(row[1].startswith("v1:") for row in rows)
    database_bytes = path.read_bytes()
    assert b"legacy_sign_in_canary" not in database_bytes
    assert b"Legacy Tablet Canary" not in database_bytes


def test_wrong_key_or_tampered_security_activity_fails_closed(tmp_path):
    path = tmp_path / "security.sqlite3"
    store = SecurityEventStore(path, clock=lambda: 1_000, encryption_key=PRIVATE_KEY)
    store.record("issue_closed", target="private/repo#42")

    with pytest.raises(SecurityEventStoreError, match="temporarily unavailable"):
        SecurityEventStore(
            path, clock=lambda: 1_001, encryption_key=b"x" * 32
        ).list(limit=10)

    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT payload FROM security_events WHERE id = 1"
        ).fetchone()[0]
        replacement = "A" if payload[-1] != "A" else "B"
        connection.execute(
            "UPDATE security_events SET payload = ? WHERE id = 1",
            (payload[:-1] + replacement,),
        )
    with pytest.raises(SecurityEventStoreError, match="temporarily unavailable"):
        store.list(limit=10)


def test_missing_security_activity_encryption_key_fails_with_store_error(
    monkeypatch, tmp_path
):
    monkeypatch.delenv("STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEY", raising=False)

    with pytest.raises(SecurityEventStoreError, match="temporarily unavailable"):
        SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 1_000)


def test_security_events_are_private_bounded_and_reverse_chronological(tmp_path):
    now = [1_000]
    store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: now[0])

    store.record(
        "sign_in",
        method="token",
        device_label="  Timmy   Phone  " + "x" * 80,
        target="dashboard",
    )
    now[0] += 1
    store.record("device_revoked", device_label="Old phone", target="device")

    page = store.list(limit=1)
    assert [(event.kind, event.device_label, event.target) for event in page.events] == [
        ("device_revoked", "Old phone", "device")
    ]
    assert page.next_cursor is not None

    older = store.list(limit=10, cursor=page.next_cursor)
    assert older.events[0].kind == "sign_in"
    assert older.events[0].method == "token"
    assert older.events[0].device_label == ("Timmy Phone " + "x" * 52)

    columns = {
        row[1] for row in sqlite3.connect(store.path).execute("PRAGMA table_info(security_events)")
    }
    assert columns == {
        "id", "payload", "created_at", "status", "operation_id",
    }


def test_security_event_retention_prunes_age_and_count(tmp_path):
    now = [0]
    store = SecurityEventStore(
        tmp_path / "security.sqlite3",
        clock=lambda: now[0],
        max_events=3,
        retention_seconds=10,
    )
    for index in range(4):
        now[0] = index
        store.record("sign_in", method="token", device_label=f"Device {index}")

    assert [event.device_label for event in store.list(limit=10).events] == [
        "Device 3", "Device 2", "Device 1"
    ]

    now[0] = 20
    store.record("sign_out", device_label="Current")
    assert [event.kind for event in store.list(limit=10).events] == ["sign_out"]


def test_security_event_reservation_is_durable_until_finalized(tmp_path):
    store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 1_000)

    operation_id = store.reserve("issue_closed", target="stackchain/api#7")

    pending = SecurityEventStore(
        tmp_path / "security.sqlite3", clock=lambda: 1_001
    ).list(limit=10).events
    assert [(event.kind, event.target, event.status) for event in pending] == [
        ("issue_closed", "stackchain/api#7", "pending")
    ]

    store.finalize(operation_id)

    completed = store.list(limit=10).events
    assert [(event.kind, event.target, event.status) for event in completed] == [
        ("issue_closed", "stackchain/api#7", "completed")
    ]


def test_failed_operation_can_discard_its_pending_reservation(tmp_path):
    store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 1_000)
    operation_id = store.reserve("issue_closed", target="stackchain/api#7")

    store.discard(operation_id)

    assert store.list(limit=10).events == []