Coverage for src/lexigram/admin/auth/services/auth_service.py: 82%
113 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Admin authentication orchestrator service.
3Coordinates the complete login flow: IP rate limiting, account lockout,
4credential verification, session issuance, and audit logging.
5"""
7from __future__ import annotations
9from datetime import UTC, datetime, timedelta
10from typing import Any
12from lexigram.admin.auth.errors import (
13 AccountLockedError,
14 AdminAuthError,
15 InvalidCredentialsError,
16 MfaNotEnabledError,
17 MfaVerificationFailedError,
18 RateLimitExceededError,
19)
20from lexigram.admin.auth.protocols import (
21 AdminAuditLogServiceProtocol,
22 AdminEmailOtpServiceProtocol,
23 AdminEmailVerificationServiceProtocol,
24 AdminLoginAttemptServiceProtocol,
25 AdminMfaServiceProtocol,
26 AdminSessionServiceProtocol,
27)
28from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
29from lexigram.admin.auth.types import AdminAuthResult, AdminSecurityEventType
30from lexigram.di.decorators import inject
31from lexigram.logging import get_logger
32from lexigram.result import Err, Ok, Result
34logger = get_logger(__name__)
37@inject
38class AdminAuthService:
39 """Top-level authentication orchestrator for the admin panel.
41 Coordinates the full login pipeline in strict order:
43 1. IP rate-limit check — blocks brute-force by origin.
44 2. Account lockout check — blocks repeated per-account failures.
45 3. Credential verification — validates email/password.
46 3b. Email verification gate — blocks login until the email is
47 verified (config-toggleable enforcement).
48 3c. Second-factor challenge — TOTP or email OTP per config; returns a
49 result with ``mfa_required=True`` (no session created).
50 4. Attempt recording + lockout clearance on success.
51 5. Session creation.
52 6. Audit logging.
54 All audit calls use ``AdminAuditLogServiceProtocol``, whose implementations
55 are guaranteed never to raise; audit failures are silently absorbed so they
56 never interrupt the authentication response.
58 Args:
59 user_store: Admin user persistence and credential verification.
60 attempt_service: IP rate limiting and account lockout enforcement.
61 audit_service: Security event recording (fire-and-forget).
62 session_service: Session lifecycle management.
63 mfa_service: Optional TOTP 2FA challenge service (None disables 2FA).
64 email_verification_service: Optional email verification gate service.
65 email_otp_service: Optional email OTP factor service.
66 mfa_factor: Second factor selected at login (``"totp"`` or
67 ``"email"``; default ``"totp"``).
68 session_lifetime: Absolute session TTL in seconds (default 86400 = 24h).
69 """
71 def __init__(
72 self,
73 user_store: AdminUserStoreProtocol,
74 attempt_service: AdminLoginAttemptServiceProtocol,
75 audit_service: AdminAuditLogServiceProtocol,
76 session_service: AdminSessionServiceProtocol,
77 mfa_service: AdminMfaServiceProtocol | None = None,
78 email_verification_service: AdminEmailVerificationServiceProtocol | None = None,
79 email_otp_service: AdminEmailOtpServiceProtocol | None = None,
80 mfa_factor: str = "totp",
81 session_lifetime: int = 86400,
82 ) -> None:
83 self._user_store = user_store
84 self._attempt_service = attempt_service
85 self._audit_service = audit_service
86 self._session_service = session_service
87 self._mfa_service = mfa_service
88 self._email_verification_service = email_verification_service
89 self._email_otp_service = email_otp_service
90 self._mfa_factor = mfa_factor
91 self._session_lifetime = session_lifetime
93 # ------------------------------------------------------------------
94 # Public API
95 # ------------------------------------------------------------------
97 async def authenticate(
98 self,
99 email: str,
100 password: str,
101 ip_address: str,
102 user_agent: str,
103 ) -> Result[AdminAuthResult, AdminAuthError]:
104 """Authenticate an admin user through the full security pipeline.
106 Steps executed in order:
108 1. ``check_ip_rate_limit`` — raises ``RateLimitExceededError`` if the
109 origin IP has exceeded the configured threshold.
110 2. ``check_account_lockout`` — raises ``AccountLockedError`` if the
111 account is temporarily or permanently locked.
112 3. ``user_store.authenticate`` — returns ``None`` on invalid credentials.
113 3b. Email verification gate — when enforcement is on and the email is
114 unverified, returns a result with ``email_verification_required=True``
115 (no session created); the caller must run the verify flow.
116 3c. Second-factor challenge — when the configured factor is active,
117 returns a result with ``mfa_required=True`` (no session created);
118 the caller must finish via ``complete_mfa_login``.
119 4. Record success attempt and clear lockout state.
120 5. Create a new session via ``session_service``.
121 6. Emit ``LOGIN_SUCCESS`` audit event.
123 Args:
124 email: Admin user email address.
125 password: Plain-text password to verify.
126 ip_address: Client IP address used for rate limiting and audit.
127 user_agent: Client user-agent string used for audit.
129 Returns:
130 ``Ok(AdminAuthResult)`` containing session details on success.
131 ``Ok(AdminAuthResult)`` with ``mfa_required=True`` (empty
132 ``session_id``) when the user must complete a 2FA challenge.
133 ``Ok(AdminAuthResult)`` with ``email_verification_required=True``
134 (empty ``session_id``) when the email is unverified.
135 ``Err(RateLimitExceededError)`` when the IP is rate-limited.
136 ``Err(AccountLockedError)`` when the account is locked.
137 ``Err(InvalidCredentialsError)`` when credentials are invalid.
138 """
139 # Step 1 — IP rate limit
140 try:
141 await self._attempt_service.check_ip_rate_limit(ip_address)
142 except RateLimitExceededError as exc:
143 await self._attempt_service.record_attempt(
144 email=email,
145 ip_address=ip_address,
146 user_agent=user_agent,
147 success=False,
148 failure_reason="ip_rate_limited",
149 )
150 await self._audit_service.log_event(
151 event_type=AdminSecurityEventType.LOGIN_BLOCKED_IP,
152 ip_address=ip_address,
153 user_agent=user_agent,
154 success=False,
155 metadata={"email": email},
156 )
157 logger.warning(
158 "admin_login_blocked_ip",
159 ip_address=ip_address,
160 email=email,
161 )
162 return Err(exc)
164 # Step 2 — Account lockout
165 try:
166 await self._attempt_service.check_account_lockout(email)
167 except AccountLockedError as exc:
168 await self._audit_service.log_event(
169 event_type=AdminSecurityEventType.LOGIN_BLOCKED_LOCKOUT,
170 ip_address=ip_address,
171 user_agent=user_agent,
172 success=False,
173 metadata={"email": email},
174 )
175 logger.warning(
176 "admin_login_blocked_lockout",
177 email=email,
178 ip_address=ip_address,
179 )
180 return Err(exc)
182 # Step 3 — Credential verification
183 user: Any | None = await self._user_store.authenticate(email, password)
184 if user is None:
185 await self._attempt_service.record_attempt(
186 email=email,
187 ip_address=ip_address,
188 user_agent=user_agent,
189 success=False,
190 failure_reason="invalid_credentials",
191 )
192 await self._audit_service.log_event(
193 event_type=AdminSecurityEventType.LOGIN_FAILURE,
194 ip_address=ip_address,
195 user_agent=user_agent,
196 success=False,
197 metadata={"email": email},
198 )
199 logger.info(
200 "admin_login_failure",
201 email=email,
202 ip_address=ip_address,
203 )
204 return Err(InvalidCredentialsError("Invalid email or password."))
206 # Step 3b — Email verification gate (when enforcement is on)
207 if (
208 self._email_verification_service is not None
209 and await self._email_verification_service.is_required(str(user.user_id))
210 ):
211 await self._audit_service.log_event(
212 event_type=AdminSecurityEventType.EMAIL_VERIFICATION_SENT,
213 ip_address=ip_address,
214 user_agent=user_agent,
215 success=True,
216 admin_user_id=str(user.user_id),
217 metadata={"email": str(user.email)},
218 )
219 roles: list[str] = list(getattr(user, "roles", []) or [])
220 return Ok(
221 AdminAuthResult(
222 session_id="",
223 user_id=str(user.user_id),
224 email=str(user.email),
225 roles=roles,
226 expires_at=datetime.now(UTC)
227 + timedelta(seconds=self._session_lifetime),
228 email_verification_required=True,
229 )
230 )
232 # Step 3c — Second-factor challenge (TOTP or email per config)
233 if self._mfa_factor == "email":
234 if self._email_otp_service is not None:
235 await self._audit_service.log_event(
236 event_type=AdminSecurityEventType.MFA_CHALLENGE_ISSUED,
237 ip_address=ip_address,
238 user_agent=user_agent,
239 success=True,
240 admin_user_id=str(user.user_id),
241 metadata={"email": str(user.email)},
242 )
243 roles = list(getattr(user, "roles", []) or [])
244 return Ok(
245 AdminAuthResult(
246 session_id="",
247 user_id=str(user.user_id),
248 email=str(user.email),
249 roles=roles,
250 expires_at=datetime.now(UTC)
251 + timedelta(seconds=self._session_lifetime),
252 mfa_required=True,
253 )
254 )
255 elif self._mfa_factor == "totp" and self._mfa_service is not None:
256 mfa_enabled = await self._mfa_service.is_enabled(str(user.user_id))
257 if mfa_enabled:
258 await self._audit_service.log_event(
259 event_type=AdminSecurityEventType.MFA_CHALLENGE_ISSUED,
260 ip_address=ip_address,
261 user_agent=user_agent,
262 success=True,
263 admin_user_id=str(user.user_id),
264 metadata={"email": str(user.email)},
265 )
266 roles = list(getattr(user, "roles", []) or [])
267 return Ok(
268 AdminAuthResult(
269 session_id="",
270 user_id=str(user.user_id),
271 email=str(user.email),
272 roles=roles,
273 expires_at=datetime.now(UTC)
274 + timedelta(seconds=self._session_lifetime),
275 mfa_required=True,
276 )
277 )
279 # Step 4 — Record success and clear lockout
280 await self._attempt_service.record_attempt(
281 email=email,
282 ip_address=ip_address,
283 user_agent=user_agent,
284 success=True,
285 )
286 await self._attempt_service.clear_lockout(email)
288 # Step 5 — Create session
289 session_roles: list[str] = list(getattr(user, "roles", []) or [])
290 session_id: str = await self._session_service.create_session(
291 user_id=str(user.user_id),
292 email=str(user.email),
293 roles=session_roles,
294 ip_address=ip_address,
295 user_agent=user_agent,
296 )
297 await self._audit_service.log_event(
298 event_type=AdminSecurityEventType.SESSION_CREATED,
299 ip_address=ip_address,
300 user_agent=user_agent,
301 success=True,
302 admin_user_id=str(user.user_id),
303 metadata={"email": str(user.email), "session_id": session_id},
304 )
306 expires_at: datetime = datetime.now(UTC) + timedelta(
307 seconds=self._session_lifetime
308 )
310 # Step 6 — Audit success
311 await self._audit_service.log_event(
312 event_type=AdminSecurityEventType.LOGIN_SUCCESS,
313 ip_address=ip_address,
314 user_agent=user_agent,
315 success=True,
316 admin_user_id=str(user.user_id),
317 metadata={"email": str(user.email), "session_id": session_id},
318 )
319 logger.info(
320 "admin_login_success",
321 user_id=str(user.user_id),
322 email=str(user.email),
323 ip_address=ip_address,
324 )
326 return Ok(
327 AdminAuthResult(
328 session_id=session_id,
329 user_id=str(user.user_id),
330 email=str(user.email),
331 roles=session_roles,
332 expires_at=expires_at,
333 )
334 )
336 async def complete_mfa_login(
337 self,
338 user_id: str,
339 email: str,
340 roles: list[str],
341 code: str,
342 ip_address: str,
343 user_agent: str,
344 ) -> Result[AdminAuthResult, AdminAuthError]:
345 """Complete a login after a successful second-factor challenge.
347 Called by the 2FA form once the user supplies a valid code. Runs
348 the post-credential pipeline that ``authenticate`` deferred when it
349 returned ``mfa_required=True``: attempt recording, lockout
350 clearance, session creation, and audit logging.
352 The code is verified against the configured factor: TOTP via
353 ``mfa_service`` or email OTP via ``email_otp_service``.
355 Args:
356 user_id: Admin user UUID (from the pending challenge).
357 email: Admin user email (from the pending challenge).
358 roles: Role names for the user (from the pending challenge).
359 code: TOTP code or email OTP code to verify.
360 ip_address: Client IP address used for rate limiting and audit.
361 user_agent: Client user-agent string used for audit.
363 Returns:
364 ``Ok(AdminAuthResult)`` with a real session on success.
365 ``Err(MfaVerificationFailedError)`` when the code is invalid.
366 ``Err(MfaNotEnabledError)`` when the selected factor is
367 unavailable.
368 ``Err(RateLimitExceededError)`` when the IP is rate-limited.
369 ``Err(AccountLockedError)`` when the account is locked.
370 """
371 # Step 0a — IP rate limit (mirrors authenticate()'s Step 1)
372 try:
373 await self._attempt_service.check_ip_rate_limit(ip_address)
374 except RateLimitExceededError as exc:
375 await self._attempt_service.record_attempt(
376 email=email,
377 ip_address=ip_address,
378 user_agent=user_agent,
379 success=False,
380 failure_reason="mfa_ip_rate_limited",
381 )
382 await self._audit_service.log_event(
383 event_type=AdminSecurityEventType.LOGIN_BLOCKED_IP,
384 ip_address=ip_address,
385 user_agent=user_agent,
386 success=False,
387 metadata={"email": email, "stage": "mfa"},
388 )
389 return Err(exc)
391 # Step 0b — Account lockout (mirrors authenticate()'s Step 2)
392 try:
393 await self._attempt_service.check_account_lockout(email)
394 except AccountLockedError as exc:
395 await self._audit_service.log_event(
396 event_type=AdminSecurityEventType.LOGIN_BLOCKED_LOCKOUT,
397 ip_address=ip_address,
398 user_agent=user_agent,
399 success=False,
400 metadata={"email": email, "stage": "mfa"},
401 )
402 return Err(exc)
404 if self._mfa_factor == "email":
405 if self._email_otp_service is None:
406 return Err(
407 MfaNotEnabledError(
408 "Email code authentication is not available for this account."
409 )
410 )
411 verification = await self._email_otp_service.verify_otp(user_id, code)
412 else:
413 if self._mfa_service is None:
414 return Err(
415 MfaNotEnabledError(
416 "Two-factor authentication is not enabled for this account."
417 )
418 )
419 verification = await self._mfa_service.verify_code(user_id, code)
421 # Step 1 — Verify the second-factor code
422 if verification.is_err():
423 await self._audit_service.log_event(
424 event_type=AdminSecurityEventType.MFA_CHALLENGE_FAILED,
425 ip_address=ip_address,
426 user_agent=user_agent,
427 success=False,
428 admin_user_id=user_id,
429 metadata={"email": email},
430 )
431 logger.warning("admin_mfa_code_not_available", user_id=user_id)
432 return Err(verification.unwrap_err())
434 if not verification.unwrap():
435 await self._audit_service.log_event(
436 event_type=AdminSecurityEventType.MFA_CHALLENGE_FAILED,
437 ip_address=ip_address,
438 user_agent=user_agent,
439 success=False,
440 admin_user_id=user_id,
441 metadata={"email": email},
442 )
443 await self._attempt_service.record_attempt(
444 email=email,
445 ip_address=ip_address,
446 user_agent=user_agent,
447 success=False,
448 failure_reason="invalid_mfa_code",
449 )
450 logger.warning("admin_mfa_code_failed", user_id=user_id, email=email)
451 return Err(MfaVerificationFailedError("Invalid verification code."))
453 await self._audit_service.log_event(
454 event_type=AdminSecurityEventType.MFA_VERIFIED,
455 ip_address=ip_address,
456 user_agent=user_agent,
457 success=True,
458 admin_user_id=user_id,
459 metadata={"email": email},
460 )
462 # Step 2 — Record success and clear lockout
463 await self._attempt_service.record_attempt(
464 email=email,
465 ip_address=ip_address,
466 user_agent=user_agent,
467 success=True,
468 )
469 await self._attempt_service.clear_lockout(email)
471 # Step 3 — Create session
472 session_id: str = await self._session_service.create_session(
473 user_id=user_id,
474 email=email,
475 roles=roles,
476 ip_address=ip_address,
477 user_agent=user_agent,
478 )
479 await self._audit_service.log_event(
480 event_type=AdminSecurityEventType.SESSION_CREATED,
481 ip_address=ip_address,
482 user_agent=user_agent,
483 success=True,
484 admin_user_id=user_id,
485 metadata={"email": email, "session_id": session_id},
486 )
488 expires_at: datetime = datetime.now(UTC) + timedelta(
489 seconds=self._session_lifetime
490 )
492 # Step 4 — Audit success
493 await self._audit_service.log_event(
494 event_type=AdminSecurityEventType.LOGIN_SUCCESS,
495 ip_address=ip_address,
496 user_agent=user_agent,
497 success=True,
498 admin_user_id=user_id,
499 metadata={"email": email, "session_id": session_id},
500 )
501 logger.info(
502 "admin_login_success",
503 user_id=user_id,
504 email=email,
505 ip_address=ip_address,
506 )
508 return Ok(
509 AdminAuthResult(
510 session_id=session_id,
511 user_id=user_id,
512 email=email,
513 roles=roles,
514 expires_at=expires_at,
515 )
516 )
518 async def invalidate_session(self, session_id: str) -> None:
519 """Revoke a single session (logout).
521 Revokes the session via ``session_service`` and emits a ``LOGOUT``
522 audit event. Audit failure is absorbed and never propagated.
524 Args:
525 session_id: Identifier of the session to revoke.
526 """
527 await self._session_service.revoke_session(session_id)
528 await self._audit_service.log_event(
529 event_type=AdminSecurityEventType.LOGOUT,
530 ip_address="",
531 user_agent="",
532 success=True,
533 metadata={"session_id": session_id},
534 )
535 logger.info("admin_session_invalidated", session_id=session_id)
537 async def invalidate_all_user_sessions(self, user_id: str) -> None:
538 """Revoke all active sessions for an admin user.
540 Called after a password change or administrative action requiring
541 full session teardown. Emits a ``SESSION_REVOKED`` audit event.
542 Audit failure is absorbed and never propagated.
544 Args:
545 user_id: UUID of the admin user whose sessions to revoke.
546 """
547 await self._session_service.revoke_all_user_sessions(user_id)
548 await self._audit_service.log_event(
549 event_type=AdminSecurityEventType.SESSION_REVOKED,
550 ip_address="",
551 user_agent="",
552 success=True,
553 admin_user_id=user_id,
554 metadata={"reason": "all_sessions_revoked"},
555 )
556 logger.info("admin_all_sessions_invalidated", user_id=user_id)
559__all__ = ["AdminAuthService"]