Coverage for src/lexigram/admin/auth/store/lockout_sql.py: 0%
77 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:18 +0800
1"""SQL-backed implementation of AdminAccountLockoutStoreProtocol.
3Owns all DDL and DML for the ``admin_account_lockouts`` table. The service
4layer depends only on ``AdminAccountLockoutStoreProtocol`` from
5``lexigram.admin.auth.protocols`` — never on this class directly.
6"""
8from __future__ import annotations
10from typing import TYPE_CHECKING, Any
11import uuid
13from lexigram.admin.sql_dialect import is_postgres, now_expr
14from lexigram.contracts.data import DatabaseProviderProtocol
15from lexigram.di.decorators import inject
16from lexigram.logging import get_logger
18if TYPE_CHECKING:
19 from datetime import datetime
21 from lexigram.admin.auth.types import AdminLockoutInfo
23from lexigram.admin.auth.types import AdminLockoutStatus
25logger = get_logger(__name__)
27_TABLE = "admin_account_lockouts"
30@inject
31class AdminAccountLockoutSqlStore:
32 """SQL-backed store for admin account lockout records.
34 Implements ``AdminAccountLockoutStoreProtocol`` via structural subtyping.
35 Manages the ``admin_account_lockouts`` table including DDL bootstrap,
36 active-lockout queries, lockout creation, and lockout clearing.
37 """
39 def __init__(self, db: DatabaseProviderProtocol) -> None:
40 """Initialise with a resolved database provider.
42 Args:
43 db: Framework database provider exposing ``execute`` and
44 ``execute_query``.
45 """
46 self._db = db
47 self._initialized = False
49 # ------------------------------------------------------------------
50 # Schema bootstrap (DDL)
51 # ------------------------------------------------------------------
53 async def ensure_schema(self) -> None:
54 """Create table and indexes if they do not exist.
56 Safe to call multiple times — idempotent after the first successful
57 run. Uses ``CREATE TABLE IF NOT EXISTS`` and
58 ``CREATE UNIQUE INDEX IF NOT EXISTS`` so concurrent callers are safe.
60 Raises:
61 Exception: Propagates any unexpected DDL failure so the caller
62 surfaces the problem rather than silently skipping persistence.
63 """
64 if self._initialized:
65 return
67 try:
68 if is_postgres(self._db):
69 create_sql = """
70 CREATE TABLE IF NOT EXISTS admin_account_lockouts (
71 id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
72 email VARCHAR(255) NOT NULL,
73 locked_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
74 unlock_at TIMESTAMPTZ,
75 consecutive_failures INTEGER NOT NULL DEFAULT 0,
76 is_permanent BOOLEAN NOT NULL DEFAULT FALSE,
77 is_active BOOLEAN NOT NULL DEFAULT TRUE,
78 unlocked_at TIMESTAMPTZ,
79 deactivated_at TIMESTAMPTZ
80 )
81 """
82 else:
83 create_sql = """
84 CREATE TABLE IF NOT EXISTS admin_account_lockouts (
85 id TEXT PRIMARY KEY,
86 email VARCHAR(255) NOT NULL,
87 locked_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
88 unlock_at TIMESTAMP,
89 consecutive_failures INTEGER NOT NULL DEFAULT 0,
90 is_permanent BOOLEAN NOT NULL DEFAULT FALSE,
91 is_active BOOLEAN NOT NULL DEFAULT TRUE,
92 unlocked_at TIMESTAMP,
93 deactivated_at TIMESTAMP
94 )
95 """
96 await self._db.execute(create_sql, [])
97 await self._db.execute(
98 """
99 CREATE UNIQUE INDEX IF NOT EXISTS idx_admin_account_lockouts_email_active
100 ON admin_account_lockouts(email) WHERE is_active = TRUE
101 """,
102 [],
103 )
104 await self._db.execute(
105 """
106 CREATE INDEX IF NOT EXISTS idx_admin_account_lockouts_email
107 ON admin_account_lockouts(email, locked_at DESC)
108 """,
109 [],
110 )
111 self._initialized = True
112 logger.info("✅ %s schema ready", _TABLE)
113 except Exception as _schema_err: # noqa: BLE001 — DDL may raise DB-specific errors; log and propagate
114 logger.exception("Failed to initialise %s schema", _TABLE)
115 raise
117 # ------------------------------------------------------------------
118 # DML
119 # ------------------------------------------------------------------
121 async def get_active_lockout(self, email: str) -> AdminLockoutInfo | None:
122 """Get active lockout for email, or ``None`` if the account is not locked.
124 Checks ``is_active = TRUE`` and whether the lockout is still in effect
125 (permanent, or ``unlock_at`` is in the future). Expired temporary
126 lockouts are deactivated transparently before returning ``None``.
128 Args:
129 email: Email address to check.
131 Returns:
132 ``AdminLockoutInfo`` when an active lockout exists, ``None``
133 otherwise.
134 """
135 await self.ensure_schema()
136 sql = (
137 "SELECT id, email, locked_at, unlock_at, consecutive_failures, is_permanent "
138 "FROM admin_account_lockouts "
139 "WHERE email = ? AND is_active = TRUE"
140 )
141 result = await self._db.execute_query(sql, [email])
142 rows = self._extract_rows(result)
143 if not rows:
144 return None
146 row = dict(rows[0])
147 is_permanent: bool = bool(row.get("is_permanent", False))
148 unlock_at: Any = row.get("unlock_at")
150 # A non-permanent lockout is only active while unlock_at is in the future.
151 if not is_permanent and unlock_at is not None:
152 # Ask the DB whether the lockout has expired — avoids timezone issues.
153 expired_sql = (
154 "UPDATE admin_account_lockouts " # noqa: S608 — now_expr yields fixed NOW()/CURRENT_TIMESTAMP, no user data interpolated
155 f"SET is_active = FALSE, deactivated_at = {now_expr(self._db)} "
156 "WHERE email = ? AND is_active = TRUE AND is_permanent = FALSE "
157 f"AND unlock_at <= {now_expr(self._db)}"
158 )
159 await self._db.execute(expired_sql, [email])
161 # Re-check: if the row is gone, the lockout has expired.
162 recheck_result = await self._db.execute_query(sql, [email])
163 recheck_rows = self._extract_rows(recheck_result)
164 if not recheck_rows:
165 logger.debug("lockout.expired_deactivated", email=email)
166 return None
167 row = dict(recheck_rows[0])
168 is_permanent = bool(row.get("is_permanent", False))
169 unlock_at = row.get("unlock_at")
171 status = (
172 AdminLockoutStatus.PERMANENT if is_permanent else AdminLockoutStatus.LOCKED
173 )
175 from lexigram.admin.auth.types import (
176 AdminLockoutInfo, # local import avoids circularity at module level
177 )
179 lockout = AdminLockoutInfo(
180 status=status,
181 consecutive_failures=int(row.get("consecutive_failures", 0)),
182 locked_at=row.get("locked_at"),
183 unlock_at=unlock_at,
184 is_permanent=is_permanent,
185 )
186 logger.debug(
187 "lockout.active_found",
188 email=email,
189 status=status,
190 consecutive_failures=lockout.consecutive_failures,
191 )
192 return lockout
194 async def create_lockout(
195 self,
196 email: str,
197 consecutive_failures: int,
198 unlock_at: datetime | None,
199 is_permanent: bool,
200 ) -> None:
201 """Create or replace the active lockout for email.
203 Deactivates any existing active lockout first so the unique partial
204 index on ``(email) WHERE is_active = TRUE`` is never violated.
206 Args:
207 email: Email address to lock.
208 consecutive_failures: Total consecutive failure count to record.
209 unlock_at: UTC datetime when the lock expires (``None`` if
210 ``is_permanent`` is ``True``).
211 is_permanent: Whether the lock requires manual admin intervention
212 to clear.
213 """
214 await self.ensure_schema()
216 # 1. Deactivate any existing active lockout.
217 deactivate_sql = (
218 "UPDATE admin_account_lockouts " # noqa: S608 — now_expr yields fixed NOW()/CURRENT_TIMESTAMP, no user data interpolated
219 f"SET is_active = FALSE, deactivated_at = {now_expr(self._db)} "
220 "WHERE email = ? AND is_active = TRUE"
221 )
222 await self._db.execute(deactivate_sql, (email,))
224 # 2. Insert the new lockout record.
225 lockout_id = str(uuid.uuid4())
226 insert_sql = (
227 "INSERT INTO admin_account_lockouts "
228 "(id, email, consecutive_failures, unlock_at, is_permanent, is_active) "
229 "VALUES (?, ?, ?, ?, ?, TRUE)"
230 )
231 await self._db.execute(
232 insert_sql,
233 (
234 lockout_id,
235 email,
236 consecutive_failures,
237 unlock_at,
238 is_permanent,
239 ),
240 )
241 logger.info(
242 "lockout.created",
243 email=email,
244 consecutive_failures=consecutive_failures,
245 is_permanent=is_permanent,
246 unlock_at=str(unlock_at) if unlock_at is not None else None,
247 )
249 async def clear_lockout(self, email: str) -> None:
250 """Deactivate any active lockout for email.
252 Called on successful login or explicit admin unlock. Safe to call
253 when no active lockout exists — the UPDATE simply affects zero rows.
255 Args:
256 email: Email address to unlock.
257 """
258 await self.ensure_schema()
259 sql = (
260 "UPDATE admin_account_lockouts " # noqa: S608 — now_expr yields fixed NOW()/CURRENT_TIMESTAMP, no user data interpolated
261 f"SET is_active = FALSE, unlocked_at = {now_expr(self._db)} "
262 "WHERE email = ? AND is_active = TRUE"
263 )
264 await self._db.execute(sql, (email,))
265 logger.debug("lockout.cleared", email=email)
267 # ------------------------------------------------------------------
268 # Internal helpers
269 # ------------------------------------------------------------------
271 @staticmethod
272 def _extract_rows(result: Any) -> list[Any]:
273 """Normalise heterogeneous query result shapes into a plain list.
275 Args:
276 result: Raw result returned by ``execute_query``.
278 Returns:
279 A list of row-like objects (dicts or record proxies).
280 """
281 if hasattr(result, "rows"):
282 return list(result.rows)
283 if isinstance(result, list):
284 return result
285 return []
288__all__ = ["AdminAccountLockoutSqlStore"]