Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/password_policy.py: 39%

1"""Password-strength policy enforcement for locally-managed proxy users. 

2 

3Applied at every path that persists a new or changed password for a DB-backed 

4user (``/user/update``, ``/user/bulk_update``, and the invitation onboarding 

5claim flow), so the strength bar is configured in one place instead of 

6per-endpoint. 

7 

8Also screens new passwords against known data breaches via the 

9haveibeenpwned.com (HIBP) k-anonymity range API: only the first 5 characters 

10of the password's SHA-1 hash ever leave the proxy, and the check fails open 

11(allows the password) when HIBP is unreachable. 

12""" 

13 

14import asyncio 

15import hashlib 

16from collections.abc import Mapping, Sequence 

17from dataclasses import dataclass 

18from types import MappingProxyType 

19from typing import Final 

20 

21from litellm._logging import verbose_proxy_logger 

22from litellm._version import version 

23from litellm.constants import HIBP_RANGE_API_BASE 

24from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_async_httpx_client 

25from litellm.proxy._types import ProxyErrorTypes, ProxyException 

26from litellm.types.llms.custom_http import httpxSpecialProvider 

27 

28HIBP_TIMEOUT_SECONDS: Final = 5.0 

29 

30DEFAULT_MIN_LENGTH: Final = 12 

31MIN_ALLOWED_LENGTH: Final = 8 

32 

33 

34def _has_uppercase(password: str) -> bool: 

35 return any(ch.isupper() for ch in password) 

36 

37 

38def _has_lowercase(password: str) -> bool: 

39 return any(ch.islower() for ch in password) 

40 

41 

42def _has_digit(password: str) -> bool: 

43 return any(ch.isdigit() for ch in password) 

44 

45 

46def _has_special_character(password: str) -> bool: 

47 """Unicode-aware: a letter or digit from ANY script counts as 

48 alphanumeric, not just ASCII, so an accented letter (e.g. the second 

49 character of "Passwörd1234") cannot be miscounted as the required 

50 special character the way an ASCII-only `[^A-Za-z0-9]` regex would.""" 

51 return any(not ch.isalnum() for ch in password) 

52 

53 

54@dataclass(frozen=True, slots=True) 

55class PasswordPolicy: 

56 min_length: int 

57 require_uppercase: bool 

58 require_lowercase: bool 

59 require_numbers: bool 

60 require_special_characters: bool 

61 

62 

63def _configured_min_length(general_settings: Mapping[str, object]) -> int: 

64 """The configured minimum, floored at MIN_ALLOWED_LENGTH so a nonpositive 

65 or too-low override (a typo, or `0`/`false` coercing through) cannot 

66 silently disable the length requirement rather than merely relaxing it.""" 

67 min_length_setting: Final = general_settings.get("password_policy_min_length") 

68 if isinstance(min_length_setting, bool) or not isinstance(min_length_setting, (int, float)): 

69 return DEFAULT_MIN_LENGTH 

70 return max(MIN_ALLOWED_LENGTH, int(min_length_setting)) 

71 

72 

73def get_password_policy(general_settings: Mapping[str, object]) -> PasswordPolicy: 

74 return PasswordPolicy( 

75 min_length=_configured_min_length(general_settings), 

76 require_uppercase=general_settings.get("password_policy_require_uppercase", True) is not False, 

77 require_lowercase=general_settings.get("password_policy_require_lowercase", True) is not False, 

78 require_numbers=general_settings.get("password_policy_require_numbers", True) is not False, 

79 require_special_characters=( 

80 general_settings.get("password_policy_require_special_characters", True) is not False 

81 ), 

82 ) 

83 

84 

85def _policy_violations(password: str, policy: PasswordPolicy) -> tuple[str, ...]: 

86 checks: Final = ( 

87 (len(password) < policy.min_length, f"be at least {policy.min_length} characters long"), 

88 (policy.require_uppercase and not _has_uppercase(password), "include an uppercase letter"), 

89 (policy.require_lowercase and not _has_lowercase(password), "include a lowercase letter"), 

90 (policy.require_numbers and not _has_digit(password), "include a number"), 

91 (policy.require_special_characters and not _has_special_character(password), "include a special character"), 

92 ) 

93 return tuple(message for failed, message in checks if failed) 

94 

95 

96def validate_password_policy(password: str, general_settings: Mapping[str, object]) -> None: 

97 """Raise ``ProxyException`` (400) if ``password`` fails the configured policy.""" 

98 policy: Final = get_password_policy(general_settings) 

99 violations: Final = _policy_violations(password, policy) 

100 if not violations: 

101 return 

102 raise ProxyException( 

103 message="Password does not meet the required policy: must " + ", ".join(violations) + ".", 

104 type=ProxyErrorTypes.validation_error, 

105 param="password", 

106 code=400, 

107 ) 

108 

109 

110def get_hibp_client() -> AsyncHTTPHandler: 

111 return get_async_httpx_client( 

112 llm_provider=httpxSpecialProvider.PasswordBreachCheck, 

113 params={"timeout": HIBP_TIMEOUT_SECONDS}, # mutable-ok: callee takes a bare dict (PEP 589) 

114 ) 

115 

116 

117def _is_suffix_in_range_response(response_body: str, hash_suffix: str) -> bool: 

118 for line in response_body.upper().splitlines(): 

119 entry_suffix, _, count = line.strip().partition(":") 

120 if entry_suffix == hash_suffix: 

121 return int(count.strip() or "0") > 0 

122 return False 

123 

124 

125async def _is_password_breached(password: str, client: AsyncHTTPHandler) -> bool: 

126 # usedforsecurity=False: SHA-1 is only a lookup key into the HIBP dataset, so no security property rests on it 

127 sha1_hex: Final = hashlib.sha1(password.encode("utf-8"), usedforsecurity=False).hexdigest().upper() 

128 headers: Final = { # mutable-ok: callee takes a bare dict (PEP 589) 

129 "Add-Padding": "true", 

130 "User-Agent": f"litellm-proxy/{version}", 

131 } 

132 try: 

133 response: Final = await client.get( 

134 f"{HIBP_RANGE_API_BASE}/{sha1_hex[:5]}", 

135 headers=headers, 

136 ) 

137 response.raise_for_status() 

138 breached: Final = _is_suffix_in_range_response(response.text, sha1_hex[5:]) 

139 except Exception as e: # noqa: BLE001 # fail-open: any HIBP failure skips the check, never breaks the caller 

140 verbose_proxy_logger.warning("Breached-password check skipped, HIBP lookup failed: %s", e) 

141 return False 

142 return breached 

143 

144 

145def is_breach_check_enabled(general_settings: Mapping[str, object]) -> bool: 

146 return general_settings.get("password_policy_check_breached_passwords", True) is not False 

147 

148 

149async def is_password_breached( 

150 password: str, 

151 general_settings: Mapping[str, object], 

152 client: AsyncHTTPHandler | None = None, 

153) -> bool: 

154 """False when the check is disabled, the password is absent from the HIBP 

155 corpus, or HIBP is unreachable (fail open).""" 

156 if not is_breach_check_enabled(general_settings): 

157 return False 

158 return await _is_password_breached(password, client if client is not None else get_hibp_client()) 

159 

160 

161def breached_password_error() -> ProxyException: 

162 return ProxyException( 

163 message=( 

164 "This password appears in known data breaches and cannot be used. Please choose a different password." 

165 ), 

166 type=ProxyErrorTypes.validation_error, 

167 param="password", 

168 code=400, 

169 ) 

170 

171 

172async def validate_password_not_breached( 

173 password: str, 

174 general_settings: Mapping[str, object], 

175 client: AsyncHTTPHandler | None = None, 

176) -> None: 

177 """Raise ``ProxyException`` (400) if ``password`` appears in a known data breach. 

178 

179 Fails open: an unreachable or misbehaving HIBP allows the password.""" 

180 if not await is_password_breached(password, general_settings, client): 

181 return 

182 raise breached_password_error() 

183 

184 

185def _strength_verdict(password: str, general_settings: Mapping[str, object]) -> ProxyException | None: 

186 try: 

187 validate_password_policy(password, general_settings) 

188 except ProxyException as e: 

189 return e 

190 return None 

191 

192 

193async def validate_passwords_bulk( 

194 passwords: Sequence[str], 

195 general_settings: Mapping[str, object], 

196 client: AsyncHTTPHandler | None = None, 

197) -> Mapping[str, ProxyException | None]: 

198 """Per-unique-password policy verdicts for a batch: the ProxyException to 

199 surface, or None when the password is acceptable. 

200 

201 Deduplicates first, then issues every needed HIBP lookup concurrently, so a 

202 batch caller pays one HIBP timeout window in the worst case instead of one 

203 per password (each lookup still fails open independently).""" 

204 unique_passwords: Final = tuple(dict.fromkeys(passwords)) 

205 strength_verdicts: Final[Mapping[str, ProxyException | None]] = MappingProxyType( 

206 {password: _strength_verdict(password, general_settings) for password in unique_passwords} 

207 ) 

208 to_screen: Final = tuple(password for password in unique_passwords if strength_verdicts[password] is None) 

209 breached_flags: Final = await asyncio.gather( 

210 *(is_password_breached(password, general_settings, client) for password in to_screen) 

211 ) 

212 breached_passwords: Final = frozenset(password for password, breached in zip(to_screen, breached_flags) if breached) 

213 return MappingProxyType( 

214 { 

215 password: breached_password_error() if password in breached_passwords else strength_verdicts[password] 

216 for password in unique_passwords 

217 } 

218 )