diff --git a/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py b/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py index cebd224a1a7..1794cd14381 100644 --- a/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py +++ b/litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py @@ -33,10 +33,12 @@ def get_request_base_url(request: Request) -> str: """ Get the base URL for the request, considering X-Forwarded-* headers. - When behind a proxy (like nginx), the proxy may set: - - X-Forwarded-Proto: The original protocol (http/https) - - X-Forwarded-Host: The original host (may include port) - - X-Forwarded-Port: The original port (if not in Host header) + X-Forwarded-Proto / X-Forwarded-Host / X-Forwarded-Port are only honoured + when the request comes from a configured trusted proxy + (``use_x_forwarded_for`` enabled AND caller in ``mcp_trusted_proxy_ranges``). + Otherwise the request's literal ``base_url`` is returned, so an + untrusted caller cannot poison OAuth-discovery / redirect_uri values + by injecting headers. Args: request: FastAPI Request object @@ -47,34 +49,28 @@ def get_request_base_url(request: Request) -> str: base_url = str(request.base_url).rstrip("/") parsed = urlparse(base_url) - # Get forwarded headers + if not IPAddressUtils.is_request_from_trusted_proxy(request): + return base_url + x_forwarded_proto = request.headers.get("X-Forwarded-Proto") x_forwarded_host = request.headers.get("X-Forwarded-Host") x_forwarded_port = request.headers.get("X-Forwarded-Port") - # Start with the original scheme scheme = x_forwarded_proto if x_forwarded_proto else parsed.scheme - # Handle host and port if x_forwarded_host: # X-Forwarded-Host may already include port (e.g., "example.com:8080") if ":" in x_forwarded_host and not x_forwarded_host.startswith("["): - # Host includes port netloc = x_forwarded_host elif x_forwarded_port: - # Port is separate netloc = f"{x_forwarded_host}:{x_forwarded_port}" else: - # Just host, no explicit port netloc = x_forwarded_host else: - # No X-Forwarded-Host, use original netloc netloc = parsed.netloc if x_forwarded_port and ":" not in netloc: - # Add forwarded port if not already in netloc netloc = f"{netloc}:{x_forwarded_port}" - # Reconstruct the URL return urlunparse((scheme, netloc, parsed.path, "", "", "")) diff --git a/litellm/proxy/auth/ip_address_utils.py b/litellm/proxy/auth/ip_address_utils.py index 34fab4849e5..39d3282942f 100644 --- a/litellm/proxy/auth/ip_address_utils.py +++ b/litellm/proxy/auth/ip_address_utils.py @@ -13,6 +13,10 @@ from fastapi import Request from litellm._logging import verbose_proxy_logger from litellm.proxy.auth.auth_utils import _get_request_ip_address +# One-shot warning so operators upgrading from the prior "always trust X-Forwarded-*" +# behaviour see an actionable message in their logs the first time it triggers. +_warned_xff_without_trusted_ranges = False + class IPAddressUtils: """Static utilities for IP-based MCP access control.""" @@ -106,6 +110,61 @@ class IPAddressUtils: return any(addr in network for network in networks) + @staticmethod + def is_request_from_trusted_proxy( + request: Request, + general_settings: Optional[Dict[str, Any]] = None, + ) -> bool: + """ + Return True if X-Forwarded-* headers on this request should be trusted. + + Trusts the headers iff both: + 1. ``use_x_forwarded_for`` is enabled in proxy settings, AND + 2. ``mcp_trusted_proxy_ranges`` is configured AND the direct + connection IP (``request.client.host``) falls inside one of + those CIDRs. + + When ``use_x_forwarded_for`` is enabled but ``mcp_trusted_proxy_ranges`` + is missing, the headers are NOT trusted: there is no way to + distinguish a trusted reverse proxy from a direct attacker, so callers + that build URLs (OAuth issuer / redirect_uri / etc.) must fall back + to the request's literal base URL instead of risking a poisoned host. + """ + if general_settings is None: + try: + from litellm.proxy.proxy_server import ( + general_settings as proxy_general_settings, + ) + + general_settings = proxy_general_settings + except ImportError: + general_settings = {} + + if general_settings is None: + general_settings = {} + + if not general_settings.get("use_x_forwarded_for", False): + return False + + trusted_ranges = general_settings.get("mcp_trusted_proxy_ranges") + if not trusted_ranges: + global _warned_xff_without_trusted_ranges + if not _warned_xff_without_trusted_ranges: + verbose_proxy_logger.warning( + "use_x_forwarded_for is enabled but mcp_trusted_proxy_ranges " + "is not configured. X-Forwarded-* headers will NOT be " + "trusted, so MCP OAuth discovery URLs will use the proxy's " + "literal base URL. Set mcp_trusted_proxy_ranges in " + "general_settings to your reverse-proxy CIDR(s) to allow " + "X-Forwarded-* through." + ) + _warned_xff_without_trusted_ranges = True + return False + + direct_ip = request.client.host if request.client else None + trusted_networks = IPAddressUtils.parse_trusted_proxy_networks(trusted_ranges) + return IPAddressUtils.is_trusted_proxy(direct_ip, trusted_networks) + @staticmethod def get_mcp_client_ip( request: Request, diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py index 558c677d2dc..85d5d6ba466 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py @@ -23,6 +23,21 @@ def mock_mcp_client_ip(): yield +@pytest.fixture +def trust_xff(): + """Force ``IPAddressUtils.is_request_from_trusted_proxy`` to True. + + Tests that exercise X-Forwarded-* parsing logic opt into this fixture. + The trust gate's own behaviour is covered by + ``test_get_request_base_url_xff_trust_gate``. + """ + with patch( + "litellm.proxy._experimental.mcp_server.discoverable_endpoints.IPAddressUtils.is_request_from_trusted_proxy", + return_value=True, + ): + yield + + @pytest.mark.asyncio async def test_authorize_endpoint_includes_response_type(): """Test that authorize endpoint includes response_type=code parameter (fixes #15684)""" @@ -505,6 +520,7 @@ async def test_register_client_remote_registration_success(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_authorize_endpoint_respects_x_forwarded_proto(): """Test that authorize endpoint uses X-Forwarded-Proto header to construct correct redirect_uri""" try: @@ -572,6 +588,7 @@ async def test_authorize_endpoint_respects_x_forwarded_proto(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_token_endpoint_respects_x_forwarded_proto(): """Test that token endpoint uses X-Forwarded-Proto header for redirect_uri""" try: @@ -650,6 +667,7 @@ async def test_token_endpoint_respects_x_forwarded_proto(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_oauth_protected_resource_respects_x_forwarded_proto(): """Test that oauth_protected_resource_mcp uses X-Forwarded-Proto for URLs""" try: @@ -704,6 +722,7 @@ async def test_oauth_protected_resource_respects_x_forwarded_proto(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_oauth_authorization_server_respects_x_forwarded_proto(): """Test that oauth_authorization_server_mcp uses X-Forwarded-Proto for URLs""" try: @@ -759,6 +778,7 @@ async def test_oauth_authorization_server_respects_x_forwarded_proto(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_register_client_respects_x_forwarded_proto(): """Test that register_client uses X-Forwarded-Proto for redirect_uris""" try: @@ -796,6 +816,7 @@ async def test_register_client_respects_x_forwarded_proto(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_authorize_endpoint_respects_x_forwarded_host(): """Test that authorize endpoint uses X-Forwarded-Host and X-Forwarded-Proto to construct correct redirect_uri""" try: @@ -869,6 +890,7 @@ async def test_authorize_endpoint_respects_x_forwarded_host(): @pytest.mark.asyncio +@pytest.mark.usefixtures("trust_xff") async def test_token_endpoint_respects_x_forwarded_host(): """Test that token endpoint uses X-Forwarded-Host and X-Forwarded-Proto for redirect_uri""" try: @@ -1071,7 +1093,12 @@ async def test_token_endpoint_respects_x_forwarded_host(): def test_get_request_base_url_comprehensive( base_url, x_forwarded_proto, x_forwarded_host, x_forwarded_port, expected_url ): - """Comprehensive test for get_request_base_url with various header combinations""" + """Comprehensive test for get_request_base_url with various header combinations. + + These cases exercise the X-Forwarded-* parsing logic, so the trust gate + is patched True; the gate's own behaviour is covered by the + ``test_get_request_base_url_xff_trust_gate`` matrix below. + """ try: from fastapi import Request @@ -1081,11 +1108,9 @@ def test_get_request_base_url_comprehensive( except ImportError: pytest.skip("MCP discoverable endpoints not available") - # Create mock request mock_request = MagicMock(spec=Request) mock_request.base_url = base_url - # Build headers dict headers = {} if x_forwarded_proto: headers["X-Forwarded-Proto"] = x_forwarded_proto @@ -1094,16 +1119,17 @@ def test_get_request_base_url_comprehensive( if x_forwarded_port: headers["X-Forwarded-Port"] = x_forwarded_port - # Mock headers.get() to return our test values def mock_get(header_name, default=None): return headers.get(header_name, default) mock_request.headers.get = mock_get - # Test the function - result = get_request_base_url(mock_request) + with patch( + "litellm.proxy._experimental.mcp_server.discoverable_endpoints.IPAddressUtils.is_request_from_trusted_proxy", + return_value=True, + ): + result = get_request_base_url(mock_request) - # Verify result assert result == expected_url, ( f"Expected '{expected_url}' but got '{result}'\n" f"Input: base_url={base_url}, " @@ -1113,6 +1139,131 @@ def test_get_request_base_url_comprehensive( ) +@pytest.mark.parametrize( + "general_settings,direct_ip,expect_xff_honoured", + [ + # Default: use_x_forwarded_for not set -> ignore X-Forwarded-* entirely. + ({}, "127.0.0.1", False), + # XFF enabled, no trusted ranges -> still ignored (no way to tell a trusted + # reverse proxy from a direct attacker). + ({"use_x_forwarded_for": True}, "127.0.0.1", False), + # XFF enabled, ranges set, but caller IP outside any range -> ignored. + ( + { + "use_x_forwarded_for": True, + "mcp_trusted_proxy_ranges": ["10.0.0.0/8"], + }, + "203.0.113.5", + False, + ), + # XFF enabled, caller in trusted range -> headers honoured. + ( + { + "use_x_forwarded_for": True, + "mcp_trusted_proxy_ranges": ["10.0.0.0/8"], + }, + "10.0.0.7", + True, + ), + # Loopback example (common dev / single-host deploy). + ( + { + "use_x_forwarded_for": True, + "mcp_trusted_proxy_ranges": ["127.0.0.0/8"], + }, + "127.0.0.1", + True, + ), + ], +) +def test_get_request_base_url_xff_trust_gate( + general_settings, direct_ip, expect_xff_honoured +): + """Verify the X-Forwarded-* trust gate. + + With XFF poisoning attempted, the helper must return either the literal + base_url (gate denies) or the forwarded URL (gate allows), never the + forwarded URL when the gate denies. + """ + try: + from fastapi import Request + + from litellm.proxy._experimental.mcp_server.discoverable_endpoints import ( + get_request_base_url, + ) + except ImportError: + pytest.skip("MCP discoverable endpoints not available") + + mock_request = MagicMock(spec=Request) + mock_request.base_url = "http://localhost:4000/" + mock_request.client = MagicMock() + mock_request.client.host = direct_ip + + headers = { + "X-Forwarded-Proto": "https", + "X-Forwarded-Host": "attacker.example.com", + } + mock_request.headers.get = lambda name, default=None: headers.get(name, default) + mock_request.headers.__contains__ = lambda self_, name: name in headers + + with patch( + "litellm.proxy.proxy_server.general_settings", + general_settings, + create=True, + ): + result = get_request_base_url(mock_request) + + if expect_xff_honoured: + assert result == "https://attacker.example.com" + else: + assert result == "http://localhost:4000" + + +def test_xff_misconfig_warning_emitted_once(caplog): + """Operators upgrading from the old "always trust X-Forwarded-*" behaviour + get a one-shot warning when they have ``use_x_forwarded_for`` enabled + but no ``mcp_trusted_proxy_ranges`` configured. The warning must NOT + spam every request.""" + try: + from fastapi import Request + + from litellm.proxy import auth as proxy_auth_pkg # noqa: F401 + from litellm.proxy._experimental.mcp_server.discoverable_endpoints import ( + get_request_base_url, + ) + from litellm.proxy.auth import ip_address_utils + except ImportError: + pytest.skip("MCP discoverable endpoints not available") + + # Reset the module-level one-shot flag so the test is deterministic. + ip_address_utils._warned_xff_without_trusted_ranges = False + + mock_request = MagicMock(spec=Request) + mock_request.base_url = "http://localhost:4000/" + mock_request.client = MagicMock() + mock_request.client.host = "203.0.113.5" + headers = {"X-Forwarded-Host": "attacker.example.com"} + mock_request.headers.get = lambda name, default=None: headers.get(name, default) + + misconfig = {"use_x_forwarded_for": True} + + import logging + + with ( + caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"), + patch("litellm.proxy.proxy_server.general_settings", misconfig, create=True), + ): + for _ in range(3): + get_request_base_url(mock_request) + + matching = [ + rec for rec in caplog.records if "mcp_trusted_proxy_ranges" in rec.getMessage() + ] + assert ( + len(matching) == 1 + ), f"expected exactly one warning, got {len(matching)}: {[r.getMessage() for r in matching]}" + + # ------------------------------------------------------------------- # Tests for scopes_supported when mcp_server.scopes is None # -------------------------------------------------------------------