Coverage for src / lexigram / admin / auth / services / auth_service.py: 32%
57 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +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 RateLimitExceededError,
17)
18from lexigram.admin.auth.protocols import (
19 AdminAuditLogServiceProtocol,
20 AdminLoginAttemptServiceProtocol,
21 AdminSessionServiceProtocol,
22)
23from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
24from lexigram.admin.auth.types import AdminAuthResult, AdminSecurityEventType
25from lexigram.di.decorators import inject
26from lexigram.logging import get_logger
27from lexigram.result import Err, Ok, Result
29logger = get_logger(__name__)
32@inject
33class AdminAuthService:
34 """Top-level authentication orchestrator for the admin panel.
36 Coordinates the full login pipeline in strict order:
38 1. IP rate-limit check — blocks brute-force by origin.
39 2. Account lockout check — blocks repeated per-account failures.
40 3. Credential verification — validates email/password.
41 4. Attempt recording + lockout clearance on success.
42 5. Session creation.
43 6. Audit logging.
45 All audit calls use ``AdminAuditLogServiceProtocol``, whose implementations
46 are guaranteed never to raise; audit failures are silently absorbed so they
47 never interrupt the authentication response.
49 Args:
50 user_store: Admin user persistence and credential verification.
51 attempt_service: IP rate limiting and account lockout enforcement.
52 audit_service: Security event recording (fire-and-forget).
53 session_service: Session lifecycle management.
54 session_lifetime: Absolute session TTL in seconds (default 86400 = 24h).
55 """
57 def __init__(
58 self,
59 user_store: AdminUserStoreProtocol,
60 attempt_service: AdminLoginAttemptServiceProtocol,
61 audit_service: AdminAuditLogServiceProtocol,
62 session_service: AdminSessionServiceProtocol,
63 session_lifetime: int = 86400,
64 ) -> None:
65 self._user_store = user_store
66 self._attempt_service = attempt_service
67 self._audit_service = audit_service
68 self._session_service = session_service
69 self._session_lifetime = session_lifetime
71 # ------------------------------------------------------------------
72 # Public API
73 # ------------------------------------------------------------------
75 async def authenticate(
76 self,
77 email: str,
78 password: str,
79 ip_address: str,
80 user_agent: str,
81 ) -> Result[AdminAuthResult, AdminAuthError]:
82 """Authenticate an admin user through the full security pipeline.
84 Steps executed in order:
86 1. ``check_ip_rate_limit`` — raises ``RateLimitExceededError`` if the
87 origin IP has exceeded the configured threshold.
88 2. ``check_account_lockout`` — raises ``AccountLockedError`` if the
89 account is temporarily or permanently locked.
90 3. ``user_store.authenticate`` — returns ``None`` on invalid credentials.
91 4. Record success attempt and clear lockout state.
92 5. Create a new session via ``session_service``.
93 6. Emit ``LOGIN_SUCCESS`` audit event.
95 Args:
96 email: Admin user email address.
97 password: Plain-text password to verify.
98 ip_address: Client IP address used for rate limiting and audit.
99 user_agent: Client user-agent string used for audit.
101 Returns:
102 ``Ok(AdminAuthResult)`` containing session details on success.
103 ``Err(RateLimitExceededError)`` when the IP is rate-limited.
104 ``Err(AccountLockedError)`` when the account is locked.
105 ``Err(InvalidCredentialsError)`` when credentials are invalid.
106 """
107 # Step 1 — IP rate limit
108 try:
109 await self._attempt_service.check_ip_rate_limit(ip_address)
110 except RateLimitExceededError as exc:
111 await self._attempt_service.record_attempt(
112 email=email,
113 ip_address=ip_address,
114 user_agent=user_agent,
115 success=False,
116 failure_reason="ip_rate_limited",
117 )
118 await self._audit_service.log_event(
119 event_type=AdminSecurityEventType.LOGIN_BLOCKED_IP,
120 ip_address=ip_address,
121 user_agent=user_agent,
122 success=False,
123 metadata={"email": email},
124 )
125 logger.warning(
126 "admin_login_blocked_ip",
127 ip_address=ip_address,
128 email=email,
129 )
130 return Err(exc)
132 # Step 2 — Account lockout
133 try:
134 await self._attempt_service.check_account_lockout(email)
135 except AccountLockedError as exc:
136 await self._audit_service.log_event(
137 event_type=AdminSecurityEventType.LOGIN_BLOCKED_LOCKOUT,
138 ip_address=ip_address,
139 user_agent=user_agent,
140 success=False,
141 metadata={"email": email},
142 )
143 logger.warning(
144 "admin_login_blocked_lockout",
145 email=email,
146 ip_address=ip_address,
147 )
148 return Err(exc)
150 # Step 3 — Credential verification
151 user: Any | None = await self._user_store.authenticate(email, password)
152 if user is None:
153 await self._attempt_service.record_attempt(
154 email=email,
155 ip_address=ip_address,
156 user_agent=user_agent,
157 success=False,
158 failure_reason="invalid_credentials",
159 )
160 await self._audit_service.log_event(
161 event_type=AdminSecurityEventType.LOGIN_FAILURE,
162 ip_address=ip_address,
163 user_agent=user_agent,
164 success=False,
165 metadata={"email": email},
166 )
167 logger.info(
168 "admin_login_failure",
169 email=email,
170 ip_address=ip_address,
171 )
172 return Err(InvalidCredentialsError("Invalid email or password."))
174 # Step 4 — Record success and clear lockout
175 await self._attempt_service.record_attempt(
176 email=email,
177 ip_address=ip_address,
178 user_agent=user_agent,
179 success=True,
180 )
181 await self._attempt_service.clear_lockout(email)
183 # Step 5 — Create session
184 roles: list[str] = list(getattr(user, "roles", []) or [])
185 session_id: str = await self._session_service.create_session(
186 user_id=str(user.user_id),
187 email=str(user.email),
188 roles=roles,
189 ip_address=ip_address,
190 user_agent=user_agent,
191 )
192 await self._audit_service.log_event(
193 event_type=AdminSecurityEventType.SESSION_CREATED,
194 ip_address=ip_address,
195 user_agent=user_agent,
196 success=True,
197 admin_user_id=str(user.user_id),
198 metadata={"email": str(user.email), "session_id": session_id},
199 )
201 expires_at: datetime = datetime.now(UTC) + timedelta(
202 seconds=self._session_lifetime
203 )
205 # Step 6 — Audit success
206 await self._audit_service.log_event(
207 event_type=AdminSecurityEventType.LOGIN_SUCCESS,
208 ip_address=ip_address,
209 user_agent=user_agent,
210 success=True,
211 admin_user_id=str(user.user_id),
212 metadata={"email": str(user.email), "session_id": session_id},
213 )
214 logger.info(
215 "admin_login_success",
216 user_id=str(user.user_id),
217 email=str(user.email),
218 ip_address=ip_address,
219 )
221 return Ok(
222 AdminAuthResult(
223 session_id=session_id,
224 user_id=str(user.user_id),
225 email=str(user.email),
226 roles=roles,
227 expires_at=expires_at,
228 )
229 )
231 async def invalidate_session(self, session_id: str) -> None:
232 """Revoke a single session (logout).
234 Revokes the session via ``session_service`` and emits a ``LOGOUT``
235 audit event. Audit failure is absorbed and never propagated.
237 Args:
238 session_id: Identifier of the session to revoke.
239 """
240 await self._session_service.revoke_session(session_id)
241 await self._audit_service.log_event(
242 event_type=AdminSecurityEventType.LOGOUT,
243 ip_address="",
244 user_agent="",
245 success=True,
246 metadata={"session_id": session_id},
247 )
248 logger.info("admin_session_invalidated", session_id=session_id)
250 async def invalidate_all_user_sessions(self, user_id: str) -> None:
251 """Revoke all active sessions for an admin user.
253 Called after a password change or administrative action requiring
254 full session teardown. Emits a ``SESSION_REVOKED`` audit event.
255 Audit failure is absorbed and never propagated.
257 Args:
258 user_id: UUID of the admin user whose sessions to revoke.
259 """
260 await self._session_service.revoke_all_user_sessions(user_id)
261 await self._audit_service.log_event(
262 event_type=AdminSecurityEventType.SESSION_REVOKED,
263 ip_address="",
264 user_agent="",
265 success=True,
266 admin_user_id=user_id,
267 metadata={"reason": "all_sessions_revoked"},
268 )
269 logger.info("admin_all_sessions_invalidated", user_id=user_id)
272__all__ = ["AdminAuthService"]