Coverage for src / lexigram / admin / auth / services / audit_log_service.py: 33%
61 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Admin security audit log service."""
3from __future__ import annotations
5from datetime import UTC, datetime
6from typing import Any
7import uuid
9from lexigram.admin.auth.protocols import (
10 AdminAuditLogServiceProtocol,
11 AdminAuditLogStoreProtocol,
12)
13from lexigram.admin.auth.types import AdminSecurityEvent, AdminSecurityEventType
14from lexigram.contracts.audit import (
15 AuditEntry,
16 AuditEventSeverity,
17 AuditLoggerProtocol,
18 AuditQuery,
19)
20from lexigram.di.decorators import inject
21from lexigram.logging import get_logger
23logger = get_logger(__name__)
26@inject
27class AdminAuditLogService:
28 """Security event audit log service.
30 All methods are fire-tolerant — exceptions are logged but never re-raised.
31 Audit failure must never block the authentication flow; a failed audit
32 write is strictly less important than completing the operation that
33 triggered it (e.g. a login or logout).
34 """
36 def __init__(
37 self,
38 store: AdminAuditLogStoreProtocol,
39 audit_logger: AuditLoggerProtocol | None = None,
40 ) -> None:
41 """Initialize with audit log store.
43 Args:
44 store: Persistence layer for security events.
45 audit_logger: Optional unified AuditLoggerProtocol bridge. When
46 provided, every event is also forwarded to the centralised
47 audit log alongside the admin-specific store.
48 """
49 self._store = store
50 self._audit_logger = audit_logger
52 async def log_event(
53 self,
54 event_type: AdminSecurityEventType,
55 ip_address: str,
56 user_agent: str,
57 success: bool,
58 admin_user_id: str | None = None,
59 metadata: dict[str, Any] | None = None,
60 ) -> None:
61 """Record a security event. Never raises.
63 Swallows all exceptions so that an audit store failure (e.g. database
64 unreachable, serialisation error) cannot interrupt the calling auth
65 flow. The failure is recorded at WARNING level so operators are
66 notified without surfacing an error to the end user.
68 Args:
69 event_type: Type of security event.
70 ip_address: Client IP.
71 user_agent: Client user agent.
72 success: Whether the operation succeeded.
73 admin_user_id: Associated admin user (None for pre-auth events).
74 metadata: Optional structured context dict.
75 """
76 try:
77 # Only primitive-safe values are stored; anything else is coerced
78 # to str so that the JSONB/TEXT column always receives serialisable
79 # data regardless of what the caller provides.
80 safe_meta: dict[str, str | int | bool | None] = {}
81 for k, v in (metadata or {}).items():
82 if isinstance(v, (str, int, bool)) or v is None:
83 safe_meta[str(k)] = v
84 else:
85 safe_meta[str(k)] = str(v)
87 event = AdminSecurityEvent(
88 id=str(uuid.uuid4()),
89 event_type=event_type,
90 admin_user_id=admin_user_id,
91 ip_address=ip_address,
92 user_agent=user_agent,
93 success=success,
94 metadata=safe_meta,
95 created_at=datetime.now(UTC),
96 )
97 await self._store.insert(event)
98 # Bridge to unified audit logger when configured.
99 if self._audit_logger is not None:
100 try:
101 unified_entry = AuditEntry(
102 action=event_type.value,
103 actor_id=admin_user_id or "",
104 resource_type="admin.security",
105 resource_id=event.id,
106 outcome="success" if success else "failure",
107 source="admin",
108 metadata=dict(safe_meta),
109 )
110 await self._audit_logger.log(unified_entry)
111 except Exception: # noqa: BLE001
112 logger.warning(
113 "audit.bridge_log_failed", event_type=event_type.value
114 )
115 logger.debug(
116 "audit.event_logged",
117 event_type=event_type.value,
118 success=success,
119 admin_user_id=admin_user_id,
120 )
121 except Exception: # noqa: BLE001 — audit failures must never propagate; the store may be unavailable for any reason (DB down, serialisation issue, schema not yet created) and we must not interrupt the auth flow that triggered this log call.
122 logger.warning(
123 "audit.log_failed",
124 event_type=(
125 event_type.value
126 if hasattr(event_type, "value")
127 else str(event_type)
128 ),
129 )
131 async def record(
132 self,
133 action: str,
134 actor_id: str,
135 resource_type: str,
136 resource_id: str,
137 outcome: str,
138 severity: AuditEventSeverity = AuditEventSeverity.MEDIUM,
139 **metadata: Any,
140 ) -> None:
141 """Record a single audit event using the framework-wide contract.
143 Maps the framework-wide ``AuditLogProtocol.record()`` interface to the
144 admin-specific ``log_event()`` method. Never raises — audit failures
145 are swallowed so that an audit store outage cannot interrupt flows.
147 Args:
148 action: Dot-notation action identifier (e.g. ``auth.login``).
149 actor_id: ID of the user or service performing the action.
150 resource_type: Type of resource affected.
151 resource_id: ID of the resource affected.
152 outcome: ``"success"`` or ``"failure"``.
153 severity: Severity level (defaults to MEDIUM).
154 **metadata: Additional key-value context forwarded to ``log_event``.
155 """
156 try:
157 event_type = AdminSecurityEventType(action)
158 except ValueError:
159 event_type = AdminSecurityEventType.SUSPICIOUS_ACTIVITY
161 meta: dict[str, Any] = {
162 "resource_type": resource_type,
163 "resource_id": resource_id,
164 "severity": severity,
165 **metadata,
166 }
167 await self.log_event(
168 event_type=event_type,
169 ip_address=str(metadata.get("ip_address", "")),
170 user_agent=str(metadata.get("user_agent", "")),
171 success=outcome == "success",
172 admin_user_id=actor_id,
173 metadata=meta,
174 )
176 async def log(self, entry: AuditEntry) -> None:
177 """Implement AuditLoggerProtocol.log — forward entry to the admin store.
179 Converts the unified AuditEntry to a security event and persists it.
180 Never raises.
182 Args:
183 entry: The canonical audit entry to record.
184 """
185 try:
186 event_type_str = entry.action
187 try:
188 event_type = AdminSecurityEventType(event_type_str)
189 except ValueError:
190 event_type = AdminSecurityEventType.SUSPICIOUS_ACTIVITY
192 meta: dict[str, Any] = {
193 "resource_type": entry.resource_type,
194 "resource_id": entry.resource_id,
195 "severity": entry.severity,
196 **(entry.metadata or {}),
197 }
198 await self.log_event(
199 event_type=event_type,
200 ip_address=str(meta.get("ip_address", "")),
201 user_agent=str(meta.get("user_agent", "")),
202 success=entry.outcome == "success",
203 admin_user_id=entry.actor_id or None,
204 metadata=meta,
205 )
206 except Exception: # noqa: BLE001
207 logger.warning("audit.log_failed", action=entry.action)
209 async def query(self, query: AuditQuery) -> list[AuditEntry]:
210 """Implement AuditLoggerProtocol.query — returns empty list (admin store uses get_recent_events).
212 Admin security events are stored in a separate schema from generic audit
213 entries. This method returns an empty list; use ``get_recent_events``
214 for admin-specific queries.
216 Args:
217 query: Filter criteria (not applied to the admin-specific store).
219 Returns:
220 Always an empty list for this implementation.
221 """
222 return []
224 async def get_recent_events(
225 self,
226 admin_user_id: str | None = None,
227 since_seconds: int = 3600,
228 limit: int = 50,
229 ) -> list[AdminSecurityEvent]:
230 """Retrieve recent security events. Returns empty list on error.
232 Swallows all exceptions for consistency with ``log_event`` — the
233 audit log is a best-effort observability layer and a read failure
234 (e.g. temporary DB outage) must not crash admin dashboards or API
235 endpoints that display security history.
237 Args:
238 admin_user_id: Filter to specific user (None = all users).
239 since_seconds: Look-back window in seconds.
240 limit: Maximum results to return.
242 Returns:
243 List of security events, newest first. Empty list on any error.
244 """
245 try:
246 return await self._store.query_recent(
247 admin_user_id=admin_user_id,
248 since_seconds=since_seconds,
249 limit=limit,
250 )
251 except Exception: # noqa: BLE001 — a read failure from the audit store (transient DB error, schema missing, etc.) must degrade gracefully; callers receive an empty list rather than a 500 error.
252 logger.warning("audit.query_failed", admin_user_id=admin_user_id)
253 return []
256# Verify structural subtyping at import time: AdminAuditLogService must
257# satisfy AdminAuditLogServiceProtocol without explicit inheritance.
258_: AdminAuditLogServiceProtocol = AdminAuditLogService.__new__(AdminAuditLogService)
260__all__ = ["AdminAuditLogService"]