fix(proxy): count failed sign-ins in Redis alone while it answers

Every worker spends one shared budget and a successful sign-in clears it for all
of them. This worker's own counter is only consulted while Redis raises, so an
outage degrades to per-worker accounting instead of switching the control off

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/auth/login_throttle.py

@@ -11,7 +11,7 @@ startup and can be reassigned later.
 
 import asyncio
 import hashlib
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from functools import cache
 from types import MappingProxyType
@@ -60,6 +60,7 @@ def _bounded_store(max_entries: int) -> DualCache:
 _FAILED_LOGIN_USERNAME_CACHE: Final = _bounded_store(_MAX_TRACKED_LOGIN_USERNAMES)
 _FAILED_LOGIN_SOURCE_CACHE: Final = _bounded_store(_MAX_TRACKED_LOGIN_SOURCES)
 _NO_SETTINGS: Final = MappingProxyType({})
+_UNAVAILABLE: Final = object()
 
 _DELAYS_IN_FLIGHT: Final[dict[str, int]] = {}  # mutable-ok: per-source slots taken and released around each held delay
 
@@ -188,16 +189,22 @@ class LoginThrottle:
             return await work
         except Exception as exc:  # noqa: BLE001  # an unreachable cache must never deny a valid credential
             verbose_proxy_logger.warning("login attempt accounting unavailable: %s", exc)
-            return None
+            return _UNAVAILABLE
 
-    async def _failures(self, store: DualCache, key: str) -> int:
-        """The larger of the shared and the process-local count, so a Redis outage degrades
-        to per-worker accounting instead of switching the control off."""
-        local: Final = _as_count(await self._outcome(store.async_get_cache(key=key)))
+    async def _shared(self, work: Callable[[RedisCache], Awaitable[object]]) -> object:
+        """The Redis result, or ``_UNAVAILABLE`` when Redis is not configured or the call raised."""
         redis_cache: Final = self.redis_cache
         if redis_cache is None:
-            return local
-        return max(local, _as_count(await self._outcome(redis_cache.async_get_cache(key))))
+            return _UNAVAILABLE
+        return await self._outcome(work(redis_cache))
+
+    async def _failures(self, store: DualCache, key: str) -> int:
+        """The shared count while Redis answers, so every worker sees one budget; this worker's
+        own count only while it does not, so an outage degrades to per-worker accounting."""
+        shared: Final = await self._shared(lambda redis_cache: redis_cache.async_get_cache(key))
+        if shared is not _UNAVAILABLE:
+            return _as_count(shared)
+        return _as_count(await self._outcome(store.async_get_cache(key=key)))
 
     async def _remaining_window(self, key: str) -> int:
         """Seconds until this counter expires.
@@ -206,13 +213,12 @@ class LoginThrottle:
         was stripped out of band (PERSIST, a restore). It is given the full window again,
         since nothing increments a key once the limit is reached.
         """
-        redis_cache: Final = self.redis_cache
-        if redis_cache is None:
+        if self.redis_cache is None:
             return self.window_seconds
-        ttl: Final = await self._outcome(redis_cache.async_get_ttl(key))
+        ttl: Final = await self._shared(lambda redis_cache: redis_cache.async_get_ttl(key))
         if isinstance(ttl, int) and ttl > 0:
             return min(ttl, self.window_seconds)
-        await self._outcome(redis_cache.async_increment_with_floor(key, 0, self.window_seconds))
+        await self._shared(lambda redis_cache: redis_cache.async_increment_with_floor(key, 0, self.window_seconds))
         return self.window_seconds
 
     def _refused(self, retry_after: int, param: str) -> ProxyException:
@@ -260,16 +266,12 @@ class LoginThrottle:
             )
 
     async def _bump(self, store: DualCache, key: str) -> int:
-        local: Final = _as_count(
-            await self._outcome(store.async_increment_cache(key=key, value=1, ttl=self.window_seconds))
+        shared: Final = await self._shared(
+            lambda redis_cache: redis_cache.async_increment_with_floor(key, 1, self.window_seconds)
         )
-        redis_cache: Final = self.redis_cache
-        if redis_cache is None:
-            return local
-        shared: Final = _as_count(
-            await self._outcome(redis_cache.async_increment_with_floor(key, 1, self.window_seconds))
-        )
-        return max(local, shared)
+        if shared is not _UNAVAILABLE:
+            return _as_count(shared)
+        return _as_count(await self._outcome(store.async_increment_cache(key=key, value=1, ttl=self.window_seconds)))
 
     async def record_failure(self, username: str) -> FailureCounts:
         """Count one rejected credential guess against this username and against this source."""
@@ -331,7 +333,5 @@ class LoginThrottle:
         if not self.enabled:
             return
         key: Final = self._username_key(username)
-        redis_cache: Final = self.redis_cache
-        if redis_cache is not None:
-            await self._outcome(redis_cache.async_delete_cache(key))
+        await self._shared(lambda redis_cache: redis_cache.async_delete_cache(key))
         await self._outcome(self.username_cache.async_delete_cache(key=key))

tests/test_litellm/proxy/auth/test_login_utils.py

@@ -1203,6 +1203,83 @@ async def test_counters_are_written_with_their_expiry_and_re_armed_if_stripped(m
     assert set(redis.ttls) >= {k for k in redis.values if ":user:" in k}, "the refusal must re-arm a stripped expiry"
 
 
+class _DownRedis(_FakeRedis):
+    """Redis whose every call raises, as during an outage or an open circuit breaker."""
+
+    async def async_get_cache(self, key, **kwargs):
+        raise ConnectionError("redis is down")
+
+    async def async_increment_with_floor(self, key, value, ttl):
+        raise ConnectionError("redis is down")
+
+    async def async_get_ttl(self, key):
+        raise ConnectionError("redis is down")
+
+    async def async_delete_cache(self, key):
+        raise ConnectionError("redis is down")
+
+
+@pytest.mark.asyncio
+async def test_redis_is_the_only_counter_while_it_answers(monkeypatch):
+    """Regression: every worker must spend the same budget, and a success must clear it for all.
+
+    Counting in this worker's memory as well as in Redis let the two drift apart: a worker
+    whose Redis write failed kept its own count while the others gave the attacker fresh
+    guesses, and a stale local count outlived the shared clear after a correct password.
+    """
+    from litellm.caching.dual_cache import DualCache
+    from litellm.proxy._types import ProxyException
+    from litellm.proxy.auth.login_throttle import _CACHE_KEY_PREFIX
+
+    monkeypatch.setenv("UI_USERNAME", "admin")
+    monkeypatch.setenv("UI_PASSWORD", "right")
+    redis = _FakeRedis()
+    first_worker_store = DualCache()
+    second_worker_store = DualCache()
+    first_worker = _throttle(max_attempts=2, cache=first_worker_store, redis_cache=redis)
+    second_worker = _throttle(max_attempts=2, cache=second_worker_store, redis_cache=redis)
+
+    for _ in range(2):
+        with pytest.raises(ProxyException, match="Invalid credentials"):
+            await _guess(first_worker)
+
+    assert not [k for k in first_worker_store.in_memory_cache.cache_dict if str(k).startswith(_CACHE_KEY_PREFIX)], (
+        "with Redis answering, no worker may keep a counter of its own"
+    )
+    with pytest.raises(ProxyException) as blocked:
+        await _guess(second_worker)
+    assert blocked.value.code == "429", "the second worker must see the budget the first one spent"
+
+    monkeypatch.setenv("DATABASE_URL", "postgresql://stub")
+    with patch("litellm.proxy.auth.login_utils.user_update", new=AsyncMock()), patch(  # test-quality-ok: success mints a UI key and persists the user; faked so no DB is needed
+        "litellm.proxy.auth.login_utils.generate_key_helper_fn", new=AsyncMock(return_value={"token": "sk-ui"})
+    ):
+        await _guess(second_worker, password="right")
+
+    assert not [k for k in redis.values if ":user:" in k], "a success must clear the shared username counter"
+    with pytest.raises(ProxyException, match="Invalid credentials"):
+        await _guess(first_worker)
+
+
+@pytest.mark.asyncio
+async def test_a_redis_outage_falls_back_to_this_workers_own_counter(monkeypatch):
+    """With Redis raising, guesses are still counted and refused, per worker, instead of unbounded."""
+    from litellm.proxy._types import ProxyException
+
+    monkeypatch.setenv("UI_USERNAME", "admin")
+    monkeypatch.setenv("UI_PASSWORD", "right")
+    throttle = _throttle(max_attempts=2, redis_cache=_DownRedis())
+
+    for _ in range(2):
+        with pytest.raises(ProxyException, match="Invalid credentials"):
+            await _guess(throttle)
+
+    with pytest.raises(ProxyException) as blocked:
+        await _guess(throttle)
+    assert blocked.value.code == "429"
+    assert blocked.value.headers.get("Retry-After") == "900"
+
+
 @pytest.mark.asyncio
 async def test_counters_do_not_share_the_key_authentication_cache(monkeypatch):
     """Regression: throttle entries must not evict cached credentials.