Coverage for src/lexigram/admin/auth/services/email_otp_service.py: 0%

1"""Admin email one-time-password (login factor) service. 

2 

3Handles issuing, delivering, and consuming emailed 6-digit codes used as 

4the config-chosen second factor. Code issuance delegates to 

5lexigram-auth's RFC 6238 primitives (``authn.mfa``, period = code TTL); 

6storage is a sha256 digest verified by atomic consume. Depends only on 

7``AdminEmailOtpStoreProtocol`` from ``lexigram.admin.auth.protocols``. 

8""" 

9 

10from __future__ import annotations 

11 

12from datetime import UTC, datetime, timedelta 

13import hashlib 

14from typing import TYPE_CHECKING 

15 

16from lexigram.admin.auth.errors import ( 

17 EmailOtpCooldownError, 

18 EmailOtpDeliveryError, 

19 MfaNotEnabledError, 

20) 

21from lexigram.admin.auth.types import AdminSecurityEventType 

22from lexigram.admin.config import AdminEmailOtpConfig 

23from lexigram.auth.authn.mfa import generate_totp_code, generate_totp_secret 

24from lexigram.di.decorators import inject 

25from lexigram.logging import get_logger 

26from lexigram.result import Err, Ok, Result 

27 

28if TYPE_CHECKING: 

29 from lexigram.admin.auth.errors import AdminAuthError 

30 from lexigram.admin.auth.protocols import ( 

31 AdminAuditLogServiceProtocol, 

32 AdminEmailOtpStoreProtocol, 

33 ) 

34 from lexigram.admin.services.notifications import AdminNotificationService 

35 

36logger = get_logger(__name__) 

37 

38 

39@inject 

40class AdminEmailOtpService: 

41 """Email OTP factor: issue, deliver, and verify 6-digit codes. 

42 

43 Codes are stored only as sha256 digests and consumed atomically 

44 (single-use, expiring). Unlike verification emails, delivery failure is 

45 NOT fail-open — the factor is unusable without a working mailer, so the 

46 service returns ``Err(EmailOtpDeliveryError)`` instead. 

47 """ 

48 

49 def __init__( 

50 self, 

51 config: AdminEmailOtpConfig, 

52 store: AdminEmailOtpStoreProtocol, 

53 notification_service: AdminNotificationService | None = None, 

54 audit_service: AdminAuditLogServiceProtocol | None = None, 

55 ) -> None: 

56 """Initialise the email OTP service. 

57 

58 Args: 

59 config: Email OTP settings (TTL, resend cooldown). 

60 store: Code persistence. 

61 notification_service: Delivery channel for the emailed code. 

62 audit_service: Optional security audit logger. 

63 """ 

64 self._config = config 

65 self._store = store 

66 self._notification_service = notification_service 

67 self._audit_service = audit_service 

68 

69 async def send_otp( 

70 self, 

71 user_id: str, 

72 email: str, 

73 user_name: str, 

74 ) -> Result[None, AdminAuthError]: 

75 """Generate, persist, and email a fresh one-time code. 

76 

77 Args: 

78 user_id: Admin user UUID. 

79 email: Email address to send the code to. 

80 user_name: Display name for the email greeting. 

81 

82 Returns: 

83 ``Ok(None)`` on success; ``Err(MfaNotEnabledError)`` when the 

84 factor is disabled, ``Err(EmailOtpCooldownError)`` when a resend 

85 is attempted too soon, ``Err(EmailOtpDeliveryError)`` when the 

86 code cannot be delivered. 

87 """ 

88 if not self._config.enabled: 

89 return Err(MfaNotEnabledError("Email OTP is not enabled.")) 

90 

91 last_sent_at = await self._store.last_sent_at(user_id) 

92 if last_sent_at is not None: 

93 elapsed = (datetime.now(UTC) - last_sent_at).total_seconds() 

94 if elapsed < self._config.resend_cooldown_seconds: 

95 return Err( 

96 EmailOtpCooldownError("Please wait before requesting another code.") 

97 ) 

98 

99 secret = generate_totp_secret() 

100 code = generate_totp_code(secret, period=self._config.ttl_minutes * 60) 

101 code_hash = hashlib.sha256(code.encode()).hexdigest() 

102 expires_at = datetime.now(UTC) + timedelta(minutes=self._config.ttl_minutes) 

103 await self._store.save(user_id, code_hash, expires_at) 

104 

105 if self._notification_service is None: 

106 await self._audit_failure(user_id, "no_mailer") 

107 return Err( 

108 EmailOtpDeliveryError("No email delivery service is configured.") 

109 ) 

110 

111 result = await self._notification_service.notify_email_otp( 

112 user_email=email, 

113 user_name=user_name, 

114 code=code, 

115 expires_in=f"{self._config.ttl_minutes} minutes", 

116 ) 

117 if result.is_err(): 

118 error = str(result.unwrap_err()) 

119 await self._audit_failure(user_id, error) 

120 return Err(EmailOtpDeliveryError(error)) 

121 

122 if self._audit_service is not None: 

123 await self._audit_service.log_event( 

124 event_type=AdminSecurityEventType.EMAIL_OTP_SENT, 

125 ip_address="", 

126 user_agent="", 

127 success=True, 

128 admin_user_id=user_id, 

129 metadata={"email": email}, 

130 ) 

131 return Ok(None) 

132 

133 async def verify_otp(self, user_id: str, code: str) -> Result[bool, AdminAuthError]: 

134 """Verify a code and consume it when valid. 

135 

136 Args: 

137 user_id: Admin user UUID. 

138 code: 6-digit code from the email. 

139 

140 Returns: 

141 ``Ok(True)`` when the code matched and was consumed; 

142 ``Ok(False)`` when it did not; ``Err(MfaNotEnabledError)`` when 

143 the factor is disabled. 

144 """ 

145 if not self._config.enabled: 

146 return Err(MfaNotEnabledError("Email OTP is not enabled.")) 

147 

148 code_hash = hashlib.sha256(code.encode()).hexdigest() 

149 consumed = await self._store.consume(user_id, code_hash) 

150 if not consumed: 

151 await self._audit_failure(user_id, "invalid_code") 

152 return Ok(False) 

153 return Ok(True) 

154 

155 async def _audit_failure(self, user_id: str, reason: str) -> None: 

156 """Record a failed OTP send/verify attempt (never raises).""" 

157 if self._audit_service is None: 

158 return 

159 await self._audit_service.log_event( 

160 event_type=AdminSecurityEventType.EMAIL_OTP_FAILED, 

161 ip_address="", 

162 user_agent="", 

163 success=False, 

164 admin_user_id=user_id, 

165 metadata={"reason": reason}, 

166 ) 

167 

168 

169__all__ = ["AdminEmailOtpService"]