[Fix] Fix Pass through routes to work with server root path (#19383)

* test_build_full_path_with_root_default

* fix pt feat
This commit is contained in:
Ishaan Jaff 2026-01-19 18:28:55 -08:00 committed by GitHub
parent 270b41b0f4
commit 818913ee23
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 173 additions and 5 deletions

View file

@ -51,6 +51,7 @@ from litellm.proxy._types import (
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_request_processing import ProxyBaseLLMRequestProcessing
from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
from litellm.proxy.utils import get_server_root_path
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.custom_http import httpxSpecialProvider
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
@ -1973,10 +1974,26 @@ class InitPassThroughEndpointHelpers:
_registered_pass_through_routes.clear()
@staticmethod
def get_registered_pass_through_endpoints_keys() -> List[str]:
def get_all_registered_pass_through_routes() -> List[str]:
"""Get all registered pass-through endpoints from the registry"""
return list(_registered_pass_through_routes.keys())
@staticmethod
def _build_full_path_with_root(path: str) -> str:
"""
Build full path by prepending server root path if needed.
Args:
path: The relative path to build
Returns:
Full path with server root prepended (if root is not "/")
"""
root_path = get_server_root_path()
if root_path == "/":
return path
return f"{root_path}{path}"
@staticmethod
def is_registered_pass_through_route(route: str) -> bool:
"""
@ -2003,7 +2020,9 @@ class InitPassThroughEndpointHelpers:
parts = key.split(":", 2) # Split into [endpoint_id, type, path]
if len(parts) == 3:
route_type = parts[1]
registered_path = parts[2]
registered_path = InitPassThroughEndpointHelpers._build_full_path_with_root(
parts[2]
)
if route_type == "exact" and route == registered_path:
return True
elif route_type == "subpath":
@ -2021,7 +2040,9 @@ class InitPassThroughEndpointHelpers:
parts = key.split(":", 2) # Split into [endpoint_id, type, path]
if len(parts) == 3:
route_type = parts[1]
registered_path = parts[2]
registered_path = InitPassThroughEndpointHelpers._build_full_path_with_root(
parts[2]
)
if route_type == "exact" and route == registered_path:
return _registered_pass_through_routes[key]
@ -2085,7 +2106,7 @@ async def initialize_pass_through_endpoints(
# mark the ones that are visited in the list
# remove the ones that are not visited from the list
registered_pass_through_endpoints = (
InitPassThroughEndpointHelpers.get_registered_pass_through_endpoints_keys()
InitPassThroughEndpointHelpers.get_all_registered_pass_through_routes()
)
visited_endpoints = set()

View file

@ -1897,8 +1897,8 @@ async def test_add_litellm_data_to_request_adds_headers_to_metadata():
The fix ensures headers are available in data["metadata"]["headers"] so
guardrails can validate User-Agent, API keys, and other header-based checks.
"""
from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
# Create mock request with headers including User-Agent
mock_request = MagicMock(spec=Request)
@ -1954,3 +1954,150 @@ async def test_add_litellm_data_to_request_adds_headers_to_metadata():
# Also verify proxy_server_request has headers (original location)
assert "proxy_server_request" in result
assert "headers" in result["proxy_server_request"]
def test_build_full_path_with_root_default():
"""
Test _build_full_path_with_root with default root path (/)
"""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
)
with patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path") as mock_get_root:
# Test with default root path
mock_get_root.return_value = "/"
result = InitPassThroughEndpointHelpers._build_full_path_with_root("/api/v1/endpoint")
assert result == "/api/v1/endpoint"
def test_build_full_path_with_root_custom():
"""
Test _build_full_path_with_root with custom root path
"""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
)
with patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path") as mock_get_root:
# Test with custom root path /proxy
mock_get_root.return_value = "/proxy"
result = InitPassThroughEndpointHelpers._build_full_path_with_root("/api/v1/endpoint")
assert result == "/proxy/api/v1/endpoint"
def test_build_full_path_with_root_nested():
"""
Test _build_full_path_with_root with nested root path
"""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
)
with patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path") as mock_get_root:
# Test with nested root path /api/v2
mock_get_root.return_value = "/api/v2"
result = InitPassThroughEndpointHelpers._build_full_path_with_root("/endpoint")
assert result == "/api/v2/endpoint"
def test_is_registered_pass_through_route_with_custom_root():
"""
Test is_registered_pass_through_route correctly handles server root path
When server has a custom root path like /proxy, the registered path
should be constructed by prepending the root to match incoming routes.
"""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
_registered_pass_through_routes,
)
# Clear the registry first
_registered_pass_through_routes.clear()
# Register a pass-through route with endpoint format: {endpoint_id}:exact:{path}
endpoint_id = "test-endpoint-123"
path = "/api/endpoint"
route_key = f"{endpoint_id}:exact:{path}"
_registered_pass_through_routes[route_key] = {
"target": "http://example.com",
"headers": {},
}
with patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path") as mock_get_root:
# Test with custom root path /proxy
mock_get_root.return_value = "/proxy"
# Should match when request route includes the root path
assert InitPassThroughEndpointHelpers.is_registered_pass_through_route("/proxy/api/endpoint") is True
# Should not match when request route doesn't include root path
assert InitPassThroughEndpointHelpers.is_registered_pass_through_route("/api/endpoint") is False
# Test with default root path
mock_get_root.return_value = "/"
# Should match with default root
assert InitPassThroughEndpointHelpers.is_registered_pass_through_route("/api/endpoint") is True
# Should not match with root prepended when root is /
assert InitPassThroughEndpointHelpers.is_registered_pass_through_route("/proxy/api/endpoint") is False
# Clean up
_registered_pass_through_routes.clear()
def test_get_registered_pass_through_route_with_custom_root():
"""
Test get_registered_pass_through_route correctly handles server root path
When server has a custom root path, the method should return the correct
endpoint configuration by matching the full path including the root.
"""
from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
InitPassThroughEndpointHelpers,
_registered_pass_through_routes,
)
# Clear the registry first
_registered_pass_through_routes.clear()
# Register a pass-through route
endpoint_id = "test-endpoint-456"
path = "/chat/completions"
target_config = {
"target": "http://api.example.com/v1/chat/completions",
"headers": {"Authorization": "Bearer token123"},
"forward_headers": True,
}
route_key = f"{endpoint_id}:exact:{path}"
_registered_pass_through_routes[route_key] = target_config
with patch("litellm.proxy.pass_through_endpoints.pass_through_endpoints.get_server_root_path") as mock_get_root:
# Test with custom root path /litellm
mock_get_root.return_value = "/litellm"
# Should return config when request route includes root path
result = InitPassThroughEndpointHelpers.get_registered_pass_through_route("/litellm/chat/completions")
assert result is not None
assert result["target"] == "http://api.example.com/v1/chat/completions"
assert result["headers"]["Authorization"] == "Bearer token123"
# Should return None when route doesn't match
result = InitPassThroughEndpointHelpers.get_registered_pass_through_route("/chat/completions")
assert result is None
# Test with default root path
mock_get_root.return_value = "/"
# Should return config with default root
result = InitPassThroughEndpointHelpers.get_registered_pass_through_route("/chat/completions")
assert result is not None
assert result["target"] == "http://api.example.com/v1/chat/completions"
# Clean up
_registered_pass_through_routes.clear()