tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-08-09 12:24:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(mcp): SSRF guard on OAuth metadata discovery follow-up fetches

Browse Source 
The OAuth discovery code in mcp_server_manager followed two
attacker-influenceable URLs without validation: the
``resource_metadata`` URL parsed out of a ``WWW-Authenticate``
challenge, and the ``authorization_servers[0]`` field of the
PRM JSON returned by the resource server.  A malicious MCP server
could point those at a cloud-instance-metadata service, an internal
admin panel, or a loopback debug endpoint and the proxy would issue
a blind GET on its behalf.

Add ``_is_safe_metadata_url(url, server_url)`` and gate both follow-
up fetch sites on it.  A URL is allowed when:

  - it shares scheme + host + port with ``server_url`` (well-known
    endpoints constructed from the admin's URL, and PRM published at
    the resource server itself per RFC 9728 §3.3), or
  - it resolves to publicly-routable IPs only (covers federated
    authorization servers — Azure Entra, Google, Okta, GitHub —
    hosted cross-origin from the resource server).

URLs that resolve to private / loopback / link-local / cloud-metadata
addresses, or that don't resolve at all, are rejected.  ``http`` and
``https`` are the only schemes accepted.  The IP block list is
provided by the existing ``_is_blocked_ip`` helper from
``litellm_core_utils.url_utils`` so the policy stays consistent with
the rest of the proxy.

The guard does not protect against active DNS rebinding between
this resolution and the subsequent httpx GET — the same-authority
pin remains the primary mitigation; the IP check is defence in
depth.  The surface only triggers on config load / add-server, not
per request, so the synchronous ``getaddrinfo`` is acceptable.

Threads ``server_url`` through ``_fetch_oauth_metadata_from_resource``,
``_fetch_authorization_server_metadata``, and
``_fetch_single_authorization_server_metadata``.  Existing tests for
those helpers updated for the new signature; new
``TestOAuthDiscoverySSRFGuard`` covers same-authority allow,
private-IP rejection across IPv4 and IPv6, multi-A-record dual-
stack rejection, unresolvable hosts, non-http schemes, and
end-to-end "no network call when guard denies".
This commit is contained in:
user
2026-04-30 02:43:30 +00:00
parent 4a7af1ff68
commit df62dd8768
2 changed files with 258 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
+89 -8
View File
@@ -12,6 +12,7 @@ import hashlib
import json
import os
import re
import socket
from typing import Any, Callable, Dict, List, Literal, Optional, Set, Tuple, Union, cast
from urllib.parse import urlparse
@@ -41,6 +42,7 @@ from litellm.constants import (
MCP_TOOL_LISTING_TIMEOUT,
)
from litellm.exceptions import BlockedPiiEntityError, GuardrailRaisedException
from litellm.litellm_core_utils.url_utils import _is_blocked_ip
from litellm.experimental_mcp_client.client import MCPClient, MCPSigV4Auth
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import (
@@ -1499,6 +1501,66 @@ class MCPServerManager:
)
return await client.get_prompt(get_prompt_request_params)
@staticmethod
def _is_safe_metadata_url(url: str, server_url: str) -> bool:
"""
Whether ``url`` is safe to fetch during OAuth discovery for ``server_url``.
OAuth metadata discovery follows attacker-influenceable URLs from a
WWW-Authenticate header and from the protected-resource-metadata JSON.
Without a guard those become an SSRF primitive: a malicious MCP server
can point the proxy at cloud metadata services, internal admin panels,
or loopback debug endpoints.
A URL is allowed when:
- it shares (scheme, host, port) with ``server_url`` — well-known
endpoints constructed from the admin's URL, and PRM published at
the resource server itself per RFC 9728 §3.3, OR
- it resolves to one or more public IPs only — covers federated
authorization servers (Azure Entra, Google, Okta, GitHub) hosted
cross-origin from the resource.
URLs that resolve to private / loopback / link-local / cloud-metadata
addresses, or that don't resolve at all, are rejected. ``http`` and
``https`` are the only schemes accepted.
"""
try:
target = urlparse(url)
base = urlparse(server_url)
except Exception:
return False
if target.scheme not in ("http", "https") or not target.hostname:
return False
target_port = target.port or (443 if target.scheme == "https" else 80)
base_port = base.port or (443 if base.scheme == "https" else 80)
same_authority = (
base.scheme == target.scheme
and (base.hostname or "").lower() == target.hostname.lower()
and base_port == target_port
)
if same_authority:
return True
try:
infos = socket.getaddrinfo(
target.hostname, target_port, type=socket.SOCK_STREAM
)
except socket.gaierror:
return False
# Reuse the proxy-wide outbound block list (private / loopback /
# link-local / multicast / reserved / cloud-fabric IPs). Defence
# in depth only — the resolution here and the one httpx performs at
# request time leave a small DNS-rebinding window; an attacker who
# also controls the resource server is already in scope, so the
# primary mitigation is the same-authority pin above.
for info in infos:
if _is_blocked_ip(info[4][0]):
return False
return True
async def _descovery_metadata(
self,
server_url: str,
@@ -1514,7 +1576,7 @@ class MCPServerManager:
resource_scopes,
) = await self._attempt_well_known_discovery(server_url)
metadata = await self._fetch_authorization_server_metadata(
authorization_servers
authorization_servers, server_url
)
if (
metadata is None
@@ -1555,7 +1617,7 @@ class MCPServerManager:
authorization_servers,
resource_scopes,
) = await self._fetch_oauth_metadata_from_resource(
resource_metadata_url
resource_metadata_url, server_url
)
else:
(
@@ -1576,7 +1638,7 @@ class MCPServerManager:
if authorization_servers:
metadata = await self._fetch_authorization_server_metadata(
authorization_servers
authorization_servers, server_url
)
preferred_scopes = scopes or resource_scopes
@@ -1616,11 +1678,20 @@ class MCPServerManager:
return resource_metadata_url, scopes
async def _fetch_oauth_metadata_from_resource(
self, resource_metadata_url: str
self, resource_metadata_url: str, server_url: str
) -> Tuple[List[str], Optional[List[str]]]:
if not resource_metadata_url:
return [], None
if not self._is_safe_metadata_url(resource_metadata_url, server_url):
verbose_logger.warning(
"MCP OAuth discovery: refusing to fetch resource metadata from %s "
"(rejected by SSRF guard for server %s)",
resource_metadata_url,
server_url,
)
return [], None
try:
client = get_async_httpx_client(
llm_provider=httpxSpecialProvider.MCP,
@@ -1677,23 +1748,25 @@ class MCPServerManager:
(
authorization_servers,
scopes,
) = await self._fetch_oauth_metadata_from_resource(url)
) = await self._fetch_oauth_metadata_from_resource(url, server_url)
if authorization_servers:
return authorization_servers, scopes
return [], None
async def _fetch_authorization_server_metadata(
self, authorization_servers: List[str]
self, authorization_servers: List[str], server_url: str
) -> Optional[MCPOAuthMetadata]:
for issuer in authorization_servers:
metadata = await self._fetch_single_authorization_server_metadata(issuer)
metadata = await self._fetch_single_authorization_server_metadata(
issuer, server_url
)
if metadata is not None:
return metadata
return None
async def _fetch_single_authorization_server_metadata(
self, issuer_url: str
self, issuer_url: str, server_url: str
) -> Optional[MCPOAuthMetadata]:
try:
parsed = urlparse(issuer_url)
@@ -1720,6 +1793,14 @@ class MCPServerManager:
candidate_urls.append(issuer_url.rstrip("/"))
for url in candidate_urls:
if not self._is_safe_metadata_url(url, server_url):
verbose_logger.warning(
"MCP OAuth discovery: refusing to fetch authorization-server "
"metadata from %s (rejected by SSRF guard for server %s)",
url,
server_url,
)
continue
try:
client = get_async_httpx_client(
llm_provider=httpxSpecialProvider.MCP,
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
+169 -5
View File
@@ -719,7 +719,8 @@ class TestMCPServerManager:
return_value=mock_client,
):
servers, scopes = await manager._fetch_oauth_metadata_from_resource(
"https://protected.example.com/.well-known/oauth"
"https://protected.example.com/.well-known/oauth",
"https://protected.example.com/mcp",
)
assert servers == [
@@ -772,7 +773,8 @@ class TestMCPServerManager:
mock_well_known.assert_awaited_once_with("http://localhost:8001/mcp")
mock_fetch_auth.assert_awaited_once_with(
["https://login.microsoftonline.com/test-tenant-id/v2.0"]
["https://login.microsoftonline.com/test-tenant-id/v2.0"],
"http://localhost:8001/mcp",
)
assert result is mock_metadata
assert result.scopes == ["api://some-scope/.default"]
@@ -810,7 +812,12 @@ class TestMCPServerManager:
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
result = await manager._fetch_single_authorization_server_metadata(issuer)
# The Azure issuer is cross-origin against the server_url — use
# the issuer itself as server_url so the test exercises the
# well-known fetch logic without needing real DNS.
result = await manager._fetch_single_authorization_server_metadata(
issuer, issuer
)
assert result is not None
assert (
@@ -846,7 +853,9 @@ class TestMCPServerManager:
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
result = await manager._fetch_single_authorization_server_metadata(issuer)
result = await manager._fetch_single_authorization_server_metadata(
issuer, issuer
)
assert result is not None
assert (
@@ -910,7 +919,7 @@ class TestMCPServerManager:
):
result = await manager._descovery_metadata(server_url)
mock_fetch_auth.assert_awaited_once_with(["https://example.com"])
mock_fetch_auth.assert_awaited_once_with(["https://example.com"], server_url)
assert result is mock_metadata
assert result.scopes == ["read"]
@@ -2947,5 +2956,160 @@ class TestMCPServerManagerExpandToolPermissions:
assert sorted(result["uuid-a"]) == ["read_file", "write_file"]
class TestOAuthDiscoverySSRFGuard:
"""SSRF guard for the OAuth metadata discovery follow-up fetches.
The vulnerability: a malicious MCP server returns a ``WWW-Authenticate``
header pointing at an attacker-chosen ``resource_metadata`` URL, then a
PRM JSON whose ``authorization_servers[0]`` points at internal/loopback
addresses, coercing the proxy into making blind GETs to cloud-metadata
services, internal admin panels, or loopback debug endpoints.
"""
@staticmethod
def _patch_resolves(monkeypatch, mapping):
"""Patch ``socket.getaddrinfo`` for a deterministic SSRF-guard test.
``mapping`` is ``{hostname: [ip-string, ...]}``; unknown hosts raise
``gaierror`` (treated as "unresolvable" -> blocked).
"""
import socket as _socket
def fake_getaddrinfo(host, port, *args, **kwargs):
if host not in mapping:
raise _socket.gaierror(f"unknown host {host}")
family = _socket.AF_INET
return [
(family, _socket.SOCK_STREAM, 0, "", (ip, port)) for ip in mapping[host]
]
monkeypatch.setattr(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.socket.getaddrinfo",
fake_getaddrinfo,
)
def test_same_authority_url_is_safe(self):
# Same scheme + host + port skips DNS entirely — the well-known
# endpoint construction in _attempt_well_known_discovery always
# produces same-authority URLs against the admin's server_url.
assert MCPServerManager._is_safe_metadata_url(
"https://example.com/.well-known/oauth-protected-resource",
"https://example.com/mcp",
)
def test_same_host_different_port_blocked_when_resolves_to_private_ip(
self, monkeypatch
):
# Cross-authority (different port). Falls through to DNS check.
# If the resolved IP is private, the guard rejects.
self._patch_resolves(monkeypatch, {"example.com": ["10.1.2.3"]})
assert not MCPServerManager._is_safe_metadata_url(
"https://example.com:9999/.well-known/oauth-protected-resource",
"https://example.com/mcp",
)
@pytest.mark.parametrize(
"ip",
[
"127.0.0.1", # loopback
"10.0.0.5", # RFC1918
"172.16.0.1", # RFC1918
"192.168.1.1", # RFC1918
"169.254.169.254", # AWS / Azure / GCP IMDS
"100.100.100.200", # Alibaba Cloud metadata
"0.0.0.0", # unspecified
"::1", # IPv6 loopback
"fe80::1", # IPv6 link-local
"fc00::1", # IPv6 ULA
],
)
def test_cross_origin_blocked_when_resolves_to_unsafe_ip(self, monkeypatch, ip):
self._patch_resolves(monkeypatch, {"attacker.example.com": [ip]})
assert not MCPServerManager._is_safe_metadata_url(
f"https://attacker.example.com/.well-known/oauth-authorization-server",
"https://legit-mcp.example.com/mcp",
)
def test_cross_origin_allowed_when_resolves_to_public_ip(self, monkeypatch):
self._patch_resolves(
monkeypatch, {"login.microsoftonline.com": ["20.190.151.7"]}
)
assert MCPServerManager._is_safe_metadata_url(
"https://login.microsoftonline.com/tenant/v2.0/.well-known/openid-configuration",
"https://atlassian-mcp.example.com/mcp",
)
def test_cross_origin_blocked_when_unresolvable(self, monkeypatch):
self._patch_resolves(monkeypatch, {})
assert not MCPServerManager._is_safe_metadata_url(
"https://nope.example.invalid/.well-known/oauth-authorization-server",
"https://legit-mcp.example.com/mcp",
)
def test_non_http_scheme_is_not_safe(self):
assert not MCPServerManager._is_safe_metadata_url(
"file:///etc/passwd", "https://example.com/mcp"
)
assert not MCPServerManager._is_safe_metadata_url(
"gopher://example.com/", "https://example.com/mcp"
)
def test_dual_resolution_blocked_if_any_ip_unsafe(self, monkeypatch):
# If the attacker controls a DNS record returning multiple A records,
# one of which is private, the guard must reject — pinning to the
# safe IP would be a TOCTOU window.
self._patch_resolves(
monkeypatch, {"dual-stack.example.com": ["8.8.8.8", "127.0.0.1"]}
)
assert not MCPServerManager._is_safe_metadata_url(
"https://dual-stack.example.com/.well-known/oauth-authorization-server",
"https://legit-mcp.example.com/mcp",
)
@pytest.mark.asyncio
async def test_fetch_oauth_metadata_refuses_unsafe_url(self, monkeypatch):
# End-to-end: a malicious WWW-Authenticate redirecting to a loopback
# resource_metadata URL must not produce a network call.
self._patch_resolves(monkeypatch, {"attacker.example.com": ["127.0.0.1"]})
manager = MCPServerManager()
mock_client = MagicMock()
mock_client.get = AsyncMock()
with patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
servers, scopes = await manager._fetch_oauth_metadata_from_resource(
"https://attacker.example.com/meta",
"https://legit-mcp.example.com/mcp",
)
assert servers == []
assert scopes is None
mock_client.get.assert_not_called()
@pytest.mark.asyncio
async def test_fetch_authorization_server_refuses_unsafe_issuer(self, monkeypatch):
# Mirrors the GHSA-mrfv repro: PRM lists a loopback issuer URL.
self._patch_resolves(monkeypatch, {"attacker.example.com": ["127.0.0.1"]})
manager = MCPServerManager()
mock_client = MagicMock()
mock_client.get = AsyncMock()
with patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
result = await manager._fetch_single_authorization_server_metadata(
"http://attacker.example.com:19999",
"https://legit-mcp.example.com/mcp",
)
assert result is None
mock_client.get.assert_not_called()
if __name__ == "__main__":
pytest.main([__file__])