Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/sso/saml_sso.py: 28%
238 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1"""
2SAML 2.0 SSO for the LiteLLM proxy admin UI.
4Supports both SP-initiated and IdP-initiated login via the HTTP-POST binding,
5using the OneLogin python3-saml toolkit for signature, audience and time
6validation. The IdP is configured from its metadata (``SAML_IDP_METADATA_URL``
7or inline ``SAML_IDP_METADATA_XML``); a successful login is mapped to a
8``CustomOpenID`` and handed to the shared post-login path used by every other
9SSO provider.
11python3-saml pulls in the native ``xmlsec``/``libxml2`` libraries, so it is an
12optional dependency. When it is not installed the SAML routes return a clear
13error instead of breaking proxy startup.
14"""
16# python3-saml ships no type stubs, so the type checker sees every onelogin call
17# as Unknown and the guarded optional import as possibly-unbound. Values crossing
18# that boundary are cast() to concrete types at each use site; these directives
19# silence only the unavoidable noise from the untyped dependency in this module.
20# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false
21# pyright: reportUnknownArgumentType=false, reportUnknownParameterType=false
22# pyright: reportMissingTypeStubs=false, reportPossiblyUnboundVariable=false
23# pyright: reportConstantRedefinition=false
25import asyncio
26import hashlib
27import os
28import secrets
29import time
30from typing import Final, cast
31from urllib.parse import parse_qsl
33from fastapi import HTTPException, Request, status
34from fastapi.responses import RedirectResponse
35from pydantic import ValidationError
37from litellm._logging import verbose_proxy_logger
38from litellm.caching.dual_cache import DualCache
39from litellm.proxy.auth.ip_address_utils import IPAddressUtils
40from litellm.proxy.management_endpoints.types import CustomOpenID, get_litellm_user_role
41from litellm.proxy.utils import get_custom_url
43try:
44 from onelogin.saml2.auth import OneLogin_Saml2_Auth
45 from onelogin.saml2.idp_metadata_parser import OneLogin_Saml2_IdPMetadataParser
46 from onelogin.saml2.settings import OneLogin_Saml2_Settings
47 from onelogin.saml2.xml_utils import OneLogin_Saml2_XML
49 SAML_AVAILABLE = True
50except ImportError:
51 SAML_AVAILABLE = False
53SAML_LOGIN_ROUTE: Final = "sso/saml/login"
54SAML_CALLBACK_ROUTE: Final = "sso/saml/callback"
55SAML_METADATA_ROUTE: Final = "sso/saml/metadata"
57_SAML_AUTHN_STATE_COOKIE: Final = "litellm_saml_authn"
58_SAML_IDP_SETTINGS_CACHE_PREFIX: Final = "saml_idp_settings"
59_SAML_AUTHN_REQUEST_CACHE_PREFIX: Final = "saml_authn_request"
60_SAML_CONSUMED_ASSERTION_CACHE_PREFIX: Final = "saml_consumed_assertion"
61_SAML_AUTHN_REQUEST_TTL_SECONDS: Final = 600
62_SAML_IDP_METADATA_TTL_SECONDS: Final = 3600
63_SAML_METADATA_FETCH_TIMEOUT_SECONDS: Final = 10
64_SAML_MAX_POST_BYTES: Final = 5 * 1024 * 1024
65# The replay guard tracks each assertion's NotOnOrAfter so it spans the full
66# validity window; the floor covers IdPs that issue hour-long assertions or omit
67# the timestamp, and the cap bounds cache growth.
68_SAML_REPLAY_GUARD_DEFAULT_TTL_SECONDS: Final = 3600
69_SAML_REPLAY_GUARD_MAX_TTL_SECONDS: Final = 86400
71_EMAIL_ATTRIBUTE_CANDIDATES: Final = (
72 "urn:oid:0.9.2342.19200300.100.1.3",
73 "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress",
74 "email",
75 "emailAddress",
76 "mail",
77 "Email",
78)
79_FIRST_NAME_ATTRIBUTE_CANDIDATES: Final = (
80 "urn:oid:2.5.4.42",
81 "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/givenname",
82 "givenName",
83 "first_name",
84 "firstName",
85)
86_LAST_NAME_ATTRIBUTE_CANDIDATES: Final = (
87 "urn:oid:2.5.4.4",
88 "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/surname",
89 "sn",
90 "surname",
91 "last_name",
92 "lastName",
93)
94_ROLE_ATTRIBUTE_CANDIDATES: Final = ("role", "roles", "litellm_role")
95_TEAM_IDS_ATTRIBUTE_CANDIDATES: Final = ("teams", "team_ids", "groups")
98def _saml_unavailable_error() -> HTTPException:
99 return HTTPException(
100 status_code=status.HTTP_501_NOT_IMPLEMENTED,
101 detail=(
102 "SAML SSO requires the optional 'python3-saml' dependency, which is "
103 "not installed. Re-install litellm with the saml extra: "
104 "'pip install litellm[saml]'. The saml extra bundles the native "
105 "xmlsec/libxml2 libraries, so no system packages are required."
106 ),
107 )
110class SAMLAuthHandler:
111 """SP- and IdP-initiated SAML 2.0 login for the admin UI."""
113 @staticmethod
114 def _env(name: str, default: str | None = None) -> str | None:
115 return os.getenv(name, default)
117 @staticmethod
118 def is_saml_configured() -> bool:
119 return bool(SAMLAuthHandler._env("SAML_IDP_METADATA_URL") or SAMLAuthHandler._env("SAML_IDP_METADATA_XML"))
121 @staticmethod
122 def _bool_env(name: str, default: bool) -> bool:
123 raw: Final = SAMLAuthHandler._env(name)
124 if raw is None:
125 return default
126 return raw.strip().lower() in ("true", "1", "yes", "on")
128 @staticmethod
129 def _base_url(request: Request) -> str:
130 base: Final = get_custom_url(request_base_url=str(request.base_url))
131 return base if base.endswith("/") else base + "/"
133 @staticmethod
134 def _is_https(request: Request) -> bool:
135 return IPAddressUtils.is_request_https(request)
137 @staticmethod
138 def _acs_url(request: Request) -> str:
139 return SAMLAuthHandler._base_url(request) + SAML_CALLBACK_ROUTE
141 @staticmethod
142 def _metadata_url(request: Request) -> str:
143 return SAMLAuthHandler._base_url(request) + SAML_METADATA_ROUTE
145 @staticmethod
146 def _sp_entity_id(request: Request) -> str:
147 return SAMLAuthHandler._env("SAML_SP_ENTITY_ID") or SAMLAuthHandler._metadata_url(request)
149 @staticmethod
150 async def _load_idp_settings(cache: DualCache) -> dict[str, object]:
151 metadata_url: Final = SAMLAuthHandler._env("SAML_IDP_METADATA_URL")
152 metadata_xml: Final = SAMLAuthHandler._env("SAML_IDP_METADATA_XML")
153 source: Final = metadata_url or metadata_xml
154 if source is None:
155 raise HTTPException(
156 status_code=status.HTTP_501_NOT_IMPLEMENTED,
157 detail="SAML SSO is not configured. Set SAML_IDP_METADATA_URL or SAML_IDP_METADATA_XML.",
158 )
160 cache_key: Final = f"{_SAML_IDP_SETTINGS_CACHE_PREFIX}:{hashlib.sha256(source.encode()).hexdigest()}"
161 cached: Final = cache.get_cache(key=cache_key)
162 if isinstance(cached, dict):
163 return cast(dict[str, object], cached) # cast-ok: untyped python3-saml
165 if metadata_url is not None:
166 parsed = await asyncio.to_thread(
167 OneLogin_Saml2_IdPMetadataParser.parse_remote,
168 metadata_url,
169 validate_cert=SAMLAuthHandler._bool_env("SAML_IDP_METADATA_VALIDATE_CERT", True),
170 timeout=_SAML_METADATA_FETCH_TIMEOUT_SECONDS,
171 )
172 else:
173 parsed = OneLogin_Saml2_IdPMetadataParser.parse(cast(str, metadata_xml)) # cast-ok: untyped python3-saml
175 idp_settings: Final = cast(dict[str, object], parsed) # cast-ok: untyped python3-saml
176 if not idp_settings.get("idp"):
177 raise HTTPException(
178 status_code=status.HTTP_502_BAD_GATEWAY,
179 detail="Could not parse an IdP entityID/SSO URL/certificate from the SAML metadata.",
180 )
181 cache.set_cache(key=cache_key, value=idp_settings, ttl=_SAML_IDP_METADATA_TTL_SECONDS)
182 return idp_settings
184 @staticmethod
185 def _build_settings(request: Request, idp_settings: dict[str, object]) -> dict[str, object]:
186 sp_settings: Final[dict[str, object]] = {
187 "strict": SAMLAuthHandler._bool_env("SAML_STRICT", True),
188 "debug": False,
189 "sp": {
190 "entityId": SAMLAuthHandler._sp_entity_id(request),
191 "assertionConsumerService": {
192 "url": SAMLAuthHandler._acs_url(request),
193 "binding": "urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST",
194 },
195 "NameIDFormat": SAMLAuthHandler._env(
196 "SAML_SP_NAME_ID_FORMAT",
197 "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress",
198 ),
199 },
200 "security": {
201 "wantAssertionsSigned": SAMLAuthHandler._bool_env("SAML_WANT_ASSERTIONS_SIGNED", True),
202 "wantMessagesSigned": SAMLAuthHandler._bool_env("SAML_WANT_MESSAGES_SIGNED", False),
203 "authnRequestsSigned": SAMLAuthHandler._bool_env("SAML_AUTHN_REQUESTS_SIGNED", False),
204 "wantNameId": True,
205 "requestedAuthnContext": False,
206 "rejectUnsolicitedResponsesWithInResponseTo": False,
207 },
208 }
209 return OneLogin_Saml2_IdPMetadataParser.merge_settings(sp_settings, idp_settings)
211 @staticmethod
212 def _prepare_request_data(request: Request, post_data: dict[str, str] | None = None) -> dict[str, object]:
213 base: Final = SAMLAuthHandler._base_url(request)
214 scheme, _, host_part = base.partition("://")
215 host: Final = host_part.split("/", 1)[0]
216 return {
217 "https": "on" if scheme == "https" else "off",
218 "http_host": host,
219 "script_name": "/" + SAML_CALLBACK_ROUTE,
220 "get_data": dict(request.query_params),
221 "post_data": post_data or {},
222 }
224 @staticmethod
225 async def _build_auth(
226 request: Request,
227 cache: DualCache,
228 post_data: dict[str, str] | None = None,
229 ) -> "OneLogin_Saml2_Auth":
230 if not SAML_AVAILABLE:
231 raise _saml_unavailable_error()
232 idp_settings: Final = await SAMLAuthHandler._load_idp_settings(cache)
233 settings: Final = SAMLAuthHandler._build_settings(request, idp_settings)
234 request_data: Final = SAMLAuthHandler._prepare_request_data(request, post_data)
235 try:
236 return OneLogin_Saml2_Auth(request_data, old_settings=settings)
237 except Exception as e: # noqa: BLE001 - toolkit exposes no common exception base; fail closed
238 raise HTTPException(
239 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
240 detail=f"Invalid SAML configuration: {e}",
241 )
243 @staticmethod
244 async def build_login_redirect(
245 request: Request, cache: DualCache, relay_state: str | None = None
246 ) -> RedirectResponse:
247 auth: Final = await SAMLAuthHandler._build_auth(request, cache)
248 redirect_url: Final = cast(str, auth.login(return_to=relay_state)) # cast-ok: untyped python3-saml
249 response: Final = RedirectResponse(url=redirect_url, status_code=303)
250 request_id: Final = cast(str | None, auth.get_last_request_id()) # cast-ok: untyped python3-saml
251 if request_id is not None:
252 cache.set_cache(
253 key=f"{_SAML_AUTHN_REQUEST_CACHE_PREFIX}:{request_id}",
254 value="1",
255 ttl=_SAML_AUTHN_REQUEST_TTL_SECONDS,
256 )
257 secure: Final = SAMLAuthHandler._is_https(request)
258 response.set_cookie(
259 key=_SAML_AUTHN_STATE_COOKIE,
260 value=request_id,
261 max_age=_SAML_AUTHN_REQUEST_TTL_SECONDS,
262 httponly=True,
263 secure=secure,
264 samesite="none" if secure else "lax",
265 )
266 return response
268 @staticmethod
269 async def build_sp_metadata(request: Request, cache: DualCache) -> str:
270 if not SAML_AVAILABLE:
271 raise _saml_unavailable_error()
272 idp_settings: Final = await SAMLAuthHandler._load_idp_settings(cache)
273 settings: Final = SAMLAuthHandler._build_settings(request, idp_settings)
274 saml_settings: Final = OneLogin_Saml2_Settings(settings, sp_validation_only=True)
275 metadata: Final = cast(str, saml_settings.get_sp_metadata()) # cast-ok: untyped python3-saml
276 errors: Final = cast(list[str], saml_settings.validate_metadata(metadata)) # cast-ok: untyped python3-saml
277 if errors:
278 raise HTTPException(
279 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
280 detail=f"Invalid SP metadata: {', '.join(errors)}",
281 )
282 return metadata
284 @staticmethod
285 async def read_acs_post_data(request: Request) -> dict[str, str]:
286 """Read the ACS POST form under a hard size cap before any base64/XML decoding.
288 Bounds both Content-Length-declared and chunked requests so an unauthenticated
289 caller cannot force unbounded buffering while decoding the SAMLResponse."""
290 declared: Final = request.headers.get("content-length")
291 if declared is not None and declared.isdigit() and int(declared) > _SAML_MAX_POST_BYTES:
292 raise HTTPException(
293 status_code=status.HTTP_413_CONTENT_TOO_LARGE,
294 detail="SAML response exceeds the maximum allowed size.",
295 )
297 body = bytearray()
298 async for chunk in request.stream():
299 body += chunk
300 if len(body) > _SAML_MAX_POST_BYTES:
301 raise HTTPException(
302 status_code=status.HTTP_413_CONTENT_TOO_LARGE,
303 detail="SAML response exceeds the maximum allowed size.",
304 )
306 return dict(parse_qsl(body.decode("utf-8", "replace")))
308 @staticmethod
309 async def handle_acs(request: Request, cache: DualCache, post_data: dict[str, str]) -> CustomOpenID:
310 auth: Final = await SAMLAuthHandler._build_auth(request, cache, post_data=post_data)
311 browser_request_id: Final = request.cookies.get(_SAML_AUTHN_STATE_COOKIE)
312 try:
313 auth.process_response(request_id=browser_request_id)
314 except Exception as e: # noqa: BLE001 - toolkit exposes no common exception base; fail closed
315 raise HTTPException(
316 status_code=status.HTTP_401_UNAUTHORIZED,
317 detail=f"Could not process SAML response: {e}",
318 )
320 errors: Final = cast(list[str], auth.get_errors()) # cast-ok: untyped python3-saml
321 if errors or not auth.is_authenticated():
322 reason: Final = auth.get_last_error_reason()
323 raise HTTPException(
324 status_code=status.HTTP_401_UNAUTHORIZED,
325 detail=f"SAML authentication failed: {reason or ', '.join(errors)}",
326 )
328 await SAMLAuthHandler._enforce_response_binding(auth, cache, browser_request_id)
329 return SAMLAuthHandler._result_from_auth(auth)
331 @staticmethod
332 def _replay_guard_ttl(auth: "OneLogin_Saml2_Auth") -> int:
333 not_on_or_after: Final = auth.get_last_assertion_not_on_or_after()
334 if not isinstance(not_on_or_after, int):
335 return _SAML_REPLAY_GUARD_DEFAULT_TTL_SECONDS
336 remaining: Final = not_on_or_after - int(time.time())
337 return min(
338 max(remaining, _SAML_REPLAY_GUARD_DEFAULT_TTL_SECONDS),
339 _SAML_REPLAY_GUARD_MAX_TTL_SECONDS,
340 )
342 @staticmethod
343 def _response_in_response_to(auth: "OneLogin_Saml2_Auth") -> str | None:
344 """The request id this response answers, read from the Response element or, when the
345 IdP only stamps it on the bearer SubjectConfirmationData, from there. A non-None value
346 marks the response as solicited (SP-initiated) and so requiring browser binding."""
347 value: Final = cast(str | None, auth.get_last_response_in_response_to()) # cast-ok: untyped python3-saml
348 if value:
349 return value
350 xml: Final = cast(bytes | None, auth.get_last_response_xml()) # cast-ok: untyped python3-saml
351 if not xml:
352 return None
353 root: Final = OneLogin_Saml2_XML.to_etree(xml)
354 for node in OneLogin_Saml2_XML.query(root, "//saml:SubjectConfirmationData[@InResponseTo]"):
355 irt = cast(str | None, node.get("InResponseTo")) # cast-ok: untyped python3-saml
356 if irt:
357 return irt
358 return None
360 @staticmethod
361 async def _enforce_response_binding(
362 auth: "OneLogin_Saml2_Auth",
363 cache: DualCache,
364 browser_request_id: str | None,
365 ) -> None:
366 in_response_to: Final = SAMLAuthHandler._response_in_response_to(auth)
368 if in_response_to is not None:
369 authn_key: Final = f"{_SAML_AUTHN_REQUEST_CACHE_PREFIX}:{in_response_to}"
370 if cache.get_cache(key=authn_key) is None:
371 raise HTTPException(
372 status_code=status.HTTP_401_UNAUTHORIZED,
373 detail="SAML response references an unknown or already-used login request.",
374 )
375 if browser_request_id is None or not secrets.compare_digest(browser_request_id, in_response_to):
376 raise HTTPException(
377 status_code=status.HTTP_401_UNAUTHORIZED,
378 detail="SAML response is not bound to this browser's login request.",
379 )
380 elif browser_request_id is not None:
381 raise HTTPException(
382 status_code=status.HTTP_401_UNAUTHORIZED,
383 detail="SAML response is not bound to this browser's login request.",
384 )
385 elif not SAMLAuthHandler._bool_env("SAML_ALLOW_UNSOLICITED", False):
386 raise HTTPException(
387 status_code=status.HTTP_401_UNAUTHORIZED,
388 detail="Unsolicited (IdP-initiated) SAML responses are disabled.",
389 )
390 elif cache.redis_cache is None:
391 raise HTTPException(
392 status_code=status.HTTP_401_UNAUTHORIZED,
393 detail=(
394 "Unsolicited (IdP-initiated) SAML responses require a shared Redis cache "
395 "so the replay guard is enforced across every worker."
396 ),
397 )
399 assertion_id: Final = cast(str | None, auth.get_last_assertion_id()) # cast-ok: untyped python3-saml
400 if assertion_id is None:
401 raise HTTPException(
402 status_code=status.HTTP_401_UNAUTHORIZED,
403 detail="SAML assertion is missing the required ID attribute.",
404 )
405 consumed_key: Final = f"{_SAML_CONSUMED_ASSERTION_CACHE_PREFIX}:{assertion_id}"
406 consumed_count: Final = await cache.async_increment_cache(
407 key=consumed_key, value=1, ttl=SAMLAuthHandler._replay_guard_ttl(auth)
408 )
409 if consumed_count is not None and consumed_count > 1:
410 raise HTTPException(
411 status_code=status.HTTP_401_UNAUTHORIZED,
412 detail="SAML assertion has already been used (replay detected).",
413 )
415 @staticmethod
416 def _result_from_auth(auth: "OneLogin_Saml2_Auth") -> CustomOpenID:
417 attributes: Final = cast(dict[str, list[str]], auth.get_attributes()) # cast-ok: untyped python3-saml
418 name_id: Final = cast(str | None, auth.get_nameid()) # cast-ok: untyped python3-saml
420 email = SAMLAuthHandler._attribute_value(attributes, "SAML_ATTRIBUTE_EMAIL", _EMAIL_ATTRIBUTE_CANDIDATES)
421 if email is None and name_id is not None and "@" in name_id:
422 email = name_id
424 if email is None and SAMLAuthHandler._env("ALLOWED_EMAIL_DOMAINS") is not None:
425 raise HTTPException(
426 status_code=status.HTTP_401_UNAUTHORIZED,
427 detail=(
428 "SAML assertion did not contain an email address, but ALLOWED_EMAIL_DOMAINS "
429 "restricts sign-in by email domain."
430 ),
431 )
433 user_id: Final = SAMLAuthHandler._attribute_value(attributes, "SAML_ATTRIBUTE_USER_ID", ()) or name_id or email
434 if user_id is None:
435 raise HTTPException(
436 status_code=status.HTTP_401_UNAUTHORIZED,
437 detail="SAML assertion did not contain a usable subject (NameID) or email.",
438 )
440 first_name: Final = SAMLAuthHandler._attribute_value(
441 attributes, "SAML_ATTRIBUTE_FIRST_NAME", _FIRST_NAME_ATTRIBUTE_CANDIDATES
442 )
443 last_name: Final = SAMLAuthHandler._attribute_value(
444 attributes, "SAML_ATTRIBUTE_LAST_NAME", _LAST_NAME_ATTRIBUTE_CANDIDATES
445 )
446 role_values: Final = SAMLAuthHandler._attribute_values(
447 attributes, "SAML_ATTRIBUTE_ROLE", _ROLE_ATTRIBUTE_CANDIDATES
448 )
449 team_ids: Final = SAMLAuthHandler._attribute_values(
450 attributes, "SAML_ATTRIBUTE_TEAM_IDS", _TEAM_IDS_ATTRIBUTE_CANDIDATES
451 )
453 display_name: Final = " ".join(part for part in (first_name, last_name) if part) or email
455 verbose_proxy_logger.info(
456 "SAML login: subject=%s, email=%s, attributes=%s", user_id, email, list(attributes.keys())
457 )
459 try:
460 return CustomOpenID(
461 id=user_id,
462 email=email,
463 first_name=first_name,
464 last_name=last_name,
465 display_name=display_name,
466 picture=None,
467 provider="saml",
468 team_ids=team_ids,
469 user_role=get_litellm_user_role(role_values),
470 )
471 except ValidationError as e:
472 raise HTTPException(
473 status_code=status.HTTP_401_UNAUTHORIZED,
474 detail=f"SAML assertion contained an invalid subject or email: {e}",
475 )
477 @staticmethod
478 def _attribute_value(
479 attributes: dict[str, list[str]],
480 env_override: str,
481 candidates: tuple[str, ...],
482 ) -> str | None:
483 values: Final = SAMLAuthHandler._attribute_values(attributes, env_override, candidates)
484 return values[0] if values else None
486 @staticmethod
487 def _attribute_values(
488 attributes: dict[str, list[str]],
489 env_override: str,
490 candidates: tuple[str, ...],
491 ) -> list[str]:
492 override: Final = SAMLAuthHandler._env(env_override)
493 keys: Final = (override, *candidates) if override else candidates
494 for key in keys:
495 values = attributes.get(key)
496 if values:
497 return [v for v in values if v]
498 return []