"""Structured audit log persisted to Redis. Every state-changing operation goes through :func:`audit_log`. We persist a compact JSON record both in: - A capped global list ``deepagent:audit`` (last 10 000 events) - A per-installation index ``deepagent:audit:install:`` (last 1 000) …and we ALSO emit a ``structlog`` event so the same data flows to Loki / your log aggregator. The :class:`AuditEvent` records: - ``actor``: GitHub login or admin token prefix - ``via``: ``github`` / ``admin_token`` / ``anonymous`` / ``system`` - ``action``: short verb, e.g. ``role.grant``, ``dlq.requeue``, ``job.create`` - ``target``: free-form descriptor (login affected, job id, repo) - ``installation_id``: optional scope - ``metadata``: free-form dict (truncated to 1 KB) Querying for the dashboard's *Activity* page uses :meth:`AuditStore.tail_global` / :meth:`AuditStore.tail_for_installation`. """ from __future__ import annotations import json import logging import os import time from dataclasses import asdict, dataclass, field from typing import Optional import redis from .logging_setup import get_logger log = logging.getLogger(__name__) _struct = get_logger("audit") @dataclass class AuditEvent: actor: str via: str action: str target: str = "" installation_id: Optional[int] = None metadata: dict = field(default_factory=dict) timestamp: float = field(default_factory=time.time) def to_json(self) -> str: # Truncate metadata to 1 KB to keep records bounded. d = asdict(self) try: meta_s = json.dumps(d["metadata"], default=str)[:1024] d["metadata"] = json.loads(meta_s) if meta_s.startswith(("{", "[")) else meta_s except Exception: d["metadata"] = {"truncated": True} return json.dumps(d, default=str) class AuditStore: GLOBAL_KEY = "deepagent:audit" PER_INSTALL_PREFIX = "deepagent:audit:install" GLOBAL_CAP = 10_000 PER_INSTALL_CAP = 1_000 def __init__(self, client: Optional[redis.Redis] = None): self._r = client or redis.Redis.from_url( os.getenv("DEEPAGENT_REDIS_URL", "redis://localhost:6379/0") ) def _per_install_key(self, iid: int) -> str: return f"{self.PER_INSTALL_PREFIX}:{iid}" def append(self, event: AuditEvent) -> None: try: payload = event.to_json() pipe = self._r.pipeline() pipe.lpush(self.GLOBAL_KEY, payload) pipe.ltrim(self.GLOBAL_KEY, 0, self.GLOBAL_CAP - 1) if event.installation_id is not None: pipe.lpush(self._per_install_key(event.installation_id), payload) pipe.ltrim(self._per_install_key(event.installation_id), 0, self.PER_INSTALL_CAP - 1) pipe.execute() except redis.RedisError as e: # pragma: no cover log.warning("audit persist failed: %s", e) # Always log too — single source of truth via structlog. _struct.info( "audit", actor=event.actor, via=event.via, action=event.action, target=event.target, installation_id=event.installation_id, **{f"meta_{k}": v for k, v in (event.metadata or {}).items()}, ) def tail_global(self, limit: int = 200) -> list[dict]: try: raw = self._r.lrange(self.GLOBAL_KEY, 0, limit - 1) except redis.RedisError: return [] return self._decode(raw) def tail_for_installation(self, iid: int, limit: int = 200) -> list[dict]: try: raw = self._r.lrange(self._per_install_key(iid), 0, limit - 1) except redis.RedisError: return [] return self._decode(raw) @staticmethod def _decode(raw: list) -> list[dict]: out = [] for b in raw: s = b.decode() if isinstance(b, bytes) else b try: out.append(json.loads(s)) except json.JSONDecodeError: continue return out # Module-level singleton _store: Optional[AuditStore] = None def get_audit_store() -> AuditStore: global _store if _store is None: _store = AuditStore() return _store def reset_audit_store() -> None: # test helper global _store _store = None # ---------------------------------------------------------------- helper def audit_log( *, actor: str = "system", via: str = "system", action: str, target: str = "", installation_id: Optional[int] = None, **metadata, ) -> None: """Convenience wrapper used everywhere in the codebase.""" get_audit_store().append(AuditEvent( actor=actor, via=via, action=action, target=target, installation_id=installation_id, metadata=metadata or {}, ))