Coverage for src / lexigram / admin / auth / store / login_attempt_sql.py: 32%

1"""SQL-backed implementation of AdminLoginAttemptStoreProtocol. 

2 

3Owns all DDL and DML for the ``admin_login_attempts`` table. The service 

4layer depends only on ``AdminLoginAttemptStoreProtocol`` from 

5``lexigram.admin.auth.protocols`` — never on this class directly. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, Any 

11import uuid 

12 

13from lexigram.contracts.data import DatabaseProviderProtocol 

14from lexigram.di.decorators import inject 

15from lexigram.logging import get_logger 

16 

17if TYPE_CHECKING: 

18 from lexigram.admin.auth.types import AdminLoginAttempt 

19 

20logger = get_logger(__name__) 

21 

22_TABLE = "admin_login_attempts" 

23 

24 

25@inject 

26class AdminLoginAttemptSqlStore: 

27 """SQL-backed store for admin login attempt records. 

28 

29 Implements ``AdminLoginAttemptStoreProtocol`` via structural subtyping. 

30 Manages the ``admin_login_attempts`` table including DDL bootstrap, 

31 failure counting for rate limiting, and attempt persistence. 

32 """ 

33 

34 def __init__(self, db: DatabaseProviderProtocol) -> None: 

35 """Initialise with a resolved database provider. 

36 

37 Args: 

38 db: Framework database provider exposing ``execute`` and 

39 ``execute_query``. 

40 """ 

41 self._db = db 

42 self._initialized = False 

43 

44 # ------------------------------------------------------------------ 

45 # Schema bootstrap (DDL) 

46 # ------------------------------------------------------------------ 

47 

48 async def ensure_schema(self) -> None: 

49 """Create table and indexes if they do not exist. 

50 

51 Safe to call multiple times — idempotent after the first successful 

52 run. Uses ``CREATE TABLE IF NOT EXISTS`` and 

53 ``CREATE INDEX IF NOT EXISTS`` so concurrent callers are safe. 

54 

55 Raises: 

56 Exception: Propagates any unexpected DDL failure so the caller 

57 surfaces the problem rather than silently skipping persistence. 

58 """ 

59 if self._initialized: 

60 return 

61 

62 try: 

63 await self._db.execute( 

64 """ 

65 CREATE TABLE IF NOT EXISTS admin_login_attempts ( 

66 id UUID PRIMARY KEY DEFAULT gen_random_uuid(), 

67 email VARCHAR(255) NOT NULL, 

68 ip_address VARCHAR(45) NOT NULL, 

69 user_agent TEXT NOT NULL DEFAULT '', 

70 success BOOLEAN NOT NULL DEFAULT FALSE, 

71 failure_reason VARCHAR(50), 

72 attempted_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 

73 ) 

74 """, 

75 [], 

76 ) 

77 await self._db.execute( 

78 """ 

79 CREATE INDEX IF NOT EXISTS idx_admin_login_attempts_email_attempted_at 

80 ON admin_login_attempts(email, attempted_at DESC) 

81 """, 

82 [], 

83 ) 

84 await self._db.execute( 

85 """ 

86 CREATE INDEX IF NOT EXISTS idx_admin_login_attempts_ip_attempted_at 

87 ON admin_login_attempts(ip_address, attempted_at DESC) 

88 """, 

89 [], 

90 ) 

91 self._initialized = True 

92 logger.info("✅ %s schema ready", _TABLE) 

93 except Exception as _schema_err: # noqa: BLE001 — DDL may raise DB-specific errors; log and propagate 

94 logger.exception("Failed to initialise %s schema", _TABLE) 

95 raise 

96 

97 # ------------------------------------------------------------------ 

98 # DML 

99 # ------------------------------------------------------------------ 

100 

101 async def insert(self, attempt: AdminLoginAttempt) -> None: 

102 """Insert a login attempt record. 

103 

104 Generates a fresh UUID when ``attempt.id`` is empty so callers may 

105 pass an uninitialised id without failing the NOT NULL constraint. 

106 

107 Args: 

108 attempt: Login attempt data to persist. 

109 """ 

110 await self.ensure_schema() 

111 attempt_id = attempt.id or str(uuid.uuid4()) 

112 sql = ( 

113 "INSERT INTO admin_login_attempts " 

114 "(id, email, ip_address, user_agent, success, failure_reason, attempted_at) " 

115 "VALUES (?, ?, ?, ?, ?, ?, ?)" 

116 ) 

117 await self._db.execute( 

118 sql, 

119 ( 

120 attempt_id, 

121 attempt.email, 

122 attempt.ip_address, 

123 attempt.user_agent, 

124 attempt.success, 

125 attempt.failure_reason, 

126 attempt.attempted_at, 

127 ), 

128 ) 

129 logger.debug( 

130 "login_attempt.inserted", 

131 email=attempt.email, 

132 success=attempt.success, 

133 failure_reason=attempt.failure_reason, 

134 ) 

135 

136 async def count_recent_failures(self, email: str, since_seconds: int) -> int: 

137 """Count failed login attempts for email within the last N seconds. 

138 

139 Args: 

140 email: Email address to query. 

141 since_seconds: Look-back window in seconds. 

142 

143 Returns: 

144 Number of failed attempts within the window. 

145 """ 

146 await self.ensure_schema() 

147 sql = ( 

148 "SELECT COUNT(*) AS count FROM admin_login_attempts " 

149 "WHERE email = ? AND success = FALSE " 

150 f"AND attempted_at > NOW() - INTERVAL '{since_seconds} seconds'" 

151 ) 

152 result = await self._db.execute_query(sql, [email]) 

153 count = self._extract_count(result) 

154 logger.debug( 

155 "login_attempt.count_recent_failures", 

156 email=email, 

157 since_seconds=since_seconds, 

158 count=count, 

159 ) 

160 return count 

161 

162 async def count_recent_failures_by_ip( 

163 self, ip_address: str, since_seconds: int 

164 ) -> int: 

165 """Count failed login attempts from IP within the last N seconds. 

166 

167 Args: 

168 ip_address: Client IP address to query. 

169 since_seconds: Look-back window in seconds. 

170 

171 Returns: 

172 Number of failed attempts within the window. 

173 """ 

174 await self.ensure_schema() 

175 sql = ( 

176 "SELECT COUNT(*) AS count FROM admin_login_attempts " 

177 "WHERE ip_address = ? AND success = FALSE " 

178 f"AND attempted_at > NOW() - INTERVAL '{since_seconds} seconds'" 

179 ) 

180 result = await self._db.execute_query(sql, [ip_address]) 

181 count = self._extract_count(result) 

182 logger.debug( 

183 "login_attempt.count_recent_failures_by_ip", 

184 ip_address=ip_address, 

185 since_seconds=since_seconds, 

186 count=count, 

187 ) 

188 return count 

189 

190 async def clear_failures(self, email: str) -> None: 

191 """Delete all failure records for email. 

192 

193 Called immediately after a successful login so subsequent failure 

194 counts start from zero. 

195 

196 Args: 

197 email: Email address whose failure records to remove. 

198 """ 

199 await self.ensure_schema() 

200 sql = "DELETE FROM admin_login_attempts WHERE email = ? AND success = FALSE" 

201 await self._db.execute(sql, (email,)) 

202 logger.debug("login_attempt.failures_cleared", email=email) 

203 

204 # ------------------------------------------------------------------ 

205 # Internal helpers 

206 # ------------------------------------------------------------------ 

207 

208 @staticmethod 

209 def _extract_count(result: Any) -> int: 

210 """Normalise heterogeneous query result shapes into an integer count. 

211 

212 Args: 

213 result: Raw result returned by ``execute_query``. 

214 

215 Returns: 

216 The integer value of the ``count`` column, or ``0`` when absent. 

217 """ 

218 if hasattr(result, "rows") and result.rows: 

219 return int(result.rows[0].get("count", 0)) 

220 if isinstance(result, list) and result: 

221 row = result[0] 

222 if isinstance(row, dict): 

223 return int(row.get("count", 0)) 

224 return 0 

225 

226 

227__all__ = ["AdminLoginAttemptSqlStore"]