Coverage for src/lexigram/admin/auth/services/password_reset_service.py: 95%
97 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Password reset orchestration service for the admin panel.
3Coordinates the full reset flow: token issuance, hashed-token persistence,
4email notification (optional), password policy validation, password update,
5token consumption, session invalidation, and audit logging.
6"""
8from __future__ import annotations
10from datetime import UTC, datetime, timedelta
11import hashlib
12import secrets
14from lexigram.admin.auth.errors import (
15 AdminAuthError,
16 PasswordPolicyError,
17 PasswordResetTokenExpiredError,
18 PasswordResetTokenInvalidError,
19 RateLimitExceededError,
20)
21from lexigram.admin.auth.protocols import (
22 AdminAuditLogServiceProtocol,
23 AdminAuthServiceProtocol,
24 AdminPasswordPolicyServiceProtocol,
25 AdminPasswordResetTokenStoreProtocol,
26)
27from lexigram.admin.auth.store.protocols import AdminUserStoreProtocol
28from lexigram.admin.auth.types import AdminSecurityEventType
29from lexigram.admin.services.notifications import AdminNotificationService
30from lexigram.contracts.auth import PasswordHasherProtocol
31from lexigram.contracts.infra.cache import CacheBackendProtocol
32from lexigram.di.decorators import inject
33from lexigram.logging import get_logger
34from lexigram.result import Err, Ok, Result
36logger = get_logger(__name__)
39@inject
40class AdminPasswordResetService:
41 """Password reset orchestrator (see module docstring).
43 Anti-enumeration: ``request_reset`` returns ``Ok(None)`` for unknown
44 emails, producing an identical "check your email" response in all cases.
46 Args:
47 user_store: Admin user persistence (lookup + password update).
48 token_store: Persistence for single-use reset tokens.
49 audit_service: Security event recording (fire-and-forget).
50 auth_service: Session invalidation after a successful reset.
51 policy_service: New-password policy validation.
52 hasher: Password hasher; defaults to ``PasswordHasher`` from
53 ``lexigram-auth`` when ``None``.
54 notification_service: Optional email notification; ``None`` skips
55 sending.
56 token_lifetime: Token validity in seconds (default 3600 = 1 hour).
57 cache: Optional cache backend for request rate limiting; ``None`` (or
58 a failing cache) skips limiting (fail open).
59 reset_request_limit: Max request-reset calls per IP per window.
60 reset_request_window_seconds: Rate-limit window length in seconds.
61 """
63 def __init__(
64 self,
65 user_store: AdminUserStoreProtocol,
66 token_store: AdminPasswordResetTokenStoreProtocol,
67 audit_service: AdminAuditLogServiceProtocol,
68 auth_service: AdminAuthServiceProtocol,
69 policy_service: AdminPasswordPolicyServiceProtocol,
70 hasher: PasswordHasherProtocol | None = None,
71 notification_service: AdminNotificationService | None = None,
72 token_lifetime: int = 3600,
73 cache: CacheBackendProtocol | None = None,
74 reset_request_limit: int = 5,
75 reset_request_window_seconds: int = 3600,
76 ) -> None:
77 self._user_store = user_store
78 self._token_store = token_store
79 self._audit_service = audit_service
80 self._auth_service = auth_service
81 self._policy_service = policy_service
82 self._hasher = hasher
83 self._notification_service = notification_service
84 self._token_lifetime = token_lifetime
85 self._cache = cache
86 self._reset_request_limit = reset_request_limit
87 self._reset_request_window_seconds = reset_request_window_seconds
89 # ------------------------------------------------------------------
90 # Public API
91 # ------------------------------------------------------------------
93 async def request_reset(
94 self,
95 email: str,
96 ip_address: str,
97 user_agent: str,
98 base_url: str,
99 ) -> Result[None, AdminAuthError]:
100 """Issue a reset token and notify the user.
102 Always returns ``Ok(None)`` for unknown emails so the response is
103 identical in all cases (anti-enumeration). A token is persisted
104 (sha256 of the raw token) only when the email matches an account;
105 the audit event is also emitted only for real accounts.
107 Args:
108 email: Email to send the reset link to.
109 ip_address: Client IP for audit and rate limiting.
110 user_agent: Client user agent for audit.
111 base_url: Request base URL used to build the reset link.
113 Returns:
114 ``Ok(None)`` — always, regardless of whether the email exists —
115 or ``Err(RateLimitExceededError)`` when this IP exceeds the
116 request limit.
117 """
118 email = email.strip().lower()
120 if self._cache is not None and await self._is_rate_limited(ip_address):
121 logger.warning("admin.password_reset_rate_limited", ip=ip_address)
122 return Err(
123 RateLimitExceededError(
124 "Too many password reset requests. Please try again later.",
125 reason="rate_limit",
126 )
127 )
129 user = await self._user_store.get_user_by_email(email)
130 if user is None:
131 logger.info("admin.password_reset_unknown_email", email=email)
132 return Ok(None)
134 raw_token = secrets.token_urlsafe(32)
135 token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
136 expires_at = datetime.now(UTC) + timedelta(seconds=self._token_lifetime)
137 await self._token_store.create(email, token_hash, expires_at)
139 if self._notification_service is not None:
140 reset_url = f"{base_url.rstrip('/')}/admin/password-reset/{raw_token}"
141 result = await self._notification_service.notify_password_reset(
142 user_email=email,
143 user_name=getattr(user, "name", email),
144 reset_url=reset_url,
145 expires_in=f"{self._token_lifetime // 60} minutes",
146 )
147 if getattr(result, "is_err", lambda: False)():
148 logger.warning(
149 "admin.password_reset_notify_failed",
150 email=email,
151 error=str(result.unwrap_err()),
152 )
154 await self._audit_service.log_event(
155 event_type=AdminSecurityEventType.PASSWORD_RESET_REQUESTED,
156 ip_address=ip_address,
157 user_agent=user_agent,
158 success=True,
159 admin_user_id=getattr(user, "user_id", None),
160 metadata={"email": email},
161 )
162 logger.info("admin.password_reset_requested", email=email)
163 return Ok(None)
165 async def _is_rate_limited(self, ip_address: str) -> bool:
166 """Check and increment the per-IP request counter. Fail open.
168 Uses a fixed-window counter keyed by a sha256 hash of the client IP
169 (avoids PII in cache key listings). Any cache failure is treated as
170 "not limited" so a cache outage never blocks password resets.
172 Args:
173 ip_address: Client IP address.
175 Returns:
176 ``True`` when the IP exceeds ``reset_request_limit``.
177 """
178 try:
179 ip_hash = hashlib.sha256(ip_address.encode()).hexdigest()[:16]
180 key = f"admin:password-reset:ip:{ip_hash}"
181 cache = self._cache
182 if cache is None:
183 return False
184 result = await cache.get(key)
185 value = result.unwrap() if result.is_ok() else None
186 count = int(value) if value else 0
187 if count >= self._reset_request_limit:
188 return True
189 await cache.set(key, str(count + 1), ttl=self._reset_request_window_seconds)
190 return False
191 except Exception: # noqa: BLE001
192 logger.warning("admin.password_reset_rate_limit_unavailable")
193 return False
195 async def confirm_reset(
196 self,
197 token: str,
198 new_password: str,
199 ip_address: str = "",
200 user_agent: str = "",
201 ) -> Result[
202 None,
203 PasswordResetTokenInvalidError
204 | PasswordResetTokenExpiredError
205 | PasswordPolicyError,
206 ]:
207 """Validate a token and apply a new password.
209 Consumes the token and revokes all active sessions for the user on
210 success.
212 Args:
213 token: Raw reset token from the emailed link.
214 new_password: New plain-text password (policy-validated).
215 ip_address: Client IP for audit.
216 user_agent: Client user agent for audit.
218 Returns:
219 ``Ok(None)`` on success, or an ``Err`` with a specific
220 ``AdminAuthError`` subclass describing the failure.
221 """
222 token_hash = hashlib.sha256(token.strip().encode()).hexdigest()
223 record = await self._token_store.find_by_hash(token_hash)
224 if record is None or record.consumed_at is not None:
225 return Err(PasswordResetTokenInvalidError())
226 if record.expires_at < datetime.now(UTC):
227 return Err(PasswordResetTokenExpiredError())
229 validation = self._policy_service.validate(new_password, record.email)
230 if not validation.is_valid:
231 message = "; ".join(v.message for v in validation.violations)
232 return Err(PasswordPolicyError(message))
234 consumed = await self._token_store.mark_consumed(token_hash)
235 if not consumed:
236 return Err(PasswordResetTokenInvalidError())
238 user = await self._user_store.get_user_by_email(record.email)
239 if user is None:
240 return Err(PasswordResetTokenInvalidError())
242 hasher = self._hasher
243 if hasher is None:
244 from lexigram.auth.authn.security import PasswordHasher
246 hasher = PasswordHasher()
247 user.hashed_password = await hasher.hash(new_password)
248 await self._user_store.update_user(user)
250 user_id = getattr(user, "user_id", None)
251 if user_id:
252 await self._auth_service.invalidate_all_user_sessions(user_id)
254 await self._audit_service.log_event(
255 event_type=AdminSecurityEventType.PASSWORD_CHANGED,
256 ip_address=ip_address,
257 user_agent=user_agent,
258 success=True,
259 admin_user_id=user_id,
260 metadata={"email": record.email},
261 )
262 logger.info("admin.password_reset_confirmed", email=record.email)
263 return Ok(None)
266__all__ = ["AdminPasswordResetService"]