Coverage for src / lexigram / admin / auth / store / audit_log_sql.py: 31%
85 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1from __future__ import annotations
3"""SQL-backed implementation of AdminAuditLogStoreProtocol.
5This is the *only* module in ``lexigram-admin`` that issues raw SQL against
6the ``admin_security_audit_log`` table. The audit service layer depends on
7the ``AdminAuditLogStoreProtocol`` protocol and never on this class directly.
8"""
10from datetime import datetime
11from typing import Any
12import uuid
14from lexigram.admin.auth.types import AdminSecurityEvent, AdminSecurityEventType
15from lexigram.contracts.data import DatabaseProviderProtocol
16from lexigram.di.decorators import inject
17from lexigram.logging import get_logger
18from lexigram.serialization import dumps_str, loads_str
20logger = get_logger(__name__)
22_TABLE = "admin_security_audit_log"
24_CREATE_TABLE_SQL = f"""
25 CREATE TABLE IF NOT EXISTS {_TABLE} (
26 id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
27 event_type VARCHAR(50) NOT NULL,
28 admin_user_id UUID,
29 ip_address VARCHAR(45) NOT NULL,
30 user_agent TEXT NOT NULL DEFAULT '',
31 success BOOLEAN NOT NULL DEFAULT FALSE,
32 metadata TEXT NOT NULL DEFAULT '{{}}',
33 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
34 )
35"""
37_CREATE_INDEX_CREATED_AT = (
38 f"CREATE INDEX IF NOT EXISTS idx_admin_audit_log_created_at"
39 f" ON {_TABLE}(created_at DESC)"
40)
42_CREATE_INDEX_USER_ID = (
43 f"CREATE INDEX IF NOT EXISTS idx_admin_audit_log_admin_user_id"
44 f" ON {_TABLE}(admin_user_id, created_at DESC) WHERE admin_user_id IS NOT NULL"
45)
47_CREATE_INDEX_EVENT_TYPE = (
48 f"CREATE INDEX IF NOT EXISTS idx_admin_audit_log_event_type"
49 f" ON {_TABLE}(event_type, created_at DESC)"
50)
53@inject
54class AdminAuditLogSqlStore:
55 """SQL implementation of AdminAuditLogStoreProtocol.
57 Stores security audit events in the ``admin_security_audit_log`` table.
58 All operations are fire-tolerant — exceptions are logged but never
59 re-raised from ``insert()`` since audit failure must not block
60 authentication.
62 Implements ``AdminAuditLogStoreProtocol`` via structural subtyping.
63 """
65 def __init__(self, db: DatabaseProviderProtocol) -> None:
66 """Initialize with a resolved database provider.
68 Args:
69 db: Framework database provider that exposes ``execute``,
70 ``execute_query``, and ``execute_insert``.
71 """
72 self._db = db
73 self._initialized = False
75 # ------------------------------------------------------------------
76 # Schema bootstrap (DDL)
77 # ------------------------------------------------------------------
79 async def ensure_schema(self) -> None:
80 """Create the audit log table and indexes if they do not exist.
82 Safe to call multiple times — idempotent after the first successful
83 run. Raises on any unexpected DDL failure so callers surface the
84 problem rather than silently skipping audit persistence.
85 """
86 if self._initialized:
87 return
89 try:
90 await self._db.execute(_CREATE_TABLE_SQL, [])
91 logger.info("✅ %s table ensured", _TABLE)
93 for index_sql in (
94 _CREATE_INDEX_CREATED_AT,
95 _CREATE_INDEX_USER_ID,
96 _CREATE_INDEX_EVENT_TYPE,
97 ):
98 try:
99 await self._db.execute(index_sql, [])
100 except (RuntimeError, ValueError, OSError) as exc:
101 logger.debug("Index creation skipped: %s", exc)
103 logger.info("✅ %s indexes ensured", _TABLE)
104 self._initialized = True
106 except Exception: # noqa: BLE001 — schema DDL may raise DB-specific errors; log and propagate
107 logger.exception("Failed to initialise %s schema", _TABLE)
108 raise
110 # ------------------------------------------------------------------
111 # AdminAuditLogStoreProtocol implementation
112 # ------------------------------------------------------------------
114 async def insert(self, event: AdminSecurityEvent) -> None:
115 """Persist a security audit event.
117 This method **never raises** — exceptions are caught and logged at
118 warning level. Audit failure must never block the authentication flow.
120 Args:
121 event: Security event to store.
122 """
123 await self.ensure_schema()
125 try:
126 payload = {
127 "id": uuid.UUID(event.id) if _is_uuid_string(event.id) else event.id,
128 "event_type": event.event_type.value,
129 "admin_user_id": (
130 uuid.UUID(event.admin_user_id)
131 if event.admin_user_id and _is_uuid_string(event.admin_user_id)
132 else event.admin_user_id
133 ),
134 "ip_address": event.ip_address,
135 "user_agent": event.user_agent,
136 "success": event.success,
137 "metadata": dumps_str(event.metadata),
138 "created_at": event.created_at,
139 }
140 await self._db.execute_insert(_TABLE, payload)
141 logger.debug(
142 "audit_log.inserted",
143 event_id=event.id,
144 event_type=event.event_type.value,
145 admin_user_id=event.admin_user_id,
146 )
148 except Exception as exc: # noqa: BLE001 — audit insert must never surface errors to callers
149 logger.warning(
150 "audit_log.insert_failed",
151 event_id=event.id,
152 event_type=event.event_type.value,
153 error=str(exc),
154 )
156 async def query_recent(
157 self,
158 admin_user_id: str | None = None,
159 event_type: AdminSecurityEventType | None = None,
160 since_seconds: int = 3600,
161 limit: int = 100,
162 ) -> list[AdminSecurityEvent]:
163 """Query recent security events with optional filters.
165 Args:
166 admin_user_id: Filter to a specific user (``None`` = all users).
167 event_type: Filter to a specific event type (``None`` = all types).
168 since_seconds: Look-back window in seconds.
169 limit: Maximum number of records to return.
171 Returns:
172 List of ``AdminSecurityEvent`` instances ordered newest first.
173 """
174 await self.ensure_schema()
176 # ``since_seconds`` is always an int — safe to interpolate directly.
177 conditions = [f"created_at > NOW() - INTERVAL '{int(since_seconds)} seconds'"]
178 params: list[Any] = []
180 if admin_user_id is not None:
181 params.append(
182 uuid.UUID(admin_user_id)
183 if _is_uuid_string(admin_user_id)
184 else admin_user_id
185 )
186 conditions.append("admin_user_id = ?")
188 if event_type is not None:
189 params.append(event_type.value)
190 conditions.append("event_type = ?")
192 where_clause = " AND ".join(conditions)
193 sql = (
194 f"SELECT id, event_type, admin_user_id, ip_address, user_agent,"
195 f" success, metadata, created_at"
196 f" FROM {_TABLE}"
197 f" WHERE {where_clause}"
198 f" ORDER BY created_at DESC"
199 f" LIMIT {int(limit)}"
200 )
202 try:
203 result = await self._db.execute_query(sql, params)
204 rows = _extract_rows(result)
205 return [_row_to_event(row) for row in rows]
207 except Exception as exc: # noqa: BLE001 — query failures must not crash callers; return empty list
208 logger.warning(
209 "audit_log.query_failed",
210 since_seconds=since_seconds,
211 admin_user_id=admin_user_id,
212 event_type=event_type.value if event_type else None,
213 error=str(exc),
214 )
215 return []
218# ------------------------------------------------------------------
219# Module-private helpers
220# ------------------------------------------------------------------
223def _is_uuid_string(value: str) -> bool:
224 """Return ``True`` when *value* is a well-formed UUID string.
226 Args:
227 value: String to test.
229 Returns:
230 ``True`` if *value* parses as a UUID, ``False`` otherwise.
231 """
232 try:
233 uuid.UUID(value)
234 return True
235 except ValueError:
236 return False
239def _extract_rows(result: Any) -> list[Any]:
240 """Normalise heterogeneous query result shapes into a plain list.
242 Args:
243 result: Raw return value from ``DatabaseProviderProtocol.execute_query``.
245 Returns:
246 Flat list of row-like objects (dicts or asyncpg ``Record`` instances).
247 """
248 if hasattr(result, "rows"):
249 return list(result.rows)
250 if isinstance(result, list):
251 return result
252 return []
255def _row_to_event(row: Any) -> AdminSecurityEvent:
256 """Construct an ``AdminSecurityEvent`` from a raw database row.
258 Args:
259 row: Row-like mapping returned by the database driver.
261 Returns:
262 Fully populated ``AdminSecurityEvent`` instance.
263 """
264 raw_user_id = row["admin_user_id"]
265 raw_metadata = row["metadata"]
266 raw_created_at = row["created_at"]
268 # Coerce admin_user_id: UUID objects → str, None stays None.
269 admin_user_id: str | None = str(raw_user_id) if raw_user_id is not None else None
271 # Deserialise metadata stored as a JSON string.
272 metadata: dict[str, str | int | bool | None] = loads_str(raw_metadata or "{}")
274 # asyncpg returns timezone-aware datetimes; ensure we always have one.
275 created_at: datetime = (
276 raw_created_at
277 if isinstance(raw_created_at, datetime)
278 else datetime.fromisoformat(str(raw_created_at))
279 )
281 return AdminSecurityEvent(
282 id=str(row["id"]),
283 event_type=AdminSecurityEventType(row["event_type"]),
284 admin_user_id=admin_user_id,
285 ip_address=row["ip_address"],
286 user_agent=row["user_agent"],
287 success=bool(row["success"]),
288 metadata=metadata,
289 created_at=created_at,
290 )
293__all__ = ["AdminAuditLogSqlStore"]