Coverage for src/lexigram/admin/auth/services/mfa_service.py: 96%
53 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""TOTP two-factor authentication service for the admin panel.
3Owns the 2FA lifecycle: secret generation, QR provisioning, code
4verification, and disable. TOTP math is delegated to lexigram-auth's
5``authn.mfa`` primitives (RFC 6238 single source of truth); this module
6keeps the admin store wiring and the QR SVG rendering. Depends only on
7the MFA store protocol and (optionally) the audit log service, so it is
8trivially testable without a database.
9"""
11from __future__ import annotations
13import segno
15from lexigram.admin.auth.errors import (
16 AdminAuthError,
17 MfaNotEnabledError,
18 MfaVerificationFailedError,
19)
20from lexigram.admin.auth.protocols import (
21 AdminAuditLogServiceProtocol,
22 AdminMfaStoreProtocol,
23)
24from lexigram.admin.auth.types import AdminSecurityEventType
25from lexigram.admin.config import AdminMfaConfig
26from lexigram.auth.authn.mfa import (
27 generate_totp_secret,
28 get_provisioning_uri,
29 verify_totp,
30)
31from lexigram.logging import get_logger
32from lexigram.result import Err, Ok, Result
34logger = get_logger(__name__)
37class AdminMfaService:
38 """TOTP 2FA lifecycle: setup, verify, disable.
40 Args:
41 config: MFA configuration (issuer, skew, enabled flag).
42 store: Persistence for per-user TOTP secrets.
43 audit_service: Optional security-event recorder; ``None`` skips
44 audit logging (e.g. in tests).
45 """
47 def __init__(
48 self,
49 config: AdminMfaConfig,
50 store: AdminMfaStoreProtocol,
51 audit_service: AdminAuditLogServiceProtocol | None = None,
52 ) -> None:
53 self._config = config
54 self._store = store
55 self._audit_service = audit_service
57 # ------------------------------------------------------------------
58 # Public API
59 # ------------------------------------------------------------------
61 def get_factor(self) -> str:
62 """Return the configured second factor (``"totp"`` or ``"email"``)."""
63 return self._config.factor
65 async def is_enabled(self, user_id: str) -> bool:
66 """Return True when 2FA is enabled for the user."""
67 return await self._store.is_enabled(user_id)
69 async def start_setup(
70 self, user_id: str, email: str
71 ) -> Result[tuple[str, str, str], AdminAuthError]:
72 """Generate a new TOTP secret, provisioning URI, and QR SVG.
74 Nothing is persisted here — the secret travels via the caller
75 (session) and is committed by ``confirm_setup``.
77 Args:
78 user_id: Admin user UUID.
79 email: Admin user email (embedded in the provisioning URI).
81 Returns:
82 ``Ok((secret, otpauth_uri, svg))`` on success.
83 ``Err(MfaNotEnabledError)`` when 2FA is disabled in config.
84 """
85 if not self._config.enabled:
86 return Err(
87 MfaNotEnabledError(
88 "Two-factor authentication is disabled for this panel."
89 )
90 )
91 secret = generate_totp_secret()
92 uri = get_provisioning_uri(secret, username=email, issuer=self._config.issuer)
93 svg = segno.make(uri).svg_inline(scale=4)
94 return Ok((secret, uri, svg))
96 async def confirm_setup(
97 self, user_id: str, secret: str, code: str
98 ) -> Result[None, AdminAuthError]:
99 """Validate a code against a new secret and persist it.
101 Args:
102 user_id: Admin user UUID.
103 secret: TOTP secret (from ``start_setup``).
104 code: TOTP code to validate.
106 Returns:
107 ``Ok(None)`` when the code is valid and the secret is stored.
108 ``Err(MfaVerificationFailedError)`` when the code is invalid.
109 ``Err(MfaNotEnabledError)`` when 2FA is disabled in config.
110 """
111 if not self._config.enabled:
112 return Err(
113 MfaNotEnabledError(
114 "Two-factor authentication is disabled for this panel."
115 )
116 )
117 if not verify_totp(secret, code, window=self._config.skew):
118 return Err(MfaVerificationFailedError("Invalid verification code."))
119 await self._store.save_secret(user_id, secret)
120 if self._audit_service is not None:
121 await self._audit_service.log_event(
122 event_type=AdminSecurityEventType.MFA_ENABLED,
123 ip_address="",
124 user_agent="",
125 success=True,
126 admin_user_id=user_id,
127 metadata={},
128 )
129 logger.info("admin_mfa_enabled", user_id=user_id)
130 return Ok(None)
132 async def verify_code(
133 self, user_id: str, code: str
134 ) -> Result[bool, AdminAuthError]:
135 """Validate a TOTP code for a user without persisting anything.
137 Args:
138 user_id: Admin user UUID.
139 code: TOTP code to validate.
141 Returns:
142 ``Ok(True)`` for a valid code, ``Ok(False)`` otherwise.
143 ``Err(MfaNotEnabledError)`` when the user has no stored secret.
144 """
145 secret = await self._store.get_secret(user_id)
146 if secret is None:
147 return Err(
148 MfaNotEnabledError(
149 "Two-factor authentication is not enabled for this account."
150 )
151 )
152 return Ok(verify_totp(secret, code, window=self._config.skew))
154 async def disable(self, user_id: str, code: str) -> Result[bool, AdminAuthError]:
155 """Disable 2FA for a user (requires a valid current code).
157 Args:
158 user_id: Admin user UUID.
159 code: Current TOTP code proving possession of the secret.
161 Returns:
162 ``Ok(True)`` when disabled.
163 ``Err(MfaNotEnabledError)`` when 2FA is not enabled.
164 ``Err(MfaVerificationFailedError)`` when the code is invalid.
165 """
166 result = await self.verify_code(user_id, code)
167 if result.is_err():
168 return result
169 if not result.unwrap():
170 return Err(MfaVerificationFailedError("Invalid verification code."))
171 await self._store.disable(user_id)
172 if self._audit_service is not None:
173 await self._audit_service.log_event(
174 event_type=AdminSecurityEventType.MFA_DISABLED,
175 ip_address="",
176 user_agent="",
177 success=True,
178 admin_user_id=user_id,
179 metadata={},
180 )
181 logger.info("admin_mfa_disabled", user_id=user_id)
182 return Ok(True)
185__all__ = ["AdminMfaService"]