Coverage for src/lexigram/admin/auth/services/email_verification_service.py: 93%
91 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Admin email verification orchestration service.
3Handles the verify-your-email flow: issuing single-use verification links,
4delivering them through the admin notification service, and consuming the
5links. Depends only on ``AdminEmailVerificationStoreProtocol`` from
6``lexigram.admin.auth.protocols``.
7"""
9from __future__ import annotations
11from datetime import UTC, datetime, timedelta
12import hashlib
13import secrets
14from typing import TYPE_CHECKING
16from lexigram.admin.auth.errors import (
17 AdminAuthError,
18 EmailVerificationTokenInvalidError,
19 RateLimitExceededError,
20)
21from lexigram.admin.auth.types import AdminSecurityEventType
22from lexigram.admin.config import AdminEmailVerificationConfig
23from lexigram.contracts.infra.cache import CacheBackendProtocol
24from lexigram.di.decorators import inject
25from lexigram.logging import get_logger
26from lexigram.result import Err, Ok, Result
28if TYPE_CHECKING:
29 from lexigram.admin.auth.protocols import (
30 AdminAuditLogServiceProtocol,
31 AdminEmailVerificationStoreProtocol,
32 )
33 from lexigram.admin.services.notifications import AdminNotificationService
35logger = get_logger(__name__)
38@inject
39class AdminEmailVerificationService:
40 """Email verification flow: links, delivery, and consumption.
42 A verification link embeds a random token; the store keeps only its
43 sha256 digest. Delivery failures are surfaced: when no notification
44 service (or mailer backend) is bound, sending fails with a descriptive
45 error telling the operator which dependency to configure, so missing
46 infrastructure is never silently swallowed.
47 """
49 def __init__(
50 self,
51 config: AdminEmailVerificationConfig,
52 store: AdminEmailVerificationStoreProtocol,
53 notification_service: AdminNotificationService | None = None,
54 audit_service: AdminAuditLogServiceProtocol | None = None,
55 cache: CacheBackendProtocol | None = None,
56 resend_request_limit: int = 5,
57 resend_window_seconds: int = 3600,
58 ) -> None:
59 """Initialise the verification service.
61 Args:
62 config: Email verification settings.
63 store: Verification state persistence.
64 notification_service: Optional delivery channel for the
65 verification email (skipped when None).
66 audit_service: Optional security audit logger.
67 cache: Optional cache backend for per-IP resend rate limiting;
68 ``None`` (or a failing cache) skips limiting (fail open).
69 resend_request_limit: Max resend requests per IP per window.
70 resend_window_seconds: Rate-limit window length in seconds.
71 """
72 self._config = config
73 self._store = store
74 self._notification_service = notification_service
75 self._audit_service = audit_service
76 self._cache = cache
77 self._resend_request_limit = resend_request_limit
78 self._resend_window_seconds = resend_window_seconds
80 async def is_verified(self, user_id: str) -> bool:
81 """Return True when the user's email is verified.
83 Args:
84 user_id: Admin user UUID.
85 """
86 return await self._store.is_verified(user_id)
88 async def is_required(self, user_id: str) -> bool:
89 """Return True when login must be gated on email verification.
91 Args:
92 user_id: Admin user UUID.
93 """
94 if not self._config.enabled or not self._config.enforcement:
95 return False
96 return not await self._store.is_verified(user_id)
98 async def send_verification(
99 self,
100 user_id: str,
101 email: str,
102 user_name: str,
103 base_url: str = "",
104 ip_address: str = "",
105 ) -> Result[None, AdminAuthError]:
106 """Issue a verification link and email it to the user.
108 No-op (Ok) when the flow is disabled or the email is already
109 verified. Delivery failures are NOT swallowed: a missing
110 notification service or a failed delivery returns
111 ``Err(AdminAuthError)`` with guidance on the missing mailer
112 dependency. Resend requests are rate limited per IP when a cache
113 backend is wired (fail open).
115 Args:
116 user_id: Admin user UUID.
117 email: Email address to verify.
118 user_name: Display name for the email greeting.
119 base_url: Origin used to build the absolute verify link
120 (e.g. ``https://panel.example.com``).
121 ip_address: Client IP for resend rate limiting.
123 Returns:
124 ``Ok(None)`` when the link was issued and delivered.
125 ``Err(RateLimitExceededError)`` when this IP exceeds the
126 resend limit.
127 ``Err(AdminAuthError)`` when no notification service is bound
128 or email delivery failed (e.g. no mailer backend configured).
129 """
130 if not self._config.enabled or await self._store.is_verified(user_id):
131 return Ok(None)
133 if self._cache is not None and await self._is_rate_limited(ip_address):
134 logger.warning("email_verification_rate_limited", ip=ip_address)
135 return Err(
136 RateLimitExceededError(
137 "Too many verification emails. Please try again later.",
138 reason="rate_limit",
139 )
140 )
142 token = secrets.token_urlsafe(32)
143 token_hash = hashlib.sha256(token.encode()).hexdigest()
144 expires_at = datetime.now(UTC) + timedelta(hours=self._config.token_ttl_hours)
145 await self._store.save_token(user_id, token_hash, expires_at)
147 verify_url = f"{base_url.rstrip('/')}/admin/verify-email/{token}"
149 if self._notification_service is None:
150 logger.error(
151 "email_verification_skipped",
152 user_id=user_id,
153 email=email,
154 verify_url=verify_url,
155 )
156 if self._audit_service is not None:
157 await self._audit_service.log_event(
158 event_type=AdminSecurityEventType.EMAIL_VERIFICATION_SENT,
159 ip_address="",
160 user_agent="",
161 success=False,
162 admin_user_id=user_id,
163 metadata={"email": email, "reason": "no_notification_service"},
164 )
165 return Err(
166 AdminAuthError(
167 "Verification email could not be delivered because no "
168 "notification/mailer dependency is configured. Configure a "
169 "mailer backend (lexigram-notification MailerModule with "
170 "driver 'smtp'/'sendgrid', or 'console' in development) "
171 "and retry.",
172 )
173 )
175 result = await self._notification_service.notify_email_verification(
176 user_email=email,
177 user_name=user_name,
178 verify_url=verify_url,
179 expires_in=f"{self._config.token_ttl_hours} hours",
180 )
181 if result.is_err():
182 logger.error(
183 "email_verification_send_failed",
184 user_id=user_id,
185 email=email,
186 error=str(result.unwrap_err()),
187 verify_url=verify_url,
188 )
189 if self._audit_service is not None:
190 await self._audit_service.log_event(
191 event_type=AdminSecurityEventType.EMAIL_VERIFICATION_SENT,
192 ip_address="",
193 user_agent="",
194 success=False,
195 admin_user_id=user_id,
196 metadata={"email": email, "reason": str(result.unwrap_err())},
197 )
198 return Err(
199 AdminAuthError(
200 "Verification email could not be delivered: "
201 f"{result.unwrap_err()} Configure a mailer backend "
202 "(lexigram-notification MailerModule with driver "
203 "'smtp'/'sendgrid', or 'console' in development) and "
204 "retry.",
205 )
206 )
208 if self._audit_service is not None:
209 await self._audit_service.log_event(
210 event_type=AdminSecurityEventType.EMAIL_VERIFICATION_SENT,
211 ip_address="",
212 user_agent="",
213 success=True,
214 admin_user_id=user_id,
215 metadata={"email": email},
216 )
217 return Ok(None)
219 async def verify_token(self, token: str) -> Result[bool, AdminAuthError]:
220 """Validate and consume a verification token.
222 Args:
223 token: Raw token from the emailed link.
225 Returns:
226 ``Ok(True)`` when the token was valid and the email is now
227 verified; ``Err(EmailVerificationTokenInvalidError)`` when the
228 token is unknown, used, or expired.
229 """
230 token_hash = hashlib.sha256(token.encode()).hexdigest()
231 user_id = await self._store.find_user_by_token_hash(token_hash)
233 if user_id is None:
234 await self._audit_failure("no_such_token")
235 return Err(
236 EmailVerificationTokenInvalidError(
237 "Invalid or expired verification link."
238 )
239 )
241 consumed = await self._store.consume_token(user_id, token_hash)
242 if not consumed:
243 await self._audit_failure("consumed_or_expired", user_id)
244 return Err(
245 EmailVerificationTokenInvalidError(
246 "Invalid or expired verification link."
247 )
248 )
250 if self._audit_service is not None:
251 await self._audit_service.log_event(
252 event_type=AdminSecurityEventType.EMAIL_VERIFIED,
253 ip_address="",
254 user_agent="",
255 success=True,
256 admin_user_id=user_id,
257 )
258 return Ok(True)
260 async def resend_verification(
261 self,
262 user_id: str,
263 email: str,
264 user_name: str,
265 base_url: str = "",
266 ip_address: str = "",
267 ) -> Result[None, AdminAuthError]:
268 """Re-issue and re-send the verification email.
270 Args:
271 user_id: Admin user UUID.
272 email: Email address to verify.
273 user_name: Display name for the email greeting.
274 base_url: Origin used to build the absolute verify link.
275 ip_address: Client IP for resend rate limiting.
277 Returns:
278 ``Ok(None)`` on success or when the flow is disabled/verified.
279 """
280 return await self.send_verification(
281 user_id, email, user_name, base_url, ip_address
282 )
284 async def _is_rate_limited(self, ip_address: str) -> bool:
285 """Check and increment the per-IP resend counter. Fail open.
287 Uses a fixed-window counter keyed by a sha256 hash of the client IP
288 (avoids PII in cache key listings). Any cache failure is treated as
289 "not limited" so a cache outage never blocks verification emails.
291 Args:
292 ip_address: Client IP address.
294 Returns:
295 ``True`` when the IP exceeds ``resend_request_limit``.
296 """
297 try:
298 ip_hash = hashlib.sha256(ip_address.encode()).hexdigest()[:16]
299 key = f"admin:email-verification:ip:{ip_hash}"
300 cache = self._cache
301 if cache is None:
302 return False
303 result = await cache.get(key)
304 value = result.unwrap() if result.is_ok() else None
305 count = int(value) if value else 0
306 if count >= self._resend_request_limit:
307 return True
308 await cache.set(key, str(count + 1), ttl=self._resend_window_seconds)
309 return False
310 except Exception: # noqa: BLE001 — fail open on cache outages
311 logger.warning("email_verification_rate_limit_unavailable")
312 return False
314 async def _audit_failure(self, reason: str, user_id: str | None = None) -> None:
315 """Record a failed verification attempt (never raises)."""
316 if self._audit_service is None:
317 return
318 await self._audit_service.log_event(
319 event_type=AdminSecurityEventType.EMAIL_VERIFICATION_FAILED,
320 ip_address="",
321 user_agent="",
322 success=False,
323 admin_user_id=user_id,
324 metadata={"reason": reason},
325 )
328__all__ = ["AdminEmailVerificationService"]