diff --git a/litellm/proxy/management_endpoints/ui_sso.py b/litellm/proxy/management_endpoints/ui_sso.py index a119bad17db..199de54ff09 100644 --- a/litellm/proxy/management_endpoints/ui_sso.py +++ b/litellm/proxy/management_endpoints/ui_sso.py @@ -7,3 +7,4600 @@ Has all /sso/* routes /sso/debug/login - handles user signing in with SSO and redirects to /sso/debug/callback /sso/debug/callback - returns the OpenID object returned by the SSO provider """ + +import asyncio +import base64 +import hashlib +import inspect +import os +import re +import secrets +from copy import deepcopy +from html import escape +from typing import ( + TYPE_CHECKING, + Any, + Dict, + List, + Literal, + NoReturn, + Optional, + Tuple, + Union, + cast, +) +from urllib.parse import parse_qs, urlencode, urlparse + +if TYPE_CHECKING: + import httpx + +import jwt +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status +from fastapi.responses import RedirectResponse + +import litellm +from litellm._logging import verbose_proxy_logger +from litellm._uuid import uuid +from litellm.caching.dual_cache import DualCache +from litellm.constants import ( + CLI_SSO_CLAIM_MAP, + CLI_SSO_CLAIM_MAX_SCALAR_LENGTH, + CLI_SSO_SESSION_CACHE_KEY_PREFIX, + CLI_SSO_SESSION_TTL_SECONDS, + LITELLM_CLI_SOURCE_IDENTIFIER, + LITELLM_UI_SESSION_DURATION, + MAX_SPENDLOG_ROWS_TO_QUERY, + MICROSOFT_USER_DISPLAY_NAME_ATTRIBUTE, + MICROSOFT_USER_EMAIL_ATTRIBUTE, + MICROSOFT_USER_FIRST_NAME_ATTRIBUTE, + MICROSOFT_USER_ID_ATTRIBUTE, + MICROSOFT_USER_LAST_NAME_ATTRIBUTE, +) +from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value +from litellm.llms.custom_httpx.http_handler import ( + AsyncHTTPHandler, + get_async_httpx_client, + httpxSpecialProvider, +) +from litellm.proxy._types import ( + CommonProxyErrors, + LiteLLM_UserTable, + LitellmUserRoles, + Member, + NewTeamRequest, + NewUserRequest, + NewUserResponse, + ProxyErrorTypes, + ProxyException, + SSOUserDefinedValues, + TeamMemberAddRequest, + UserAPIKeyAuth, +) +from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken, get_user_object +from litellm.proxy.auth.auth_utils import ( + _get_request_ip_address, + _has_user_setup_sso, +) +from litellm.proxy.auth.handle_jwt import JWTHandler +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.proxy.common_utils.admin_ui_utils import ( + admin_ui_disabled, + show_missing_vars_in_env, +) +from litellm.proxy.common_utils.html_forms.jwt_display_template import ( + jwt_display_template, +) +from litellm.proxy.common_utils.html_forms.ui_login import build_ui_login_form +from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache +from litellm.proxy.management_endpoints.internal_user_endpoints import new_user +from litellm.proxy.management_endpoints.sso import CustomMicrosoftSSO +from litellm.proxy.management_endpoints.sso_helper_utils import ( + check_is_admin_only_access, + has_admin_ui_access, +) +from litellm.proxy.management_endpoints.team_endpoints import new_team, team_member_add +from litellm.proxy.management_endpoints.types import ( + CustomOpenID, + get_litellm_user_role, + is_valid_litellm_user_role, +) +from litellm.proxy.utils import ( + PrismaClient, + ProxyLogging, + get_custom_url, + get_server_root_path, +) +from litellm.repositories.table_repositories import SSOConfigRepository +from litellm.repositories.team_repository import TeamRepository +from litellm.repositories.user_repository import UserRepository +from litellm.secret_managers.main import get_secret_bool, str_to_bool +from litellm.types.proxy.management_endpoints.ui_sso import * # noqa: F403 +from litellm.types.proxy.management_endpoints.ui_sso import ( + DefaultTeamSSOParams, + MicrosoftGraphAPIUserGroupDirectoryObject, + MicrosoftGraphAPIUserGroupResponse, + MicrosoftServicePrincipalTeam, + RoleMappings, + TeamMappings, +) +from litellm.types.proxy.ui_sso import ParsedOpenIDResult + +if TYPE_CHECKING: + from fastapi_sso.sso.base import OpenID +else: + from typing import Any as OpenID + +router = APIRouter() + +# OAuth bearer credential fields that must not appear in SSO debug responses +# (received_response is included in restricted-group error messages). +# Metadata fields (token_type, expires_in, scope) are intentionally kept so +# response convertors see the same fields in the PKCE path as in the non-PKCE path. +_OAUTH_TOKEN_FIELDS = frozenset({"access_token", "id_token", "refresh_token"}) +_CLI_SSO_FLOW_CACHE_KEY_PREFIX = f"{CLI_SSO_SESSION_CACHE_KEY_PREFIX}:flow" +_CLI_SSO_START_RATE_LIMIT_CACHE_KEY_PREFIX = ( + f"{_CLI_SSO_FLOW_CACHE_KEY_PREFIX}:start_rate_limit" +) +_CLI_SSO_START_RATE_LIMIT_WINDOW_SECONDS = 60 +_CLI_SSO_START_RATE_LIMIT_MAX_ATTEMPTS = 30 +_CLI_SSO_USER_CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" +_CLI_SSO_LOGIN_ID_RE = re.compile(r"^cli-[A-Za-z0-9_-]{12,124}$") +_CLI_SSO_USER_CODE_RE = re.compile( + rf"^[{_CLI_SSO_USER_CODE_ALPHABET}]{{4}}-[{_CLI_SSO_USER_CODE_ALPHABET}]{{4}}$" +) +_CLI_SSO_SCALAR_TYPES = (str, int, float, bool) +_CLI_SSO_DEST_KEY_RE = re.compile(r"^[A-Za-z0-9_.-]+$") +_CLI_SSO_SECRET_KEY_FRAGMENTS = frozenset( + { + "access_token", + "api_key", + "client_secret", + "id_token", + "password", + "private_key", + "refresh_token", + "secret", + } +) + + +def _hash_cli_sso_secret(secret: str) -> str: + return hashlib.sha256(secret.encode("utf-8")).hexdigest() + + +def _normalize_cli_sso_user_code(user_code: str) -> str: + return "".join(ch for ch in user_code.upper() if ch.isalnum()) + + +def _generate_cli_sso_user_code() -> str: + user_code = "".join(secrets.choice(_CLI_SSO_USER_CODE_ALPHABET) for _ in range(8)) + return f"{user_code[:4]}-{user_code[4:]}" + + +def _get_cli_sso_flow_cache_key(login_id: str) -> str: + return f"{_CLI_SSO_FLOW_CACHE_KEY_PREFIX}:{login_id}" + + +def _is_valid_cli_sso_login_id(login_id: Optional[str]) -> bool: + return isinstance(login_id, str) and bool(_CLI_SSO_LOGIN_ID_RE.fullmatch(login_id)) + + +def _is_valid_cli_sso_user_code(user_code: str | None) -> bool: + return isinstance(user_code, str) and bool( + _CLI_SSO_USER_CODE_RE.fullmatch(user_code) + ) + + +def _cli_sso_verification_uri_complete_enabled() -> bool: + from litellm.proxy.proxy_server import general_settings + + return bool(general_settings.get("allow_cli_sso_verification_uri_complete", False)) + + +def _cli_sso_start_response_body( + *, + login_id: str, + poll_secret: str, + user_code: str, + verification_uri_complete: str | None, +) -> dict[str, str | int]: + if verification_uri_complete is None: + return { + "login_id": login_id, + "poll_secret": poll_secret, + "user_code": user_code, + "expires_in": CLI_SSO_SESSION_TTL_SECONDS, + } + return { + "login_id": login_id, + "poll_secret": poll_secret, + "user_code": user_code, + "verification_uri_complete": verification_uri_complete, + "expires_in": CLI_SSO_SESSION_TTL_SECONDS, + } + + +def _get_cli_sso_start_rate_limit_cache_key( + request: Request, use_x_forwarded_for: Optional[bool] = False +) -> str: + client_ip = ( + _get_request_ip_address( + request=request, use_x_forwarded_for=use_x_forwarded_for + ) + or "unknown" + ) + client_ip_hash = _hash_cli_sso_secret(client_ip) + return f"{_CLI_SSO_START_RATE_LIMIT_CACHE_KEY_PREFIX}:{client_ip_hash}" + + +def _check_cli_sso_start_rate_limit( + request: Request, + cache: DualCache, + use_x_forwarded_for: Optional[bool] = False, +) -> None: + rate_limit_cache_key = _get_cli_sso_start_rate_limit_cache_key( + request=request, use_x_forwarded_for=use_x_forwarded_for + ) + current_attempts = cache.increment_cache( + key=rate_limit_cache_key, + value=1, + ttl=_CLI_SSO_START_RATE_LIMIT_WINDOW_SECONDS, + ) + if current_attempts > _CLI_SSO_START_RATE_LIMIT_MAX_ATTEMPTS: + raise HTTPException( + status_code=429, + detail="Too many CLI login attempts. Try again later.", + ) + + +def _get_cli_sso_flow_or_raise(login_id: Optional[str], cache: DualCache) -> dict: + if not _is_valid_cli_sso_login_id(login_id): + raise HTTPException(status_code=400, detail="Invalid CLI login session") + + cache_key = _get_cli_sso_flow_cache_key(cast(str, login_id)) + flow = cache.get_cache(key=cache_key) + if not isinstance(flow, dict) or "poll_secret_hash" not in flow: + raise HTTPException(status_code=400, detail="Invalid CLI login session") + return flow + + +def _set_cli_sso_flow(login_id: str, cache: DualCache, flow: dict) -> None: + cache.set_cache( + key=_get_cli_sso_flow_cache_key(login_id), + value=flow, + ttl=CLI_SSO_SESSION_TTL_SECONDS, + ) + + +def _verify_cli_sso_poll_secret(flow: dict, poll_secret: Optional[str]) -> bool: + expected_poll_secret_hash = flow.get("poll_secret_hash") + if not isinstance(expected_poll_secret_hash, str) or not isinstance( + poll_secret, str + ): + return False + supplied_poll_secret_hash = _hash_cli_sso_secret(poll_secret) + return secrets.compare_digest(supplied_poll_secret_hash, expected_poll_secret_hash) + + +def _parse_cli_sso_claim_map() -> List[Tuple[str, str]]: + """ + Parse CLI_SSO_CLAIM_MAP / LITELLM_CLI_SSO_CLAIM_MAP. + + Format: comma-separated ``source_claim->metadata_key`` pairs, e.g. + ``employment_type->acme_employment_type,org_info.department->department``. + Destination keys may use an optional ``metadata.`` prefix; values are stored + on the LiteLLM user's ``metadata`` JSON column. + """ + claim_map_raw = CLI_SSO_CLAIM_MAP.strip() + if not claim_map_raw: + return [] + + parsed: List[Tuple[str, str]] = [] + for entry in claim_map_raw.split(","): + entry = entry.strip() + if not entry or "->" not in entry: + continue + source_claim, dest_key = entry.split("->", 1) + source_claim = source_claim.strip() + dest_key = dest_key.strip() + if dest_key.startswith("metadata."): + dest_key = dest_key[len("metadata.") :] + if source_claim and dest_key: + parsed.append((source_claim, dest_key)) + return parsed + + +def _is_safe_cli_sso_metadata_dest_key(dest_key: str) -> bool: + if not dest_key or not _CLI_SSO_DEST_KEY_RE.fullmatch(dest_key): + return False + lowered = dest_key.lower() + return not any(fragment in lowered for fragment in _CLI_SSO_SECRET_KEY_FRAGMENTS) + + +def _is_safe_cli_sso_scalar_claim_value(value: Any) -> bool: + if not isinstance(value, _CLI_SSO_SCALAR_TYPES): + return False + if isinstance(value, str): + if len(value) > CLI_SSO_CLAIM_MAX_SCALAR_LENGTH: + return False + if value.startswith("eyJ") and value.count(".") >= 2: + return False + return True + + +def _sso_result_to_dict(result: Union[CustomOpenID, OpenID, dict]) -> Dict[str, Any]: + if isinstance(result, dict): + return result + if hasattr(result, "model_dump"): + dumped = result.model_dump() + if isinstance(dumped, dict): + return cast(Dict[str, Any], dumped) + return {} + + +def _get_nested_claim_value(data: Dict[str, Any], claim_path: str) -> Any: + """Resolve a dot-notation claim path against an SSO result dict. + + Unlike ``get_nested_value``, this does not strip a leading ``metadata.`` + prefix, since OIDC claims may legitimately use ``metadata`` as a top-level + key. + """ + if not claim_path: + return None + if claim_path in data: + return data[claim_path] + placeholder = "\x00" + parts = claim_path.replace("\\.", placeholder).split(".") + parts = [p.replace(placeholder, ".") for p in parts] + current: Any = data + for part in parts: + if isinstance(current, dict) and part in current: + current = current[part] + else: + return None + return current + + +def _extract_sso_claim_value( + result: Union[CustomOpenID, OpenID, dict], claim_path: str +) -> Any: + extra_fields = getattr(result, "extra_fields", None) + if isinstance(extra_fields, dict): + if claim_path in extra_fields: + return extra_fields[claim_path] + nested = _get_nested_claim_value(extra_fields, claim_path) + if nested is not None: + return nested + + if isinstance(result, dict): + return _get_nested_claim_value(result, claim_path) + + result_dict = _sso_result_to_dict(result) + return _get_nested_claim_value(result_dict, claim_path) + + +def _set_nested_metadata_value( + metadata: Dict[str, Any], key_path: str, value: Any +) -> None: + placeholder = "\x00" + parts = key_path.replace("\\.", placeholder).split(".") + parts = [p.replace(placeholder, ".") for p in parts] + current: Any = metadata + for part in parts[:-1]: + existing = current.get(part) + if not isinstance(existing, dict): + existing = {} + current[part] = existing + current = existing + current[parts[-1]] = value + + +def _flatten_cli_sso_metadata_for_poll( + metadata: Dict[str, Any], +) -> Dict[str, Union[str, int, float, bool]]: + """Expose scalar attribution metadata as a flat dict for CLI poll responses.""" + flattened: Dict[str, Union[str, int, float, bool]] = {} + stack: List[Tuple[str, Any]] = [("", metadata)] + while stack: + prefix, value = stack.pop() + if isinstance(value, dict): + for key, nested in value.items(): + nested_prefix = f"{prefix}.{key}" if prefix else key + stack.append((nested_prefix, nested)) + elif _is_safe_cli_sso_scalar_claim_value(value): + flattened[prefix] = value + return flattened + + +def build_cli_sso_attribution_metadata( + result: Union[CustomOpenID, OpenID, dict], +) -> Dict[str, Any]: + """ + Build allowlisted, non-secret scalar attribution metadata from an SSO result. + + Sources are configured via CLI_SSO_CLAIM_MAP / LITELLM_CLI_SSO_CLAIM_MAP and + may include claims captured by GENERIC_USER_EXTRA_ATTRIBUTES on CustomOpenID. + """ + claim_map = _parse_cli_sso_claim_map() + if not claim_map: + return {} + + metadata: Dict[str, Any] = {} + for source_claim, dest_key in claim_map: + if not _is_safe_cli_sso_metadata_dest_key(dest_key): + verbose_proxy_logger.debug( + f"Skipping unsafe CLI SSO metadata destination key: {dest_key}" + ) + continue + + raw_value = _extract_sso_claim_value(result=result, claim_path=source_claim) + if not _is_safe_cli_sso_scalar_claim_value(raw_value): + continue + + _set_nested_metadata_value( + metadata=metadata, key_path=dest_key, value=raw_value + ) + + return metadata + + +def _merge_cli_sso_attribution_metadata( + existing_metadata: Dict[str, Any], attribution_metadata: Dict[str, Any] +) -> Dict[str, Any]: + """Merge attribution metadata into existing user metadata in-place. + + Preserves original value types (in particular, string claim values that + happen to look numeric are NOT coerced to ``int``/``float``). Nested dicts + are merged iteratively so attribution claims do not clobber unrelated keys + under the same parent. + """ + pending: List[Tuple[Dict[str, Any], Dict[str, Any]]] = [ + (existing_metadata, attribution_metadata) + ] + while pending: + target, source = pending.pop() + for key, value in source.items(): + if value is None: + continue + existing_value = target.get(key) + if isinstance(value, dict) and isinstance(existing_value, dict): + pending.append((existing_value, value)) + else: + target[key] = value + return existing_metadata + + +async def _persist_cli_sso_user_metadata( + prisma_client: PrismaClient, + user_id: str, + attribution_metadata: Dict[str, Any], +) -> None: + if not attribution_metadata: + return + + try: + user_row = await UserRepository(prisma_client).table.find_unique( + where={"user_id": user_id} + ) + existing_metadata: Dict[str, Any] = {} + if user_row is not None: + row_metadata = user_row.metadata + if isinstance(row_metadata, dict): + existing_metadata = deepcopy(row_metadata) + + merged_metadata = _merge_cli_sso_attribution_metadata( + existing_metadata=existing_metadata, + attribution_metadata=attribution_metadata, + ) + await UserRepository(prisma_client).table.update_many( + where={"user_id": user_id}, + data={"metadata": merged_metadata}, + ) + verbose_proxy_logger.info( + f"Persisted CLI SSO attribution metadata for user {user_id}: " + f"{list(_flatten_cli_sso_metadata_for_poll(attribution_metadata).keys())}" + ) + except Exception as e: + verbose_proxy_logger.error( + f"Failed to persist CLI SSO attribution metadata for user {user_id}: {e}" + ) + + +def _cli_poll_attribution_metadata_from_session( + session_data: Dict[str, Any], +) -> Dict[str, Union[str, int, float, bool]]: + stored = session_data.get("attribution_metadata") + if isinstance(stored, dict): + return _flatten_cli_sso_metadata_for_poll(stored) + return {} + + +def _render_cli_sso_verification_page( + verify_url: str, + browser_complete_token: str, + prefill_user_code: str | None = None, +) -> str: + escaped_verify_url = escape(verify_url, quote=True) + escaped_browser_complete_token = escape(browser_complete_token, quote=True) + user_code_value_attr = ( + f' value="{escape(prefill_user_code, quote=True)}"' if prefill_user_code else "" + ) + instructions = ( + "Confirm the verification code below to finish this login." + if prefill_user_code + else "Enter the verification code shown in your terminal to finish this login." + ) + return f""" + + + + LiteLLM CLI Login + + + +
+

Complete CLI Login

+

{instructions}

+
+ + + + +
+
+ + + """ + + +@router.post("/sso/cli/start", tags=["experimental"], include_in_schema=False) +async def cli_sso_start(request: Request): + from litellm.proxy.proxy_server import general_settings, user_api_key_cache + + _check_cli_sso_start_rate_limit( + request=request, + cache=user_api_key_cache, + use_x_forwarded_for=bool( + (general_settings or {}).get("use_x_forwarded_for", False) + ), + ) + + login_id = f"cli-{secrets.token_urlsafe(24)}" + poll_secret = secrets.token_urlsafe(32) + user_code = _generate_cli_sso_user_code() + + flow = { + "poll_secret_hash": _hash_cli_sso_secret(poll_secret), + "user_code_hash": _hash_cli_sso_secret(_normalize_cli_sso_user_code(user_code)), + "sso_complete": False, + "user_code_verified": False, + "session_data": None, + } + _set_cli_sso_flow(login_id=login_id, cache=user_api_key_cache, flow=flow) + + verification_uri_complete: str | None = ( + ( + get_custom_url( + request_base_url=str(request.base_url), route="sso/key/generate" + ) + + "?" + + urlencode( + { + "source": LITELLM_CLI_SOURCE_IDENTIFIER, + "key": login_id, + "user_code": user_code, + } + ) + ) + if _cli_sso_verification_uri_complete_enabled() + else None + ) + return _cli_sso_start_response_body( + login_id=login_id, + poll_secret=poll_secret, + user_code=user_code, + verification_uri_complete=verification_uri_complete, + ) + + +@router.post( + "/sso/cli/complete/{login_id}", tags=["experimental"], include_in_schema=False +) +async def cli_sso_complete(request: Request, login_id: str): + from fastapi.responses import HTMLResponse + + from litellm.proxy.common_utils.html_forms.cli_sso_success import ( + render_cli_sso_success_page, + ) + from litellm.proxy.proxy_server import user_api_key_cache + + flow = _get_cli_sso_flow_or_raise(login_id=login_id, cache=user_api_key_cache) + if not flow.get("sso_complete") or not flow.get("session_data"): + raise HTTPException(status_code=400, detail="CLI login is not ready") + + body = (await request.body()).decode("utf-8") + form_values = parse_qs(body) + supplied_user_code = (form_values.get("user_code") or [""])[0] + supplied_browser_complete_token = ( + form_values.get("browser_complete_token") or [""] + )[0] + supplied_user_code_hash = _hash_cli_sso_secret( + _normalize_cli_sso_user_code(supplied_user_code) + ) + supplied_browser_complete_token_hash = _hash_cli_sso_secret( + supplied_browser_complete_token + ) + + expected_user_code_hash = flow.get("user_code_hash") + if not isinstance(expected_user_code_hash, str) or not secrets.compare_digest( + supplied_user_code_hash, expected_user_code_hash + ): + raise HTTPException(status_code=400, detail="Invalid verification code") + + expected_browser_complete_token_hash = flow.get("browser_complete_token_hash") + if not isinstance( + expected_browser_complete_token_hash, str + ) or not secrets.compare_digest( + supplied_browser_complete_token_hash, expected_browser_complete_token_hash + ): + raise HTTPException(status_code=400, detail="Invalid verification code") + + flow["user_code_verified"] = True + _set_cli_sso_flow(login_id=login_id, cache=user_api_key_cache, flow=flow) + + html_content = render_cli_sso_success_page() + return HTMLResponse(content=html_content, status_code=200) + + +def normalize_email(email: Optional[str]) -> Optional[str]: + """ + Normalize email address to lowercase for consistent storage and comparison. + + Email addresses should be treated as case-insensitive for SSO purposes, + even though RFC 5321 technically allows case-sensitive local parts. + This prevents issues where SSO providers return emails with different casing + than what's stored in the database. + + Args: + email: Email address to normalize, can be None + + Returns: + Lowercased email address, or None if input is None + """ + if email is None: + return None + return email.lower() if isinstance(email, str) else email + + +def determine_role_from_groups( + user_groups: List[str], + role_mappings: "RoleMappings", +) -> Optional[LitellmUserRoles]: + """ + Determine the highest privilege role for a user based on their groups. + + Role hierarchy (highest to lowest): + - proxy_admin + - proxy_admin_viewer + - internal_user + - internal_user_viewer + + Args: + user_groups: List of group names from the SSO token + role_mappings: RoleMappings configuration object + + Returns: + The highest privilege role found, or default_role if no matches, or None + """ + if not role_mappings.roles: + # No role mappings configured, return default_role + return role_mappings.default_role + + # Role hierarchy (highest to lowest) + role_hierarchy = [ + LitellmUserRoles.PROXY_ADMIN, + LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY, + LitellmUserRoles.INTERNAL_USER, + LitellmUserRoles.INTERNAL_USER_VIEW_ONLY, + ] + + # Convert user_groups to a set for efficient lookup + user_groups_set = set(user_groups) if isinstance(user_groups, list) else set() + + # Find the highest privilege role the user belongs to + for role in role_hierarchy: + if role in role_mappings.roles: + role_groups = role_mappings.roles[role] + if isinstance(role_groups, list) and user_groups_set.intersection( + set(role_groups) + ): + verbose_proxy_logger.debug( + f"User groups {user_groups} matched role '{role.value}' via groups: {role_groups}" + ) + return role + + # No matching groups found, return default_role + verbose_proxy_logger.debug( + f"User groups {user_groups} did not match any role mappings, using default_role: {role_mappings.default_role}" + ) + return role_mappings.default_role + + +def process_sso_jwt_access_token( + access_token_str: Optional[str], + sso_jwt_handler: Optional[JWTHandler], + result: Union[OpenID, dict, None], + role_mappings: Optional["RoleMappings"] = None, +) -> Optional[dict]: + """ + Process SSO JWT access token and extract team IDs and user role if available. + + This function decodes the JWT access token and extracts team IDs and user + role, then sets them on the result object. Role extraction from the access + token is needed because some SSO providers (e.g., Keycloak) do not include + role claims in the UserInfo endpoint response. + + Args: + access_token_str: The JWT access token string + sso_jwt_handler: SSO-specific JWT handler for team ID extraction + result: The SSO result object to update with team IDs and role + role_mappings: Optional role mappings configuration for group-based role determination + + Returns: + The decoded access token payload dict, or None if decoding failed or + inputs were missing. Callers can pass this to _sync_user_role_from_jwt_role_map + so it has access to custom role claims (e.g. custom_roles) that are + encoded inside the JWT but stripped from received_response. + """ + if access_token_str and result: + import jwt + + try: + access_token_payload = jwt.decode( + access_token_str, options={"verify_signature": False} + ) + except jwt.exceptions.DecodeError: + verbose_proxy_logger.debug( + "Access token is not a valid JWT (possibly an opaque token), skipping JWT-based extraction" + ) + return None + + # Extract team IDs from access token if sso_jwt_handler is available + if sso_jwt_handler: + if isinstance(result, dict): + result_team_ids: Optional[List[str]] = result.get("team_ids", []) + if not result_team_ids: + team_ids = sso_jwt_handler.get_team_ids_from_jwt( + access_token_payload + ) + result["team_ids"] = team_ids + else: + result_team_ids = getattr(result, "team_ids", []) if result else [] + if not result_team_ids: + team_ids = sso_jwt_handler.get_team_ids_from_jwt( + access_token_payload + ) + setattr(result, "team_ids", team_ids) + + # Extract user role from access token if not already set from UserInfo + existing_role = ( + result.get("user_role") + if isinstance(result, dict) + else getattr(result, "user_role", None) + ) + if existing_role is None: + user_role: Optional[LitellmUserRoles] = None + + # Try role_mappings first (group-based role determination) + if role_mappings is not None and role_mappings.roles: + group_claim = role_mappings.group_claim + user_groups_raw: Any = get_nested_value( + access_token_payload, group_claim + ) + + user_groups: List[str] = [] + if isinstance(user_groups_raw, list): + user_groups = [str(g) for g in user_groups_raw] + elif isinstance(user_groups_raw, str): + user_groups = [ + g.strip() for g in user_groups_raw.split(",") if g.strip() + ] + elif user_groups_raw is not None: + user_groups = [str(user_groups_raw)] + + if user_groups: + user_role = determine_role_from_groups(user_groups, role_mappings) + verbose_proxy_logger.debug( + f"Determined role '{user_role}' from access token groups '{user_groups}' using role_mappings" + ) + elif role_mappings.default_role: + user_role = role_mappings.default_role + + # Fallback: try GENERIC_USER_ROLE_ATTRIBUTE on the access token payload + if user_role is None: + generic_user_role_attribute_name = os.getenv( + "GENERIC_USER_ROLE_ATTRIBUTE", "role" + ) + user_role_from_token = get_nested_value( + access_token_payload, generic_user_role_attribute_name + ) + if user_role_from_token is not None: + user_role = get_litellm_user_role(user_role_from_token) + verbose_proxy_logger.debug( + f"Extracted role '{user_role}' from access token field '{generic_user_role_attribute_name}'" + ) + + if user_role is not None: + if isinstance(result, dict): + result["user_role"] = user_role + else: + setattr(result, "user_role", user_role) + verbose_proxy_logger.debug( + f"Set user_role='{user_role}' from JWT access token" + ) + + return access_token_payload + + return None + + +@router.get("/sso/key/generate", tags=["experimental"], include_in_schema=False) +async def google_login( + request: Request, + source: Optional[str] = None, + key: Optional[str] = None, + existing_key: Optional[str] = None, + return_to: Optional[str] = None, + user_code: str | None = None, +): + """ + Create Proxy API Keys using Google Workspace SSO. Requires setting PROXY_BASE_URL in .env + PROXY_BASE_URL should be the your deployed proxy endpoint, e.g. PROXY_BASE_URL="https://litellm-production-7002.up.railway.app/" + Example: + """ + from litellm.proxy.proxy_server import ( + general_settings, + premium_user, + prisma_client, + user_api_key_cache, + user_custom_ui_sso_sign_in_handler, + ) + + microsoft_client_id = os.getenv("MICROSOFT_CLIENT_ID", None) + google_client_id = os.getenv("GOOGLE_CLIENT_ID", None) + generic_client_id = os.getenv("GENERIC_CLIENT_ID", None) + + ####### Check if UI is disabled ####### + _disable_ui_flag = os.getenv("DISABLE_ADMIN_UI") + if _disable_ui_flag is not None: + is_disabled = str_to_bool(value=_disable_ui_flag) + if is_disabled: + return admin_ui_disabled() + + ####### Check if user is a Enterprise / Premium User ####### + if ( + microsoft_client_id is not None + or google_client_id is not None + or generic_client_id is not None + ): + if premium_user is not True: + # Check if under 'free SSO user' limit + if prisma_client is not None: + total_users = await UserRepository(prisma_client).table.count() + if total_users and total_users > 5: + raise ProxyException( + message="You must be a LiteLLM Enterprise user to use SSO for more than 5 users. If you have a license please set `LITELLM_LICENSE` in your env. If you want to obtain a license meet with us here: https://enterprise.litellm.ai/demo You are seeing this error message because You set one of `MICROSOFT_CLIENT_ID`, `GOOGLE_CLIENT_ID`, or `GENERIC_CLIENT_ID` in your env. Please unset this", + type=ProxyErrorTypes.auth_error, + param="premium_user", + code=status.HTTP_403_FORBIDDEN, + ) + else: + raise ProxyException( + message=CommonProxyErrors.db_not_connected_error.value, + type=ProxyErrorTypes.auth_error, + param="premium_user", + code=status.HTTP_403_FORBIDDEN, + ) + + ####### Detect DB + MASTER KEY in .env ####### + missing_env_vars = show_missing_vars_in_env() + if missing_env_vars is not None: + return missing_env_vars + + # get url from request - always use regular callback, but set state for CLI + redirect_url = SSOAuthenticationHandler.get_redirect_url_for_sso( + request=request, + sso_callback_route="sso/callback", + ) + + if source == LITELLM_CLI_SOURCE_IDENTIFIER: + _get_cli_sso_flow_or_raise(login_id=key, cache=user_api_key_cache) + + # Store CLI login handle in state for OAuth flow + cli_state: Optional[str] = SSOAuthenticationHandler._get_cli_state( + source=source, + key=key, + user_code=(user_code if _cli_sso_verification_uri_complete_enabled() else None), + ) + + # check if user defined a custom auth sso sign in handler, if yes, use it + if user_custom_ui_sso_sign_in_handler is not None: + try: + from litellm_enterprise.proxy.auth.custom_sso_handler import ( # type: ignore[import-untyped] + EnterpriseCustomSSOHandler, + ) + + return await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in( + request=request, + ) + except ImportError: + raise ValueError( + "Enterprise features are not available. Custom UI SSO sign-in requires LiteLLM Enterprise." + ) + + # Check if we should use SSO handler + if ( + SSOAuthenticationHandler.should_use_sso_handler( + microsoft_client_id=microsoft_client_id, + google_client_id=google_client_id, + generic_client_id=generic_client_id, + ) + is True + ): + verbose_proxy_logger.info(f"Redirecting to SSO login for {redirect_url}") + sso_redirect = await SSOAuthenticationHandler.get_sso_login_redirect( + redirect_url=redirect_url, + microsoft_client_id=microsoft_client_id, + google_client_id=google_client_id, + generic_client_id=generic_client_id, + state=cli_state, + request=request, + ) + if return_to is not None and sso_redirect is not None: + if SSOAuthenticationHandler._validate_return_to(return_to): + sso_redirect.set_cookie( + key="litellm_cp_return_to", + value=return_to, + max_age=600, + httponly=True, + samesite="lax", + ) + return sso_redirect + + from fastapi.responses import HTMLResponse + + hide_default_credentials_hint = ( + os.getenv("LITELLM_HIDE_DEFAULT_CREDENTIALS_HINT", "false").lower() == "true" + or general_settings.get("hide_default_credentials_hint", False) is True + ) + return HTMLResponse( + content=build_ui_login_form( + show_deprecation_banner=True, + hide_default_credentials_hint=hide_default_credentials_hint, + ), + status_code=200, + ) + + +def generic_response_convertor( + response, + jwt_handler: JWTHandler, + sso_jwt_handler: Optional[JWTHandler] = None, + role_mappings: Optional["RoleMappings"] = None, + team_mappings: Optional["TeamMappings"] = None, +) -> CustomOpenID: + generic_user_id_attribute_name = os.getenv( + "GENERIC_USER_ID_ATTRIBUTE", "preferred_username" + ) + generic_user_display_name_attribute_name = os.getenv( + "GENERIC_USER_DISPLAY_NAME_ATTRIBUTE", "sub" + ) + generic_user_email_attribute_name = os.getenv( + "GENERIC_USER_EMAIL_ATTRIBUTE", "email" + ) + + generic_user_first_name_attribute_name = os.getenv( + "GENERIC_USER_FIRST_NAME_ATTRIBUTE", "first_name" + ) + generic_user_last_name_attribute_name = os.getenv( + "GENERIC_USER_LAST_NAME_ATTRIBUTE", "last_name" + ) + + generic_provider_attribute_name = os.getenv( + "GENERIC_USER_PROVIDER_ATTRIBUTE", "provider" + ) + + generic_user_role_attribute_name = os.getenv("GENERIC_USER_ROLE_ATTRIBUTE", "role") + + generic_user_extra_attributes = os.getenv("GENERIC_USER_EXTRA_ATTRIBUTES", None) + + verbose_proxy_logger.debug( + f" generic_user_id_attribute_name: {generic_user_id_attribute_name}\n generic_user_email_attribute_name: {generic_user_email_attribute_name}" + ) + + all_teams = [] + if sso_jwt_handler is not None: + team_ids = sso_jwt_handler.get_all_jwt_team_ids(cast(dict, response)) + all_teams.extend(team_ids) + + if team_mappings is not None and team_mappings.team_ids_jwt_field is not None: + team_ids_from_db_mapping: Optional[List[str]] = get_nested_value( + data=cast(dict, response), + key_path=team_mappings.team_ids_jwt_field, + default=[], + ) + if team_ids_from_db_mapping: + all_teams.extend(team_ids_from_db_mapping) + verbose_proxy_logger.debug( + f"Loaded team_ids from DB team_mappings.team_ids_jwt_field='{team_mappings.team_ids_jwt_field}': {team_ids_from_db_mapping}" + ) + else: + team_ids = jwt_handler.get_all_jwt_team_ids(cast(dict, response)) + all_teams.extend(team_ids) + + # Determine user role based on role_mappings if available + # Only apply role_mappings for GENERIC SSO provider + user_role: Optional[LitellmUserRoles] = None + + if role_mappings is not None and role_mappings.provider.lower() in [ + "generic", + "okta", + ]: + # Use role_mappings to determine role from groups + group_claim = role_mappings.group_claim + user_groups_raw: Any = get_nested_value(response, group_claim) + + # Handle different formats: could be a list, string (comma-separated), or single value + user_groups: List[str] = [] + if isinstance(user_groups_raw, list): + user_groups = [str(g) for g in user_groups_raw] + elif isinstance(user_groups_raw, str): + # Handle comma-separated string + user_groups = [g.strip() for g in user_groups_raw.split(",") if g.strip()] + elif user_groups_raw is not None: + # Single value + user_groups = [str(user_groups_raw)] + + if user_groups: + user_role = determine_role_from_groups(user_groups, role_mappings) + verbose_proxy_logger.debug( + f"Determined role '{user_role.value if user_role else None}' from groups '{user_groups}' using role_mappings" + ) + else: + # No groups found, use default_role + user_role = role_mappings.default_role + verbose_proxy_logger.debug( + f"No groups found in '{group_claim}', using default_role: {role_mappings.default_role}" + ) + + # Fallback to existing logic if role_mappings not used + if user_role is None: + user_role_from_sso = get_nested_value( + response, generic_user_role_attribute_name + ) + if user_role_from_sso is not None: + role = get_litellm_user_role(user_role_from_sso) + if role is not None: + user_role = role + verbose_proxy_logger.debug( + f"Found valid LitellmUserRoles '{role.value}' from SSO attribute '{generic_user_role_attribute_name}'" + ) + + # Build extra_fields dict from GENERIC_USER_EXTRA_ATTRIBUTES if specified + extra_fields: Optional[Dict[str, Any]] = None + if generic_user_extra_attributes: + extra_fields = {} + for attr_name in generic_user_extra_attributes.split(","): + attr_name = attr_name.strip() + extra_fields[attr_name] = get_nested_value(response, attr_name) + + return CustomOpenID( + id=get_nested_value(response, generic_user_id_attribute_name), + display_name=get_nested_value( + response, generic_user_display_name_attribute_name + ), + email=normalize_email( + get_nested_value(response, generic_user_email_attribute_name) + ), + first_name=get_nested_value(response, generic_user_first_name_attribute_name), + last_name=get_nested_value(response, generic_user_last_name_attribute_name), + provider=get_nested_value(response, generic_provider_attribute_name), + team_ids=all_teams, + user_role=user_role, + extra_fields=extra_fields, + ) + + +def _setup_generic_sso_env_vars( + generic_client_id: str, redirect_url: str +) -> Tuple[str, List[str], str, str, str, bool]: + """Setup and validate Generic SSO environment variables.""" + generic_client_secret = os.getenv("GENERIC_CLIENT_SECRET", None) + generic_scope = os.getenv("GENERIC_SCOPE", "openid email profile").split(" ") + generic_authorization_endpoint = os.getenv("GENERIC_AUTHORIZATION_ENDPOINT", None) + generic_token_endpoint = os.getenv("GENERIC_TOKEN_ENDPOINT", None) + generic_userinfo_endpoint = os.getenv("GENERIC_USERINFO_ENDPOINT", None) + generic_include_client_id = ( + os.getenv("GENERIC_INCLUDE_CLIENT_ID", "false").lower() == "true" + ) + + # Validate required environment variables + if generic_client_secret is None: + raise ProxyException( + message="GENERIC_CLIENT_SECRET not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_CLIENT_SECRET", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if generic_authorization_endpoint is None: + raise ProxyException( + message="GENERIC_AUTHORIZATION_ENDPOINT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_AUTHORIZATION_ENDPOINT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if generic_token_endpoint is None: + raise ProxyException( + message="GENERIC_TOKEN_ENDPOINT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_TOKEN_ENDPOINT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if generic_userinfo_endpoint is None: + raise ProxyException( + message="GENERIC_USERINFO_ENDPOINT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_USERINFO_ENDPOINT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + + verbose_proxy_logger.debug( + f"authorization_endpoint: {generic_authorization_endpoint}\ntoken_endpoint: {generic_token_endpoint}\nuserinfo_endpoint: {generic_userinfo_endpoint}" + ) + verbose_proxy_logger.debug( + f"GENERIC_REDIRECT_URI: {redirect_url}\nGENERIC_CLIENT_ID: {generic_client_id}\n" + ) + + return ( + generic_client_secret, + generic_scope, + generic_authorization_endpoint, + generic_token_endpoint, + generic_userinfo_endpoint, + generic_include_client_id, + ) + + +async def _setup_team_mappings() -> Optional["TeamMappings"]: + """Setup team mappings from SSO database settings.""" + team_mappings: Optional["TeamMappings"] = None + try: + from litellm.proxy.utils import get_prisma_client_or_throw + + prisma_client = get_prisma_client_or_throw( + "Prisma client is None, connect a database to your proxy" + ) + + sso_db_record = await SSOConfigRepository(prisma_client).table.find_unique( + where={"id": "sso_config"} + ) + + if sso_db_record and sso_db_record.sso_settings: + sso_settings_dict = dict(sso_db_record.sso_settings) + team_mappings_data = sso_settings_dict.get("team_mappings") + + if team_mappings_data: + from litellm.types.proxy.management_endpoints.ui_sso import TeamMappings + + if isinstance(team_mappings_data, dict): + team_mappings = TeamMappings(**team_mappings_data) + elif isinstance(team_mappings_data, TeamMappings): + team_mappings = team_mappings_data + + if team_mappings and team_mappings.team_ids_jwt_field: + verbose_proxy_logger.debug( + f"Loaded team_mappings with team_ids_jwt_field: '{team_mappings.team_ids_jwt_field}'" + ) + except Exception as e: + verbose_proxy_logger.debug( + f"Could not load team_mappings from database: {e}. Continuing with config-based team mapping." + ) + + return team_mappings + + +async def _setup_role_mappings() -> Optional["RoleMappings"]: + """Setup role mappings from SSO database settings.""" + role_mappings: Optional["RoleMappings"] = None + try: + from litellm.proxy.utils import get_prisma_client_or_throw + + prisma_client = get_prisma_client_or_throw( + "Prisma client is None, connect a database to your proxy" + ) + + sso_db_record = await SSOConfigRepository(prisma_client).table.find_unique( + where={"id": "sso_config"} + ) + + if sso_db_record and sso_db_record.sso_settings: + sso_settings_dict = dict(sso_db_record.sso_settings) + role_mappings_data = sso_settings_dict.get("role_mappings") + + if role_mappings_data: + from litellm.types.proxy.management_endpoints.ui_sso import RoleMappings + + if isinstance(role_mappings_data, dict): + role_mappings = RoleMappings(**role_mappings_data) + elif isinstance(role_mappings_data, RoleMappings): + role_mappings = role_mappings_data + + if role_mappings: + verbose_proxy_logger.debug( + f"Loaded role_mappings for provider '{role_mappings.provider}'" + ) + except Exception as e: + verbose_proxy_logger.debug( + f"Could not load role_mappings from database: {e}. Continuing with existing role logic." + ) + + generic_role_mappings = os.getenv("GENERIC_ROLE_MAPPINGS_ROLES", None) + generic_role_mappings_group_claim = os.getenv( + "GENERIC_ROLE_MAPPINGS_GROUP_CLAIM", None + ) + generic_role_mappings_default_role = os.getenv( + "GENERIC_ROLE_MAPPINGS_DEFAULT_ROLE", None + ) + if generic_role_mappings is not None: + verbose_proxy_logger.debug( + "Found role_mappings for generic provider in environment variables" + ) + import ast + + try: + generic_user_role_mappings_data: Dict[LitellmUserRoles, List[str]] = ( + ast.literal_eval(generic_role_mappings) + ) + if isinstance(generic_user_role_mappings_data, dict): + from litellm.types.proxy.management_endpoints.ui_sso import RoleMappings + + role_mappings_data = { + "provider": "generic", + "group_claim": generic_role_mappings_group_claim, + "default_role": generic_role_mappings_default_role, + "roles": generic_user_role_mappings_data, + } + + role_mappings = RoleMappings(**role_mappings_data) + verbose_proxy_logger.debug( + f"Loaded role_mappings from environments for provider '{role_mappings.provider}'." + ) + return role_mappings + except TypeError as e: + verbose_proxy_logger.warning( + f"Error decoding role mappings from environment variables: {e}. Continuing with existing role logic." + ) + return role_mappings + + +def _parse_generic_sso_headers() -> dict: + """Parse comma-separated GENERIC_SSO_HEADERS env var into a dict.""" + raw = os.getenv("GENERIC_SSO_HEADERS", None) + if raw is None: + return {} + result: Dict[str, str] = {} + for header in raw.split(","): + header = header.strip() + if header: + key, value = header.split("=") + result[key] = value + return result + + +def _handle_generic_sso_error( + e: Exception, + generic_authorization_endpoint: Optional[str], + generic_token_endpoint: Optional[str], + additional_headers: dict, +) -> NoReturn: + """Handle errors from generic SSO verify_and_process. Always re-raises.""" + error_message = str(e) + + # Surface a helpful PKCE misconfiguration hint only when: + # 1. The error mentions PKCE/code verifier, AND + # 2. PKCE is not currently configured (GENERIC_CLIENT_USE_PKCE != true) + pkce_configured = os.getenv("GENERIC_CLIENT_USE_PKCE", "false").lower() == "true" + if not pkce_configured and ( + "PKCE" in error_message or "code verifier" in error_message.lower() + ): + is_okta = ( + generic_authorization_endpoint + and "okta" in generic_authorization_endpoint.lower() + ) or (generic_token_endpoint and "okta" in generic_token_endpoint.lower()) + provider_name = "Okta" if is_okta else "Your OAuth provider" + + detailed_message = ( + f"SSO authentication failed: {provider_name} requires PKCE (Proof Key for Code Exchange) " + f"but it's not enabled in your LiteLLM configuration.\n\n" + f"SOLUTION: Add this environment variable and restart your proxy:\n" + f" GENERIC_CLIENT_USE_PKCE=true\n\n" + ) + if is_okta: + detailed_message += ( + "For AWS ECS: Add the environment variable to your task definition.\n" + "For Docker: Add -e GENERIC_CLIENT_USE_PKCE=true to your docker run command.\n" + "For .env file: Add GENERIC_CLIENT_USE_PKCE=true to your .env file.\n\n" + ) + detailed_message += f"Original error: {error_message}" + + raise ProxyException( + message=detailed_message, + type=ProxyErrorTypes.auth_error, + param="GENERIC_CLIENT_USE_PKCE", + code=status.HTTP_401_UNAUTHORIZED, + ) + + if isinstance(e, ProxyException): + verbose_proxy_logger.error( + "SSO authentication failed: %s. Passed in headers: %s", + e, + additional_headers, + ) + else: + verbose_proxy_logger.exception( + "Error verifying and processing generic SSO: %s. Passed in headers: %s", + e, + additional_headers, + ) + raise e + + +async def get_generic_sso_response( + request: Request, + jwt_handler: JWTHandler, + sso_jwt_handler: Optional[ + JWTHandler + ], # sso specific jwt handler - used for restricted sso group access control + generic_client_id: str, + redirect_url: str, +) -> Tuple[ + Union[OpenID, dict], Optional[dict], Optional[dict] +]: # (result, received_response, access_token_payload) + # make generic sso provider + from fastapi_sso.sso.base import DiscoveryDocument + from fastapi_sso.sso.generic import create_provider + + received_response: Optional[dict] = None + + # Setup environment variables + ( + generic_client_secret, + generic_scope, + generic_authorization_endpoint, + generic_token_endpoint, + generic_userinfo_endpoint, + generic_include_client_id, + ) = _setup_generic_sso_env_vars(generic_client_id, redirect_url) + + discovery = DiscoveryDocument( + authorization_endpoint=generic_authorization_endpoint, + token_endpoint=generic_token_endpoint, + userinfo_endpoint=generic_userinfo_endpoint, + ) + + role_mappings = await _setup_role_mappings() + team_mappings = await _setup_team_mappings() + + def response_convertor(response, client): + nonlocal received_response # return for user debugging + received_response = response + return generic_response_convertor( + response=response, + jwt_handler=jwt_handler, + sso_jwt_handler=sso_jwt_handler, + role_mappings=role_mappings, + team_mappings=team_mappings, + ) + + SSOProvider = create_provider( + name="oidc", + discovery_document=discovery, + response_convertor=response_convertor, + ) + generic_sso = SSOProvider( + client_id=generic_client_id, + client_secret=generic_client_secret, + redirect_uri=redirect_url, + allow_insecure_http=True, + scope=generic_scope, + ) + verbose_proxy_logger.debug("calling generic_sso.verify_and_process") + additional_generic_sso_headers_dict = _parse_generic_sso_headers() + + code_verifier: Optional[str] = ( + None # assigned inside try; initialized for type tracking + ) + access_token_payload: Optional[dict] = None # decoded JWT access token claims + + try: + token_exchange_params = ( + await SSOAuthenticationHandler.prepare_token_exchange_parameters( + request=request, + generic_include_client_id=generic_include_client_id, + ) + ) + + # Extract code_verifier (and the cache key for deferred deletion) before calling fastapi-sso + code_verifier = token_exchange_params.pop("code_verifier", None) + pkce_cache_key = token_exchange_params.pop("_pkce_cache_key", None) + + # Get authorization code from query params (only used in the PKCE path below; + # the non-PKCE path delegates to verify_and_process which handles OAuth error + # callbacks — user-denied, CSRF mismatch — internally). + authorization_code = request.query_params.get("code") + + if code_verifier: + # State-to-session-cookie binding. The non-PKCE branch below + # delegates to fastapi-sso's ``verify_and_process``, which + # performs its own session-cookie check. The PKCE branch + # bypasses that helper, so we validate the URL ``state`` + # against the ``litellm_oauth_state`` cookie set on the + # redirect response — without this an attacker can pre-mint + # a state + cached PKCE verifier and hijack a victim's auth + # code (Login-CSRF / token theft). + url_state = request.query_params.get("state") + cookie_state = request.cookies.get("litellm_oauth_state") + if ( + not url_state + or not cookie_state + or not secrets.compare_digest(url_state, cookie_state) + ): + raise ProxyException( + message=( + "Invalid OAuth state parameter — does not match " + "the browser-bound state cookie." + ), + type=ProxyErrorTypes.auth_error, + param="state", + code=status.HTTP_400_BAD_REQUEST, + ) + if not authorization_code: + raise ProxyException( + message="Missing authorization code in callback", + type=ProxyErrorTypes.auth_error, + param="code", + code=status.HTTP_400_BAD_REQUEST, + ) + if not generic_client_id: + raise ProxyException( + message="GENERIC_CLIENT_ID must be set when PKCE is enabled", + type=ProxyErrorTypes.auth_error, + param="GENERIC_CLIENT_ID", + code=status.HTTP_401_UNAUTHORIZED, + ) + if not generic_token_endpoint: + raise ProxyException( + message="GENERIC_TOKEN_ENDPOINT must be set when PKCE is enabled", + type=ProxyErrorTypes.auth_error, + param="GENERIC_TOKEN_ENDPOINT", + code=status.HTTP_401_UNAUTHORIZED, + ) + # All guards above raise, so authorization_code is a non-empty str here. + # Use an explicit type guard rather than assert (assert is a no-op with -O). + if not isinstance(authorization_code, str): + raise ProxyException( + message="Missing authorization code in callback", + type=ProxyErrorTypes.auth_error, + param="code", + code=status.HTTP_400_BAD_REQUEST, + ) + combined_response = await SSOAuthenticationHandler._pkce_token_exchange( + authorization_code=authorization_code, + code_verifier=code_verifier, + client_id=generic_client_id, + client_secret=generic_client_secret, + token_endpoint=generic_token_endpoint, + userinfo_endpoint=generic_userinfo_endpoint, + include_client_id=generic_include_client_id, + redirect_url=redirect_url, + additional_headers=additional_generic_sso_headers_dict, + ) + # Pass the full response so custom response_convertor implementations + # can access all fields (including id_token for claim extraction). + result = response_convertor(combined_response, generic_sso) + # Strip bearer credentials from combined_response before storing in + # received_response. received_response may appear in restricted-group + # error messages — bearer tokens (access_token, id_token, refresh_token) + # must not be exposed to callers. + # Assign directly rather than relying on nonlocal mutation so that Pyright + # can track that received_response is non-None from this point on. + received_response = { + k: v + for k, v in combined_response.items() + if k not in _OAUTH_TOKEN_FIELDS + } + # In the PKCE path verify_and_process is skipped, so generic_sso.access_token + # is never set. Read the token directly from the exchange response instead so + # process_sso_jwt_access_token can extract JWT-embedded roles/teams. + access_token_str: Optional[str] = combined_response.get("access_token") + else: + result = await generic_sso.verify_and_process( + request, + params=token_exchange_params, + headers=additional_generic_sso_headers_dict, + ) + access_token_str = generic_sso.access_token + + access_token_payload = process_sso_jwt_access_token( + access_token_str, sso_jwt_handler, result, role_mappings=role_mappings + ) + # Delete the single-use PKCE verifier only after all downstream processing + # (response_convertor and process_sso_jwt_access_token) has completed + # successfully. Deleting earlier would consume the verifier on a transient + # failure, forcing the user to restart the entire OAuth flow from scratch. + if pkce_cache_key: + await SSOAuthenticationHandler._delete_pkce_verifier(pkce_cache_key) + + except Exception as e: + _handle_generic_sso_error( + e, + generic_authorization_endpoint, + generic_token_endpoint, + additional_generic_sso_headers_dict, + ) + verbose_proxy_logger.debug("generic result: %s", result) + return result or {}, received_response, access_token_payload + + +async def create_team_member_add_task(team_id, user_info): + """Create a task for adding a member to a team.""" + try: + member = Member(user_id=user_info.user_id, role="user") + team_member_add_request = TeamMemberAddRequest( + member=member, + team_id=team_id, + ) + return await team_member_add( + data=team_member_add_request, + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN), + ) + except Exception as e: + verbose_proxy_logger.debug( + f"[Non-Blocking] Error trying to add sso user to db: {e}" + ) + + +async def add_missing_team_member( + user_info: Union[NewUserResponse, LiteLLM_UserTable], sso_teams: List[str] +): + """ + - Get missing teams (diff b/w user_info.team_ids and sso_teams) + - Add missing user to missing teams + """ + # Handle None as empty list for new users + user_teams = user_info.teams if user_info.teams is not None else [] + missing_teams = set(sso_teams) - set(user_teams) + missing_teams_list = list(missing_teams) + tasks = [] + tasks = [ + create_team_member_add_task(team_id, user_info) + for team_id in missing_teams_list + ] + + try: + await asyncio.gather(*tasks) + except Exception as e: + verbose_proxy_logger.debug( + f"[Non-Blocking] Error trying to add sso user to db: {e}" + ) + + +def get_disabled_non_admin_personal_key_creation(): + key_generation_settings = litellm.key_generation_settings + if key_generation_settings is None: + return False + personal_key_generation = ( + key_generation_settings.get("personal_key_generation") or {} + ) + allowed_user_roles = personal_key_generation.get("allowed_user_roles") or [] + return bool("proxy_admin" in allowed_user_roles) + + +async def get_existing_user_info_from_db( + user_id: Optional[str], + user_email: Optional[str], + prisma_client: PrismaClient, + user_api_key_cache: UserApiKeyCache, + proxy_logging_obj: ProxyLogging, +) -> Optional[LiteLLM_UserTable]: + try: + user_info = await get_user_object( + user_id=user_id, + user_email=user_email, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + user_id_upsert=False, + parent_otel_span=None, + proxy_logging_obj=proxy_logging_obj, + sso_user_id=user_id, + ) + except Exception as e: + verbose_proxy_logger.debug(f"Error getting user object: {e}") + user_info = None + + return user_info + + +async def get_user_info_from_db( + result: Union[CustomOpenID, OpenID, dict], + prisma_client: PrismaClient, + user_api_key_cache: UserApiKeyCache, + proxy_logging_obj: ProxyLogging, + user_email: Optional[str], + user_defined_values: Optional[SSOUserDefinedValues], + alternate_user_id: Optional[str] = None, +) -> Optional[Union[LiteLLM_UserTable, NewUserResponse]]: + try: + potential_user_ids = [] + if alternate_user_id is not None: + potential_user_ids.append(alternate_user_id) + if not isinstance(result, dict): + _id = getattr(result, "id", None) + if _id is not None and isinstance(_id, str): + potential_user_ids.append(_id) + else: + _id = result.get("id", None) + if _id is not None and isinstance(_id, str): + potential_user_ids.append(_id) + + user_email = normalize_email( + getattr(result, "email", None) + if not isinstance(result, dict) + else result.get("email", None) + ) + + user_info: Optional[Union[LiteLLM_UserTable, NewUserResponse]] = None + + for user_id in potential_user_ids: + user_info = await get_existing_user_info_from_db( + user_id=user_id, + user_email=user_email, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=proxy_logging_obj, + ) + if user_info is not None: + break + + verbose_proxy_logger.debug( + f"user_info: {user_info}; litellm.default_internal_user_params: {litellm.default_internal_user_params}" + ) + + # Upsert SSO User to LiteLLM DB + user_info = await SSOAuthenticationHandler.upsert_sso_user( + result=result, + user_info=user_info, + user_email=user_email, + user_defined_values=user_defined_values, + prisma_client=prisma_client, + ) + + await SSOAuthenticationHandler.add_user_to_teams_from_sso_response( + result=result, + user_info=user_info, + ) + + return user_info + except Exception as e: + verbose_proxy_logger.exception( + f"[Non-Blocking] Error trying to add sso user to db: {e}" + ) + + return None + + +def _should_use_role_from_sso_response(sso_role: Optional[str]) -> bool: + """returns true if SSO upsert should use the 'role' defined on the SSO response""" + if sso_role is None: + return False + + if not is_valid_litellm_user_role(sso_role): + verbose_proxy_logger.debug( + f"SSO role '{sso_role}' is not a valid LiteLLM user role. " + "Ignoring role from SSO response. See LitellmUserRoles enum for valid roles." + ) + return False + return True + + +def _build_sso_user_update_data( + result: Optional[Union["CustomOpenID", OpenID, dict]], + user_email: Optional[str], + user_id: Optional[str], +) -> dict: + """ + Build the update data dictionary for SSO user upsert. + + Args: + result: The SSO response containing user information + user_email: The user's email from SSO + user_id: The user's ID for logging purposes + + Returns: + dict: Update data containing user_email and optionally user_role if valid + """ + update_data: dict = {"user_email": normalize_email(user_email)} + + # Get SSO role from result and include if valid + sso_role = getattr(result, "user_role", None) + if sso_role is not None: + # Convert enum to string if needed + sso_role_str = ( + sso_role.value if isinstance(sso_role, LitellmUserRoles) else sso_role + ) + + # Only include if it's a valid LiteLLM role + if _should_use_role_from_sso_response(sso_role_str): + update_data["user_role"] = sso_role_str + verbose_proxy_logger.info( + f"Updating user {user_id} role from SSO: {sso_role_str}" + ) + + return update_data + + +async def _sync_user_role_from_jwt_role_map( + jwt_handler: Optional[JWTHandler], + received_response: Optional[dict], + user_info: Optional[Union[LiteLLM_UserTable, NewUserResponse]], + prisma_client: PrismaClient, + user_api_key_cache: UserApiKeyCache, + user_defined_values: Optional[SSOUserDefinedValues], +) -> None: + """ + Apply jwt_litellm_role_map during SSO login. + + When jwt_litellm_role_map is configured with sync_user_role_and_teams=True, + this ensures SSO users get the same role mapping as API/JWT users. Without + this, the SSO path falls back to INTERNAL_USER_VIEW_ONLY for roles that + don't directly match LitellmUserRoles enum values. + """ + if jwt_handler is None or received_response is None: + return + if not jwt_handler.litellm_jwtauth.sync_user_role_and_teams: + return + if not jwt_handler.litellm_jwtauth.jwt_litellm_role_map: + return + + mapped_role = jwt_handler.map_jwt_role_to_litellm_role(received_response) + if mapped_role is None: + return + + verbose_proxy_logger.info( + f"SSO jwt_litellm_role_map matched role: {mapped_role.value}" + ) + + # Update user_defined_values so downstream code uses the mapped role + if user_defined_values is not None: + user_defined_values["user_role"] = mapped_role.value + + # Update existing DB record if role differs + if user_info is not None and user_info.user_role != mapped_role.value: + await UserRepository(prisma_client).table.update( + where={"user_id": user_info.user_id}, + data={"user_role": mapped_role.value}, + ) + user_info.user_role = mapped_role.value + await user_api_key_cache.async_set_cache( + key=user_info.user_id, + value=user_info, + model_type=LiteLLM_UserTable, + ) + + +def apply_user_info_values_to_sso_user_defined_values( + user_info: Optional[Union[LiteLLM_UserTable, NewUserResponse]], + user_defined_values: Optional[SSOUserDefinedValues], +) -> Optional[SSOUserDefinedValues]: + if user_defined_values is None: + return None + if user_info is not None and user_info.user_id is not None: + user_defined_values["user_id"] = user_info.user_id + + # SSO role takes precedence - only use DB role if SSO didn't provide one + # This ensures SSO is the authoritative source for user roles + sso_role = user_defined_values.get("user_role") + db_role = user_info.user_role if user_info else None + + if _should_use_role_from_sso_response(sso_role): + # SSO provided a valid role, keep it and log that we're using it + verbose_proxy_logger.info( + f"Using SSO role: {sso_role} (DB role was: {db_role})" + ) + else: + # SSO didn't provide a valid role, fall back to DB role or default + if user_info is None or user_info.user_role is None: + user_defined_values["user_role"] = ( + LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value + ) + verbose_proxy_logger.debug( + "No SSO or DB role found, using default: INTERNAL_USER_VIEW_ONLY" + ) + else: + user_defined_values["user_role"] = user_info.user_role + verbose_proxy_logger.debug(f"Using DB role: {user_info.user_role}") + + # Preserve the user's existing models from the database + if user_info is not None and hasattr(user_info, "models") and user_info.models: + user_defined_values["models"] = user_info.models + + return user_defined_values + + +async def check_and_update_if_proxy_admin_id( + user_role: str, user_id: str, prisma_client: Optional[PrismaClient] +): + """ + - Check if user role in DB is admin + - If not, update user role in DB to admin role + """ + proxy_admin_id = os.getenv("PROXY_ADMIN_ID") + if proxy_admin_id is not None and proxy_admin_id == user_id: + if user_role and user_role == LitellmUserRoles.PROXY_ADMIN.value: + return user_role + + if prisma_client: + await UserRepository(prisma_client).table.update( + where={"user_id": user_id}, + data={"user_role": LitellmUserRoles.PROXY_ADMIN.value}, + ) + + user_role = LitellmUserRoles.PROXY_ADMIN.value + + return user_role + + +@router.get("/sso/callback", tags=["experimental"], include_in_schema=False) +async def auth_callback(request: Request, state: Optional[str] = None): + """Verify login""" + verbose_proxy_logger.info(f"Starting SSO callback with state: {state}") + + # Check if this is a CLI login (state starts with our CLI prefix) + from litellm.constants import LITELLM_CLI_SESSION_TOKEN_PREFIX + from litellm.proxy._types import LiteLLM_JWTAuth + from litellm.proxy.auth.handle_jwt import JWTHandler + from litellm.proxy.proxy_server import ( + general_settings, + jwt_handler, + master_key, + prisma_client, + user_api_key_cache, + ) + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + sso_jwt_handler: Optional[JWTHandler] = None + ui_access_mode = general_settings.get("ui_access_mode", None) + if ui_access_mode is not None and isinstance(ui_access_mode, dict): + sso_jwt_handler = JWTHandler() + sso_jwt_handler.update_environment( + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + litellm_jwtauth=LiteLLM_JWTAuth( + team_ids_jwt_field=general_settings.get("ui_access_mode", {}).get( + "sso_group_jwt_field", None + ), + ), + leeway=0, + ) + + microsoft_client_id = os.getenv("MICROSOFT_CLIENT_ID", None) + google_client_id = os.getenv("GOOGLE_CLIENT_ID", None) + generic_client_id = os.getenv("GENERIC_CLIENT_ID", None) + received_response: Optional[dict] = None + access_token_payload: Optional[dict] = None + # get url from request + if master_key is None: + raise ProxyException( + 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.", + type=ProxyErrorTypes.auth_error, + param="master_key", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + redirect_url = SSOAuthenticationHandler.get_redirect_url_for_sso( + request=request, sso_callback_route="sso/callback" + ) + + verbose_proxy_logger.info(f"Redirecting to {redirect_url}") + result = None + if google_client_id is not None: + result = await GoogleSSOHandler.get_google_callback_response( + request=request, + google_client_id=google_client_id, + redirect_url=redirect_url, + ) + elif microsoft_client_id is not None: + result = await MicrosoftSSOHandler.get_microsoft_callback_response( + request=request, + microsoft_client_id=microsoft_client_id, + redirect_url=redirect_url, + ) + + elif generic_client_id is not None: + ( + result, + received_response, + access_token_payload, + ) = await get_generic_sso_response( + request=request, + jwt_handler=jwt_handler, + generic_client_id=generic_client_id, + redirect_url=redirect_url, + sso_jwt_handler=sso_jwt_handler, + ) + + if result is None: + raise HTTPException( + status_code=401, + detail="Result not returned by SSO provider.", + ) + + if state and state.startswith(f"{LITELLM_CLI_SESSION_TOKEN_PREFIX}:"): + # State format: {PREFIX}:{login_id}[:{user_code}] + state_parts = state.split(":", 2) + key_id = state_parts[1] if len(state_parts) > 1 else None + prefill_user_code = state_parts[2] if len(state_parts) > 2 else None + + verbose_proxy_logger.info("CLI SSO callback detected") + return await cli_sso_callback( + request=request, + key=key_id, + prefill_user_code=prefill_user_code, + result=result, + received_response=received_response, + ) + + # Control-plane cross-origin: read return_to from cookie. + # Starlette's cookie_parser already handles RFC 2109 unquoting. + cp_return_to: Optional[str] = request.cookies.get("litellm_cp_return_to") + + return await SSOAuthenticationHandler.get_redirect_response_from_openid( + result=result, + request=request, + received_response=received_response, + generic_client_id=generic_client_id, + ui_access_mode=ui_access_mode, + access_token_payload=access_token_payload, + jwt_handler=jwt_handler, + return_to=cp_return_to, + ) + + +async def _build_cli_sso_user_defined_values( + result: Union[OpenID, dict], + parsed_openid_result: ParsedOpenIDResult, +) -> Optional[SSOUserDefinedValues]: + from litellm.proxy.proxy_server import user_custom_sso + + user_id = parsed_openid_result.get("user_id") + if user_custom_sso is not None: + if inspect.iscoroutinefunction(user_custom_sso): + return await user_custom_sso(result) # type: ignore + raise ValueError("user_custom_sso must be a coroutine function") + if user_id is None: + return None + return SSOUserDefinedValues( + models=[], + user_id=user_id, + user_email=parsed_openid_result.get("user_email"), + max_budget=litellm.max_internal_user_budget, + user_role=parsed_openid_result.get("user_role"), + budget_duration=litellm.internal_user_budget_duration, + ) + + +async def _fetch_cli_sso_team_details( + prisma_client: PrismaClient, + teams: List[str], +) -> List[Dict[str, Any]]: + team_details: List[Dict[str, Any]] = [] + try: + if teams: + prisma_teams = await TeamRepository(prisma_client).table.find_many( + where={"team_id": {"in": teams}} + ) + for team_row in prisma_teams: + team_dict = team_row.model_dump() + team_details.append( + { + "team_id": team_dict.get("team_id"), + "team_alias": team_dict.get("team_alias"), + } + ) + except Exception as e: + verbose_proxy_logger.error( + f"Error fetching team details for CLI SSO session: {e}" + ) + return team_details + + +async def _complete_cli_sso_callback_session( + *, + request: Request, + key: str, + flow: dict, + result: Union[OpenID, dict], + parsed_openid_result: ParsedOpenIDResult, + user_defined_values: Optional[SSOUserDefinedValues], + prisma_client: PrismaClient, + user_api_key_cache: UserApiKeyCache, + proxy_logging_obj: ProxyLogging, + prefill_user_code: str | None = None, +): + from fastapi.responses import HTMLResponse + + user_id = parsed_openid_result.get("user_id") + user_email = parsed_openid_result.get("user_email") + user_info = await get_user_info_from_db( + result=result, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=proxy_logging_obj, + user_email=user_email, + user_defined_values=user_defined_values, + alternate_user_id=user_id, + ) + if user_info is None: + raise HTTPException( + status_code=500, detail="Failed to retrieve user information from SSO" + ) + if not user_info.user_id: + raise HTTPException( + status_code=500, detail="Failed to retrieve user information from SSO" + ) + + teams: List[str] = [] + if hasattr(user_info, "teams") and user_info.teams: + teams = user_info.teams if isinstance(user_info.teams, list) else [] + + team_details = await _fetch_cli_sso_team_details( + prisma_client=prisma_client, teams=teams + ) + attribution_metadata = build_cli_sso_attribution_metadata(result=result) + if attribution_metadata: + await _persist_cli_sso_user_metadata( + prisma_client=prisma_client, + user_id=cast(str, user_info.user_id), + attribution_metadata=attribution_metadata, + ) + + flow["session_data"] = { + "user_id": cast(str, user_info.user_id), + "user_role": user_info.user_role, + "models": user_info.models if hasattr(user_info, "models") else [], + "user_email": user_email, + "teams": teams, + "team_details": team_details, + "attribution_metadata": attribution_metadata, + } + flow["sso_complete"] = True + browser_complete_token = secrets.token_urlsafe(32) + flow["browser_complete_token_hash"] = _hash_cli_sso_secret(browser_complete_token) + _set_cli_sso_flow(login_id=key, cache=user_api_key_cache, flow=flow) + + verbose_proxy_logger.info( + f"Stored CLI SSO session for user: {user_info.user_id}, teams: {teams}, num_teams: {len(teams)}" + ) + verify_url = get_custom_url( + request_base_url=str(request.base_url), + route=f"sso/cli/complete/{key}", + ) + return HTMLResponse( + content=_render_cli_sso_verification_page( + verify_url=verify_url, + browser_complete_token=browser_complete_token, + prefill_user_code=prefill_user_code, + ), + status_code=200, + ) + + +async def cli_sso_callback( + request: Request, + key: Optional[str] = None, + result: Optional[Union[OpenID, dict]] = None, + received_response: Optional[dict] = None, + prefill_user_code: str | None = None, +): + """CLI SSO callback - stores session info for JWT generation on polling""" + verbose_proxy_logger.info("CLI SSO callback") + + from litellm.proxy.proxy_server import ( + general_settings, + prisma_client, + proxy_logging_obj, + user_api_key_cache, + ) + + flow = _get_cli_sso_flow_or_raise(login_id=key, cache=user_api_key_cache) + + if prisma_client is None: + raise HTTPException( + status_code=500, detail=CommonProxyErrors.db_not_connected_error.value + ) + + if result is None: + raise HTTPException( + status_code=500, + detail="SSO authentication failed - no result returned from provider", + ) + + # After None check, cast to non-None type for type checker + result_non_none: Union[OpenID, dict] = cast(Union[OpenID, dict], result) + + try: + parsed_openid_result = ( + SSOAuthenticationHandler._get_user_email_and_id_from_result( + result=result_non_none, + generic_client_id=os.getenv("GENERIC_CLIENT_ID", None), + ) + ) + verbose_proxy_logger.debug(f"parsed_openid_result: {parsed_openid_result}") + user_defined_values = await _build_cli_sso_user_defined_values( + result=result_non_none, + parsed_openid_result=parsed_openid_result, + ) + + SSOAuthenticationHandler.verify_user_in_restricted_sso_group( + general_settings=general_settings, + result=result_non_none, + received_response=received_response, + ) + + return await _complete_cli_sso_callback_session( + request=request, + key=cast(str, key), + flow=flow, + result=result_non_none, + parsed_openid_result=parsed_openid_result, + user_defined_values=user_defined_values, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=proxy_logging_obj, + prefill_user_code=prefill_user_code, + ) + except ProxyException: + raise + except HTTPException: + raise + except Exception as e: + verbose_proxy_logger.error(f"Error with CLI SSO callback: {e}") + raise HTTPException( + status_code=500, detail=f"Failed to process CLI SSO: {str(e)}" + ) + + +@router.get("/sso/cli/poll/{key_id}", tags=["experimental"], include_in_schema=False) +async def cli_poll_key( + key_id: str, + team_id: Optional[str] = None, + x_litellm_cli_poll_secret: Optional[str] = Header(default=None), +): + """ + CLI polling endpoint - retrieves session from cache and generates JWT. + + Flow: + 1. First poll (no team_id): Returns teams list without generating JWT + 2. Second poll (with team_id): Generates JWT with selected team and deletes session + + Args: + key_id: The CLI login session ID + team_id: Optional team ID to assign to the JWT. If provided, must be one of user's teams. + """ + from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken + from litellm.proxy.proxy_server import user_api_key_cache + + try: + flow = _get_cli_sso_flow_or_raise(login_id=key_id, cache=user_api_key_cache) + if not _verify_cli_sso_poll_secret( + flow=flow, poll_secret=x_litellm_cli_poll_secret + ): + raise HTTPException(status_code=403, detail="Invalid CLI polling secret") + + if not flow.get("sso_complete") or not flow.get("user_code_verified"): + return {"status": "pending"} + + session_data = flow.get("session_data") + + if isinstance(session_data, dict): + user_teams = session_data.get("teams", []) + user_team_details = session_data.get("team_details") + user_id = session_data["user_id"] + + verbose_proxy_logger.info( + f"CLI poll: user={user_id}, team_id={team_id}, user_teams={user_teams}, num_teams={len(user_teams)}" + ) + + # If no team_id provided and user has teams, return teams list for selection + # Don't generate JWT yet - let CLI select a team first. For newer + # clients we return rich team details (id + alias); older clients + # can continue to rely on the simple "teams" list. + if team_id is None and len(user_teams) > 1: + verbose_proxy_logger.info( + f"Returning teams list for user {user_id} to select from: {user_teams}" + ) + # Best-effort construction of team_details if it wasn't + # already cached for some reason. + team_details_response: Optional[List[Dict[str, Any]]] = None + if isinstance(user_team_details, list) and user_team_details: + team_details_response = user_team_details + elif user_teams: + team_details_response = [ + {"team_id": t, "team_alias": None} for t in user_teams + ] + poll_response: Dict[str, Any] = { + "status": "ready", + "user_id": user_id, + "teams": user_teams, + "team_details": team_details_response, + "requires_team_selection": True, + } + attribution_metadata = _cli_poll_attribution_metadata_from_session( + session_data + ) + if attribution_metadata: + poll_response["attribution_metadata"] = attribution_metadata + return poll_response + + # Validate team_id if provided + if team_id is not None: + if team_id not in user_teams: + raise HTTPException( + status_code=403, + detail=f"User does not belong to team: {team_id}. Available teams: {user_teams}", + ) + else: + # If no team_id provided and user has 0 or 1 team, use first team (or None) + team_id = user_teams[0] if len(user_teams) > 0 else None + + team_alias = None + if team_id and isinstance(user_team_details, list): + team_alias = next( + ( + team.get("team_alias") + for team in user_team_details + if team.get("team_id") == team_id + ), + None, + ) + + # Create user object for JWT generation + user_info = LiteLLM_UserTable( + user_id=user_id, + user_role=session_data["user_role"], + models=session_data.get("models", []), + max_budget=litellm.max_ui_session_budget, + ) + + # Generate CLI JWT on-demand (expiration configurable via LITELLM_CLI_JWT_EXPIRATION_HOURS) + # Pass selected team_id to ensure JWT has correct team + jwt_token = ExperimentalUIJWTToken.get_cli_jwt_auth_token( + user_info=user_info, team_id=team_id, team_alias=team_alias + ) + + # Delete cache entry (single-use) + user_api_key_cache.delete_cache(key=_get_cli_sso_flow_cache_key(key_id)) + + verbose_proxy_logger.info( + f"CLI JWT generated for user: {user_id}, team: {team_id}" + ) + poll_response = { + "status": "ready", + "key": jwt_token, + "user_id": user_id, + "team_id": team_id, + "teams": user_teams, + # Echo back any team details we have so clients can + # present nicer information if needed. + "team_details": user_team_details, + } + attribution_metadata = _cli_poll_attribution_metadata_from_session( + session_data + ) + if attribution_metadata: + poll_response["attribution_metadata"] = attribution_metadata + return poll_response + else: + return {"status": "pending"} + + except HTTPException: + raise + except Exception as e: + verbose_proxy_logger.error(f"Error polling for CLI JWT: {e}") + raise HTTPException( + status_code=500, detail=f"Error checking session status: {str(e)}" + ) + + +async def insert_sso_user( + result_openid: Optional[Union[OpenID, dict]], + user_defined_values: Optional[SSOUserDefinedValues] = None, +) -> NewUserResponse: + """ + Helper function to create a New User in LiteLLM DB after a successful SSO login + + Args: + result_openid (OpenID): User information in OpenID format if the login was successful. + user_defined_values (Optional[SSOUserDefinedValues], optional): LiteLLM SSOValues / fields that were read + + Returns: + Tuple[str, str]: User ID and User Role + """ + verbose_proxy_logger.debug( + f"Inserting SSO user into DB. User values: {user_defined_values}" + ) + if result_openid is None: + raise ValueError("result_openid is None") + if isinstance(result_openid, dict): + result_openid = OpenID(**result_openid) + + if user_defined_values is None: + raise ValueError("user_defined_values is None") + + # Apply default_internal_user_params + if litellm.default_internal_user_params: + # Preserve the SSO-extracted role if it's a valid LiteLLM role, + # regardless of how it was determined (role_mappings, Microsoft app_roles, + # GENERIC_USER_ROLE_ATTRIBUTE, custom SSO handler, etc.) + sso_role = user_defined_values.get("user_role") + if _should_use_role_from_sso_response(sso_role): + # Preserve the SSO-extracted role, but apply other defaults + preserved_role = sso_role + user_defined_values.update(litellm.default_internal_user_params) # type: ignore + user_defined_values["user_role"] = preserved_role # Restore preserved role + verbose_proxy_logger.debug( + f"Preserved SSO-extracted role '{preserved_role}'" + ) + else: + # SSO didn't provide a valid role, apply all defaults including role + user_defined_values.update(litellm.default_internal_user_params) # type: ignore + + # Set budget for internal users + if user_defined_values.get("user_role") == LitellmUserRoles.INTERNAL_USER.value: + if user_defined_values.get("max_budget") is None: + user_defined_values["max_budget"] = litellm.max_internal_user_budget + if user_defined_values.get("budget_duration") is None: + user_defined_values["budget_duration"] = ( + litellm.internal_user_budget_duration + ) + + if user_defined_values["user_role"] is None: + user_defined_values["user_role"] = LitellmUserRoles.INTERNAL_USER_VIEW_ONLY + + new_user_request = NewUserRequest( + user_id=user_defined_values["user_id"], + user_email=normalize_email(user_defined_values["user_email"]), + user_role=user_defined_values["user_role"], # type: ignore + max_budget=user_defined_values["max_budget"], + budget_duration=user_defined_values["budget_duration"], + sso_user_id=user_defined_values["user_id"], + auto_create_key=False, + ) + + if result_openid and hasattr(result_openid, "provider"): + new_user_request.metadata = { + "auth_provider": getattr(result_openid, "provider") + } + + response = await new_user( + data=new_user_request, + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN), + ) + + return response + + +@router.get( + "/sso/get/ui_settings", + tags=["experimental"], + include_in_schema=False, + dependencies=[Depends(user_api_key_auth)], +) +async def get_ui_settings(request: Request): + from litellm.proxy.proxy_server import general_settings, proxy_state + + _proxy_base_url = os.getenv("PROXY_BASE_URL", None) + _logout_url = os.getenv("PROXY_LOGOUT_URL", None) + _api_doc_base_url = os.getenv("LITELLM_UI_API_DOC_BASE_URL", None) + _is_sso_enabled = _has_user_setup_sso() + disable_expensive_db_queries = ( + proxy_state.get_proxy_state_variable("spend_logs_row_count") + > MAX_SPENDLOG_ROWS_TO_QUERY + ) + default_team_disabled = general_settings.get("default_team_disabled", False) + if "PROXY_DEFAULT_TEAM_DISABLED" in os.environ: + if os.environ["PROXY_DEFAULT_TEAM_DISABLED"].lower() == "true": + default_team_disabled = True + + return { + "PROXY_BASE_URL": _proxy_base_url, + "PROXY_LOGOUT_URL": _logout_url, + "LITELLM_UI_API_DOC_BASE_URL": _api_doc_base_url, + "DEFAULT_TEAM_DISABLED": default_team_disabled, + "SSO_ENABLED": _is_sso_enabled, + "NUM_SPEND_LOGS_ROWS": proxy_state.get_proxy_state_variable( + "spend_logs_row_count" + ), + "DISABLE_EXPENSIVE_DB_QUERIES": disable_expensive_db_queries, + } + + +@router.get( + "/sso/readiness", + tags=["experimental"], + dependencies=[Depends(user_api_key_auth)], +) +async def sso_readiness(): + """ + Health endpoint for checking SSO readiness. + Checks if the configured SSO provider has all required environment variables set in memory. + """ + microsoft_client_id = os.getenv("MICROSOFT_CLIENT_ID", None) + google_client_id = os.getenv("GOOGLE_CLIENT_ID", None) + generic_client_id = os.getenv("GENERIC_CLIENT_ID", None) + + # Determine which SSO provider is configured + configured_provider = None + if google_client_id is not None: + configured_provider = "google" + elif microsoft_client_id is not None: + configured_provider = "microsoft" + elif generic_client_id is not None: + configured_provider = "generic" + + # If no SSO is configured, return healthy (SSO is optional) + if configured_provider is None: + return { + "status": "healthy", + "sso_configured": False, + "message": "No SSO provider configured", + } + + # Check required environment variables for the configured provider + missing_vars = [] + + if configured_provider == "google": + google_client_secret = os.getenv("GOOGLE_CLIENT_SECRET", None) + if google_client_secret is None: + missing_vars.append("GOOGLE_CLIENT_SECRET") + + elif configured_provider == "microsoft": + microsoft_client_secret = os.getenv("MICROSOFT_CLIENT_SECRET", None) + microsoft_tenant = os.getenv("MICROSOFT_TENANT", None) + if microsoft_client_secret is None: + missing_vars.append("MICROSOFT_CLIENT_SECRET") + if microsoft_tenant is None: + missing_vars.append("MICROSOFT_TENANT") + + elif configured_provider == "generic": + generic_client_secret = os.getenv("GENERIC_CLIENT_SECRET", None) + generic_authorization_endpoint = os.getenv( + "GENERIC_AUTHORIZATION_ENDPOINT", None + ) + generic_token_endpoint = os.getenv("GENERIC_TOKEN_ENDPOINT", None) + generic_userinfo_endpoint = os.getenv("GENERIC_USERINFO_ENDPOINT", None) + if generic_client_secret is None: + missing_vars.append("GENERIC_CLIENT_SECRET") + if generic_authorization_endpoint is None: + missing_vars.append("GENERIC_AUTHORIZATION_ENDPOINT") + if generic_token_endpoint is None: + missing_vars.append("GENERIC_TOKEN_ENDPOINT") + if generic_userinfo_endpoint is None: + missing_vars.append("GENERIC_USERINFO_ENDPOINT") + + # If all required variables are present, return healthy + if len(missing_vars) == 0: + return { + "status": "healthy", + "sso_configured": True, + "provider": configured_provider, + "message": f"{configured_provider.capitalize()} SSO is properly configured", + } + + # If some variables are missing, return unhealthy + raise HTTPException( + status_code=503, + detail={ + "status": "unhealthy", + "sso_configured": True, + "provider": configured_provider, + "missing_environment_variables": missing_vars, + "message": f"{configured_provider.capitalize()} SSO is configured but missing required environment variables: {', '.join(missing_vars)}", + }, + ) + + +class SSOAuthenticationHandler: + """ + Handler for SSO Authentication across all SSO providers + """ + + @staticmethod + def _validate_return_to(return_to: str) -> bool: + """ + Validate that return_to matches the configured control_plane_url origin. + + Returns True if return_to is valid and should be used. + Returns False if control_plane_url is not configured (return_to is ignored). + Raises HTTPException(400) if return_to origin does not match control_plane_url origin. + """ + from litellm.proxy.proxy_server import general_settings + + control_plane_url = general_settings.get("control_plane_url") + if control_plane_url is None: + return False + + def _origin(url: str) -> tuple: + parsed = urlparse(url) + scheme = (parsed.scheme or "").lower() + hostname = (parsed.hostname or "").lower() + default_port = 443 if scheme == "https" else 80 + port = parsed.port if parsed.port is not None else default_port + return (scheme, hostname, port) + + if _origin(return_to) != _origin(control_plane_url): + raise HTTPException( + status_code=400, + detail="return_to does not match the configured control_plane_url", + ) + + return True + + @staticmethod + async def get_sso_login_redirect( + redirect_url: str, + google_client_id: Optional[str] = None, + microsoft_client_id: Optional[str] = None, + generic_client_id: Optional[str] = None, + state: Optional[str] = None, + request: Optional[Request] = None, + ) -> Optional[RedirectResponse]: + """ + Step 1. Call Get Login Redirect for the SSO provider. Send the redirect response to `redirect_url` + + Args: + redirect_url (str): The URL to redirect the user to after login + google_client_id (Optional[str], optional): The Google Client ID. Defaults to None. + microsoft_client_id (Optional[str], optional): The Microsoft Client ID. Defaults to None. + generic_client_id (Optional[str], optional): The Generic Client ID. Defaults to None. + request: Optional FastAPI request, used to drive the ``Secure`` + attribute on the ``litellm_oauth_state`` CSRF cookie. + + Returns: + RedirectResponse: The redirect response from the SSO provider. + """ + # Google SSO Auth + if google_client_id is not None: + from fastapi_sso.sso.google import GoogleSSO + + google_client_secret = os.getenv("GOOGLE_CLIENT_SECRET", None) + if google_client_secret is None: + raise ProxyException( + message="GOOGLE_CLIENT_SECRET not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GOOGLE_CLIENT_SECRET", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + google_sso = GoogleSSO( + client_id=google_client_id, + client_secret=google_client_secret, + redirect_uri=redirect_url, + ) + verbose_proxy_logger.info( + f"In /google-login/key/generate, \nGOOGLE_REDIRECT_URI: {redirect_url}\nGOOGLE_CLIENT_ID: {google_client_id}" + ) + with google_sso: + return await google_sso.get_login_redirect(state=state) + # Microsoft SSO Auth + elif microsoft_client_id is not None: + microsoft_client_secret = os.getenv("MICROSOFT_CLIENT_SECRET", None) + microsoft_tenant = os.getenv("MICROSOFT_TENANT", None) + if microsoft_client_secret is None: + raise ProxyException( + message="MICROSOFT_CLIENT_SECRET not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="MICROSOFT_CLIENT_SECRET", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + microsoft_sso = CustomMicrosoftSSO( + client_id=microsoft_client_id, + client_secret=microsoft_client_secret, + tenant=microsoft_tenant, + redirect_uri=redirect_url, + allow_insecure_http=True, + ) + with microsoft_sso: + return await microsoft_sso.get_login_redirect(state=state) + elif generic_client_id is not None: + from fastapi_sso.sso.base import DiscoveryDocument + from fastapi_sso.sso.generic import create_provider + + generic_client_secret = os.getenv("GENERIC_CLIENT_SECRET", None) + generic_scope = os.getenv("GENERIC_SCOPE", "openid email profile").split( + " " + ) + generic_authorization_endpoint = os.getenv( + "GENERIC_AUTHORIZATION_ENDPOINT", None + ) + generic_token_endpoint = os.getenv("GENERIC_TOKEN_ENDPOINT", None) + generic_userinfo_endpoint = os.getenv("GENERIC_USERINFO_ENDPOINT", None) + if generic_client_secret is None: + raise ProxyException( + message="GENERIC_CLIENT_SECRET not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_CLIENT_SECRET", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if generic_authorization_endpoint is None: + raise ProxyException( + message="GENERIC_AUTHORIZATION_ENDPOINT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_AUTHORIZATION_ENDPOINT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if generic_token_endpoint is None: + raise ProxyException( + message="GENERIC_TOKEN_ENDPOINT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_TOKEN_ENDPOINT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if generic_userinfo_endpoint is None: + raise ProxyException( + message="GENERIC_USERINFO_ENDPOINT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GENERIC_USERINFO_ENDPOINT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + verbose_proxy_logger.debug( + f"authorization_endpoint: {generic_authorization_endpoint}\ntoken_endpoint: {generic_token_endpoint}\nuserinfo_endpoint: {generic_userinfo_endpoint}" + ) + verbose_proxy_logger.debug( + f"GENERIC_REDIRECT_URI: {redirect_url}\nGENERIC_CLIENT_ID: {generic_client_id}\n" + ) + discovery = DiscoveryDocument( + authorization_endpoint=generic_authorization_endpoint, + token_endpoint=generic_token_endpoint, + userinfo_endpoint=generic_userinfo_endpoint, + ) + SSOProvider = create_provider(name="oidc", discovery_document=discovery) + generic_sso = SSOProvider( + client_id=generic_client_id, + client_secret=generic_client_secret, + redirect_uri=redirect_url, + allow_insecure_http=True, + scope=generic_scope, + ) + return await SSOAuthenticationHandler.get_generic_sso_redirect_response( + generic_sso=generic_sso, + state=state, + generic_authorization_endpoint=generic_authorization_endpoint, + request=request, + ) + raise ValueError( + "Unknown SSO provider. Please setup SSO with client IDs https://docs.litellm.ai/docs/proxy/admin_ui_sso" + ) + + @staticmethod + async def get_generic_sso_redirect_response( + generic_sso: Any, + state: Optional[str] = None, + generic_authorization_endpoint: Optional[str] = None, + request: Optional[Request] = None, + ) -> Optional[RedirectResponse]: + """ + Get the redirect response for Generic SSO + """ + from urllib.parse import parse_qs, urlencode, urlparse, urlunparse + + from litellm.proxy.proxy_server import redis_usage_cache, user_api_key_cache + + with generic_sso: + # State is bound to the caller's browser via a ``litellm_oauth_state`` + # HttpOnly cookie set on the redirect response below; the SSO + # callback validates the URL ``state`` against that cookie before + # completing the PKCE token exchange. Without this binding, an + # attacker who pre-mints a state + a cached PKCE verifier can hand + # the link to a victim and capture the resulting access token + # (Login CSRF / token theft). + ( + redirect_params, + code_verifier, + ) = SSOAuthenticationHandler._get_generic_sso_redirect_params( + state=state, + generic_authorization_endpoint=generic_authorization_endpoint, + ) + + # Separate PKCE params from state params (fastapi-sso doesn't accept code_challenge) + pkce_params = {} + state_only_params = {} + for key, value in redirect_params.items(): + if key in ("code_challenge", "code_challenge_method"): + pkce_params[key] = value + else: + state_only_params[key] = value + + # Get the redirect response from fastapi-sso with only state param + redirect_response = await generic_sso.get_login_redirect(**state_only_params) # type: ignore + + # If PKCE is enabled, add PKCE parameters to the redirect URL + if code_verifier and "state" in redirect_params: + # Store code_verifier in cache (10 min TTL). Wrap in dict for proper + # JSON serialization in Redis. Use Redis when available so callbacks + # landing on another pod can retrieve it (multi-pod SSO). + cache_key = f"pkce_verifier:{redirect_params['state']}" + if redis_usage_cache is not None: + await redis_usage_cache.async_set_cache( + key=cache_key, + value={"code_verifier": code_verifier}, + ttl=600, + ) + else: + await user_api_key_cache.async_set_cache( + key=cache_key, + value={"code_verifier": code_verifier}, + ttl=600, + ) + verbose_proxy_logger.debug( + "PKCE code_verifier stored in cache (TTL: 600s)" + ) + + # Add PKCE parameters to the authorization URL + if pkce_params: + parsed_url = urlparse(str(redirect_response.headers["location"])) + query_params = parse_qs(parsed_url.query) + + # Add PKCE parameters + for key, value in pkce_params.items(): + query_params[key] = [value] + + # Reconstruct the URL with PKCE parameters + new_query = urlencode(query_params, doseq=True) + new_url = urlunparse( + ( + parsed_url.scheme, + parsed_url.netloc, + parsed_url.path, + parsed_url.params, + new_query, + parsed_url.fragment, + ) + ) + + # Update the redirect response + redirect_response.headers["location"] = new_url + + # Bind state to the user's browser session. The /callback + # handler validates the URL ``state`` against this cookie via + # ``secrets.compare_digest`` before exchanging the PKCE + # code_verifier. Only set the cookie when PKCE is in use + # (i.e. inside this ``code_verifier`` branch) so two + # concurrent SSO sessions — one PKCE, one plain — cannot + # overwrite each other's state cookie. + state_value = redirect_params.get("state") + if state_value and redirect_response is not None: + # Production-safe default: require HTTPS for the + # CSRF-protection cookie unless we can prove the + # incoming request is HTTP (local dev). Without + # ``Secure`` the cookie is sent over plain HTTP, + # letting a network observer read and replay the + # state value and bypass this protection. + secure_flag = request is None or request.url.scheme == "https" + redirect_response.set_cookie( + key="litellm_oauth_state", + value=state_value, + max_age=600, + httponly=True, + samesite="lax", + secure=secure_flag, + ) + return redirect_response + + @staticmethod + def _get_generic_sso_redirect_params( + state: Optional[str] = None, + generic_authorization_endpoint: Optional[str] = None, + ) -> Tuple[dict, Optional[str]]: + """ + Get redirect parameters for Generic SSO with proper state priority handling. + Optionally generates PKCE parameters if GENERIC_CLIENT_USE_PKCE is enabled. + + Priority order: + 1. CLI state (if provided) + 2. GENERIC_CLIENT_STATE environment variable + 3. Generated UUID (required by Okta and most OAuth providers) + + + Args: + state: Optional state parameter (e.g., CLI state) + generic_authorization_endpoint: Authorization endpoint URL + + Returns: + Tuple[dict, Optional[str]]: + - Redirect parameters for SSO login (may include PKCE params) + - code_verifier (if PKCE is enabled, None otherwise) + """ + redirect_params = {} + code_verifier: Optional[str] = None + + if state: + # CLI state takes priority + # the litellm proxy cli sends the "state" parameter to the proxy server for auth. We should maintain the state parameter for the cli if it is provided + redirect_params["state"] = state + else: + generic_client_state = os.getenv("GENERIC_CLIENT_STATE", None) + if generic_client_state: + redirect_params["state"] = generic_client_state + else: + redirect_params["state"] = uuid.uuid4().hex + + # Handle PKCE (Proof Key for Code Exchange) if enabled + # Set GENERIC_CLIENT_USE_PKCE=true to enable PKCE for enhanced OAuth security + use_pkce = os.getenv("GENERIC_CLIENT_USE_PKCE", "false").lower() == "true" + + if use_pkce: + ( + code_verifier, + code_challenge, + ) = SSOAuthenticationHandler.generate_pkce_params() + redirect_params["code_challenge"] = code_challenge + redirect_params["code_challenge_method"] = "S256" + verbose_proxy_logger.debug("PKCE enabled for authorization request") + + return redirect_params, code_verifier + + @staticmethod + def should_use_sso_handler( + google_client_id: Optional[str] = None, + microsoft_client_id: Optional[str] = None, + generic_client_id: Optional[str] = None, + ) -> bool: + if ( + google_client_id is not None + or microsoft_client_id is not None + or generic_client_id is not None + ): + return True + return False + + @staticmethod + def get_redirect_url_for_sso( + request: Request, + sso_callback_route: str, + existing_key: Optional[str] = None, + ) -> str: + """ + Get the redirect URL for SSO + + Note: existing_key is not added to the URL to avoid changing the callback URL. + It should be passed via the state parameter instead. + """ + from litellm.proxy.utils import get_custom_url + + redirect_url = get_custom_url(request_base_url=str(request.base_url)) + if redirect_url.endswith("/"): + redirect_url += sso_callback_route + else: + redirect_url += "/" + sso_callback_route + + return redirect_url + + @staticmethod + async def upsert_sso_user( + result: Optional[Union[CustomOpenID, OpenID, dict]], + user_info: Optional[Union[NewUserResponse, LiteLLM_UserTable]], + user_email: Optional[str], + user_defined_values: Optional[SSOUserDefinedValues], + prisma_client: PrismaClient, + ): + """ + Connects the SSO Users to the User Table in LiteLLM DB + + - If user on LiteLLM DB, update the user_email and user_role (if SSO provides valid role) with the SSO values + - If user not on LiteLLM DB, insert the user into LiteLLM DB + """ + try: + if user_info is not None: + user_id = user_info.user_id + update_data = _build_sso_user_update_data( + result=result, + user_email=user_email, + user_id=user_id, + ) + + await UserRepository(prisma_client).table.update_many( + where={"user_id": user_id}, data=update_data + ) + else: + verbose_proxy_logger.info( + "user not in DB, inserting user into LiteLLM DB" + ) + # user not in DB, insert User into LiteLLM DB + user_info = await insert_sso_user( + result_openid=result, + user_defined_values=user_defined_values, + ) + return user_info + except Exception as e: + verbose_proxy_logger.exception( + f"Error upserting SSO user into LiteLLM DB: {e}" + ) + return user_info + + @staticmethod + async def add_user_to_teams_from_sso_response( + result: Optional[Union[CustomOpenID, OpenID, dict]], + user_info: Optional[Union[NewUserResponse, LiteLLM_UserTable]], + ): + """ + Adds the user as a team member to the teams specified in the SSO responses `team_ids` field + + + The `team_ids` field is populated by litellm after processing the SSO response + """ + if user_info is None: + verbose_proxy_logger.debug( + "User not found in LiteLLM DB, skipping team member addition" + ) + return + sso_teams = getattr(result, "team_ids", []) + await add_missing_team_member(user_info=user_info, sso_teams=sso_teams) + + @staticmethod + def verify_user_in_restricted_sso_group( + general_settings: Dict, + result: Optional[Union[CustomOpenID, OpenID, dict]], + received_response: Optional[dict], + ) -> Literal[True]: + """ + when ui_access_mode.type == "restricted_sso_group": + + - result.team_ids should contain the restricted_sso_group + - if not, raise a ProxyException + - if so, return True + - if result.team_ids is None, return False + - if result.team_ids is an empty list, return False + - if result.team_ids is a list, return True if the restricted_sso_group is in the list, otherwise return False + """ + + ui_access_mode = cast( + Optional[Union[Dict, str]], general_settings.get("ui_access_mode") + ) + + if ui_access_mode is None: + return True + if isinstance(ui_access_mode, str): + return True + team_ids = getattr(result, "team_ids", []) + + if ui_access_mode.get("type") == "restricted_sso_group": + restricted_sso_group = ui_access_mode.get("restricted_sso_group") + if restricted_sso_group not in team_ids: + raise ProxyException( + message=f"User is not in the restricted SSO group: {restricted_sso_group}. User groups: {team_ids}. Received SSO response: {received_response}", + type=ProxyErrorTypes.auth_error, + param="restricted_sso_group", + code=status.HTTP_403_FORBIDDEN, + ) + return True + + @staticmethod + async def create_litellm_team_from_sso_group( + litellm_team_id: str, + litellm_team_name: Optional[str] = None, + ): + """ + Creates a Litellm Team from a SSO Group ID + + Your SSO provider might have groups that should be created on LiteLLM + + Use this helper to create a Litellm Team from a SSO Group ID + + Args: + litellm_team_id (str): The ID of the Litellm Team + litellm_team_name (Optional[str]): The name of the Litellm Team + """ + from litellm.proxy.proxy_server import prisma_client + + if prisma_client is None: + raise ProxyException( + message="Prisma client not found. Set it in the proxy_server.py file", + type=ProxyErrorTypes.auth_error, + param="prisma_client", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + try: + team_obj = await TeamRepository(prisma_client).table.find_first( + where={"team_id": litellm_team_id} + ) + verbose_proxy_logger.debug(f"Team object: {team_obj}") + + # only create a new team if it doesn't exist + if team_obj: + verbose_proxy_logger.debug( + f"Team already exists: {litellm_team_id} - {litellm_team_name}" + ) + return + + team_request: NewTeamRequest = NewTeamRequest( + team_id=litellm_team_id, + team_alias=litellm_team_name, + ) + if litellm.default_team_params: + team_request = SSOAuthenticationHandler._cast_and_deepcopy_litellm_default_team_params( + default_team_params=litellm.default_team_params, + litellm_team_id=litellm_team_id, + litellm_team_name=litellm_team_name, + team_request=team_request, + ) + + await new_team( + data=team_request, + # params used for Audit Logging + http_request=Request(scope={"type": "http", "method": "POST"}), + user_api_key_dict=UserAPIKeyAuth( + token="", + key_alias=f"litellm.{MicrosoftSSOHandler.__name__}", + ), + ) + except Exception as e: + verbose_proxy_logger.exception(f"Error creating Litellm Team: {e}") + + @staticmethod + def _cast_and_deepcopy_litellm_default_team_params( + default_team_params: Union[DefaultTeamSSOParams, Dict], + team_request: NewTeamRequest, + litellm_team_id: str, + litellm_team_name: Optional[str] = None, + ) -> NewTeamRequest: + """ + Casts and deepcopies the litellm.default_team_params to a NewTeamRequest object + + - Ensures we create a new DefaultTeamSSOParams object + - Handle the case where litellm.default_team_params is a dict or a DefaultTeamSSOParams object + - Adds the litellm_team_id and litellm_team_name to the DefaultTeamSSOParams object + """ + if isinstance(default_team_params, dict): + _team_request = deepcopy(default_team_params) + _team_request["team_id"] = litellm_team_id + _team_request["team_alias"] = litellm_team_name + team_request = NewTeamRequest(**_team_request) + elif isinstance(litellm.default_team_params, DefaultTeamSSOParams): + _default_team_params = deepcopy(litellm.default_team_params) + _new_team_request = team_request.model_dump() + _new_team_request.update(_default_team_params) + team_request = NewTeamRequest(**_new_team_request) + return team_request + + @staticmethod + def _get_cli_state( + source: str | None, + key: str | None, + existing_key: str | None = None, + user_code: str | None = None, + ) -> Optional[str]: + """ + Checks the request 'source' if a cli state token was passed in + + This is used to authenticate through the CLI login flow. + + The state parameter format is: {PREFIX}:{login_id}[:{user_code}] + - The state parameter is used to pass data through the OAuth flow without changing the callback URL + - user_code is appended only for the opt-in verification_uri_complete flow so the verify page can pre-fill it + """ + from litellm.constants import ( + LITELLM_CLI_SESSION_TOKEN_PREFIX, + ) + + if source == LITELLM_CLI_SOURCE_IDENTIFIER and key: + if _is_valid_cli_sso_user_code(user_code): + return f"{LITELLM_CLI_SESSION_TOKEN_PREFIX}:{key}:{user_code}" + return f"{LITELLM_CLI_SESSION_TOKEN_PREFIX}:{key}" + else: + return None + + @staticmethod + def _get_user_email_and_id_from_result( + result: Optional[Union[OpenID, dict]], + generic_client_id: Optional[str] = None, + ) -> ParsedOpenIDResult: + """ + Gets the user email and id from the OpenID result after validating the email domain + """ + user_email: Optional[str] = normalize_email(getattr(result, "email", None)) + user_id: Optional[str] = ( + getattr(result, "id", None) if result is not None else None + ) + user_role: Optional[str] = None + + if user_email is not None and os.getenv("ALLOWED_EMAIL_DOMAINS") is not None: + email_domain = user_email.split("@")[1] + allowed_domains = os.getenv("ALLOWED_EMAIL_DOMAINS").split(",") # type: ignore + if email_domain not in allowed_domains: + raise HTTPException( + status_code=401, + detail={ + "message": "The email domain={}, is not an allowed email domain={}. Contact your admin to change this.".format( + email_domain, allowed_domains + ) + }, + ) + + # Extract user_role from result (works for all SSO providers) + if result is not None: + _user_role = getattr(result, "user_role", None) + if _user_role is not None: + # Convert enum to string if needed + user_role = ( + _user_role.value + if isinstance(_user_role, LitellmUserRoles) + else _user_role + ) + verbose_proxy_logger.debug( + f"Extracted user_role from SSO result: {user_role}" + ) + + # generic client id - override with custom attribute name if specified + if generic_client_id is not None and result is not None: + generic_user_role_attribute_name = os.getenv( + "GENERIC_USER_ROLE_ATTRIBUTE", "role" + ) + user_id = getattr(result, "id", None) + user_email = normalize_email(getattr(result, "email", None)) + if user_role is None: + _role_from_attr = getattr(result, generic_user_role_attribute_name, None) # type: ignore + if _role_from_attr is not None: + # Convert enum to string if needed + user_role = ( + _role_from_attr.value + if isinstance(_role_from_attr, LitellmUserRoles) + else _role_from_attr + ) + + if user_id is None and result is not None: + _first_name = getattr(result, "first_name", "") or "" + _last_name = getattr(result, "last_name", "") or "" + user_id = _first_name + _last_name + + if user_email is not None and (user_id is None or len(user_id) == 0): + user_id = user_email + + return ParsedOpenIDResult( + user_email=user_email, + user_id=user_id, + user_role=user_role, + ) + + @staticmethod + async def get_redirect_response_from_openid( + result: Union[OpenID, dict, CustomOpenID], + request: Request, + received_response: Optional[dict] = None, + generic_client_id: Optional[str] = None, + ui_access_mode: Optional[Dict] = None, + access_token_payload: Optional[dict] = None, + jwt_handler: Optional[JWTHandler] = None, + return_to: Optional[str] = None, + ) -> RedirectResponse: + import jwt + + from litellm.proxy.proxy_server import ( + general_settings, + generate_key_helper_fn, + master_key, + premium_user, + proxy_logging_obj, + redis_usage_cache, + user_api_key_cache, + user_custom_sso, + ) + from litellm.proxy.utils import get_prisma_client_or_throw + from litellm.types.proxy.ui_sso import ReturnedUITokenObject + + prisma_client = get_prisma_client_or_throw( + "Prisma client is None, connect a database to your proxy" + ) + + # User is Authe'd in - generate key for the UI to access Proxy + parsed_openid_result = ( + SSOAuthenticationHandler._get_user_email_and_id_from_result( + result=result, generic_client_id=generic_client_id + ) + ) + user_email = parsed_openid_result.get("user_email") + user_id = parsed_openid_result.get("user_id") + user_role = parsed_openid_result.get("user_role") + verbose_proxy_logger.info(f"SSO callback result: {result}") + + user_info = None + user_id_models: List = [] + max_internal_user_budget = litellm.max_internal_user_budget + internal_user_budget_duration = litellm.internal_user_budget_duration + + # User might not be already created on first generation of key + # But if it is, we want their models preferences + default_ui_key_values: Dict[str, Any] = { + "duration": LITELLM_UI_SESSION_DURATION, + "key_max_budget": litellm.max_ui_session_budget, + "aliases": {}, + "config": {}, + "spend": 0, + "team_id": "litellm-dashboard", + } + user_defined_values: Optional[SSOUserDefinedValues] = None + + if user_custom_sso is not None: + if inspect.iscoroutinefunction(user_custom_sso): + user_defined_values = await user_custom_sso(result) # type: ignore + else: + raise ValueError("user_custom_sso must be a coroutine function") + elif user_id is not None: + user_defined_values = SSOUserDefinedValues( + models=user_id_models, + user_id=user_id, + user_email=user_email, + max_budget=max_internal_user_budget, + user_role=user_role, + budget_duration=internal_user_budget_duration, + ) + + # (IF SET) Verify user is in restricted SSO group + SSOAuthenticationHandler.verify_user_in_restricted_sso_group( + general_settings=general_settings, + result=result, + received_response=received_response, + ) + + user_info = await get_user_info_from_db( + result=result, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + proxy_logging_obj=proxy_logging_obj, + user_email=user_email, + user_defined_values=user_defined_values, + alternate_user_id=user_id, + ) + + # Sync user role from JWT claims via jwt_litellm_role_map (if configured). + # This ensures SSO users get the same role mapping as API/JWT users. + # Use the decoded access_token_payload (not received_response) because + # custom role claims (e.g. custom_roles) are encoded inside the JWT + # access token, which is stripped from received_response. + await _sync_user_role_from_jwt_role_map( + jwt_handler=jwt_handler, + received_response=access_token_payload or received_response, + user_info=user_info, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + user_defined_values=user_defined_values, + ) + + user_defined_values = apply_user_info_values_to_sso_user_defined_values( + user_info=user_info, user_defined_values=user_defined_values + ) + + if user_defined_values is None: + raise Exception( + "Unable to map user identity to known values. 'user_defined_values' is None. File an issue - https://github.com/BerriAI/litellm/issues" + ) + + verbose_proxy_logger.info( + f"user_defined_values for creating ui key: {user_defined_values}" + ) + + default_ui_key_values.update(user_defined_values) + default_ui_key_values["request_type"] = "key" + response = await generate_key_helper_fn( + **default_ui_key_values, # type: ignore + table_name="key", + ) + + key = response["token"] # type: ignore + user_id = response["user_id"] # type: ignore + + user_role = ( + user_defined_values["user_role"] + or LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value + ) + if user_id and isinstance(user_id, str): + user_role = await check_and_update_if_proxy_admin_id( + user_role=user_role, user_id=user_id, prisma_client=prisma_client + ) + + verbose_proxy_logger.debug( + f"user_role: {user_role}; ui_access_mode: {ui_access_mode}" + ) + ## CHECK IF ROLE ALLOWED TO USE PROXY ## + is_admin_only_access = check_is_admin_only_access(ui_access_mode or {}) + if is_admin_only_access: + has_access = has_admin_ui_access(user_role or "") + if not has_access: + raise HTTPException( + status_code=401, + detail={ + "error": f"User not allowed to access proxy. User role={user_role}, proxy mode={ui_access_mode}" + }, + ) + + disabled_non_admin_personal_key_creation = ( + get_disabled_non_admin_personal_key_creation() + ) + litellm_dashboard_ui = get_custom_url( + request_base_url=str(request.base_url), route="ui/" + ) + + if get_secret_bool("EXPERIMENTAL_UI_LOGIN"): + _user_info: Optional[LiteLLM_UserTable] = None + if ( + user_defined_values is not None + and user_defined_values["user_id"] is not None + ): + _user_info = LiteLLM_UserTable( + user_id=user_defined_values["user_id"], + user_role=user_defined_values["user_role"] or user_role, + models=[], + max_budget=litellm.max_ui_session_budget, + ) + if _user_info is None: + raise HTTPException( + status_code=401, + detail={ + "error": "User Information is required for experimental UI login" + }, + ) + + key = ExperimentalUIJWTToken.get_experimental_ui_login_jwt_auth_token( + _user_info + ) + + returned_ui_token_object = ReturnedUITokenObject( + user_id=cast(str, user_id), + key=key, + user_email=user_email, + user_role=user_role or LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value, + login_method="sso", + premium_user=premium_user, + auth_header_name=general_settings.get( + "litellm_key_header_name", "Authorization" + ), + disabled_non_admin_personal_key_creation=disabled_non_admin_personal_key_creation, + server_root_path=get_server_root_path(), + ) + + jwt_token = jwt.encode( + cast(dict, returned_ui_token_object), + master_key or "", + algorithm="HS256", + ) + + # Control-plane cross-origin: store JWT behind a single-use opaque + # code (60s TTL) so the token never appears in browser history / logs. + # The control plane redeems it via POST /v3/login/exchange. + if return_to is not None and SSOAuthenticationHandler._validate_return_to( + return_to + ): + code = secrets.token_urlsafe(32) + cache_key = f"login_code:{code}" + cache_value = {"token": jwt_token, "redirect_url": return_to} + if redis_usage_cache is not None: + await redis_usage_cache.async_set_cache( + key=cache_key, value=cache_value, ttl=60 + ) + else: + await user_api_key_cache.async_set_cache( + key=cache_key, value=cache_value, ttl=60 + ) + + separator = "&" if "?" in return_to else "?" + redirect_url = ( + return_to + separator + urlencode({"login": "success", "code": code}) + ) + verbose_proxy_logger.info( + "Cross-origin SSO: redirecting to control plane with login code" + ) + redirect_response = RedirectResponse(url=redirect_url, status_code=303) + redirect_response.delete_cookie("litellm_cp_return_to") + return redirect_response + + if user_id is not None and isinstance(user_id, str): + litellm_dashboard_ui += "?login=success" + verbose_proxy_logger.info(f"Redirecting to {litellm_dashboard_ui}") + redirect_response = RedirectResponse(url=litellm_dashboard_ui, status_code=303) + redirect_response.set_cookie(key="token", value=jwt_token) + return redirect_response + + @staticmethod + async def prepare_token_exchange_parameters( + request: Request, + generic_include_client_id: bool, + ) -> dict: + """ + Prepare token exchange parameters for Generic SSO. + + Args: + request: Request object + generic_include_client_id: Generic OAuth Client ID + + Returns: + dict: Token exchange parameters + """ + # Prepare token exchange parameters (may add code_verifier: str later) + token_params: Dict[str, Any] = {"include_client_id": generic_include_client_id} + + # Retrieve PKCE code_verifier if PKCE was used in authorization. + # Gate on GENERIC_CLIENT_USE_PKCE to avoid an unnecessary Redis round-trip + # on every non-PKCE SSO callback. + query_params = dict(request.query_params) + state = query_params.get("state") + + use_pkce = os.getenv("GENERIC_CLIENT_USE_PKCE", "false").lower() == "true" + + if use_pkce and not state: + verbose_proxy_logger.warning( + "PKCE is enabled (GENERIC_CLIENT_USE_PKCE=true) but no 'state' parameter " + "was found in the callback. The PKCE verifier cannot be retrieved without " + "a state value — the token exchange will proceed without code_verifier, " + "which the provider may reject. Ensure your OAuth provider returns 'state' " + "in the callback redirect." + ) + + if state and use_pkce: + from litellm.proxy.proxy_server import redis_usage_cache, user_api_key_cache + + cache_key = f"pkce_verifier:{state}" + if redis_usage_cache is not None: + cached_data = await redis_usage_cache.async_get_cache(key=cache_key) + else: + cached_data = await user_api_key_cache.async_get_cache(key=cache_key) + + code_verifier = None + # Track why code_verifier is absent for accurate strict-mode diagnostics. + _empty_value_in_dict = False # dict format correct but value is empty/null + + if cached_data: + # Extract code_verifier from dict (stored as dict for JSON serialization) + if isinstance(cached_data, dict) and "code_verifier" in cached_data: + code_verifier = cached_data["code_verifier"] + if not code_verifier: + # Dict format is correct but value is empty or null. This is + # a distinct case from an unrecognized format — the entry exists + # but was stored with an empty/null verifier (data integrity issue). + _empty_value_in_dict = True + verbose_proxy_logger.warning( + "PKCE verifier dict for state '%s' has an empty/null code_verifier " + "value — may indicate a storage bug. Treating as a cache miss.", + state, + ) + else: + verbose_proxy_logger.debug( + "PKCE code_verifier retrieved from cache" + ) + elif isinstance(cached_data, str): + # Handle legacy format (plain string) for backward compatibility + code_verifier = cached_data + verbose_proxy_logger.warning( + "Retrieved code_verifier in legacy plain-string format. " + "Future storage will use dict format." + ) + else: + # Defer the detailed ERROR log to the strict-mode branch below + # (which includes state and a diagnostic message). Log at DEBUG + # here to avoid duplicate ERROR entries in the same request. + verbose_proxy_logger.debug( + "Unexpected PKCE verifier cache format (type=%s); skipping.", + type(cached_data).__name__, + ) + + if code_verifier: + # Add code_verifier to token exchange parameters. + token_params["code_verifier"] = code_verifier + # Return the cache key so the caller can delete it *after* a + # successful token exchange (avoids losing the verifier on retry + # if the exchange fails partway through). + token_params["_pkce_cache_key"] = cache_key + else: + await SSOAuthenticationHandler._handle_missing_pkce_verifier( + state=state, + cache_key=cache_key, + cached_data=cached_data, + empty_value_in_dict=_empty_value_in_dict, + redis_usage_cache=redis_usage_cache, + user_api_key_cache=user_api_key_cache, + ) + return token_params + + @staticmethod + async def _handle_missing_pkce_verifier( + state: Optional[str], + cache_key: str, + cached_data: object, + empty_value_in_dict: bool, + redis_usage_cache: object, + user_api_key_cache: object, + ) -> None: + """Handle the case where PKCE verifier could not be extracted from cache. + + In strict mode (PKCE_STRICT_CACHE_MISS=true) raises ProxyException. + Otherwise logs a warning and returns (token exchange proceeds without verifier). + """ + active_cache = ( + redis_usage_cache if redis_usage_cache is not None else user_api_key_cache + ) + strict_cache_miss = ( + os.getenv("PKCE_STRICT_CACHE_MISS", "false").lower() == "true" + ) + if strict_cache_miss: + if empty_value_in_dict: + await SSOAuthenticationHandler._delete_pkce_verifier(cache_key) + raise ProxyException( + message=( + f"PKCE verifier for state '{state}' was found in cache but " + f"has an empty or null code_verifier value — possible storage bug." + ), + type=ProxyErrorTypes.auth_error, + param="PKCE_CACHE_MISS", + code=status.HTTP_401_UNAUTHORIZED, + ) + elif cached_data is not None: + await SSOAuthenticationHandler._delete_pkce_verifier(cache_key) + verbose_proxy_logger.error( + "PKCE verifier for state '%s' has an unrecognized format (type=%s); " + "treating as a cache miss. Investigate the cached value — it may be " + "a corrupt or stale entry.", + state, + type(cached_data).__name__, + ) + raise ProxyException( + message=( + f"PKCE verifier for state '{state}' has an unrecognized format " + f"(type={type(cached_data).__name__}). The cached entry may be corrupt." + ), + type=ProxyErrorTypes.auth_error, + param="PKCE_CACHE_MISS", + code=status.HTTP_401_UNAUTHORIZED, + ) + else: + if redis_usage_cache is not None: + cause = ( + "The authorization and callback were likely handled by different " + "instances — the verifier was stored on one pod but not found on another." + ) + else: + cause = ( + "The verifier may have expired (TTL), been lost on a pod restart, " + "or the PKCE authorization step was never completed. " + "Configure Redis so all proxy instances share the PKCE verifier." + ) + verbose_proxy_logger.error( + "PKCE is enabled but no verifier found in cache for state '%s'. " + "%s Cache type: %s.", + state, + cause, + type(active_cache).__name__, + ) + raise ProxyException( + message=f"PKCE verifier not found in cache for state '{state}'. {cause}", + type=ProxyErrorTypes.auth_error, + param="PKCE_CACHE_MISS", + code=status.HTTP_401_UNAUTHORIZED, + ) + else: + if cached_data is not None: + await SSOAuthenticationHandler._delete_pkce_verifier(cache_key) + verbose_proxy_logger.warning( + "PKCE is enabled but verifier not found in cache for state '%s' " + "(cache type: %s, raw data present: %s). " + "Continuing without code_verifier — set PKCE_STRICT_CACHE_MISS=true to fail fast instead.", + state, + type(active_cache).__name__, + cached_data is not None, + ) + + @staticmethod + async def _delete_pkce_verifier(cache_key: str) -> None: + """Delete a single-use PKCE verifier from cache after a successful exchange. + + Failure is non-fatal: a leftover verifier is a minor security concern + (unused key in cache) but not worth aborting an otherwise-successful login. + """ + from litellm.proxy.proxy_server import redis_usage_cache, user_api_key_cache + + try: + if redis_usage_cache is not None: + await redis_usage_cache.async_delete_cache(key=cache_key) + else: + await user_api_key_cache.async_delete_cache(key=cache_key) + except Exception as exc: + verbose_proxy_logger.warning( + "PKCE: failed to delete verifier cache key '%s' (best-effort cleanup): %s", + cache_key, + exc, + ) + + @staticmethod + def generate_pkce_params() -> Tuple[str, str]: + """ + Generate PKCE (Proof Key for Code Exchange) parameters for OAuth 2.0. + + Returns: + Tuple[str, str]: (code_verifier, code_challenge) + - code_verifier: Random 43-128 character string (we use 43 for efficiency) + - code_challenge: Base64-URL-encoded SHA256 hash of the code_verifier + + Reference: https://datatracker.ietf.org/doc/html/rfc7636 + """ + # Generate a cryptographically random code_verifier (43 characters) + # Using 32 random bytes which becomes 43 characters when base64-url-encoded + code_verifier = ( + base64.urlsafe_b64encode(secrets.token_bytes(32)) + .decode("utf-8") + .rstrip("=") + ) + + # Generate code_challenge using S256 method (SHA256) + code_challenge_bytes = hashlib.sha256(code_verifier.encode("utf-8")).digest() + code_challenge = ( + base64.urlsafe_b64encode(code_challenge_bytes).decode("utf-8").rstrip("=") + ) + + return code_verifier, code_challenge + + @staticmethod + def _validate_token_response(response: "httpx.Response") -> dict: + """ + Parse and validate the token endpoint response. + + Ensures the response is valid JSON, a dict, and contains a non-null + access_token string. Raises ProxyException on any validation failure. + """ + try: + token_response_raw = response.json() + except Exception as json_err: + verbose_proxy_logger.error( + "Failed to parse token response as JSON: %s. Body: %s", + json_err, + response.text[:500], + ) + raise ProxyException( + message=f"Token endpoint returned invalid JSON: {json_err}", + type=ProxyErrorTypes.auth_error, + param="token_exchange", + code=status.HTTP_401_UNAUTHORIZED, + ) + + if not isinstance(token_response_raw, dict): + verbose_proxy_logger.error( + "Token endpoint returned non-dict JSON (type=%s). Body: %s", + type(token_response_raw).__name__, + response.text[:500], + ) + raise ProxyException( + message=( + f"Token endpoint returned unexpected response format " + f"(expected JSON object, got {type(token_response_raw).__name__})" + ), + type=ProxyErrorTypes.auth_error, + param="token_exchange", + code=status.HTTP_401_UNAUTHORIZED, + ) + token_response: dict = token_response_raw + + access_token_val = token_response.get("access_token") + if not isinstance(access_token_val, str) or not access_token_val: + error = token_response.get("error") + error_desc = token_response.get("error_description", "") + if error: + detail = f"{error} - {error_desc}" if error_desc else error + else: + detail = ( + "token endpoint returned HTTP 200 but no access_token " + f"(response keys: {sorted(token_response.keys())})" + ) + verbose_proxy_logger.error( + "Token response missing or null access_token. detail=%s", detail + ) + raise ProxyException( + message=f"Token exchange failed: {detail}", + type=ProxyErrorTypes.auth_error, + param="token_exchange", + code=status.HTTP_401_UNAUTHORIZED, + ) + + return token_response + + @staticmethod + async def _pkce_token_exchange( + authorization_code: str, + code_verifier: str, + client_id: str, + client_secret: Optional[str], + token_endpoint: str, + userinfo_endpoint: Optional[str], + include_client_id: bool, + redirect_url: Optional[str], + additional_headers: Dict[str, str], + ) -> dict: + """ + Performs a direct OAuth token exchange including the PKCE code_verifier. + + fastapi-sso does not forward code_verifier, so when PKCE is enabled we + bypass it and call the token endpoint ourselves, then fetch user info. + + Returns a combined dict of the token response and user info, suitable + for passing to a response_convertor. + """ + verbose_proxy_logger.debug( + "PKCE: performing direct token exchange (code_verifier length=%d)", + len(code_verifier), + ) + + token_data: Dict[str, str] = { + "grant_type": "authorization_code", + "code": authorization_code, + "code_verifier": code_verifier, + } + # Only include redirect_uri when set — omitting it avoids sending the + # literal string "None" to the provider if the env var is missing. + if redirect_url: + token_data["redirect_uri"] = redirect_url + + request_headers = { + **additional_headers, + "Content-Type": "application/x-www-form-urlencoded", # must not be overridden + "Accept": "application/json", + } + + if not include_client_id: + # Use Basic Auth only when a secret is available; public PKCE clients omit it. + if client_secret: + credentials = base64.b64encode( + f"{client_id}:{client_secret}".encode() + ).decode() + request_headers["Authorization"] = f"Basic {credentials}" + else: + token_data["client_id"] = client_id + else: + token_data["client_id"] = client_id + if client_secret: + token_data["client_secret"] = client_secret + + http_client = get_async_httpx_client( + llm_provider=httpxSpecialProvider.SSO_HANDLER + ) + try: + response = await http_client.post( + url=token_endpoint, + data=token_data, + headers=request_headers, + timeout=30.0, + ) + except Exception as exc: + # Catch network-level errors (SSL, DNS, TCP, timeout, etc.) and + # wrap them as a clean ProxyException rather than leaking raw + # httpx or OS exceptions to callers. + verbose_proxy_logger.error("PKCE token endpoint unreachable: %s", exc) + raise ProxyException( + message=f"Token endpoint request failed: {exc}", + type=ProxyErrorTypes.auth_error, + param="token_exchange", + code=status.HTTP_401_UNAUTHORIZED, + ) from exc + if response.status_code != 200: + verbose_proxy_logger.error( + "PKCE token exchange failed. status=%s body=%s", + response.status_code, + response.text[:500], + ) + raise ProxyException( + message=f"Token exchange failed: {response.status_code} - {response.text[:500]}", + type=ProxyErrorTypes.auth_error, + param="token_exchange", + code=status.HTTP_401_UNAUTHORIZED, + ) + + token_response = SSOAuthenticationHandler._validate_token_response(response) + + verbose_proxy_logger.debug( + "PKCE token exchange successful. id_token_present=%s", + bool(token_response.get("id_token")), + ) + # Bearer credentials (access_token, id_token, refresh_token) are always sourced + # from token_response — not from userinfo — in the merge step below. + userinfo = await SSOAuthenticationHandler._get_pkce_userinfo( + access_token=token_response["access_token"], + id_token=token_response.get("id_token"), + userinfo_endpoint=userinfo_endpoint, + additional_headers=additional_headers, + ) + + # Merge: userinfo takes precedence for identity claims (sub, email, name, …) per + # the OpenID Connect spec (userinfo is the authoritative source for identity). + # Bearer credentials (access_token, id_token, refresh_token) from the token endpoint + # take precedence over same-named fields in userinfo — non-standard providers sometimes + # include token fields in userinfo, which must not shadow the real bearer token. + # If a bearer field is absent from the token response, any userinfo-provided value + # is preserved as a fallback (useful for non-standard providers that omit id_token + # from the token response but include it in userinfo). + # + # Three-way merge semantics for each bearer-credential field: + # 1. token_response has a non-null value → use it (token endpoint is authoritative) + # 2. token_response explicitly sent null → remove the key so callers get a clean + # absence signal; the null from the token endpoint overrides userinfo too + # 3. field absent from token_response → leave whatever userinfo provided as-is + # (e.g. userinfo-provided id_token from a non-standard provider) + merged = {**token_response, **userinfo} + for field in _OAUTH_TOKEN_FIELDS: + if token_response.get(field) is not None: + # Case 1: non-null in token_response — restore authoritative value. + merged[field] = token_response[field] + elif field in token_response: + # Case 2: key exists but value is explicitly null — remove from merged. + merged.pop(field, None) + # Case 3: field absent from token_response — leave userinfo value as-is. + return merged + + @staticmethod + async def _get_pkce_userinfo( + access_token: str, + id_token: Optional[str], + userinfo_endpoint: Optional[str], + additional_headers: Dict[str, str], + ) -> dict: + """ + Fetches user info from the userinfo endpoint. + Falls back to decoding the id_token if the endpoint is unavailable. + """ + # None = request not yet attempted, failed, or returned empty/null (treated as failure + # so the id_token fallback can be attempted instead of returning a session with no claims). + userinfo: Optional[dict] = None + + if userinfo_endpoint: + try: + client = get_async_httpx_client( + llm_provider=httpxSpecialProvider.SSO_HANDLER + ) + resp = await client.get( + url=userinfo_endpoint, + headers={ + **additional_headers, + "Authorization": f"Bearer {access_token}", # must not be overridden + }, + ) + if resp.status_code == 200: + try: + userinfo_raw = resp.json() + if not userinfo_raw: + # JSON null (None) or empty dict ({}) — no identity claims. + # Treat as failure so id_token fallback can be attempted. + verbose_proxy_logger.warning( + "Userinfo endpoint returned an empty or null response " + "(type=%s); treating as failure and attempting id_token fallback. " + "Check your provider's userinfo endpoint configuration.", + type(userinfo_raw).__name__, + ) + userinfo = None + else: + userinfo = userinfo_raw + except Exception as json_err: + verbose_proxy_logger.warning( + "Userinfo endpoint returned non-JSON response (status 200): %s", + json_err, + ) + else: + verbose_proxy_logger.warning( + "Userinfo endpoint returned %s (body: %s), falling back to id_token", + resp.status_code, + resp.text[:500], + ) + except Exception as e: + verbose_proxy_logger.warning( + "Userinfo endpoint error: %s, falling back to id_token", e + ) + + # Only fall back to id_token when the userinfo request failed (None). + # Empty dict ({}) and JSON null are both treated as failure (set to None above) since + # they contain no identity claims — id_token fallback is attempted in that case too. + # Explicitly check for a non-empty string to avoid attempting JWT decode on + # a blank or non-string id_token field from a misbehaving provider. + if userinfo is None and isinstance(id_token, str) and id_token: + try: + userinfo = jwt.decode(id_token, options={"verify_signature": False}) + if not userinfo: + # jwt.decode returned an empty dict (payload-free JWT or provider bug). + # Treat this the same as a missing userinfo — the session would have no + # identity claims, which is equivalent to a broken session. + verbose_proxy_logger.warning( + "id_token decoded to an empty payload — treating as failure." + ) + userinfo = None + except Exception as decode_err: + verbose_proxy_logger.error("Failed to decode id_token: %s", decode_err) + raise ProxyException( + message=f"Failed to decode id_token JWT: {decode_err}", + type=ProxyErrorTypes.auth_error, + param="userinfo", + code=status.HTTP_401_UNAUTHORIZED, + ) + + if userinfo is None: + id_token_attempted = isinstance(id_token, str) and bool(id_token) + if userinfo_endpoint: + if id_token_attempted: + detail = ( + "userinfo endpoint failed and id_token was present but " + "decoded to an empty payload — no identity claims available" + ) + else: + detail = "userinfo endpoint failed and no id_token was present in the token response" + else: + if id_token_attempted: + detail = ( + "no userinfo endpoint is configured (GENERIC_USERINFO_ENDPOINT) " + "and id_token decoded to an empty payload — no identity claims available" + ) + else: + detail = "no userinfo endpoint is configured (GENERIC_USERINFO_ENDPOINT) and no id_token was present" + raise ProxyException( + message=f"SSO user info unavailable: {detail}.", + type=ProxyErrorTypes.auth_error, + param="userinfo", + code=status.HTTP_401_UNAUTHORIZED, + ) + + return userinfo + + +class MicrosoftSSOHandler: + """ + Handles Microsoft SSO callback response and returns a CustomOpenID object + """ + + graph_api_base_url = "https://graph.microsoft.com/v1.0" + graph_api_user_groups_endpoint = f"{graph_api_base_url}/me/memberOf" + + """ + Constants + """ + MAX_GRAPH_API_PAGES = 200 + + # used for debugging to show the user groups litellm found from Graph API + GRAPH_API_RESPONSE_KEY = "graph_api_user_groups" + + @staticmethod + async def get_microsoft_callback_response( + request: Request, + microsoft_client_id: str, + redirect_url: str, + return_raw_sso_response: bool = False, + ) -> Union[CustomOpenID, OpenID, dict]: + """ + Get the Microsoft SSO callback response + + Args: + return_raw_sso_response: If True, return the raw SSO response + """ + microsoft_client_secret = os.getenv("MICROSOFT_CLIENT_SECRET", None) + microsoft_tenant = os.getenv("MICROSOFT_TENANT", None) + if microsoft_client_secret is None: + raise ProxyException( + message="MICROSOFT_CLIENT_SECRET not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="MICROSOFT_CLIENT_SECRET", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + if microsoft_tenant is None: + raise ProxyException( + message="MICROSOFT_TENANT not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="MICROSOFT_TENANT", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + microsoft_sso = CustomMicrosoftSSO( + client_id=microsoft_client_id, + client_secret=microsoft_client_secret, + tenant=microsoft_tenant, + redirect_uri=redirect_url, + allow_insecure_http=True, + ) + original_msft_result = ( + await microsoft_sso.verify_and_process( + request=request, + convert_response=False, # type: ignore + ) + or {} + ) + + user_team_ids = await MicrosoftSSOHandler.get_user_groups_from_graph_api( + access_token=microsoft_sso.access_token + ) + + # Extract app roles from the id_token JWT + app_roles = MicrosoftSSOHandler.get_app_roles_from_id_token( + id_token=microsoft_sso.id_token + ) + verbose_proxy_logger.debug(f"Extracted app roles from id_token: {app_roles}") + + # Combine groups and app roles + user_role: Optional[LitellmUserRoles] = None + if app_roles: + # Check if any app role is a valid LitellmUserRoles + for role_str in app_roles: + role = get_litellm_user_role(role_str) + if role is not None: + user_role = role + verbose_proxy_logger.debug( + f"Found valid LitellmUserRoles '{role.value}' in app_roles" + ) + break + + verbose_proxy_logger.debug( + f"Combined team_ids (groups + app roles): {user_team_ids}" + ) + + # if user is trying to get the raw sso response for debugging, return the raw sso response + if return_raw_sso_response: + original_msft_result[MicrosoftSSOHandler.GRAPH_API_RESPONSE_KEY] = ( + user_team_ids + ) + original_msft_result["app_roles"] = app_roles + return original_msft_result or {} + + result = MicrosoftSSOHandler.openid_from_response( + response=original_msft_result, + team_ids=user_team_ids, + user_role=user_role, + ) + return result + + @staticmethod + def openid_from_response( + response: Optional[dict], + team_ids: List[str], + user_role: Optional[LitellmUserRoles], + ) -> CustomOpenID: + response = response or {} + verbose_proxy_logger.debug(f"Microsoft SSO Callback Response: {response}") + openid_response = CustomOpenID( + email=normalize_email( + response.get(MICROSOFT_USER_EMAIL_ATTRIBUTE) or response.get("mail") + ), + display_name=response.get(MICROSOFT_USER_DISPLAY_NAME_ATTRIBUTE), + provider="microsoft", + id=response.get(MICROSOFT_USER_ID_ATTRIBUTE), + first_name=response.get(MICROSOFT_USER_FIRST_NAME_ATTRIBUTE), + last_name=response.get(MICROSOFT_USER_LAST_NAME_ATTRIBUTE), + team_ids=team_ids, + user_role=user_role, + ) + verbose_proxy_logger.debug(f"Microsoft SSO OpenID Response: {openid_response}") + return openid_response + + @staticmethod + def get_app_roles_from_id_token(id_token: Optional[str]) -> List[str]: + """ + Extract app roles from the Microsoft Entra ID (Azure AD) id_token JWT. + + App roles are assigned in the Azure AD Enterprise Application and appear + in the 'app_roles' claim of the id_token. + + Args: + id_token (Optional[str]): The JWT id_token from Microsoft SSO + + Returns: + List[str]: List of app role names assigned to the user + """ + if not id_token: + verbose_proxy_logger.debug("No id_token provided for app role extraction") + return [] + + try: + import jwt + + # Decode the JWT without signature verification + # (signature is already verified by fastapi_sso) + decoded_token = jwt.decode(id_token, options={"verify_signature": False}) + + # Extract app_roles claim from the token + ## check for both 'roles' and 'app_roles' claims + roles = decoded_token.get("app_roles", []) or decoded_token.get("roles", []) + + if roles and isinstance(roles, list): + verbose_proxy_logger.debug( + f"Found {len(roles)} app role(s) in id_token: {roles}" + ) + return roles + else: + verbose_proxy_logger.debug( + "No app roles found in id_token or roles claim is not a list" + ) + return [] + + except Exception as e: + verbose_proxy_logger.error(f"Error extracting app roles from id_token: {e}") + return [] + + @staticmethod + async def get_user_groups_from_graph_api( + access_token: Optional[str] = None, + ) -> List[str]: + """ + Returns a list of `team_ids` the user belongs to from the Microsoft Graph API + + Args: + access_token (Optional[str]): Microsoft Graph API access token + + Returns: + List[str]: List of group IDs the user belongs to + """ + try: + async_client = get_async_httpx_client( + llm_provider=httpxSpecialProvider.SSO_HANDLER + ) + + # Handle MSFT Enterprise Application Groups + service_principal_id = os.getenv("MICROSOFT_SERVICE_PRINCIPAL_ID", None) + service_principal_group_ids: Optional[List[str]] = [] + service_principal_teams: Optional[List[MicrosoftServicePrincipalTeam]] = [] + if service_principal_id: + ( + service_principal_group_ids, + service_principal_teams, + ) = await MicrosoftSSOHandler.get_group_ids_from_service_principal( + service_principal_id=service_principal_id, + async_client=async_client, + access_token=access_token, + ) + verbose_proxy_logger.debug( + f"Service principal group IDs: {service_principal_group_ids}" + ) + if len(service_principal_group_ids) > 0: + await MicrosoftSSOHandler.create_litellm_teams_from_service_principal_team_ids( + service_principal_teams=service_principal_teams, + ) + + # Fetch user membership from Microsoft Graph API + all_group_ids = [] + next_link: Optional[str] = ( + MicrosoftSSOHandler.graph_api_user_groups_endpoint + ) + auth_headers = {"Authorization": f"Bearer {access_token}"} + page_count = 0 + + while ( + next_link is not None + and page_count < MicrosoftSSOHandler.MAX_GRAPH_API_PAGES + ): + group_ids, next_link = await MicrosoftSSOHandler.fetch_and_parse_groups( + url=next_link, headers=auth_headers, async_client=async_client + ) + all_group_ids.extend(group_ids) + page_count += 1 + + if ( + next_link is not None + and page_count >= MicrosoftSSOHandler.MAX_GRAPH_API_PAGES + ): + verbose_proxy_logger.warning( + f"Reached maximum page limit of {MicrosoftSSOHandler.MAX_GRAPH_API_PAGES}. Some groups may not be included." + ) + + # If service_principal_group_ids is not empty, only return group_ids that are in both all_group_ids and service_principal_group_ids + if service_principal_group_ids and len(service_principal_group_ids) > 0: + all_group_ids = [ + group_id + for group_id in all_group_ids + if group_id in service_principal_group_ids + ] + + return all_group_ids + + except Exception as e: + verbose_proxy_logger.error( + f"Error getting user groups from Microsoft Graph API: {e}" + ) + return [] + + @staticmethod + async def fetch_and_parse_groups( + url: str, headers: dict, async_client: AsyncHTTPHandler + ) -> Tuple[List[str], Optional[str]]: + """Helper function to fetch and parse group data from a URL""" + response = await async_client.get(url, headers=headers) + response_json = response.json() + response_typed = await MicrosoftSSOHandler._cast_graph_api_response_dict( + response=response_json + ) + group_ids = MicrosoftSSOHandler._get_group_ids_from_graph_api_response( + response=response_typed + ) + return group_ids, response_typed.get("odata_nextLink") + + @staticmethod + def _get_group_ids_from_graph_api_response( + response: MicrosoftGraphAPIUserGroupResponse, + ) -> List[str]: + group_ids = [] + for _object in response.get("value", []) or []: + _group_id = _object.get("id") + if _group_id is not None: + group_ids.append(_group_id) + return group_ids + + @staticmethod + async def _cast_graph_api_response_dict( + response: dict, + ) -> MicrosoftGraphAPIUserGroupResponse: + directory_objects: List[MicrosoftGraphAPIUserGroupDirectoryObject] = [] + for _object in response.get("value", []): + directory_objects.append( + MicrosoftGraphAPIUserGroupDirectoryObject( + odata_type=_object.get("@odata.type"), + id=_object.get("id"), + deletedDateTime=_object.get("deletedDateTime"), + description=_object.get("description"), + displayName=_object.get("displayName"), + roleTemplateId=_object.get("roleTemplateId"), + ) + ) + return MicrosoftGraphAPIUserGroupResponse( + odata_context=response.get("@odata.context"), + odata_nextLink=response.get("@odata.nextLink"), + value=directory_objects, + ) + + @staticmethod + async def get_group_ids_from_service_principal( + service_principal_id: str, + async_client: AsyncHTTPHandler, + access_token: Optional[str] = None, + ) -> Tuple[List[str], List[MicrosoftServicePrincipalTeam]]: + """ + Gets the groups belonging to the Service Principal Application + + Service Principal Id is an `Enterprise Application` in Azure AD + + Users use Enterprise Applications to manage Groups and Users on Microsoft Entra ID + """ + base_url = "https://graph.microsoft.com/v1.0" + # Endpoint to get app role assignments for the given service principal + endpoint = f"/servicePrincipals/{service_principal_id}/appRoleAssignedTo" + url = base_url + endpoint + + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } + + response = await async_client.get(url, headers=headers) + response_json = response.json() + verbose_proxy_logger.debug( + f"Response from service principal app role assigned to: {response_json}" + ) + group_ids: List[str] = [] + service_principal_teams: List[MicrosoftServicePrincipalTeam] = [] + + for _object in response_json.get("value", []): + if _object.get("principalType") == "Group": + # Append the group ID to the list + group_ids.append(_object.get("principalId")) + # Append the service principal team to the list + service_principal_teams.append( + MicrosoftServicePrincipalTeam( + principalDisplayName=_object.get("principalDisplayName"), + principalId=_object.get("principalId"), + ) + ) + + return group_ids, service_principal_teams + + @staticmethod + async def create_litellm_teams_from_service_principal_team_ids( + service_principal_teams: List[MicrosoftServicePrincipalTeam], + ): + """ + Creates Litellm Teams from the Service Principal Group IDs + + When a user sets a `SERVICE_PRINCIPAL_ID` in the env, litellm will fetch groups under that service principal and create Litellm Teams from them + """ + verbose_proxy_logger.debug( + f"Creating Litellm Teams from Service Principal Teams: {service_principal_teams}" + ) + for service_principal_team in service_principal_teams: + litellm_team_id: Optional[str] = service_principal_team.get("principalId") + litellm_team_name: Optional[str] = service_principal_team.get( + "principalDisplayName" + ) + if not litellm_team_id: + verbose_proxy_logger.debug( + f"Skipping team creation for {litellm_team_name} because it has no principalId" + ) + continue + + await SSOAuthenticationHandler.create_litellm_team_from_sso_group( + litellm_team_id=litellm_team_id, + litellm_team_name=litellm_team_name, + ) + + +class GoogleSSOHandler: + """ + Handles Google SSO callback response and returns a CustomOpenID object + """ + + @staticmethod + async def get_google_callback_response( + request: Request, + google_client_id: str, + redirect_url: str, + return_raw_sso_response: bool = False, + ) -> Union[OpenID, dict]: + """ + Get the Google SSO callback response + + Args: + return_raw_sso_response: If True, return the raw SSO response + """ + from fastapi_sso.sso.google import GoogleSSO + + google_client_secret = os.getenv("GOOGLE_CLIENT_SECRET", None) + if google_client_secret is None: + raise ProxyException( + message="GOOGLE_CLIENT_SECRET not set. Set it in .env file", + type=ProxyErrorTypes.auth_error, + param="GOOGLE_CLIENT_SECRET", + code=status.HTTP_500_INTERNAL_SERVER_ERROR, + ) + google_sso = GoogleSSO( + client_id=google_client_id, + redirect_uri=redirect_url, + client_secret=google_client_secret, + ) + + # if user is trying to get the raw sso response for debugging, return the raw sso response + if return_raw_sso_response: + return ( + await google_sso.verify_and_process( + request=request, + convert_response=False, # type: ignore + ) + or {} + ) + + result = await google_sso.verify_and_process(request) + return result or {} + + +@router.get("/sso/debug/login", tags=["experimental"], include_in_schema=False) +async def debug_sso_login(request: Request): + """ + Create Proxy API Keys using Google Workspace SSO. Requires setting PROXY_BASE_URL in .env + PROXY_BASE_URL should be the your deployed proxy endpoint, e.g. PROXY_BASE_URL="https://litellm-production-7002.up.railway.app/" + Example: + """ + from litellm.proxy.proxy_server import premium_user + + microsoft_client_id = os.getenv("MICROSOFT_CLIENT_ID", None) + google_client_id = os.getenv("GOOGLE_CLIENT_ID", None) + generic_client_id = os.getenv("GENERIC_CLIENT_ID", None) + + ####### Check if user is a Enterprise / Premium User ####### + if ( + microsoft_client_id is not None + or google_client_id is not None + or generic_client_id is not None + ): + if premium_user is not True: + raise ProxyException( + message="You must be a LiteLLM Enterprise user to use SSO. If you have a license please set `LITELLM_LICENSE` in your env. If you want to obtain a license meet with us here: https://enterprise.litellm.ai/demo You are seeing this error message because You set one of `MICROSOFT_CLIENT_ID`, `GOOGLE_CLIENT_ID`, or `GENERIC_CLIENT_ID` in your env. Please unset this", + type=ProxyErrorTypes.auth_error, + param="premium_user", + code=status.HTTP_403_FORBIDDEN, + ) + + # get url from request + redirect_url = SSOAuthenticationHandler.get_redirect_url_for_sso( + request=request, + sso_callback_route="sso/debug/callback", + ) + + # Check if we should use SSO handler + if ( + SSOAuthenticationHandler.should_use_sso_handler( + microsoft_client_id=microsoft_client_id, + google_client_id=google_client_id, + generic_client_id=generic_client_id, + ) + is True + ): + return await SSOAuthenticationHandler.get_sso_login_redirect( + redirect_url=redirect_url, + microsoft_client_id=microsoft_client_id, + google_client_id=google_client_id, + generic_client_id=generic_client_id, + request=request, + ) + + +@router.get("/sso/debug/callback", tags=["experimental"], include_in_schema=False) +async def debug_sso_callback(request: Request): + """ + Returns the OpenID object returned by the SSO provider + """ + import json + + from fastapi.responses import HTMLResponse + + from litellm.proxy._types import LiteLLM_JWTAuth + from litellm.proxy.auth.handle_jwt import JWTHandler + from litellm.proxy.proxy_server import ( + general_settings, + jwt_handler, + prisma_client, + user_api_key_cache, + ) + + sso_jwt_handler: Optional[JWTHandler] = None + ui_access_mode = general_settings.get("ui_access_mode", None) + if ui_access_mode is not None and isinstance(ui_access_mode, dict): + sso_jwt_handler = JWTHandler() + sso_jwt_handler.update_environment( + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + litellm_jwtauth=LiteLLM_JWTAuth( + team_ids_jwt_field=general_settings.get("ui_access_mode", {}).get( + "sso_group_jwt_field", None + ), + ), + leeway=0, + ) + + microsoft_client_id = os.getenv("MICROSOFT_CLIENT_ID", None) + google_client_id = os.getenv("GOOGLE_CLIENT_ID", None) + generic_client_id = os.getenv("GENERIC_CLIENT_ID", None) + + redirect_url = os.getenv("PROXY_BASE_URL", str(request.base_url)) + if redirect_url.endswith("/"): + redirect_url += "sso/debug/callback" + else: + redirect_url += "/sso/debug/callback" + + result = None + received_response: Optional[dict] = None + access_token_payload: Optional[dict] = None + if google_client_id is not None: + result = await GoogleSSOHandler.get_google_callback_response( + request=request, + google_client_id=google_client_id, + redirect_url=redirect_url, + return_raw_sso_response=True, + ) + elif microsoft_client_id is not None: + result = await MicrosoftSSOHandler.get_microsoft_callback_response( + request=request, + microsoft_client_id=microsoft_client_id, + redirect_url=redirect_url, + return_raw_sso_response=True, + ) + + elif generic_client_id is not None: + result, received_response, access_token_payload = ( + await get_generic_sso_response( + request=request, + jwt_handler=jwt_handler, + generic_client_id=generic_client_id, + redirect_url=redirect_url, + sso_jwt_handler=sso_jwt_handler, + ) + ) + + # If result is None, return a basic error message + if result is None: + return HTMLResponse( + content="

SSO Authentication Failed

No data was returned from the SSO provider.

", + status_code=400, + ) + + # Convert the OpenID object to a dictionary + if hasattr(result, "__dict__"): + result_dict = result.__dict__ + else: + result_dict = dict(result) + + # Filter out any None values and convert to JSON serializable format + filtered_result = {} + for key, value in result_dict.items(): + if value is not None and not key.startswith("_"): + if isinstance(value, (str, int, float, bool)) or value is None: + filtered_result[key] = value + else: + try: + # Try to convert to string or another JSON serializable format + filtered_result[key] = str(value) + except Exception as e: + filtered_result[key] = f"Complex value (not displayable): {str(e)}" + + # Defense-in-depth: ensure no bearer tokens leak into the rendered HTML even if + # a non-conforming IdP places them in its userinfo response. + safe_raw_claims = { + k: v + for k, v in (received_response or {}).items() + if k not in _OAUTH_TOKEN_FIELDS + } + safe_access_token_claims = { + k: v + for k, v in (access_token_payload or {}).items() + if k not in _OAUTH_TOKEN_FIELDS + } + + sso_payload = { + "parsed_by_proxy": filtered_result, + "raw_claims": safe_raw_claims, + "access_token_claims": safe_access_token_claims, + } + + # Replace the placeholder in the template with the actual data + sso_payload_json = json.dumps(sso_payload, indent=2, default=str).replace( + "