Coverage for src/lexigram/auth/authn/security.py: 81%
146 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 00:58 +0800
1"""Password security utilities using Passlib with a lightweight fallback."""
3from __future__ import annotations
5import asyncio
6import base64
7import hashlib
8import importlib
9from pathlib import Path
10from types import ModuleType
11from typing import TYPE_CHECKING, Any, cast
13from lexigram.contracts.auth import PasswordHasherProtocol, PasswordPolicyProtocol
14from lexigram.logging import get_logger
16if TYPE_CHECKING:
17 from lexigram.auth.config import PasswordConfig
19__all__ = [
20 "DUMMY_PASSWORD_HASH",
21 "PasswordHasher",
22 "PasswordPolicy",
23 "UnknownHashError",
24]
26logger = get_logger(__name__)
28try:
29 _bcrypt: ModuleType | None = importlib.import_module("bcrypt")
30except ImportError:
31 _bcrypt = None
32bcrypt_available = _bcrypt is not None
34if not bcrypt_available:
35 logger.warning("bcrypt not available - password hashing will be limited")
37try:
38 pyotp: ModuleType | None = importlib.import_module("pyotp")
39except ImportError:
40 pyotp = None
41pyotp_available = pyotp is not None
43UnknownHashError = Exception
46_MAX_PASSWORD_BYTES = 72
47_WARN_PASSWORD_BYTES = 64
48_DEFAULT_BCRYPT_ROUNDS = 12
51def _prehash(password: str) -> str:
52 """Pre-hash *password* with SHA-256 to avoid bcrypt's 72-byte truncation.
54 The SHA-256 digest (32 bytes) is base64-encoded to a 44-byte ASCII string,
55 which is always well within bcrypt's 72-byte limit. This must be applied
56 consistently in both :meth:`PasswordHasher.hash` and
57 :meth:`PasswordHasher.verify` so that hashes computed with pre-hashing can
58 be verified correctly.
60 Args:
61 password: Plain-text password string.
63 Returns:
64 Base64-encoded SHA-256 digest of the UTF-8 encoded password.
65 """
66 return base64.b64encode(hashlib.sha256(password.encode()).digest()).decode()
69def _prepare_password_bytes(password: str) -> bytes:
70 """Return the bytes that bcrypt should hash for *password*.
72 * Passwords **≤ 64 bytes**: encoded directly — no transformation.
73 * Passwords **> 64 bytes and ≤ 72 bytes**: a structured warning is emitted
74 because these are in the borderline range where bcrypt's limit is close.
75 * Passwords **> 72 bytes**: pre-hashed with SHA-256 via :func:`_prehash`
76 so that bcrypt sees a 44-byte ASCII string rather than silently
77 truncating the input.
79 Args:
80 password: Plain-text password string.
82 Returns:
83 Bytes ready to pass to ``bcrypt.hashpw`` / ``bcrypt.checkpw``.
85 Raises:
86 RuntimeError: If bcrypt is not installed.
87 """
88 if not bcrypt_available:
89 raise RuntimeError("bcrypt library is not available")
91 password_bytes = password.encode("utf-8")
92 byte_length = len(password_bytes)
94 if byte_length > _MAX_PASSWORD_BYTES:
95 # Pre-hash to avoid silent bcrypt truncation.
96 return _prehash(password).encode("ascii")
98 if byte_length > _WARN_PASSWORD_BYTES:
99 logger.warning(
100 "password_near_bcrypt_limit",
101 byte_length=byte_length,
102 limit=_MAX_PASSWORD_BYTES,
103 message=(
104 "Password is between 64 and 72 bytes; "
105 "it is close to bcrypt's 72-byte truncation limit."
106 ),
107 )
109 return password_bytes
112class PasswordHasher(PasswordHasherProtocol):
113 """Bcrypt password hasher implementing the PasswordHasherProtocol.
115 Provides secure password hashing using bcrypt with UTF-8 aware truncation.
116 The bcrypt cost factor is per-instance (``rounds``) so the configured
117 value is honored both at construction and by the container singleton.
118 """
120 MAX_PASSWORD_BYTES = _MAX_PASSWORD_BYTES
122 def __init__(self, rounds: int = _DEFAULT_BCRYPT_ROUNDS) -> None:
123 self._rounds = rounds
125 async def hash(self, password: str) -> str:
126 """Hash a password using bcrypt with UTF-8 aware truncation.
128 Args:
129 password: Plain text password.
131 Returns:
132 Bcrypt hash string (``$2b$<cost>$...``) using the configured
133 cost factor.
135 Raises:
136 RuntimeError: If bcrypt is not available.
137 """
138 if not bcrypt_available:
139 raise RuntimeError("bcrypt library is not available")
141 password_bytes = _prepare_password_bytes(password)
143 def _hash_sync(pwd_bytes: bytes) -> str:
144 bcrypt = cast("Any", _bcrypt)
145 salt = bcrypt.gensalt(rounds=self._rounds)
146 hashed = bcrypt.hashpw(pwd_bytes, salt)
147 return str(hashed.decode("ascii"))
149 return await asyncio.to_thread(_hash_sync, password_bytes)
151 async def verify(self, password: str, hashed_password: str | bytes) -> bool:
152 """Verify a password against its hash asynchronously.
154 Args:
155 password: Plain text password.
156 hashed_password: Stored hash to compare against.
158 Returns:
159 True if the password matches the hash, False otherwise.
161 Raises:
162 RuntimeError: If bcrypt is not available.
163 """
164 if not bcrypt_available:
165 raise RuntimeError("bcrypt library is not available")
167 def _verify_sync(pwd_bytes: bytes, hash_bytes: bytes) -> bool:
168 try:
169 bcrypt = cast("Any", _bcrypt)
170 return bool(bcrypt.checkpw(pwd_bytes, hash_bytes))
171 except ValueError:
172 return False
174 password_bytes = _prepare_password_bytes(password)
176 if isinstance(hashed_password, str):
177 hashed_bytes = hashed_password.encode("ascii")
178 else:
179 hashed_bytes = hashed_password
181 return await asyncio.to_thread(_verify_sync, password_bytes, hashed_bytes)
183 def needs_rehash(self, hashed_password: str) -> bool:
184 """Compare the stored hash's cost to the configured target.
186 Parses the self-describing bcrypt prefix (``$2b$<cost>$...``); returns
187 ``True`` when the stored cost is below the configured ``rounds``.
188 Unparseable or unknown formats return ``True`` (fail-closed) — safe
189 because rehashing only ever runs after a successful ``verify()``.
191 Args:
192 hashed_password: Stored bcrypt hash string.
194 Returns:
195 True when the hash should be re-computed at the current cost.
196 """
197 if not isinstance(hashed_password, str) or not hashed_password:
198 return True
199 try:
200 parts = hashed_password.split("$")
201 if len(parts) >= 3 and parts[1] in ("2a", "2b", "2y", "2x"):
202 cost = int(parts[2])
203 return cost < self._rounds
204 except (ValueError, IndexError, TypeError):
205 logger.warning(
206 "password_hash_cost_unparseable",
207 hash_prefix=hashed_password[:7],
208 )
209 return True
211 async def rehash_if_needed(
212 self,
213 password: str,
214 hashed_password: str | None,
215 ) -> str | None:
216 """Rehash the password if needed."""
217 if not hashed_password:
218 return None
219 if self.needs_rehash(hashed_password):
220 return await self.hash(password)
221 return None
224# A constant dummy hash used to ensure password verification is always executed
225# to prevent timing side-channels that reveal whether a username exists.
226# This must be a valid bcrypt hash string to avoid verification errors.
227#
228# SECURITY: This is intentionally weak (not a real hash) - it is used ONLY to
229# ensure that password verification runs for non-existent users, preventing
230# attackers from distinguishing between "user not found" and "wrong password"
231# via timing attacks. The hash is valid bcrypt format but the password that
232# produces it is unknown/impossible to guess.
233DUMMY_PASSWORD_HASH = "$2b$12$OMAqo55i5DcmvOMAqo55i5DcmvOMAqo55i5DcmvOMAqo55i5Dcmv" # noqa: S105 # timing-equalization dummy, no real password matches
236class PasswordPolicy(PasswordPolicyProtocol):
237 """Password policy configuration.
239 Lazy-loads common passwords file only when needed.
240 Implements PasswordPolicyProtocol for dependency injection compatibility.
241 """
243 _DEFAULT_COMMON_PASSWORDS: set[str] | None = None
245 def __init__(
246 self,
247 min_length: int = 8, # NIST SP 800-63B minimum
248 max_length: int = 128,
249 require_uppercase: bool = True,
250 require_lowercase: bool = True,
251 require_digits: bool = False,
252 require_special: bool = False,
253 prevent_common: bool = True, # NIST SP 800-63B: check against breach lists
254 prevent_reuse: bool = False,
255 history_size: int = 5,
256 common_passwords_file: str | None = None,
257 banned_patterns: list[str] | None = None,
258 ):
259 self.min_length = min_length
260 self.max_length = max_length
261 self.require_uppercase = require_uppercase
262 self.require_lowercase = require_lowercase
263 self.require_digits = require_digits
264 self.require_special = require_special
265 self.prevent_common = prevent_common
266 self.prevent_reuse = prevent_reuse
267 self.history_size = history_size
268 self._common_passwords_file = common_passwords_file
269 self._common_passwords: set[str] | None = None
270 self.banned_patterns: list[str] = banned_patterns or []
272 @classmethod
273 def from_config(cls, config: PasswordConfig) -> PasswordPolicy:
274 """Build a ``PasswordPolicy`` from a ``PasswordConfig`` dataclass.
276 Args:
277 config: Password complexity configuration from ``AuthConfig.password``.
279 Returns:
280 A configured :class:`PasswordPolicy` instance.
281 """
282 return cls(
283 min_length=config.min_length,
284 max_length=config.max_length,
285 require_uppercase=config.require_uppercase,
286 require_lowercase=config.require_lowercase,
287 require_digits=config.require_digits,
288 require_special=config.require_special,
289 banned_patterns=list(config.banned_patterns),
290 )
292 def _load_common_passwords(self, file_path: str | None) -> set[str]:
293 """Load common passwords from file (lazy-loaded)."""
294 if not file_path:
295 if PasswordPolicy._DEFAULT_COMMON_PASSWORDS is None:
296 PasswordPolicy._DEFAULT_COMMON_PASSWORDS = {
297 "password",
298 "password1",
299 "password123",
300 "123456",
301 "123456789",
302 "12345678",
303 "1234567890",
304 "qwerty",
305 "qwerty123",
306 "abc123",
307 "letmein",
308 "monkey",
309 "dragon",
310 "master",
311 "sunshine",
312 "princess",
313 "welcome",
314 "shadow",
315 "superman",
316 "michael",
317 "football",
318 "baseball",
319 "iloveyou",
320 "trustno1",
321 "hunter2",
322 "admin",
323 "admin123",
324 "administrator",
325 "root",
326 "toor",
327 "passw0rd",
328 "p@ssword",
329 "p@ssw0rd",
330 "pass@word",
331 "test",
332 "test123",
333 "demo",
334 "demo123",
335 "guest",
336 "guest123",
337 "login",
338 "login123",
339 "changeme",
340 "change_me",
341 "default",
342 "secret",
343 "secret123",
344 "temp",
345 "temp123",
346 "temporary",
347 "letmein1",
348 "letmein123",
349 "qwertyuiop",
350 "asdfghjkl",
351 "zxcvbnm",
352 "1q2w3e4r",
353 "1q2w3e",
354 "11111111",
355 "22222222",
356 "33333333",
357 "00000000",
358 "111111111",
359 "password2",
360 "password1!",
361 "p@ssword1",
362 "admin@123",
363 "welcome1",
364 "welcome@1",
365 "hello123",
366 "summer2023",
367 "winter2023",
368 "spring2023",
369 "autumn2023",
370 "january1",
371 "february1",
372 "march2023",
373 "welcome123",
374 }
375 return PasswordPolicy._DEFAULT_COMMON_PASSWORDS
377 try:
378 with Path(file_path).open() as f:
379 return {line.strip().lower() for line in f if line.strip()}
380 except FileNotFoundError:
381 return set()
383 def _get_common_passwords(self) -> set[str]:
384 """Return common passwords set, loading lazily on first access."""
385 if self._common_passwords is None:
386 self._common_passwords = self._load_common_passwords(
387 self._common_passwords_file
388 )
389 return self._common_passwords
391 def validate(self, password: str) -> None:
392 """Validate password against policy.
394 Args:
395 password: Plain text password to validate.
397 Raises:
398 ValueError: If the password violates the policy.
399 """
400 errors = []
402 # Length check
403 if len(password) < self.min_length:
404 errors.append(
405 f"Password must be at least {self.min_length} characters long",
406 )
408 if len(password) > self.max_length:
409 errors.append(
410 f"Password must be at most {self.max_length} characters long",
411 )
413 # Banned pattern check
414 lower = password.lower()
415 for pattern in self.banned_patterns:
416 if pattern.lower() in lower:
417 errors.append(
418 f"Password must not contain the substring '{pattern}'",
419 )
421 # Character requirements
422 if self.require_uppercase and not any(c.isupper() for c in password):
423 errors.append("Password must contain at least one uppercase letter")
425 if self.require_lowercase and not any(c.islower() for c in password):
426 errors.append("Password must contain at least one lowercase letter")
428 if self.require_digits and not any(c.isdigit() for c in password):
429 errors.append("Password must contain at least one digit")
431 if self.require_special and not any(not c.isalnum() for c in password):
432 errors.append("Password must contain at least one special character")
434 # Common password check
435 if self.prevent_common and password.lower() in self._get_common_passwords():
436 errors.append("Password is too common, please choose a different one")
438 if errors:
439 raise ValueError("; ".join(errors))
441 def is_valid(self, password: str) -> bool:
442 """Return True if the password satisfies the policy without raising.
444 Args:
445 password: Plain text password.
447 Returns:
448 True if valid, False otherwise.
449 """
450 try:
451 self.validate(password)
452 return True
453 except ValueError:
454 return False