Coverage for src/lexigram/auth/authn/services.py: 84%
176 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Authentication services for user login, registration, and token management."""
3from __future__ import annotations
5import asyncio
6from dataclasses import dataclass, field
7from typing import TYPE_CHECKING, Any
9from lexigram import serialization as _json
10from lexigram.auth.authn.security import (
11 DUMMY_PASSWORD_HASH,
12 PasswordHasher,
13 PasswordPolicy,
14)
15from lexigram.auth.events import (
16 UserLockedOut,
17 UserLoggedIn,
18 UserLoginFailed,
19 UserRegistered,
20)
21from lexigram.auth.exceptions import (
22 AccountLockedError,
23 EmailExistsError,
24 InvalidCredentialsError,
25 PasswordPolicyError,
26)
27from lexigram.auth.models.user import User, UserCredentials
28from lexigram.auth.services.activity_tracker import AuthActivityTracker
29from lexigram.contracts.auth.protocols import (
30 LoginAttemptTrackerProtocol,
31 PasswordHasherProtocol,
32)
33from lexigram.di.decorators import inject
34from lexigram.logging import get_logger
35from lexigram.primitives import clock as ambient_clock
37if TYPE_CHECKING:
38 from lexigram.auth.authn.jwt import JWTTokenManager
39 from lexigram.auth.authn.schemas import RegisterRequest
40 from lexigram.auth.models import AuthToken
41 from lexigram.auth.storage.token_store import UserStoreProtocol
42 from lexigram.contracts.auth.exceptions import TokenError
43 from lexigram.contracts.auth.token import VerifiedToken
44 from lexigram.contracts.core import HookRegistryProtocol
45 from lexigram.contracts.events.protocols import EventBusProtocol
46 from lexigram.contracts.infra.cache.protocols import CacheBackendProtocol
47 from lexigram.result import Result
49logger = get_logger(__name__)
52@dataclass
53class LockoutConfig:
54 """Configuration for account lockout on repeated failed login attempts.
56 Attributes:
57 max_failed_attempts: Number of failed attempts within the observation
58 window that triggers a lockout. Defaults to 5.
59 lockout_duration_seconds: Length of the rolling observation window in
60 seconds. The lockout is lifted automatically once all recorded
61 failures fall outside this window. Defaults to 300 (5 minutes).
62 max_attempts: Canonical alias for ``max_failed_attempts``. Both
63 fields default to 5 and are kept in sync by ``__post_init__``.
65 Note:
66 For distributed deployments pass a ``CacheBackendProtocol`` to
67 :class:`LoginAttemptTracker` instead of relying on the in-process
68 dict managed by ``AuthenticationService``.
69 """
71 max_failed_attempts: int = 5
72 lockout_duration_seconds: int = 300
73 max_attempts: int = field(default=5)
75 def __post_init__(self) -> None:
76 # max_attempts is the canonical name; max_failed_attempts exists for
77 # backward compatibility. Sync rules:
78 # • If max_attempts was set to a non-default value → it wins.
79 # • Otherwise max_failed_attempts is the source of truth.
80 if self.max_attempts != 5:
81 self.max_failed_attempts = self.max_attempts
82 else:
83 self.max_attempts = self.max_failed_attempts
86class LoginAttemptTracker(LoginAttemptTrackerProtocol):
87 """Tracks failed login attempts and enforces account lockout.
89 Supports both in-process state (no external dependency) and distributed
90 state via an injected :class:`~lexigram.contracts.cache.protocols.CacheBackendProtocol`.
91 When a cache backend is provided all attempt records are stored there so
92 that multiple application instances share the same view — preventing an
93 attacker from bypassing lockout by rotating across instances.
95 Args:
96 max_attempts: Consecutive failures within *lockout_duration_seconds*
97 that trigger a lockout. Defaults to 5.
98 lockout_duration_seconds: Rolling window size **and** cache TTL.
99 Defaults to 900 seconds (15 minutes).
100 cache: Optional cache backend for distributed state. When ``None``
101 an in-process ``dict`` is used instead.
103 Example::
105 tracker = LoginAttemptTracker(max_attempts=5, lockout_duration_seconds=900)
107 await tracker.record_failure("user@example.com")
108 locked = await tracker.is_locked("user@example.com") # False until 5 failures
110 await tracker.clear("user@example.com") # reset on successful login
111 """
113 _CACHE_KEY_PREFIX = "lexigram:auth:attempts:"
115 def __init__(
116 self,
117 max_attempts: int = 5,
118 lockout_duration_seconds: int = 900,
119 cache: CacheBackendProtocol | None = None,
120 ) -> None:
121 self.max_attempts = max_attempts
122 self.lockout_duration_seconds = lockout_duration_seconds
123 self._cache = cache
124 # In-memory fallback: identifier → list of monotonic timestamps
125 self._local: dict[str, list[float]] = {}
127 # ------------------------------------------------------------------
128 # Public API
129 # ------------------------------------------------------------------
131 async def is_locked(self, identifier: str) -> bool:
132 """Return ``True`` if *identifier* has exceeded the failure threshold.
134 Args:
135 identifier: Username, e-mail address, or IP used as the tracking key.
136 """
137 if self._cache is not None:
138 return await self._is_locked_cache(identifier)
139 return self._is_locked_local(identifier)
141 async def record_failure(self, identifier: str) -> None:
142 """Record a failed authentication attempt for *identifier*.
144 Args:
145 identifier: Username, e-mail address, or IP used as the tracking key.
146 """
147 if self._cache is not None:
148 await self._record_failure_cache(identifier)
149 else:
150 self._record_failure_local(identifier)
152 async def clear(self, identifier: str) -> None:
153 """Remove all recorded failures for *identifier* (call on success).
155 Args:
156 identifier: Username, e-mail address, or IP used as the tracking key.
157 """
158 if self._cache is not None:
159 await self._cache.delete(self._CACHE_KEY_PREFIX + identifier)
160 else:
161 self._local.pop(identifier, None)
163 # ------------------------------------------------------------------
164 # Cache-backed helpers
165 # ------------------------------------------------------------------
167 async def _is_locked_cache(self, identifier: str) -> bool:
168 key = self._CACHE_KEY_PREFIX + identifier
169 raw: Any = await self._cache.get(key) # type: ignore[union-attr]
170 if raw is None:
171 return False
172 data: dict[str, Any] = (
173 _json.loads(raw) if isinstance(raw, (str, bytes)) else raw
174 )
175 now = ambient_clock.monotonic()
176 window_start = now - self.lockout_duration_seconds
177 recent = [t for t in data.get("timestamps", []) if t > window_start]
178 return len(recent) >= self.max_attempts
180 async def _record_failure_cache(self, identifier: str) -> None:
181 key = self._CACHE_KEY_PREFIX + identifier
182 raw: Any = await self._cache.get(key) # type: ignore[union-attr]
183 now = ambient_clock.monotonic()
184 window_start = now - self.lockout_duration_seconds
185 if raw is None:
186 timestamps: list[float] = []
187 else:
188 data: dict[str, Any] = (
189 _json.loads(raw) if isinstance(raw, (str, bytes)) else raw
190 )
191 timestamps = [t for t in data.get("timestamps", []) if t > window_start]
192 timestamps.append(now)
193 await self._cache.set( # type: ignore[union-attr]
194 key,
195 _json.dumps({"timestamps": timestamps}),
196 ttl=self.lockout_duration_seconds,
197 )
199 # ------------------------------------------------------------------
200 # In-memory helpers
201 # ------------------------------------------------------------------
203 def _is_locked_local(self, identifier: str) -> bool:
204 now = ambient_clock.monotonic()
205 window_start = now - self.lockout_duration_seconds
206 recent = [t for t in self._local.get(identifier, []) if t > window_start]
207 return len(recent) >= self.max_attempts
209 def _record_failure_local(self, identifier: str) -> None:
210 now = ambient_clock.monotonic()
211 window_start = now - self.lockout_duration_seconds
212 existing = self._local.get(identifier, [])
213 self._local[identifier] = [t for t in existing if t > window_start] + [now]
216@inject
217class AuthenticationService:
218 """Service for core authentication operations.
220 Handles user login, registration, token creation and validation.
221 """
223 def __init__(
224 self,
225 password_policy: PasswordPolicy,
226 user_store: UserStoreProtocol,
227 token_manager: JWTTokenManager,
228 lockout_config: LockoutConfig | None = None,
229 event_bus: EventBusProtocol | None = None,
230 tracker: LoginAttemptTracker | None = None,
231 hooks: HookRegistryProtocol | None = None,
232 password_hasher: PasswordHasherProtocol | None = None,
233 activity_tracker: AuthActivityTracker | None = None,
234 ) -> None:
235 self.password_policy = password_policy
236 self.user_store = user_store
237 self.token_manager = token_manager
238 self.lockout_config: LockoutConfig = lockout_config or LockoutConfig()
239 self._event_bus = event_bus
240 # Use the explicitly supplied tracker, or build a default in-memory one
241 # from lockout_config so callers that pre-date LoginAttemptTracker are
242 # not affected.
243 self._tracker: LoginAttemptTracker = tracker or LoginAttemptTracker(
244 max_attempts=self.lockout_config.max_attempts,
245 lockout_duration_seconds=self.lockout_config.lockout_duration_seconds,
246 )
247 self._hooks = hooks
248 # The composed hasher injected by the provider; falls back to the
249 # bcrypt hasher for callers that construct the service directly.
250 self._password_hasher: PasswordHasherProtocol = (
251 password_hasher or PasswordHasher()
252 )
253 # Tracks dashboard auth activity when bound by the DI container;
254 # ``None`` disables recording (e.g. tests or bare constructions).
255 self._activity_tracker = activity_tracker
256 # Background tasks kept alive to prevent GC before completion
257 self._background_tasks: set[asyncio.Task[object]] = set()
259 def __repr__(self) -> str:
260 """Return a string representation of this service."""
261 return f"AuthenticationService(user_store={type(self.user_store).__name__})"
263 def _emit(self, event: object) -> None:
264 """Fire-and-forget event publication.
266 Schedules a background task to publish *event* via the event bus (if
267 one is configured). The task reference is stored in
268 ``_background_tasks`` to prevent premature garbage collection; it
269 removes itself from the set upon completion.
270 """
271 if self._event_bus is None:
272 return
273 task: asyncio.Task[object] = asyncio.create_task(self._event_bus.publish(event))
274 self._background_tasks.add(task)
275 task.add_done_callback(self._background_tasks.discard)
277 def set_hook_registry(self, hooks: HookRegistryProtocol | None) -> None:
278 """Attach an optional hook registry after provider boot wiring."""
279 self._hooks = hooks
281 async def _emit_action(self, hook_name: str, payload: object) -> None:
282 """Emit an auth action hook when a registry is available."""
283 if self._hooks is None:
284 return
286 await self._hooks.call_action(hook_name, payload=payload)
288 async def authenticate_user(
289 self,
290 email: str,
291 password: str,
292 ) -> Result[User, InvalidCredentialsError | AccountLockedError]:
293 """Authenticate a user with email and password.
295 Always performs a password hash verification regardless of whether the
296 user exists. This constant-time behaviour prevents user-enumeration
297 attacks via timing side-channels.
299 If the account has exceeded ``LockoutConfig.max_attempts``
300 within the observation window an ``AccountLockedError`` is returned
301 immediately without performing credential verification.
303 Returns:
304 ``Ok(User)`` if authentication succeeds,
305 ``Err(AccountLockedError)`` if the account is temporarily locked,
306 ``Err(InvalidCredentialsError)`` otherwise.
307 """
308 from lexigram.result import Err, Ok
310 # Lockout check — must happen before any credential work
311 if await self._tracker.is_locked(email):
312 # Best-effort user fetch to populate the richer UserLockedOut event.
313 locked_user = await self.user_store.get_user_by_email(email)
314 if locked_user:
315 self._emit(UserLockedOut(user_id=locked_user.user_id, email=email))
316 else:
317 self._emit(UserLoginFailed(email=email, reason="Account locked"))
318 if self._activity_tracker is not None:
319 self._activity_tracker.record_failed_login()
320 return Err(AccountLockedError(email))
322 user = await self.user_store.get_user_by_email(email)
324 # Resolve credentials separately — credentials are never stored on User.
325 creds = await self.user_store.get_credentials(user.user_id) if user else None
327 # Always verify against *some* hash so the code path takes constant
328 # time whether the user exists or not.
329 hashed = (
330 creds.hashed_password
331 if creds and creds.hashed_password
332 else DUMMY_PASSWORD_HASH
333 )
334 verified = await self._password_hasher.verify(password, hashed)
336 if user and user.is_active and verified:
337 from lexigram.auth.hooks import AuthUserAuthenticatedHook
339 # Successful login — clear failure history
340 await self._tracker.clear(email)
341 user.record_login()
342 await self._rehash_password_if_needed(user, password, creds)
343 self._emit(UserLoggedIn(user_id=user.user_id, email=user.email))
344 await self._emit_action(
345 "auth.login",
346 AuthUserAuthenticatedHook(user_id=user.user_id, method="password"),
347 )
348 return Ok(user)
350 # Record the failure
351 await self._tracker.record_failure(email)
352 self._emit(UserLoginFailed(email=email, reason="Invalid credentials"))
353 if self._activity_tracker is not None:
354 self._activity_tracker.record_failed_login()
355 return Err(InvalidCredentialsError())
357 async def _rehash_password_if_needed(
358 self,
359 user: User,
360 password: str,
361 creds: UserCredentials | None,
362 ) -> None:
363 """Rehash password in background if needed."""
364 if not creds or not creds.hashed_password:
365 return
367 try:
368 new_hash = await self._password_hasher.rehash_if_needed(
369 password,
370 creds.hashed_password,
371 )
372 if new_hash:
373 updated_creds = UserCredentials(
374 user_id=user.user_id,
375 hashed_password=new_hash,
376 previous_hashes=creds.previous_hashes,
377 )
378 await self.user_store.update_credentials(updated_creds)
379 except (RuntimeError, ValueError) as exc:
380 logger.warning("password_rehash_failed", reason=str(exc))
382 async def register_user(
383 self, request: RegisterRequest
384 ) -> Result[User, EmailExistsError | PasswordPolicyError]:
385 """Register a new user.
387 Args:
388 request: Registration request with email, name, and password.
390 Returns:
391 ``Ok(User)`` on success.
392 ``Err(PasswordPolicyError)`` if passwords do not match or the
393 password violates the policy.
394 ``Err(EmailExistsError)`` if the email is already registered.
395 """
396 from lexigram.result import Err, Ok
398 if request.password != request.confirm_password:
399 return Err(PasswordPolicyError("Passwords do not match"))
401 try:
402 self.password_policy.validate(request.password)
403 except ValueError as e:
404 return Err(PasswordPolicyError(str(e)))
406 if await self.user_store.get_user_by_email(request.email):
407 return Err(EmailExistsError(request.email))
409 hashed_password = await self._password_hasher.hash(request.password)
411 user = await self.user_store.create_user(
412 name=request.name,
413 email=request.email,
414 hashed_password=hashed_password,
415 roles=["user"],
416 profile=request.profile,
417 )
418 self._emit(UserRegistered(user_id=user.user_id, email=user.email))
419 return Ok(user)
421 def create_token(self, user: User) -> AuthToken:
422 """Create an authentication token for a user."""
423 return self.token_manager.create_token_pair(user)
425 async def verify_token(self, token: str) -> Result[VerifiedToken, TokenError]:
426 """Verify and decode an authentication token.
428 Returns:
429 ``Ok(VerifiedToken)`` if valid, ``Err(TokenError)`` otherwise.
430 """
431 return await self.token_manager.verify_token(token)
433 async def refresh_token(self, refresh_token: str) -> Result[AuthToken, TokenError]:
434 """Refresh an access token using a refresh token.
436 Args:
437 refresh_token: The refresh token string.
439 Returns:
440 ``Ok(AuthToken)`` if the refresh succeeds.
441 ``Err(TokenError)`` if the refresh token is invalid or expired.
442 """
443 from lexigram.contracts.auth.exceptions import TokenError as _TokenError
444 from lexigram.result import Err, Ok
446 try:
447 token = await self.token_manager.refresh_access_token(refresh_token)
448 if self._activity_tracker is not None:
449 self._activity_tracker.record_refresh()
450 return Ok(token)
451 except _TokenError as e:
452 return Err(e)
454 async def get_user_from_token(
455 self, token: str
456 ) -> Result[VerifiedToken, TokenError]:
457 """Get user information from token.
459 Returns:
460 ``Ok(VerifiedToken)`` if valid, ``Err(TokenError)`` otherwise.
461 """
462 return await self.token_manager.get_user_from_token(token)
464 async def shutdown(self) -> None:
465 """Cancel and await all pending background event tasks."""
466 tasks = list(self._background_tasks)
467 for task in tasks:
468 task.cancel()
469 if tasks:
470 await asyncio.gather(*tasks, return_exceptions=True)
471 self._background_tasks.clear()
474__all__ = ["AuthenticationService", "LockoutConfig", "LoginAttemptTracker"]