fix(proxy): take FIPS logins one at a time and recheck the throttle first

A login that waited for the KDF thread asked the throttle again only after
deriving, so a burst for one user kept that thread busy long after the user
was blocked. Logins now wait for their turn on a lock and ask the throttle
when it comes, so a blocked login costs no PBKDF2 and only one login at a time
holds a place in the KDF queue. Password changes and invite claims no longer
wait behind a whole burst of logins.

litellm/proxy/auth/login_utils.py

@@ -5,6 +5,7 @@ This module contains the core login logic that can be reused across different
 login endpoints (e.g., /login and /v2/login).
 """
 
+import asyncio
 import os
 import secrets
 from collections.abc import Mapping
@@ -107,6 +108,18 @@ async def screen_login_password_for_breach(
     return breached
 
 
+_ONE_FIPS_LOGIN_AT_A_TIME: Final = asyncio.Lock()
+
+
+async def _password_matches(password: str, stored: str, attempt: LoginAttempt) -> bool:
+    """Checks inline, or in FIPS mode one login at a time, refusing a login whose user got blocked while it waited."""
+    if not openssl_enforces_fips():
+        return verify_password(password, stored)
+    async with _ONE_FIPS_LOGIN_AT_A_TIME:
+        await attempt.refuse_if_blocked_while_waiting()
+        return await run_off_loop_if_fips(verify_password, password, stored)
+
+
 async def _rehash_password_if_needed(user_id: str, password: str, stored: str) -> None:
     """Rehash legacy password (SHA256) to scrypt on successful login."""
     if stored.startswith(("scrypt:", "pbkdf2:")):
@@ -413,10 +426,7 @@ async def _sign_in(
                 code=401,
             )
 
-        password_matches: Final = await run_off_loop_if_fips(verify_password, password, _password)
-        if openssl_enforces_fips():
-            await attempt.refuse_if_blocked_while_waiting()
-        if password_matches:
+        if await _password_matches(password, _password, attempt):
             await _rehash_password_if_needed(_user_row.user_id, password, _password)
             breached_now: Final = prisma_client is not None and await screen_login_password_for_breach(
                 user_id=_user_row.user_id,

tests/unit/proxy/auth/test_login_utils.py

@@ -5,6 +5,7 @@ This module tests the refactored login logic that was moved from proxy_server.py
 to login_utils.py for better reusability.
 """
 
+import asyncio
 import hashlib
 import os
 from collections.abc import Mapping
@@ -2471,7 +2472,7 @@ async def _login_whose_user_gets_blocked_during_the_row_lookup(monkeypatch: pyte
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("fips_openssl")
 async def test_fips_login_hides_the_right_password_when_its_user_got_blocked_while_it_waited(
-    monkeypatch: pytest.MonkeyPatch,
+    monkeypatch: pytest.MonkeyPatch, pbkdf2_on_event_loop: list[bool]
 ) -> None:
     from litellm.proxy._types import ProxyException
 
@@ -2479,6 +2480,40 @@ async def test_fips_login_hides_the_right_password_when_its_user_got_blocked_whi
         await _login_whose_user_gets_blocked_during_the_row_lookup(monkeypatch, _PBKDF2_ROW)
 
     assert blocked.value.code == "429"
+    assert pbkdf2_on_event_loop == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("fips_openssl")
+async def test_fips_burst_of_wrong_passwords_stops_deriving_once_its_user_is_blocked(
+    monkeypatch: pytest.MonkeyPatch, pbkdf2_on_event_loop: list[bool]
+) -> None:
+    from litellm.proxy.auth.login_utils import authenticate_user
+
+    monkeypatch.setenv("DATABASE_URL", "postgresql://stub")
+    repository = _known_user()
+    repository.return_value.table.find_first.return_value.password = _PBKDF2_ROW
+    throttle = _throttle(user_limit=1)
+    with patch(  # test-quality-ok: the user lookup is the database boundary
+        "litellm.proxy.auth.login_utils.UserRepository", repository
+    ):
+        refusals = await asyncio.gather(
+            *(
+                authenticate_user(
+                    username="known@example.com",
+                    password="Wrong-Lamp-4-River",
+                    master_key="sk-master",
+                    prisma_client=MagicMock(),
+                    throttle=throttle,
+                    general_settings=_POLICY_NO_BREACH_CHECK,
+                )
+                for _ in range(6)
+            ),
+            return_exceptions=True,
+        )
+
+    assert [refusal.code for refusal in refusals] == ["401", "401", "429", "429", "429", "429"]
+    assert pbkdf2_on_event_loop == [False, False]
 
 
 @pytest.mark.asyncio