scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-06 02:48:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

Merge pull request #26841 from stuxf/fix/mcp-xff-trust-gate

Browse source 
chore(mcp): require trusted-proxy gate before honouring X-Forwarded-* on OAuth discovery
This commit is contained in:
yuneng-jiang 2026-05-01 15:08:40 -07:00 • committed by GitHub
commit 8363fe00e1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 226 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
View file

@ -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, "", "", ""))

59
litellm/proxy/auth/ip_address_utils.py
View file

@ -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,

165
tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py
View file

@ -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
# -------------------------------------------------------------------