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 == []