Coverage for agentos/tools/audit_logger.py: 38%
108 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 23:17 +0800
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 23:17 +0800
1"""
2AuditLogger — ring-buffer audit log with JSON export, level filtering, and callbacks.
4Supports:
5 - Structured audit events (actor, action, resource, outcome, details)
6 - Severity levels (INFO, WARNING, ERROR, CRITICAL)
7 - Ring buffer with configurable capacity
8 - JSON export (to file or string)
9 - Level-based filtering
10 - Subscription callbacks for real-time forwarding
11 - Thread-safe append
12"""
14from __future__ import annotations
16import json
17import threading
18import time
19from collections.abc import Callable
20from dataclasses import asdict, dataclass, field
21from enum import Enum
22from pathlib import Path
23from typing import Any
25# ============================================================================
26# Severity
27# ============================================================================
30class Severity(Enum):
31 INFO = 10
32 WARNING = 20
33 ERROR = 30
34 CRITICAL = 40
36 @classmethod
37 def from_str(cls, s: str) -> Severity:
38 return getattr(cls, s.upper(), cls.INFO)
41# ============================================================================
42# AuditEvent
43# ============================================================================
46@dataclass
47class AuditEvent:
48 """Single audit log entry."""
50 actor: str = "" # who performed the action
51 action: str = "" # what was done (e.g., "user.delete", "config.update")
52 resource: str = "" # what was acted upon (e.g., "user:123", "/etc/config.yaml")
53 outcome: str = "" # "success", "failure", "denied"
54 severity: Severity = Severity.INFO
55 details: dict[str, Any] = field(default_factory=dict)
56 timestamp: float = field(default_factory=time.time)
58 def to_dict(self) -> dict[str, Any]:
59 d = asdict(self)
60 d["severity"] = self.severity.name
61 d["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(self.timestamp))
62 return d
64 @classmethod
65 def from_dict(cls, d: dict[str, Any]) -> AuditEvent:
66 return cls(
67 actor=d.get("actor", ""),
68 action=d.get("action", ""),
69 resource=d.get("resource", ""),
70 outcome=d.get("outcome", ""),
71 severity=Severity.from_str(d.get("severity", "INFO")),
72 details=d.get("details", {}),
73 timestamp=d.get("timestamp", time.time()),
74 )
77# ============================================================================
78# AuditLogger
79# ============================================================================
82class AuditLogger:
83 """Ring-buffer audit logger.
85 Usage:
86 audit = AuditLogger(capacity=2000)
88 # Record an event
89 audit.log(
90 actor="admin",
91 action="user.delete",
92 resource="user:42",
93 outcome="success",
94 severity=Severity.WARNING,
95 details={"reason": "GDPR request"},
96 )
98 # Export as JSON
99 audit.export_json("audit_2026.json")
101 # Subscribe to events in real-time
102 audit.subscribe(lambda event: forward_to_siem(event))
104 # Query with filter
105 recent_failures = audit.query(
106 min_severity=Severity.ERROR,
107 limit=50,
108 )
109 """
111 def __init__(self, capacity: int = 5000):
112 if capacity <= 0:
113 raise ValueError("capacity must be positive")
114 self._capacity = capacity
115 self._buffer: list[AuditEvent] = []
116 self._lock = threading.RLock()
117 self._subscribers: list[Callable[[AuditEvent], None]] = []
119 # ---------- log ----------
121 def log(
122 self,
123 actor: str = "",
124 action: str = "",
125 resource: str = "",
126 outcome: str = "",
127 severity: Severity = Severity.INFO,
128 details: dict[str, Any] | None = None,
129 event: AuditEvent | None = None,
130 ) -> AuditEvent:
131 """Record an audit event. Accepts either field args or an AuditEvent object."""
132 if event is None:
133 event = AuditEvent(
134 actor=actor,
135 action=action,
136 resource=resource,
137 outcome=outcome,
138 severity=severity,
139 details=details or {},
140 )
141 with self._lock:
142 self._buffer.append(event)
143 # Ring buffer eviction
144 excess = len(self._buffer) - self._capacity
145 if excess > 0:
146 self._buffer = self._buffer[excess:]
148 # Notify subscribers outside lock
149 self._notify(event)
150 return event
152 # ---------- query ----------
154 def query(
155 self,
156 actor: str | None = None,
157 action: str | None = None,
158 resource: str | None = None,
159 outcome: str | None = None,
160 min_severity: Severity | None = None,
161 max_severity: Severity | None = None,
162 since: float | None = None,
163 until: float | None = None,
164 limit: int | None = None,
165 ) -> list[AuditEvent]:
166 """Query audit events with optional filters."""
167 with self._lock:
168 results = list(self._buffer)
170 if actor:
171 results = [e for e in results if e.actor == actor]
172 if action:
173 results = [e for e in results if e.action == action]
174 if resource:
175 results = [e for e in results if e.resource == resource]
176 if outcome:
177 results = [e for e in results if e.outcome == outcome]
178 if min_severity:
179 results = [e for e in results if e.severity.value >= min_severity.value]
180 if max_severity:
181 results = [e for e in results if e.severity.value <= max_severity.value]
182 if since is not None:
183 results = [e for e in results if e.timestamp >= since]
184 if until is not None:
185 results = [e for e in results if e.timestamp <= until]
187 if limit is not None and limit > 0:
188 results = results[-limit:]
190 return results
192 def recent(self, count: int = 20) -> list[AuditEvent]:
193 """Return the most recent N events."""
194 with self._lock:
195 return self._buffer[-count:] if count < len(self._buffer) else list(self._buffer)
197 # ---------- export ----------
199 def export_json(self, path: str | None = None) -> str:
200 """Export all events as JSON. If path given, writes to file."""
201 with self._lock:
202 data = [e.to_dict() for e in self._buffer]
204 json_str = json.dumps(data, indent=2, ensure_ascii=False)
206 if path:
207 Path(path).write_text(json_str, encoding="utf-8")
209 return json_str
211 # ---------- subscription ----------
213 def subscribe(self, callback: Callable[[AuditEvent], None]) -> None:
214 """Register a callback for real-time event forwarding."""
215 with self._lock:
216 self._subscribers.append(callback)
218 def unsubscribe(self, callback: Callable[[AuditEvent], None]) -> bool:
219 with self._lock:
220 if callback in self._subscribers:
221 self._subscribers.remove(callback)
222 return True
223 return False
225 def _notify(self, event: AuditEvent) -> None:
226 with self._lock:
227 subs = list(self._subscribers)
228 for cb in subs:
229 try:
230 cb(event)
231 except Exception:
232 pass
234 # ---------- info ----------
236 @property
237 def count(self) -> int:
238 with self._lock:
239 return len(self._buffer)
241 @property
242 def capacity(self) -> int:
243 return self._capacity