fix(mcp): warn loudly when X-Forwarded-For is present but use_x_forwarded_for is off

litellm/proxy/auth/ip_address_utils.py

@@ -17,6 +17,14 @@
 # behaviour see an actionable message in their logs the first time it triggers.
 _warned_xff_without_trusted_ranges = False
 
+# Error for the inverse footgun: requests arrive with an X-Forwarded-For header
+# but use_x_forwarded_for is off, so the real client IP is silently dropped and
+# "internal network only" access control trusts the load balancer's IP instead.
+# Logged once per misconfiguration window (not per-request) so a flood of crafted
+# XFF headers can't spam the logs; re-arms whenever use_x_forwarded_for is observed
+# enabled, so a later rollback to disabled warns again.
+_warned_xff_present_but_disabled = False
+
 
 class IPAddressUtils:
     """Static utilities for IP-based MCP access control."""
@@ -198,6 +206,28 @@ def get_mcp_client_ip(
 
         use_xff = general_settings.get("use_x_forwarded_for", False)
 
+        global _warned_xff_present_but_disabled
+        if use_xff:
+            _warned_xff_present_but_disabled = False
+        elif "x-forwarded-for" in request.headers:
+            if not _warned_xff_present_but_disabled:
+                verbose_proxy_logger.error(
+                    "Received a request with an X-Forwarded-For header but "
+                    "use_x_forwarded_for is not enabled. The real client IP is "
+                    "being ignored and the direct peer's IP (typically your load "
+                    "balancer / reverse proxy) is used for MCP access control. "
+                    "Because that peer almost always falls within "
+                    "general_settings.mcp_internal_ip_ranges, every external caller "
+                    "is treated as internal and 'available_on_public_internet: "
+                    "false' MCP servers are effectively exposed. Set "
+                    "use_x_forwarded_for: true (and mcp_trusted_proxy_ranges to "
+                    "your proxy CIDRs) in general_settings to honor the real "
+                    "client IP. Not failing the request: if there is no load "
+                    "balancer, a crafted X-Forwarded-For header must not be able "
+                    "to take down the service."
+                )
+                _warned_xff_present_but_disabled = True
+
         # 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(

tests/test_litellm/proxy/auth/test_mcp_ip_filtering.py

@@ -128,6 +128,116 @@ def test_ignores_xff_from_untrusted_direct_caller(self):
         assert result == "203.0.113.5"
 
 
+class TestXffPresentButDisabledWarning:
+    """When an XFF header arrives but use_x_forwarded_for is off, the proxy must
+    loudly warn (the internal-only check is silently trusting the load balancer's
+    IP) yet still serve the request, so a crafted header can't DoS a no-LB deploy."""
+
+    def _reset_warning_flag(self):
+        from litellm.proxy.auth import ip_address_utils
+
+        ip_address_utils._warned_xff_present_but_disabled = False
+
+    def _request_with_xff(self):
+        request = MagicMock(spec=Request)
+        request.client = MagicMock()
+        request.client.host = "10.0.0.7"
+        request.headers = {"x-forwarded-for": "8.8.8.8"}
+        return request
+
+    def test_warns_and_does_not_fail_when_xff_present_but_disabled(self):
+        self._reset_warning_flag()
+        request = self._request_with_xff()
+
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger.error"
+        ) as mock_error:
+            result = IPAddressUtils.get_mcp_client_ip(
+                request, general_settings={"use_x_forwarded_for": False}
+            )
+
+        # Does not hard-fail: falls back to the direct peer (the load balancer).
+        assert result == "10.0.0.7"
+        mock_error.assert_called_once()
+        assert "use_x_forwarded_for" in str(mock_error.call_args)
+
+    def test_warning_is_one_shot(self):
+        self._reset_warning_flag()
+
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger.error"
+        ) as mock_error:
+            IPAddressUtils.get_mcp_client_ip(
+                self._request_with_xff(),
+                general_settings={"use_x_forwarded_for": False},
+            )
+            IPAddressUtils.get_mcp_client_ip(
+                self._request_with_xff(),
+                general_settings={"use_x_forwarded_for": False},
+            )
+
+        # One-shot so a flood of crafted XFF headers cannot spam the logs.
+        mock_error.assert_called_once()
+
+    def test_re_arms_after_xff_is_enabled_then_disabled_again(self):
+        self._reset_warning_flag()
+
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger.error"
+        ) as mock_error:
+            IPAddressUtils.get_mcp_client_ip(
+                self._request_with_xff(),
+                general_settings={"use_x_forwarded_for": False},
+            )
+            # Operator fixes the config; observing it enabled re-arms the warning.
+            IPAddressUtils.get_mcp_client_ip(
+                self._request_with_xff(),
+                general_settings={
+                    "use_x_forwarded_for": True,
+                    "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                },
+            )
+            # Config rolls back to disabled: the misconfiguration must warn again.
+            IPAddressUtils.get_mcp_client_ip(
+                self._request_with_xff(),
+                general_settings={"use_x_forwarded_for": False},
+            )
+
+        assert mock_error.call_count == 2
+
+    def test_no_warning_without_xff_header(self):
+        self._reset_warning_flag()
+        request = MagicMock(spec=Request)
+        request.client = MagicMock()
+        request.client.host = "10.0.0.7"
+        request.headers = {}
+
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger.error"
+        ) as mock_error:
+            IPAddressUtils.get_mcp_client_ip(
+                request, general_settings={"use_x_forwarded_for": False}
+            )
+
+        mock_error.assert_not_called()
+
+    def test_no_warning_when_xff_enabled(self):
+        self._reset_warning_flag()
+
+        with patch(
+            "litellm.proxy.auth.ip_address_utils.verbose_proxy_logger.error"
+        ) as mock_error:
+            IPAddressUtils.get_mcp_client_ip(
+                self._request_with_xff(),
+                general_settings={
+                    "use_x_forwarded_for": True,
+                    "mcp_trusted_proxy_ranges": ["10.0.0.0/8"],
+                },
+            )
+
+        mock_error.assert_not_called()
+
+
 class TestMCPServerIPFiltering:
     """Tests that external callers only see public MCP servers."""