scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-22 00:31:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 51

fix(proxy): fail closed on pass-through route collisions (#27339)

Browse source 
Squash-merged by litellm-agent from stuxf's PR.
This commit is contained in:
stuxf 2026-05-11 18:54:20 -07:00 • committed by GitHub
parent a4442be11a
commit 47c0dd2e11
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 293 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/proxy/auth/user_api_key_auth.py
View file

@ -558,6 +558,19 @@ async def check_api_key_for_custom_headers_or_pass_through_endpoints(
# is also True, but raw config dicts skip that path —
# so this runtime check has to default to True too.
if endpoint.get("auth", True) is not True:
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
)
registered_route = InitPassThroughEndpointHelpers.get_registered_pass_through_route(
route=route, method=request.method
)
if registered_route is None:
verbose_proxy_logger.warning(
"Ignoring auth=false for pass-through endpoint %s because it is not registered as a pass-through route",
route,
)
continue
return UserAPIKeyAuth()
## IF AUTH ENABLED
### IF CUSTOM PARSER REQUIRED

143
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -1380,6 +1380,7 @@ def create_pass_through_route(
if hasattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY):
delattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY)
setattr(endpoint_func, "_litellm_pass_through_route", True)
return endpoint_func
@ -1938,28 +1939,61 @@ class SafeRouteAdder:
"""
@staticmethod
def _is_path_registered(app: FastAPI, path: str, methods: List[str]) -> bool:
"""
Check if a path with any of the specified methods is already registered on the app.
def _get_matching_routes_by_method(
app: FastAPI, path: str, methods: List[str]
) -> Dict[str, Any]:
"""
Get the first existing FastAPI route for each path/method pair.
Args:
app: The FastAPI application instance
path: The path to check (e.g., "/v1/chat/completions")
methods: List of HTTP methods to check (e.g., ["GET", "POST"])
Returns:
True if the path is already registered with any of the methods, False otherwise
"""
for route in app.routes:
# Use getattr to safely access route attributes
route_path = getattr(route, "path", None)
route_methods = getattr(route, "methods", None)
Returns:
Mapping of method to the route that would receive that method.
"""
routes_by_method: Dict[str, Any] = {}
for method in {method.upper() for method in methods}:
for route in app.routes:
route_path = getattr(route, "path", None)
route_methods = getattr(route, "methods", None)
if route_path == path and route_methods is not None:
if method in route_methods:
routes_by_method[method] = route
break
return routes_by_method
if route_path == path and route_methods is not None:
# Check if any of the methods overlap
if any(method in route_methods for method in methods):
return True
return False
@staticmethod
def _is_path_registered(app: FastAPI, path: str, methods: List[str]) -> bool:
"""
Check if a path with any of the specified methods is already registered on the app.
"""
return bool(
SafeRouteAdder._get_matching_routes_by_method(
app=app, path=path, methods=methods
)
)
@staticmethod
def _is_pass_through_route(route: Any) -> bool:
route_endpoint = getattr(route, "endpoint", None)
return bool(getattr(route_endpoint, "_litellm_pass_through_route", False))
@staticmethod
def is_registered_route_pass_through(
app: FastAPI, path: str, methods: List[str]
) -> bool:
"""
Check whether the existing FastAPI route was created by the pass-through router.
"""
routes_by_method = SafeRouteAdder._get_matching_routes_by_method(
app=app, path=path, methods=methods
)
return bool(routes_by_method) and all(
SafeRouteAdder._is_pass_through_route(route)
for route in routes_by_method.values()
)
@staticmethod
def add_api_route_if_not_exists(
@ -2019,7 +2053,7 @@ class InitPassThroughEndpointHelpers:
guardrails: Optional[dict] = None,
methods: Optional[List[str]] = None,
default_query_params: Optional[dict] = None,
):
) -> bool:
"""Add exact path route for pass-through endpoint"""
# Default to all methods if none specified (backward compatibility)
if methods is None or len(methods) == 0:
@ -2045,7 +2079,7 @@ class InitPassThroughEndpointHelpers:
)
# Use SafeRouteAdder to only add route if it doesn't exist on the app
SafeRouteAdder.add_api_route_if_not_exists(
route_added = SafeRouteAdder.add_api_route_if_not_exists(
app=app,
path=path,
endpoint=create_pass_through_route( # type: ignore
@ -2063,7 +2097,21 @@ class InitPassThroughEndpointHelpers:
dependencies=dependencies,
)
# Always register/update the route metadata (headers, target) even if FastAPI route exists
if (
route_added is False
and not SafeRouteAdder.is_registered_route_pass_through(
app=app, path=path, methods=methods
)
):
_registered_pass_through_routes.pop(route_key, None)
verbose_proxy_logger.warning(
"Skipping pass-through metadata registration for %s with methods %s because the path is already registered by another route",
path,
methods,
)
return False
# Register/update metadata only when the path is backed by a pass-through handler.
_registered_pass_through_routes[route_key] = {
"endpoint_id": endpoint_id,
"path": path,
@ -2080,6 +2128,7 @@ class InitPassThroughEndpointHelpers:
"guardrails": guardrails,
},
}
return True
@staticmethod
def add_subpath_route(
@ -2095,7 +2144,7 @@ class InitPassThroughEndpointHelpers:
guardrails: Optional[dict] = None,
methods: Optional[List[str]] = None,
default_query_params: Optional[dict] = None,
):
) -> bool:
"""Add wildcard route for sub-paths"""
# Default to all methods if none specified (backward compatibility)
if methods is None or len(methods) == 0:
@ -2121,7 +2170,7 @@ class InitPassThroughEndpointHelpers:
)
# Use SafeRouteAdder to only add route if it doesn't exist on the app
SafeRouteAdder.add_api_route_if_not_exists(
route_added = SafeRouteAdder.add_api_route_if_not_exists(
app=app,
path=wildcard_path,
endpoint=create_pass_through_route( # type: ignore
@ -2140,7 +2189,21 @@ class InitPassThroughEndpointHelpers:
dependencies=dependencies,
)
# Register the route to prevent duplicates only if it was added
if (
route_added is False
and not SafeRouteAdder.is_registered_route_pass_through(
app=app, path=wildcard_path, methods=methods
)
):
_registered_pass_through_routes.pop(route_key, None)
verbose_proxy_logger.warning(
"Skipping pass-through metadata registration for %s with methods %s because the path is already registered by another route",
wildcard_path,
methods,
)
return False
# Register/update metadata only when the path is backed by a pass-through handler.
_registered_pass_through_routes[route_key] = {
"endpoint_id": endpoint_id,
"path": path,
@ -2157,6 +2220,7 @@ class InitPassThroughEndpointHelpers:
"guardrails": guardrails,
},
}
return True
@staticmethod
def remove_endpoint_routes(endpoint_id: str):
@ -2323,14 +2387,13 @@ async def _register_pass_through_endpoint(
auth = endpoint_data.get("auth")
dependencies = None
if auth is not None and str(auth).lower() == "true":
# 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,
# and unauthenticated forwarding should require explicit opt-in.
auth_enabled = auth is not None and str(auth).lower() == "true"
if auth_enabled:
# 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,
# and unauthenticated forwarding should require explicit opt-in.
dependencies = [Depends(user_api_key_auth)]
if path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(path)
if target is None:
return
@ -2342,7 +2405,7 @@ async def _register_pass_through_endpoint(
verbose_proxy_logger.debug(
"Initializing pass through endpoint: %s (ID: %s)", path, endpoint_id
)
InitPassThroughEndpointHelpers.add_exact_path_route(
exact_route_registered = InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path=path,
target=target,
@ -2356,17 +2419,20 @@ async def _register_pass_through_endpoint(
methods=methods,
default_query_params=default_query_params,
)
if (
auth_enabled
and exact_route_registered
and path not in LiteLLMRoutes.openai_routes.value
):
LiteLLMRoutes.openai_routes.value.append(path)
methods_for_key = methods if methods else ["GET", "POST", "PUT", "DELETE", "PATCH"]
methods_str = ",".join(sorted(methods_for_key))
visited_endpoints.add(f"{endpoint_id}:exact:{path}:{methods_str}")
if exact_route_registered:
visited_endpoints.add(f"{endpoint_id}:exact:{path}:{methods_str}")
if endpoint_data.get("include_subpath", False) is True:
if auth is not None and str(auth).lower() == "true":
wildcard_path = path.rstrip("/") + "/*"
if wildcard_path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(wildcard_path)
InitPassThroughEndpointHelpers.add_subpath_route(
subpath_route_registered = InitPassThroughEndpointHelpers.add_subpath_route(
app=app,
path=path,
target=target,
@ -2380,7 +2446,12 @@ async def _register_pass_through_endpoint(
methods=methods,
default_query_params=default_query_params,
)
visited_endpoints.add(f"{endpoint_id}:subpath:{path}:{methods_str}")
if auth_enabled and subpath_route_registered:
wildcard_path = path.rstrip("/") + "/*"
if wildcard_path not in LiteLLMRoutes.openai_routes.value:
LiteLLMRoutes.openai_routes.value.append(wildcard_path)
if subpath_route_registered:
visited_endpoints.add(f"{endpoint_id}:subpath:{path}:{methods_str}")
verbose_proxy_logger.debug(
"Added new pass through endpoint: %s (ID: %s)", path, endpoint_id

179
tests/test_litellm/proxy/pass_through_endpoints/test_passthrough_auth_default.py
View file

@ -33,10 +33,23 @@ from litellm.proxy.auth.user_api_key_auth import (
check_api_key_for_custom_headers_or_pass_through_endpoints,
)
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
SafeRouteAdder,
_register_pass_through_endpoint,
)
@pytest.fixture(autouse=True)
def reset_passthrough_route_state(monkeypatch):
monkeypatch.setattr(
"litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path",
lambda: "/",
)
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
yield
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
def test_passthrough_auth_defaults_to_true():
# Regression: an admin who configures a pass-through without setting
# auth explicitly used to ship an unauthenticated forwarder. The
@ -120,17 +133,171 @@ async def test_runtime_check_explicit_auth_false_still_skips_validation():
request = MagicMock()
request.headers = {}
request.method = "POST"
raw_endpoint_auth_false = {
"path": "/public-webhook",
"target": "https://example.com",
"auth": False,
}
result = await check_api_key_for_custom_headers_or_pass_through_endpoints(
request=request,
route="/public-webhook",
pass_through_endpoints=[raw_endpoint_auth_false],
api_key="",
)
app = FastAPI()
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
try:
InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path="/public-webhook",
target="https://example.com",
custom_headers=None,
forward_headers=None,
merge_query_params=None,
dependencies=None,
cost_per_request=None,
endpoint_id="public-webhook",
methods=["POST"],
)
result = await check_api_key_for_custom_headers_or_pass_through_endpoints(
request=request,
route="/public-webhook",
pass_through_endpoints=[raw_endpoint_auth_false],
api_key="",
)
finally:
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
assert isinstance(result, UserAPIKeyAuth)
def test_colliding_passthrough_route_does_not_register_metadata():
app = FastAPI()
@app.post("/customer/block")
async def existing_management_route():
return {"ok": True}
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
try:
route_registered = InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path="/customer/block",
target="https://example.com",
custom_headers=None,
forward_headers=None,
merge_query_params=None,
dependencies=None,
cost_per_request=None,
endpoint_id="colliding-forwarder",
methods=["POST"],
)
assert route_registered is False
assert (
InitPassThroughEndpointHelpers.get_registered_pass_through_route(
route="/customer/block", method="POST"
)
is None
)
finally:
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
@pytest.mark.asyncio
async def test_unregistered_auth_false_passthrough_does_not_skip_validation():
request = MagicMock()
request.headers = {}
request.method = "POST"
raw_endpoint_auth_false = {
"path": "/customer/block",
"target": "https://example.com",
"auth": False,
}
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
result = await check_api_key_for_custom_headers_or_pass_through_endpoints(
request=request,
route="/customer/block",
pass_through_endpoints=[raw_endpoint_auth_false],
api_key="sk-1234",
)
assert result == "sk-1234"
def test_existing_passthrough_route_metadata_can_be_updated():
app = FastAPI()
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
try:
first_registration = InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path="/forwarder",
target="https://example.com/old",
custom_headers=None,
forward_headers=None,
merge_query_params=None,
dependencies=None,
cost_per_request=None,
endpoint_id="forwarder",
methods=["POST"],
)
second_registration = InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path="/forwarder",
target="https://example.com/new",
custom_headers=None,
forward_headers=None,
merge_query_params=None,
dependencies=None,
cost_per_request=None,
endpoint_id="forwarder",
methods=["POST"],
)
route_info = InitPassThroughEndpointHelpers.get_registered_pass_through_route(
route="/forwarder", method="POST"
)
assert first_registration is True
assert second_registration is True
assert route_info is not None
assert route_info["passthrough_params"]["target"] == "https://example.com/new"
finally:
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
def test_passthrough_route_detection_is_method_aware_for_split_paths():
app = FastAPI()
@app.get("/split-route")
async def existing_get_route():
return {"ok": True}
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()
try:
route_registered = InitPassThroughEndpointHelpers.add_exact_path_route(
app=app,
path="/split-route",
target="https://example.com",
custom_headers=None,
forward_headers=None,
merge_query_params=None,
dependencies=None,
cost_per_request=None,
endpoint_id="split-route-forwarder",
methods=["POST"],
)
assert route_registered is True
assert (
SafeRouteAdder.is_registered_route_pass_through(
app=app, path="/split-route", methods=["POST"]
)
is True
)
assert (
SafeRouteAdder.is_registered_route_pass_through(
app=app, path="/split-route", methods=["GET", "POST"]
)
is False
)
finally:
InitPassThroughEndpointHelpers.clear_all_pass_through_routes()