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

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

When a request carries an X-Forwarded-For header but use_x_forwarded_for is
unset, get_mcp_client_ip silently falls back to the direct peer's IP (the load
balancer / reverse proxy). That peer almost always sits inside
mcp_internal_ip_ranges, so the 'Internal network only'
(available_on_public_internet: false) restriction trusts every external caller
as internal and effectively exposes those servers.

Emit a one-shot loud error pointing the operator at use_x_forwarded_for instead
of hard-failing: on a deployment with no load balancer, a crafted
X-Forwarded-For header must not be able to take the service down, and a one-shot
log keeps a flood of crafted headers from spamming the logs.

* fix(mcp): re-arm XFF-disabled warning on config change and harden test assertion

Address PR review: tie the one-shot warning flag to the observed
use_x_forwarded_for value so it re-arms whenever the setting is seen enabled,
restoring the diagnostic on a later rollback to disabled. Also assert against
str(call_args) so the test survives a positional-to-keyword logger refactor.
This commit is contained in:
Mateo Wang 2026-06-24 20:49:32 -07:00 • committed by GitHub
parent 257d67167f
commit 9c41077786
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 140 additions and 0 deletions

View file

@ -17,6 +17,14 @@ from litellm.proxy.auth.auth_utils import _get_request_ip_address
# 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 @@ class IPAddressUtils:
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(

View file

@ -169,6 +169,116 @@ class TestMCPClientIPExtraction:
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."""