"""Audit - security audit logging and evidence tracking. Audit tracks security-relevant events across the system: - user, agent, task, model, tool, action - policy decision, approval, result, error - resource usage, latency, network action - memory write, security alert **What "immutable" means here, precisely.** Records are hash-chained: each one carries the digest of its predecessor, so an in-place edit, a deletion, a reordering, or a truncation of the beginning is detected and localised by :func:`~cyberbrain26.security_plane.audit.chain.verify_chain`. The chain does **not** make the log impossible to alter. The digest is plain SHA-256 with no secret, so an adversary holding write access to the database can rewrite the log and recompute every link, presenting a chain that verifies perfectly; deleting the end of the chain has the same effect. That is a residual risk tracked in the plan's F6 section, not an oversight. This is also why :meth:`AuditLog.clear` and :meth:`AuditLog.prune_expired` exist and why they are not silent: they are the operations a chain alone cannot make safe, so they mark what they did instead of pretending they did nothing. """ from __future__ import annotations import json import time import uuid from dataclasses import dataclass, field from enum import Enum from typing import Any from cyberbrain26.data_plane.sqlite_store import SQLiteDatabase from cyberbrain26.security_plane.audit.chain import ( GENESIS, TamperClass, VerificationReport, compute_event_hash, verify_chain, ) class AuditEventType(str, Enum): """Types of audit events.""" USER_ACTION = "user_action" AGENT_STARTED = "agent_started" AGENT_FINISHED = "agent_finished" TASK_CREATED = "task_created" TASK_STARTED = "task_started" TASK_FINISHED = "task_finished" TOOL_CALLED = "tool_called" TOOL_RESULT = "tool_result" MODEL_DISCOVERED = "model_discovered" MODEL_VALIDATED = "model_validated" MODEL_FAILED = "model_failed" POLICY_DECISION = "policy_decision" APPROVAL_REQUIRED = "approval_required" APPROVAL_GRANTED = "approval_granted" APPROVAL_DENIED = "approval_denied" SECURITY_ALERT = "security_alert" RESOURCE_LIMIT = "resource_limit" MEMORY_UPDATED = "memory_updated" IDENTITY_CHECK = "identity_check" PERMISSION_CHECK = "permission_check" FIREWALL_CHECK = "firewall_check" DATA_ACCESS = "data_access" CONFIG_CHANGE = "config_change" ERROR = "error" class AuditEventStatus(str, Enum): """Status of an audit event.""" RECORDED = "recorded" CORRELATED = "correlated" ARCHIVED = "archived" @dataclass class AuditEvent: """A single audit event.""" event_id: str event_type: AuditEventType timestamp: float actor: str = "" agent_id: str = "" task_id: str = "" model: str = "" tool: str = "" action: str = "" resource: str = "" policy_decision: str = "" approval_id: str = "" result: str = "" error: str = "" resource_usage: dict[str, Any] = field(default_factory=dict) latency_ms: float = 0.0 network_action: str = "" memory_write: str = "" security_alert: str = "" metadata: dict[str, Any] = field(default_factory=dict) status: AuditEventStatus = AuditEventStatus.RECORDED @dataclass class AuditQuery: """Query parameters for filtering audit events.""" event_type: AuditEventType | None = None actor: str | None = None agent_id: str | None = None task_id: str | None = None action: str | None = None resource: str | None = None since: float | None = None until: float | None = None limit: int = 100 class AuditLog: """Append-only, hash-chained audit log for security events. Events are written to `audit_events` and kept in an in-memory list. Each record commits to its predecessor's digest via ``seq``, ``prev_hash`` and ``entry_hash`` (computed with ``compute_event_hash`` from ``cyberbrain26.security_plane.audit.chain``), so corruption, deletion, reordering and prefix truncation are all detectable and localisable. The chain is tamper-evident, not tamper-proof: an attacker with write access can rewrite the log and recompute any digest without the HMAC key (``CYBERBRAIN_CHAIN_KEY``). `clear()` and `prune_expired()` both DELETE rows; the docstring on each method states this explicitly rather than pretending the log is immutable. Usage: audit = AuditLog() audit.record(AuditEventType.SECURITY_ALERT, agent_id="agent-1", security_alert="suspicious activity detected") events = audit.query(agent_id="agent-1") report = audit.verify_chain() if not report.intact: ... **Single writer.** The head digest is read and the new link appended within one transaction, so concurrent writers cannot fork the chain. Two ``AuditLog`` instances writing to the same database in the same process would not be safe; that is a precondition, not something this class enforces. """ def __init__(self, database: SQLiteDatabase | None = None) -> None: self.database = database self._events: list[AuditEvent] = [] if self.database is not None: self._events = self._load_persisted() def record( self, event_type: AuditEventType, actor: str = "", agent_id: str = "", task_id: str = "", model: str = "", tool: str = "", action: str = "", resource: str = "", policy_decision: str = "", approval_id: str = "", result: str = "", error: str = "", resource_usage: dict[str, Any] | None = None, latency_ms: float = 0.0, network_action: str = "", memory_write: str = "", security_alert: str = "", metadata: dict[str, Any] | None = None, ) -> AuditEvent: """Record a new audit event, linked to the current head of the chain. The head digest is read and the new link appended inside one transaction, so two writers cannot both claim the same predecessor. Every field that goes into the digest is the same value that is persisted, which is what lets a verifier recompute the chain from the stored row alone. """ event = AuditEvent( event_id=f"audit_{uuid.uuid4().hex[:16]}", event_type=event_type, timestamp=time.time(), actor=actor, agent_id=agent_id, task_id=task_id, model=model, tool=tool, action=action, resource=resource, policy_decision=policy_decision, approval_id=approval_id, result=result, error=error, resource_usage=resource_usage or {}, latency_ms=latency_ms, network_action=network_action, memory_write=memory_write, security_alert=security_alert, metadata=metadata or {}, ) if self.database is None: # In-memory log only: no persistence, therefore no chain. Recorded # honestly rather than chained against nothing. self._events.append(event) return event row = self._chain_fields(event) with self.database.transaction() as connection: head = connection.execute( "SELECT seq, entry_hash FROM audit_events " "WHERE entry_hash IS NOT NULL ORDER BY seq DESC LIMIT 1" ).fetchone() if head is None: previous_seq, prev_hash = None, GENESIS next_seq = 1 else: previous_seq = head["seq"] prev_hash = head["entry_hash"] next_seq = int(previous_seq) + 1 row["seq"] = next_seq row["prev_hash"] = prev_hash row["entry_hash"] = compute_event_hash(row, prev_hash) connection.execute( """ INSERT INTO audit_events( event_id, event_type, timestamp, actor, agent_id, task_id, model, tool, action, resource, policy_decision, approval_id, result, error, resource_usage_json, latency_ms, network_action, memory_write, security_alert, metadata_json, status, seq, prev_hash, entry_hash ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( event.event_id, event.event_type.value, event.timestamp, event.actor, event.agent_id, event.task_id, event.model, event.tool, event.action, event.resource, event.policy_decision, event.approval_id, event.result, event.error, json.dumps(event.resource_usage, separators=(",", ":")), event.latency_ms, event.network_action, event.memory_write, event.security_alert, json.dumps(event.metadata, separators=(",", ":")), event.status.value, row["seq"], row["prev_hash"], row["entry_hash"], ), ) self._events.append(event) return event @staticmethod def _chain_fields(event: AuditEvent) -> dict[str, Any]: """The record as the chain sees it: every column, JSON columns flattened. The digest must be computable from the stored row alone, so nested structures are encoded exactly as ``_load_persisted`` will decode them. """ return { "event_id": event.event_id, "event_type": event.event_type.value, "timestamp": event.timestamp, "actor": event.actor, "agent_id": event.agent_id, "task_id": event.task_id, "model": event.model, "tool": event.tool, "action": event.action, "resource": event.resource, "policy_decision": event.policy_decision, "approval_id": event.approval_id, "result": event.result, "error": event.error, "resource_usage_json": json.dumps( event.resource_usage, separators=(",", ":") ), "latency_ms": event.latency_ms, "network_action": event.network_action, "memory_write": event.memory_write, "security_alert": event.security_alert, "metadata_json": json.dumps(event.metadata, separators=(",", ":")), "status": event.status.value, } def get(self, event_id: str) -> AuditEvent | None: """Retrieve an event by its ID.""" self._refresh_persisted() for event in self._events: if event.event_id == event_id: return event return None def query(self, params: AuditQuery | None = None) -> list[AuditEvent]: """Query events with optional filters. Returns events matching all specified criteria, most recent first, limited by params.limit. """ self._refresh_persisted() params = params or AuditQuery() results: list[AuditEvent] = [] for event in self._events: if params.event_type is not None and event.event_type != params.event_type: continue if params.actor is not None and event.actor != params.actor: continue if params.agent_id is not None and event.agent_id != params.agent_id: continue if params.task_id is not None and event.task_id != params.task_id: continue if params.action is not None and event.action != params.action: continue if params.resource is not None and event.resource != params.resource: continue if params.since is not None and event.timestamp < params.since: continue if params.until is not None and event.timestamp > params.until: continue results.append(event) results.sort(key=lambda e: e.timestamp, reverse=True) return results[: params.limit] def count(self) -> int: """Total number of recorded events.""" self._refresh_persisted() return len(self._events) def count_by_type(self, event_type: AuditEventType) -> int: """Count events of a specific type.""" self._refresh_persisted() return sum(1 for e in self._events if e.event_type == event_type) def list_all(self, limit: int = 100) -> list[AuditEvent]: """List all events, most recent first.""" self._refresh_persisted() events = sorted(self._events, key=lambda e: e.timestamp, reverse=True) return events[:limit] def clear(self) -> None: """Delete every event. NOT a chain operation - it severs the chain. This exists for tests and fixtures. Calling it on a live log destroys the evidence a chain exists to protect, and no chain can report that, because there is nothing left to verify. That is the honest limitation, and it is why this method is not silent about what it does. """ self._events.clear() if self.database is not None: with self.database.transaction() as connection: connection.execute("DELETE FROM audit_events") def chain_rows(self) -> list[dict[str, Any]]: """Return persisted records in chain order, shaped as the chain sees them. Ordered by ``seq`` when a chain exists, and by timestamp otherwise, so a pre-chain (epoch 0) log still verifies as a declared boundary instead of as a broken link. Two literal queries rather than one interpolated ``ORDER BY``: an f-string in SQL is not injectable here (the clause is a fixed literal), but it reads as if it were, and the next person to "fix" that reading would introduce the hole. Interpolation is avoided for that reason alone. """ if self.database is None: return [] with self.database.connection() as connection: has_seq = connection.execute( "SELECT COUNT(*) AS n FROM audit_events WHERE seq IS NOT NULL" ).fetchone()["n"] if has_seq: rows = connection.execute( "SELECT * FROM audit_events ORDER BY seq ASC" ).fetchall() else: rows = connection.execute( "SELECT * FROM audit_events ORDER BY timestamp ASC, rowid ASC" ).fetchall() shaped: list[dict[str, Any]] = [] for row in rows: shaped.append( { "event_id": row["event_id"], "event_type": row["event_type"], "timestamp": row["timestamp"], "actor": row["actor"], "agent_id": row["agent_id"], "task_id": row["task_id"], "model": row["model"], "tool": row["tool"], "action": row["action"], "resource": row["resource"], "policy_decision": row["policy_decision"], "approval_id": row["approval_id"], "result": row["result"], "error": row["error"], "resource_usage_json": row["resource_usage_json"], "latency_ms": row["latency_ms"], "network_action": row["network_action"], "memory_write": row["memory_write"], "security_alert": row["security_alert"], "metadata_json": row["metadata_json"], "status": row["status"], "seq": row["seq"], "prev_hash": row["prev_hash"], "entry_hash": row["entry_hash"], } ) return shaped def verify_chain(self) -> VerificationReport: """Verify the persisted chain and report what, if anything, is wrong. Reports ``linkage_ok`` and ``integrity_ok`` separately - collapsing them hides which attack occurred. A log written before the chain existed returns ``unverified_premise=True`` and ``intact=False``: "never chained" is a declared boundary, never a clean pass. """ return verify_chain(self.chain_rows()) def prune_expired(self, *, now: float | None = None) -> int: """Apply retention, marking the range if it was purged over an open break. Retention is a deadline, so it is not refused when the chain is already broken. But it is never bridged silently, because a silent bridge turns a loud finding into a green verdict. When the chain is intact the purge is clean. When it is broken, this records a chained event naming the purged range and the reason, so the fact outlives the rows it describes. """ if self.database is None: return 0 if now is None: now = time.time() cutoff = now - self.database.retention_days * 24 * 60 * 60 rows_going = self.database.count_audit_before(cutoff) if rows_going == 0: return 0 # D2: know whether the chain is broken BEFORE removing the evidence. before = self.verify_chain() deleted = self.database.prune_expired(now=now)["audit_events"] # D4: the record is the chain event. It is linked like any other record, # so it cannot be removed without breaking the chain that names it. self.record( AuditEventType.SECURITY_ALERT, actor="audit.retention", action="chain_range_purged", result="purged" if before.intact else "purged_over_unexplained_break", security_alert=( f"retention removed {deleted} audit record(s) older than {cutoff:.0f}; " f"chain_intact_before_purge={before.intact}; " f"first_broken_seq={before.first_broken_seq}; " f"tamper_class={before.tamper_class.value}" ), metadata={ "cutoff": cutoff, "deleted": deleted, "chain_intact_before_purge": before.intact, "first_broken_seq": before.first_broken_seq, "tamper_class": before.tamper_class.value, }, ) self._events = self._load_persisted() return deleted def _load_persisted(self) -> list[AuditEvent]: assert self.database is not None with self.database.connection() as connection: rows = connection.execute( "SELECT * FROM audit_events ORDER BY timestamp ASC" ).fetchall() return [self._row_to_event(row) for row in rows] def _refresh_persisted(self) -> None: if self.database is not None: self._events = self._load_persisted() @staticmethod def _row_to_event(row: Any) -> AuditEvent: return AuditEvent( event_id=row["event_id"], event_type=AuditEventType(row["event_type"]), timestamp=row["timestamp"], actor=row["actor"], agent_id=row["agent_id"], task_id=row["task_id"], model=row["model"], tool=row["tool"], action=row["action"], resource=row["resource"], policy_decision=row["policy_decision"], approval_id=row["approval_id"], result=row["result"], error=row["error"], resource_usage=json.loads(row["resource_usage_json"]), latency_ms=row["latency_ms"], network_action=row["network_action"], memory_write=row["memory_write"], security_alert=row["security_alert"], metadata=json.loads(row["metadata_json"]), status=AuditEventStatus(row["status"]), ) __all__ = [ "GENESIS", "AuditEvent", "AuditEventStatus", "AuditEventType", "AuditLog", "AuditQuery", "TamperClass", "VerificationReport", ]