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", principal_id=42, 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",
        "principal_id",
        "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(principal_id=42, **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(principal_id=42, limit=10).events[0]
    assert (event.kind, event.method, event.device_label, event.target) == tuple(
        canaries.values()
    )


def test_legacy_security_events_migrate_encrypted_but_remain_unattributed(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)
    assert store.list(principal_id=42, limit=10).events == []
    store.finalize("operation-9")
    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", principal_id=42, target="private/repo#42")

    with pytest.raises(SecurityEventStoreError, match="temporarily unavailable"):
        SecurityEventStore(
            path, clock=lambda: 1_001, encryption_key=b"x" * 32
        ).list(principal_id=42, 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(principal_id=42, 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",
        principal_id=42,
        method="token",
        device_label="  Timmy   Phone  " + "x" * 80,
        target="dashboard",
    )
    now[0] += 1
    store.record("device_revoked", principal_id=42, device_label="Old phone", target="device")

    page = store.list(principal_id=42, 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(principal_id=42, 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", "principal_id", "created_at", "status", "operation_id",
    }


def test_security_events_are_isolated_by_principal_across_cursor_pages(tmp_path):
    now = [1_000]
    store = SecurityEventStore(
        tmp_path / "security.sqlite3", clock=lambda: now[0], encryption_key=PRIVATE_KEY
    )
    store.record("sign_in", principal_id=42, device_label="Timmy phone")
    now[0] += 1
    store.record("sign_in", principal_id=84, device_label="Other phone")
    now[0] += 1
    store.record("device_revoked", principal_id=42, device_label="Old Timmy phone")

    first = store.list(principal_id=42, limit=1)
    second = store.list(principal_id=42, limit=10, cursor=first.next_cursor)

    assert [event.device_label for event in first.events] == ["Old Timmy phone"]
    assert [event.device_label for event in second.events] == ["Timmy phone"]
    assert all(event.device_label != "Other phone" for event in first.events + second.events)


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", principal_id=42, method="token", device_label=f"Device {index}")

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

    now[0] = 20
    store.record("sign_out", principal_id=42, device_label="Current")
    assert [event.kind for event in store.list(principal_id=42, 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", principal_id=42, target="stackchain/api#7")

    pending = SecurityEventStore(
        tmp_path / "security.sqlite3", clock=lambda: 1_001
    ).list(principal_id=42, 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(principal_id=42, 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", principal_id=42, target="stackchain/api#7")

    store.discard(operation_id)

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