Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/login_utils.py: 27%
161 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2Login utilities for handling user authentication in the proxy server.
4This module contains the core login logic that can be reused across different
5login endpoints (e.g., /login and /v2/login).
6"""
8import os
9import secrets
10from collections.abc import Mapping
11from datetime import datetime, timedelta, timezone
12from types import MappingProxyType
13from typing import TYPE_CHECKING, Final, Literal, cast
15import jwt
16from fastapi import HTTPException
18import litellm
19from litellm._logging import verbose_proxy_logger
20from litellm.constants import LITELLM_PROXY_ADMIN_NAME, LITELLM_UI_SESSION_DURATION
21from litellm.litellm_core_utils.duration_parser import duration_in_seconds
22from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
23from litellm.proxy._types import (
24 LiteLLM_UserTable,
25 LitellmUserRoles,
26 ProxyErrorTypes,
27 ProxyException,
28 UpdateUserRequest,
29 UserAPIKeyAuth,
30)
31from litellm.proxy.auth.auth_utils import is_sso_provider_fully_configured
32from litellm.proxy.auth.login_throttle import LoginAttempt, LoginThrottle
33from litellm.proxy.auth.password_policy import is_breach_check_enabled, is_password_breached
34from litellm.proxy.management_endpoints.internal_user_endpoints import user_update
35from litellm.proxy.management_endpoints.key_management_endpoints import (
36 generate_key_helper_fn,
37)
38from litellm.proxy.management_endpoints.ui_sso import (
39 get_disabled_non_admin_personal_key_creation,
40)
41from litellm.proxy.utils import (
42 PrismaClient,
43 get_server_root_path,
44 hash_password,
45 verify_password,
46)
47from litellm.repositories.user_repository import UserRepository
48from litellm.secret_managers.main import get_secret_bool
49from litellm.types.proxy.ui_sso import ReturnedUITokenObject
51INVALID_UI_CREDENTIALS_MESSAGE: Final = (
52 "Invalid credentials used to access UI. Check 'UI_USERNAME' and 'UI_PASSWORD', or the password set for your user"
53)
54INVALID_USER_PASSWORD_MESSAGE: Final = "Invalid credentials used to access UI. Check the password set for your user"
56if TYPE_CHECKING: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 from prisma import types as prisma_types
59BREACH_RECHECK_INTERVAL: Final = timedelta(hours=24)
60PASSWORD_RESET_ALLOWED_ROUTES: Final = ("/user/password/change", "/session/logout")
61PASSWORD_SESSION_METADATA: Final = MappingProxyType({"login_method": "username_password"})
64def _breach_recheck_due(last_breach_check_at: datetime | None) -> bool:
65 if last_breach_check_at is None:
66 return True
67 last_checked_utc: Final = (
68 last_breach_check_at
69 if last_breach_check_at.tzinfo is not None
70 else last_breach_check_at.replace(tzinfo=timezone.utc)
71 )
72 return datetime.now(timezone.utc) - last_checked_utc >= BREACH_RECHECK_INTERVAL
75async def screen_login_password_for_breach(
76 user_id: str,
77 password: str,
78 last_breach_check_at: datetime | None,
79 general_settings: Mapping[str, object],
80 prisma_client: PrismaClient,
81 client: AsyncHTTPHandler | None = None,
82) -> bool:
83 """Screens a successfully verified login password against HIBP, stamps
84 ``password_reset_required`` when breached, and returns whether a breach was
85 found so the login it runs in can restrict the session it is about to mint.
86 Fails open (HIBP or DB trouble never fails the login) and rechecks a given
87 user at most once per ``BREACH_RECHECK_INTERVAL``."""
88 if not is_breach_check_enabled(general_settings):
89 return False
90 if not _breach_recheck_due(last_breach_check_at):
91 return False
92 breached: Final = await is_password_breached(password, general_settings, client)
93 checked_at: Final = datetime.now(timezone.utc)
94 breached_update: Final[prisma_types.LiteLLM_UserTableUpdateInput] = {
95 "last_breach_check_at": checked_at,
96 "password_reset_required": True,
97 }
98 recheck_update: Final[prisma_types.LiteLLM_UserTableUpdateInput] = {"last_breach_check_at": checked_at}
99 update_data: Final = breached_update if breached else recheck_update
100 find_user: Final[prisma_types.LiteLLM_UserTableWhereInput] = {"user_id": user_id}
101 try:
102 await UserRepository(prisma_client).table.update(where=find_user, data=update_data)
103 except Exception as e: # noqa: BLE001 # a failed stamp must never surface into the login
104 verbose_proxy_logger.warning("Login-time breach screening could not update user %s: %s", user_id, e)
105 return breached
108async def _rehash_password_if_needed(user_id: str, password: str, stored: str) -> None:
109 """Rehash legacy password (SHA256) to scrypt on successful login."""
110 if stored.startswith("scrypt:"):
111 return
112 from litellm.proxy.proxy_server import prisma_client
114 if prisma_client is not None:
115 await UserRepository(prisma_client).table.update(
116 where={"user_id": user_id},
117 data={"password": hash_password(password)},
118 )
121def get_ui_credentials(master_key: str | None) -> tuple[str, str]:
122 """
123 Get UI username and password from environment variables or master key.
125 Args:
126 master_key: Master key for the proxy (used as fallback for password)
128 Returns:
129 tuple[str, str]: A tuple containing (ui_username, ui_password)
131 Raises:
132 ProxyException: If neither UI_PASSWORD nor master_key is available
133 """
134 ui_username: Final = os.getenv("UI_USERNAME", "admin")
135 ui_password = os.getenv("UI_PASSWORD", None)
136 if ui_password is None:
137 ui_password = str(master_key) if master_key is not None else None
138 if ui_password is None:
139 raise ProxyException(
140 message="set Proxy master key to use UI. https://docs.litellm.ai/docs/proxy/virtual_keys. If set, use `--detailed_debug` to debug issue.",
141 type=ProxyErrorTypes.auth_error,
142 param="UI_PASSWORD",
143 code=500,
144 )
145 return ui_username, ui_password
148def _matches_env_credentials(username: str, password: str, master_key: str | None) -> bool:
149 ui_username, ui_password = get_ui_credentials(master_key)
150 return secrets.compare_digest(username.encode("utf-8"), ui_username.encode("utf-8")) and secrets.compare_digest(
151 password.encode("utf-8"), ui_password.encode("utf-8")
152 )
155def _admin_credentials_match(
156 username: str, password: str, master_key: str, general_settings: Mapping[str, object]
157) -> bool:
158 return general_settings.get("disable_env_credential_login") is not True and _matches_env_credentials(
159 username, password, master_key
160 )
163def _invalid_credentials_message(general_settings: Mapping[str, object]) -> str:
164 """One rejection message for unknown usernames and wrong passwords alike, so neither can be enumerated."""
165 if is_env_credential_login_enabled(general_settings):
166 return INVALID_UI_CREDENTIALS_MESSAGE
167 return INVALID_USER_PASSWORD_MESSAGE
170def is_env_credential_login_enabled(general_settings: Mapping[str, object]) -> bool:
171 """Whether a login with UI_USERNAME/UI_PASSWORD (or the master-key fallback) can succeed.
173 Two settings can turn it off: `disable_env_credential_login` unconditionally, and
174 `disable_password_login_when_sso_enabled` as a side effect, since its gate rejects
175 every username/password login before the env comparison runs. Feeds both the
176 `authenticate_user` gate and the Admin UI warning banner, so the banner never nags
177 about a login path that is already unreachable.
178 """
179 if general_settings.get("disable_env_credential_login") is True: 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true
180 return False
181 if general_settings.get("disable_password_login_when_sso_enabled") is True and is_sso_provider_fully_configured(): 181 ↛ 182line 181 didn't jump to line 182 because the condition on line 181 was never true
182 return False
183 return True
186class LoginResult:
187 """Result object containing authentication data from login."""
189 user_id: str
190 key: str
191 user_email: str | None
192 user_role: str
193 login_method: Literal["sso", "username_password"]
194 password_reset_required: bool
196 def __init__(
197 self,
198 user_id: str,
199 key: str,
200 user_email: str | None,
201 user_role: str,
202 login_method: Literal["sso", "username_password"] = "username_password",
203 password_reset_required: bool = False,
204 ):
205 self.user_id = user_id
206 self.key = key
207 self.user_email = user_email
208 self.user_role = user_role
209 self.login_method = login_method
210 self.password_reset_required = password_reset_required
213async def authenticate_user(
214 username: str,
215 password: str,
216 master_key: str | None,
217 prisma_client: PrismaClient | None,
218 throttle: LoginThrottle,
219 general_settings: Mapping[str, object] = MappingProxyType({}),
220) -> LoginResult:
221 """
222 Authenticate a user and generate an API key for UI access.
224 This function handles two login scenarios:
225 1. Admin login using UI_USERNAME and UI_PASSWORD
226 2. User login using email and password from database
228 Args:
229 username: Username or email from the login form
230 password: Password from the login form
231 master_key: Master key for the proxy (required)
232 prisma_client: Prisma database client (optional)
233 throttle: Failed sign-in accounting for this request's source address
234 general_settings: Proxy general_settings, checked for
235 `disable_password_login_when_sso_enabled` and
236 `disable_env_credential_login`
238 Returns:
239 LoginResult: Object containing authentication data
241 Raises:
242 ProxyException: If authentication fails or required configuration is missing,
243 or if username/password login is disabled while SSO is configured
245 Recovery: an admin locked out of the UI by
246 `disable_password_login_when_sso_enabled`, or by the failed sign-in block in
247 `throttle`, can still administer the proxy over the API with the master key
248 (Authorization: Bearer <master_key>), which never goes through this function.
249 No credential, the env admin credentials and the master key included, is
250 exempt from the block. To restore UI username/password login, unset the
251 setting in config.yaml (or the DB-persisted general_settings) and restart the
252 proxy; this is a deliberate, auditable config change rather than a hidden
253 bypass.
255 The gate below requires the SSO provider to be FULLY configured (every
256 companion secret/endpoint an actual sign-in needs), not merely that a
257 client id is present, so an incomplete SSO setup can never disable the
258 only working login path.
259 """
260 if general_settings.get("disable_password_login_when_sso_enabled") is True and is_sso_provider_fully_configured():
261 raise ProxyException(
262 message=(
263 "Username/password login is disabled because SSO is configured "
264 "and 'disable_password_login_when_sso_enabled' is set. Sign in via SSO."
265 ),
266 type=ProxyErrorTypes.auth_error,
267 param="disable_password_login_when_sso_enabled",
268 code=403,
269 )
271 if master_key is None:
272 raise ProxyException(
273 message="Master Key not set for Proxy. Please set Master Key to use Admin UI. Set `LITELLM_MASTER_KEY` in .env or set general_settings:master_key in config.yaml. https://docs.litellm.ai/docs/proxy/virtual_keys. If set, use `--detailed_debug` to debug issue.",
274 type=ProxyErrorTypes.auth_error,
275 param="master_key",
276 code=500,
277 )
279 attempt: Final = await throttle.attempt(username)
280 return await _sign_in(username, password, master_key, prisma_client, attempt, general_settings)
283async def _sign_in(
284 username: str,
285 password: str,
286 master_key: str,
287 prisma_client: PrismaClient | None,
288 attempt: LoginAttempt,
289 general_settings: Mapping[str, object],
290) -> LoginResult:
291 admin_credentials_match: Final = _admin_credentials_match(username, password, master_key, general_settings)
292 # Check if we can find the `username` in the db. On the UI, users can enter username=their email
293 _user_row: LiteLLM_UserTable | None = None
294 user_role: (
295 Literal[
296 LitellmUserRoles.PROXY_ADMIN,
297 LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
298 LitellmUserRoles.INTERNAL_USER,
299 LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
300 ]
301 | None
302 ) = None
304 if prisma_client is not None:
305 _user_row = cast(
306 LiteLLM_UserTable | None,
307 await UserRepository(prisma_client).table.find_first(
308 where={"user_email": {"equals": username, "mode": "insensitive"}}
309 ),
310 )
312 """
313 To login to Admin UI, we support the following
314 - Login with UI_USERNAME and UI_PASSWORD
315 - Login with Invite Link `user_email` and `password` combination
316 """
317 if admin_credentials_match:
318 # Non SSO -> If user is using UI_USERNAME and UI_PASSWORD they are Proxy admin
319 user_role = LitellmUserRoles.PROXY_ADMIN
320 user_id = LITELLM_PROXY_ADMIN_NAME
322 # we want the key created to have PROXY_ADMIN_PERMISSIONS
323 key_user_id: Final = os.getenv("PROXY_ADMIN_ID", LITELLM_PROXY_ADMIN_NAME)
325 # Admin is Authe'd in - generate key for the UI to access Proxy
327 # ensure this user is set as the proxy admin, in this route there is no sso, we can assume this user is only the admin
328 await user_update(
329 data=UpdateUserRequest(
330 user_id=key_user_id,
331 user_role=user_role,
332 ),
333 user_api_key_dict=UserAPIKeyAuth(
334 user_role=LitellmUserRoles.PROXY_ADMIN,
335 ),
336 )
338 if os.getenv("DATABASE_URL") is not None:
339 response = await generate_key_helper_fn(
340 llm_router=None,
341 request_type="key",
342 **{
343 "user_role": LitellmUserRoles.PROXY_ADMIN,
344 "duration": LITELLM_UI_SESSION_DURATION,
345 "key_max_budget": litellm.max_ui_session_budget,
346 "models": [],
347 "aliases": {},
348 "config": {},
349 "spend": 0,
350 "user_id": key_user_id,
351 "team_id": "litellm-dashboard",
352 },
353 )
354 else:
355 raise ProxyException(
356 message="No Database connected. Set DATABASE_URL in .env. If set, use `--detailed_debug` to debug issue.",
357 type=ProxyErrorTypes.auth_error,
358 param="DATABASE_URL",
359 code=500,
360 )
362 key = response["token"]
364 if get_secret_bool("EXPERIMENTAL_UI_LOGIN"):
365 from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken
367 user_info: LiteLLM_UserTable | None = None
368 if _user_row is not None:
369 user_info = _user_row
370 elif user_id is not None: # if user_id is not None, we are using the UI_USERNAME and UI_PASSWORD
371 user_info = LiteLLM_UserTable(
372 user_id=user_id,
373 user_role=user_role,
374 models=[],
375 max_budget=litellm.max_ui_session_budget,
376 )
377 if user_info is None:
378 raise HTTPException(
379 status_code=401,
380 detail={"error": "User Information is required for experimental UI login"},
381 )
383 key = ExperimentalUIJWTToken.get_experimental_ui_login_jwt_auth_token(user_info)
385 await attempt.succeeded()
387 return LoginResult(
388 user_id=user_id,
389 key=key,
390 user_email=None,
391 user_role=user_role,
392 login_method="username_password",
393 )
395 elif _user_row is not None:
396 """
397 When sharing invite links
399 -> if the user has no role in the DB assume they are only a viewer
400 """
401 user_id = getattr(_user_row, "user_id", "unknown")
402 user_role = getattr(_user_row, "user_role", LitellmUserRoles.INTERNAL_USER_VIEW_ONLY)
403 user_email: Final = getattr(_user_row, "user_email", "unknown")
404 _password: Final = getattr(_user_row, "password", "unknown")
406 if _password is None:
407 raise ProxyException(
408 message="User has no password set. Please set a password for the user via `/user/update`.",
409 type=ProxyErrorTypes.auth_error,
410 param="password",
411 code=401,
412 )
414 if verify_password(password, _password):
415 await _rehash_password_if_needed(_user_row.user_id, password, _password)
416 breached_now: Final = prisma_client is not None and await screen_login_password_for_breach(
417 user_id=_user_row.user_id,
418 password=password,
419 last_breach_check_at=getattr(_user_row, "last_breach_check_at", None),
420 general_settings=general_settings,
421 prisma_client=prisma_client,
422 )
423 password_reset_required: Final = breached_now or getattr(_user_row, "password_reset_required", None) is True
424 if os.getenv("DATABASE_URL") is not None:
425 response = await generate_key_helper_fn(
426 llm_router=None,
427 request_type="key",
428 user_role=user_role,
429 duration=LITELLM_UI_SESSION_DURATION,
430 key_max_budget=litellm.max_ui_session_budget,
431 spend=0,
432 user_id=user_id,
433 team_id="litellm-dashboard",
434 allowed_routes=list(PASSWORD_RESET_ALLOWED_ROUTES) if password_reset_required else None,
435 metadata={
436 **PASSWORD_SESSION_METADATA,
437 **({"password_reset_required": True} if password_reset_required else {}),
438 },
439 )
440 else:
441 raise ProxyException(
442 message="No Database connected. Set DATABASE_URL in .env. If set, use `--detailed_debug` to debug issue.",
443 type=ProxyErrorTypes.auth_error,
444 param="DATABASE_URL",
445 code=500,
446 )
448 key = response["token"]
450 await attempt.succeeded()
452 return LoginResult(
453 user_id=user_id,
454 key=key,
455 user_email=user_email,
456 user_role=cast(str, user_role),
457 login_method="username_password",
458 password_reset_required=password_reset_required,
459 )
460 else:
461 await attempt.failed()
462 raise ProxyException(
463 message=_invalid_credentials_message(general_settings),
464 type=ProxyErrorTypes.auth_error,
465 param="invalid_credentials",
466 code=401,
467 )
468 else:
469 await attempt.failed()
470 raise ProxyException(
471 message=_invalid_credentials_message(general_settings),
472 type=ProxyErrorTypes.auth_error,
473 param="invalid_credentials",
474 code=401,
475 )
478def _ui_session_exp_timestamp() -> int:
479 """The ``exp`` claim (unix seconds) for a UI session cookie, ``LITELLM_UI_SESSION_DURATION``
480 from now. The virtual key sealed inside the cookie already expires after this same
481 duration; stamping the JWT itself gives the cookie the bounded lifetime the dashboard's
482 client-side expiry check and the server-side session-cookie readers both assume, instead
483 of a token that stays signature-valid until the master key rotates."""
484 ttl_seconds: Final = duration_in_seconds(LITELLM_UI_SESSION_DURATION)
485 return int((datetime.now(timezone.utc) + timedelta(seconds=ttl_seconds)).timestamp())
488def encode_ui_session_jwt(returned_ui_token_object: ReturnedUITokenObject, master_key: str) -> str:
489 """Encode a UI session cookie JWT with a bounded ``exp``.
491 The single choke point every UI login path (SSO and username/password /login, /v2,
492 /v3) uses to mint the ``token`` cookie, so the cookie's lifetime is set in exactly one
493 place and cannot drift between paths. Without the ``exp`` the cookie is valid until the
494 master key rotates, and the session-cookie readers that require a bounded lifetime
495 (the MCP interactive sign-in) reject it.
496 """
497 claims: Final = {**cast(dict, returned_ui_token_object), "exp": _ui_session_exp_timestamp()}
498 return jwt.encode(claims, master_key, algorithm="HS256")
501def create_ui_token_object(
502 login_result: LoginResult,
503 general_settings: dict,
504 premium_user: bool,
505) -> ReturnedUITokenObject:
506 """
507 Create a ReturnedUITokenObject from a LoginResult.
509 Args:
510 login_result: The result from authenticate_user
511 general_settings: General proxy settings dictionary
512 premium_user: Whether premium features are enabled
514 Returns:
515 ReturnedUITokenObject: Token object ready for JWT encoding
516 """
517 disabled_non_admin_personal_key_creation: Final = get_disabled_non_admin_personal_key_creation()
519 return ReturnedUITokenObject(
520 user_id=login_result.user_id,
521 key=login_result.key,
522 user_email=login_result.user_email,
523 user_role=login_result.user_role,
524 login_method=login_result.login_method,
525 premium_user=premium_user,
526 auth_header_name=general_settings.get("litellm_key_header_name", "Authorization"),
527 disabled_non_admin_personal_key_creation=disabled_non_admin_personal_key_creation,
528 server_root_path=get_server_root_path(),
529 password_reset_required=login_result.password_reset_required,
530 )