Coverage for src/lexigram/auth/authn/password_hasher.py: 69%

1"""Argon2id-based password hashing implementations. 

2 

3This module provides: 

4- Argon2idKeyDerivation: implements KeyDerivationProtocol (core security) 

5- Argon2idPasswordHasher: implements PasswordHasherProtocol (auth-domain) 

6- ComposedPasswordHasher: Argon2id-default hasher with a bcrypt legacy shim 

7""" 

8 

9from __future__ import annotations 

10 

11import asyncio 

12import re 

13from typing import TYPE_CHECKING, Any, cast 

14 

15from lexigram.contracts.auth import PasswordHasherProtocol 

16from lexigram.contracts.security.protocols import KeyDerivationProtocol 

17from lexigram.logging import get_logger 

18 

19if TYPE_CHECKING: 

20 from lexigram.auth.config import PasswordConfig 

21 

22__all__ = [ 

23 "Argon2idKeyDerivation", 

24 "Argon2idPasswordHasher", 

25 "ComposedPasswordHasher", 

26] 

27 

28logger = get_logger(__name__) 

29 

30_ARGON2_MEMORY_FLOOR = 19456 

31 

32 

33_argon2: Any 

34try: 

35 import argon2 

36 import argon2.exceptions 

37 

38 _argon2 = argon2 

39 _argon2_available = True 

40except ImportError: 

41 _argon2 = None 

42 _argon2_available = False 

43 

44 

45class Argon2idKeyDerivation(KeyDerivationProtocol): 

46 """Argon2id key derivation implementation. 

47 

48 Implements KeyDerivationProtocol using argon2-cffi. 

49 Follows OWASP 2024 recommendations for parameters. 

50 """ 

51 

52 def __init__(self, config: PasswordConfig | None = None) -> None: 

53 if not _argon2_available: 

54 raise RuntimeError("argon2-cffi is not installed") 

55 self._ph = _argon2.PasswordHasher( 

56 memory_cost=65536, # 64 MiB 

57 time_cost=3, 

58 parallelism=4, 

59 hash_len=32, 

60 salt_len=16, 

61 ) 

62 

63 async def derive(self, secret: str, *, salt: bytes | None = None) -> str: 

64 """Derive a key from a secret using Argon2id.""" 

65 

66 def _derive_sync() -> str: 

67 return cast("str", self._ph.hash(secret)) 

68 

69 return await asyncio.to_thread(_derive_sync) 

70 

71 async def verify(self, secret: str, encoded: str) -> bool: 

72 """Verify a secret against an Argon2id hash.""" 

73 

74 def _verify_sync() -> bool: 

75 try: 

76 self._ph.verify(encoded, secret) 

77 return True 

78 except ( 

79 _argon2.exceptions.VerifyMismatchError, 

80 _argon2.exceptions.VerificationError, 

81 ): 

82 return False 

83 

84 return await asyncio.to_thread(_verify_sync) 

85 

86 async def hash(self, secret: str, *, salt: bytes | None = None) -> str: 

87 """Backward-compatible alias for derive.""" 

88 return await self.derive(secret, salt=salt) 

89 

90 

91class Argon2idPasswordHasher(PasswordHasherProtocol): 

92 """Auth-domain password hasher using Argon2id. 

93 

94 Implements PasswordHasherProtocol (hash/verify) and delegates 

95 to KeyDerivationProtocol internally. 

96 """ 

97 

98 def __init__(self, kdf: KeyDerivationProtocol) -> None: 

99 self._kdf = kdf 

100 

101 async def hash(self, password: str) -> str: 

102 """Hash a password using Argon2id.""" 

103 return await self._kdf.derive(password) 

104 

105 async def verify(self, password: str, hashed_password: str) -> bool: 

106 """Verify a password against its hash.""" 

107 return await self._kdf.verify(password, hashed_password) 

108 

109 def needs_rehash(self, hashed_password: str) -> bool: 

110 """Compare the stored hash's memory cost against the OWASP floor. 

111 

112 Parses the Argon2id encoded prefix (``$argon2id$v=19$m=...,t=...,p=...``); 

113 returns ``True`` when the stored memory cost is below the 19456 KiB 

114 floor. Unparseable or unknown formats return ``True`` (fail-closed). 

115 

116 Args: 

117 hashed_password: Stored Argon2id hash string. 

118 

119 Returns: 

120 True when the hash should be re-computed at current parameters. 

121 """ 

122 if not isinstance(hashed_password, str) or not hashed_password: 

123 return True 

124 try: 

125 parts = hashed_password.split("$") 

126 if parts[1] == "argon2id": 

127 match = re.search(r"m=(\d+)", parts[3]) 

128 if match: 

129 return int(match.group(1)) < _ARGON2_MEMORY_FLOOR 

130 except (ValueError, IndexError, TypeError): 

131 logger.warning( 

132 "password_hash_cost_unparseable", 

133 hash_prefix=hashed_password[:7], 

134 ) 

135 return True 

136 

137 async def rehash_if_needed( 

138 self, 

139 password: str, 

140 hashed_password: str | None, 

141 ) -> str | None: 

142 """Rehash the password when the stored hash is below the cost floor. 

143 

144 Args: 

145 password: Plain text password (already verified). 

146 hashed_password: Stored hash string, or None. 

147 

148 Returns: 

149 A fresh hash when an upgrade is needed, else None. 

150 """ 

151 if not hashed_password: 

152 return None 

153 if self.needs_rehash(hashed_password): 

154 return await self.hash(password) 

155 return None 

156 

157 

158class ComposedPasswordHasher(PasswordHasherProtocol): 

159 """Argon2id-default composed hasher with a bcrypt legacy shim (ODD-1 A). 

160 

161 New hashes use the primary (Argon2id) hasher. Stored bcrypt hashes 

162 continue to verify through the legacy shim and are flagged by 

163 :meth:`needs_rehash` (algorithm differs from the default) so 

164 ``rehash_if_needed`` upgrades them on the user's next successful login. 

165 """ 

166 

167 def __init__( 

168 self, 

169 primary: PasswordHasherProtocol, 

170 legacy: PasswordHasherProtocol, 

171 ) -> None: 

172 self._primary = primary 

173 self._legacy = legacy 

174 

175 async def hash(self, password: str) -> str: 

176 """Hash a password with the primary (Argon2id) hasher.""" 

177 return await self._primary.hash(password) 

178 

179 async def verify(self, password: str, hashed_password: str) -> bool: 

180 """Verify a password, dispatching on the stored hash's algorithm. 

181 

182 Bcrypt-prefixed (``$2*$``) hashes route to the legacy shim; Argon2id 

183 hashes route to the primary hasher; anything else fails closed. 

184 """ 

185 if not isinstance(hashed_password, str) or not hashed_password: 

186 return False 

187 if hashed_password.startswith("$2"): 

188 return await self._legacy.verify(password, hashed_password) 

189 return await self._primary.verify(password, hashed_password) 

190 

191 def needs_rehash(self, hashed_password: str) -> bool: 

192 """Return True when the stored hash is not at current parameters. 

193 

194 Argon2id hashes below the memory floor are flagged by the primary; 

195 any bcrypt hash differs from the Argon2id default and is upgraded on 

196 the next successful login. Unknown formats return True (fail-closed). 

197 """ 

198 if not isinstance(hashed_password, str) or not hashed_password: 

199 return True 

200 if hashed_password.startswith("$argon2id$"): 

201 return self._primary.needs_rehash(hashed_password) 

202 return True 

203 

204 async def rehash_if_needed( 

205 self, 

206 password: str, 

207 hashed_password: str | None, 

208 ) -> str | None: 

209 """Rehash the password when the stored hash is below the cost target. 

210 

211 Args: 

212 password: Plain text password (already verified). 

213 hashed_password: Stored hash string, or None. 

214 

215 Returns: 

216 A fresh hash when an upgrade is needed, else None. 

217 """ 

218 if not hashed_password: 

219 return None 

220 if self.needs_rehash(hashed_password): 

221 return await self.hash(password) 

222 return None