Coverage for agentos/enterprise/audit.py: 49%
122 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 23:17 +0800
1"""
2AgentOS Enterprise — Audit Logging.
4功能：
5 - 全量审计事件记录
6 - 事件分类（认证/操作/数据/系统）
7 - 可配置保留策略
8 - 合规报告导出（CSV/JSON）
9 - GDPR/CCPA 数据删除支持
10"""
12from __future__ import annotations
14import csv
15import io
16import json
17import time
18import uuid
19from dataclasses import dataclass, field
20from enum import StrEnum
23class AuditCategory(StrEnum):
24 """审计事件分类。"""
26 AUTH = "auth" # 登录/登出/Token
27 API_KEY = "api_key" # Key 创建/撤销/轮转
28 AGENT = "agent" # Agent 创建/运行/删除
29 TENANT = "tenant" # 租户管理
30 CONFIG = "config" # 配置变更
31 SYSTEM = "system" # 系统事件
32 DATA = "data" # 数据访问/导出
33 SECURITY = "security" # 安全事件（违规/攻击）
36class AuditSeverity(StrEnum):
37 INFO = "info"
38 WARNING = "warning"
39 ERROR = "error"
40 CRITICAL = "critical"
43@dataclass
44class AuditEvent:
45 """一条审计事件。"""
47 event_id: str
48 timestamp: float
49 category: AuditCategory
50 action: str # 如 "api_key.created", "agent.run"
51 severity: AuditSeverity
52 actor_type: str # "user" / "agent" / "system" / "api_key"
53 actor_id: str # user_id / agent_id / key_id
54 tenant_id: str
55 resource_type: str # "agent" / "api_key" / "tenant" / ...
56 resource_id: str
57 ip_address: str
58 user_agent: str
59 status: str # "success" / "failure"
60 details: dict = field(default_factory=dict)
61 metadata: dict = field(default_factory=dict)
64@dataclass
65class RetentionPolicy:
66 """审计日志保留策略。"""
68 max_events: int = 100_000 # 最大事件数
69 max_age_days: int = 90 # 最大保留天数
70 auto_prune: bool = True # 是否自动清理过期事件
73class AuditLogger:
74 """审计日志引擎。
76 特性：
77 - 全量事件记录
78 - 内存 + 文件双存储模式
79 - 可配置保留策略
80 - 过滤/搜索/导出
81 - 合规支持（GDPR 数据删除）
82 """
84 def __init__(self, retention: RetentionPolicy = None):
85 self._events: list[AuditEvent] = []
86 self.retention = retention or RetentionPolicy()
88 # ── 记录 ──
90 def log(
91 self,
92 category: AuditCategory,
93 action: str,
94 severity: AuditSeverity = AuditSeverity.INFO,
95 actor_type: str = "system",
96 actor_id: str = "",
97 tenant_id: str = "",
98 resource_type: str = "",
99 resource_id: str = "",
100 ip_address: str = "",
101 user_agent: str = "",
102 status: str = "success",
103 details: dict = None,
104 metadata: dict = None,
105 ) -> AuditEvent:
106 """记录一条审计事件。"""
107 event = AuditEvent(
108 event_id=f"evt_{uuid.uuid4().hex[:12]}",
109 timestamp=time.time(),
110 category=category,
111 action=action,
112 severity=severity,
113 actor_type=actor_type,
114 actor_id=actor_id,
115 tenant_id=tenant_id,
116 resource_type=resource_type,
117 resource_id=resource_id,
118 ip_address=ip_address,
119 user_agent=user_agent,
120 status=status,
121 details=details or {},
122 metadata=metadata or {},
123 )
124 self._events.append(event)
126 # 自动清理
127 if self.retention.auto_prune:
128 self._prune()
130 return event
132 def log_auth(
133 self,
134 action: str,
135 user_id: str,
136 tenant_id: str,
137 status: str,
138 ip: str = "",
139 ua: str = "",
140 **kwargs,
141 ):
142 """便捷：记录认证事件。"""
143 return self.log(
144 category=AuditCategory.AUTH,
145 action=action,
146 actor_type="user",
147 actor_id=user_id,
148 tenant_id=tenant_id,
149 resource_type="session",
150 resource_id=user_id,
151 ip_address=ip,
152 user_agent=ua,
153 status=status,
154 details=kwargs,
155 )
157 def log_api_key(self, action: str, key_id: str, tenant_id: str, actor_id: str, **kwargs):
158 """便捷：记录 API Key 事件。"""
159 return self.log(
160 category=AuditCategory.API_KEY,
161 action=action,
162 severity=AuditSeverity.WARNING if action.endswith(".revoked") else AuditSeverity.INFO,
163 actor_type="user",
164 actor_id=actor_id,
165 tenant_id=tenant_id,
166 resource_type="api_key",
167 resource_id=key_id,
168 details=kwargs,
169 )
171 def log_agent_run(self, agent_id: str, tenant_id: str, status: str, **kwargs):
172 """便捷：记录 Agent 运行事件。"""
173 return self.log(
174 category=AuditCategory.AGENT,
175 action="agent.run",
176 actor_type="agent",
177 actor_id=agent_id,
178 tenant_id=tenant_id,
179 resource_type="agent",
180 resource_id=agent_id,
181 status=status,
182 details=kwargs,
183 )
185 def log_security(self, action: str, severity: AuditSeverity, details: dict, **kwargs):
186 """便捷：记录安全事件。"""
187 return self.log(
188 category=AuditCategory.SECURITY,
189 action=action,
190 severity=severity,
191 actor_type=kwargs.pop("actor_type", "system"),
192 actor_id=kwargs.pop("actor_id", ""),
193 tenant_id=kwargs.pop("tenant_id", ""),
194 resource_type=kwargs.pop("resource_type", ""),
195 resource_id=kwargs.pop("resource_id", ""),
196 details=details,
197 **kwargs,
198 )
200 # ── 查询 ──
202 def query(
203 self,
204 category: AuditCategory | None = None,
205 severity: AuditSeverity | None = None,
206 tenant_id: str | None = None,
207 actor_id: str | None = None,
208 status: str | None = None,
209 since: float | None = None,
210 until: float | None = None,
211 limit: int = 100,
212 ) -> list[AuditEvent]:
213 """多条件过滤查询。"""
214 results = self._events
216 if category:
217 results = [e for e in results if e.category == category]
218 if severity:
219 results = [e for e in results if e.severity == severity]
220 if tenant_id:
221 results = [e for e in results if e.tenant_id == tenant_id]
222 if actor_id:
223 results = [e for e in results if e.actor_id == actor_id]
224 if status:
225 results = [e for e in results if e.status == status]
226 if since:
227 results = [e for e in results if e.timestamp >= since]
228 if until:
229 results = [e for e in results if e.timestamp <= until]
231 return sorted(results, key=lambda e: e.timestamp, reverse=True)[:limit]
233 def get_recent(self, n: int = 50) -> list[AuditEvent]:
234 """最近 N 条事件。"""
235 return sorted(self._events, key=lambda e: e.timestamp, reverse=True)[:n]
237 # ── 导出 ──
239 def export_json(self, events: list[AuditEvent] = None) -> str:
240 """导出为 JSON 字符串。"""
241 target = events or self._events
242 return json.dumps(
243 [_event_to_dict(e) for e in target],
244 ensure_ascii=False,
245 indent=2,
246 default=str,
247 )
249 def export_csv(self, events: list[AuditEvent] = None) -> str:
250 """导出为 CSV 字符串。"""
251 target = events or self._events
252 output = io.StringIO()
253 writer = csv.DictWriter(
254 output,
255 fieldnames=[
256 "event_id",
257 "timestamp",
258 "category",
259 "action",
260 "severity",
261 "actor_type",
262 "actor_id",
263 "tenant_id",
264 "resource_type",
265 "resource_id",
266 "ip_address",
267 "status",
268 "details",
269 ],
270 )
271 writer.writeheader()
272 for e in target:
273 writer.writerow(
274 {
275 "event_id": e.event_id,
276 "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(e.timestamp)),
277 "category": e.category.value,
278 "action": e.action,
279 "severity": e.severity.value,
280 "actor_type": e.actor_type,
281 "actor_id": e.actor_id,
282 "tenant_id": e.tenant_id,
283 "resource_type": e.resource_type,
284 "resource_id": e.resource_id,
285 "ip_address": e.ip_address,
286 "status": e.status,
287 "details": json.dumps(e.details, ensure_ascii=False),
288 }
289 )
290 return output.getvalue()
292 # ── 合规 ──
294 def delete_user_data(self, user_id: str) -> int:
295 """GDPR / CCPA：删除指定用户相关的所有审计记录。返回删除条数。"""
296 before = len(self._events)
297 self._events = [e for e in self._events if e.actor_id != user_id]
298 return before - len(self._events)
300 def compliance_report(self, tenant_id: str, start: float, end: float) -> dict:
301 """生成合规报告摘要。"""
302 events = self.query(tenant_id=tenant_id, since=start, until=end, limit=10000)
303 by_category = {}
304 by_severity = {}
305 failure_count = 0
306 for e in events:
307 by_category[e.category.value] = by_category.get(e.category.value, 0) + 1
308 by_severity[e.severity.value] = by_severity.get(e.severity.value, 0) + 1
309 if e.status == "failure":
310 failure_count += 1
311 return {
312 "tenant_id": tenant_id,
313 "period": {
314 "start": time.strftime("%Y-%m-%d", time.gmtime(start)),
315 "end": time.strftime("%Y-%m-%d", time.gmtime(end)),
316 },
317 "total_events": len(events),
318 "by_category": by_category,
319 "by_severity": by_severity,
320 "failure_rate": f"{failure_count / len(events) * 100:.1f}%" if events else "0%",
321 "generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
322 }
324 # ── 统计 ──
326 def stats(self) -> dict:
327 total = len(self._events)
328 by_category = {}
329 by_severity = {}
330 for e in self._events:
331 by_category[e.category.value] = by_category.get(e.category.value, 0) + 1
332 by_severity[e.severity.value] = by_severity.get(e.severity.value, 0) + 1
333 return {
334 "total_events": total,
335 "by_category": by_category,
336 "by_severity": by_severity,
337 "retention_policy": {
338 "max_events": self.retention.max_events,
339 "max_age_days": self.retention.max_age_days,
340 },
341 }
343 # ── 内部 ──
345 def _prune(self):
346 """按保留策略清理过期事件。"""
347 # 按数量
348 if len(self._events) > self.retention.max_events:
349 self._events = self._events[-self.retention.max_events :]
351 # 按时间
352 cutoff = time.time() - self.retention.max_age_days * 86400
353 self._events = [e for e in self._events if e.timestamp > cutoff]
356def _event_to_dict(e: AuditEvent) -> dict:
357 return {
358 "event_id": e.event_id,
359 "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(e.timestamp)),
360 "category": e.category.value,
361 "action": e.action,
362 "severity": e.severity.value,
363 "actor_type": e.actor_type,
364 "actor_id": e.actor_id,
365 "tenant_id": e.tenant_id,
366 "resource_type": e.resource_type,
367 "resource_id": e.resource_id,
368 "ip_address": e.ip_address,
369 "user_agent": e.user_agent,
370 "status": e.status,
371 "details": e.details,
372 "metadata": e.metadata,
373 }