From 225fd3bcdfb7d16e71fd72bfb8ff9c56e81fe21e Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 7 Oct 2026 16:19:53 -0700 Subject: [PATCH] feat(proxy): add denied_passthrough_routes deny list for custom pass-through endpoints (#44924) * feat(proxy): add denied_passthrough_routes deny list for custom pass-through endpoints Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(proxy): harden denied_passthrough_routes against non-admin clears and dot-segment paths Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * chore(ui): regenerate schema.d.ts for typed denied_passthrough_routes Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(proxy): match trailing-slash deny entries, block null metadata from dropping denies Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(proxy): close bulk update, encoded ?/# and ordering gaps in passthrough deny list Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(proxy): check denied pass-through routes against the path the forwarder sends upstream Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(proxy): type matching_denied_passthrough_route metadata mappings * fix(proxy): treat a `/` deny entry as denying every pass-through route Also types the new deny-list tests and drops their get_server_root_path mock in favour of unsetting SERVER_ROOT_PATH. * test(proxy): type the deny-list tests and drop unrelated test reformatting --------- Co-authored-by: mrinal Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Co-authored-by: Yucheng He --- litellm/proxy/_types.py | 4 + litellm/proxy/auth/handle_jwt.py | 25 +- litellm/proxy/auth/route_checks.py | 73 +++- .../management_endpoints/common_utils.py | 90 ++++- .../key_management_endpoints.py | 31 +- .../management_endpoints/team_endpoints.py | 9 +- .../pass_through_endpoints.py | 91 +++-- .../test_passthrough_route_denylist.py | 333 ++++++++++++++++++ tests/unit/proxy/auth/test_handle_jwt.py | 51 +++ tests/unit/proxy/auth/test_route_checks.py | 205 +++++++++++ .../management_endpoints/test_common_utils.py | 115 +++++- .../test_key_management_endpoints.py | 31 ++ .../test_pass_through_endpoints.py | 93 ++++- ui/litellm-dashboard/src/lib/http/schema.d.ts | 20 ++ 14 files changed, 1093 insertions(+), 78 deletions(-) create mode 100644 tests/integration/authorization/test_passthrough_route_denylist.py diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index a314fdbf060..1627c5b63ec 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -1365,6 +1365,7 @@ class KeyRequestBase(GenerateRequestBase): enforced_params: list[str] | None = None allowed_routes: list | None = [] allowed_passthrough_routes: list | None = None + denied_passthrough_routes: list[str] | None = None allowed_vector_store_indexes: list[AllowedVectorStoreIndexItem] | None = None rpm_limit_type: Literal["guaranteed_throughput", "best_effort_throughput", "dynamic"] | None = ( None # raise an error if 'guaranteed_throughput' is set and we're overallocating rpm @@ -2212,6 +2213,7 @@ class NewTeamRequest(TeamBase): prompts: list[str] | None = None object_permission: LiteLLM_ObjectPermissionBase | None = None allowed_passthrough_routes: list | None = None + denied_passthrough_routes: list[str] | None = None disable_global_guardrails: bool | None = None secret_manager_settings: dict | None = None model_rpm_limit: dict[str, int] | None = None @@ -2293,6 +2295,7 @@ class UpdateTeamRequest(LiteLLMPydanticObjectBase): team_member_tpm_limit: int | None = None team_member_key_duration: str | None = None allowed_passthrough_routes: list | None = None + denied_passthrough_routes: list[str] | None = None secret_manager_settings: dict | None = None prompts: list[str] | None = None model_rpm_limit: dict[str, int] | None = None @@ -5104,6 +5107,7 @@ LiteLLM_ManagementEndpoint_MetadataFields_Premium: Final = [ "logging", "secret_manager_settings", "allowed_passthrough_routes", + "denied_passthrough_routes", ] # Metadata keys that are immutable once set: preserved when an update omits them, diff --git a/litellm/proxy/auth/handle_jwt.py b/litellm/proxy/auth/handle_jwt.py index d600fb3486b..a57eacdd133 100644 --- a/litellm/proxy/auth/handle_jwt.py +++ b/litellm/proxy/auth/handle_jwt.py @@ -1648,6 +1648,10 @@ class JWTAuthManager: ): return True + team_metadata: Final = (team_object.metadata or {}) if team_object else {} + if RouteChecks.matching_denied_passthrough_route(route=route, metadata_sources=(team_metadata,)) is not None: + return False + if RouteChecks.jwt_team_routes_grant_pass_through(route=route, team_allowed_routes=team_allowed_routes): return True @@ -1655,11 +1659,16 @@ class JWTAuthManager: # so beyond the JWT config grant above, only the selected team's metadata grants access. return RouteChecks.check_passthrough_route_access( route=route, - user_api_key_dict=UserAPIKeyAuth(team_metadata=(team_object.metadata or {}) if team_object else {}), + user_api_key_dict=UserAPIKeyAuth(team_metadata=team_metadata), ) @staticmethod - def _raise_team_passthrough_route_denial(route: str) -> None: + def _raise_team_passthrough_route_denial(route: str, team_object: LiteLLM_TeamTable | None) -> None: + denied_route: Final = RouteChecks.matching_denied_passthrough_route( + route=route, metadata_sources=((team_object.metadata if team_object else None),) + ) + if denied_route is not None: + raise RouteChecks.passthrough_route_denied_exception(route=route, denied_route=denied_route) raise HTTPException( status_code=403, detail=( @@ -1683,7 +1692,7 @@ class JWTAuthManager: """Find first team with access to the requested model""" from litellm.proxy.proxy_server import llm_router - denied_auth_enforced_pass_through_route = False + denied_pass_through_team: LiteLLM_TeamTable | None = None if not team_ids: if ( @@ -1733,7 +1742,7 @@ class JWTAuthManager: team_allowed_routes=jwt_handler.litellm_jwtauth.team_allowed_routes, ): is_allowed = False - denied_auth_enforced_pass_through_route = True + denied_pass_through_team = team_object verbose_proxy_logger.debug( "JWT team route check: team_id=%s, route=%s, is_allowed=%s", team_id, route, is_allowed ) @@ -1742,8 +1751,8 @@ class JWTAuthManager: except Exception: continue - if denied_auth_enforced_pass_through_route: - JWTAuthManager._raise_team_passthrough_route_denial(route=route) + if denied_pass_through_team is not None: + JWTAuthManager._raise_team_passthrough_route_denial(route=route, team_object=denied_pass_through_team) if requested_model and (any_claim_team_resolved or not jwt_handler.litellm_jwtauth.team_claim_fallback): # Claim resolved but no model access, or fallback disabled — deny. @@ -2788,7 +2797,7 @@ class JWTAuthManager: request_method=request_method, team_allowed_routes=handler.litellm_jwtauth.team_allowed_routes, ): - JWTAuthManager._raise_team_passthrough_route_denial(route=route) + JWTAuthManager._raise_team_passthrough_route_denial(route=route, team_object=selected_team_object) # Extract alias fields for resolution (if configured) org_alias: Final = handler.get_org_alias(token=jwt_valid_token, default_value=None) @@ -2858,7 +2867,7 @@ class JWTAuthManager: request_method=request_method, team_allowed_routes=handler.litellm_jwtauth.team_allowed_routes, ): - JWTAuthManager._raise_team_passthrough_route_denial(route=route) + JWTAuthManager._raise_team_passthrough_route_denial(route=route, team_object=team_object) elif selected_team_id is None: ( team_id, diff --git a/litellm/proxy/auth/route_checks.py b/litellm/proxy/auth/route_checks.py index 8b2e0beb10f..4a4913fd65b 100644 --- a/litellm/proxy/auth/route_checks.py +++ b/litellm/proxy/auth/route_checks.py @@ -1,6 +1,7 @@ +import itertools import re -from collections.abc import Collection -from typing import Final +from collections.abc import Collection, Iterable, Mapping +from typing import Final, cast from fastapi import HTTPException, Request, status @@ -693,6 +694,70 @@ class RouteChecks: ), ) + @staticmethod + def _route_matches_denied_route(route: str, denied_route: str) -> bool: + """A `/` entry denies every route, since every route sits under the root.""" + normalized_denied_route: Final = denied_route.rstrip("/") or "/" + return ( + normalized_denied_route == "/" + or RouteChecks._route_matches_allowed_route(route=route, allowed_route=normalized_denied_route) + or RouteChecks.route_matches_wildcard_pattern(route=route, pattern=denied_route) + ) + + @staticmethod + def matching_denied_passthrough_route( + route: str, metadata_sources: Iterable[Mapping[str, object] | None] + ) -> str | None: + """ + First ``denied_passthrough_routes`` entry across ``metadata_sources`` that matches ``route``. + Unlike the allowlist (key list, else team list), every source's deny list applies. + """ + denied_routes: Final = tuple( + itertools.chain.from_iterable( + cast( # cast-ok: management endpoints validate this metadata key as a list of route strings on write + "list[str]", (metadata or {}).get("denied_passthrough_routes") or [] + ) + for metadata in metadata_sources + ) + ) + if not denied_routes: + return None + from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( + InitPassThroughEndpointHelpers, + ) + + forwarded_routes: Final = InitPassThroughEndpointHelpers.forwarded_routes(route) + return next( + ( + denied_route + for denied_route in denied_routes + if any( + RouteChecks._route_matches_denied_route(route=candidate, denied_route=denied_route) + for candidate in forwarded_routes + ) + ), + None, + ) + + @staticmethod + def passthrough_route_denied_exception(route: str, denied_route: str) -> HTTPException: + return HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=( + f"Key/team denied access to passthrough route {route}. " + f"Matched `{denied_route}` in `denied_passthrough_routes`." + ), + ) + + @staticmethod + def _raise_if_passthrough_route_denied(route: str, valid_token: UserAPIKeyAuth) -> None: + denied_route: Final = RouteChecks.matching_denied_passthrough_route( + route=route, + metadata_sources=(valid_token.metadata, valid_token.team_metadata), + ) + if denied_route is not None: + raise RouteChecks.passthrough_route_denied_exception(route=route, denied_route=denied_route) + @staticmethod def jwt_team_routes_grant_pass_through(route: str, team_allowed_routes: Collection[str]) -> bool: """ @@ -724,8 +789,10 @@ class RouteChecks: ) -> None: """ Require an explicit grant for auth=true pass-through: ``allowed_passthrough_routes`` on the - key or team, or an explicit JWT ``team_allowed_routes`` entry. + key or team, or an explicit JWT ``team_allowed_routes`` entry. A key or team + ``denied_passthrough_routes`` match blocks the route even when one of those grants it. """ + RouteChecks._raise_if_passthrough_route_denied(route=route, valid_token=valid_token) if RouteChecks.check_passthrough_route_access(route=route, user_api_key_dict=valid_token): return if RouteChecks.jwt_team_routes_grant_pass_through(route=route, team_allowed_routes=jwt_team_allowed_routes): diff --git a/litellm/proxy/management_endpoints/common_utils.py b/litellm/proxy/management_endpoints/common_utils.py index be2423fb596..6d458feef57 100644 --- a/litellm/proxy/management_endpoints/common_utils.py +++ b/litellm/proxy/management_endpoints/common_utils.py @@ -4,7 +4,7 @@ from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Optional, Union from fastapi import HTTPException, status -from pydantic import BaseModel +from pydantic import BaseModel, TypeAdapter, ValidationError # Defined above the `litellm.proxy.*` imports so the name is bound even when @@ -150,31 +150,89 @@ def require_caller_user_id_for_non_admin( return user_api_key_dict.user_id +_ROUTE_LIST: Final = TypeAdapter(list[str] | None) + + +def _passthrough_routes_permission_error(field: str, entity: str) -> HTTPException: + return HTTPException( + status_code=403, + detail={"error": f"Only proxy admins can set `{field}` on a {entity}."}, + ) + + def _check_passthrough_routes_caller_permission( - data: BaseModel, + data: BaseModel | None, + user_api_key_dict: UserAPIKeyAuth, + *, + entity: str = "key", + existing_metadata: Mapping[str, object] | None = None, +) -> None: + """ + Only proxy admins may set `allowed_passthrough_routes` or `denied_passthrough_routes` + (top-level or under `metadata`), since the runtime route checker reads both from key and + team metadata. + """ + check_allowed_passthrough_routes_caller_permission(data, user_api_key_dict, entity=entity) + check_denied_passthrough_routes_caller_permission( + data, user_api_key_dict, entity=entity, existing_metadata=existing_metadata + ) + + +def check_allowed_passthrough_routes_caller_permission( + data: BaseModel | None, user_api_key_dict: UserAPIKeyAuth, *, entity: str = "key", ) -> None: - """ - Only proxy admins may set `allowed_passthrough_routes` (top-level or under - `metadata`) — it short-circuits the role-based route gate, so keys and teams - must be gated identically. - """ + if data is None: + return + metadata: Final = getattr(data, "metadata", None) + if isinstance(metadata, dict): + try: + _ROUTE_LIST.validate_python(metadata.get("denied_passthrough_routes")) + except ValidationError as e: + raise HTTPException( + status_code=400, + detail={"error": "`metadata.denied_passthrough_routes` must be a list of route strings."}, + ) from e # view-only admins excluded by design; blocked upstream from writes anyway if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value: return if getattr(data, "allowed_passthrough_routes", None): - raise HTTPException( - status_code=403, - detail={"error": f"Only proxy admins can set `allowed_passthrough_routes` on a {entity}."}, - ) - metadata: Final = getattr(data, "metadata", None) + raise _passthrough_routes_permission_error("allowed_passthrough_routes", entity) if isinstance(metadata, dict) and metadata.get("allowed_passthrough_routes"): - raise HTTPException( - status_code=403, - detail={"error": f"Only proxy admins can set `metadata.allowed_passthrough_routes` on a {entity}."}, - ) + raise _passthrough_routes_permission_error("metadata.allowed_passthrough_routes", entity) + + +def check_denied_passthrough_routes_caller_permission( + data: BaseModel | None, + user_api_key_dict: UserAPIKeyAuth, + *, + entity: str = "key", + existing_metadata: Mapping[str, object] | None = None, +) -> None: + """ + A non-admin request must leave an existing deny list as it is: clearing it, or replacing + `metadata` without it, would widen access. The outcome depends on the stored deny list, so + run this only after the caller is known to be allowed to edit the object. + """ + if data is None or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value: + return + metadata: Final = getattr(data, "metadata", None) + existing_denied: Final = (existing_metadata or {}).get("denied_passthrough_routes") or None + if ( + "denied_passthrough_routes" in data.model_fields_set + and (getattr(data, "denied_passthrough_routes", None) or None) != existing_denied + ): + raise _passthrough_routes_permission_error("denied_passthrough_routes", entity) + if _metadata_changes_denied_routes(data, metadata, existing_denied): + raise _passthrough_routes_permission_error("metadata.denied_passthrough_routes", entity) + + +def _metadata_changes_denied_routes(data: BaseModel, metadata: object, existing_denied: object) -> bool: + if isinstance(metadata, dict): + return (metadata.get("denied_passthrough_routes") or None) != existing_denied + return metadata is None and "metadata" in data.model_fields_set and existing_denied is not None def _check_disable_global_guardrails_caller_permission( diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index 72844de2549..6063ee21250 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -94,6 +94,8 @@ from litellm.proxy.management_endpoints.common_utils import ( _set_object_metadata_field, _team_member_has_permission, _user_has_admin_view, + check_allowed_passthrough_routes_caller_permission, + check_denied_passthrough_routes_caller_permission, validate_budget_duration, validate_finite_spend, ) @@ -2012,6 +2014,7 @@ async def generate_key_fn( - prompts: Optional[List[str]] - List of prompts that the key is allowed to use. - allowed_routes: Optional[list] - List of allowed routes for the key. Store the actual route or store a wildcard pattern for a set of routes. Example - ["/chat/completions", "/embeddings", "/keys/*"] - allowed_passthrough_routes: Optional[list] - List of allowed pass through endpoints for the key. Store the actual endpoint or store a wildcard pattern for a set of endpoints. Example - ["/my-custom-endpoint"]. Use this instead of allowed_routes, if you just want to specify which pass through endpoints the key can access, without specifying the routes. If allowed_routes is specified, allowed_pass_through_endpoints is ignored. + - denied_passthrough_routes: Optional[list] - List of pass through routes the key may not call, even if allowed by `allowed_passthrough_routes` or `allowed_routes`. Matches exact paths, path prefixes, and trailing `*` wildcards. Applies together with the team's `denied_passthrough_routes`. Example - ["/my-custom-endpoint/admin"]. - object_permission: Optional[LiteLLM_ObjectPermissionBase] - key-specific object permission. Example - {"vector_stores": ["vector_store_1", "vector_store_2"], "agents": ["agent_1", "agent_2"], "agent_access_groups": ["dev_group"]}. IF null or {} then no object permission. - key_type: Optional[str] - Type of key that determines default allowed routes. Options: "llm_api" (can call LLM API routes), "management" (can call management routes), "read_only" (can only call info/read routes), "default" (uses default allowed routes). Defaults to "default". - prompts: Optional[List[str]] - List of allowed prompts for the key. If specified, the key will only be able to use these specific prompts. @@ -2774,6 +2777,11 @@ async def _process_single_key_update( existing_key_row=existing_key_row, user_api_key_cache=user_api_key_cache, ) + check_denied_passthrough_routes_caller_permission( + update_key_request, + user_api_key_dict, + existing_metadata=existing_key_row.metadata, # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # LiteLLM_VerificationToken.metadata is a bare dict + ) # Custom key update hook if user_custom_key_update is not None: @@ -3091,10 +3099,7 @@ async def _validate_update_key_data( existing_key_row=existing_key_row, user_api_key_dict=user_api_key_dict, ) - _check_passthrough_routes_caller_permission( - data=data, - user_api_key_dict=user_api_key_dict, - ) + check_allowed_passthrough_routes_caller_permission(data, user_api_key_dict) _check_permissions_caller_permission( data=data, user_api_key_dict=user_api_key_dict, @@ -3240,6 +3245,11 @@ async def _validate_update_key_data( user_api_key_cache=user_api_key_cache, route=("/key/update (max_budget/spend)" if _is_budget_change else "/key/update"), ) + check_denied_passthrough_routes_caller_permission( + data, + user_api_key_dict, + existing_metadata=existing_key_row.metadata, # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # LiteLLM_VerificationToken.metadata is a bare dict + ) # Check team limits if key has a team_id (from request or existing key) team_obj: LiteLLM_TeamTableCachedObj | None = None @@ -3428,6 +3438,7 @@ async def update_key_fn( - temp_budget_expiry: Optional[str] - Expiry time for the temporary budget increase (Enterprise only). - allowed_routes: Optional[list] - List of allowed routes for the key. Store the actual route or store a wildcard pattern for a set of routes. Example - ["/chat/completions", "/embeddings", "/keys/*"] - allowed_passthrough_routes: Optional[list] - List of allowed pass through routes for the key. Store the actual route or store a wildcard pattern for a set of routes. Example - ["/my-custom-endpoint"]. Use this instead of allowed_routes, if you just want to specify which pass through routes the key can access, without specifying the routes. If allowed_routes is specified, allowed_passthrough_routes is ignored. + - denied_passthrough_routes: Optional[list] - List of pass through routes the key may not call, even if allowed by `allowed_passthrough_routes` or `allowed_routes`. Matches exact paths, path prefixes, and trailing `*` wildcards. Applies together with the team's `denied_passthrough_routes`. Example - ["/my-custom-endpoint/admin"]. - prompts: Optional[List[str]] - List of allowed prompts for the key. If specified, the key will only be able to use these specific prompts. - object_permission: Optional[LiteLLM_ObjectPermissionBase] - key-specific object permission. Example - {"vector_stores": ["vector_store_1", "vector_store_2"], "agents": ["agent_1", "agent_2"], "agent_access_groups": ["dev_group"]}. IF null or {} then no object permission. - auto_rotate: Optional[bool] - Whether this key should be automatically rotated @@ -3944,7 +3955,7 @@ async def bulk_update_team_keys( # Block metadata.allowed_passthrough_routes for non-admins — the runtime # route checker reads it from key/team metadata to grant passthrough. - _check_passthrough_routes_caller_permission(data=data.update_fields, user_api_key_dict=user_api_key_dict) + check_allowed_passthrough_routes_caller_permission(data.update_fields, user_api_key_dict) if not requested_tokens: raise HTTPException( @@ -5824,10 +5835,7 @@ async def regenerate_key_fn( user_api_key_dict=user_api_key_dict, allowed_routes_was_provided="allowed_routes" in data.model_fields_set, ) - _check_passthrough_routes_caller_permission( - data=data, - user_api_key_dict=user_api_key_dict, - ) + check_allowed_passthrough_routes_caller_permission(data, user_api_key_dict) _check_permissions_caller_permission( data=data, user_api_key_dict=user_api_key_dict, @@ -5945,6 +5953,11 @@ async def regenerate_key_fn( status_code=status.HTTP_403_FORBIDDEN, detail={"error": "You are not authorized to regenerate this key"}, ) + check_denied_passthrough_routes_caller_permission( + data, + user_api_key_dict, + existing_metadata=_key_in_db.metadata, # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # LiteLLM_VerificationToken.metadata is a bare dict + ) if data is not None and (data.access_group_ids or data.object_permission is not None): regenerate_team_table: LiteLLM_TeamTableCachedObj | None = None diff --git a/litellm/proxy/management_endpoints/team_endpoints.py b/litellm/proxy/management_endpoints/team_endpoints.py index aeb317c8905..dfd3b93ff23 100644 --- a/litellm/proxy/management_endpoints/team_endpoints.py +++ b/litellm/proxy/management_endpoints/team_endpoints.py @@ -1389,6 +1389,7 @@ async def new_team( - team_member_tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for individual team members. - team_member_key_duration: Optional[str] - The duration for a team member's key. e.g. "1d", "1w", "1mo" - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team. + - denied_passthrough_routes: Optional[List[str]] - List of pass through routes the team's keys may not call, even if allowed. Applies together with each key's `denied_passthrough_routes`. - allowed_vector_store_indexes: Optional[List[dict]] - List of allowed vector store indexes for the key. Example - [{"index_name": "my-index", "index_permissions": ["write", "read"]}]. If specified, the key will only be able to use these specific vector store indexes. Create index, using `/v1/indexes` endpoint. - secret_manager_settings: Optional[dict] - Secret manager settings for the team. [Docs](https://docs.litellm.ai/docs/secret_managers/overview) - router_settings: Optional[UpdateRouterConfig] - team-specific router settings. Example - {"model_group_retry_policy": {"gpt-4": {"RateLimitErrorRetries": 5}}}. IF null or {} then no router settings. @@ -2151,6 +2152,7 @@ async def update_team( - team_member_tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for individual team members. - team_member_key_duration: Optional[str] - The duration for a team member's key. e.g. "1d", "1w", "1mo" - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team. + - denied_passthrough_routes: Optional[List[str]] - List of pass through routes the team's keys may not call, even if allowed. Applies together with each key's `denied_passthrough_routes`. - model_rpm_limit: Optional[Dict[str, int]] - The RPM (Requests Per Minute) limit per model for this team. Example: {"gpt-4": 100, "gpt-3.5-turbo": 200} - model_tpm_limit: Optional[Dict[str, int]] - The TPM (Tokens Per Minute) limit per model for this team. Example: {"gpt-4": 10000, "gpt-3.5-turbo": 20000} - default_estimated_output_tokens: Optional[int] - Expected output tokens reserved for TPM limiting when a request omits max_tokens, for keys on this team that do not set their own. Positive integer. @@ -2283,7 +2285,12 @@ async def update_team( entity="team", ) - _check_passthrough_routes_caller_permission(data, user_api_key_dict, entity="team") + _check_passthrough_routes_caller_permission( + data, + user_api_key_dict, + entity="team", + existing_metadata=_existing_team_metadata if isinstance(_existing_team_metadata, dict) else None, + ) _check_disable_global_guardrails_caller_permission( data.disable_global_guardrails, data.metadata, # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # request models declare `metadata` as bare dict diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 3f4a30e9179..fc70ffe8697 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -716,23 +716,29 @@ class HttpPassThroughEndpointHelpers(BasePassthroughUtils): if not subpath: return base_target - # Ensure base_target ends with / and subpath doesn't start with / - if not base_target.endswith("/"): - base_target = base_target + "/" - subpath = subpath.removeprefix("/") + target_root: Final = base_target if base_target.endswith("/") else base_target + "/" + return target_root + HttpPassThroughEndpointHelpers.resolve_subpath(subpath) - # Resolve any '..' segments in the subpath so it cannot climb above - # the base_target prefix that the operator configured. Preserve a - # trailing slash on the original subpath since some upstreams treat - # `/foo` and `/foo/` as different resources. - trailing_slash: Final = subpath.endswith("/") - safe_subpath = posixpath.normpath("/" + subpath).lstrip("/") - if safe_subpath == ".": - safe_subpath = "" - if trailing_slash and safe_subpath and not safe_subpath.endswith("/"): - safe_subpath += "/" + @staticmethod + def resolve_subpath(subpath: str) -> str: + """ + ``subpath`` with ``.``, ``..`` and empty segments resolved, so it cannot climb above the target the + operator configured. A trailing slash is kept since some upstreams treat `/foo` and `/foo/` differently. + """ + resolved: Final = posixpath.normpath("/" + subpath.removeprefix("/")).lstrip("/") + return resolved + "/" if resolved and subpath.endswith("/") else resolved - return base_target + safe_subpath + @staticmethod + def forwarded_route(endpoint_path: str, subpath: str) -> str: + """ + The proxy route as the upstream sees it: the subpath resolved like the forwarder resolves it, then + parsed by ``httpx`` like the forwarded URL is, so a decoded ``?`` or ``#`` ends the path there too. + """ + route: Final = f"{endpoint_path.rstrip('/')}/{HttpPassThroughEndpointHelpers.resolve_subpath(subpath)}" + try: + return httpx.URL(route).path + except httpx.InvalidURL: + return route @staticmethod def join_base_and_endpoint_path(base_url: httpx.URL, endpoint_path: str) -> str: @@ -3272,6 +3278,31 @@ class InitPassThroughEndpointHelpers: return False + @staticmethod + def forwarded_routes(route: str) -> tuple[str, ...]: + """ + ``route`` as each registered endpoint it falls under would forward it. An exact endpoint, or no + endpoint at all, sees ``route`` itself. + """ + comparison_route: Final = InitPassThroughEndpointHelpers._route_for_registry_lookup(route) + registered: Final = tuple( + (parts[1], parts[2]) + for parts in (key.split(":", 3) for key in _registered_pass_through_routes) + if len(parts) >= 3 + ) + subpath_endpoint_paths: Final = tuple( + path + for route_type, path in registered + if route_type == "subpath" and (comparison_route == path or comparison_route.startswith(path + "/")) + ) + forwarded: Final = tuple( + HttpPassThroughEndpointHelpers.forwarded_route(endpoint_path=path, subpath=comparison_route[len(path) :]) + for path in subpath_endpoint_paths + ) + if subpath_endpoint_paths and ("exact", comparison_route) not in registered: + return forwarded + return (comparison_route, *forwarded) + @staticmethod def get_registered_pass_through_route(route: str, method: str | None = None) -> dict[str, Any] | None: """Get passthrough params for a given route and optionally filter by HTTP method""" @@ -3576,7 +3607,8 @@ async def _filter_endpoints_by_team_allowed_routes( prisma_client, ) -> list[PassThroughGenericEndpoint]: """ - Filter pass-through endpoints based on team's allowed_passthrough_routes metadata. + Filter pass-through endpoints based on team's allowed_passthrough_routes and + denied_passthrough_routes metadata. Args: team_id: The team ID to check permissions for @@ -3603,18 +3635,23 @@ async def _filter_endpoints_by_team_allowed_routes( team_metadata: Final = cast( # cast-ok: prisma types the Json column as str; reads hand back the decoded value "Mapping[str, object] | None", team.metadata ) - if team_metadata is not None and team_metadata.get("allowed_passthrough_routes") is not None: - ## FILTER pass_through_endpoints by allowed_passthrough_routes - pass_through_endpoints = [ - endpoint - for endpoint in pass_through_endpoints - if endpoint.path - in cast( # cast-ok: guarded above; team metadata stores this key as a list of route paths - Sequence[str], team_metadata.get("allowed_passthrough_routes") - ) - ] + if team_metadata is None: + return pass_through_endpoints - return pass_through_endpoints + from litellm.proxy.auth.route_checks import RouteChecks + + allowed_routes: Final = cast( # cast-ok: team metadata stores this key as a list of route paths + "Sequence[str] | None", team_metadata.get("allowed_passthrough_routes") + ) + return [ + endpoint + for endpoint in pass_through_endpoints + if (allowed_routes is None or endpoint.path in allowed_routes) + and not ( + endpoint.auth + and RouteChecks.matching_denied_passthrough_route(route=endpoint.path, metadata_sources=(team_metadata,)) + ) + ] @router.get( diff --git a/tests/integration/authorization/test_passthrough_route_denylist.py b/tests/integration/authorization/test_passthrough_route_denylist.py new file mode 100644 index 00000000000..0ad93eedbc5 --- /dev/null +++ b/tests/integration/authorization/test_passthrough_route_denylist.py @@ -0,0 +1,333 @@ +import json +import uuid +from typing import Final, Literal + +import httpx +import pytest +from integration._support.client import Gateway, JsonValue, Scenario, eventually, object_value, string_value +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import TypeAdapter + +ENDPOINTS: Final = TypeAdapter(list[JsonValue]) +JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue]) +Owner = Literal["key", "team"] + + +def _echo(request: Request) -> Reply: + return Reply(body=json.dumps({"target": request.target}).encode()) + + +def _registered_endpoint(gateway: Gateway, scenario: Scenario, wire: Wire, *, auth: bool = True) -> str: + path: Final = f"/integration-deny-{uuid.uuid4().hex}" + created: Final = gateway.post( + "/config/pass_through_endpoint", + {"path": path, "target": f"{wire.url}/upstream", "auth": auth, "include_subpath": True}, + ) + endpoint_id: Final = object_value(ENDPOINTS.validate_python(created["endpoints"])[0])["id"] + scenario.cleanups.callback( + lambda: gateway.request("DELETE", "/config/pass_through_endpoint", params={"endpoint_id": str(endpoint_id)}) + ) + return path + + +def _call(gateway: Gateway, route: str, key: str) -> httpx.Response: + return gateway.request("POST", route, {"probe": "denylist"}, key=key) + + +def _upstream_targets(wire: Wire) -> tuple[str, ...]: + return tuple(request.target for request in wire.drain()) + + +def _assert_denied(response: httpx.Response, denied_entry: str) -> None: + assert response.status_code == 403, response.text + assert f"Matched `{denied_entry}` in `denied_passthrough_routes`" in response.text, response.text + + +def _key_with_routes( + scenario: Scenario, allow_on: Owner, deny_on: Owner, allowed: list[JsonValue], denied: list[JsonValue] +) -> str: + team_fields: Final[dict[str, JsonValue]] = { + **({"allowed_passthrough_routes": allowed} if allow_on == "team" else {}), + **({"denied_passthrough_routes": denied} if deny_on == "team" else {}), + } + key_fields: Final[dict[str, JsonValue]] = { + **({"allowed_passthrough_routes": allowed} if allow_on == "key" else {}), + **({"denied_passthrough_routes": denied} if deny_on == "key" else {}), + } + return scenario.key(team_id=scenario.team(**team_fields), **key_fields) + + +@pytest.mark.parametrize( + ("allow_on", "deny_on"), + [("key", "key"), ("team", "key"), ("key", "team")], +) +def test_denied_subpath_is_blocked_even_when_allowed_while_its_sibling_still_reaches_upstream( + gateway: Gateway, allow_on: Owner, deny_on: Owner +) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + key: Final = _key_with_routes(scenario, allow_on, deny_on, [path], [f"{path}/admin"]) + + _assert_denied(_call(gateway, f"{path}/admin/users", key), f"{path}/admin") + sibling: Final = _call(gateway, f"{path}/public", key) + + assert sibling.status_code == 200, sibling.text + assert _upstream_targets(wire) == ("/upstream/public",) + + +@pytest.mark.parametrize( + "subpath", + [ + "public/%2e%2e/admin/users", + "/admin/users", + "admin%3F", + "admin%3F/users", + "admin%23", + "admin%23/users", + "public%3Fx/%2e%2e/admin%3F", + "public%23x/%2e%2e/admin%23", + ], + ids=[ + "encoded_dot_dot_segment", + "empty_segment", + "encoded_query_mark", + "encoded_query_mark_then_subpath", + "encoded_fragment_mark", + "encoded_fragment_mark_then_subpath", + "encoded_query_mark_then_dot_dot", + "encoded_fragment_mark_then_dot_dot", + ], +) +def test_dot_and_empty_segments_cannot_reach_a_denied_subpath(gateway: Gateway, subpath: str) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + key: Final = scenario.key(allowed_passthrough_routes=[path], denied_passthrough_routes=[f"{path}/admin"]) + + response: Final = _call(gateway, f"{path}/{subpath}", key) + + _assert_denied(response, f"{path}/admin") + assert _upstream_targets(wire) == () + + +def test_trailing_slash_deny_entry_blocks_the_route_and_everything_under_it(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + key: Final = scenario.key(allowed_passthrough_routes=[path], denied_passthrough_routes=[f"{path}/admin/"]) + + _assert_denied(_call(gateway, f"{path}/admin", key), f"{path}/admin/") + _assert_denied(_call(gateway, f"{path}/admin/", key), f"{path}/admin/") + _assert_denied(_call(gateway, f"{path}/admin/users", key), f"{path}/admin/") + sibling: Final = _call(gateway, f"{path}/public", key) + + assert sibling.status_code == 200, sibling.text + assert _upstream_targets(wire) == ("/upstream/public",) + + +def test_trailing_wildcard_deny_blocks_every_route_with_that_prefix(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + key: Final = scenario.key(allowed_passthrough_routes=[path], denied_passthrough_routes=[f"{path}/adm*"]) + + _assert_denied(_call(gateway, f"{path}/admin", key), f"{path}/adm*") + _assert_denied(_call(gateway, f"{path}/adm-console/x", key), f"{path}/adm*") + sibling: Final = _call(gateway, f"{path}/public", key) + + assert sibling.status_code == 200, sibling.text + assert _upstream_targets(wire) == ("/upstream/public",) + + +def test_deny_entry_does_not_match_a_longer_segment_that_shares_its_prefix(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + key: Final = scenario.key(allowed_passthrough_routes=[path], denied_passthrough_routes=[f"{path}/admin"]) + + response: Final = _call(gateway, f"{path}/administrator", key) + + assert response.status_code == 200, response.text + assert _upstream_targets(wire) == ("/upstream/administrator",) + + +def test_proxy_admin_key_reaches_a_route_its_key_and_team_both_deny(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + admin: Final = scenario.user(user_role="proxy_admin") + team: Final = scenario.team(denied_passthrough_routes=[path]) + gateway.post("/team/member_add", {"team_id": team, "member": {"role": "user", "user_id": admin}}) + key: Final = scenario.key(user_id=admin, team_id=team, denied_passthrough_routes=[path]) + + response: Final = _call(gateway, f"{path}/ops", key) + + assert response.status_code == 200, response.text + assert _upstream_targets(wire) == ("/upstream/ops",) + + +@pytest.mark.parametrize("deny_on", ["key", "team"]) +def test_deny_added_and_cleared_through_update_takes_effect_on_the_next_request( + gateway: Gateway, deny_on: Owner +) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + team: Final = scenario.team() + key: Final = scenario.key(team_id=team, allowed_passthrough_routes=[path]) + + def set_denied(routes: list[JsonValue]) -> None: + if deny_on == "key": + gateway.post("/key/update", {"key": key, "denied_passthrough_routes": routes}) + else: + gateway.post("/team/update", {"team_id": team, "denied_passthrough_routes": routes}) + + def probe() -> httpx.Response: + return _call(gateway, f"{path}/admin", key) + + before: Final = probe() + assert before.status_code == 200, before.text + set_denied([path]) + _assert_denied(eventually(probe, lambda response: response.status_code == 403, seconds=10), path) + set_denied([]) + restored: Final = eventually(probe, lambda response: response.status_code == 200, seconds=10) + + assert restored.status_code == 200, restored.text + targets: Final = _upstream_targets(wire) + assert len(targets) >= 2 and set(targets) == {"/upstream/admin"}, targets + + +@pytest.mark.parametrize( + "body", + [{"denied_passthrough_routes": ["/integration-deny-probe"]}, {"metadata": {"denied_passthrough_routes": ["/x"]}}], + ids=["top_level", "metadata"], +) +def test_internal_user_cannot_set_denied_routes_while_proxy_admin_can( + gateway: Gateway, body: dict[str, JsonValue] +) -> None: + with gateway.scenario() as scenario: + user: Final = scenario.user(user_role="internal_user") + user_key: Final = scenario.key(user_id=user) + + refused: Final = gateway.request("POST", "/key/generate", {"user_id": user, **body}, key=user_key) + if refused.status_code == 200: + scenario.cleanups.callback( + scenario.delete_key, string_value(JSON_OBJECT.validate_json(refused.content)["key"]) + ) + + assert refused.status_code == 403, refused.text + assert "denied_passthrough_routes" in refused.text, refused.text + admin_key: Final = scenario.key(denied_passthrough_routes=["/integration-deny-probe"]) + info: Final = object_value(gateway.get("/key/info", {"key": admin_key})["info"]) + assert object_value(info["metadata"])["denied_passthrough_routes"] == ["/integration-deny-probe"], info + + +def test_deny_entries_leave_open_passthroughs_and_llm_routes_untouched(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + open_path: Final = _registered_endpoint(gateway, scenario, wire, auth=False) + model: Final = scenario.model() + key: Final = scenario.key(denied_passthrough_routes=[open_path, "/v1/chat/completions", "/chat/completions"]) + + opened: Final = _call(gateway, open_path, key) + chat: Final = gateway.request( + "POST", "/v1/chat/completions", {"model": model, "messages": [{"role": "user", "content": "x"}]}, key=key + ) + + assert opened.status_code == 200, opened.text + assert _upstream_targets(wire) == ("/upstream",) + assert chat.status_code == 200, chat.text + + +def test_team_endpoint_listing_hides_routes_the_team_denies(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + denied: Final = _registered_endpoint(gateway, scenario, wire) + visible: Final = _registered_endpoint(gateway, scenario, wire) + team: Final = scenario.team(denied_passthrough_routes=[denied]) + + listed: Final = gateway.get("/config/pass_through_endpoint", {"team_id": team})["endpoints"] + + paths: Final = {string_value(object_value(endpoint)["path"]) for endpoint in ENDPOINTS.validate_python(listed)} + assert visible in paths, paths + assert denied not in paths, paths + + +def test_team_admin_cannot_clear_or_drop_a_deny_a_proxy_admin_set_on_a_team_key(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + team_admin: Final = scenario.user(user_role="internal_user") + team: Final = scenario.team(members_with_roles=[{"role": "admin", "user_id": team_admin}]) + team_admin_key: Final = scenario.key(user_id=team_admin) + denied: Final[list[JsonValue]] = [f"{path}/admin"] + key: Final = scenario.key(team_id=team, allowed_passthrough_routes=[path], denied_passthrough_routes=denied) + + def update(body: dict[str, JsonValue]) -> httpx.Response: + return gateway.request("POST", "/key/update", {"key": key, **body}, key=team_admin_key) + + cleared: Final = update({"denied_passthrough_routes": []}) + dropped: Final = update({"metadata": {}}) + unchanged: Final = update({"denied_passthrough_routes": denied}) + + assert cleared.status_code == 403 and "denied_passthrough_routes" in cleared.text, cleared.text + assert dropped.status_code == 403 and "metadata.denied_passthrough_routes" in dropped.text, dropped.text + assert unchanged.status_code == 200, unchanged.text + _assert_denied(_call(gateway, f"{path}/admin/users", key), f"{path}/admin") + assert _upstream_targets(wire) == () + + +def test_team_admin_bulk_update_cannot_drop_a_deny_a_proxy_admin_set_on_a_team_key(gateway: Gateway) -> None: + with wire_server(_echo) as wire, gateway.scenario() as scenario: + path: Final = _registered_endpoint(gateway, scenario, wire) + team_admin: Final = scenario.user(user_role="internal_user") + team: Final = scenario.team(members_with_roles=[{"role": "admin", "user_id": team_admin}]) + team_admin_key: Final = scenario.key(user_id=team_admin) + guarded: Final = scenario.key( + team_id=team, allowed_passthrough_routes=[path], denied_passthrough_routes=[f"{path}/admin"] + ) + plain: Final = scenario.key(team_id=team, allowed_passthrough_routes=[path]) + + response: Final = gateway.request( + "POST", + "/team/key/bulk_update", + {"team_id": team, "key_ids": [guarded, plain], "update_fields": {"metadata": {}}}, + key=team_admin_key, + ) + + assert response.status_code == 200, response.text + body: Final = JSON_OBJECT.validate_python(response.json()) + failed: Final = tuple(object_value(item) for item in ENDPOINTS.validate_python(body["failed_updates"])) + succeeded: Final = tuple(object_value(item) for item in ENDPOINTS.validate_python(body["successful_updates"])) + assert [string_value(item["key"]) for item in failed] == [guarded], response.text + assert "metadata.denied_passthrough_routes" in string_value(failed[0]["failed_reason"]), response.text + assert [string_value(item["key"]) for item in succeeded] == [plain], response.text + _assert_denied(_call(gateway, f"{path}/admin/users", guarded), f"{path}/admin") + assert _upstream_targets(wire) == () + + +@pytest.mark.parametrize("route", ["/key/update", "/key/regenerate"]) +def test_non_owner_gets_the_same_refusal_whether_or_not_another_users_key_has_a_deny( + gateway: Gateway, route: str +) -> None: + with gateway.scenario() as scenario: + owner: Final = scenario.user(user_role="internal_user") + guarded: Final = scenario.key(user_id=owner, denied_passthrough_routes=["/integration-deny-probe"]) + plain: Final = scenario.key(user_id=owner) + outsider_key: Final = scenario.key(user_id=scenario.user(user_role="internal_user")) + + def probe(key: str) -> httpx.Response: + return gateway.request("POST", route, {"key": key, "denied_passthrough_routes": []}, key=outsider_key) + + on_guarded: Final = probe(guarded) + on_plain: Final = probe(plain) + + assert on_guarded.status_code == on_plain.status_code != 200, (on_guarded.text, on_plain.text) + assert "denied_passthrough_routes" not in on_guarded.text, on_guarded.text + assert on_guarded.text.replace(guarded, "KEY") == on_plain.text.replace(plain, "KEY") + + +def test_non_admin_setting_allowed_routes_on_regenerate_is_refused_before_the_key_lookup(gateway: Gateway) -> None: + with gateway.scenario() as scenario: + user_key: Final = scenario.key(user_id=scenario.user(user_role="internal_user")) + + response: Final = gateway.request( + "POST", + "/key/regenerate", + {"key": f"sk-missing-{uuid.uuid4().hex}", "allowed_passthrough_routes": ["/integration-deny-probe"]}, + key=user_key, + ) + + assert response.status_code == 403, response.text + assert "allowed_passthrough_routes" in response.text, response.text diff --git a/tests/unit/proxy/auth/test_handle_jwt.py b/tests/unit/proxy/auth/test_handle_jwt.py index 39c2466033d..69eb01cabcd 100644 --- a/tests/unit/proxy/auth/test_handle_jwt.py +++ b/tests/unit/proxy/auth/test_handle_jwt.py @@ -8252,3 +8252,54 @@ async def test_check_admin_access_names_the_route_and_the_expanded_allow_list_wh "Admin not allowed to access this route. Route=/key/generate, " f"Allowed Routes={[*LiteLLMRoutes.info_routes.value, '/custom/admin/route']}" ) + + +@pytest.mark.parametrize( + "team_allowed_routes, team_metadata", + [ + ((), {"allowed_passthrough_routes": ["/model-host"], "denied_passthrough_routes": ["/model-host/v1"]}), + (("/model-host/*",), {"denied_passthrough_routes": ["/model-host/v1/*"]}), + ], + ids=["team-metadata-allow", "jwt-team-allowed-routes-grant"], +) +def test_team_has_passthrough_route_access_denied_route_wins( + team_allowed_routes: tuple[str, ...], + team_metadata: dict[str, list[str]], + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("SERVER_ROOT_PATH", raising=False) + team: Final = LiteLLM_TeamTable(team_id="team-a", metadata=team_metadata) + + with patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes", + _AUTH_ENFORCED_MODEL_HOST_ROUTES, + ): + assert not JWTAuthManager._team_has_passthrough_route_access( + team_object=team, + route="/model-host/v1/extractor/predict", + request_method="POST", + team_allowed_routes=team_allowed_routes, + ) + + +@pytest.mark.parametrize( + "team_metadata, expected_detail", + [ + ( + {"allowed_passthrough_routes": ["/model-host"], "denied_passthrough_routes": ["/model-host/v1"]}, + "Matched `/model-host/v1` in `denied_passthrough_routes`", + ), + ({}, "Team not allowed to access passthrough route"), + ], + ids=["team-deny-names-the-entry", "no-grant-keeps-generic-message"], +) +def test_team_passthrough_route_denial_names_the_matched_deny_entry( + team_metadata: dict[str, list[str]], expected_detail: str +) -> None: + team: Final = LiteLLM_TeamTable(team_id="team-a", metadata=team_metadata) + + with pytest.raises(HTTPException) as exc_info: + JWTAuthManager._raise_team_passthrough_route_denial(route="/model-host/v1/predict", team_object=team) + + assert exc_info.value.status_code == 403 + assert expected_detail in str(exc_info.value.detail) diff --git a/tests/unit/proxy/auth/test_route_checks.py b/tests/unit/proxy/auth/test_route_checks.py index fb63c4ce425..36799b73876 100644 --- a/tests/unit/proxy/auth/test_route_checks.py +++ b/tests/unit/proxy/auth/test_route_checks.py @@ -13,6 +13,7 @@ from litellm.proxy._types import ( LitellmUserRoles, UserAPIKeyAuth, ) +from litellm.proxy.auth.auth_checks import _is_api_route_allowed from litellm.proxy.auth.auth_checks_organization import _user_is_org_admin from litellm.proxy.auth.route_checks import RouteChecks from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import router as llm_passthrough_router @@ -4464,6 +4465,210 @@ def test_non_admin_trace_reads_reach_endpoint_visibility_checks(route: str) -> N ) +_DENY_TEST_REGISTERED_ROUTES: Final = { + "test-uuid-1:subpath:/svc:GET,POST": { + "endpoint_id": "test-uuid-1", + "path": "/svc", + "type": "subpath", + "auth": True, + }, +} + + +def _check_route_with_registered_routes( + route: str, valid_token: UserAPIKeyAuth, user_role: LitellmUserRoles = LitellmUserRoles.INTERNAL_USER +) -> None: + request: Final = MagicMock(spec=Request) + request.method = "POST" + with ( + pytest.MonkeyPatch.context() as env, + patch( + "litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes", + _DENY_TEST_REGISTERED_ROUTES, + ), + ): + env.delenv("SERVER_ROOT_PATH", raising=False) + _is_api_route_allowed( + route=route, + request=request, + request_data={}, + valid_token=valid_token, + user_obj=LiteLLM_UserTable(user_id="test_user", user_role=user_role.value), + ) + + +@pytest.mark.parametrize( + "metadata, team_metadata, denied_route", + [ + ({"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin"]}, {}, "/svc/admin"), + ({"denied_passthrough_routes": ["/svc/admin"]}, {"allowed_passthrough_routes": ["/svc"]}, "/svc/admin"), + ({"allowed_passthrough_routes": ["/svc"]}, {"denied_passthrough_routes": ["/svc/admin"]}, "/svc/admin"), + ({"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/adm*"]}, {}, "/svc/adm*"), + ], + ids=["key-deny-beats-key-allow", "key-deny-beats-team-allow", "team-deny-beats-key-allow", "wildcard-deny"], +) +def test_denied_passthrough_routes_win_over_allow( + metadata: dict[str, list[str]], team_metadata: dict[str, list[str]], denied_route: str +) -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata=metadata, + team_metadata=team_metadata, + ) + + with pytest.raises(HTTPException) as exc_info: + _check_route_with_registered_routes(route="/svc/admin/users", valid_token=valid_token) + + assert exc_info.value.status_code == 403 + assert f"Matched `{denied_route}` in `denied_passthrough_routes`" in exc_info.value.detail + + +@pytest.mark.parametrize( + "route", + ["/svc/public", "/svc/administrator", "/anthropic/v1/messages", "/chat/completions"], + ids=["allowed-sibling", "no-false-prefix-match", "built-in-provider-route", "llm-api-route"], +) +def test_denied_passthrough_routes_leave_other_routes_untouched(route: str) -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={ + "allowed_passthrough_routes": ["/svc"], + "denied_passthrough_routes": ["/svc/admin", "/anthropic", "/chat/completions"], + }, + ) + + _check_route_with_registered_routes(route=route, valid_token=valid_token) + + +def test_denied_passthrough_routes_do_not_restrict_proxy_admins() -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.PROXY_ADMIN.value, + metadata={"denied_passthrough_routes": ["/svc"]}, + team_metadata={"denied_passthrough_routes": ["/svc"]}, + ) + + _check_route_with_registered_routes( + route="/svc/admin/users", valid_token=valid_token, user_role=LitellmUserRoles.PROXY_ADMIN + ) + + +@pytest.mark.parametrize( + "route", + [ + "/svc/public/../admin/users", + "/svc/public/../../admin/users", + "/svc//admin/users", + "/svc/./admin", + "/svc/admin?", + "/svc/admin?/users", + "/svc/admin#", + "/svc/admin#/users", + "/svc/public/../admin?x", + "/svc/public?x/../admin?", + "/svc/public#x/../admin#", + ], + ids=[ + "dot-dot-segment", + "dot-dot-past-endpoint-root", + "empty-segment", + "dot-segment", + "query-mark", + "query-mark-then-subpath", + "fragment-mark", + "fragment-mark-then-subpath", + "dot-dot-then-query-mark", + "query-mark-then-dot-dot", + "fragment-mark-then-dot-dot", + ], +) +def test_dot_and_empty_segments_cannot_reach_a_denied_route(route: str) -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin"]}, + ) + + with pytest.raises(HTTPException) as exc_info: + _check_route_with_registered_routes(route=route, valid_token=valid_token) + + assert exc_info.value.status_code == 403 + assert "Matched `/svc/admin` in `denied_passthrough_routes`" in exc_info.value.detail + + +def test_dot_dot_out_of_a_denied_route_is_checked_as_the_route_it_forwards_to() -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin"]}, + ) + + _check_route_with_registered_routes(route="/svc/admin/../public", valid_token=valid_token) + + +@pytest.mark.parametrize("denied_route", ["/", "//"]) +@pytest.mark.parametrize("route", ["/svc", "/svc/public", "/svc/admin/users"]) +def test_root_deny_entry_blocks_every_route(route: str, denied_route: str) -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": [denied_route]}, + ) + + with pytest.raises(HTTPException) as exc_info: + _check_route_with_registered_routes(route=route, valid_token=valid_token) + + assert exc_info.value.status_code == 403 + assert f"Matched `{denied_route}` in `denied_passthrough_routes`" in exc_info.value.detail + + +@pytest.mark.parametrize("route", ["/svc/admin", "/svc/admin/", "/svc/admin/users"]) +def test_trailing_slash_deny_entry_blocks_the_route_and_everything_under_it(route: str) -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin/"]}, + ) + + with pytest.raises(HTTPException) as exc_info: + _check_route_with_registered_routes(route=route, valid_token=valid_token) + + assert exc_info.value.status_code == 403 + assert "Matched `/svc/admin/` in `denied_passthrough_routes`" in exc_info.value.detail + + +def test_trailing_slash_deny_entry_does_not_match_a_longer_segment() -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin/"]}, + ) + + _check_route_with_registered_routes(route="/svc/administrator", valid_token=valid_token) + + +def test_dot_segments_resolving_outside_a_denied_route_still_pass() -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin"]}, + ) + + _check_route_with_registered_routes(route="/svc/public/./docs", valid_token=valid_token) + + +def test_query_text_naming_a_denied_route_still_passes() -> None: + valid_token: Final = UserAPIKeyAuth( + user_id="test_user", + user_role=LitellmUserRoles.INTERNAL_USER.value, + metadata={"allowed_passthrough_routes": ["/svc"], "denied_passthrough_routes": ["/svc/admin"]}, + ) + + _check_route_with_registered_routes(route="/svc/public?next=/svc/admin", valid_token=valid_token) + + def test_is_llm_api_route(): assert RouteChecks.is_llm_api_route("/v1/chat/completions") is True assert RouteChecks.is_llm_api_route("/v1/completions") is True diff --git a/tests/unit/proxy/management_endpoints/test_common_utils.py b/tests/unit/proxy/management_endpoints/test_common_utils.py index 15bd1bb6690..67920c9c7fe 100644 --- a/tests/unit/proxy/management_endpoints/test_common_utils.py +++ b/tests/unit/proxy/management_endpoints/test_common_utils.py @@ -9,22 +9,24 @@ users can intentionally clear previously-set fields. from datetime import datetime, timezone from types import SimpleNamespace - -from fastapi import HTTPException -from litellm import Router +from typing import Final from unittest.mock import AsyncMock, MagicMock, patch import pytest +from fastapi import HTTPException +from pydantic import BaseModel +from litellm import Router from litellm.proxy._types import ( - Member, LiteLLM_OrganizationMembershipTable, LiteLLM_TeamTable, LiteLLM_UserTable, LitellmUserRoles, + Member, UserAPIKeyAuth, ) from litellm.proxy.management_endpoints.common_utils import ( + _has_non_empty_value, _org_admin_can_invite_user, _set_object_metadata_field, _team_admin_can_invite_user, @@ -33,7 +35,6 @@ from litellm.proxy.management_endpoints.common_utils import ( _user_has_admin_view, admin_can_invite_user, ) -from litellm.proxy.management_endpoints.common_utils import _has_non_empty_value from litellm.types.utils import BudgetConfig @@ -726,11 +727,109 @@ class TestCheckPassthroughRoutesCallerPermission: class _Bare(BaseModel): unrelated: str = "x" - assert ( - _check_passthrough_routes_caller_permission(_Bare(), self._non_admin()) - is None + assert _check_passthrough_routes_caller_permission(_Bare(), self._non_admin()) is None + + @pytest.mark.parametrize( + "kwargs, field", + [ + ({"denied_passthrough_routes": ["/v1/foo"]}, "denied_passthrough_routes"), + ({"metadata": {"denied_passthrough_routes": ["/v1/foo"]}}, "metadata.denied_passthrough_routes"), + ], + ) + def test_denied_routes_rejected_for_non_admin(self, kwargs: dict[str, object], field: str) -> None: + from fastapi import HTTPException + from pydantic import BaseModel + + from litellm.proxy.management_endpoints.common_utils import ( + _check_passthrough_routes_caller_permission, ) + class _RouteData(BaseModel): + denied_passthrough_routes: list[str] | None = None + metadata: dict[str, object] | None = None + + with pytest.raises(HTTPException) as exc_info: + _check_passthrough_routes_caller_permission( + _RouteData.model_validate(kwargs), self._non_admin(), entity="team" + ) + + assert exc_info.value.detail == {"error": f"Only proxy admins can set `{field}` on a team."} + + +class _DenyRouteData(BaseModel): + denied_passthrough_routes: list[str] | None = None + metadata: dict[str, object] | None = None + max_budget: float | None = None + + +_EXISTING_DENY: Final = {"denied_passthrough_routes": ["/v1/foo"]} + + +class TestDeniedPassthroughRoutesCallerPermission: + @pytest.mark.parametrize( + "kwargs, field", + [ + ({"denied_passthrough_routes": []}, "denied_passthrough_routes"), + ({"denied_passthrough_routes": ["/v1/other"]}, "denied_passthrough_routes"), + ({"metadata": {"team": "core"}}, "metadata.denied_passthrough_routes"), + ({"metadata": None}, "metadata.denied_passthrough_routes"), + ], + ids=["cleared", "replaced", "dropped-by-metadata-replace", "dropped-by-null-metadata"], + ) + def test_non_admin_cannot_change_an_existing_deny_list(self, kwargs: dict[str, object], field: str) -> None: + from litellm.proxy.management_endpoints.common_utils import _check_passthrough_routes_caller_permission + + with pytest.raises(HTTPException) as exc_info: + _check_passthrough_routes_caller_permission( + _DenyRouteData.model_validate(kwargs), + UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), + existing_metadata=_EXISTING_DENY, + ) + + assert exc_info.value.status_code == 403 + assert exc_info.value.detail == {"error": f"Only proxy admins can set `{field}` on a key."} + + @pytest.mark.parametrize( + "kwargs", + [ + {"denied_passthrough_routes": ["/v1/foo"]}, + {"metadata": {"team": "core", "denied_passthrough_routes": ["/v1/foo"]}}, + {"max_budget": 10.0}, + ], + ids=["resent-top-level", "resent-in-metadata", "unrelated-field"], + ) + def test_non_admin_may_leave_an_existing_deny_list_unchanged(self, kwargs: dict[str, object]) -> None: + from litellm.proxy.management_endpoints.common_utils import _check_passthrough_routes_caller_permission + + _check_passthrough_routes_caller_permission( + _DenyRouteData.model_validate(kwargs), + UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), + existing_metadata=_EXISTING_DENY, + ) + + def test_non_admin_may_send_null_metadata_when_no_deny_list_exists(self) -> None: + from litellm.proxy.management_endpoints.common_utils import _check_passthrough_routes_caller_permission + + _check_passthrough_routes_caller_permission( + _DenyRouteData(metadata=None), + UserAPIKeyAuth(user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER), + existing_metadata={"team": "core"}, + ) + + def test_malformed_metadata_deny_entries_are_rejected_even_for_proxy_admins(self) -> None: + from litellm.proxy.management_endpoints.common_utils import _check_passthrough_routes_caller_permission + + with pytest.raises(HTTPException) as exc_info: + _check_passthrough_routes_caller_permission( + _DenyRouteData(metadata={"denied_passthrough_routes": [123, None]}), + UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN), + ) + + assert exc_info.value.status_code == 400 + assert exc_info.value.detail == { + "error": "`metadata.denied_passthrough_routes` must be a list of route strings." + } + class TestCheckDisableGlobalGuardrailsCallerPermission: """Only proxy admins may set disable_global_guardrails (top-level or under diff --git a/tests/unit/proxy/management_endpoints/test_key_management_endpoints.py b/tests/unit/proxy/management_endpoints/test_key_management_endpoints.py index 48ecdb287ab..20672e67358 100644 --- a/tests/unit/proxy/management_endpoints/test_key_management_endpoints.py +++ b/tests/unit/proxy/management_endpoints/test_key_management_endpoints.py @@ -14781,6 +14781,37 @@ async def test_process_single_key_update_non_admin_permissions_explicit_empty_re assert "permissions" in str(exc_info.value.detail) +@pytest.mark.asyncio +@pytest.mark.parametrize( + "fields", + [{"metadata": {}}, {"metadata": None}, {"denied_passthrough_routes": []}], + ids=["metadata_replaced", "metadata_null", "denies_cleared"], +) +async def test_process_single_key_update_non_admin_cannot_drop_stored_denied_passthrough_routes( + fields: dict[str, object], +) -> None: + stored_key: Final = LiteLLM_VerificationToken( + token="hashed-key", user_id="key-owner", metadata={"denied_passthrough_routes": ["/svc/admin"]} + ) + prisma_client: Final = AsyncMock() + + with pytest.raises(HTTPException) as exc_info: + await _process_single_key_update( + update_key_request=UpdateKeyRequest.model_validate({"key": "sk-owned-key", **fields}), + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="team-admin"), + litellm_changed_by=None, + prisma_client=prisma_client, + user_api_key_cache=MagicMock(), + proxy_logging_obj=MagicMock(), + llm_router=MagicMock(), + existing_key_row=stored_key, + ) + + assert exc_info.value.status_code == 403 + assert "denied_passthrough_routes" in str(exc_info.value.detail) + prisma_client.update_data.assert_not_called() + + @pytest.mark.asyncio async def test_execute_virtual_key_regeneration_cache_invalidation_with_token_hash(): """ diff --git a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py index 13ccef94908..4142e5b94c6 100644 --- a/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py +++ b/tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py @@ -56,9 +56,9 @@ from litellm.proxy.pass_through_endpoints.success_handler import ( from litellm.proxy.route_llm_request import ProxyModelNotFoundError from litellm.types import utils as types_utils from litellm.types.passthrough_endpoints.pass_through_endpoints import ( - EndpointType, LITELLM_PASS_THROUGH_DEPLOYMENT_MODEL_INFO_STATE_KEY, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, + EndpointType, ) from tests._master_key import MASTER_KEY from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome @@ -713,6 +713,28 @@ def test_construct_target_url_with_subpath(): ) assert result == "http://example.com/api/v1" + result = HttpPassThroughEndpointHelpers.construct_target_url_with_subpath( + base_target="http://example.com", subpath="api/../v1/", include_subpath=True + ) + assert result == "http://example.com/v1/" + + +@pytest.mark.parametrize( + "subpath", + ["admin/users", "public/../admin", "../../admin", "/admin/", "./admin", "admin?", "public?x/../admin#"], +) +def test_forwarded_route_is_the_path_the_forwarder_sends_upstream(subpath: str) -> None: + from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( + HttpPassThroughEndpointHelpers, + ) + + target: Final = HttpPassThroughEndpointHelpers.construct_target_url_with_subpath( + base_target="http://upstream.test/base", subpath=subpath, include_subpath=True + ) + forwarded: Final = HttpPassThroughEndpointHelpers.forwarded_route(endpoint_path="/svc", subpath=subpath) + + assert "/svc" + httpx.URL(target).path.removeprefix("/base") == forwarded + def test_add_exact_path_route(): """ @@ -7019,12 +7041,21 @@ async def test_user_defined_passthrough_is_neither_tracked_nor_enforced(metadata api_key="custom-key", token="custom-key", team_id="shared-team", team_model_max_budget=budget, ) endpoint: Final = create_pass_through_route( - endpoint="/custom-budget-test", target="https://upstream.test/echo", custom_headers={}, cost_per_request=0.25, + endpoint="/custom-budget-test", + target="https://upstream.test/echo", + custom_headers={}, + cost_per_request=0.25, + ) + request: Final = Request( + { + "type": "http", + "method": "POST", + "path": "/custom-budget-test", + "headers": [], + "query_string": b"", + "endpoint": endpoint, + } ) - request: Final = Request({ - "type": "http", "method": "POST", "path": "/custom-budget-test", "headers": [], - "query_string": b"", "endpoint": endpoint, - }) body: Final = { "model": "upstream-only-model", metadata_slot: { "model_group": "managed-model", "customer_label": "retained", @@ -8368,6 +8399,56 @@ def test_a_pass_through_added_after_a_lazy_feature_loaded_takes_over_its_path(mo assert client.post("/v1/decider").json() == {"served_by": "pass-through"} +@pytest.mark.asyncio +async def test_filter_endpoints_by_team_allowed_routes_drops_denied() -> None: + from litellm.proxy._types import PassThroughGenericEndpoint + from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( + _filter_endpoints_by_team_allowed_routes, + ) + + endpoints: Final = [ + PassThroughGenericEndpoint(id="endpoint-1", path="/api/public", target="http://example.com/api1"), + PassThroughGenericEndpoint(id="endpoint-2", path="/api/admin", target="http://example.com/api2"), + ] + mock_prisma_client: Final = MagicMock() + mock_team: Final = MagicMock() + mock_team.metadata = {"denied_passthrough_routes": ["/api/admin"]} + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=mock_team) + + result: Final = await _filter_endpoints_by_team_allowed_routes( + team_id="test-team-123", + pass_through_endpoints=endpoints, + prisma_client=mock_prisma_client, + ) + + assert [endpoint.path for endpoint in result] == ["/api/public"] + + +@pytest.mark.asyncio +async def test_filter_endpoints_by_team_allowed_routes_keeps_public_endpoints_the_team_denies() -> None: + from litellm.proxy._types import PassThroughGenericEndpoint + from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( + _filter_endpoints_by_team_allowed_routes, + ) + + endpoints: Final = [ + PassThroughGenericEndpoint(id="endpoint-1", path="/api/webhook", target="http://example.com/a", auth=False), + PassThroughGenericEndpoint(id="endpoint-2", path="/api/admin", target="http://example.com/b"), + ] + mock_prisma_client: Final = MagicMock() + mock_team: Final = MagicMock() + mock_team.metadata = {"denied_passthrough_routes": ["/api"]} + mock_prisma_client.db.litellm_teamtable.find_unique = AsyncMock(return_value=mock_team) + + result: Final = await _filter_endpoints_by_team_allowed_routes( + team_id="test-team-123", + pass_through_endpoints=endpoints, + prisma_client=mock_prisma_client, + ) + + assert [endpoint.path for endpoint in result] == ["/api/webhook"] + + @pytest.fixture() async def _drain_logging_worker(): """ diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index c1107a7e910..bce470a36f1 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -8275,6 +8275,7 @@ export interface paths { * - prompts: Optional[List[str]] - List of prompts that the key is allowed to use. * - allowed_routes: Optional[list] - List of allowed routes for the key. Store the actual route or store a wildcard pattern for a set of routes. Example - ["/chat/completions", "/embeddings", "/keys/*"] * - allowed_passthrough_routes: Optional[list] - List of allowed pass through endpoints for the key. Store the actual endpoint or store a wildcard pattern for a set of endpoints. Example - ["/my-custom-endpoint"]. Use this instead of allowed_routes, if you just want to specify which pass through endpoints the key can access, without specifying the routes. If allowed_routes is specified, allowed_pass_through_endpoints is ignored. + * - denied_passthrough_routes: Optional[list] - List of pass through routes the key may not call, even if allowed by `allowed_passthrough_routes` or `allowed_routes`. Matches exact paths, path prefixes, and trailing `*` wildcards. Applies together with the team's `denied_passthrough_routes`. Example - ["/my-custom-endpoint/admin"]. * - object_permission: Optional[LiteLLM_ObjectPermissionBase] - key-specific object permission. Example - {"vector_stores": ["vector_store_1", "vector_store_2"], "agents": ["agent_1", "agent_2"], "agent_access_groups": ["dev_group"]}. IF null or {} then no object permission. * - key_type: Optional[str] - Type of key that determines default allowed routes. Options: "llm_api" (can call LLM API routes), "management" (can call management routes), "read_only" (can only call info/read routes), "default" (uses default allowed routes). Defaults to "default". * - prompts: Optional[List[str]] - List of allowed prompts for the key. If specified, the key will only be able to use these specific prompts. @@ -8741,6 +8742,7 @@ export interface paths { * - temp_budget_expiry: Optional[str] - Expiry time for the temporary budget increase (Enterprise only). * - allowed_routes: Optional[list] - List of allowed routes for the key. Store the actual route or store a wildcard pattern for a set of routes. Example - ["/chat/completions", "/embeddings", "/keys/*"] * - allowed_passthrough_routes: Optional[list] - List of allowed pass through routes for the key. Store the actual route or store a wildcard pattern for a set of routes. Example - ["/my-custom-endpoint"]. Use this instead of allowed_routes, if you just want to specify which pass through routes the key can access, without specifying the routes. If allowed_routes is specified, allowed_passthrough_routes is ignored. + * - denied_passthrough_routes: Optional[list] - List of pass through routes the key may not call, even if allowed by `allowed_passthrough_routes` or `allowed_routes`. Matches exact paths, path prefixes, and trailing `*` wildcards. Applies together with the team's `denied_passthrough_routes`. Example - ["/my-custom-endpoint/admin"]. * - prompts: Optional[List[str]] - List of allowed prompts for the key. If specified, the key will only be able to use these specific prompts. * - object_permission: Optional[LiteLLM_ObjectPermissionBase] - key-specific object permission. Example - {"vector_stores": ["vector_store_1", "vector_store_2"], "agents": ["agent_1", "agent_2"], "agent_access_groups": ["dev_group"]}. IF null or {} then no object permission. * - auto_rotate: Optional[bool] - Whether this key should be automatically rotated @@ -17220,6 +17222,7 @@ export interface paths { * - team_member_tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for individual team members. * - team_member_key_duration: Optional[str] - The duration for a team member's key. e.g. "1d", "1w", "1mo" * - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team. + * - denied_passthrough_routes: Optional[List[str]] - List of pass through routes the team's keys may not call, even if allowed. Applies together with each key's `denied_passthrough_routes`. * - allowed_vector_store_indexes: Optional[List[dict]] - List of allowed vector store indexes for the key. Example - [{"index_name": "my-index", "index_permissions": ["write", "read"]}]. If specified, the key will only be able to use these specific vector store indexes. Create index, using `/v1/indexes` endpoint. * - secret_manager_settings: Optional[dict] - Secret manager settings for the team. [Docs](https://docs.litellm.ai/docs/secret_managers/overview) * - router_settings: Optional[UpdateRouterConfig] - team-specific router settings. Example - {"model_group_retry_policy": {"gpt-4": {"RateLimitErrorRetries": 5}}}. IF null or {} then no router settings. @@ -17447,6 +17450,7 @@ export interface paths { * - team_member_tpm_limit: Optional[int] - The TPM (Tokens Per Minute) limit for individual team members. * - team_member_key_duration: Optional[str] - The duration for a team member's key. e.g. "1d", "1w", "1mo" * - allowed_passthrough_routes: Optional[List[str]] - List of allowed pass through routes for the team. + * - denied_passthrough_routes: Optional[List[str]] - List of pass through routes the team's keys may not call, even if allowed. Applies together with each key's `denied_passthrough_routes`. * - model_rpm_limit: Optional[Dict[str, int]] - The RPM (Requests Per Minute) limit per model for this team. Example: {"gpt-4": 100, "gpt-3.5-turbo": 200} * - model_tpm_limit: Optional[Dict[str, int]] - The TPM (Tokens Per Minute) limit per model for this team. Example: {"gpt-4": 10000, "gpt-3.5-turbo": 20000} * - default_estimated_output_tokens: Optional[int] - Expected output tokens reserved for TPM limiting when a request omits max_tokens, for keys on this team that do not set their own. Positive integer. @@ -32879,6 +32883,8 @@ export interface components { default_estimated_output_tokens_per_model?: { [key: string]: number; } | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Duration */ @@ -33045,6 +33051,8 @@ export interface components { default_estimated_output_tokens_per_model?: { [key: string]: number; } | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Duration */ @@ -39478,6 +39486,8 @@ export interface components { } | null; /** Default Team Member Models */ default_team_member_models?: string[] | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Enforced Batch Output Expires After */ @@ -39772,6 +39782,8 @@ export interface components { default_estimated_output_tokens_per_model?: { [key: string]: number; } | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Duration */ @@ -40590,6 +40602,8 @@ export interface components { } | null; /** Default Team Member Models */ default_team_member_models?: string[] | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Enforced Batch Output Expires After */ @@ -42171,6 +42185,8 @@ export interface components { default_estimated_output_tokens_per_model?: { [key: string]: number; } | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Duration */ @@ -48341,6 +48357,8 @@ export interface components { default_estimated_output_tokens_per_model?: { [key: string]: number; } | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Duration */ @@ -48835,6 +48853,8 @@ export interface components { } | null; /** Default Team Member Models */ default_team_member_models?: string[] | null; + /** Denied Passthrough Routes */ + denied_passthrough_routes?: string[] | null; /** Disable Global Guardrails */ disable_global_guardrails?: boolean | null; /** Enforced Batch Output Expires After */