Coverage for src/lexigram/admin/auth/protocols.py: 100%
102 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:39 +0800
1"""Admin authentication service protocols.
3All protocols remain in lexigram-admin (not lexigram-contracts) because they
4are admin-specific and not consumed by other extension packages.
6``AdminAuditLogServiceProtocol`` extends the framework-wide
7``AuditLoggerProtocol`` from ``lexigram.contracts.audit`` so that admin audit
8implementations satisfy the cross-package contract.
9"""
11from __future__ import annotations
13from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
15from lexigram.contracts.audit import AuditLoggerProtocol
17if TYPE_CHECKING:
18 from datetime import datetime
20 from lexigram.admin.auth.errors import AdminAuthError
21 from lexigram.admin.auth.types import (
22 AdminAuthResult,
23 AdminLockoutInfo,
24 AdminLoginAttempt,
25 AdminPasswordResetToken,
26 AdminPasswordValidationResult,
27 AdminSecurityEvent,
28 AdminSecurityEventType,
29 )
30 from lexigram.result import Result
33@runtime_checkable
34class AdminAuthServiceProtocol(Protocol):
35 """Main authentication orchestration service protocol.
37 Coordinates credential verification, rate limiting, lockout checks,
38 session issuance, and audit logging.
39 """
41 async def authenticate(
42 self,
43 email: str,
44 password: str,
45 ip_address: str,
46 user_agent: str,
47 ) -> Result[AdminAuthResult, AdminAuthError]:
48 """Authenticate an admin user with full security pipeline.
50 Args:
51 email: Admin user email.
52 password: Plain-text password.
53 ip_address: Client IP for rate limiting.
54 user_agent: Client user agent for audit.
56 Returns:
57 Ok(AdminAuthResult) with session details on success.
58 Err with specific AdminAuthError subclass on failure.
59 """
60 ...
62 async def invalidate_session(self, session_id: str) -> None:
63 """Invalidate a session (logout).
65 Args:
66 session_id: Session identifier to revoke.
67 """
68 ...
70 async def invalidate_all_user_sessions(self, user_id: str) -> None:
71 """Revoke all active sessions for a user (e.g., after password change).
73 Args:
74 user_id: Admin user UUID whose sessions to revoke.
75 """
76 ...
78 async def complete_mfa_login(
79 self,
80 user_id: str,
81 email: str,
82 roles: list[str],
83 code: str,
84 ip_address: str,
85 user_agent: str,
86 ) -> Result[AdminAuthResult, AdminAuthError]:
87 """Complete a login after a successful TOTP challenge.
89 Verifies the code, then runs the post-credential pipeline
90 (attempt recording, lockout clearance, session creation, audits)
91 that was deferred when ``authenticate`` returned ``mfa_required``.
93 Args:
94 user_id: Admin user UUID (from the pending challenge).
95 email: Admin user email (from the pending challenge).
96 roles: Role names for the user (from the pending challenge).
97 code: TOTP code to verify.
98 ip_address: Client IP for rate limiting.
99 user_agent: Client user agent for audit.
101 Returns:
102 Ok(AdminAuthResult) with a real session on success.
103 Err(MfaVerificationFailedError) when the code is invalid.
104 Err(MfaNotEnabledError) when 2FA is unavailable.
105 """
106 ...
109@runtime_checkable
110class AdminLoginAttemptStoreProtocol(Protocol):
111 """Persistence protocol for login attempt records."""
113 async def ensure_schema(self) -> None:
114 """Create the admin_login_attempts table if it does not exist."""
115 ...
117 async def insert(self, attempt: AdminLoginAttempt) -> None:
118 """Persist a login attempt record.
120 Args:
121 attempt: The attempt to store.
122 """
123 ...
125 async def count_recent_failures(self, email: str, since_seconds: int) -> int:
126 """Count failed attempts for email within the given window.
128 Args:
129 email: Email address to query.
130 since_seconds: Look-back window in seconds.
132 Returns:
133 Number of failed attempts.
134 """
135 ...
137 async def count_recent_failures_by_ip(
138 self, ip_address: str, since_seconds: int
139 ) -> int:
140 """Count failed attempts from an IP within the given window.
142 Args:
143 ip_address: IP address to query.
144 since_seconds: Look-back window in seconds.
146 Returns:
147 Number of failed attempts.
148 """
149 ...
151 async def clear_failures(self, email: str) -> None:
152 """Clear failure records for email (called on successful login).
154 Args:
155 email: Email to clear.
156 """
157 ...
160@runtime_checkable
161class AdminAccountLockoutStoreProtocol(Protocol):
162 """Persistence protocol for account lockout records."""
164 async def ensure_schema(self) -> None:
165 """Create the admin_account_lockouts table if it does not exist."""
166 ...
168 async def get_active_lockout(self, email: str) -> AdminLockoutInfo | None:
169 """Get active lockout for email, or None if not locked.
171 Args:
172 email: Email to check.
174 Returns:
175 AdminLockoutInfo if active lockout exists, None otherwise.
176 """
177 ...
179 async def create_lockout(
180 self,
181 email: str,
182 consecutive_failures: int,
183 unlock_at: Any | None,
184 is_permanent: bool,
185 ) -> None:
186 """Create or update a lockout record for email.
188 Args:
189 email: Email to lock.
190 consecutive_failures: Total consecutive failures.
191 unlock_at: UTC datetime when lock expires (None if permanent).
192 is_permanent: Whether this requires manual admin unlock.
193 """
194 ...
196 async def clear_lockout(self, email: str) -> None:
197 """Remove active lockout for email (on successful login or admin unlock).
199 Args:
200 email: Email to unlock.
201 """
202 ...
205@runtime_checkable
206class AdminLoginAttemptServiceProtocol(Protocol):
207 """Service for IP rate limiting and account lockout enforcement."""
209 async def check_ip_rate_limit(self, ip_address: str) -> None:
210 """Check IP rate limit. Raises RateLimitExceededError if exceeded.
212 Args:
213 ip_address: Client IP to check.
215 Raises:
216 RateLimitExceededError: When the IP is rate-limited.
217 """
218 ...
220 async def check_account_lockout(self, email: str) -> None:
221 """Check account lockout status. Raises AccountLockedError if locked.
223 Args:
224 email: Email address to check.
226 Raises:
227 AccountLockedError: When the account is locked.
228 """
229 ...
231 async def record_attempt(
232 self,
233 email: str,
234 ip_address: str,
235 user_agent: str,
236 success: bool,
237 failure_reason: str | None = None,
238 ) -> None:
239 """Record a login attempt and update lockout state on failure.
241 Args:
242 email: Email that attempted login.
243 ip_address: Client IP.
244 user_agent: Client user agent.
245 success: Whether the attempt succeeded.
246 failure_reason: Short failure code when success=False.
247 """
248 ...
250 async def clear_lockout(self, email: str) -> None:
251 """Clear lockout and failure records on successful login.
253 Args:
254 email: Email to clear.
255 """
256 ...
259@runtime_checkable
260class AdminAuditLogStoreProtocol(Protocol):
261 """Persistence protocol for security audit log entries."""
263 async def ensure_schema(self) -> None:
264 """Create the admin_security_audit_log table if it does not exist."""
265 ...
267 async def insert(self, event: AdminSecurityEvent) -> None:
268 """Persist a security event.
270 Args:
271 event: Security event to store.
272 """
273 ...
275 async def query_recent(
276 self,
277 admin_user_id: str | None = None,
278 event_type: AdminSecurityEventType | None = None,
279 since_seconds: int = 3600,
280 limit: int = 100,
281 ) -> list[AdminSecurityEvent]:
282 """Query recent security events with optional filters.
284 Args:
285 admin_user_id: Filter to specific user (None = all users).
286 event_type: Filter to specific event type (None = all types).
287 since_seconds: Look-back window in seconds.
288 limit: Maximum records to return.
290 Returns:
291 List of matching security events, newest first.
292 """
293 ...
296@runtime_checkable
297class AdminAuditLogServiceProtocol(AuditLoggerProtocol, Protocol):
298 """Service for recording admin security events.
300 Extends the framework-wide ``AuditLoggerProtocol`` so that admin audit
301 implementations satisfy the cross-package contract. Adds admin-specific
302 methods (``log_event``, ``get_recent_events``) on top of the base
303 ``log()`` and ``query()`` methods from ``AuditLoggerProtocol``.
305 Implementations must never raise — audit failures are swallowed so that
306 an audit store outage cannot interrupt authentication flows.
307 """
309 async def log_event(
310 self,
311 event_type: AdminSecurityEventType,
312 ip_address: str,
313 user_agent: str,
314 success: bool,
315 admin_user_id: str | None = None,
316 metadata: dict[str, Any] | None = None,
317 ) -> None:
318 """Record a security event. Implementation must never raise.
320 Args:
321 event_type: Type of security event.
322 ip_address: Client IP.
323 user_agent: Client user agent.
324 success: Whether the operation succeeded.
325 admin_user_id: Associated admin user (None for pre-auth events).
326 metadata: Optional structured context.
327 """
328 ...
330 async def get_recent_events(
331 self,
332 admin_user_id: str | None = None,
333 since_seconds: int = 3600,
334 limit: int = 50,
335 ) -> list[AdminSecurityEvent]:
336 """Retrieve recent security events for display.
338 Args:
339 admin_user_id: Filter to specific user.
340 since_seconds: Look-back window.
341 limit: Maximum results.
343 Returns:
344 List of security events, newest first.
345 """
346 ...
349@runtime_checkable
350class AdminPasswordPolicyServiceProtocol(Protocol):
351 """Password policy validation service."""
353 def validate(
354 self,
355 password: str,
356 email: str | None = None,
357 ) -> AdminPasswordValidationResult:
358 """Validate a password against all configured policy rules.
360 Returns ALL violations, not just the first one.
362 Args:
363 password: Plain-text password to validate.
364 email: Optional email — used to check if password contains it.
366 Returns:
367 AdminPasswordValidationResult with is_valid and full violations list.
368 """
369 ...
372@runtime_checkable
373class AdminCsrfServiceProtocol(Protocol):
374 """CSRF token generation and validation service."""
376 def generate_token(self, session_id: str) -> str:
377 """Generate a CSRF token scoped to the given session.
379 Token format: base64url(timestamp:nonce:hmac_signature)
381 Args:
382 session_id: Session ID to scope the token to.
384 Returns:
385 CSRF token string.
386 """
387 ...
389 def validate_token(self, session_id: str, token: str) -> bool:
390 """Validate a CSRF token against the session.
392 Uses hmac.compare_digest for timing-safe comparison.
394 Args:
395 session_id: Session ID the token was generated for.
396 token: Token to validate.
398 Returns:
399 True if valid and not expired, False otherwise.
400 """
401 ...
404@runtime_checkable
405class AdminSessionServiceProtocol(Protocol):
406 """Admin session lifecycle management service."""
408 async def create_session(
409 self,
410 user_id: str,
411 email: str,
412 roles: list[str],
413 ip_address: str,
414 user_agent: str,
415 ) -> str:
416 """Create a new session and return the session ID.
418 Args:
419 user_id: Admin user UUID.
420 email: Admin user email.
421 roles: User's roles.
422 ip_address: Client IP.
423 user_agent: Client user agent.
425 Returns:
426 New session identifier (secrets.token_urlsafe(32)).
427 """
428 ...
430 async def get_session(self, session_id: str) -> dict[str, Any] | None:
431 """Retrieve session data if valid (not expired, not revoked).
433 Checks both idle timeout and absolute expiry.
435 Args:
436 session_id: Session to retrieve.
438 Returns:
439 Session data dict or None if not found/expired.
440 """
441 ...
443 async def touch_session(self, session_id: str) -> None:
444 """Update session last-active timestamp.
446 Args:
447 session_id: Session to touch.
448 """
449 ...
451 async def revoke_session(self, session_id: str) -> None:
452 """Revoke a single session.
454 Args:
455 session_id: Session to revoke.
456 """
457 ...
459 async def revoke_all_user_sessions(self, user_id: str) -> None:
460 """Revoke all sessions for a user.
462 Args:
463 user_id: Admin user UUID.
464 """
465 ...
468@runtime_checkable
469class AdminPasswordResetTokenStoreProtocol(Protocol):
470 """Persistence contract for password reset tokens.
472 Implementations:
473 - :class:`~lexigram.admin.auth.store.password_reset_token_sql.AdminPasswordResetTokenSqlStore`
474 """
476 async def ensure_schema(self) -> None:
477 """Create the token table if it does not exist."""
478 ...
480 async def create(self, email: str, token_hash: str, expires_at: datetime) -> None:
481 """Persist a new token record.
483 Args:
484 email: Email the token is issued for.
485 token_hash: sha256 hex digest of the raw token.
486 expires_at: UTC expiry timestamp.
487 """
488 ...
490 async def find_by_hash(self, token_hash: str) -> AdminPasswordResetToken | None:
491 """Look up a token by its sha256 hash.
493 Args:
494 token_hash: sha256 hex digest of the raw token.
496 Returns:
497 Token record or ``None`` when unknown.
498 """
499 ...
501 async def mark_consumed(self, token_hash: str) -> bool:
502 """Atomically verify-and-consume a token in one statement.
504 Args:
505 token_hash: sha256 hex digest of the raw token.
507 Returns:
508 ``True`` only when the token existed, was unconsumed, and had
509 not expired at the instant of the write; ``False`` otherwise
510 — the caller cannot distinguish missing, already-consumed,
511 or expired without a separate lookup.
512 """
513 ...
516@runtime_checkable
517class AdminMfaStoreProtocol(Protocol):
518 """Persistence contract for per-user TOTP secrets.
520 Implementations:
521 - :class:`~lexigram.admin.auth.store.mfa_sql.AdminMfaSqlStore`
522 """
524 async def ensure_schema(self) -> None:
525 """Create the MFA table if it does not exist."""
526 ...
528 async def is_enabled(self, user_id: str) -> bool:
529 """Return True when 2FA is enabled for the user."""
530 ...
532 async def get_secret(self, user_id: str) -> str | None:
533 """Return the stored TOTP secret (None when disabled)."""
534 ...
536 async def save_secret(self, user_id: str, secret: str) -> None:
537 """Persist (or refresh) the TOTP secret for a user."""
538 ...
540 async def disable(self, user_id: str) -> None:
541 """Remove the TOTP secret (2FA off)."""
542 ...
545@runtime_checkable
546class AdminMfaServiceProtocol(Protocol):
547 """TOTP 2FA orchestration contract.
549 Implementations:
550 - :class:`~lexigram.admin.auth.services.mfa_service.AdminMfaService`
551 """
553 async def is_enabled(self, user_id: str) -> bool:
554 """Return True when 2FA is enabled for the user."""
555 ...
557 async def start_setup(
558 self, user_id: str, email: str
559 ) -> Result[tuple[str, str, str], AdminAuthError]:
560 """Generate a TOTP secret, provisioning URI, and QR SVG (no persist).
562 Returns:
563 ``Ok((secret, otpauth_uri, svg))`` on success; ``Err`` when 2FA
564 is disabled in configuration.
565 """
566 ...
568 async def confirm_setup(
569 self, user_id: str, secret: str, code: str
570 ) -> Result[None, AdminAuthError]:
571 """Validate a code against a new secret and persist it."""
572 ...
574 async def verify_code(
575 self, user_id: str, code: str
576 ) -> Result[bool, AdminAuthError]:
577 """Validate a TOTP code; ``Err`` when 2FA is not enabled."""
578 ...
580 async def disable(self, user_id: str, code: str) -> Result[bool, AdminAuthError]:
581 """Disable 2FA (requires a valid current code)."""
582 ...
584 def get_factor(self) -> str:
585 """Return the configured second factor (``"totp"`` or ``"email"``)."""
586 ...
589@runtime_checkable
590class AdminEmailVerificationStoreProtocol(Protocol):
591 """Persistence contract for admin email verification state.
593 Implementations:
594 - :class:`~lexigram.admin.auth.store.email_verification_sql.AdminEmailVerificationSqlStore`
595 """
597 async def ensure_schema(self) -> None:
598 """Create the verification table if it does not exist."""
599 ...
601 async def is_verified(self, user_id: str) -> bool:
602 """Return True when the user's email is verified.
604 Args:
605 user_id: Admin user UUID.
606 """
607 ...
609 async def find_user_by_token_hash(self, token_hash: str) -> str | None:
610 """Look up the user owning an unconsumed token.
612 Args:
613 token_hash: sha256 hex digest of the raw token.
615 Returns:
616 User UUID or ``None`` when no unconsumed token matches.
617 """
618 ...
620 async def save_token(
621 self, user_id: str, token_hash: str, expires_at: datetime
622 ) -> None:
623 """Persist (or refresh) the verification token for a user.
625 Args:
626 user_id: Admin user UUID.
627 token_hash: sha256 hex digest of the raw token.
628 expires_at: UTC expiry timestamp.
629 """
630 ...
632 async def consume_token(self, user_id: str, token_hash: str) -> bool:
633 """Atomically verify + consume a token.
635 Marks the email verified and clears the token when the hash matches,
636 the token is unexpired, and the email is not already verified.
638 Args:
639 user_id: Admin user UUID.
640 token_hash: sha256 hex digest of the raw token.
642 Returns:
643 ``True`` when the token was valid and consumed.
644 """
645 ...
647 async def clear_token(self, user_id: str) -> None:
648 """Remove the pending verification token for a user.
650 Args:
651 user_id: Admin user UUID.
652 """
653 ...
656@runtime_checkable
657class AdminEmailOtpStoreProtocol(Protocol):
658 """Persistence contract for email one-time-password codes.
660 Implementations:
661 - :class:`~lexigram.admin.auth.store.email_otp_sql.AdminEmailOtpSqlStore`
662 """
664 async def ensure_schema(self) -> None:
665 """Create the OTP table if it does not exist."""
666 ...
668 async def save(self, user_id: str, code_hash: str, expires_at: datetime) -> None:
669 """Persist a new emailed code.
671 Args:
672 user_id: Admin user UUID.
673 code_hash: sha256 hex digest of the raw code.
674 expires_at: UTC expiry timestamp.
675 """
676 ...
678 async def consume(self, user_id: str, code_hash: str) -> bool:
679 """Atomically consume a matching unexpired code.
681 Args:
682 user_id: Admin user UUID.
683 code_hash: sha256 hex digest of the raw code.
685 Returns:
686 ``True`` when an unexpired, unused code matched and was consumed.
687 """
688 ...
690 async def last_sent_at(self, user_id: str) -> datetime | None:
691 """Return the creation time of the most recent code.
693 Args:
694 user_id: Admin user UUID.
696 Returns:
697 UTC datetime of the newest code, or ``None`` when none exists.
698 """
699 ...
702@runtime_checkable
703class AdminEmailOtpServiceProtocol(Protocol):
704 """Email one-time-password factor contract.
706 Implementations:
707 - :class:`~lexigram.admin.auth.services.email_otp_service.AdminEmailOtpService`
708 """
710 async def send_otp(
711 self, user_id: str, email: str, user_name: str
712 ) -> Result[None, AdminAuthError]:
713 """Generate, persist, and email a fresh one-time code.
715 Returns:
716 ``Ok(None)`` on success; ``Err`` when disabled, in cooldown, or
717 undeliverable.
718 """
719 ...
721 async def verify_otp(self, user_id: str, code: str) -> Result[bool, AdminAuthError]:
722 """Verify a code and consume it when valid.
724 Returns:
725 ``Ok(True)`` on match; ``Ok(False)`` otherwise;
726 ``Err`` when the factor is disabled.
727 """
728 ...
731@runtime_checkable
732class AdminEmailVerificationServiceProtocol(Protocol):
733 """Email verification orchestration contract.
735 Implementations:
736 - :class:`~lexigram.admin.auth.services.email_verification_service.AdminEmailVerificationService`
737 """
739 async def is_verified(self, user_id: str) -> bool:
740 """Return True when the user's email is verified."""
741 ...
743 async def is_required(self, user_id: str) -> bool:
744 """Return True when login must be gated on email verification."""
745 ...
747 async def send_verification(
748 self,
749 user_id: str,
750 email: str,
751 user_name: str,
752 base_url: str = "",
753 ip_address: str = "",
754 ) -> Result[None, AdminAuthError]:
755 """Issue a verification link and email it to the user.
757 No-op (Ok) when disabled or already verified; fail-open on delivery.
758 Rate limited per IP when a cache backend is wired (fail open).
759 """
760 ...
762 async def verify_token(self, token: str) -> Result[bool, AdminAuthError]:
763 """Validate and consume a verification token.
765 Returns:
766 ``Ok(True)`` on success; ``Err(EmailVerificationTokenInvalidError)``
767 for unknown/used/expired tokens.
768 """
769 ...
771 async def resend_verification(
772 self,
773 user_id: str,
774 email: str,
775 user_name: str,
776 base_url: str = "",
777 ip_address: str = "",
778 ) -> Result[None, AdminAuthError]:
779 """Re-issue and re-send the verification email."""
780 ...
783@runtime_checkable
784class AdminPasswordResetServiceProtocol(Protocol):
785 """Password reset orchestration contract."""
787 async def request_reset(
788 self,
789 email: str,
790 ip_address: str,
791 user_agent: str,
792 base_url: str,
793 ) -> Result[None, AdminAuthError]:
794 """Issue a reset token and notify the user.
796 Always returns ``Ok(None)`` for unknown emails (anti-enumeration).
797 """
798 ...
800 async def confirm_reset(
801 self,
802 token: str,
803 new_password: str,
804 ip_address: str = "",
805 user_agent: str = "",
806 ) -> Result[None, AdminAuthError]:
807 """Validate a token and apply a new password.
809 Consumes the token on success and invalidates all user sessions.
810 """
811 ...
814__all__ = [
815 "AdminAccountLockoutStoreProtocol",
816 "AdminAuditLogServiceProtocol",
817 "AdminAuditLogStoreProtocol",
818 "AdminAuthServiceProtocol",
819 "AdminCsrfServiceProtocol",
820 "AdminEmailOtpServiceProtocol",
821 "AdminEmailOtpStoreProtocol",
822 "AdminEmailVerificationServiceProtocol",
823 "AdminEmailVerificationStoreProtocol",
824 "AdminLoginAttemptServiceProtocol",
825 "AdminLoginAttemptStoreProtocol",
826 "AdminMfaServiceProtocol",
827 "AdminMfaStoreProtocol",
828 "AdminPasswordPolicyServiceProtocol",
829 "AdminPasswordResetServiceProtocol",
830 "AdminPasswordResetTokenStoreProtocol",
831 "AdminSessionServiceProtocol",
832]