"""Hash-chained, append-only audit log (SOC 2 / SIEM ready). Every security-relevant operation — add, search, delete, erasure, key management, snapshot, restore — appends one record: {seq, ts, actor, role, action, resource, outcome, meta, prev_hash, hash} ``hash = BLAKE2b(prev_hash || canonical-record)`` — tampering with any record breaks the chain and ``verify()`` pinpoints the first damaged sequence number. Records are also exported as JSONL (one per line) for Splunk / Elastic / Datadog ingestion, and a syslog-style single-line format for legacy collectors. GDPR note: the audit chain is intentionally exempt from user erasure (accounting/legitimate-interest records). It stores actor + resource ids, never raw conversation text. """ from __future__ import annotations import json import threading from datetime import datetime, timezone from cortexm.security.hashes import HashProvider # actions that are always audited AUDITED_ACTIONS = { "memory.add", "memory.search", "memory.update", "memory.delete", "memory.delete_all", "memory.verify", "governance.erase", "governance.retention", "governance.snapshot", "governance.restore", "governance.pitr", "keys.create", "keys.revoke", "auth.failure", "security.quarantine", } def _now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="milliseconds") class AuditLog: def __init__(self, store, enabled: bool = True) -> None: self.store = store self.enabled = enabled self._hasher = HashProvider("blake2b") self._lock = threading.Lock() self._ensure_table() def _ensure_table(self) -> None: self.store.conn.execute(""" CREATE TABLE IF NOT EXISTS audit_log ( seq INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, actor TEXT NOT NULL, role TEXT, action TEXT NOT NULL, resource TEXT, outcome TEXT NOT NULL, meta TEXT, prev_hash TEXT NOT NULL, hash TEXT NOT NULL )""") # indexes for the common tail() filters (actor, action, ts) self.store.conn.execute( "CREATE INDEX IF NOT EXISTS idx_audit_actor_action " "ON audit_log(actor, action)") self.store.conn.execute( "CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(ts)") self.store.conn.commit() # ------------------------------------------------------------- append def log(self, action: str, *, actor: str = "system", role: str | None = None, resource: str | None = None, outcome: str = "success", meta: dict | None = None) -> dict | None: if not self.enabled: return None with self._lock: row = self.store.conn.execute( "SELECT seq, hash FROM audit_log ORDER BY seq DESC LIMIT 1" ).fetchone() prev = row["hash"] if row else "genesis" rec = { "ts": _now_iso(), "actor": actor, "role": role or "-", "action": action, "resource": resource or "-", "outcome": outcome, "meta": json.dumps(meta or {}, sort_keys=True), } canonical = json.dumps(rec, sort_keys=True, separators=(",", ":")) digest = self._hasher.hash_text(f"{prev}|{canonical}") self.store.conn.execute( "INSERT INTO audit_log(ts, actor, role, action, resource," " outcome, meta, prev_hash, hash) VALUES(?,?,?,?,?,?,?,?,?)", (rec["ts"], rec["actor"], rec["role"], rec["action"], rec["resource"], rec["outcome"], rec["meta"], prev, digest)) self.store.conn.commit() return {"seq": self.store.conn.execute( "SELECT last_insert_rowid() AS s").fetchone()["s"], **rec, "prev_hash": prev, "hash": digest} # ------------------------------------------------------------- read def tail(self, n: int = 100, actor: str | None = None, action: str | None = None) -> list[dict]: q = "SELECT * FROM audit_log" conds, args = [], [] if actor: conds.append("actor=?"); args.append(actor) if action: conds.append("action=?"); args.append(action) if conds: q += " WHERE " + " AND ".join(conds) q += " ORDER BY seq DESC LIMIT ?" args.append(int(n)) return [dict(r) for r in self.store.conn.execute(q, args)] def verify(self) -> dict: """Recompute the chain; report the first broken seq.""" prev = "genesis" broken_at = None n = 0 for row in self.store.conn.execute( "SELECT * FROM audit_log ORDER BY seq ASC"): n += 1 rec = {"ts": row["ts"], "actor": row["actor"], "role": row["role"], "action": row["action"], "resource": row["resource"], "outcome": row["outcome"], "meta": row["meta"]} canonical = json.dumps(rec, sort_keys=True, separators=(",", ":")) digest = self._hasher.hash_text(f"{prev}|{canonical}") if digest != row["hash"] or prev != row["prev_hash"]: broken_at = row["seq"] break prev = row["hash"] return {"records": n, "intact": broken_at is None, "first_broken_seq": broken_at, "head_hash": prev if broken_at is None else None} # ------------------------------------------------------------- export def export_jsonl(self, path: str) -> int: """SIEM ingestion export (Splunk/Elastic/Datadog friendly).""" count = 0 with open(path, "w", encoding="utf-8") as fh: for row in self.store.conn.execute( "SELECT * FROM audit_log ORDER BY seq ASC"): rec = {k: row[k] for k in row.keys()} try: rec["meta"] = json.loads(rec["meta"]) except Exception: pass fh.write(json.dumps(rec, sort_keys=True) + "\n") count += 1 return count def export_syslog(self, path: str) -> int: """RFC-3164-style lines: <134>ts actor action resource outcome.""" count = 0 with open(path, "w", encoding="utf-8") as fh: for row in self.store.conn.execute( "SELECT * FROM audit_log ORDER BY seq ASC"): ts = row["ts"][:19].replace("T", " ") fh.write(f"<134>{ts} context-m audit: actor={row['actor']} " f"action={row['action']} resource={row['resource']} " f"outcome={row['outcome']} seq={row['seq']}\n") count += 1 return count class AuditContext: """Request-scoped audit binding (actor/role propagated by the server).""" def __init__(self, audit: AuditLog | None, actor: str = "system", role: str | None = None) -> None: self.audit = audit self.actor = actor self.role = role def log(self, action: str, **kw) -> None: if self.audit is not None: self.audit.log(action, actor=self.actor, role=self.role, **kw)