Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/auth/route_checks.py: 43%
338 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:01 +0000
1import re
2from collections.abc import Collection
3from typing import Final
5from fastapi import HTTPException, Request, status
7from litellm._logging import verbose_proxy_logger
8from litellm.proxy._types import (
9 CommonProxyErrors,
10 KeyManagementRoutes,
11 LiteLLM_UserTable,
12 LiteLLMRoutes,
13 LitellmUserRoles,
14 UserAPIKeyAuth,
15)
17from .auth_checks_organization import _user_is_org_admin
19# Management write routes denied to PROXY_ADMIN_VIEW_ONLY. Adding a new write
20# endpoint to a management router REQUIRES adding it here too — the surrounding
21# check falls through to "allow" if the route is not matched, which previously
22# let view-only admins call /team/block, /team/unblock, /key/bulk_update, etc.
23_PROXY_ADMIN_VIEW_ONLY_BLOCKED_ROUTES: Final = frozenset(
24 [
25 # user
26 "/user/new",
27 "/management/v1/users/bulk",
28 "/user/delete",
29 "/management/v1/users/bulk_delete",
30 "/user/bulk_update",
31 # team
32 "/team/new",
33 "/management/v1/teams/{team_id}/members/bulk_delete",
34 "/management/v1/teams/{team_id}/members/bulk_update",
35 "/team/update",
36 "/team/delete",
37 "/team/block",
38 "/team/unblock",
39 "/team/permissions_update",
40 "/team/permissions_bulk_update",
41 # model
42 "/model/new",
43 "/model/update",
44 "/model/delete",
45 # JWT key mapping
46 "/jwt/key/mapping/new",
47 "/jwt/key/mapping/update",
48 "/jwt/key/mapping/delete",
49 # key management — keep in sync with KeyManagementRoutes write entries
50 KeyManagementRoutes.KEY_GENERATE.value,
51 KeyManagementRoutes.KEY_UPDATE.value,
52 KeyManagementRoutes.KEY_DELETE.value,
53 KeyManagementRoutes.KEY_REGENERATE.value,
54 KeyManagementRoutes.KEY_GENERATE_SERVICE_ACCOUNT.value,
55 KeyManagementRoutes.KEY_BLOCK.value,
56 KeyManagementRoutes.KEY_UNBLOCK.value,
57 KeyManagementRoutes.KEY_BULK_UPDATE.value,
58 KeyManagementRoutes.TEAM_KEY_BULK_UPDATE.value,
59 ]
60)
62# Suffixes for `/key/{key_id}/...` path-parameterized write routes that the
63# enum templates with `{key_id}`. The blocklist above can't match templated
64# paths directly because the request route carries the resolved key id.
65_PROXY_ADMIN_VIEW_ONLY_BLOCKED_KEY_SUFFIXES: Final = ("/regenerate", "/reset_spend")
67_AUTH_ENFORCED_PASS_THROUGH_ROUTE_GROUPS: Final = frozenset(("openai_routes", "llm_api_routes"))
70class RouteChecks:
71 @staticmethod
72 def should_call_route(
73 route: str,
74 valid_token: UserAPIKeyAuth,
75 request: Request | None = None,
76 ):
77 """
78 Check if management route is disabled and raise exception
79 """
80 try:
81 from litellm_enterprise.proxy.auth.route_checks import EnterpriseRouteChecks
83 EnterpriseRouteChecks.should_call_route(route=route)
84 except HTTPException as e:
85 raise e
86 except Exception:
87 pass
89 # Check if Virtual Key is allowed to call the route - Applies to all Roles
90 RouteChecks.is_virtual_key_allowed_to_call_route(route=route, valid_token=valid_token, request=request)
91 return True
93 @staticmethod
94 def is_virtual_key_allowed_to_call_route(
95 route: str,
96 valid_token: UserAPIKeyAuth,
97 request: Request | None = None,
98 ) -> bool:
99 """
100 Raises Exception if Virtual Key is not allowed to call the route
101 """
103 # Only check if valid_token.allowed_routes is set and is a list with at least one item
104 if valid_token.allowed_routes is None: 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true
105 return True
106 if not isinstance(valid_token.allowed_routes, list): 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true
107 return True
108 if len(valid_token.allowed_routes) == 0: 108 ↛ 111line 108 didn't jump to line 111 because the condition on line 108 was always true
109 return True
111 denied_auth_enforced_pass_through_route = False
113 # explicit check for allowed routes (exact match or prefix match)
114 for allowed_route in valid_token.allowed_routes:
115 if RouteChecks._route_matches_allowed_route(route=route, allowed_route=allowed_route):
116 return True
118 ## check if 'allowed_route' is a field name in LiteLLMRoutes
119 if any(allowed_route in LiteLLMRoutes._member_names_ for allowed_route in valid_token.allowed_routes):
120 for allowed_route in valid_token.allowed_routes:
121 if allowed_route in LiteLLMRoutes._member_names_:
122 if RouteChecks.check_route_access(
123 route=route,
124 allowed_routes=LiteLLMRoutes._member_map_[allowed_route].value,
125 ):
126 if (
127 allowed_route in _AUTH_ENFORCED_PASS_THROUGH_ROUTE_GROUPS
128 and RouteChecks.is_auth_enforced_pass_through_route(
129 route=route,
130 method=RouteChecks._get_request_method(request=request),
131 )
132 ):
133 if RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token):
134 return True
135 denied_auth_enforced_pass_through_route = True
136 else:
137 return True
139 ################################################
140 # For llm_api_routes, also check registered pass-through endpoints
141 ################################################
142 if allowed_route == "llm_api_routes":
143 if route == "/auto_router/session" and RouteChecks._get_request_method(request) == "GET":
144 return True
146 from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
147 InitPassThroughEndpointHelpers,
148 )
150 if InitPassThroughEndpointHelpers.is_registered_pass_through_route(route=route):
151 if RouteChecks.is_auth_enforced_pass_through_route(
152 route=route,
153 method=RouteChecks._get_request_method(request=request),
154 ):
155 if RouteChecks.check_passthrough_route_access(
156 route=route, user_api_key_dict=valid_token
157 ):
158 return True
159 denied_auth_enforced_pass_through_route = True
160 else:
161 return True
163 # Method-aware carve-out: allow GET on the two
164 # read-only MCP-server discovery endpoints
165 # (`/v1/mcp/server` and `/v1/mcp/server/{server_id}`)
166 # so virtual keys with allowed_routes=["llm_api_routes"]
167 # can list/inspect MCP servers. The GET handlers in
168 # mcp_management_endpoints.py sanitize the response
169 # for restricted virtual keys (stripping url,
170 # headers, env, credentials). POST/PUT/DELETE on
171 # these paths are admin-only management writes and
172 # are intentionally not covered.
173 if RouteChecks._is_get_mcp_server_discovery_route(route=route, request=request):
174 return True
176 # Agent registry CRUD moved from llm_api_routes into
177 # management_routes so DISABLE_LLM_API_ENDPOINTS stops
178 # blocking it. Keys configured with
179 # allowed_routes=["llm_api_routes"] before that split
180 # could reach these paths, so keep them reachable here;
181 # the handlers in agent_endpoints/endpoints.py still
182 # enforce proxy-admin on writes and scope reads by role.
183 if RouteChecks.check_route_access(
184 route=route,
185 allowed_routes=LiteLLMRoutes.agent_management_routes.value,
186 ):
187 return True
189 # check if wildcard pattern is allowed
190 for allowed_route in valid_token.allowed_routes:
191 if RouteChecks.route_matches_wildcard_pattern(route=route, pattern=allowed_route):
192 return True
194 if denied_auth_enforced_pass_through_route:
195 raise RouteChecks._auth_pass_through_denied_exception(route=route)
197 if valid_token.metadata.get("password_reset_required") is True:
198 raise HTTPException(
199 status_code=status.HTTP_403_FORBIDDEN,
200 detail=(
201 "This account's password must be changed before the session can be used: "
202 "it was either found in a known data breach or set by an admin. "
203 "Change it via POST /user/password/change (UI: /ui/change-password), then log in again."
204 ),
205 )
207 raise HTTPException(
208 status_code=status.HTTP_403_FORBIDDEN,
209 detail=f"Virtual key is not allowed to call this route. Only allowed to call routes: {valid_token.allowed_routes}. Tried to call route: {route}",
210 )
212 @staticmethod
213 def _mask_user_id(user_id: str) -> str:
214 """
215 Mask user_id to prevent leaking sensitive information in error messages
217 Args:
218 user_id (str): The user_id to mask
220 Returns:
221 str: Masked user_id showing only first 2 and last 2 characters
222 """
223 from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
225 if not user_id or len(user_id) <= 4:
226 return "***"
228 # Use SensitiveDataMasker with custom configuration for user_id
229 masker: Final = SensitiveDataMasker(visible_prefix=6, visible_suffix=2, mask_char="*")
231 return masker._mask_value(user_id)
233 @staticmethod
234 def _raise_admin_only_route_exception(
235 user_obj: LiteLLM_UserTable | None,
236 route: str,
237 ) -> None:
238 """
239 Raise exception for routes that require proxy admin access
241 Args:
242 user_obj (Optional[LiteLLM_UserTable]): The user object
243 route (str): The route being accessed
245 Raises:
246 Exception: With user role and masked user_id information
247 """
248 user_role = "unknown"
249 user_id = "unknown"
250 if user_obj is not None:
251 user_role = user_obj.user_role or "unknown"
252 user_id = user_obj.user_id or "unknown"
254 masked_user_id: Final = RouteChecks._mask_user_id(user_id)
255 raise Exception(
256 f"Only proxy admin can be used to generate, delete, update info for new keys/users/teams. Route={route}. Your role={user_role}. Your user_id={masked_user_id}"
257 )
259 @staticmethod
260 def non_proxy_admin_allowed_routes_check(
261 user_obj: LiteLLM_UserTable | None,
262 _user_role: LitellmUserRoles | None,
263 route: str,
264 request: Request,
265 valid_token: UserAPIKeyAuth,
266 request_data: dict,
267 ):
268 """
269 Checks if Non Proxy Admin User is allowed to access the route
270 """
272 # Check user has defined custom admin routes
273 RouteChecks.custom_admin_only_route_check(
274 route=route,
275 )
277 if RouteChecks.is_auth_enforced_pass_through_route(
278 route=route,
279 method=RouteChecks._get_request_method(request=request),
280 ):
281 RouteChecks._require_auth_pass_through_access(
282 route=route,
283 valid_token=valid_token,
284 jwt_team_allowed_routes=RouteChecks._jwt_team_allowed_routes(valid_token=valid_token),
285 )
286 elif RouteChecks.is_llm_api_route(route=route): 286 ↛ 288line 286 didn't jump to line 288 because the condition on line 286 was always true
287 pass
288 elif RouteChecks.is_info_route(route=route):
289 # check if user allowed to call an info route
290 if route == "/key/info":
291 # handled by function itself
292 pass
293 elif route == "/user/info":
294 # check if user can access this route
295 query_params: Final = request.query_params
296 user_id: Final = query_params.get("user_id")
297 verbose_proxy_logger.debug("user_id: %s & valid_token.user_id: %s", user_id, valid_token.user_id)
298 if (
299 user_id
300 and user_id != valid_token.user_id
301 and _user_role != LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value
302 ):
303 raise HTTPException(
304 status_code=status.HTTP_403_FORBIDDEN,
305 detail=f"key not allowed to access this user's info. user_id={user_id}, key's user_id={valid_token.user_id}",
306 )
307 elif route == "/v2/user/info":
308 # handled by the endpoint itself (full RBAC in handler)
309 pass
310 elif route == "/model/info":
311 # /model/info just shows models user has access to
312 pass
313 elif route == "/team/info":
314 pass # handled by function itself
315 elif (
316 route in LiteLLMRoutes.global_spend_tracking_routes.value
317 and getattr(valid_token, "permissions", None) is not None
318 and "get_spend_routes" in getattr(valid_token, "permissions", [])
319 ):
320 pass
321 elif _user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value:
322 RouteChecks._check_proxy_admin_viewer_access(
323 route=route,
324 _user_role=_user_role,
325 request_data=request_data,
326 request=request,
327 )
328 elif (
329 _user_role == LitellmUserRoles.INTERNAL_USER.value
330 and RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.internal_user_routes.value)
331 or _user_is_org_admin(request_data=request_data, user_object=user_obj)
332 and RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.org_admin_allowed_routes.value)
333 or _user_role == LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value
334 and RouteChecks.check_route_access(
335 route=route,
336 allowed_routes=LiteLLMRoutes.internal_user_view_only_routes.value,
337 )
338 or RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.self_managed_routes.value)
339 ):
340 pass
341 elif route.startswith("/v1/mcp/") or route.startswith("/mcp-rest/"):
342 pass # authN/authZ handled by api itself
343 elif RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token) or (
344 valid_token.is_team_service_account
345 and RouteChecks.check_route_access(
346 route=route, allowed_routes=LiteLLMRoutes.team_service_account_key_routes.value
347 )
348 ):
349 pass
350 elif valid_token.allowed_routes is not None:
351 # check if route is in allowed_routes (exact match or prefix match)
352 route_allowed = False
353 for allowed_route in valid_token.allowed_routes:
354 if RouteChecks._route_matches_allowed_route(route=route, allowed_route=allowed_route):
355 route_allowed = True
356 break
358 if RouteChecks.route_matches_wildcard_pattern(route=route, pattern=allowed_route):
359 route_allowed = True
360 break
362 if not route_allowed:
363 RouteChecks._raise_admin_only_route_exception(user_obj=user_obj, route=route)
364 else:
365 RouteChecks._raise_admin_only_route_exception(user_obj=user_obj, route=route)
367 @staticmethod
368 def custom_admin_only_route_check(route: str):
369 from litellm.proxy.proxy_server import general_settings, premium_user
371 if "admin_only_routes" in general_settings: 371 ↛ 372line 371 didn't jump to line 372 because the condition on line 371 was never true
372 if premium_user is not True:
373 verbose_proxy_logger.error(
374 "Trying to use 'admin_only_routes' this is an Enterprise only feature. %s",
375 CommonProxyErrors.not_premium_user.value,
376 )
377 return
378 if route in general_settings["admin_only_routes"]:
379 raise HTTPException(
380 status_code=status.HTTP_403_FORBIDDEN,
381 detail=f"user not allowed to access this route. Route={route} is an admin only route",
382 )
384 @staticmethod
385 def is_llm_api_route(route: str) -> bool:
386 """
387 Helper to checks if provided route is an OpenAI route
390 Returns:
391 - True: if route is an OpenAI route
392 - False: if route is not an OpenAI route
393 """
394 # Ensure route is a string before performing checks
395 if not isinstance(route, str): 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true
396 return False
398 if route in LiteLLMRoutes.openai_routes.value:
399 return True
401 if route in LiteLLMRoutes.anthropic_routes.value:
402 return True
404 if route in LiteLLMRoutes.google_routes.value:
405 return True
407 if RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.mcp_inference_routes.value):
408 return True
410 if RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.agent_inference_routes.value):
411 return True
413 if route in LiteLLMRoutes.litellm_native_routes.value:
414 return True
416 # fuzzy match routes like "/v1/threads/thread_49EIN5QF32s4mH20M7GFKdlZ"
417 # Check for routes with placeholders or wildcard patterns
418 for openai_route in LiteLLMRoutes.openai_routes.value:
419 # Replace placeholders with regex pattern
420 # placeholders are written as "/threads/{thread_id}"
421 if "{" in openai_route:
422 if RouteChecks._route_matches_pattern(route=route, pattern=openai_route):
423 return True
424 # Check for wildcard patterns like "/containers/*"
425 if RouteChecks._is_wildcard_pattern(pattern=openai_route):
426 if RouteChecks.route_matches_wildcard_pattern(route=route, pattern=openai_route):
427 return True
429 # Check for Google routes with placeholders like "/v1beta/models/{model_name}:generateContent"
430 for google_route in LiteLLMRoutes.google_routes.value:
431 if "{" in google_route:
432 if RouteChecks._route_matches_pattern(route=route, pattern=google_route):
433 return True
435 # Check for Anthropic routes with placeholders
436 for anthropic_route in LiteLLMRoutes.anthropic_routes.value:
437 if "{" in anthropic_route:
438 if RouteChecks._route_matches_pattern(route=route, pattern=anthropic_route):
439 return True
441 if RouteChecks._is_azure_openai_route(route=route): 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true
442 return True
444 for _llm_passthrough_route in LiteLLMRoutes.mapped_pass_through_routes.value:
445 if route == _llm_passthrough_route or route.startswith(_llm_passthrough_route + "/"):
446 return True
447 return False
449 @staticmethod
450 def _is_get_mcp_server_discovery_route(route: str, request: Request | None) -> bool:
451 """
452 Returns True if `request` is a GET against one of the two read-only
453 MCP-server discovery paths:
455 - GET `/v1/mcp/server` (list)
456 - GET `/v1/mcp/server/{server_id}` (single server, single segment)
458 Multi-segment paths (`/v1/mcp/server/{id}/approve`, etc.) and any
459 non-GET method return False, so admin-only management writes on the
460 same path prefix are not reachable through this carve-out.
461 """
462 if request is None or request.method.upper() != "GET":
463 return False
464 if route == "/v1/mcp/server":
465 return True
466 prefix: Final = "/v1/mcp/server/"
467 if not route.startswith(prefix):
468 return False
469 remainder: Final = route[len(prefix) :]
470 return bool(remainder) and "/" not in remainder
472 @staticmethod
473 def is_management_route(route: str) -> bool:
474 """
475 Check if route is a management route
476 """
477 return RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.management_routes.value)
479 @staticmethod
480 def is_info_route(route: str) -> bool:
481 """
482 Check if route is an info route
483 """
484 return route in LiteLLMRoutes.info_routes.value
486 @staticmethod
487 def _is_azure_openai_route(route: str) -> bool:
488 """
489 Check if route is a route from AzureOpenAI SDK client
491 eg.
492 route='/openai/deployments/vertex_ai/gemini-1.5-flash/chat/completions'
493 """
494 # Ensure route is a string before attempting regex matching
495 if not isinstance(route, str): 495 ↛ 496line 495 didn't jump to line 496 because the condition on line 495 was never true
496 return False
497 # Add support for deployment and engine model paths
498 deployment_pattern: Final = r"^/openai/deployments/[^/]+/[^/]+/chat/completions$"
499 engine_pattern: Final = r"^/engines/[^/]+/chat/completions$"
501 if re.match(deployment_pattern, route) or re.match(engine_pattern, route): 501 ↛ 502line 501 didn't jump to line 502 because the condition on line 501 was never true
502 return True
503 return False
505 @staticmethod
506 def _route_matches_pattern(route: str, pattern: str) -> bool:
507 """
508 Check if route matches the pattern placed in proxy/_types.py
510 Example:
511 - pattern: "/threads/{thread_id}"
512 - route: "/threads/thread_49EIN5QF32s4mH20M7GFKdlZ"
513 - returns: True
516 - pattern: "/key/{token_id}/regenerate"
517 - route: "/key/regenerate/82akk800000000jjsk"
518 - returns: False, pattern is "/key/{token_id}/regenerate"
519 """
520 # Ensure route is a string before attempting regex matching
521 if not isinstance(route, str): 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 return False
524 def _placeholder_to_regex(match: re.Match) -> str:
525 placeholder: Final = match.group(0).strip("{}")
526 if not placeholder.endswith(":path"):
527 return r"[^/]+"
528 # A ":path" placeholder takes whatever the router's own path
529 # converter takes, slashes and colons alike, so an id spelled with
530 # either (or both) still matches the template it was mounted under.
531 #
532 # Unless the template puts a ":" literal of its own after the
533 # placeholder: the Google routes end in ":generateContent" and
534 # friends, and there the value has to stop before that suffix
535 # rather than swallow it and match a different verb.
536 #
537 # "[\s\S]" rather than ".", because "." stops at a newline and the
538 # path converter does not: a %0A anywhere in the value would leave
539 # the route unmatched here while still reaching the handler, which
540 # turns this gate into a bypass for the lists built on it.
541 return r"[^:]+" if ":" in match.string[match.end() :] else r"[\s\S]+"
543 pattern = re.sub(r"\{[^}]+\}", _placeholder_to_regex, pattern)
544 # Anchor the pattern to match the entire string
545 pattern = f"^{pattern}$"
546 if re.match(pattern, route):
547 return True
548 return False
550 @staticmethod
551 def _is_wildcard_pattern(pattern: str) -> bool:
552 """
553 Check if pattern is a wildcard pattern
554 """
555 return pattern.endswith("*")
557 @staticmethod
558 def route_matches_wildcard_pattern(route: str, pattern: str) -> bool:
559 """
560 Check if route matches the wildcard pattern
562 eg.
564 pattern: "/scim/v2/*"
565 route: "/scim/v2/Users"
566 - returns: True
568 pattern: "/scim/v2/*"
569 route: "/chat/completions"
570 - returns: False
573 pattern: "/scim/v2/*"
574 route: "/scim/v2/Users/123"
575 - returns: True
577 """
578 if pattern.endswith("*"):
579 # Get the prefix (everything before the wildcard)
580 prefix: Final = pattern[:-1]
581 return route.startswith(prefix)
582 else:
583 # If there's no wildcard, the pattern and route should match exactly
584 return route == pattern
586 @staticmethod
587 def _route_matches_allowed_route(route: str, allowed_route: str) -> bool:
588 """
589 Check if route matches the allowed_route pattern.
590 Supports both exact match and prefix match.
592 Examples:
593 - allowed_route="/fake-openai-proxy-6", route="/fake-openai-proxy-6" -> True (exact match)
594 - allowed_route="/fake-openai-proxy-6", route="/fake-openai-proxy-6/v1/chat/completions" -> True (prefix match)
595 - allowed_route="/fake-openai-proxy-6", route="/fake-openai-proxy-600" -> False (not a valid prefix)
597 Args:
598 route: The actual route being accessed
599 allowed_route: The allowed route pattern
601 Returns:
602 bool: True if route matches (exact or prefix), False otherwise
603 """
604 # Exact match
605 if route == allowed_route:
606 return True
607 # Prefix match - ensure we add "/" to prevent false matches like /fake-openai-proxy-600
608 if route.startswith(allowed_route + "/"):
609 return True
610 return False
612 @staticmethod
613 def check_route_access(route: str, allowed_routes: Collection[str]) -> bool:
614 """
615 Check if a route has access by checking both exact matches and patterns
617 Args:
618 route (str): The route to check
619 allowed_routes (Sequence): Allowed routes/patterns
621 Returns:
622 bool: True if route is allowed, False otherwise
623 """
624 #########################################################
625 # exact match route is in allowed_routes
626 #########################################################
627 if route in allowed_routes:
628 return True
630 #########################################################
631 # wildcard match route is in allowed_routes
632 # e.g calling /anthropic/v1/messages is allowed if allowed_routes has /anthropic/*
633 #########################################################
634 if any( 634 ↛ 639line 634 didn't jump to line 639 because the condition on line 634 was never true
635 RouteChecks.route_matches_wildcard_pattern(route=route, pattern=allowed_route)
636 for allowed_route in allowed_routes
637 if RouteChecks._is_wildcard_pattern(pattern=allowed_route)
638 ):
639 return True
641 #########################################################
642 # pattern match route is in allowed_routes
643 # pattern: "/threads/{thread_id}"
644 # route: "/threads/thread_49EIN5QF32s4mH20M7GFKdlZ"
645 # returns: True
646 #########################################################
647 if any( # Check pattern match
648 RouteChecks._route_matches_pattern(route=route, pattern=allowed_route) for allowed_route in allowed_routes
649 ):
650 return True
652 return False
654 @staticmethod
655 def _get_request_method(request: Request | None) -> str | None:
656 if request is None: 656 ↛ 657line 656 didn't jump to line 657 because the condition on line 656 was never true
657 return None
659 try:
660 method: Final = request.method
661 except (AttributeError, KeyError):
662 return None
663 if not isinstance(method, str): 663 ↛ 664line 663 didn't jump to line 664 because the condition on line 663 was never true
664 return None
666 return method.upper()
668 @staticmethod
669 def is_auth_enforced_pass_through_route(route: str, method: str | None = None) -> bool:
670 """
671 True for config/DB pass-through endpoints registered with auth=true.
673 These routes are injected into ``openai_routes`` for spend/budget hooks but
674 must not inherit blanket ``openai_routes`` RBAC; access is gated by
675 ``allowed_passthrough_routes`` on the key or team.
676 """
677 from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
678 InitPassThroughEndpointHelpers,
679 )
681 route_info: Final = InitPassThroughEndpointHelpers.get_registered_pass_through_route(route=route, method=method)
682 if route_info is None: 682 ↛ 683line 682 didn't jump to line 683 because the condition on line 682 was never true
683 return False
684 return route_info.get("auth") is True
686 @staticmethod
687 def _auth_pass_through_denied_exception(route: str) -> HTTPException:
688 return HTTPException(
689 status_code=status.HTTP_403_FORBIDDEN,
690 detail=(
691 f"Key/team not allowed to access passthrough route {route}. "
692 "Configure `allowed_passthrough_routes` on the team or key."
693 ),
694 )
696 @staticmethod
697 def jwt_team_routes_grant_pass_through(route: str, team_allowed_routes: Collection[str]) -> bool:
698 """
699 Explicit paths and trailing-wildcard prefixes grant auth=true pass-through. Blanket grants never do:
700 a named route group like ``openai_routes`` is only ever compared as a path, and an entry that names
701 no path segment (``*``, ``/*``) is skipped.
702 """
703 return any(
704 RouteChecks.route_matches_wildcard_pattern(route=route, pattern=allowed_route)
705 for allowed_route in team_allowed_routes
706 if allowed_route.rstrip("*").strip("/")
707 )
709 @staticmethod
710 def _jwt_team_allowed_routes(valid_token: UserAPIKeyAuth) -> Collection[str]:
711 """``team_allowed_routes`` for team tokens built by JWT auth; JWT-mapped virtual keys stay key-scoped."""
712 if valid_token.jwt_claims is None or valid_token.token is not None or valid_token.team_id is None: 712 ↛ 715line 712 didn't jump to line 715 because the condition on line 712 was always true
713 return ()
715 from litellm.proxy.proxy_server import jwt_handler
717 return jwt_handler.litellm_jwtauth.team_allowed_routes
719 @staticmethod
720 def _require_auth_pass_through_access(
721 route: str,
722 valid_token: UserAPIKeyAuth,
723 jwt_team_allowed_routes: Collection[str] = (),
724 ) -> None:
725 """
726 Require an explicit grant for auth=true pass-through: ``allowed_passthrough_routes`` on the
727 key or team, or an explicit JWT ``team_allowed_routes`` entry.
728 """
729 if RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token): 729 ↛ 730line 729 didn't jump to line 730 because the condition on line 729 was never true
730 return
731 if RouteChecks.jwt_team_routes_grant_pass_through(route=route, team_allowed_routes=jwt_team_allowed_routes): 731 ↛ 732line 731 didn't jump to line 732 because the condition on line 731 was never true
732 return
733 raise RouteChecks._auth_pass_through_denied_exception(route=route)
735 @staticmethod
736 def check_passthrough_route_access(route: str, user_api_key_dict: UserAPIKeyAuth) -> bool:
737 """
738 Check if route is a passthrough route.
739 Supports both exact match and prefix match.
740 """
741 metadata: Final = user_api_key_dict.metadata
742 team_metadata: Final = user_api_key_dict.team_metadata or {}
743 if metadata is None and team_metadata is None: 743 ↛ 744line 743 didn't jump to line 744 because the condition on line 743 was never true
744 return False
745 if "allowed_passthrough_routes" not in metadata and "allowed_passthrough_routes" not in team_metadata: 745 ↛ 747line 745 didn't jump to line 747 because the condition on line 745 was always true
746 return False
747 if (
748 metadata.get("allowed_passthrough_routes") is None
749 and team_metadata.get("allowed_passthrough_routes") is None
750 ):
751 return False
753 allowed_passthrough_routes: Final = (
754 metadata.get("allowed_passthrough_routes") or team_metadata.get("allowed_passthrough_routes") or []
755 )
757 # Check if route matches any allowed passthrough route (exact or prefix match)
758 for allowed_route in allowed_passthrough_routes:
759 if RouteChecks._route_matches_allowed_route(route=route, allowed_route=allowed_route):
760 return True
762 return False
764 @staticmethod
765 def _is_assistants_api_request(request: Request) -> bool:
766 """
767 Returns True if `thread` or `assistant` is in the request path
769 Args:
770 request (Request): The request object
772 Returns:
773 bool: True if `thread` or `assistant` is in the request path, False otherwise
774 """
775 # Inline import — auth_utils participates in a proxy import cycle.
776 from .auth_utils import get_request_route # noqa: PLC0415
778 route: Final = get_request_route(request)
779 if "thread" in route or "assistant" in route:
780 return True
781 return False
783 @staticmethod
784 def is_generate_content_route(route: str) -> bool:
785 """
786 Returns True if this is a google generateContent or streamGenerateContent route
788 These routes from google allow passing key=api_key in the query params
789 """
790 if "generateContent" in route:
791 return True
792 if "streamGenerateContent" in route:
793 return True
794 return False
796 # HTTP methods that are intrinsically read-only and therefore safe to
797 # default-allow for PROXY_ADMIN_VIEW_ONLY. Anything else (POST/PUT/PATCH/
798 # DELETE) is treated as a write attempt and goes through the explicit
799 # write-allowlist below.
800 _SAFE_HTTP_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
802 # Explicit write routes that PROXY_ADMIN_VIEW_ONLY must NEVER call. The
803 # role-principle is "no writes, ever" — the management_routes list is the
804 # authoritative source for which non-llm routes are writes; we just need
805 # to filter out the read endpoints (info / list) that share the prefix.
806 # A cleaner approach is to denylist by HTTP verb (POST/PUT/PATCH/DELETE);
807 # this block stays as a backstop in case a write is implemented as GET.
808 _ADMIN_VIEWER_BLOCKED_WRITE_ROUTES = frozenset(
809 [
810 "/user/new",
811 "/management/v1/users/bulk",
812 "/user/delete",
813 "/management/v1/users/bulk_delete",
814 "/user/bulk_update",
815 "/team/new",
816 "/management/v1/teams/{team_id}/members/bulk_delete",
817 "/management/v1/teams/{team_id}/members/bulk_update",
818 "/team/update",
819 "/team/delete",
820 "/model/new",
821 "/model/update",
822 "/model/delete",
823 "/key/generate",
824 "/key/delete",
825 "/key/update",
826 "/key/regenerate",
827 "/key/service-account/generate",
828 "/key/block",
829 "/key/unblock",
830 "/team/key/bulk_update",
831 ]
832 )
834 @staticmethod
835 def _check_proxy_admin_viewer_access(
836 route: str,
837 _user_role: str,
838 request_data: dict,
839 request: Request | None = None,
840 ) -> None:
841 """
842 Check access for PROXY_ADMIN_VIEW_ONLY role.
844 Admin Viewer follows a read-parity-with-Proxy-Admin rule: anything Proxy
845 Admin can read/list/get, Admin Viewer can read/list/get. The only
846 exclusions are cost-incurring inference routes (Playground, /chat/
847 completions, etc.) and any state-mutating request.
849 Implementation:
850 1. LLM/inference routes → 403 (cost-incurring).
851 2. Safe HTTP method (GET/HEAD/OPTIONS) → allow by default. This is
852 the read-parity guarantee — every new GET endpoint added anywhere
853 in the codebase is automatically readable by Admin Viewer
854 without needing to remember to add it to an allowlist.
855 3. Unsafe HTTP method (POST/PUT/PATCH/DELETE):
856 - Allow `/user/update` only when restricted to user_email.
857 - Allow `/user/password/change` (endpoint only writes the caller's own row).
858 - Block all explicit writes in `_ADMIN_VIEWER_BLOCKED_WRITE_ROUTES`.
859 - Otherwise allow only if the route is in admin_viewer_routes /
860 global_spend_tracking_routes (legacy explicit-allow set).
861 - Else 403.
862 """
863 if RouteChecks.is_llm_api_route(route=route):
864 raise HTTPException(
865 status_code=status.HTTP_403_FORBIDDEN,
866 detail=f"user not allowed to access this OpenAI routes, role= {_user_role}",
867 )
869 # Check if this is a write operation on management routes
870 if RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.management_routes.value):
871 # For management routes, only allow read operations or specific allowed updates
872 if route == "/user/update":
873 # Check the Request params are valid for PROXY_ADMIN_VIEW_ONLY
874 if request_data is not None and isinstance(request_data, dict):
875 _params_updated: Final = request_data.keys()
876 for param in _params_updated:
877 if param != "user_email":
878 raise HTTPException(
879 status_code=status.HTTP_403_FORBIDDEN,
880 detail=f"user not allowed to access this route, role= {_user_role}. Trying to access: {route} and updating invalid param: {param}. only user_email can be updated",
881 )
882 elif RouteChecks.check_route_access(route=route, allowed_routes=_PROXY_ADMIN_VIEW_ONLY_BLOCKED_ROUTES) or (
883 route.startswith("/key/") and route.endswith(_PROXY_ADMIN_VIEW_ONLY_BLOCKED_KEY_SUFFIXES)
884 ):
885 # Block write operations for PROXY_ADMIN_VIEW_ONLY
886 raise HTTPException(
887 status_code=status.HTTP_403_FORBIDDEN,
888 detail=f"user not allowed to access this route, role= {_user_role}. Trying to access: {route}",
889 )
890 # Allow read operations on management routes (like /user/info, /team/info, /model/info)
891 method: Final = request.method.upper() if request is not None else "GET"
892 is_safe_method: Final = method in RouteChecks._SAFE_HTTP_METHODS
894 # ── Safe HTTP method: default-allow ──────────────────────────────
895 if is_safe_method:
896 return
898 # ── Unsafe HTTP method: explicit checks ──────────────────────────
899 # Allow `/user/update` for self-service email change.
900 if route == "/user/update":
901 if request_data is not None and isinstance(request_data, dict):
902 for param in request_data:
903 if param != "user_email":
904 raise HTTPException(
905 status_code=status.HTTP_403_FORBIDDEN,
906 detail=(
907 f"user not allowed to access this route, role= {_user_role}. "
908 f"Trying to access: {route} and updating invalid param: {param}. "
909 "only user_email can be updated"
910 ),
911 )
912 return
914 # Self-service password change; the endpoint only writes the caller's own row.
915 if route == "/user/password/change":
916 return
918 # Self-service logout; the endpoint only revokes the caller's own session key.
919 if route == "/session/logout":
920 return
922 # Hard-block known write routes regardless of HTTP method (defensive
923 # — these are POSTs in practice, but pinning them here protects
924 # against future GET-shaped writes).
925 if RouteChecks.check_route_access(
926 route=route, allowed_routes=RouteChecks._ADMIN_VIEWER_BLOCKED_WRITE_ROUTES
927 ) or (route.startswith("/key/") and route.endswith("/regenerate")):
928 raise HTTPException(
929 status_code=status.HTTP_403_FORBIDDEN,
930 detail=f"user not allowed to access this route, role= {_user_role}. Trying to access: {route}",
931 )
933 # Legacy explicit-allow sets (kept for routes that are POST but
934 # semantically read-only, e.g. /spend/calculate). Both admin_viewer_routes
935 # and global_spend_tracking_routes are reads/listings.
936 if RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.admin_viewer_routes.value):
937 return
938 if RouteChecks.check_route_access(route=route, allowed_routes=LiteLLMRoutes.global_spend_tracking_routes.value):
939 return
941 # NOTE: We intentionally do NOT fall back to allowing all
942 # `management_routes`. That set is a mix of reads (info/list — handled
943 # via the safe-method branch above) and writes (`/team/block`,
944 # `/team/permissions_update`, `/jwt/key/mapping/{new,update,delete}`,
945 # `/key/bulk_update`, `/key/{id}/reset_spend`). A blanket allow would
946 # let Admin Viewer POST these write endpoints — violating the
947 # "no writes, ever" rule. Default-deny instead.
948 raise HTTPException(
949 status_code=status.HTTP_403_FORBIDDEN,
950 detail=f"user not allowed to access this route, role= {_user_role}. Trying to access: {route}",
951 )