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