scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

feat(mcp): issuer-anchored OAuth discovery (RFC 8414 §3.3) as the trust anchor

Browse source 
Adds an admin-configured issuer to MCP servers. When set, OAuth metadata is
fetched from the issuer's own origin and adopted only when the document
self-attests that same issuer (RFC 8414 §3.3), making token_endpoint,
registration_endpoint, and scopes authoritative for the pinned issuer instead
of a document the MCP resource server chose. This closes the mix-up where a
compromised resource echoes a pinned authorization_url to smuggle its own
token endpoint and inflated scopes past the corroboration gate. Discovery is
same-authority against the issuer origin, fails closed on a §3.3 mismatch, and
does not fall back to resource-rooted discovery. Rows without an issuer keep
the existing corroboration-gate behavior unchanged.

Backend + schema only; UI field and live-proxy proof follow.
This commit is contained in:
Tin Chi Lo 2026-07-15 13:37:10 -07:00
parent 923c325e64
commit 8e73ff057f
11 changed files with 260 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm-proxy-extras/litellm_proxy_extras/migrations/20260715000000_add_issuer_to_mcp_server_table/migration.sql Normal file
View file

@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "LiteLLM_MCPServerTable" ADD COLUMN IF NOT EXISTS "issuer" TEXT;

1
litellm-proxy-extras/litellm_proxy_extras/schema.prisma
View file

@ -325,6 +325,7 @@ model LiteLLM_MCPServerTable {
command String?
args String[] @default([])
env Json? @default("{}")
issuer String?
authorization_url String?
token_url String?
registration_url String?

1
litellm/models/mcp_server.py
View file

@ -79,6 +79,7 @@ class LiteLLM_MCPServerTable(LiteLLMPydanticObjectBase):
command: Optional[str] = None
args: List[str] = Field(default_factory=list)
env: Dict[str, str] = Field(default_factory=dict)
issuer: Optional[str] = None
authorization_url: Optional[str] = None
token_url: Optional[str] = None
registration_url: Optional[str] = None

2
litellm/proxy/_experimental/mcp_server/db.py
View file

@ -48,6 +48,7 @@ if TYPE_CHECKING:
_AUTH_FLOW_SCOPED_FIELDS: frozenset = frozenset(
{
"issuer",
"authorization_url",
"token_url",
"registration_url",
@ -1181,6 +1182,7 @@ def mcp_oauth_token_identity(server: object) -> tuple[object, ...]:
getattr(server, "spec_path", None),
getattr(server, "auth_type", None),
getattr(server, "oauth2_flow", None),
getattr(server, "issuer", None),
getattr(server, "authorization_url", None),
getattr(server, "token_url", None),
getattr(server, "registration_url", None),

113
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -217,6 +217,17 @@ def _normalized_authorize_endpoint(url: str) -> str:
return f"{scheme}://{authority}{parsed.path.rstrip('/')}"
def _issuer_matches(claimed_issuer: object, configured_issuer: str) -> bool:
"""RFC 8414 §3.3 issuer equality between the metadata document's self-attested ``issuer`` and the
admin-configured issuer, tolerant only of URL-insignificant differences (scheme/host case, the
default port, a trailing slash). A non-string or empty claimed issuer never matches, so a
document that omits ``issuer`` fails closed under issuer-anchored discovery.
"""
if not isinstance(claimed_issuer, str) or not claimed_issuer:
return False
return _normalized_authorize_endpoint(claimed_issuer) == _normalized_authorize_endpoint(configured_issuer)
def _endpoints_corroborate_authorization_url(
source_authorization_url: str | None,
trusted_authorization_url: str | None,
@ -1137,34 +1148,41 @@ class MCPServerManager:
)
auth_type = server_config.get("auth_type", None)
manual_issuer = _blank_to_none(server_config.get("issuer"))
manual_authorization_url = _blank_to_none(server_config.get("authorization_url"))
manual_token_url = _blank_to_none(server_config.get("token_url"))
manual_registration_url = _blank_to_none(server_config.get("registration_url"))
if server_url and (
auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
is_discovery_auth_type = auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
use_issuer_anchor = manual_issuer is not None and is_discovery_auth_type
should_discover = bool(server_url) and (
is_discovery_auth_type
or self._obo_needs_endpoint_discovery(
auth_type,
server_config.get("token_exchange_endpoint"),
manual_token_url,
)
):
)
if not should_discover:
mcp_oauth_metadata = None
elif manual_issuer is not None and is_discovery_auth_type:
mcp_oauth_metadata = await self._fetch_issuer_anchored_oauth_metadata(manual_issuer)
else:
mcp_oauth_metadata = await self._descovery_metadata(
server_url=server_url,
allow_origin_fallback=auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES,
allow_origin_fallback=is_discovery_auth_type,
)
else:
mcp_oauth_metadata = None
gated_oauth_metadata = (
_restrict_discovery_to_corroborated_authorization_server(
if use_issuer_anchor:
gated_oauth_metadata = mcp_oauth_metadata
elif is_discovery_auth_type:
gated_oauth_metadata = _restrict_discovery_to_corroborated_authorization_server(
mcp_oauth_metadata,
manual_authorization_url,
server_name or server_id,
bool(server_config.get("dcr_bridge")),
)
if auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
else mcp_oauth_metadata
)
else:
gated_oauth_metadata = mcp_oauth_metadata
# Filter blank scopes (e.g. YAML ``scopes: [""]``) the same way the DB-build path does, so
# an all-blank list normalizes to None rather than a ``("",)`` tuple that skips the
@ -1227,6 +1245,7 @@ class MCPServerManager:
client_secret=server_config.get("client_secret", None),
oauth2_flow=self._explicit_oauth2_flow(config_oauth2_flow),
scopes=resolved_scopes,
issuer=manual_issuer,
authorization_url=resolved_authorization_url,
token_url=resolved_token_url,
registration_url=resolved_registration_url,
@ -1570,12 +1589,15 @@ class MCPServerManager:
auth_type = cast(MCPAuthType, mcp_server.auth_type)
server_url = mcp_server.url
manual_issuer = _blank_to_none(mcp_server.issuer)
manual_authorization_url = _blank_to_none(mcp_server.authorization_url)
manual_token_url = _blank_to_none(mcp_server.token_url)
manual_registration_url = _blank_to_none(mcp_server.registration_url)
is_discovery_auth_type = auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
use_issuer_anchor = manual_issuer is not None and is_discovery_auth_type
has_all_upstream_oauth_fields = bool(manual_authorization_url and manual_token_url and scopes)
needs_discovery = bool(server_url) and (
(auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES and not has_all_upstream_oauth_fields)
(is_discovery_auth_type and not has_all_upstream_oauth_fields)
or self._obo_needs_endpoint_discovery(
auth_type,
mcp_server.token_exchange_endpoint
@ -1583,31 +1605,33 @@ class MCPServerManager:
manual_token_url,
)
)
mcp_oauth_metadata = (
await self._descovery_metadata(
if not needs_discovery:
mcp_oauth_metadata = None
elif manual_issuer is not None and is_discovery_auth_type:
mcp_oauth_metadata = await self._fetch_issuer_anchored_oauth_metadata(manual_issuer)
else:
mcp_oauth_metadata = await self._descovery_metadata(
server_url=server_url, # type: ignore[arg-type]
allow_origin_fallback=auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES,
allow_origin_fallback=is_discovery_auth_type,
)
if needs_discovery
else None
)
if needs_discovery and mcp_oauth_metadata is None:
if needs_discovery and not use_issuer_anchor and mcp_oauth_metadata is None:
verbose_logger.warning(
"MCP OAuth discovery yielded no metadata for server %s (%s); "
"OAuth endpoints/scopes stay unresolved until a rebuild succeeds",
mcp_server.server_id,
server_url,
)
gated_oauth_metadata = (
_restrict_discovery_to_corroborated_authorization_server(
if use_issuer_anchor:
gated_oauth_metadata = mcp_oauth_metadata
elif is_discovery_auth_type:
gated_oauth_metadata = _restrict_discovery_to_corroborated_authorization_server(
mcp_oauth_metadata,
manual_authorization_url,
mcp_server.server_id,
bool(getattr(mcp_server, "dcr_bridge", None)),
)
if auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
else mcp_oauth_metadata
)
else:
gated_oauth_metadata = mcp_oauth_metadata
resolved_scopes = scopes or (gated_oauth_metadata.scopes if gated_oauth_metadata else None)
@ -1629,6 +1653,7 @@ class MCPServerManager:
client_secret=client_secret_value or getattr(mcp_server, "client_secret", None),
oauth2_flow=self._explicit_oauth2_flow(getattr(mcp_server, "oauth2_flow", None)),
scopes=resolved_scopes,
issuer=manual_issuer,
authorization_url=manual_authorization_url or getattr(gated_oauth_metadata, "authorization_url", None),
token_url=manual_token_url or getattr(gated_oauth_metadata, "token_url", None),
registration_url=manual_registration_url or getattr(gated_oauth_metadata, "registration_url", None),
@ -3337,8 +3362,28 @@ class MCPServerManager:
return metadata
return None
async def _fetch_issuer_anchored_oauth_metadata(self, issuer: str) -> Optional[MCPOAuthMetadata]:
"""RFC 8414 issuer-anchored discovery.
Fetch authorization-server metadata from the admin-configured issuer's own origin and adopt
it only when the document self-attests that same issuer (RFC 8414 §3.3). Because the trust
anchor is the pinned issuer rather than anything the MCP resource server advertises, the
resulting token_endpoint/registration_endpoint/scopes are authoritative for that issuer and
cannot be substituted by a compromised resource. Fails closed (returns None) on a §3.3
mismatch or a fetch failure. The issuer is passed as its own ``server_url`` so the fetch is
treated as same-authority and is not subject to the resource-scoped SSRF shortcut.
"""
metadata = await self._fetch_single_authorization_server_metadata(issuer, issuer, require_issuer=issuer)
if metadata is None:
verbose_logger.warning(
"MCP OAuth issuer-anchored discovery for issuer %s yielded no metadata whose issuer "
"matched (RFC 8414 §3.3); OAuth endpoints/scopes stay unresolved until a rebuild succeeds",
issuer,
)
return metadata
async def _fetch_single_authorization_server_metadata(
self, issuer_url: str, server_url: str
self, issuer_url: str, server_url: str, require_issuer: Optional[str] = None
) -> Optional[MCPOAuthMetadata]:
try:
parsed = urlparse(issuer_url)
@ -3382,15 +3427,27 @@ class MCPServerManager:
)
continue
scopes = self._extract_scopes(data.get("scopes_supported"))
claimed_issuer = data.get("issuer")
verbose_logger.debug(
"Authorization server metadata from %s: issuer=%s grant_types_supported=%s "
"token_endpoint_auth_methods_supported=%s",
url,
data.get("issuer"),
claimed_issuer,
data.get("grant_types_supported"),
data.get("token_endpoint_auth_methods_supported"),
)
if require_issuer is not None and not _issuer_matches(claimed_issuer, require_issuer):
verbose_logger.warning(
"MCP OAuth issuer-anchored discovery: metadata at %s self-attests issuer %r, which "
"does not match the configured issuer %r (RFC 8414 §3.3); rejecting so a compromised "
"resource cannot substitute an attacker authorization server",
url,
claimed_issuer,
require_issuer,
)
continue
scopes = self._extract_scopes(data.get("scopes_supported"))
metadata = MCPOAuthMetadata(
scopes=scopes,
authorization_url=data.get("authorization_endpoint"),
@ -3408,6 +3465,8 @@ class MCPServerManager:
):
return metadata
if require_issuer is not None:
return None
return self._build_azure_authorization_server_metadata(parsed)
@staticmethod

1
litellm/proxy/_experimental/mcp_server/rest_endpoints.py
View file

@ -1138,6 +1138,7 @@ if MCP_AVAILABLE:
static_headers=request.static_headers,
client_id=client_id,
client_secret=client_secret,
issuer=request.issuer,
token_url=request.token_url,
scopes=scopes,
authorization_url=request.authorization_url,

2
litellm/proxy/_types.py
View file

@ -1263,6 +1263,7 @@ class NewMCPServerRequest(LiteLLMPydanticObjectBase):
command: Optional[str] = None
args: List[str] = Field(default_factory=list)
env: Dict[str, str] = Field(default_factory=dict)
issuer: Optional[str] = None
authorization_url: Optional[str] = None
token_url: Optional[str] = None
registration_url: Optional[str] = None
@ -1368,6 +1369,7 @@ class UpdateMCPServerRequest(LiteLLMPydanticObjectBase):
command: Optional[str] = None
args: List[str] = Field(default_factory=list)
env: Dict[str, str] = Field(default_factory=dict)
issuer: Optional[str] = None
authorization_url: Optional[str] = None
token_url: Optional[str] = None
registration_url: Optional[str] = None

1
litellm/proxy/schema.prisma
View file

@ -325,6 +325,7 @@ model LiteLLM_MCPServerTable {
command String?
args String[] @default([])
env Json? @default("{}")
issuer String?
authorization_url String?
token_url String?
registration_url String?

1
litellm/types/mcp_server/mcp_server_manager.py
View file

@ -60,6 +60,7 @@ class MCPServer(BaseModel):
# OAuth-specific fields
client_id: Optional[str] = None
client_secret: Optional[str] = None
issuer: Optional[str] = None
scopes: Optional[List[str]] = None
authorization_url: Optional[str] = None
token_url: Optional[str] = None

1
schema.prisma
View file

@ -325,6 +325,7 @@ model LiteLLM_MCPServerTable {
command String?
args String[] @default([])
env Json? @default("{}")
issuer String?
authorization_url String?
token_url String?
registration_url String?

162
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
View file

@ -1230,6 +1230,80 @@ class TestMCPServerManager:
assert built.registration_url == "https://idp.example.com/register"
assert built.scopes == ["read"]
@pytest.mark.asyncio
async def test_build_from_table_issuer_anchor_ignores_resource_rooted_discovery(self):
"""When an admin configures an issuer, discovery is anchored on that issuer's own metadata
(RFC 8414), not on the resource-rooted RFC 9728 chain a compromised MCP resource controls.
The resource-rooted _descovery_metadata must not run at all, and the authoritative token_url,
registration_url, and scopes come from the issuer document. This closes the mix-up where a
compromised resource echoes the pinned authorize endpoint to smuggle its own token_url."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="issuer-anchored-1",
alias="issuer_anchored",
description="issuer configured, blank endpoints",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
issuer="https://idp.example.com",
created_at=datetime.now(),
updated_at=datetime.now(),
)
authoritative = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
registration_url="https://idp.example.com/register",
scopes=["read", "write"],
authorization_server_scopes=["read", "write"],
)
resource_rooted = AsyncMock(return_value=MCPOAuthMetadata(token_url="https://attacker.example.com/steal"))
with (
patch.object(manager, "_fetch_issuer_anchored_oauth_metadata", new=AsyncMock(return_value=authoritative)) as anchored,
patch.object(manager, "_descovery_metadata", new=resource_rooted),
):
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
anchored.assert_awaited_once_with("https://idp.example.com")
resource_rooted.assert_not_awaited()
assert built.issuer == "https://idp.example.com"
assert built.authorization_url == "https://idp.example.com/authorize"
assert built.token_url == "https://idp.example.com/token"
assert built.registration_url == "https://idp.example.com/register"
assert built.scopes == ["read", "write"]
@pytest.mark.asyncio
async def test_build_from_table_issuer_anchor_fails_closed_without_falling_back_to_resource(self):
"""A configured issuer whose metadata does not validate (RFC 8414 §3.3 mismatch or fetch
failure) yields None from the anchored fetch. The build must adopt nothing and must NOT fall
back to resource-rooted discovery, or the fail-closed guarantee would be defeated by the very
resource the issuer anchor exists to distrust."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="issuer-anchored-2",
alias="issuer_anchored_failclosed",
description="issuer configured, upstream fails validation",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
issuer="https://idp.example.com",
created_at=datetime.now(),
updated_at=datetime.now(),
)
resource_rooted = AsyncMock(return_value=MCPOAuthMetadata(token_url="https://attacker.example.com/steal"))
with (
patch.object(manager, "_fetch_issuer_anchored_oauth_metadata", new=AsyncMock(return_value=None)),
patch.object(manager, "_descovery_metadata", new=resource_rooted),
):
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
resource_rooted.assert_not_awaited()
assert built.issuer == "https://idp.example.com"
assert built.token_url is None
assert built.registration_url is None
assert built.scopes is None
@pytest.mark.asyncio
@pytest.mark.parametrize(
"advertised_authorization_url",
@ -2462,6 +2536,78 @@ class TestMCPServerManager:
assert result.token_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/token"
assert result.scopes == ["api://some-scope/.default"]
@staticmethod
def _issuer_doc_response_builder(well_known_url: str, document: dict):
def build_response(url: str, **kwargs):
mock_response = MagicMock()
if url == well_known_url:
mock_response.json.return_value = document
mock_response.raise_for_status = MagicMock()
else:
request = httpx.Request("GET", url)
response_obj = httpx.Response(status_code=404, request=request)
mock_response.raise_for_status = MagicMock(
side_effect=httpx.HTTPStatusError("not found", request=request, response=response_obj)
)
return mock_response
return build_response
@pytest.mark.asyncio
async def test_fetch_single_authorization_server_metadata_adopts_document_with_matching_issuer(self):
"""RFC 8414 §3.3: under require_issuer, a document that self-attests the same issuer it was
fetched from is authoritative and its endpoints and scopes are adopted."""
manager = MCPServerManager()
issuer = "https://idp.example.com"
build_response = self._issuer_doc_response_builder(
f"{issuer}/.well-known/oauth-authorization-server",
{
"issuer": issuer,
"authorization_endpoint": "https://idp.example.com/authorize",
"token_endpoint": "https://idp.example.com/token",
"scopes_supported": ["read", "write"],
},
)
mock_client = MagicMock()
mock_client.get = AsyncMock(side_effect=build_response)
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(issuer, issuer, require_issuer=issuer)
assert result is not None
assert result.authorization_url == "https://idp.example.com/authorize"
assert result.token_url == "https://idp.example.com/token"
assert result.scopes == ["read", "write"]
@pytest.mark.asyncio
async def test_fetch_single_authorization_server_metadata_rejects_issuer_mismatch(self):
"""RFC 8414 §3.3 fail-closed: a document self-attesting a DIFFERENT issuer than the one it was
fetched from is rejected even though it carries valid-looking endpoints, so a compromised
resource cannot point the issuer-anchored fetch at an attacker authorization server that
smuggles its own token_endpoint and inflated scopes."""
manager = MCPServerManager()
issuer = "https://idp.example.com"
build_response = self._issuer_doc_response_builder(
f"{issuer}/.well-known/oauth-authorization-server",
{
"issuer": "https://attacker.example.com",
"authorization_endpoint": "https://idp.example.com/authorize",
"token_endpoint": "https://attacker.example.com/steal",
"scopes_supported": ["admin"],
},
)
mock_client = MagicMock()
mock_client.get = AsyncMock(side_effect=build_response)
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(issuer, issuer, require_issuer=issuer)
assert result is None
@pytest.mark.asyncio
async def test_fetch_single_authorization_server_metadata_derives_azure_metadata(
self,
@ -5487,6 +5633,22 @@ class TestMCPServerTimestamps:
assert _normalized_authorize_endpoint("https://idp.example.com/authorize?prompt=consent") == canonical
assert _normalized_authorize_endpoint("https://idp.example.com:8443/authorize") != canonical
def test_issuer_matches_rfc8414_section_3_3(self):
"""Issuer equality tolerates only URL-insignificant differences (scheme/host case, default
port, a trailing slash). A different host, a non-string, an empty string, or a None issuer
never matches, so a document that omits issuer fails closed under issuer-anchored discovery."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import _issuer_matches
assert _issuer_matches("https://mcp.slack.com", "https://mcp.slack.com")
assert _issuer_matches("https://MCP.slack.com/", "https://mcp.slack.com")
assert _issuer_matches("https://mcp.slack.com:443", "https://mcp.slack.com")
assert _issuer_matches("https://login.example.com/tenant/v2.0", "https://login.example.com/tenant/v2.0")
assert not _issuer_matches("https://attacker.example.com", "https://mcp.slack.com")
assert not _issuer_matches("https://login.example.com/other/v2.0", "https://login.example.com/tenant/v2.0")
assert not _issuer_matches(None, "https://mcp.slack.com")
assert not _issuer_matches("", "https://mcp.slack.com")
assert not _issuer_matches(123, "https://mcp.slack.com")
def test_build_mcp_server_table_preserves_timestamps(self):
"""_build_mcp_server_table must use the MCPServer's stored timestamps, not datetime.now()."""
manager = MCPServerManager()