Coverage for src / lexigram / admin / auth / protocols.py: 100%

1"""Admin authentication service protocols. 

2 

3All protocols remain in lexigram-admin (not lexigram-contracts) because they 

4are admin-specific and not consumed by other extension packages. 

5 

6``AdminAuditLogServiceProtocol`` extends the framework-wide 

7``AuditLoggerProtocol`` from ``lexigram.contracts.audit`` so that admin audit 

8implementations satisfy the cross-package contract. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

14 

15from lexigram.contracts.audit import AuditLoggerProtocol 

16 

17if TYPE_CHECKING: 

18 from lexigram.admin.auth.errors import AdminAuthError 

19 from lexigram.admin.auth.types import ( 

20 AdminAuthResult, 

21 AdminLockoutInfo, 

22 AdminLoginAttempt, 

23 AdminPasswordValidationResult, 

24 AdminSecurityEvent, 

25 AdminSecurityEventType, 

26 ) 

27 from lexigram.result import Result 

28 

29 

30@runtime_checkable 

31class AdminAuthServiceProtocol(Protocol): 

32 """Main authentication orchestration service protocol. 

33 

34 Coordinates credential verification, rate limiting, lockout checks, 

35 session issuance, and audit logging. 

36 """ 

37 

38 async def authenticate( 

39 self, 

40 email: str, 

41 password: str, 

42 ip_address: str, 

43 user_agent: str, 

44 ) -> Result[AdminAuthResult, AdminAuthError]: 

45 """Authenticate an admin user with full security pipeline. 

46 

47 Args: 

48 email: Admin user email. 

49 password: Plain-text password. 

50 ip_address: Client IP for rate limiting. 

51 user_agent: Client user agent for audit. 

52 

53 Returns: 

54 Ok(AdminAuthResult) with session details on success. 

55 Err with specific AdminAuthError subclass on failure. 

56 """ 

57 ... 

58 

59 async def invalidate_session(self, session_id: str) -> None: 

60 """Invalidate a session (logout). 

61 

62 Args: 

63 session_id: Session identifier to revoke. 

64 """ 

65 ... 

66 

67 async def invalidate_all_user_sessions(self, user_id: str) -> None: 

68 """Revoke all active sessions for a user (e.g., after password change). 

69 

70 Args: 

71 user_id: Admin user UUID whose sessions to revoke. 

72 """ 

73 ... 

74 

75 

76@runtime_checkable 

77class AdminLoginAttemptStoreProtocol(Protocol): 

78 """Persistence protocol for login attempt records.""" 

79 

80 async def ensure_schema(self) -> None: 

81 """Create the admin_login_attempts table if it does not exist.""" 

82 ... 

83 

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

85 """Persist a login attempt record. 

86 

87 Args: 

88 attempt: The attempt to store. 

89 """ 

90 ... 

91 

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

93 """Count failed attempts for email within the given window. 

94 

95 Args: 

96 email: Email address to query. 

97 since_seconds: Look-back window in seconds. 

98 

99 Returns: 

100 Number of failed attempts. 

101 """ 

102 ... 

103 

104 async def count_recent_failures_by_ip( 

105 self, ip_address: str, since_seconds: int 

106 ) -> int: 

107 """Count failed attempts from an IP within the given window. 

108 

109 Args: 

110 ip_address: IP address to query. 

111 since_seconds: Look-back window in seconds. 

112 

113 Returns: 

114 Number of failed attempts. 

115 """ 

116 ... 

117 

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

119 """Clear failure records for email (called on successful login). 

120 

121 Args: 

122 email: Email to clear. 

123 """ 

124 ... 

125 

126 

127@runtime_checkable 

128class AdminAccountLockoutStoreProtocol(Protocol): 

129 """Persistence protocol for account lockout records.""" 

130 

131 async def ensure_schema(self) -> None: 

132 """Create the admin_account_lockouts table if it does not exist.""" 

133 ... 

134 

135 async def get_active_lockout(self, email: str) -> AdminLockoutInfo | None: 

136 """Get active lockout for email, or None if not locked. 

137 

138 Args: 

139 email: Email to check. 

140 

141 Returns: 

142 AdminLockoutInfo if active lockout exists, None otherwise. 

143 """ 

144 ... 

145 

146 async def create_lockout( 

147 self, 

148 email: str, 

149 consecutive_failures: int, 

150 unlock_at: Any | None, 

151 is_permanent: bool, 

152 ) -> None: 

153 """Create or update a lockout record for email. 

154 

155 Args: 

156 email: Email to lock. 

157 consecutive_failures: Total consecutive failures. 

158 unlock_at: UTC datetime when lock expires (None if permanent). 

159 is_permanent: Whether this requires manual admin unlock. 

160 """ 

161 ... 

162 

163 async def clear_lockout(self, email: str) -> None: 

164 """Remove active lockout for email (on successful login or admin unlock). 

165 

166 Args: 

167 email: Email to unlock. 

168 """ 

169 ... 

170 

171 

172@runtime_checkable 

173class AdminLoginAttemptServiceProtocol(Protocol): 

174 """Service for IP rate limiting and account lockout enforcement.""" 

175 

176 async def check_ip_rate_limit(self, ip_address: str) -> None: 

177 """Check IP rate limit. Raises RateLimitExceededError if exceeded. 

178 

179 Args: 

180 ip_address: Client IP to check. 

181 

182 Raises: 

183 RateLimitExceededError: When the IP is rate-limited. 

184 """ 

185 ... 

186 

187 async def check_account_lockout(self, email: str) -> None: 

188 """Check account lockout status. Raises AccountLockedError if locked. 

189 

190 Args: 

191 email: Email address to check. 

192 

193 Raises: 

194 AccountLockedError: When the account is locked. 

195 """ 

196 ... 

197 

198 async def record_attempt( 

199 self, 

200 email: str, 

201 ip_address: str, 

202 user_agent: str, 

203 success: bool, 

204 failure_reason: str | None = None, 

205 ) -> None: 

206 """Record a login attempt and update lockout state on failure. 

207 

208 Args: 

209 email: Email that attempted login. 

210 ip_address: Client IP. 

211 user_agent: Client user agent. 

212 success: Whether the attempt succeeded. 

213 failure_reason: Short failure code when success=False. 

214 """ 

215 ... 

216 

217 async def clear_lockout(self, email: str) -> None: 

218 """Clear lockout and failure records on successful login. 

219 

220 Args: 

221 email: Email to clear. 

222 """ 

223 ... 

224 

225 

226@runtime_checkable 

227class AdminAuditLogStoreProtocol(Protocol): 

228 """Persistence protocol for security audit log entries.""" 

229 

230 async def ensure_schema(self) -> None: 

231 """Create the admin_security_audit_log table if it does not exist.""" 

232 ... 

233 

234 async def insert(self, event: AdminSecurityEvent) -> None: 

235 """Persist a security event. 

236 

237 Args: 

238 event: Security event to store. 

239 """ 

240 ... 

241 

242 async def query_recent( 

243 self, 

244 admin_user_id: str | None = None, 

245 event_type: AdminSecurityEventType | None = None, 

246 since_seconds: int = 3600, 

247 limit: int = 100, 

248 ) -> list[AdminSecurityEvent]: 

249 """Query recent security events with optional filters. 

250 

251 Args: 

252 admin_user_id: Filter to specific user (None = all users). 

253 event_type: Filter to specific event type (None = all types). 

254 since_seconds: Look-back window in seconds. 

255 limit: Maximum records to return. 

256 

257 Returns: 

258 List of matching security events, newest first. 

259 """ 

260 ... 

261 

262 

263@runtime_checkable 

264class AdminAuditLogServiceProtocol(AuditLoggerProtocol, Protocol): 

265 """Service for recording admin security events. 

266 

267 Extends the framework-wide ``AuditLoggerProtocol`` so that admin audit 

268 implementations satisfy the cross-package contract. Adds admin-specific 

269 methods (``log_event``, ``get_recent_events``) on top of the base 

270 ``log()`` and ``query()`` methods from ``AuditLoggerProtocol``. 

271 

272 Implementations must never raise — audit failures are swallowed so that 

273 an audit store outage cannot interrupt authentication flows. 

274 """ 

275 

276 async def log_event( 

277 self, 

278 event_type: AdminSecurityEventType, 

279 ip_address: str, 

280 user_agent: str, 

281 success: bool, 

282 admin_user_id: str | None = None, 

283 metadata: dict[str, Any] | None = None, 

284 ) -> None: 

285 """Record a security event. Implementation must never raise. 

286 

287 Args: 

288 event_type: Type of security event. 

289 ip_address: Client IP. 

290 user_agent: Client user agent. 

291 success: Whether the operation succeeded. 

292 admin_user_id: Associated admin user (None for pre-auth events). 

293 metadata: Optional structured context. 

294 """ 

295 ... 

296 

297 async def get_recent_events( 

298 self, 

299 admin_user_id: str | None = None, 

300 since_seconds: int = 3600, 

301 limit: int = 50, 

302 ) -> list[AdminSecurityEvent]: 

303 """Retrieve recent security events for display. 

304 

305 Args: 

306 admin_user_id: Filter to specific user. 

307 since_seconds: Look-back window. 

308 limit: Maximum results. 

309 

310 Returns: 

311 List of security events, newest first. 

312 """ 

313 ... 

314 

315 

316@runtime_checkable 

317class AdminPasswordPolicyServiceProtocol(Protocol): 

318 """Password policy validation service.""" 

319 

320 def validate( 

321 self, 

322 password: str, 

323 email: str | None = None, 

324 ) -> AdminPasswordValidationResult: 

325 """Validate a password against all configured policy rules. 

326 

327 Returns ALL violations, not just the first one. 

328 

329 Args: 

330 password: Plain-text password to validate. 

331 email: Optional email — used to check if password contains it. 

332 

333 Returns: 

334 AdminPasswordValidationResult with is_valid and full violations list. 

335 """ 

336 ... 

337 

338 

339@runtime_checkable 

340class AdminCsrfServiceProtocol(Protocol): 

341 """CSRF token generation and validation service.""" 

342 

343 def generate_token(self, session_id: str) -> str: 

344 """Generate a CSRF token scoped to the given session. 

345 

346 Token format: base64url(timestamp:nonce:hmac_signature) 

347 

348 Args: 

349 session_id: Session ID to scope the token to. 

350 

351 Returns: 

352 CSRF token string. 

353 """ 

354 ... 

355 

356 def validate_token(self, session_id: str, token: str) -> bool: 

357 """Validate a CSRF token against the session. 

358 

359 Uses hmac.compare_digest for timing-safe comparison. 

360 

361 Args: 

362 session_id: Session ID the token was generated for. 

363 token: Token to validate. 

364 

365 Returns: 

366 True if valid and not expired, False otherwise. 

367 """ 

368 ... 

369 

370 

371@runtime_checkable 

372class AdminSessionServiceProtocol(Protocol): 

373 """Admin session lifecycle management service.""" 

374 

375 async def create_session( 

376 self, 

377 user_id: str, 

378 email: str, 

379 roles: list[str], 

380 ip_address: str, 

381 user_agent: str, 

382 ) -> str: 

383 """Create a new session and return the session ID. 

384 

385 Args: 

386 user_id: Admin user UUID. 

387 email: Admin user email. 

388 roles: User's roles. 

389 ip_address: Client IP. 

390 user_agent: Client user agent. 

391 

392 Returns: 

393 New session identifier (secrets.token_urlsafe(32)). 

394 """ 

395 ... 

396 

397 async def get_session(self, session_id: str) -> dict[str, Any] | None: 

398 """Retrieve session data if valid (not expired, not revoked). 

399 

400 Checks both idle timeout and absolute expiry. 

401 

402 Args: 

403 session_id: Session to retrieve. 

404 

405 Returns: 

406 Session data dict or None if not found/expired. 

407 """ 

408 ... 

409 

410 async def touch_session(self, session_id: str) -> None: 

411 """Update session last-active timestamp. 

412 

413 Args: 

414 session_id: Session to touch. 

415 """ 

416 ... 

417 

418 async def revoke_session(self, session_id: str) -> None: 

419 """Revoke a single session. 

420 

421 Args: 

422 session_id: Session to revoke. 

423 """ 

424 ... 

425 

426 async def revoke_all_user_sessions(self, user_id: str) -> None: 

427 """Revoke all sessions for a user. 

428 

429 Args: 

430 user_id: Admin user UUID. 

431 """ 

432 ... 

433 

434 

435__all__ = [ 

436 "AdminAccountLockoutStoreProtocol", 

437 "AdminAuditLogServiceProtocol", 

438 "AdminAuditLogStoreProtocol", 

439 "AdminAuthServiceProtocol", 

440 "AdminCsrfServiceProtocol", 

441 "AdminLoginAttemptServiceProtocol", 

442 "AdminLoginAttemptStoreProtocol", 

443 "AdminPasswordPolicyServiceProtocol", 

444 "AdminSessionServiceProtocol", 

445]