Coverage for agentos/tools/audit_logger.py: 38%

1""" 

2AuditLogger — ring-buffer audit log with JSON export, level filtering, and callbacks. 

3 

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""" 

13 

14from __future__ import annotations 

15 

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 

24 

25# ============================================================================ 

26# Severity 

27# ============================================================================ 

28 

29 

30class Severity(Enum): 

31 INFO = 10 

32 WARNING = 20 

33 ERROR = 30 

34 CRITICAL = 40 

35 

36 @classmethod 

37 def from_str(cls, s: str) -> Severity: 

38 return getattr(cls, s.upper(), cls.INFO) 

39 

40 

41# ============================================================================ 

42# AuditEvent 

43# ============================================================================ 

44 

45 

46@dataclass 

47class AuditEvent: 

48 """Single audit log entry.""" 

49 

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) 

57 

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 

63 

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 ) 

75 

76 

77# ============================================================================ 

78# AuditLogger 

79# ============================================================================ 

80 

81 

82class AuditLogger: 

83 """Ring-buffer audit logger. 

84 

85 Usage: 

86 audit = AuditLogger(capacity=2000) 

87 

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 ) 

97 

98 # Export as JSON 

99 audit.export_json("audit_2026.json") 

100 

101 # Subscribe to events in real-time 

102 audit.subscribe(lambda event: forward_to_siem(event)) 

103 

104 # Query with filter 

105 recent_failures = audit.query( 

106 min_severity=Severity.ERROR, 

107 limit=50, 

108 ) 

109 """ 

110 

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]] = [] 

118 

119 # ---------- log ---------- 

120 

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:] 

147 

148 # Notify subscribers outside lock 

149 self._notify(event) 

150 return event 

151 

152 # ---------- query ---------- 

153 

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) 

169 

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] 

186 

187 if limit is not None and limit > 0: 

188 results = results[-limit:] 

189 

190 return results 

191 

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) 

196 

197 # ---------- export ---------- 

198 

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] 

203 

204 json_str = json.dumps(data, indent=2, ensure_ascii=False) 

205 

206 if path: 

207 Path(path).write_text(json_str, encoding="utf-8") 

208 

209 return json_str 

210 

211 # ---------- subscription ---------- 

212 

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) 

217 

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 

224 

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 

233 

234 # ---------- info ---------- 

235 

236 @property 

237 def count(self) -> int: 

238 with self._lock: 

239 return len(self._buffer) 

240 

241 @property 

242 def capacity(self) -> int: 

243 return self._capacity