Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/login_throttle.py: 39%
243 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""Failed-login accounting for the Admin UI sign-in path.
3Wrong passwords are counted over a short window per source address and per source-and-username
4pair; too many in one window blocks that key for a fixed time. While a key is blocked every attempt
5from it, right or wrong, is refused with 429 before the password is checked. A blocked pair stops
6counting against its source, so one script stuck on one account does not block the whole office.
7Recovery is the master key over the API, which never passes through here, or waiting out the block.
8"""
10from __future__ import annotations
12import asyncio
13import hashlib
14import ipaddress
15import math
16import time
17from collections.abc import Mapping
18from dataclasses import dataclass
19from functools import cache
20from typing import Final, Literal, NamedTuple, Protocol, TypeAlias
22from fastapi import Request, status
23from pydantic import TypeAdapter, ValidationError
24from redis.exceptions import RedisError
26from litellm._logging import verbose_proxy_logger
27from litellm.caching.in_memory_cache import InMemoryCache
28from litellm.caching.redis_cache import RedisCache, RedisCircuitBreakerOpenError
29from litellm.constants import (
30 EMPTY_MAPPING,
31 LOGIN_THROTTLE_CACHE_KEY_PREFIX,
32 LOGIN_THROTTLE_MAX_TRACKED_BLOCKS,
33 LOGIN_THROTTLE_MAX_TRACKED_COUNTERS,
34 LOGIN_THROTTLE_NOT_BLOCKED,
35 LOGIN_THROTTLE_UNKNOWN_SOURCE,
36)
37from litellm.proxy._types import ProxyErrorTypes, ProxyException
38from litellm.proxy.auth.network import TrustedProxyConfig, resolve_client_ip
39from litellm.secret_managers.main import get_secret_bool
41DEFAULT_MAX_FAILED_LOGIN_ATTEMPTS_PER_SOURCE: Final = 10
42DEFAULT_FAILED_LOGIN_WINDOW_SECONDS: Final = 60
43DEFAULT_FAILED_LOGIN_BLOCK_SECONDS: Final = 300
45IPV6_SOURCE_PREFIX_LENGTH: Final = 64
46EXEMPT: Final = 0
48SOURCE_LIMIT_KEY: Final = "max_failed_login_attempts_per_source"
49SOURCE_LIMIT_OVERRIDES_KEY: Final = "max_failed_login_attempts_per_source_overrides"
50WINDOW_KEY: Final = "failed_login_window_seconds"
51BLOCK_KEY: Final = "failed_login_block_seconds"
52TRUSTED_PROXY_RANGES_KEY: Final = "trusted_proxy_ranges"
54_REDIS_FAILURES: Final = (RedisError, RedisCircuitBreakerOpenError, OSError, asyncio.TimeoutError)
55_LOCAL_BLOCK_EXPIRY: Final = TypeAdapter[float | None](float | None)
56_SOURCE_LIMIT_OVERRIDES: Final = TypeAdapter[Mapping[str, object]](Mapping[str, object])
57_RANGE_ENTRIES: Final = TypeAdapter[tuple[object, ...]](tuple[object, ...])
59Scope: TypeAlias = Literal["user", "source"]
61_BlockTtls: TypeAlias = tuple[int, int]
62_LUA_BLOCK_TTLS: Final = TypeAdapter[_BlockTtls](_BlockTtls)
63_Network: TypeAlias = ipaddress.IPv4Network | ipaddress.IPv6Network
66class LocalStore(Protocol):
67 """The per-worker store behind the counters and blocks; ``InMemoryCache`` satisfies it."""
69 def get_cache(self, key: str) -> object: ... 69 ↛ exitline 69 didn't return from function 'get_cache' because
71 def set_cache(self, key: str, value: float, *, ttl: int) -> None: ... 71 ↛ exitline 71 didn't return from function 'set_cache' because
73 def increment_cache(self, key: str, value: float, *, ttl: int) -> float: ... 73 ↛ exitline 73 didn't return from function 'increment_cache' because
75 def delete_cache(self, key: str) -> None: ... 75 ↛ exitline 75 didn't return from function 'delete_cache' because
78# KEYS: pair counter, pair block, source counter, source block (one cluster slot via the source hash tag)
79# ARGV: pair limit, source limit (0 = source scope off), window seconds, block seconds
80# Both scripts return {pair block TTL, source block TTL}; 0 or below means not blocked
81_BLOCK_TTLS_LUA: Final = "return {redis.call('TTL', KEYS[2]), redis.call('TTL', KEYS[4])}"
82_RECORD_FAILURE_LUA: Final = (
83 "local function bump(count_key, block_key, limit) "
84 "local blocked = redis.call('TTL', block_key) "
85 "if blocked > 0 then return blocked end "
86 "local count = redis.call('INCR', count_key) "
87 "if redis.call('TTL', count_key) < 0 then redis.call('EXPIRE', count_key, ARGV[3]) end "
88 "if count > limit then redis.call('SET', block_key, '1', 'EX', ARGV[4]) return tonumber(ARGV[4]) end "
89 "return 0 end "
90 "local user_block = bump(KEYS[1], KEYS[2], tonumber(ARGV[1])) "
91 "local source_block = 0 "
92 "if tonumber(ARGV[2]) > 0 and user_block == 0 then "
93 "source_block = bump(KEYS[3], KEYS[4], tonumber(ARGV[2])) end "
94 "return {user_block, source_block}"
95)
97_COUNTERS: Final = InMemoryCache(
98 max_size_in_memory=LOGIN_THROTTLE_MAX_TRACKED_COUNTERS, default_ttl=DEFAULT_FAILED_LOGIN_WINDOW_SECONDS
99)
100_BLOCKS: Final = InMemoryCache(
101 max_size_in_memory=LOGIN_THROTTLE_MAX_TRACKED_BLOCKS, default_ttl=DEFAULT_FAILED_LOGIN_BLOCK_SECONDS
102)
105@cache
106def _rate_limit_disabled() -> bool:
107 return get_secret_bool("LITELLM_DISABLE_LOGIN_RATE_LIMIT", default_value=False) is True
110@cache
111def warn_login_counters_are_per_worker(num_workers: str) -> None:
112 verbose_proxy_logger.warning(
113 "Running %s workers but Redis is not configured. Failed Admin UI sign-in attempts are counted "
114 "per worker, so the effective limits are %s times the configured values. Configure Redis "
115 "to share one count across workers.",
116 num_workers,
117 num_workers,
118 )
121@cache
122def warn_source_login_limit_is_off() -> None:
123 verbose_proxy_logger.warning(
124 "%s is not set or not a valid list of ranges, so failed Admin UI sign-in attempts are limited per "
125 "source address and username only. Set it to the address ranges of the proxies in front of LiteLLM, "
126 "or to an empty list when clients connect directly, to also limit each source address across usernames.",
127 TRUSTED_PROXY_RANGES_KEY,
128 )
131def declared_proxy_ranges(settings: Mapping[str, object]) -> tuple[str, ...] | None:
132 """What the operator says fronts LiteLLM: the proxy ranges, an empty tuple for none, None when unsaid.
134 Only a declared topology makes the source address trustworthy enough to limit across usernames.
135 An unset key, a value that is not a list of ranges, or a list with an entry that is not an address
136 or range leaves it unknown and the source scope off.
137 """
138 entries: Final = _configured_range_entries(settings.get(TRUSTED_PROXY_RANGES_KEY))
139 if entries is None or any(_parse_network(entry, TRUSTED_PROXY_RANGES_KEY) is None for entry in entries): 139 ↛ 141line 139 didn't jump to line 141 because the condition on line 139 was always true
140 return None
141 return entries
144def _configured_range_entries(raw_ranges: object) -> tuple[str, ...] | None:
145 """Every configured entry, blanks included, so a stray empty string fails validation like any other typo."""
146 if raw_ranges is None: 146 ↛ 148line 146 didn't jump to line 148 because the condition on line 146 was always true
147 return None
148 if isinstance(raw_ranges, str):
149 return tuple(part.strip() for part in raw_ranges.split(","))
150 try:
151 return tuple(str(entry).strip() for entry in _RANGE_ENTRIES.validate_python(raw_ranges))
152 except ValidationError:
153 verbose_proxy_logger.warning(
154 "Invalid %s value: expected a list of address ranges, got %s",
155 TRUSTED_PROXY_RANGES_KEY,
156 type(raw_ranges).__name__,
157 )
158 return None
161def _positive_int(raw: object, key: str, default: int) -> int:
162 if raw is None:
163 return default
164 try:
165 value: Final = int(str(raw))
166 except (TypeError, ValueError):
167 verbose_proxy_logger.warning("Invalid %s value %r; using %s", key, raw, default)
168 return default
169 if value < 1:
170 verbose_proxy_logger.warning("Invalid %s value %s (must be >= 1); using %s", key, value, default)
171 return default
172 return value
175def _int_setting(settings: Mapping[str, object], key: str, default: int) -> int:
176 return _positive_int(settings.get(key), key, default)
179def _override_limit(raw: object, default: int) -> int:
180 """A per-address override: a limit of 1 or more, or ``EXEMPT`` (0) to leave that address unlimited."""
181 if str(raw).strip() == str(EXEMPT):
182 return EXEMPT
183 return _positive_int(raw, SOURCE_LIMIT_OVERRIDES_KEY, default)
186def _parse_address(client_ip: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
187 """The address as it is limited and counted: an IPv4-mapped IPv6 address is its IPv4 address."""
188 try:
189 address: Final = ipaddress.ip_address(client_ip)
190 except ValueError:
191 return None
192 if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None:
193 return address.ipv4_mapped
194 return address
197def _parse_network(raw_range: str, setting_name: str = SOURCE_LIMIT_OVERRIDES_KEY) -> _Network | None:
198 try:
199 return ipaddress.ip_network(raw_range.strip(), strict=False)
200 except ValueError:
201 verbose_proxy_logger.warning("Invalid address or range %r in %s; skipping", raw_range, setting_name)
202 return None
205def _precedence(network: _Network, limit: int) -> tuple[int, bool, int]:
206 """Sort key for competing overrides: the longest prefix wins, then an exemption, then the higher limit."""
207 return (network.prefixlen, limit == EXEMPT, limit)
210def _source_limit(settings: Mapping[str, object], client_ip: str) -> int:
211 """Failure allowance for this address: the most specific configured range containing it, else the default.
213 ``EXEMPT`` (0) means the operator opted this address out of both limits. Between equivalent keys such as
214 ``1.2.3.4`` and ``1.2.3.4/32`` an exemption wins, then the higher limit.
215 """
216 default: Final = _int_setting(settings, SOURCE_LIMIT_KEY, DEFAULT_MAX_FAILED_LOGIN_ATTEMPTS_PER_SOURCE)
217 raw_overrides: Final = settings.get(SOURCE_LIMIT_OVERRIDES_KEY)
218 if raw_overrides is None:
219 return default
220 try:
221 overrides: Final = _SOURCE_LIMIT_OVERRIDES.validate_python(raw_overrides)
222 except ValidationError:
223 verbose_proxy_logger.warning(
224 "Invalid %s value; expected a mapping of address or range to limit", SOURCE_LIMIT_OVERRIDES_KEY
225 )
226 return default
227 address: Final = _parse_address(client_ip)
228 if address is None:
229 return default
230 matches: Final = sorted(
231 _precedence(network, _override_limit(raw_limit, default))
232 for raw_range, raw_limit in overrides.items()
233 if (network := _parse_network(raw_range)) is not None and address in network
234 )
235 return matches[-1][-1] if matches else default
238def user_limit_for(source_limit: int) -> int:
239 """Failures allowed for one username from one address: half the address allowance, rounded down, at least 1."""
240 return max(source_limit // 2, 1)
243def source_group(client_ip: str) -> str:
244 """The bucket an address is counted in: IPv4 as is, IPv6 by its /64, so one prefix holder cannot rotate."""
245 address: Final = _parse_address(client_ip)
246 if address is None:
247 return client_ip
248 if isinstance(address, ipaddress.IPv6Address):
249 return str(ipaddress.ip_network((address, IPV6_SOURCE_PREFIX_LENGTH), strict=False))
250 return str(address)
253class _Keys(NamedTuple):
254 pair_counter: str
255 pair_block: str
256 source_counter: str
257 source_block: str
260@dataclass(frozen=True, slots=True)
261class Block:
262 scope: Scope
263 retry_after: int
266@dataclass(frozen=True, slots=True)
267class LoginThrottle:
268 """Failed-login limits for one request's source address.
270 ``source_limit`` is None when the source scope is off: ``trusted_proxy_ranges`` is unset, so the peer
271 address may be a shared ingress. An empty list means clients connect directly and the peer is the source.
272 ``user_limit`` is derived from the address allowance either way, see ``user_limit_for``. An address whose
273 override is ``EXEMPT`` gets a disabled throttle: nothing is counted or blocked for it.
274 """
276 client_ip: str
277 source_limit: int | None
278 user_limit: int
279 window_seconds: int
280 block_seconds: int
281 counters: LocalStore
282 blocks: LocalStore
283 redis_cache: RedisCache | None = None
284 enabled: bool = True
286 @classmethod
287 def from_request(
288 cls,
289 request: Request,
290 general_settings: Mapping[str, object] | None,
291 redis_cache: RedisCache | None,
292 ) -> LoginThrottle:
293 settings: Final[Mapping[str, object]] = general_settings if general_settings is not None else EMPTY_MAPPING
294 proxies: Final = declared_proxy_ranges(settings)
295 resolved, _ = resolve_client_ip(
296 request, TrustedProxyConfig(use_forwarded_for=bool(proxies), trusted_proxy_cidrs=proxies or ())
297 )
298 source_limit: Final = _source_limit(settings, resolved or LOGIN_THROTTLE_UNKNOWN_SOURCE)
299 exempt: Final = source_limit == EXEMPT
300 return cls(
301 client_ip=resolved or LOGIN_THROTTLE_UNKNOWN_SOURCE,
302 source_limit=source_limit if proxies is not None and resolved is not None and not exempt else None,
303 user_limit=user_limit_for(source_limit),
304 window_seconds=_int_setting(settings, WINDOW_KEY, DEFAULT_FAILED_LOGIN_WINDOW_SECONDS),
305 block_seconds=_int_setting(settings, BLOCK_KEY, DEFAULT_FAILED_LOGIN_BLOCK_SECONDS),
306 counters=_COUNTERS,
307 blocks=_BLOCKS,
308 redis_cache=redis_cache,
309 enabled=not exempt and not _rate_limit_disabled(),
310 )
312 def _keys(self, username: str) -> _Keys:
313 group: Final = source_group(self.client_ip)
314 user: Final = hashlib.sha256(username.casefold().encode("utf-8")).hexdigest()
315 return _Keys(
316 pair_counter=f"{LOGIN_THROTTLE_CACHE_KEY_PREFIX}:{{{group}}}:user:{user}",
317 pair_block=f"{LOGIN_THROTTLE_CACHE_KEY_PREFIX}:{{{group}}}:block:user:{user}",
318 source_counter=f"{LOGIN_THROTTLE_CACHE_KEY_PREFIX}:{{{group}}}:source",
319 source_block=f"{LOGIN_THROTTLE_CACHE_KEY_PREFIX}:{{{group}}}:block:source",
320 )
322 async def attempt(self, username: str) -> LoginAttempt:
323 """Refuses a blocked key before any credential is looked at; otherwise hands back the attempt to settle."""
324 if not self.enabled:
325 return LoginAttempt(throttle=self, username=username)
326 block: Final = await self._active_block(self._keys(username))
327 if block is None:
328 return LoginAttempt(throttle=self, username=username)
329 verbose_proxy_logger.warning(
330 "Admin UI sign-in refused: the %s is blocked for %s more seconds; username=%r source=%s",
331 block.scope,
332 block.retry_after,
333 username,
334 self.client_ip,
335 )
336 raise self.refused(block.retry_after)
338 async def _active_block(self, keys: _Keys) -> Block | None:
339 local: Final = self._local_block_ttls(keys)
340 shared: Final = await self._shared_block_ttls(keys)
341 user_ttl: Final = max(local[0], shared[0])
342 source_ttl: Final = max(local[1], shared[1])
343 if self.source_limit is not None and source_ttl > 0:
344 return Block(scope="source", retry_after=source_ttl)
345 if user_ttl > 0:
346 return Block(scope="user", retry_after=user_ttl)
347 return None
349 async def _shared_block_ttls(self, keys: _Keys) -> _BlockTtls:
350 if self.redis_cache is None:
351 return LOGIN_THROTTLE_NOT_BLOCKED
352 try:
353 return _LUA_BLOCK_TTLS.validate_python(
354 await self.redis_cache.async_register_script(_BLOCK_TTLS_LUA)(keys, ())
355 )
356 except _REDIS_FAILURES as err:
357 self._warn_redis(err)
358 return LOGIN_THROTTLE_NOT_BLOCKED
360 def _local_block_ttls(self, keys: _Keys) -> _BlockTtls:
361 return self._local_block_ttl(keys.pair_block), self._local_block_ttl(keys.source_block)
363 def _local_block_ttl(self, block_key: str) -> int:
364 expires_at: Final = _LOCAL_BLOCK_EXPIRY.validate_python(self.blocks.get_cache(block_key))
365 if expires_at is None:
366 return 0
367 return max(math.ceil(expires_at - time.time()), 0)
369 async def record_failure(self, username: str) -> _BlockTtls:
370 keys: Final = self._keys(username)
371 source_limit: Final = self.source_limit or 0
372 if self.redis_cache is not None:
373 try:
374 return _LUA_BLOCK_TTLS.validate_python(
375 await self.redis_cache.async_register_script(_RECORD_FAILURE_LUA)(
376 keys, (self.user_limit, source_limit, self.window_seconds, self.block_seconds)
377 )
378 )
379 except _REDIS_FAILURES as err:
380 self._warn_redis(err)
381 user_block: Final = self._local_bump(keys.pair_counter, keys.pair_block, self.user_limit)
382 if source_limit == 0 or user_block > 0:
383 return user_block, 0
384 return user_block, self._local_bump(keys.source_counter, keys.source_block, source_limit)
386 def _local_bump(self, count_key: str, block_key: str, limit: int) -> int:
387 blocked: Final = self._local_block_ttl(block_key)
388 if blocked > 0:
389 return blocked
390 count: Final = int(self.counters.increment_cache(count_key, 1, ttl=self.window_seconds))
391 if count <= limit:
392 return 0
393 self.blocks.set_cache(block_key, time.time() + self.block_seconds, ttl=self.block_seconds)
394 return self.block_seconds
396 async def clear_pair(self, username: str) -> None:
397 pair_counter: Final = self._keys(username).pair_counter
398 if self.redis_cache is not None:
399 try:
400 await self.redis_cache.async_delete_cache(pair_counter)
401 except _REDIS_FAILURES as err:
402 self._warn_redis(err)
403 self.counters.delete_cache(pair_counter)
405 def _warn_redis(self, err: Exception) -> None:
406 verbose_proxy_logger.warning(
407 "Redis failed while counting Admin UI sign-in attempts; using this worker's own counters "
408 "until it recovers: %s",
409 err,
410 )
412 @staticmethod
413 def refused(retry_after: int) -> ProxyException:
414 return ProxyException(
415 message="Too many failed sign-in attempts. Try again later.",
416 type=ProxyErrorTypes.auth_error,
417 param="username",
418 code=status.HTTP_429_TOO_MANY_REQUESTS,
419 headers={"Retry-After": str(retry_after)}, # mutable-ok: ProxyException writes into its headers dict
420 )
423@dataclass(frozen=True, slots=True)
424class LoginAttempt:
425 throttle: LoginThrottle
426 username: str
428 async def succeeded(self) -> None:
429 if not self.throttle.enabled:
430 return
431 await self.throttle.clear_pair(self.username)
433 async def failed(self) -> None:
434 if not self.throttle.enabled:
435 return
436 user_block, source_block = await self.throttle.record_failure(self.username)
437 if user_block == 0 and source_block == 0:
438 return
439 verbose_proxy_logger.warning(
440 "Admin UI sign-in blocked for %s seconds after too many failures; scope=%s username=%r source=%s",
441 user_block or source_block,
442 "user" if user_block else "source",
443 self.username,
444 self.throttle.client_ip,
445 )