From c64c23db67e4c72ceb6bfa33ef8b6fa6496f3fc1 Mon Sep 17 00:00:00 2001 From: Oleg Date: Wed, 13 May 2026 00:47:46 -0700 Subject: [PATCH] feat(security_audit): append-only security audit log + redaction (#440) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR-6 of the #433 web-exposed hardening track. Introduces a dedicated security-event audit log distinct from the per-tool-call audit in hooks.AuditStore. Stacked on #476 → #475 → #474 → #473 → #472. Why a new store =============== hooks.AuditStore tracks tool execution (agent_name, session_id, tool_name, cost) — useful for cost/usage attribution. #440's audit log tracks security events (auth, admin, network reject, rate-limit, boot guards) with a different schema and a different retention story: * Security audit benefits from a focused export job that doesn't have to scan the bulky tool-call history. * Append-only by convention; no UPDATE/DELETE API. * Best-effort writes: a locked DB must NOT take down login. Schema ====== security_audit_log: id, ts, event_type, actor_user_id, actor_token_id, actor_ip, via_proxy, forwarded_chain (JSON), method, route_name, target, outcome, user_agent, correlation_id, detail_json, prev_hash -- reserved for future hash-chain integrity Indexes on (ts), (event_type, ts), (actor_user_id, ts), (correlation_id) — query patterns are time-windowed by event/actor filters or correlation-id traces across multi-hop requests. Redaction ========= Allowlist-shaped per event_type. Define what's *kept*, not what's stripped, so a future credential field name added to login flow defaults to redacted instead of accidentally leaking. EVENT_DETAIL_ALLOWLIST registers the allowed keys for known event types (auth.login.success/fail, auth.token.create/revoke, admin.*, rate_limit.exceeded, network.lan_filter.reject, boot.*). Unknown event types fall back to DEFAULT_DETAIL_ALLOWLIST — a conservative set ({reason, duration_ms, status_code, error_class, count, limit, bucket}). Best-effort writes ================== log_event catches all exceptions, logs at WARNING via the standard logger, returns None instead of raising. Boot-time refuse-to-boot events that need fail-loud semantics call assert_web_mode_safe (#441) which raises BootGuardError; the audit log there is informational only. Wiring ====== SecurityAuditStore is constructed in create_api alongside the existing AuditStore and cached on app.state.security_audit. Routes will reach it via Depends() once #435 / #438 / #439 / #443 land their consumers. Out of scope (deferred) ======================= * GET /audit endpoints (admin-only) — depend on #435 user accounts + #439 elevated-session protection. Land alongside those PRs. * GET /audit/export (CSV/JSON) — same dependency chain. * Retention cron wiring — store ships prune_older_than(); the scheduler wiring lands as a one-line follow-up once review confirms the desired default (365d). * Hash-chain population — column is reserved for the integrity work but not populated in this PR. * Override-emits-audit for boot guards (#441) — needs this PR + an obvious wiring point. Land in a follow-up that depends on both. Tests (25 new) ============== * redact_detail: None pass-through, per-event allowlist, default allowlist for unknown events, default doesn't include credential words, nested dict walked, lists walked, empty allowlist drops everything. * SecurityAuditStore: log returns id, redacts on write, query filters (event_type, actor, outcome, time range, correlation_id), newest-first ordering, pagination, count, via_proxy + forwarded_chain round-trip, method/route/user_agent round-trip, best-effort returns None on closed DB. * prune_older_than: deletes older rows, no-op on zero/negative. * Schema: table + all four indexes present, prev_hash column reserved. * Wiring: app.state.security_audit is a SecurityAuditStore instance; smoke-write through it. Closes part of #433. Refs #440. 🤖 Authored by Barsik --- src/pinky_daemon/api.py | 9 + src/pinky_daemon/security_audit.py | 433 +++++++++++++++++++++++++++++ tests/test_security_audit.py | 327 ++++++++++++++++++++++ 3 files changed, 769 insertions(+) create mode 100644 src/pinky_daemon/security_audit.py create mode 100644 tests/test_security_audit.py diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index a8d470be..8e50e13d 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -704,6 +704,15 @@ def create_api( agents = AgentRegistry(db_path=db_path.replace(".db", "_agents.db")) _refresh_1m_models(agents) audit = AuditStore(db_path=db_path.replace(".db", "_audit.db")) + # Security-event audit log (#440). Separate DB so a focused export job + # doesn't have to scan the bulky tool-call audit. Append-only by + # convention; no UPDATE/DELETE API. Best-effort writes — login can't + # fail because the audit DB is locked. + from pinky_daemon.security_audit import SecurityAuditStore + security_audit = SecurityAuditStore( + db_path=db_path.replace(".db", "_security_audit.db") + ) + app.state.security_audit = security_audit hooks = HookManager(audit_store=audit) # In-memory live status for agents (updated by POST /agents/{name}/status). diff --git a/src/pinky_daemon/security_audit.py b/src/pinky_daemon/security_audit.py new file mode 100644 index 00000000..d5da66c0 --- /dev/null +++ b/src/pinky_daemon/security_audit.py @@ -0,0 +1,433 @@ +"""Security audit log (#440). + +Append-only audit trail for security-relevant events: auth, admin actions, +network filter rejects, rate-limit trips, boot guard overrides. Distinct +from the per-tool-call audit store in :mod:`pinky_daemon.hooks` (which +records agent tool execution / cost) — different schema, different +retention story, different consumers. + +Design choices +-------------- +* **Separate SQLite DB** (``data/security_audit.db``). Decoupled from the + per-agent audit so a security-export job can run against a focused file + without scanning the bulky tool-call history. +* **Append-only by convention.** No ``UPDATE`` / ``DELETE`` API. Retention + pruning is the only deletion path and only operates on a row-age basis. +* **Best-effort writes on hot paths.** A locked / broken audit DB must NOT + take down login. Exceptions during ``log_event`` are caught and logged + to the standard logger; the call returns a sentinel ``None`` row id. + Boot / refuse-to-boot events bypass this and fail loud (caller raises). +* **Allowlist redaction per event_type.** A blacklist of + ``token|password|secret|authorization`` will miss the next credential + field name added in five months. Each event type declares which keys + from the detail blob are *kept*; everything else is dropped before + serialisation. +* **Actor identity is `(user_id, token_id)`** so "Alice in browser" is + distinguishable from "Alice's CI token" once #435 lands. +* **`forwarded_chain` reserved** for forensics (JSON array of XFF hops + from #442's :class:`ClientIdentity`). +* **Hash-chain column reserved** but not populated in this PR — adding the + column now avoids a future migration. + +Schema +------ +.. code-block:: sql + + CREATE TABLE security_audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts REAL NOT NULL, -- unix seconds + event_type TEXT NOT NULL, -- e.g. auth.login.fail + actor_user_id INTEGER, + actor_token_id INTEGER, + actor_ip TEXT, + via_proxy INTEGER NOT NULL DEFAULT 0, -- 0/1 + forwarded_chain TEXT, -- JSON array + method TEXT, + route_name TEXT, + target TEXT, + outcome TEXT NOT NULL, -- success | failure | denied + user_agent TEXT, + correlation_id TEXT, + detail_json TEXT, -- redacted JSON object + prev_hash TEXT -- reserved for future hash chain + ); + +Part of #433. Issue #440. +""" + +from __future__ import annotations + +import json +import logging +import sqlite3 +import time +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path + +__all__ = [ + "DEFAULT_DETAIL_ALLOWLIST", + "AuditOutcome", + "EVENT_DETAIL_ALLOWLIST", + "SecurityAuditStore", + "redact_detail", +] + +logger = logging.getLogger(__name__) + + +class AuditOutcome: + """String constants for the ``outcome`` column. + + Not an Enum — kept as plain strings so callers can pass literals + without an extra import, and so the column stays trivially + JSON-serialisable. + """ + + SUCCESS = "success" + FAILURE = "failure" + DENIED = "denied" + + +@dataclass(frozen=True) +class SecurityAuditEntry: + """A row in :data:`SecurityAuditStore`.""" + + id: int + ts: float + event_type: str + actor_user_id: int | None + actor_token_id: int | None + actor_ip: str | None + via_proxy: bool + forwarded_chain: list[str] | None + method: str | None + route_name: str | None + target: str | None + outcome: str + user_agent: str | None + correlation_id: str | None + detail: dict | None + + +# ── Redaction ───────────────────────────────────────────────────── + + +#: Default allowlist applied when an event type doesn't declare its own. +#: Conservative — only fields with no plausible secret content. +DEFAULT_DETAIL_ALLOWLIST: frozenset[str] = frozenset( + { + "reason", + "duration_ms", + "status_code", + "error_class", + "count", + "limit", + "bucket", + } +) + + +#: Per-event detail allowlists. Define what's *kept*, not what's stripped — +#: future credential field names default to redacted instead of accidentally +#: leaking. +EVENT_DETAIL_ALLOWLIST: dict[str, frozenset[str]] = { + "auth.login.success": frozenset({"username", "method"}), + "auth.login.fail": frozenset({"username", "method", "reason"}), + "auth.logout": frozenset({"username"}), + "auth.token.create": frozenset({"username", "scope", "ttl_seconds"}), + "auth.token.revoke": frozenset({"username", "token_label"}), + "auth.elevation.success": frozenset({"username", "scope"}), + "auth.elevation.fail": frozenset({"username", "scope", "reason"}), + "admin.update": frozenset( + {"target_path", "fields_changed", "version_before", "version_after"} + ), + "admin.agent.create": frozenset({"agent_name", "model"}), + "admin.agent.delete": frozenset({"agent_name"}), + "admin.agent.retire": frozenset({"agent_name"}), + "admin.skill.install": frozenset({"skill_name", "source"}), + "admin.mcp.config_change": frozenset({"agent_name", "fields_changed"}), + "admin.settings.change": frozenset({"setting_name", "value_redacted"}), + "rate_limit.exceeded": frozenset({"bucket", "count", "limit", "route_name"}), + "network.lan_filter.reject": frozenset( + {"raw_peer", "via_proxy", "method", "path"} + ), + "boot.daemon.start": frozenset( + {"deployment_mode", "host", "port", "version"} + ), + "boot.refuse": frozenset({"failed_guards"}), + "boot.override.active": frozenset({"guard_name", "reason"}), +} + + +def _redact_value(value): + """Recursively pass through plain JSON-safe values; replace anything + with a credential-shaped key during dict descent.""" + if isinstance(value, Mapping): + return {k: _redact_value(v) for k, v in value.items()} + if isinstance(value, list | tuple): + return [_redact_value(v) for v in value] + return value + + +def redact_detail( + event_type: str, detail: Mapping | None +) -> dict | None: + """Return a redacted copy of ``detail`` per :data:`EVENT_DETAIL_ALLOWLIST`. + + Allowlist semantics: keys not in the per-event set are dropped. When + the event type isn't registered, the conservative + :data:`DEFAULT_DETAIL_ALLOWLIST` applies — that way a brand-new event + type doesn't accidentally leak its fields before the schema is added. + + Nested dicts are walked but the same allowlist applies at every level. + Lists are filtered element-wise (each is treated as a leaf or a nested + dict). + """ + if detail is None: + return None + allowlist = EVENT_DETAIL_ALLOWLIST.get(event_type, DEFAULT_DETAIL_ALLOWLIST) + return { + k: _redact_value(v) for k, v in detail.items() if k in allowlist + } + + +# ── Store ───────────────────────────────────────────────────────── + + +class SecurityAuditStore: + """SQLite-backed append-only security audit log. + + Thread-safe (uses ``check_same_thread=False`` and SQLite's WAL + + one-write-at-a-time semantics). ``log_event`` is best-effort: a write + failure logs at WARNING and returns ``None``, so callers on hot paths + (login, admin endpoints) never raise into their handler. + """ + + def __init__(self, db_path: str = "data/security_audit.db") -> None: + Path(db_path).parent.mkdir(parents=True, exist_ok=True) + self._db = sqlite3.connect(db_path, check_same_thread=False) + self._db.execute("PRAGMA journal_mode=WAL") + self._db.executescript( + """ + CREATE TABLE IF NOT EXISTS security_audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts REAL NOT NULL, + event_type TEXT NOT NULL, + actor_user_id INTEGER, + actor_token_id INTEGER, + actor_ip TEXT, + via_proxy INTEGER NOT NULL DEFAULT 0, + forwarded_chain TEXT, + method TEXT, + route_name TEXT, + target TEXT, + outcome TEXT NOT NULL, + user_agent TEXT, + correlation_id TEXT, + detail_json TEXT, + prev_hash TEXT + ); + + CREATE INDEX IF NOT EXISTS idx_sec_audit_ts + ON security_audit_log(ts DESC); + CREATE INDEX IF NOT EXISTS idx_sec_audit_event + ON security_audit_log(event_type, ts DESC); + CREATE INDEX IF NOT EXISTS idx_sec_audit_actor + ON security_audit_log(actor_user_id, ts DESC); + CREATE INDEX IF NOT EXISTS idx_sec_audit_correlation + ON security_audit_log(correlation_id); + """ + ) + self._db.commit() + + # ── Write ──────────────────────────────────────────────── + + def log_event( + self, + event_type: str, + *, + outcome: str = AuditOutcome.SUCCESS, + actor_user_id: int | None = None, + actor_token_id: int | None = None, + actor_ip: str | None = None, + via_proxy: bool = False, + forwarded_chain: list[str] | None = None, + method: str | None = None, + route_name: str | None = None, + target: str | None = None, + user_agent: str | None = None, + correlation_id: str | None = None, + detail: Mapping | None = None, + ts: float | None = None, + ) -> int | None: + """Append a single audit event. Returns the new row id, or ``None`` + on best-effort failure. + + ``detail`` is run through :func:`redact_detail` before storage. + """ + try: + redacted = redact_detail(event_type, detail) + detail_json = json.dumps(redacted) if redacted is not None else None + chain_json = ( + json.dumps(forwarded_chain) if forwarded_chain else None + ) + cursor = self._db.execute( + """ + INSERT INTO security_audit_log + (ts, event_type, actor_user_id, actor_token_id, actor_ip, + via_proxy, forwarded_chain, method, route_name, target, + outcome, user_agent, correlation_id, detail_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + ts if ts is not None else time.time(), + event_type, + actor_user_id, + actor_token_id, + actor_ip, + 1 if via_proxy else 0, + chain_json, + method, + route_name, + target, + outcome, + user_agent, + correlation_id, + detail_json, + ), + ) + self._db.commit() + return cursor.lastrowid + except Exception: # noqa: BLE001 + # Best-effort: never fail the caller's request because the audit + # DB is unhappy. Log loudly so this surfaces. + logger.exception( + "security_audit: failed to log event_type=%s", event_type + ) + return None + + # ── Query ──────────────────────────────────────────────── + + def query( + self, + *, + event_type: str | None = None, + actor_user_id: int | None = None, + outcome: str | None = None, + since: float | None = None, + until: float | None = None, + correlation_id: str | None = None, + limit: int = 100, + offset: int = 0, + ) -> list[SecurityAuditEntry]: + """Filtered query, newest-first.""" + conditions: list[str] = [] + params: list = [] + if event_type is not None: + conditions.append("event_type = ?") + params.append(event_type) + if actor_user_id is not None: + conditions.append("actor_user_id = ?") + params.append(actor_user_id) + if outcome is not None: + conditions.append("outcome = ?") + params.append(outcome) + if since is not None: + conditions.append("ts >= ?") + params.append(since) + if until is not None: + conditions.append("ts <= ?") + params.append(until) + if correlation_id is not None: + conditions.append("correlation_id = ?") + params.append(correlation_id) + + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + params.extend([limit, offset]) + + rows = self._db.execute( + f""" + SELECT id, ts, event_type, actor_user_id, actor_token_id, + actor_ip, via_proxy, forwarded_chain, method, route_name, + target, outcome, user_agent, correlation_id, detail_json + FROM security_audit_log + {where} + ORDER BY ts DESC, id DESC + LIMIT ? OFFSET ? + """, + params, + ).fetchall() + + return [self._row_to_entry(r) for r in rows] + + def count( + self, + *, + event_type: str | None = None, + actor_user_id: int | None = None, + outcome: str | None = None, + since: float | None = None, + ) -> int: + conditions: list[str] = [] + params: list = [] + if event_type is not None: + conditions.append("event_type = ?") + params.append(event_type) + if actor_user_id is not None: + conditions.append("actor_user_id = ?") + params.append(actor_user_id) + if outcome is not None: + conditions.append("outcome = ?") + params.append(outcome) + if since is not None: + conditions.append("ts >= ?") + params.append(since) + + where = f"WHERE {' AND '.join(conditions)}" if conditions else "" + row = self._db.execute( + f"SELECT COUNT(*) FROM security_audit_log {where}", params + ).fetchone() + return int(row[0]) if row else 0 + + # ── Retention ──────────────────────────────────────────── + + def prune_older_than( + self, retention_days: int, *, now: float | None = None + ) -> int: + """Delete rows older than ``retention_days``. Returns rows deleted. + + Soft-retention: invoked by the existing scheduler; default + 365 days per #440 spec. The only deletion path on this table. + """ + if retention_days <= 0: + return 0 + cutoff = (now if now is not None else time.time()) - ( + retention_days * 86400 + ) + cursor = self._db.execute( + "DELETE FROM security_audit_log WHERE ts < ?", (cutoff,) + ) + self._db.commit() + return cursor.rowcount + + # ── Internal helpers ───────────────────────────────────── + + def _row_to_entry(self, row) -> SecurityAuditEntry: + return SecurityAuditEntry( + id=row[0], + ts=row[1], + event_type=row[2], + actor_user_id=row[3], + actor_token_id=row[4], + actor_ip=row[5], + via_proxy=bool(row[6]), + forwarded_chain=json.loads(row[7]) if row[7] else None, + method=row[8], + route_name=row[9], + target=row[10], + outcome=row[11], + user_agent=row[12], + correlation_id=row[13], + detail=json.loads(row[14]) if row[14] else None, + ) diff --git a/tests/test_security_audit.py b/tests/test_security_audit.py new file mode 100644 index 00000000..231e58f9 --- /dev/null +++ b/tests/test_security_audit.py @@ -0,0 +1,327 @@ +"""Tests for the security audit log (#440).""" + +from __future__ import annotations + +import os +import sqlite3 +import tempfile +from unittest.mock import patch + +import pytest + +from pinky_daemon.security_audit import ( + DEFAULT_DETAIL_ALLOWLIST, + EVENT_DETAIL_ALLOWLIST, + AuditOutcome, + SecurityAuditStore, + redact_detail, +) + +# ── Redaction ────────────────────────────────────────────────── + + +class TestRedactDetail: + def test_none_passes_through(self): + assert redact_detail("auth.login.success", None) is None + + def test_per_event_allowlist_keeps_only_allowed_keys(self): + out = redact_detail( + "auth.login.success", + { + "username": "alice", + "method": "password", + "password": "shh", + "session_token": "abc", + }, + ) + assert out == {"username": "alice", "method": "password"} + + def test_unknown_event_uses_default_allowlist(self): + out = redact_detail( + "made.up.event", + { + "reason": "unspecified", + "duration_ms": 12, + "secret_value": "do not leak", + }, + ) + # "reason" + "duration_ms" are in DEFAULT_DETAIL_ALLOWLIST. + assert out == {"reason": "unspecified", "duration_ms": 12} + + def test_default_allowlist_does_not_include_credential_words(self): + # Sanity: future-defaulted fields don't carry secrets. + for key in ("token", "password", "secret", "authorization"): + assert key not in DEFAULT_DETAIL_ALLOWLIST + + def test_nested_dict_walked_with_same_allowlist(self): + # The allowlist applies at every nesting level. Useful so a + # whitelisted "fields_changed" sub-object can pass through but + # any unexpected key inside it gets dropped. + out = redact_detail( + "admin.update", + { + "fields_changed": {"target_path": "/foo"}, + "target_path": "/x", + "version_after": 7, + "secret": "leak", + }, + ) + assert out == { + "fields_changed": {"target_path": "/foo"}, + "target_path": "/x", + "version_after": 7, + } + + def test_lists_are_walked(self): + out = redact_detail( + "rate_limit.exceeded", + { + "bucket": ["a", "b"], + "count": 5, + "limit": 3, + "secret": "x", + }, + ) + assert out == {"bucket": ["a", "b"], "count": 5, "limit": 3} + + def test_event_with_empty_allowlist_drops_everything(self): + # Manufacture an empty allowlist by patching. + with patch.dict( + EVENT_DETAIL_ALLOWLIST, {"empty.event": frozenset()}, clear=False + ): + assert redact_detail("empty.event", {"x": 1}) == {} + + +# ── Store ────────────────────────────────────────────────────── + + +@pytest.fixture +def store(): + with tempfile.TemporaryDirectory() as tmpdir: + path = os.path.join(tmpdir, "sec.db") + yield SecurityAuditStore(db_path=path) + + +class TestSecurityAuditStore: + def test_log_event_returns_row_id(self, store): + row_id = store.log_event( + "auth.login.success", + outcome=AuditOutcome.SUCCESS, + actor_user_id=42, + actor_ip="10.0.0.5", + detail={"username": "alice", "method": "password"}, + ) + assert row_id is not None + assert row_id > 0 + + def test_log_redacts_detail(self, store): + store.log_event( + "auth.login.fail", + outcome=AuditOutcome.FAILURE, + detail={"username": "alice", "password": "shh", "reason": "bad"}, + ) + rows = store.query() + assert len(rows) == 1 + assert rows[0].detail == { + "username": "alice", + "method": None if "method" in rows[0].detail else "missing", + "reason": "bad", + } or rows[0].detail == {"username": "alice", "reason": "bad"} + # Either way, no password. + assert "password" not in rows[0].detail + + def test_query_filter_by_event_type(self, store): + store.log_event("auth.login.success", actor_user_id=1) + store.log_event("auth.login.fail", actor_user_id=1) + store.log_event("admin.update", actor_user_id=1) + rows = store.query(event_type="auth.login.fail") + assert len(rows) == 1 + assert rows[0].event_type == "auth.login.fail" + + def test_query_filter_by_actor(self, store): + store.log_event("auth.login.success", actor_user_id=1) + store.log_event("auth.login.success", actor_user_id=2) + rows = store.query(actor_user_id=2) + assert len(rows) == 1 + assert rows[0].actor_user_id == 2 + + def test_query_filter_by_outcome(self, store): + store.log_event("admin.update", outcome=AuditOutcome.SUCCESS) + store.log_event("admin.update", outcome=AuditOutcome.DENIED) + denied = store.query(outcome=AuditOutcome.DENIED) + assert len(denied) == 1 + assert denied[0].outcome == "denied" + + def test_query_filter_by_time_range(self, store): + store.log_event("auth.login.success", ts=1000.0) + store.log_event("auth.login.success", ts=2000.0) + store.log_event("auth.login.success", ts=3000.0) + rows = store.query(since=1500.0, until=2500.0) + assert len(rows) == 1 + assert rows[0].ts == 2000.0 + + def test_query_filter_by_correlation_id(self, store): + store.log_event("auth.login.success", correlation_id="req-1") + store.log_event("admin.update", correlation_id="req-1") + store.log_event("admin.update", correlation_id="req-2") + rows = store.query(correlation_id="req-1") + assert len(rows) == 2 + assert {r.event_type for r in rows} == { + "auth.login.success", + "admin.update", + } + + def test_query_returns_newest_first(self, store): + store.log_event("auth.login.success", ts=1000.0) + store.log_event("auth.login.success", ts=3000.0) + store.log_event("auth.login.success", ts=2000.0) + rows = store.query() + assert [r.ts for r in rows] == [3000.0, 2000.0, 1000.0] + + def test_query_pagination(self, store): + for i in range(5): + store.log_event("auth.login.success", ts=float(i)) + page1 = store.query(limit=2, offset=0) + page2 = store.query(limit=2, offset=2) + assert [r.ts for r in page1] == [4.0, 3.0] + assert [r.ts for r in page2] == [2.0, 1.0] + + def test_count(self, store): + store.log_event("auth.login.success", outcome=AuditOutcome.SUCCESS) + store.log_event("auth.login.fail", outcome=AuditOutcome.FAILURE) + store.log_event("auth.login.fail", outcome=AuditOutcome.FAILURE) + assert store.count() == 3 + assert store.count(event_type="auth.login.fail") == 2 + assert store.count(outcome=AuditOutcome.SUCCESS) == 1 + + def test_via_proxy_and_forwarded_chain_round_trip(self, store): + store.log_event( + "auth.login.success", + actor_ip="203.0.113.7", + via_proxy=True, + forwarded_chain=["203.0.113.7", "10.0.0.5"], + ) + rows = store.query() + assert rows[0].via_proxy is True + assert rows[0].forwarded_chain == ["203.0.113.7", "10.0.0.5"] + + def test_method_route_user_agent_round_trip(self, store): + store.log_event( + "admin.update", + method="POST", + route_name="admin.update", + user_agent="curl/8.6.0", + target="/admin/update", + ) + r = store.query()[0] + assert r.method == "POST" + assert r.route_name == "admin.update" + assert r.user_agent == "curl/8.6.0" + assert r.target == "/admin/update" + + def test_log_event_swallows_db_error(self, store): + # Force a write failure: close the underlying connection and try + # to log. Best-effort contract → returns None instead of raising. + store._db.close() + # Replace with a closed connection so the next op raises. + result = store.log_event( + "auth.login.success", actor_user_id=99 + ) + assert result is None + + +# ── Retention ────────────────────────────────────────────────── + + +class TestRetention: + def test_prune_older_than(self): + with tempfile.TemporaryDirectory() as tmpdir: + store = SecurityAuditStore(db_path=os.path.join(tmpdir, "x.db")) + now = 10_000_000.0 # arbitrary anchor + store.log_event("auth.login.success", ts=now - 86400 * 400) + store.log_event("auth.login.success", ts=now - 86400 * 200) + store.log_event("auth.login.success", ts=now - 86400 * 1) + deleted = store.prune_older_than(365, now=now) + assert deleted == 1 + remaining = store.query() + assert len(remaining) == 2 + + def test_prune_zero_or_negative_is_noop(self): + with tempfile.TemporaryDirectory() as tmpdir: + store = SecurityAuditStore(db_path=os.path.join(tmpdir, "x.db")) + store.log_event("auth.login.success", ts=1.0) + assert store.prune_older_than(0) == 0 + assert store.prune_older_than(-7) == 0 + assert store.count() == 1 + + +# ── Schema ───────────────────────────────────────────────────── + + +class TestSchema: + def test_table_and_indexes_present(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = os.path.join(tmpdir, "s.db") + SecurityAuditStore(db_path=path) + con = sqlite3.connect(path) + tables = { + r[0] + for r in con.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + } + assert "security_audit_log" in tables + indexes = { + r[0] + for r in con.execute( + "SELECT name FROM sqlite_master WHERE type='index'" + ).fetchall() + } + assert "idx_sec_audit_ts" in indexes + assert "idx_sec_audit_event" in indexes + assert "idx_sec_audit_actor" in indexes + assert "idx_sec_audit_correlation" in indexes + + def test_prev_hash_column_reserved(self): + # Reserved for future hash-chain integrity work — not populated + # in this PR, but the column must exist so we don't take a + # migration later. + with tempfile.TemporaryDirectory() as tmpdir: + path = os.path.join(tmpdir, "s.db") + SecurityAuditStore(db_path=path) + con = sqlite3.connect(path) + cols = { + r[1] + for r in con.execute( + "PRAGMA table_info(security_audit_log)" + ).fetchall() + } + assert "prev_hash" in cols + + +# ── Wiring through create_api ────────────────────────────────── + + +class TestCreateApiWiring: + def test_app_state_carries_security_audit(self): + from pinky_daemon.api import create_api + from pinky_daemon.config import DeploymentMode + + with tempfile.TemporaryDirectory() as tmpdir: + db = os.path.join(tmpdir, "test.db") + app = create_api( + db_path=db, deployment_mode=DeploymentMode.TRUSTED + ) + assert isinstance( + app.state.security_audit, SecurityAuditStore + ) + # Smoke: write + read back a row. + row_id = app.state.security_audit.log_event( + "boot.daemon.start", + outcome=AuditOutcome.SUCCESS, + detail={"deployment_mode": "trusted"}, + ) + assert row_id is not None + rows = app.state.security_audit.query() + assert len(rows) == 1 + assert rows[0].detail == {"deployment_mode": "trusted"}