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

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 dataclasses import dataclass, field, asdict 

20from enum import Enum 

21from pathlib import Path 

22from typing import Any, Callable, Dict, List, Optional 

23 

24 

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

26# Severity 

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

28 

29class Severity(Enum): 

30 INFO = 10 

31 WARNING = 20 

32 ERROR = 30 

33 CRITICAL = 40 

34 

35 @classmethod 

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

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

38 

39 

40# ============================================================================ 

41# AuditEvent 

42# ============================================================================ 

43 

44@dataclass 

45class AuditEvent: 

46 """Single audit log entry.""" 

47 actor: str = "" # who performed the action 

48 action: str = "" # what was done (e.g., "user.delete", "config.update") 

49 resource: str = "" # what was acted upon (e.g., "user:123", "/etc/config.yaml") 

50 outcome: str = "" # "success", "failure", "denied" 

51 severity: Severity = Severity.INFO 

52 details: Dict[str, Any] = field(default_factory=dict) 

53 timestamp: float = field(default_factory=time.time) 

54 

55 def to_dict(self) -> Dict[str, Any]: 

56 d = asdict(self) 

57 d["severity"] = self.severity.name 

58 d["timestamp"] = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(self.timestamp)) 

59 return d 

60 

61 @classmethod 

62 def from_dict(cls, d: Dict[str, Any]) -> "AuditEvent": 

63 return cls( 

64 actor=d.get("actor", ""), 

65 action=d.get("action", ""), 

66 resource=d.get("resource", ""), 

67 outcome=d.get("outcome", ""), 

68 severity=Severity.from_str(d.get("severity", "INFO")), 

69 details=d.get("details", {}), 

70 timestamp=d.get("timestamp", time.time()), 

71 ) 

72 

73 

74# ============================================================================ 

75# AuditLogger 

76# ============================================================================ 

77 

78class AuditLogger: 

79 """Ring-buffer audit logger. 

80 

81 Usage: 

82 audit = AuditLogger(capacity=2000) 

83 

84 # Record an event 

85 audit.log( 

86 actor="admin", 

87 action="user.delete", 

88 resource="user:42", 

89 outcome="success", 

90 severity=Severity.WARNING, 

91 details={"reason": "GDPR request"}, 

92 ) 

93 

94 # Export as JSON 

95 audit.export_json("audit_2026.json") 

96 

97 # Subscribe to events in real-time 

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

99 

100 # Query with filter 

101 recent_failures = audit.query( 

102 min_severity=Severity.ERROR, 

103 limit=50, 

104 ) 

105 """ 

106 

107 def __init__(self, capacity: int = 5000): 

108 if capacity <= 0: 

109 raise ValueError("capacity must be positive") 

110 self._capacity = capacity 

111 self._buffer: List[AuditEvent] = [] 

112 self._lock = threading.RLock() 

113 self._subscribers: List[Callable[[AuditEvent], None]] = [] 

114 

115 # ---------- log ---------- 

116 

117 def log( 

118 self, 

119 actor: str = "", 

120 action: str = "", 

121 resource: str = "", 

122 outcome: str = "", 

123 severity: Severity = Severity.INFO, 

124 details: Optional[Dict[str, Any]] = None, 

125 event: AuditEvent | None = None, 

126 ) -> AuditEvent: 

127 """Record an audit event. Accepts either field args or an AuditEvent object.""" 

128 if event is None: 

129 event = AuditEvent( 

130 actor=actor, 

131 action=action, 

132 resource=resource, 

133 outcome=outcome, 

134 severity=severity, 

135 details=details or {}, 

136 ) 

137 with self._lock: 

138 self._buffer.append(event) 

139 # Ring buffer eviction 

140 excess = len(self._buffer) - self._capacity 

141 if excess > 0: 

142 self._buffer = self._buffer[excess:] 

143 

144 # Notify subscribers outside lock 

145 self._notify(event) 

146 return event 

147 

148 # ---------- query ---------- 

149 

150 def query( 

151 self, 

152 actor: Optional[str] = None, 

153 action: Optional[str] = None, 

154 resource: Optional[str] = None, 

155 outcome: Optional[str] = None, 

156 min_severity: Optional[Severity] = None, 

157 max_severity: Optional[Severity] = None, 

158 since: Optional[float] = None, 

159 until: Optional[float] = None, 

160 limit: Optional[int] = None, 

161 ) -> List[AuditEvent]: 

162 """Query audit events with optional filters.""" 

163 with self._lock: 

164 results = list(self._buffer) 

165 

166 if actor: 

167 results = [e for e in results if e.actor == actor] 

168 if action: 

169 results = [e for e in results if e.action == action] 

170 if resource: 

171 results = [e for e in results if e.resource == resource] 

172 if outcome: 

173 results = [e for e in results if e.outcome == outcome] 

174 if min_severity: 

175 results = [e for e in results if e.severity.value >= min_severity.value] 

176 if max_severity: 

177 results = [e for e in results if e.severity.value <= max_severity.value] 

178 if since is not None: 

179 results = [e for e in results if e.timestamp >= since] 

180 if until is not None: 

181 results = [e for e in results if e.timestamp <= until] 

182 

183 if limit is not None and limit > 0: 

184 results = results[-limit:] 

185 

186 return results 

187 

188 def recent(self, count: int = 20) -> List[AuditEvent]: 

189 """Return the most recent N events.""" 

190 with self._lock: 

191 return self._buffer[-count:] if count < len(self._buffer) else list(self._buffer) 

192 

193 # ---------- export ---------- 

194 

195 def export_json(self, path: Optional[str] = None) -> str: 

196 """Export all events as JSON. If path given, writes to file.""" 

197 with self._lock: 

198 data = [e.to_dict() for e in self._buffer] 

199 

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

201 

202 if path: 

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

204 

205 return json_str 

206 

207 # ---------- subscription ---------- 

208 

209 def subscribe(self, callback: Callable[[AuditEvent], None]) -> None: 

210 """Register a callback for real-time event forwarding.""" 

211 with self._lock: 

212 self._subscribers.append(callback) 

213 

214 def unsubscribe(self, callback: Callable[[AuditEvent], None]) -> bool: 

215 with self._lock: 

216 if callback in self._subscribers: 

217 self._subscribers.remove(callback) 

218 return True 

219 return False 

220 

221 def _notify(self, event: AuditEvent) -> None: 

222 with self._lock: 

223 subs = list(self._subscribers) 

224 for cb in subs: 

225 try: 

226 cb(event) 

227 except Exception: 

228 pass 

229 

230 # ---------- info ---------- 

231 

232 @property 

233 def count(self) -> int: 

234 with self._lock: 

235 return len(self._buffer) 

236 

237 @property 

238 def capacity(self) -> int: 

239 return self._capacity