Coverage for src/lexigram/auth/authn/account_verification.py: 72%

1"""Account verification service for Lexigram Auth.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime, timedelta 

6import secrets 

7from typing import TYPE_CHECKING, Any 

8 

9from lexigram.auth.exceptions import AlreadyVerifiedError, UserNotFoundError 

10from lexigram.di.decorators import inject 

11from lexigram.logging import get_logger 

12from lexigram.primitives import clock as ambient_clock 

13from lexigram.result import Err, Ok, Result 

14 

15if TYPE_CHECKING: 

16 from lexigram.auth.storage.token_store import UserStoreProtocol 

17 from lexigram.contracts.auth.exceptions import ( 

18 VerificationError as ContractsVerificationError, 

19 ) 

20 

21logger = get_logger(__name__) 

22 

23 

24class AccountVerificationError(Exception): 

25 """Error during account verification operations.""" 

26 

27 _code = "LEX_ERR_AUTH_027" 

28 

29 

30@inject 

31class AccountVerificationService: 

32 """Service for handling account verification flows. 

33 

34 Provides email verification with time-limited tokens. 

35 """ 

36 

37 TOKEN_EXPIRY_DAYS = 7 

38 TOKEN_LENGTH = 32 

39 

40 def __init__( 

41 self, 

42 user_store: UserStoreProtocol, 

43 token_ttl_days: int = TOKEN_EXPIRY_DAYS, 

44 ) -> None: 

45 self._store = user_store 

46 self._token_ttl = timedelta(days=token_ttl_days) 

47 

48 def generate_verification_token(self) -> tuple[str, datetime]: 

49 """Generate a secure random verification token. 

50 

51 Returns: 

52 Tuple of (token, expiry_datetime) 

53 """ 

54 token = secrets.token_urlsafe(self.TOKEN_LENGTH) 

55 expiry = (ambient_clock.now()) + self._token_ttl 

56 return token, expiry 

57 

58 async def send_verification( 

59 self, user_id: str 

60 ) -> Result[tuple[str, datetime], AlreadyVerifiedError | UserNotFoundError]: 

61 """Send verification email for a user. 

62 

63 Args: 

64 user_id: The user's ID. 

65 

66 Returns: 

67 Ok((token, expiry)) if the verification token was generated. 

68 Err(UserNotFoundError) if the user does not exist. 

69 Err(AlreadyVerifiedError) if the user is already verified. 

70 """ 

71 user = await self._store.get_user_by_id(user_id) 

72 

73 if not user: 

74 logger.debug("Verification requested for non-existent user: %s", user_id) 

75 return Err(UserNotFoundError(user_id)) 

76 

77 if getattr(user, "is_verified", False): 

78 logger.debug("User already verified: %s", user_id) 

79 return Err(AlreadyVerifiedError(user_id)) 

80 

81 token, expiry = self.generate_verification_token() 

82 

83 user.verification_token = token # type: ignore[attr-defined] 

84 user.verification_expires_at = expiry # type: ignore[attr-defined] 

85 

86 await self._store.update_user(user) 

87 

88 logger.info("Verification token generated for user: %s", user_id) 

89 return Ok((token, expiry)) 

90 

91 async def verify(self, token: str) -> Result[None, ContractsVerificationError]: 

92 """Verify user account with token. 

93 

94 Args: 

95 token: The verification token. 

96 

97 Returns: 

98 Ok(None) if verification succeeded. 

99 Err(VerificationError) if the token is invalid or expired. 

100 """ 

101 from lexigram.contracts.auth.exceptions import VerificationError as VErr 

102 

103 user = await self._find_user_by_token(token) 

104 

105 if not user: 

106 logger.warning("Invalid verification token used") 

107 return Err(VErr("Invalid or expired verification token")) 

108 

109 user.is_verified = True 

110 user.verification_token = None 

111 user.verification_expires_at = None 

112 

113 await self._store.update_user(user) 

114 

115 logger.info("Account verified for user: %s", user.user_id) 

116 return Ok(None) 

117 

118 async def _find_user_by_token(self, token: str) -> Any | None: 

119 """Find user by verification token.""" 

120 users = await self._store.list_users(skip=0, limit=1000) 

121 

122 for user in users: 

123 if hasattr(user, "verification_token") and user.verification_token == token: 

124 if hasattr(user, "verification_expires_at") and ( 

125 user.verification_expires_at 

126 and user.verification_expires_at > (ambient_clock.now()) 

127 ): 

128 return user 

129 break 

130 

131 return None 

132 

133 async def resend_verification( 

134 self, email: str 

135 ) -> Result[tuple[str, datetime], AlreadyVerifiedError | UserNotFoundError]: 

136 """Resend verification token. 

137 

138 Args: 

139 email: The user's email. 

140 

141 Returns: 

142 Ok((token, expiry)) if notified successfully. 

143 Err(UserNotFoundError) if the email is not registered. 

144 Err(AlreadyVerifiedError) if the user is already verified. 

145 """ 

146 user = await self._store.get_user_by_email(email) 

147 

148 if not user: 

149 return Err(UserNotFoundError(email)) 

150 

151 if getattr(user, "is_verified", False): 

152 return Err(AlreadyVerifiedError(user.user_id)) 

153 

154 return await self.send_verification(user.user_id) 

155 

156 def __repr__(self) -> str: 

157 return f"AccountVerificationService(user_store={type(self._store).__name__})" 

158 

159 

160__all__ = [ 

161 "AccountVerificationError", 

162 "AccountVerificationService", 

163 "logger", 

164]