Coverage for src/lexigram/auth/authn/password_reset.py: 41%

1"""Password reset 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.authn.security import PasswordHasher 

10from lexigram.auth.models.user import UserCredentials 

11from lexigram.di.decorators import inject 

12from lexigram.logging import get_logger 

13from lexigram.primitives import clock as ambient_clock 

14from lexigram.result import Err, Ok, Result 

15 

16if TYPE_CHECKING: 

17 from lexigram.auth.storage.token_store import UserStoreProtocol 

18 

19logger = get_logger(__name__) 

20 

21 

22class PasswordResetTokenError(Exception): 

23 """Error during password reset token operations.""" 

24 

25 _code = "LEX_ERR_AUTH_028" 

26 

27 

28@inject 

29class PasswordResetService: 

30 """Service for handling password reset flows. 

31 

32 Provides secure password reset with time-limited tokens. 

33 """ 

34 

35 TOKEN_EXPIRY_HOURS = 24 

36 TOKEN_LENGTH = 32 

37 

38 def __init__( 

39 self, 

40 user_store: UserStoreProtocol, 

41 token_ttl_hours: int = TOKEN_EXPIRY_HOURS, 

42 ) -> None: 

43 self._store = user_store 

44 self._token_ttl = timedelta(hours=token_ttl_hours) 

45 

46 def generate_reset_token(self) -> tuple[str, datetime]: 

47 """Generate a secure random reset token. 

48 

49 Returns: 

50 Tuple of (token, expiry_datetime) 

51 """ 

52 token = secrets.token_urlsafe(self.TOKEN_LENGTH) 

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

54 return token, expiry 

55 

56 async def request_reset( 

57 self, email: str 

58 ) -> Result[tuple[str, datetime] | None, PasswordResetTokenError]: 

59 """Request a password reset for a user email. 

60 

61 Args: 

62 email: The user's email address. 

63 

64 Returns: 

65 ``Ok((token, expiry))`` when the user exists and a token was created. 

66 ``Ok(None)`` when no user with that email exists (silent, prevents 

67 email enumeration). 

68 ``Err(PasswordResetTokenError)`` on store failure. 

69 """ 

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

71 

72 if not user: 

73 logger.debug("Password reset requested for non-existent email: %s", email) 

74 return Ok(None) 

75 

76 token, expiry = self.generate_reset_token() 

77 

78 await self._store.update_user(user) 

79 

80 logger.info("Password reset token generated for user: %s", user.user_id) 

81 return Ok((token, expiry)) 

82 

83 async def confirm_reset( 

84 self, token: str, new_password: str 

85 ) -> Result[None, PasswordResetTokenError]: 

86 """Confirm password reset with new password. 

87 

88 Args: 

89 token: The reset token. 

90 new_password: The new password to set. 

91 

92 Returns: 

93 ``Ok(None)`` when the reset succeeded. 

94 ``Err(PasswordResetTokenError)`` when the token is invalid or expired. 

95 """ 

96 user = await self._find_user_by_token(token) 

97 

98 if not user: 

99 logger.warning("Invalid password reset token used") 

100 return Err(PasswordResetTokenError("Invalid or expired reset token")) 

101 

102 hasher = PasswordHasher() 

103 hashed_password = await hasher.hash(new_password) 

104 

105 new_creds = UserCredentials( 

106 user_id=user.user_id, 

107 hashed_password=hashed_password, 

108 ) 

109 await self._store.update_credentials(new_creds) 

110 

111 logger.info("Password reset completed for user: %s", user.user_id) 

112 return Ok(None) 

113 

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

115 """Find user by reset token.""" 

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

117 

118 for user in users: 

119 if ( 

120 hasattr(user, "password_reset_token") 

121 and user.password_reset_token == token 

122 ): 

123 if hasattr(user, "password_reset_expires_at") and ( 

124 user.password_reset_expires_at 

125 and user.password_reset_expires_at > (ambient_clock.now()) 

126 ): 

127 return user 

128 break 

129 

130 return None 

131 

132 async def invalidate_token(self, user_id: str) -> None: 

133 """Invalidate any pending reset token for a user.""" 

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

135 

136 if user: 

137 user.password_reset_token = None # type: ignore[attr-defined] 

138 user.password_reset_expires_at = None # type: ignore[attr-defined] 

139 await self._store.update_user(user) 

140 

141 def __repr__(self) -> str: 

142 return f"PasswordResetService(user_store={type(self._store).__name__})" 

143 

144 

145__all__ = [ 

146 "PasswordResetService", 

147 "PasswordResetTokenError", 

148 "logger", 

149]