ai-workspace-services/litellm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(proxy): enforce allowed_passthrough_routes for auth=true pass-thr… (#29256)

Browse Source 
* fix(proxy): enforce allowed_passthrough_routes for auth=true pass-through

Pass-through endpoints with auth=true were injected into openai_routes,
so teams with openai_routes access bypassed per-team allowed_passthrough_routes.
Gate auth-enforced pass-through at JWT, virtual-key, and non-admin route checks.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(proxy): clarify JWT passthrough denial

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(proxy): make pass-through auth checks method-aware

Prevent allowlist bypass when the same path is registered with different auth settings per HTTP method.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Fix passthrough route auth checks

* fix(proxy): reject unregistered pass-through HTTP methods

Enforce method-aware JWT checks and return 405 when stale FastAPI routes accept requests outside the current pass-through registry.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(proxy): remove duplicate request_method in JWT team lookup

Fixes SyntaxError on proxy startup caused by passing request_method twice to find_team_with_model_access.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Fix passthrough route auth enforcement

* fix(proxy): raise passthrough-specific 403 directly in virtual-key path

* fix(proxy): load team for RBAC role-claim JWT passthrough gating

* Revert "chore(tests): migrate Bedrock CI to AWS account 941277531214 (#28728)" (#29326)

This reverts the Bedrock CI account migration (#28728). The original account
(888602223428) was put under an AWS security restriction after a leaked key
and has since been reactivated, while the replacement account (941277531214)
lacks access to several models the suites exercise (legacy Bedrock Claude 3
models, Cohere, Nova Canvas image gen, Bedrock batch inference, and flagship
Opus). Pointing CI back at the reactivated account restores that coverage.

This is the exact inverse of #28728: all hardcoded 941277531214 references go
back to 888602223428 (provisioned/imported-model ARNs, AgentCore runtime ARNs
and their suffixes, batch execution role ARN, and the example proxy config),
the S3 buckets revert to litellm-proxy and load-testing-oct, the guardrail IDs
revert to wf0hkdb5x07f and ff6ujrregl1q, the SageMaker endpoint and Knowledge
Base revert to their original ids, and the live-call tests go back to the
legacy model strings. The grid_spec fail_reason workaround for the unentitled
Opus cells is dropped while keeping the unrelated bedrock_effort_ceiling field
added after the migration.

The CircleCI AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY env vars still point at
941277531214 and must be set to the reactivated account's fresh credentials
separately via the CircleCI API; AWS_REGION_NAME stays us-west-2.

(cherry picked from commit f11c12d157)

* fix(proxy): scope pass-through 405 to registry routes; grant rerank passthrough in rpm tests

The auth=true pass-through 405 guard fired for mapped provider routes
(e.g. /assemblyai/*) that are not in the in-memory registry, since
get_registered_pass_through_route returns None for them while
is_registered_pass_through_route matches via mapped_pass_through_routes.
Only raise 405 when the path is registered but the request method is not
allowed, so mapped provider pass-throughs fall through to the default
target params as before.

The rpm-limit pass-through tests register /v1/rerank with auth=true but
gave their keys no allowed_passthrough_routes, so the new default-deny
returned 403 before the rate limiter ran (non-deterministically,
depending on registry insertion order). Grant the keys explicit
passthrough access so the tests exercise rate limiting under the new
auth model.

* fix(proxy): guard request method lookup against scopes without a method

Starlette's Request.method property reads scope["method"] and raises
KeyError when the scope omits it (e.g. minimally-constructed test
requests). getattr only swallows AttributeError, so the new
_get_request_method helper propagated the KeyError up through
user_api_key_auth and surfaced as a ProxyException. Catch KeyError
(and AttributeError) and fall back to None.

* test(passthrough): pin SERVER_ROOT_PATH in unregistered-method test

test_custom_proxy.py sets os.environ['SERVER_ROOT_PATH'] = '/my-custom-path'
at module import with no cleanup. When that module is collected into the same
xdist worker as this test, the leaked root path is prepended to registered
pass-through paths, so is_registered_pass_through_route misses '/test/path'
and the handler returns 404 instead of the expected 405 (order-dependent).
Pin SERVER_ROOT_PATH to '' so the test is deterministic.

* test(passthrough): restore regression coverage for non-auth-enforced pass-through via llm_api_routes

* fix(proxy): record auth flag in pass-through registry for allowlist enforcement

Auth-enforced pass-through detection inferred enforcement from the FastAPI
dependency stored at registration time. The management create and update
endpoints register routes with dependencies=None even though auth defaults to
true, so is_auth_enforced_pass_through_route treated those DB-created routes as
unenforced. A key allowed for llm_api_routes could then call a management-created
auth-enabled pass-through route without matching allowed_passthrough_routes.

Store the auth setting on each registry entry and read it directly when deciding
whether the allowlist applies, instead of deriving it from dependency metadata.

* fix(proxy): include bool in pass-through registry value type for auth flag

The auth flag stored in _registered_pass_through_routes is a bool, which
was not part of the registry value Union, so mypy rejected the dict literal.
Add bool to the Union and narrow route_methods to a list before the
membership check so the in-operator stays valid.

* fix(proxy): preserve stored auth flag on pass-through endpoint update

model_dump(exclude_none=True) re-included the auth=True default whenever
a partial update omitted auth, silently flipping an existing auth=false
pass-through to auth-enforced and 403ing every team/key without
allowed_passthrough_routes. Merge only explicitly set fields via
exclude_unset so omitted fields keep their stored value.

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Shivam Rawat 2026-05-30 17:07:24 -07:00 committed by GitHub
parent ba2699740c
commit dc4f5b12ef
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 1252 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
litellm/proxy/auth/handle_jwt.py
View File

@ -1110,6 +1110,40 @@ class JWTAuthManager:
return all_team_ids
@staticmethod
def _team_has_passthrough_route_access(
team_object: Optional[LiteLLM_TeamTable],
route: str,
request_method: Optional[str] = None,
) -> bool:
normalized_request_method = (
request_method.upper() if isinstance(request_method, str) else None
)
if not RouteChecks.is_auth_enforced_pass_through_route(
route=route,
method=normalized_request_method,
):
return True
# JWT team selection is team-scoped; key metadata is not available here,
# so passthrough access is granted only by the selected team's metadata.
return RouteChecks.check_passthrough_route_access(
route=route,
user_api_key_dict=UserAPIKeyAuth(
team_metadata=(team_object.metadata or {}) if team_object else {}
),
)
@staticmethod
def _raise_team_passthrough_route_denial(route: str) -> None:
raise HTTPException(
status_code=403,
detail=(
f"Team not allowed to access passthrough route {route}. "
"Configure `allowed_passthrough_routes` on the team."
),
)
@staticmethod
async def find_team_with_model_access(
team_ids: Set[str],
@ -1120,10 +1154,13 @@ class JWTAuthManager:
user_api_key_cache: UserApiKeyCache,
parent_otel_span: Optional[Span],
proxy_logging_obj: ProxyLogging,
request_method: Optional[str] = None,
) -> Tuple[Optional[str], Optional[LiteLLM_TeamTable]]:
"""Find first team with access to the requested model"""
from litellm.proxy.proxy_server import llm_router
denied_auth_enforced_pass_through_route = False
if not team_ids:
if jwt_handler.litellm_jwtauth.enforce_team_based_model_access:
raise HTTPException(
@ -1158,6 +1195,16 @@ class JWTAuthManager:
user_route=route,
litellm_proxy_roles=jwt_handler.litellm_jwtauth,
)
if (
is_allowed
and not JWTAuthManager._team_has_passthrough_route_access(
team_object=team_object,
route=route,
request_method=request_method,
)
):
is_allowed = False
denied_auth_enforced_pass_through_route = True
verbose_proxy_logger.debug(
f"JWT team route check: team_id={team_id}, route={route}, is_allowed={is_allowed}"
)
@ -1166,6 +1213,9 @@ class JWTAuthManager:
except Exception:
continue
if denied_auth_enforced_pass_through_route:
JWTAuthManager._raise_team_passthrough_route_denial(route=route)
if requested_model:
raise HTTPException(
status_code=403,
@ -1581,7 +1631,7 @@ class JWTAuthManager:
return None, None, None
@staticmethod
async def auth_builder(
async def auth_builder( # noqa: PLR0915
api_key: str,
jwt_handler: JWTHandler,
request_data: dict,
@ -1592,6 +1642,7 @@ class JWTAuthManager:
parent_otel_span: Optional[Span],
proxy_logging_obj: ProxyLogging,
request_headers: Optional[dict] = None,
request_method: Optional[str] = None,
) -> JWTAuthBuilderResult:
"""Main authentication and authorization builder"""
# Check if OIDC UserInfo endpoint is enabled, but fall back to standard
@ -1723,12 +1774,43 @@ class JWTAuthManager:
team_ids=all_team_ids,
requested_model=request_data.get("model"),
route=route,
request_method=request_method,
jwt_handler=jwt_handler,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
# The RBAC role-claim path (rbac_role == TEAM) sets team_id without
# loading team_object, so fetch it here before gating an auth-enforced
# passthrough route on the team's allowed_passthrough_routes.
if (
team_id
and team_object is None
and RouteChecks.is_auth_enforced_pass_through_route(
route=route,
method=(
request_method.upper() if isinstance(request_method, str) else None
),
)
):
team_object = await get_team_object(
team_id=team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
team_id_upsert=jwt_handler.litellm_jwtauth.team_id_upsert,
)
if team_id and not JWTAuthManager._team_has_passthrough_route_access(
team_object=team_object,
route=route,
request_method=request_method,
):
JWTAuthManager._raise_team_passthrough_route_denial(route=route)
# Extract alias fields for resolution (if configured)
org_alias = jwt_handler.get_org_alias(token=jwt_valid_token, default_value=None)

105
litellm/proxy/auth/route_checks.py
View File

@ -59,6 +59,10 @@ _PROXY_ADMIN_VIEW_ONLY_BLOCKED_ROUTES = frozenset(
# paths directly because the request route carries the resolved key id.
_PROXY_ADMIN_VIEW_ONLY_BLOCKED_KEY_SUFFIXES = ("/regenerate", "/reset_spend")
_AUTH_ENFORCED_PASS_THROUGH_ROUTE_GROUPS = frozenset(
("openai_routes", "llm_api_routes")
)
class RouteChecks:
@staticmethod
@ -103,6 +107,8 @@ class RouteChecks:
if len(valid_token.allowed_routes) == 0:
return True
denied_auth_enforced_pass_through_route = False
# explicit check for allowed routes (exact match or prefix match)
for allowed_route in valid_token.allowed_routes:
if RouteChecks._route_matches_allowed_route(
@ -121,7 +127,20 @@ class RouteChecks:
route=route,
allowed_routes=LiteLLMRoutes._member_map_[allowed_route].value,
):
return True
if (
allowed_route in _AUTH_ENFORCED_PASS_THROUGH_ROUTE_GROUPS
and RouteChecks.is_auth_enforced_pass_through_route(
route=route,
method=RouteChecks._get_request_method(request=request),
)
):
if RouteChecks.check_passthrough_route_access(
route=route, user_api_key_dict=valid_token
):
return True
denied_auth_enforced_pass_through_route = True
else:
return True
################################################
# For llm_api_routes, also check registered pass-through endpoints
@ -134,7 +153,17 @@ class RouteChecks:
if InitPassThroughEndpointHelpers.is_registered_pass_through_route(
route=route
):
return True
if RouteChecks.is_auth_enforced_pass_through_route(
route=route,
method=RouteChecks._get_request_method(request=request),
):
if RouteChecks.check_passthrough_route_access(
route=route, user_api_key_dict=valid_token
):
return True
denied_auth_enforced_pass_through_route = True
else:
return True
# Method-aware carve-out: allow GET on the two
# read-only MCP-server discovery endpoints
@ -158,6 +187,9 @@ class RouteChecks:
):
return True
if denied_auth_enforced_pass_through_route:
raise RouteChecks._auth_pass_through_denied_exception(route=route)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
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}",
@ -228,7 +260,14 @@ class RouteChecks:
route=route,
)
if RouteChecks.is_llm_api_route(route=route):
if RouteChecks.is_auth_enforced_pass_through_route(
route=route,
method=RouteChecks._get_request_method(request=request),
):
RouteChecks._require_auth_pass_through_access(
route=route, valid_token=valid_token
)
elif RouteChecks.is_llm_api_route(route=route):
pass
elif RouteChecks.is_info_route(route=route):
# check if user allowed to call an info route
@ -624,6 +663,66 @@ class RouteChecks:
return False
@staticmethod
def _get_request_method(request: Optional[Request]) -> Optional[str]:
if request is None:
return None
try:
method = request.method
except (AttributeError, KeyError):
return None
if not isinstance(method, str):
return None
return method.upper()
@staticmethod
def is_auth_enforced_pass_through_route(
route: str, method: Optional[str] = None
) -> bool:
"""
True for config/DB pass-through endpoints registered with auth=true.
These routes are injected into ``openai_routes`` for spend/budget hooks but
must not inherit blanket ``openai_routes`` RBAC; access is gated by
``allowed_passthrough_routes`` on the key or team.
"""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
)
route_info = InitPassThroughEndpointHelpers.get_registered_pass_through_route(
route=route, method=method
)
if route_info is None:
return False
return route_info.get("auth") is True
@staticmethod
def _auth_pass_through_denied_exception(route: str) -> HTTPException:
return HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=(
f"Key/team not allowed to access passthrough route {route}. "
"Configure `allowed_passthrough_routes` on the team or key."
),
)
@staticmethod
def _require_auth_pass_through_access(
route: str,
valid_token: UserAPIKeyAuth,
) -> None:
"""
Require an explicit ``allowed_passthrough_routes`` match for auth=true pass-through.
"""
if RouteChecks.check_passthrough_route_access(
route=route, user_api_key_dict=valid_token
):
return
raise RouteChecks._auth_pass_through_denied_exception(route=route)
@staticmethod
def check_passthrough_route_access(
route: str, user_api_key_dict: UserAPIKeyAuth

3
litellm/proxy/auth/user_api_key_auth.py
View File

@ -925,6 +925,9 @@ async def _user_api_key_auth_builder( # noqa: PLR0915
proxy_logging_obj=proxy_logging_obj,
parent_otel_span=parent_otel_span,
request_headers=_safe_get_request_headers(request),
request_method=RouteChecks._get_request_method(
request=request
),
)
is_proxy_admin = result["is_proxy_admin"]

47
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View File

@ -78,7 +78,7 @@ pass_through_endpoint_logging = PassThroughEndpointLogging()
# Global registry to track registered pass-through routes and prevent memory leaks
_registered_pass_through_routes: Dict[
str, Dict[str, Union[str, List[str], Dict[str, Any]]]
str, Dict[str, Union[str, bool, List[str], Dict[str, Any]]]
] = {}
@ -1539,6 +1539,17 @@ def create_pass_through_route(
route=path, method=request.method
)
)
if (
passthrough_params is None
and InitPassThroughEndpointHelpers.get_registered_pass_through_route(
route=path
)
is not None
):
raise HTTPException(
status_code=status.HTTP_405_METHOD_NOT_ALLOWED,
detail=f"Method {request.method} is not allowed for pass-through endpoint {path}.",
)
target_params = {
"target": target,
"custom_headers": custom_headers,
@ -2266,6 +2277,7 @@ class InitPassThroughEndpointHelpers:
methods: Optional[List[str]] = None,
default_query_params: Optional[dict] = None,
config_file_path: Optional[str] = None,
auth: bool = False,
):
"""Add exact path route for pass-through endpoint"""
# Default to all methods if none specified (backward compatibility)
@ -2317,6 +2329,7 @@ class InitPassThroughEndpointHelpers:
"path": path,
"type": "exact",
"methods": methods,
"auth": auth,
"passthrough_params": {
"target": target,
"custom_headers": custom_headers,
@ -2344,6 +2357,7 @@ class InitPassThroughEndpointHelpers:
methods: Optional[List[str]] = None,
default_query_params: Optional[dict] = None,
config_file_path: Optional[str] = None,
auth: bool = False,
):
"""Add wildcard route for sub-paths"""
# Default to all methods if none specified (backward compatibility)
@ -2396,6 +2410,7 @@ class InitPassThroughEndpointHelpers:
"path": path,
"type": "subpath",
"methods": methods,
"auth": auth,
"passthrough_params": {
"target": target,
"custom_headers": custom_headers,
@ -2514,8 +2529,15 @@ class InitPassThroughEndpointHelpers:
InitPassThroughEndpointHelpers._build_full_path_with_root(parts[2])
)
# Get the methods for this route
route_methods = _registered_pass_through_routes[key].get("methods", [])
# Get the methods for this route. Prefer the registered metadata,
# but keep supporting test fixtures / older registry entries that
# only encoded methods in the route key.
methods_entry = _registered_pass_through_routes[key].get("methods", [])
route_methods: List[str] = (
methods_entry if isinstance(methods_entry, list) else []
)
if not route_methods and len(parts) == 4:
route_methods = parts[3].split(",")
# Check if path matches
path_matches = False
@ -2573,8 +2595,9 @@ async def _register_pass_through_endpoint(
default_query_params = endpoint_data.get("default_query_params")
auth = endpoint_data.get("auth")
dependencies = None
auth_enforced = auth is not None and str(auth).lower() == "true"
if auth is not None and str(auth).lower() == "true":
if auth_enforced:
# Authentication on a pass-through endpoint used to be enterprise-only.
# That left OSS with no safe configuration: auth=True raised at startup
# unless the operator had a license. The safe option must always be free,
@ -2607,6 +2630,7 @@ async def _register_pass_through_endpoint(
methods=methods,
default_query_params=default_query_params,
config_file_path=config_file_path,
auth=auth_enforced,
)
methods_for_key = methods if methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
@ -2632,6 +2656,7 @@ async def _register_pass_through_endpoint(
methods=methods,
default_query_params=default_query_params,
config_file_path=config_file_path,
auth=auth_enforced,
)
visited_endpoints.add(f"{endpoint_id}:subpath:{path}:{methods_str}")
@ -2955,14 +2980,18 @@ async def update_pass_through_endpoints(
},
)
# Get the update data as dict, excluding None values for partial updates
# Only merge fields the caller explicitly sent so omitted fields keep their
# stored value. Without exclude_unset, defaults like auth=True would overwrite
# an existing auth=false entry on any unrelated edit.
# Exclude is_from_config as it's a response-only field (computed at read time)
update_data = data.model_dump(exclude_none=True, exclude={"is_from_config"})
update_data = data.model_dump(
exclude_unset=True, exclude_none=True, exclude={"is_from_config"}
)
# Start with existing endpoint data
endpoint_dict = found_endpoint.model_dump()
# Update with new data (only non-None values)
# Update with new data (only explicitly provided values)
endpoint_dict.update(update_data)
# Preserve existing ID if not provided in update and endpoint has ID
@ -3010,6 +3039,7 @@ async def update_pass_through_endpoints(
guardrails=getattr(updated_endpoint, "guardrails", None),
methods=updated_endpoint.methods,
default_query_params=updated_endpoint.default_query_params,
auth=updated_endpoint.auth,
)
else:
InitPassThroughEndpointHelpers.add_exact_path_route(
@ -3025,6 +3055,7 @@ async def update_pass_through_endpoints(
guardrails=getattr(updated_endpoint, "guardrails", None),
methods=updated_endpoint.methods,
default_query_params=updated_endpoint.default_query_params,
auth=updated_endpoint.auth,
)
return PassThroughEndpointResponse(
@ -3103,6 +3134,7 @@ async def create_pass_through_endpoints(
guardrails=getattr(created_endpoint, "guardrails", None),
methods=created_endpoint.methods,
default_query_params=created_endpoint.default_query_params,
auth=created_endpoint.auth,
)
else:
InitPassThroughEndpointHelpers.add_exact_path_route(
@ -3118,6 +3150,7 @@ async def create_pass_through_endpoints(
guardrails=getattr(created_endpoint, "guardrails", None),
methods=created_endpoint.methods,
default_query_params=created_endpoint.default_query_params,
auth=created_endpoint.auth,
)
return PassThroughEndpointResponse(endpoints=[created_endpoint])

1
tests/llm_translation/reasoning_effort_grid/grid_spec.py
View File

@ -1,6 +1,7 @@
from dataclasses import dataclass, field
from typing import Dict, FrozenSet, List, Optional, Tuple
OMIT = object()

1
tests/llm_translation/reasoning_effort_grid/test_reasoning_effort_grid.py
View File

@ -15,6 +15,7 @@ from .grid_spec import (
all_cells,
)
_PROMPT_MESSAGES: List[Dict[str, str]] = [
{"role": "user", "content": "Step by step, calculate 47 * 53. Show your work."}
]

8
tests/local_testing/test_pass_through_endpoints.py
View File

@ -218,7 +218,9 @@ async def test_pass_through_endpoint_rpm_limit(
for mock_api_key in mock_api_keys:
cache_value = UserAPIKeyAuth(
token=hash_token(mock_api_key), rpm_limit=rpm_limit
token=hash_token(mock_api_key),
rpm_limit=rpm_limit,
metadata={"allowed_passthrough_routes": ["/v1/rerank"]},
)
user_api_key_cache.set_cache(key=hash_token(mock_api_key), value=cache_value)
@ -320,7 +322,9 @@ async def test_pass_through_endpoint_sequential_rpm_limit(
for mock_api_key in mock_api_keys:
cache_value = UserAPIKeyAuth(
token=hash_token(mock_api_key), rpm_limit=rpm_limit
token=hash_token(mock_api_key),
rpm_limit=rpm_limit,
metadata={"allowed_passthrough_routes": ["/v1/rerank"]},
)
user_api_key_cache.set_cache(key=hash_token(mock_api_key), value=cache_value)

425
tests/test_litellm/proxy/auth/test_handle_jwt.py
View File

@ -1,6 +1,7 @@
from typing import Optional
from unittest.mock import AsyncMock, MagicMock, patch
from fastapi import HTTPException
import pytest
from litellm.proxy._types import (
@ -132,6 +133,141 @@ async def test_map_user_to_teams_null_inputs():
await JWTAuthManager.map_user_to_teams(user_object=None, team_object=None)
@pytest.mark.asyncio
async def test_find_team_with_model_access_reports_passthrough_allowlist_denial():
jwt_handler = JWTHandler()
jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth()
team = LiteLLM_TeamTable(
team_id="team-a",
models=["gpt-4"],
metadata={},
)
with (
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team,
),
patch(
"litellm.proxy.auth.handle_jwt.can_team_access_model",
new_callable=AsyncMock,
return_value=True,
),
patch(
"litellm.proxy.auth.handle_jwt.allowed_routes_check",
return_value=True,
),
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.is_auth_enforced_pass_through_route",
return_value=True,
) as mock_is_auth_enforced_pass_through_route,
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.check_passthrough_route_access",
return_value=False,
) as mock_passthrough_check,
):
with pytest.raises(HTTPException) as exc_info:
await JWTAuthManager.find_team_with_model_access(
team_ids={"team-a"},
requested_model="gpt-4",
route="/my-pass-through",
request_method="POST",
jwt_handler=jwt_handler,
prisma_client=None,
user_api_key_cache=MagicMock(),
parent_otel_span=None,
proxy_logging_obj=MagicMock(),
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
assert "requested model" not in exc_info.value.detail
mock_is_auth_enforced_pass_through_route.assert_called_once_with(
route="/my-pass-through", method="POST"
)
user_api_key_dict = mock_passthrough_check.call_args.kwargs["user_api_key_dict"]
assert user_api_key_dict.metadata == {}
assert user_api_key_dict.team_metadata == {}
@pytest.mark.asyncio
async def test_find_team_with_model_access_uses_request_method_for_passthrough_auth():
jwt_handler = JWTHandler()
jwt_handler.litellm_jwtauth = LiteLLM_JWTAuth()
team = LiteLLM_TeamTable(
team_id="team-a",
models=["gpt-4"],
metadata={},
)
mock_registered_routes = {
"test-uuid-1:exact:/custom:GET": {
"endpoint_id": "test-uuid-1",
"path": "/custom",
"type": "exact",
"methods": ["GET"],
"auth": False,
},
"test-uuid-2:exact:/custom:POST": {
"endpoint_id": "test-uuid-2",
"path": "/custom",
"type": "exact",
"methods": ["POST"],
"auth": True,
},
}
with (
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team,
),
patch(
"litellm.proxy.auth.handle_jwt.allowed_routes_check",
return_value=True,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
mock_registered_routes,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
team_id, team_obj = await JWTAuthManager.find_team_with_model_access(
team_ids={"team-a"},
requested_model=None,
route="/custom",
jwt_handler=jwt_handler,
prisma_client=None,
user_api_key_cache=MagicMock(),
parent_otel_span=None,
proxy_logging_obj=MagicMock(),
request_method="GET",
)
assert team_id == "team-a"
assert team_obj == team
with pytest.raises(HTTPException) as exc_info:
await JWTAuthManager.find_team_with_model_access(
team_ids={"team-a"},
requested_model=None,
route="/custom",
jwt_handler=jwt_handler,
prisma_client=None,
user_api_key_cache=MagicMock(),
parent_otel_span=None,
proxy_logging_obj=MagicMock(),
request_method="POST",
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
@pytest.mark.asyncio
async def test_auth_builder_proxy_admin_user_role():
"""Test that is_proxy_admin is True when user_object.user_role is PROXY_ADMIN"""
@ -1576,6 +1712,295 @@ async def test_auth_builder_uses_team_from_header_e2e():
assert result["team_object"] == team_object
@pytest.mark.asyncio
async def test_auth_builder_header_team_denies_auth_passthrough_without_allowlist():
"""Header-selected JWT teams must enforce team allowed_passthrough_routes."""
from litellm.caching import DualCache
from litellm.proxy.utils import ProxyLogging
jwt_handler = JWTHandler()
user_api_key_cache = DualCache()
jwt_handler.update_environment(
prisma_client=None,
user_api_key_cache=user_api_key_cache,
litellm_jwtauth=LiteLLM_JWTAuth(
team_ids_jwt_field="groups",
user_id_jwt_field="sub",
),
)
team_object = LiteLLM_TeamTable(team_id="team-2", metadata={})
with (
patch.object(jwt_handler, "auth_jwt", new_callable=AsyncMock) as mock_auth_jwt,
patch.object(JWTAuthManager, "check_rbac_role", new_callable=AsyncMock),
patch.object(
JWTAuthManager,
"check_admin_access",
new_callable=AsyncMock,
return_value=None,
),
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team_object,
),
patch.object(
JWTAuthManager,
"get_objects",
new_callable=AsyncMock,
) as mock_get_objects,
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.is_auth_enforced_pass_through_route",
return_value=True,
),
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.check_passthrough_route_access",
return_value=False,
) as mock_passthrough_check,
):
mock_auth_jwt.return_value = {
"sub": "user-1",
"scope": "",
"groups": ["team-1", "team-2"],
}
with pytest.raises(HTTPException) as exc_info:
await JWTAuthManager.auth_builder(
api_key="jwt-token",
jwt_handler=jwt_handler,
request_data={"model": "gpt-4"},
general_settings={},
route="/my-pass-through",
prisma_client=None,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=ProxyLogging(user_api_key_cache=user_api_key_cache),
request_headers={"x-litellm-team-id": "team-2"},
request_method="POST",
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
mock_get_objects.assert_not_called()
user_api_key_dict = mock_passthrough_check.call_args.kwargs["user_api_key_dict"]
assert user_api_key_dict.team_metadata == {}
@pytest.mark.asyncio
async def test_auth_builder_specific_team_denies_auth_passthrough_without_allowlist():
"""JWT-field-selected teams must enforce team allowed_passthrough_routes."""
from litellm.caching import DualCache
from litellm.proxy.utils import ProxyLogging
jwt_handler = JWTHandler()
user_api_key_cache = DualCache()
jwt_handler.update_environment(
prisma_client=None,
user_api_key_cache=user_api_key_cache,
litellm_jwtauth=LiteLLM_JWTAuth(
team_id_jwt_field="team_id",
user_id_jwt_field="sub",
),
)
team_object = LiteLLM_TeamTable(team_id="team-1", metadata={})
with (
patch.object(jwt_handler, "auth_jwt", new_callable=AsyncMock) as mock_auth_jwt,
patch.object(JWTAuthManager, "check_rbac_role", new_callable=AsyncMock),
patch.object(
JWTAuthManager,
"check_admin_access",
new_callable=AsyncMock,
return_value=None,
),
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team_object,
),
patch.object(
JWTAuthManager,
"get_objects",
new_callable=AsyncMock,
) as mock_get_objects,
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.is_auth_enforced_pass_through_route",
return_value=True,
),
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.check_passthrough_route_access",
return_value=False,
) as mock_passthrough_check,
):
mock_auth_jwt.return_value = {
"sub": "user-1",
"scope": "",
"team_id": "team-1",
}
with pytest.raises(HTTPException) as exc_info:
await JWTAuthManager.auth_builder(
api_key="jwt-token",
jwt_handler=jwt_handler,
request_data={"model": "gpt-4"},
general_settings={},
route="/my-pass-through",
prisma_client=None,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=ProxyLogging(user_api_key_cache=user_api_key_cache),
request_method="POST",
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
mock_get_objects.assert_not_called()
user_api_key_dict = mock_passthrough_check.call_args.kwargs["user_api_key_dict"]
assert user_api_key_dict.team_metadata == {}
@pytest.mark.asyncio
async def test_auth_builder_rbac_team_loads_team_for_passthrough_allowlist():
"""RBAC role-claim teams (team_object unset) must load team metadata before gating."""
from litellm.caching import DualCache
from litellm.proxy.utils import ProxyLogging
jwt_handler = JWTHandler()
user_api_key_cache = DualCache()
jwt_handler.update_environment(
prisma_client=None,
user_api_key_cache=user_api_key_cache,
litellm_jwtauth=LiteLLM_JWTAuth(),
)
team_object = LiteLLM_TeamTable(
team_id="team-rbac",
metadata={"allowed_passthrough_routes": ["/my-pass-through"]},
)
with (
patch.object(jwt_handler, "auth_jwt", new_callable=AsyncMock) as mock_auth_jwt,
patch.object(jwt_handler, "get_rbac_role", return_value=LitellmUserRoles.TEAM),
patch.object(jwt_handler, "get_object_id", return_value="team-rbac"),
patch.object(JWTAuthManager, "check_rbac_role", new_callable=AsyncMock),
patch.object(
JWTAuthManager,
"check_admin_access",
new_callable=AsyncMock,
return_value=None,
),
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team_object,
) as mock_get_team,
patch.object(
JWTAuthManager,
"get_objects",
new_callable=AsyncMock,
return_value=(None, None, None, None),
),
patch.object(JWTAuthManager, "map_user_to_teams", new_callable=AsyncMock),
patch.object(
JWTAuthManager, "sync_user_role_and_teams", new_callable=AsyncMock
),
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.is_auth_enforced_pass_through_route",
return_value=True,
),
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.check_passthrough_route_access",
return_value=True,
) as mock_passthrough_check,
):
mock_auth_jwt.return_value = {"scope": ""}
result = await JWTAuthManager.auth_builder(
api_key="jwt-token",
jwt_handler=jwt_handler,
request_data={"model": "gpt-4"},
general_settings={},
route="/my-pass-through",
prisma_client=None,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=ProxyLogging(user_api_key_cache=user_api_key_cache),
request_method="POST",
)
assert result["team_id"] == "team-rbac"
mock_get_team.assert_awaited_once()
assert mock_get_team.await_args.kwargs["team_id"] == "team-rbac"
user_api_key_dict = mock_passthrough_check.call_args.kwargs["user_api_key_dict"]
assert user_api_key_dict.team_metadata == {
"allowed_passthrough_routes": ["/my-pass-through"]
}
@pytest.mark.asyncio
async def test_auth_builder_rbac_team_denies_passthrough_without_allowlist():
"""RBAC role-claim teams without an allowlist are still denied for passthrough."""
from litellm.caching import DualCache
from litellm.proxy.utils import ProxyLogging
jwt_handler = JWTHandler()
user_api_key_cache = DualCache()
jwt_handler.update_environment(
prisma_client=None,
user_api_key_cache=user_api_key_cache,
litellm_jwtauth=LiteLLM_JWTAuth(),
)
team_object = LiteLLM_TeamTable(team_id="team-rbac", metadata={})
with (
patch.object(jwt_handler, "auth_jwt", new_callable=AsyncMock) as mock_auth_jwt,
patch.object(jwt_handler, "get_rbac_role", return_value=LitellmUserRoles.TEAM),
patch.object(jwt_handler, "get_object_id", return_value="team-rbac"),
patch.object(JWTAuthManager, "check_rbac_role", new_callable=AsyncMock),
patch.object(
JWTAuthManager,
"check_admin_access",
new_callable=AsyncMock,
return_value=None,
),
patch(
"litellm.proxy.auth.handle_jwt.get_team_object",
new_callable=AsyncMock,
return_value=team_object,
) as mock_get_team,
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.is_auth_enforced_pass_through_route",
return_value=True,
),
patch(
"litellm.proxy.auth.handle_jwt.RouteChecks.check_passthrough_route_access",
return_value=False,
),
):
mock_auth_jwt.return_value = {"scope": ""}
with pytest.raises(HTTPException) as exc_info:
await JWTAuthManager.auth_builder(
api_key="jwt-token",
jwt_handler=jwt_handler,
request_data={"model": "gpt-4"},
general_settings={},
route="/my-pass-through",
prisma_client=None,
user_api_key_cache=user_api_key_cache,
parent_otel_span=None,
proxy_logging_obj=ProxyLogging(user_api_key_cache=user_api_key_cache),
request_method="POST",
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
mock_get_team.assert_awaited_once()
@pytest.mark.asyncio
async def test_auth_builder_admin_on_llm_route_honors_team_header():
"""JWT proxy_admin + x-litellm-team-id on an LLM API route -> team context is

306
tests/test_litellm/proxy/auth/test_route_checks.py
View File

@ -224,6 +224,28 @@ def test_virtual_key_mcp_routes_allows_v1_mcp_server():
assert result is True
def test_auth_enforced_passthrough_check_does_not_apply_to_info_routes():
"""Auth-enforced passthrough gating only applies to OpenAI/LLM route groups."""
valid_token = UserAPIKeyAuth(
user_id="test_user",
allowed_routes=["info_routes"],
)
with patch.object(
RouteChecks,
"is_auth_enforced_pass_through_route",
return_value=True,
) as mock_is_auth_enforced_pass_through_route:
result = RouteChecks.is_virtual_key_allowed_to_call_route(
route="/team/info",
valid_token=valid_token,
)
assert result is True
mock_is_auth_enforced_pass_through_route.assert_not_called()
@pytest.mark.parametrize(
"route",
[
@ -686,24 +708,88 @@ def test_anthropic_count_tokens_route_accessible_to_internal_users():
def test_virtual_key_llm_api_routes_allows_registered_pass_through_endpoints():
"""
Test that virtual keys with llm_api_routes permission can access registered pass-through endpoints.
This tests the scenario where a pass-through endpoint is registered from the DB
(e.g., /azure-assistant) and a virtual key with llm_api_routes permission should be able to access
both the exact path and subpaths (e.g., /azure-assistant/openai/assistants).
Virtual keys with llm_api_routes can access auth=true pass-through endpoints only when
allowed_passthrough_routes is configured on the key or team.
"""
# Mock the registered pass-through routes
mock_registered_routes = {
"test-uuid-1:exact:/azure-assistant": {
"test-uuid-1:exact:/azure-assistant:DELETE,GET,PATCH,POST,PUT": {
"endpoint_id": "test-uuid-1",
"path": "/azure-assistant",
"type": "exact",
"auth": True,
},
"test-uuid-2:subpath:/custom-endpoint": {
"test-uuid-2:subpath:/custom-endpoint:DELETE,GET,PATCH,POST,PUT": {
"endpoint_id": "test-uuid-2",
"path": "/custom-endpoint",
"type": "subpath",
"auth": True,
},
}
with (
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
mock_registered_routes,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
valid_token = UserAPIKeyAuth(
user_id="test_user",
allowed_routes=["llm_api_routes"],
metadata={
"allowed_passthrough_routes": [
"/azure-assistant",
"/custom-endpoint",
]
},
)
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/azure-assistant",
valid_token=valid_token,
)
is True
)
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom-endpoint/openai/assistants",
valid_token=valid_token,
)
is True
)
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom-endpoint",
valid_token=valid_token,
)
is True
)
def test_virtual_key_llm_api_routes_allows_non_auth_enforced_pass_through_endpoints():
"""
Virtual keys with llm_api_routes can access registered pass-through endpoints that
are NOT auth-enforced (auth=false) without configuring allowed_passthrough_routes.
This is the original behaviour and must not regress.
"""
mock_registered_routes = {
"test-uuid-1:exact:/azure-assistant:DELETE,GET,PATCH,POST,PUT": {
"endpoint_id": "test-uuid-1",
"path": "/azure-assistant",
"type": "exact",
"auth": False,
},
"test-uuid-2:subpath:/custom-endpoint:DELETE,GET,PATCH,POST,PUT": {
"endpoint_id": "test-uuid-2",
"path": "/custom-endpoint",
"type": "subpath",
"auth": False,
},
}
@ -717,32 +803,202 @@ def test_virtual_key_llm_api_routes_allows_registered_pass_through_endpoints():
return_value="/",
),
):
# Create a virtual key with llm_api_routes permission
valid_token = UserAPIKeyAuth(
user_id="test_user",
allowed_routes=["llm_api_routes"],
)
# Test exact match for registered pass-through endpoint
result1 = RouteChecks.is_virtual_key_allowed_to_call_route(
route="/azure-assistant",
valid_token=valid_token,
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/azure-assistant",
valid_token=valid_token,
)
is True
)
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom-endpoint/openai/assistants",
valid_token=valid_token,
)
is True
)
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom-endpoint",
valid_token=valid_token,
)
is True
)
assert result1 is True
# Test subpath for registered pass-through endpoint with subpath type
result2 = RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom-endpoint/openai/assistants",
valid_token=valid_token,
)
assert result2 is True
# Test exact match for subpath type
result3 = RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom-endpoint",
valid_token=valid_token,
def test_virtual_key_llm_api_routes_denies_auth_pass_through_without_allowlist():
"""auth=true pass-through must not be reachable via llm_api_routes alone."""
mock_registered_routes = {
"test-uuid-1:exact:/azure-assistant:GET,POST": {
"endpoint_id": "test-uuid-1",
"path": "/azure-assistant",
"type": "exact",
"auth": True,
},
}
with (
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
mock_registered_routes,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
valid_token = UserAPIKeyAuth(
user_id="test_user",
allowed_routes=["llm_api_routes"],
)
with pytest.raises(HTTPException) as exc_info:
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/azure-assistant",
valid_token=valid_token,
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
def test_virtual_key_llm_api_routes_uses_method_specific_auth_setting():
"""Same-path pass-through routes must be checked against the request method."""
mock_registered_routes = {
"test-uuid-1:exact:/custom:GET": {
"endpoint_id": "test-uuid-1",
"path": "/custom",
"type": "exact",
"methods": ["GET"],
"auth": False,
},
"test-uuid-2:exact:/custom:POST": {
"endpoint_id": "test-uuid-2",
"path": "/custom",
"type": "exact",
"methods": ["POST"],
"auth": True,
},
}
with (
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
mock_registered_routes,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
valid_token = UserAPIKeyAuth(
user_id="test_user",
allowed_routes=["llm_api_routes"],
)
get_request = MagicMock(spec=Request)
get_request.method = "GET"
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom",
valid_token=valid_token,
request=get_request,
)
is True
)
post_request = MagicMock(spec=Request)
post_request.method = "POST"
with pytest.raises(HTTPException) as exc_info:
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/custom",
valid_token=valid_token,
request=post_request,
)
assert exc_info.value.status_code == 403
def test_non_proxy_admin_denies_auth_pass_through_without_allowlist():
"""Internal users must not bypass allowed_passthrough_routes via openai_routes."""
mock_registered_routes = {
"test-uuid-1:exact:/my-pass-through:GET,POST": {
"endpoint_id": "test-uuid-1",
"path": "/my-pass-through",
"type": "exact",
"auth": True,
},
}
valid_token = UserAPIKeyAuth(
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
)
with (
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
mock_registered_routes,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
with pytest.raises(HTTPException) as exc_info:
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=None,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route="/my-pass-through",
request=MagicMock(spec=Request),
valid_token=valid_token,
request_data={},
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
def test_non_proxy_admin_allows_auth_pass_through_with_team_allowlist():
mock_registered_routes = {
"test-uuid-1:exact:/my-pass-through:GET,POST": {
"endpoint_id": "test-uuid-1",
"path": "/my-pass-through",
"type": "exact",
"auth": True,
},
}
valid_token = UserAPIKeyAuth(
user_id="test_user",
user_role=LitellmUserRoles.INTERNAL_USER.value,
team_metadata={"allowed_passthrough_routes": ["/my-pass-through"]},
)
with (
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
mock_registered_routes,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=None,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route="/my-pass-through",
request=MagicMock(spec=Request),
valid_token=valid_token,
request_data={},
)
assert result3 is True
def test_virtual_key_without_llm_api_routes_cannot_access_pass_through():

2
tests/test_litellm/proxy/auth/test_user_api_key_auth.py
View File

@ -1513,6 +1513,7 @@ class TestJWTOAuth2Coexistence:
mock_request = MagicMock()
mock_request.url.path = "/v1/chat/completions"
mock_request.method = "POST"
mock_request.headers = {"authorization": f"Bearer {jwt_token}"}
mock_request.query_params = {}
@ -1546,6 +1547,7 @@ class TestJWTOAuth2Coexistence:
mock_oauth2.assert_not_called()
# JWT auth SHOULD be called
mock_jwt_auth.assert_called_once()
assert mock_jwt_auth.call_args.kwargs["request_method"] == "POST"
assert result.user_id == "jwt-human-user"
@pytest.mark.asyncio

308
tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py
View File

@ -519,6 +519,64 @@ def test_add_subpath_route():
assert callable(call_args["endpoint"])
@pytest.mark.asyncio
async def test_pass_through_handler_rejects_unregistered_method():
"""
Stale FastAPI routes can remain after an endpoint is updated from all methods
to a restricted method list. The handler must enforce the current registry.
"""
from fastapi import HTTPException
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
create_pass_through_route,
)
endpoint_func = create_pass_through_route(
endpoint="/test/path",
target="http://example.com",
)
request = MagicMock(spec=Request)
request.method = "GET"
with (
patch.dict(os.environ, {"SERVER_ROOT_PATH": ""}),
patch(
"litellm.proxy.auth.auth_utils.get_request_route",
return_value="/test/path",
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._parse_request_data_by_content_type",
new_callable=AsyncMock,
return_value=({}, {}, None, False),
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
{
"test-endpoint-id:exact:/test/path:POST": {
"endpoint_id": "test-endpoint-id",
"path": "/test/path",
"type": "exact",
"methods": ["POST"],
"passthrough_params": {
"target": "http://example.com",
"custom_headers": {},
"forward_headers": False,
"merge_query_params": False,
},
}
},
),
):
with pytest.raises(HTTPException) as exc_info:
await endpoint_func(
request=request,
fastapi_response=MagicMock(),
user_api_key_dict=MagicMock(),
)
assert exc_info.value.status_code == 405
@pytest.mark.asyncio
async def test_initialize_pass_through_endpoints_with_include_subpath():
"""
@ -1171,6 +1229,256 @@ async def test_update_pass_through_endpoint():
assert updated_data["cost_per_request"] == 0.75
@pytest.mark.asyncio
async def test_create_pass_through_endpoint_auth_true_enforces_allowlist():
"""
Regression: a pass-through endpoint created through the management API with
auth=true (the model default) must be treated as allowlist-enforced. The
create path registers FastAPI routes with dependencies=None, so deriving
enforcement from dependency metadata let a key with broad llm_api_routes
access call the route without an allowed_passthrough_routes match.
"""
from fastapi import HTTPException
from litellm.proxy._types import (
ConfigFieldInfo,
PassThroughGenericEndpoint,
UserAPIKeyAuth,
)
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
create_pass_through_endpoints,
)
registry: dict = {}
with (
patch(
"litellm.proxy.proxy_server.get_config_general_settings"
) as mock_get_config,
patch("litellm.proxy.proxy_server.update_config_general_settings"),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
registry,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
mock_get_config.return_value = ConfigFieldInfo(
field_name="pass_through_endpoints", field_value=[]
)
# auth is not passed -> defaults to True on PassThroughGenericEndpoint
endpoint = PassThroughGenericEndpoint(
path="/secure-passthrough",
target="http://example.com/api",
methods=["POST"],
)
await create_pass_through_endpoints(
data=endpoint,
request=MagicMock(spec=Request),
user_api_key_dict=MagicMock(spec=UserAPIKeyAuth),
)
assert any(value.get("auth") is True for value in registry.values())
assert (
RouteChecks.is_auth_enforced_pass_through_route(
route="/secure-passthrough", method="POST"
)
is True
)
post_request = MagicMock(spec=Request)
post_request.method = "POST"
without_allowlist = UserAPIKeyAuth(
user_id="u", allowed_routes=["llm_api_routes"]
)
with pytest.raises(HTTPException) as exc_info:
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/secure-passthrough",
valid_token=without_allowlist,
request=post_request,
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
with_allowlist = UserAPIKeyAuth(
user_id="u",
allowed_routes=["llm_api_routes"],
metadata={"allowed_passthrough_routes": ["/secure-passthrough"]},
)
assert (
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/secure-passthrough",
valid_token=with_allowlist,
request=post_request,
)
is True
)
@pytest.mark.asyncio
async def test_update_pass_through_endpoint_auth_true_enforces_allowlist():
"""
Regression: editing a pass-through endpoint through the management API must
keep an auth=true route allowlist-enforced. remove_endpoint_routes drops the
old registry entry, so the re-registration has to record the auth flag.
"""
from fastapi import HTTPException
from litellm.proxy._types import (
ConfigFieldInfo,
PassThroughGenericEndpoint,
UserAPIKeyAuth,
)
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
update_pass_through_endpoints,
)
registry: dict = {}
existing_endpoint_id = "edit-me-123"
existing_endpoints = [
{
"id": existing_endpoint_id,
"path": "/edited-passthrough",
"target": "http://example.com/api",
"auth": True,
"methods": ["POST"],
}
]
with (
patch(
"litellm.proxy.proxy_server.get_config_general_settings"
) as mock_get_config,
patch("litellm.proxy.proxy_server.update_config_general_settings"),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
registry,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
mock_get_config.return_value = ConfigFieldInfo(
field_name="pass_through_endpoints", field_value=existing_endpoints
)
update_data = PassThroughGenericEndpoint(
path="/edited-passthrough",
target="http://newapi.com/v2",
methods=["POST"],
)
await update_pass_through_endpoints(
endpoint_id=existing_endpoint_id,
data=update_data,
request=MagicMock(spec=Request),
user_api_key_dict=MagicMock(spec=UserAPIKeyAuth),
)
assert (
RouteChecks.is_auth_enforced_pass_through_route(
route="/edited-passthrough", method="POST"
)
is True
)
post_request = MagicMock(spec=Request)
post_request.method = "POST"
without_allowlist = UserAPIKeyAuth(
user_id="u", allowed_routes=["llm_api_routes"]
)
with pytest.raises(HTTPException) as exc_info:
RouteChecks.is_virtual_key_allowed_to_call_route(
route="/edited-passthrough",
valid_token=without_allowlist,
request=post_request,
)
assert exc_info.value.status_code == 403
assert "allowed_passthrough_routes" in exc_info.value.detail
@pytest.mark.asyncio
async def test_update_pass_through_endpoint_preserves_auth_false():
"""
Regression: editing an unrelated field on an auth=false pass-through must not
silently flip it to auth=true. auth defaults to True on the request model, so a
naive exclude_none merge would overwrite the stored auth=false and start
rejecting every team/key that lacks allowed_passthrough_routes.
"""
from litellm.proxy._types import (
ConfigFieldInfo,
PassThroughGenericEndpoint,
UserAPIKeyAuth,
)
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
update_pass_through_endpoints,
)
registry: dict = {}
existing_endpoint_id = "public-forwarder-123"
existing_endpoints = [
{
"id": existing_endpoint_id,
"path": "/public-passthrough",
"target": "http://example.com/api",
"auth": False,
"methods": ["POST"],
}
]
with (
patch(
"litellm.proxy.proxy_server.get_config_general_settings"
) as mock_get_config,
patch(
"litellm.proxy.proxy_server.update_config_general_settings"
) as mock_update_config,
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
registry,
),
patch(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
return_value="/",
),
):
mock_get_config.return_value = ConfigFieldInfo(
field_name="pass_through_endpoints", field_value=existing_endpoints
)
update_data = PassThroughGenericEndpoint(
path="/public-passthrough",
target="http://newapi.com/v2",
methods=["POST"],
)
result = await update_pass_through_endpoints(
endpoint_id=existing_endpoint_id,
data=update_data,
request=MagicMock(spec=Request),
user_api_key_dict=MagicMock(spec=UserAPIKeyAuth),
)
assert result.endpoints[0].auth is False
persisted = mock_update_config.call_args[1]["data"].field_value[0]
assert persisted["auth"] is False
assert (
RouteChecks.is_auth_enforced_pass_through_route(
route="/public-passthrough", method="POST"
)
is False
)
@pytest.mark.asyncio
async def test_update_pass_through_endpoint_not_found():
"""