Coverage for src/lexigram/admin/auth/store/audit_log_sql.py: 0%
89 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +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.admin.sql_dialect import is_postgres, since_expr
16from lexigram.contracts.data import DatabaseProviderProtocol
17from lexigram.di.decorators import inject
18from lexigram.logging import get_logger
19from lexigram.serialization import dumps_str, loads_str
21logger = get_logger(__name__)
23_TABLE = "admin_security_audit_log"
25_CREATE_TABLE_SQL_POSTGRES = f"""
26 CREATE TABLE IF NOT EXISTS {_TABLE} (
27 id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
28 event_type VARCHAR(50) NOT NULL,
29 admin_user_id UUID,
30 ip_address VARCHAR(45) NOT NULL,
31 user_agent TEXT NOT NULL DEFAULT '',
32 success BOOLEAN NOT NULL DEFAULT FALSE,
33 metadata TEXT NOT NULL DEFAULT '{{}}',
34 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
35 )
36"""
38_CREATE_TABLE_SQL_SQLITE = f"""
39 CREATE TABLE IF NOT EXISTS {_TABLE} (
40 id TEXT PRIMARY KEY,
41 event_type VARCHAR(50) NOT NULL,
42 admin_user_id TEXT,
43 ip_address VARCHAR(45) NOT NULL,
44 user_agent TEXT NOT NULL DEFAULT '',
45 success BOOLEAN NOT NULL DEFAULT FALSE,
46 metadata TEXT NOT NULL DEFAULT '{{}}',
47 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
48 )
49"""
51_CREATE_INDEX_CREATED_AT = (
52 f"CREATE INDEX IF NOT EXISTS idx_admin_audit_log_created_at"
53 f" ON {_TABLE}(created_at DESC)"
54)
56_CREATE_INDEX_USER_ID = (
57 f"CREATE INDEX IF NOT EXISTS idx_admin_audit_log_admin_user_id"
58 f" ON {_TABLE}(admin_user_id, created_at DESC) WHERE admin_user_id IS NOT NULL"
59)
61_CREATE_INDEX_EVENT_TYPE = (
62 f"CREATE INDEX IF NOT EXISTS idx_admin_audit_log_event_type"
63 f" ON {_TABLE}(event_type, created_at DESC)"
64)
67@inject
68class AdminAuditLogSqlStore:
69 """SQL implementation of AdminAuditLogStoreProtocol.
71 Stores security audit events in the ``admin_security_audit_log`` table.
72 All operations are fire-tolerant — exceptions are logged but never
73 re-raised from ``insert()`` since audit failure must not block
74 authentication.
76 Implements ``AdminAuditLogStoreProtocol`` via structural subtyping.
77 """
79 def __init__(self, db: DatabaseProviderProtocol) -> None:
80 """Initialize with a resolved database provider.
82 Args:
83 db: Framework database provider that exposes ``execute``,
84 ``execute_query``, and ``execute_insert``.
85 """
86 self._db = db
87 self._initialized = False
89 # ------------------------------------------------------------------
90 # Schema bootstrap (DDL)
91 # ------------------------------------------------------------------
93 async def ensure_schema(self) -> None:
94 """Create the audit log table and indexes if they do not exist.
96 Safe to call multiple times — idempotent after the first successful
97 run. Raises on any unexpected DDL failure so callers surface the
98 problem rather than silently skipping audit persistence.
99 """
100 if self._initialized:
101 return
103 try:
104 create_table_sql = (
105 _CREATE_TABLE_SQL_POSTGRES
106 if is_postgres(self._db)
107 else _CREATE_TABLE_SQL_SQLITE
108 )
109 await self._db.execute(create_table_sql, [])
110 logger.info("✅ %s table ensured", _TABLE)
112 for index_sql in (
113 _CREATE_INDEX_CREATED_AT,
114 _CREATE_INDEX_USER_ID,
115 _CREATE_INDEX_EVENT_TYPE,
116 ):
117 try:
118 await self._db.execute(index_sql, [])
119 except (RuntimeError, ValueError, OSError) as exc:
120 logger.debug("Index creation skipped: %s", exc)
122 logger.info("✅ %s indexes ensured", _TABLE)
123 self._initialized = True
125 except Exception: # noqa: BLE001 — schema DDL may raise DB-specific errors; log and propagate
126 logger.exception("Failed to initialise %s schema", _TABLE)
127 raise
129 # ------------------------------------------------------------------
130 # AdminAuditLogStoreProtocol implementation
131 # ------------------------------------------------------------------
133 async def insert(self, event: AdminSecurityEvent) -> None:
134 """Persist a security audit event.
136 This method **never raises** — exceptions are caught and logged at
137 warning level. Audit failure must never block the authentication flow.
139 Args:
140 event: Security event to store.
141 """
142 await self.ensure_schema()
144 try:
145 use_uuid_ids = is_postgres(self._db)
146 payload = {
147 "id": (
148 uuid.UUID(event.id)
149 if use_uuid_ids and _is_uuid_string(event.id)
150 else event.id
151 ),
152 "event_type": event.event_type.value,
153 "admin_user_id": (
154 uuid.UUID(event.admin_user_id)
155 if use_uuid_ids
156 and event.admin_user_id
157 and _is_uuid_string(event.admin_user_id)
158 else event.admin_user_id
159 ),
160 "ip_address": event.ip_address,
161 "user_agent": event.user_agent,
162 "success": event.success,
163 "metadata": dumps_str(event.metadata),
164 "created_at": event.created_at,
165 }
166 await self._db.execute_insert(_TABLE, payload)
167 logger.debug(
168 "audit_log.inserted",
169 event_id=event.id,
170 event_type=event.event_type.value,
171 admin_user_id=event.admin_user_id,
172 )
174 except Exception as exc: # noqa: BLE001 — audit insert must never surface errors to callers
175 logger.warning(
176 "audit_log.insert_failed",
177 event_id=event.id,
178 event_type=event.event_type.value,
179 error=str(exc),
180 )
182 async def query_recent(
183 self,
184 admin_user_id: str | None = None,
185 event_type: AdminSecurityEventType | None = None,
186 since_seconds: int = 3600,
187 limit: int = 100,
188 ) -> list[AdminSecurityEvent]:
189 """Query recent security events with optional filters.
191 Args:
192 admin_user_id: Filter to a specific user (``None`` = all users).
193 event_type: Filter to a specific event type (``None`` = all types).
194 since_seconds: Look-back window in seconds.
195 limit: Maximum number of records to return.
197 Returns:
198 List of ``AdminSecurityEvent`` instances ordered newest first.
199 """
200 await self.ensure_schema()
202 # ``since_seconds`` is always an int — safe to interpolate directly.
203 conditions = [f"created_at > {since_expr(self._db, int(since_seconds))}"]
204 params: list[Any] = []
206 if admin_user_id is not None:
207 params.append(
208 uuid.UUID(admin_user_id)
209 if is_postgres(self._db) and _is_uuid_string(admin_user_id)
210 else admin_user_id
211 )
212 conditions.append("admin_user_id = ?")
214 if event_type is not None:
215 params.append(event_type.value)
216 conditions.append("event_type = ?")
218 where_clause = " AND ".join(conditions)
219 sql = (
220 f"SELECT id, event_type, admin_user_id, ip_address, user_agent," # noqa: S608 — table const, where_clause built from fixed conditions, int(limit)
221 f" success, metadata, created_at"
222 f" FROM {_TABLE}"
223 f" WHERE {where_clause}"
224 f" ORDER BY created_at DESC"
225 f" LIMIT {int(limit)}"
226 )
228 try:
229 result = await self._db.execute_query(sql, params)
230 rows = _extract_rows(result)
231 return [_row_to_event(row) for row in rows]
233 except Exception as exc: # noqa: BLE001 — query failures must not crash callers; return empty list
234 logger.warning(
235 "audit_log.query_failed",
236 since_seconds=since_seconds,
237 admin_user_id=admin_user_id,
238 event_type=event_type.value if event_type else None,
239 error=str(exc),
240 )
241 return []
244# ------------------------------------------------------------------
245# Module-private helpers
246# ------------------------------------------------------------------
249def _is_uuid_string(value: str) -> bool:
250 """Return ``True`` when *value* is a well-formed UUID string.
252 Args:
253 value: String to test.
255 Returns:
256 ``True`` if *value* parses as a UUID, ``False`` otherwise.
257 """
258 try:
259 uuid.UUID(value)
260 return True
261 except ValueError:
262 return False
265def _extract_rows(result: Any) -> list[Any]:
266 """Normalise heterogeneous query result shapes into a plain list.
268 Args:
269 result: Raw return value from ``DatabaseProviderProtocol.execute_query``.
271 Returns:
272 Flat list of row-like objects (dicts or asyncpg ``Record`` instances).
273 """
274 if hasattr(result, "rows"):
275 return list(result.rows)
276 if isinstance(result, list):
277 return result
278 return []
281def _row_to_event(row: Any) -> AdminSecurityEvent:
282 """Construct an ``AdminSecurityEvent`` from a raw database row.
284 Args:
285 row: Row-like mapping returned by the database driver.
287 Returns:
288 Fully populated ``AdminSecurityEvent`` instance.
289 """
290 raw_user_id = row["admin_user_id"]
291 raw_metadata = row["metadata"]
292 raw_created_at = row["created_at"]
294 # Coerce admin_user_id: UUID objects → str, None stays None.
295 admin_user_id: str | None = str(raw_user_id) if raw_user_id is not None else None
297 # Deserialise metadata stored as a JSON string.
298 metadata: dict[str, str | int | bool | None] = loads_str(raw_metadata or "{}")
300 # asyncpg returns timezone-aware datetimes; ensure we always have one.
301 created_at: datetime = (
302 raw_created_at
303 if isinstance(raw_created_at, datetime)
304 else datetime.fromisoformat(str(raw_created_at))
305 )
307 return AdminSecurityEvent(
308 id=str(row["id"]),
309 event_type=AdminSecurityEventType(row["event_type"]),
310 admin_user_id=admin_user_id,
311 ip_address=row["ip_address"],
312 user_agent=row["user_agent"],
313 success=bool(row["success"]),
314 metadata=metadata,
315 created_at=created_at,
316 )
319__all__ = ["AdminAuditLogSqlStore"]