fix(mcp): stop 404ing internal-only MCP servers when XFF is on without trusted proxy ranges

litellm/proxy/auth/ip_address_utils.py

@@ -13,9 +13,10 @@
 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.
+# One-shot warnings so operators upgrading from the prior "always trust X-Forwarded-*"
+# behaviour see an actionable message in their logs the first time each path triggers.
 _warned_xff_without_trusted_ranges = False
+_warned_xff_access_control_without_trusted_ranges = False
 
 
 class IPAddressUtils:
@@ -152,12 +153,11 @@ def is_request_from_trusted_proxy(
             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 and access-control "
-                    "client IPs will use the proxy's literal request values. "
-                    "Set mcp_trusted_proxy_ranges in "
-                    "general_settings to your reverse-proxy CIDR(s) to allow "
-                    "X-Forwarded-* through."
+                    "is not configured. X-Forwarded-Host/Proto will NOT be "
+                    "trusted when building MCP OAuth discovery URLs, which fall "
+                    "back to the proxy's literal request 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
@@ -174,9 +174,16 @@ def get_mcp_client_ip(
         """
         Extract client IP from a FastAPI request for MCP access control.
 
-        Security: Only trusts X-Forwarded-For if:
-        1. use_x_forwarded_for is enabled in settings
-        2. The direct connection is from a trusted proxy (if mcp_trusted_proxy_ranges configured)
+        Honors X-Forwarded-For whenever ``use_x_forwarded_for`` is enabled, so a
+        client behind a reverse proxy is classified by its real (forwarded) IP
+        rather than the proxy's peer address. When ``mcp_trusted_proxy_ranges``
+        is also configured, the forwarded header is only honored if the direct
+        peer falls inside a trusted range; an untrusted peer falls back to its
+        literal IP so a spoofed header cannot grant internal access. With
+        ``use_x_forwarded_for`` enabled but no trusted ranges configured the
+        header is trusted as-is (matching the proxy's wider X-Forwarded-For
+        contract), and a one-shot warning is logged so operators know
+        access control rests on an unvalidated header.
 
         Args:
             request: FastAPI request object
@@ -197,22 +204,41 @@ def get_mcp_client_ip(
             general_settings = {}
 
         use_xff = general_settings.get("use_x_forwarded_for", False)
-
-        # If XFF is enabled, validate the request comes from a trusted proxy
-        if use_xff and "x-forwarded-for" in request.headers:
-            if not IPAddressUtils.is_request_from_trusted_proxy(
+        xff_present = use_xff and "x-forwarded-for" in request.headers
+        trusted_ranges_configured = bool(
+            general_settings.get("mcp_trusted_proxy_ranges")
+        )
+
+        if (
+            xff_present
+            and trusted_ranges_configured
+            and not IPAddressUtils.is_request_from_trusted_proxy(
                 request, general_settings=general_settings
-            ):
-                direct_ip = request.client.host if request.client else None
-                if general_settings.get("mcp_trusted_proxy_ranges"):
-                    # Direct connection isn't in any configured trusted CIDR.
-                    verbose_proxy_logger.warning(
-                        "XFF header from untrusted IP %s, ignoring", direct_ip
-                    )
-                    return direct_ip
-                # XFF enabled but no trusted proxy ranges configured: the direct
-                # peer is typically the reverse proxy's own (private) IP, so
-                # returning it would mis-classify external callers as internal.
-                # Fail closed for access control.
-                return ""
+            )
+        ):
+            direct_ip = request.client.host if request.client else None
+            verbose_proxy_logger.warning(
+                "XFF header from untrusted IP %s, ignoring", direct_ip
+            )
+            return direct_ip
+
+        if xff_present and not trusted_ranges_configured:
+            IPAddressUtils._warn_xff_trusted_for_access_control()
+
         return _get_request_ip_address(request, use_x_forwarded_for=use_xff)
+
+    @staticmethod
+    def _warn_xff_trusted_for_access_control() -> None:
+        global _warned_xff_access_control_without_trusted_ranges
+        if _warned_xff_access_control_without_trusted_ranges:
+            return
+        verbose_proxy_logger.warning(
+            "use_x_forwarded_for is enabled but mcp_trusted_proxy_ranges is not "
+            "configured. X-Forwarded-For is trusted as-is for MCP IP-based "
+            "access control, so a caller with direct network access to the proxy "
+            "could spoof an internal IP and reach available_on_public_internet="
+            "false servers. Set mcp_trusted_proxy_ranges in general_settings to "
+            "your reverse-proxy CIDR(s) to validate the proxy before trusting "
+            "X-Forwarded-For."
+        )
+        _warned_xff_access_control_without_trusted_ranges = True

tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py

@@ -60,7 +60,11 @@ def test_fails_closed_on_bad_input(self):
 
 
 class TestMCPClientIPExtraction:
-    def test_fails_closed_when_xff_enabled_without_trusted_proxy_ranges(self):
+    def test_honours_xff_for_internal_client_without_trusted_proxy_ranges(self):
+        # LIT-3964 regression: use_x_forwarded_for enabled, no mcp_trusted_proxy_ranges,
+        # internal client behind a load balancer. The forwarded client IP must be
+        # honored (matching use_x_forwarded_for semantics and pre-1.89 behavior) so
+        # the caller is classified internal, not fail-closed to "" and then 404'd.
         request = MagicMock(spec=Request)
         request.client = MagicMock()
         request.client.host = "203.0.113.5"
@@ -71,17 +75,13 @@ def test_fails_closed_when_xff_enabled_without_trusted_proxy_ranges(self):
             general_settings={"use_x_forwarded_for": True},
         )
 
-        # XFF is untrusted (no mcp_trusted_proxy_ranges) so it must be ignored,
-        # and we must not trust the direct peer either: fail closed so the caller
-        # is classified as external and is_internal_ip("") is False.
-        assert result == ""
-        assert IPAddressUtils.is_internal_ip(result) is False
+        assert result == "10.0.0.1"
+        assert IPAddressUtils.is_internal_ip(result) is True
 
-    def test_private_proxy_peer_does_not_grant_internal_access(self):
-        # Regression: behind an internal reverse proxy with use_x_forwarded_for
-        # enabled but mcp_trusted_proxy_ranges unset, the direct peer is the
-        # proxy's private IP. Returning it would mis-classify an external caller
-        # as internal and expose available_on_public_internet=false servers.
+    def test_external_client_via_xff_stays_external_without_trusted_proxy_ranges(self):
+        # External caller forwarded through an internal reverse proxy: the forwarded
+        # IP is the real public client, so it is still classified external and cannot
+        # reach available_on_public_internet=false servers.
         request = MagicMock(spec=Request)
         request.client = MagicMock()
         request.client.host = "10.0.0.7"
@@ -92,9 +92,64 @@ def test_private_proxy_peer_does_not_grant_internal_access(self):
             general_settings={"use_x_forwarded_for": True},
         )
 
-        assert result == ""
+        assert result == "8.8.8.8"
         assert IPAddressUtils.is_internal_ip(result) is False
 
+    def test_warns_once_that_xff_is_trusted_for_access_control_without_ranges(self):
+        # Security observability: when XFF is honored for access control without
+        # mcp_trusted_proxy_ranges, operators must be told (once) that the gate
+        # rests on an unvalidated header that a direct caller could spoof.
+        request = MagicMock(spec=Request)
+        request.client = MagicMock()
+        request.client.host = "203.0.113.5"
+        request.headers = {"x-forwarded-for": "10.0.0.1"}
+
+        with (
+            patch(
+                "litellm.proxy.auth.ip_address_utils._warned_xff_access_control_without_trusted_ranges",
+                False,
+            ),
+            patch(
+                "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger"
+            ) as mock_logger,
+        ):
+            IPAddressUtils.get_mcp_client_ip(
+                request, general_settings={"use_x_forwarded_for": True}
+            )
+            IPAddressUtils.get_mcp_client_ip(
+                request, general_settings={"use_x_forwarded_for": True}
+            )
+
+        assert mock_logger.warning.call_count == 1
+        message = mock_logger.warning.call_args[0][0]
+        assert "access control" in message.lower()
+        assert "mcp_trusted_proxy_ranges" in message
+
+    def test_no_access_control_warning_when_trusted_ranges_configured(self):
+        request = MagicMock(spec=Request)
+        request.client = MagicMock()
+        request.client.host = "10.0.0.5"
+        request.headers = {"x-forwarded-for": "192.168.1.10"}
+
+        with (
+            patch(
+                "litellm.proxy.auth.ip_address_utils._warned_xff_access_control_without_trusted_ranges",
+                False,
+            ),
+            patch(
+                "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger"
+            ) as mock_logger,
+        ):
+            IPAddressUtils.get_mcp_client_ip(
+                request,
+                general_settings={
+                    "use_x_forwarded_for": True,
+                    "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                },
+            )
+
+        mock_logger.warning.assert_not_called()
+
     def test_honours_xff_from_trusted_proxy(self):
         request = MagicMock(spec=Request)
         request.client = MagicMock()
@@ -128,6 +183,54 @@ def test_ignores_xff_from_untrusted_direct_caller(self):
         assert result == "203.0.113.5"
 
 
+class TestInternalOnlyServerResolutionRegression:
+    """LIT-3964: with use_x_forwarded_for enabled and no mcp_trusted_proxy_ranges,
+    internal-only servers must still resolve for forwarded internal callers
+    (previously failed closed to "" and 404'd every internal-only server)."""
+
+    @patch("litellm.public_mcp_servers", [])
+    @patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"use_x_forwarded_for": True},
+    )
+    def test_internal_only_server_found_for_forwarded_internal_client(self):
+        request = MagicMock(spec=Request)
+        request.client = MagicMock()
+        request.client.host = "10.0.0.7"  # load balancer peer
+        request.headers = {"x-forwarded-for": "192.168.1.50"}  # real internal client
+
+        client_ip = IPAddressUtils.get_mcp_client_ip(request)
+        assert client_ip == "192.168.1.50"
+
+        manager = _make_manager(
+            [_make_server("internal_math", available_on_public_internet=False)]
+        )
+        server = manager.get_mcp_server_by_name("internal_math", client_ip=client_ip)
+        assert server is not None
+        assert server.server_id == "internal_math"
+
+    @patch("litellm.public_mcp_servers", [])
+    @patch(
+        "litellm.proxy.proxy_server.general_settings",
+        {"use_x_forwarded_for": True},
+    )
+    def test_internal_only_server_hidden_from_forwarded_external_client(self):
+        request = MagicMock(spec=Request)
+        request.client = MagicMock()
+        request.client.host = "10.0.0.7"
+        request.headers = {"x-forwarded-for": "8.8.8.8"}  # real external client
+
+        client_ip = IPAddressUtils.get_mcp_client_ip(request)
+        assert client_ip == "8.8.8.8"
+
+        manager = _make_manager(
+            [_make_server("internal_math", available_on_public_internet=False)]
+        )
+        assert (
+            manager.get_mcp_server_by_name("internal_math", client_ip=client_ip) is None
+        )
+
+
 class TestMCPServerIPFiltering:
     """Tests that external callers only see public MCP servers."""