"""Unified metadata-only audit trail (WARDEN-WP-0022). Every ops-warden action appends a JSONL event. Secret values are rejected at write time. """ from __future__ import annotations import json import os import re from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Iterable, Optional _AUDIT_FILENAME = "audit.jsonl" _MAX_BYTES = 5 * 1024 * 1024 _SECRET_PREFIXES = ( "ghp_", "gho_", "ghs_", "github_pat_", "sk-", "sk_live_", "sk_test_", "xoxb-", "xoxp-", "AKIA", "ASIA", "hvs.", "hvb.", "s.", "AIza", "eyJ", ) _HIGH_ENTROPY_RUN = re.compile(r"[A-Za-z0-9_\-]{32,}") class AuditError(Exception): """Raised when audit metadata looks like a secret value.""" def _assert_metadata_safe(blob: str) -> None: lowered = blob.lower() for prefix in _SECRET_PREFIXES: if prefix.lower() in lowered: raise AuditError( f"audit field appears to contain a literal secret (matched {prefix!r})" ) for run in _HIGH_ENTROPY_RUN.findall(blob): if "<" in run or ">" in run: continue if run.replace("_", "").replace("-", "").isalpha(): continue raise AuditError( f"audit field contains high-entropy token ({run[:8]}…) — suspected secret" ) def _audit_path(state_dir: Path) -> Path: return state_dir / _AUDIT_FILENAME def _maybe_rotate(path: Path) -> None: if path.exists() and path.stat().st_size > _MAX_BYTES: backup = path.with_suffix(".jsonl.1") backup.unlink(missing_ok=True) path.rename(backup) def record_event( state_dir: Path, *, kind: str, action: str, subject: str = "", target: str = "", decision_id: Optional[str] = None, outcome: str = "ok", source: str = "audit", **extra: Any, ) -> Path: """Append one metadata-only audit event. Never pass secret values in any field.""" event = { "ts": datetime.now(timezone.utc).isoformat(), "kind": kind, "action": action, "subject": subject, "target": target, "decision_id": decision_id, "outcome": outcome, "source": source, } for key, value in extra.items(): if value is None: continue event[key] = value _assert_metadata_safe(json.dumps(event, default=str)) state_dir.mkdir(parents=True, exist_ok=True) path = _audit_path(state_dir) _maybe_rotate(path) with path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(event, default=str) + "\n") return path def read_events( state_dir: Path, *, since: Optional[datetime] = None, kinds: Optional[set[str]] = None, ) -> list[dict[str, Any]]: """Read unified audit events newer than ``since`` (UTC), optionally filtered by kind.""" path = _audit_path(state_dir) if not path.exists(): return [] events: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: event = json.loads(line) except json.JSONDecodeError: continue if kinds and event.get("kind") not in kinds: continue if since: ts_raw = event.get("ts") if not ts_raw: continue try: ts = datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) except ValueError: continue if ts < since: continue events.append(event) return events def _legacy_sign_events(state_dir: Path, since: Optional[datetime]) -> list[dict[str, Any]]: path = state_dir / "signatures.log" if not path.exists(): return [] out: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: raw = json.loads(line) except json.JSONDecodeError: continue ts_raw = raw.get("timestamp") if since and ts_raw: try: ts = datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) except ValueError: continue if ts < since: continue out.append( { "ts": ts_raw, "kind": "sign", "action": "issue", "subject": raw.get("actor", ""), "target": raw.get("actor", ""), "decision_id": raw.get("policy_decision_id"), "outcome": "ok", "source": "signatures.log", "backend": raw.get("backend"), "actor_type": raw.get("actor_type"), } ) return out def _legacy_access_events(state_dir: Path, since: Optional[datetime]) -> list[dict[str, Any]]: path = state_dir / "access-audit.log" if not path.exists(): return [] out: list[dict[str, Any]] = [] for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: raw = json.loads(line) except json.JSONDecodeError: continue ts_raw = raw.get("timestamp") if since and ts_raw: try: ts = datetime.fromisoformat(str(ts_raw).replace("Z", "+00:00")) except ValueError: continue if ts < since: continue out.append( { "ts": ts_raw, "kind": "access", "action": raw.get("action", "fetch"), "subject": raw.get("subject", ""), "target": raw.get("need_id", ""), "decision_id": raw.get("policy_decision_id"), "outcome": "ok" if raw.get("exit_code", 0) == 0 else "error", "source": "access-audit.log", "owner_repo": raw.get("owner_repo"), } ) return out def collect_activity( state_dir: Path, *, days: int = 7, kinds: Optional[set[str]] = None, include_legacy: bool = True, ) -> list[dict[str, Any]]: """Merge unified audit + legacy logs into one chronological list.""" since = datetime.now(timezone.utc) - timedelta(days=days) events = read_events(state_dir, since=since, kinds=kinds) if include_legacy: legacy_kinds = kinds or {"sign", "access", "worker"} if not kinds or "sign" in kinds: events.extend(_legacy_sign_events(state_dir, since)) if not kinds or "access" in kinds: events.extend(_legacy_access_events(state_dir, since)) # De-dupe unified vs legacy: prefer audit.jsonl when same ts+kind+action+target seen: set[tuple[str, str, str, str]] = set() unique: list[dict[str, Any]] = [] for event in events: key = ( str(event.get("ts", "")), str(event.get("kind", "")), str(event.get("action", "")), str(event.get("target", "")), ) if key in seen and event.get("source") != "audit": continue seen.add(key) unique.append(event) unique.sort(key=lambda e: str(e.get("ts", ""))) return unique def fetch_hub_notes(*, days: int = 7, hub_url: Optional[str] = None) -> list[dict[str, Any]]: """Best-effort pull of recent ops-warden-related State Hub progress notes.""" import httpx base = (hub_url or os.environ.get("STATE_HUB_URL", "http://127.0.0.1:8000")).rstrip("/") since = datetime.now(timezone.utc) - timedelta(days=days) try: resp = httpx.get(f"{base}/progress/", params={"limit": 100}, timeout=5.0) resp.raise_for_status() payload = resp.json() except Exception: return [] items = payload if isinstance(payload, list) else payload.get("items", []) notes: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue summary = str(item.get("summary", "")) if "ops-warden" not in summary.lower() and "[worker]" not in summary: continue created = item.get("created_at") if created: try: ts = datetime.fromisoformat(str(created).replace("Z", "+00:00")) if ts < since: continue except ValueError: pass notes.append( { "ts": created, "kind": "hub", "action": item.get("event_type", "note"), "subject": item.get("author", ""), "target": "state-hub", "outcome": "ok", "source": "state-hub", "summary": summary, } ) return notes