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

chore(mcp): gate X-Forwarded-* trust on get_request_base_url

Browse source 
get_request_base_url unconditionally honoured X-Forwarded-Proto / Host /
Port to build OAuth issuer / redirect_uri / authorization_endpoint
values for the MCP discovery endpoints.  In a deployment where the
proxy is reachable from a caller that can send those headers (direct
internet exposure, or a reverse proxy that does not strip them), an
attacker could poison the OAuth metadata and steer MCP clients at an
attacker-controlled host.

Apply the same trusted-proxy gate the codebase already uses for
get_mcp_client_ip: only honour the headers when use_x_forwarded_for is
enabled in proxy settings AND the direct connection IP falls inside
mcp_trusted_proxy_ranges.  When that's not configured, fall back to
the request's literal base_url, so an untrusted caller cannot poison
the discovery metadata.

The existing X-Forwarded-* parsing test cases now opt into a
trust_xff fixture (the parsing logic itself is unchanged).  Adds a
matrix for the new gate covering: XFF disabled, XFF enabled with no
ranges, caller outside ranges, caller inside ranges, and the
loopback-dev-deployment case.
This commit is contained in:
user 2026-04-30 01:30:41 +00:00
parent 4a7af1ff68
commit 463781d8cb
3 changed files with 166 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, "", "", ""))

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

@ -106,6 +106,50 @@ 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:
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,

120
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)"""
@ -506,6 +521,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:
@ -573,6 +589,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:
@ -652,6 +669,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:
@ -706,6 +724,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:
@ -761,6 +780,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:
@ -798,6 +818,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:
@ -871,6 +892,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:
@ -1074,7 +1096,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
@ -1084,11 +1111,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
@ -1097,16 +1122,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}, "
@ -1116,6 +1142,86 @@ 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"
# -------------------------------------------------------------------
# Tests for scopes_supported when mcp_server.scopes is None
# -------------------------------------------------------------------