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

Merge pull request #33317 from BerriAI/litellm_mcp_field_granular_oauth_discovery

Browse source 
fix(mcp): discover missing OAuth scopes and token_url when authorization_url is set manually
This commit is contained in:
tin-berri 2026-07-15 16:56:26 -07:00 • committed by GitHub
commit 923c325e64
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 568 additions and 31 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -186,6 +186,61 @@ _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES: tuple[MCPAuth, ...] = (
)
def _blank_to_none(value: str | None) -> str | None:
"""Collapse an absent, empty, or whitespace-only string to ``None``.
OAuth endpoint fields are consumed by truthiness-based merges (``row or discovered``) and by the
corroboration gate. A whitespace-only value is truthy to ``or`` but is not a usable endpoint, so
without this the merge would keep the blank value for redirects while the gate treats it as
unpinned and backfills the other fields, yielding a broken half-discovered config. Normalizing
the pinned fields once, at each build entry point, gives every downstream consumer a single
notion of "blank" so those code paths cannot disagree.
"""
if not isinstance(value, str):
return None
return value.strip() or None
def _normalized_authorize_endpoint(url: str) -> str:
"""Compare authorize endpoints on scheme, host, and path only. The default port is elided and
the host is lowercased so ``https://IDP.example.com:443/authorize/`` and
``https://idp.example.com/authorize`` are the same identity; query and trailing slash are not."""
parsed = urlparse(url)
scheme = parsed.scheme.lower()
host = (parsed.hostname or "").lower()
default_port = {"https": 443, "http": 80}.get(scheme)
try:
port = parsed.port
except ValueError:
port = None
authority = host if port is None or port == default_port else f"{host}:{port}"
return f"{scheme}://{authority}{parsed.path.rstrip('/')}"
def _endpoints_corroborate_authorization_url(
source_authorization_url: str | None,
trusted_authorization_url: str | None,
) -> bool:
"""Whether a source's ``token_url``/``registration_url`` may be paired with a trusted authorize
endpoint. This is the single trust rule for adopting OAuth endpoints from any non-manual source.
Discovery is rooted at the MCP resource (RFC 9728), so a compromised upstream can advertise an
attacker-run authorization server. When ``authorization_url`` is admin-pinned, pairing it with a
``token_url`` from a different source is the RFC 9700 authorization-server mix-up: the user signs
in at the trusted authorize endpoint while the gateway redeems the code, with the stored client
secret and PKCE verifier, at the attacker's token endpoint. Endpoints are trustworthy together
only when they share an authorization server, so a source's endpoints are adopted only when the
same source advertised an ``authorization_endpoint`` matching the pinned value. With no pinned
value (``trusted_authorization_url is None``) there is nothing to protect: the authorize endpoint
comes from the same source as the token endpoint, so they corroborate each other by construction.
"""
if not (trusted_authorization_url and trusted_authorization_url.strip()):
return True
return bool(source_authorization_url) and _normalized_authorize_endpoint(
source_authorization_url
) == _normalized_authorize_endpoint(trusted_authorization_url)
def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_server: MCPServer | None) -> None:
"""Keep the last known good OAuth endpoints when a rebuild's re-discovery comes back empty.
@ -193,26 +248,82 @@ def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_serv
during re-discovery downgrades a working server (``authorization_url`` set) to a broken one
(``None``, /authorize 400s) with no configuration change. Mirrors the ``short_prefix``
carry-forward. Skipped when the server's ``url`` or ``auth_type`` changed, since the previous
endpoints may then belong to a different upstream. ``registration_url`` IS carried here even
though ``_persist_discovered_oauth_endpoints`` refuses to write it to the row: carrying only
restores the same in-memory value the previous build already ran with, while persisting it
would flip ``_dcr_bridge_relays_client_registration`` (which keys off the stored column) for
dcr_bridge servers that never had one configured.
endpoints may then belong to a different upstream. ``registration_url`` IS carried even though
``_persist_discovered_oauth_endpoints`` refuses to write it to the row: carrying only restores
the same in-memory value the previous build already ran with, while persisting it would flip
``_dcr_bridge_relays_client_registration`` (which keys off the stored column) for dcr_bridge
servers that never had one configured.
Carry-forward is a non-manual endpoint source, so the same trust rule as discovery applies: the
previous ``token_url``/``registration_url``/``scopes`` are carried only when the previous
``authorization_url`` corroborates the authorize endpoint this build will use, i.e. when the
incoming build has no pinned authorize endpoint (``None`` -> we adopt the previous one too, a
consistent group) or pins the same one. An admin re-pointing ``authorization_url`` to a different
server must not keep serving the old server's token endpoint or granted scopes.
"""
if previous_server is None:
return
if previous_server.url != new_server.url or previous_server.auth_type != new_server.auth_type:
return
may_carry = _endpoints_corroborate_authorization_url(
previous_server.authorization_url, new_server.authorization_url
)
if new_server.authorization_url is None and previous_server.authorization_url:
new_server.authorization_url = previous_server.authorization_url
if new_server.token_url is None and previous_server.token_url:
if may_carry and new_server.token_url is None and previous_server.token_url:
new_server.token_url = previous_server.token_url
if new_server.registration_url is None and previous_server.registration_url:
if may_carry and new_server.registration_url is None and previous_server.registration_url:
new_server.registration_url = previous_server.registration_url
if not new_server.scopes and previous_server.scopes:
if may_carry and not new_server.scopes and previous_server.scopes:
new_server.scopes = previous_server.scopes
def _restrict_discovery_to_corroborated_authorization_server(
metadata: MCPOAuthMetadata | None,
manual_authorization_url: str | None,
server_identifier: str,
is_dcr_bridge: bool,
) -> MCPOAuthMetadata | None:
"""Reject discovered token/registration endpoints a manually pinned authorize endpoint cannot
vouch for (the RFC 9700 authorization-server mix-up).
Discovery is rooted at the MCP resource, so a compromised upstream can advertise an attacker
``token_endpoint``: with ``authorization_url`` admin-pinned but ``token_url`` blank, the merge
would pair the trusted authorize endpoint with that attacker token endpoint, and the gateway would
post the authorization code and client secret there. So the discovered ``token_url`` and
``registration_url`` are kept only if the document corroborates the pin (its
``authorization_endpoint`` matches). ``scopes`` are deliberately NOT gated here: per the MCP
authorization spec Scope Selection Strategy and RFC 9700 §2.3, the scopes a client requests are
resource-driven (the WWW-Authenticate challenge or the RFC 9728 protected-resource
``scopes_supported``), and scope inflation by a compromised resource is bounded by the
authorization server and user consent (RFC 6749 §3.3), not by the client second-guessing the
request. With no pin there is no trust anchor to protect, so discovery is returned as-is.
"""
if metadata is None or not (manual_authorization_url and manual_authorization_url.strip()):
return metadata
if _endpoints_corroborate_authorization_url(metadata.authorization_url, manual_authorization_url):
return metadata
if not metadata.token_url and not metadata.registration_url:
return metadata
bridge_note = (
" The discovered registration_url is rejected with it, so this dcr_bridge server stays on the"
" short-circuit registration arm."
if is_dcr_bridge and metadata.registration_url
else ""
)
verbose_logger.warning(
"MCP OAuth discovery for server %s advertised authorization_endpoint %s, which does not match the "
"manually configured authorization_url %s; rejecting the discovered token_url/registration_url so "
"authorization codes and client credentials only follow the configured authorization server. "
"Configure Token URL manually if the mismatch is intentional.%s",
server_identifier,
_normalized_authorize_endpoint(metadata.authorization_url) if metadata.authorization_url else "<absent>",
_normalized_authorize_endpoint(manual_authorization_url),
bridge_note,
)
return metadata.model_copy(update={"token_url": None, "registration_url": None})
def invalidate_user_env_vars_cache(user_id: str, server_id: str) -> None:
"""Drop a cached entry after the user stores or clears their env var values
so the next request reads the fresh value instead of a stale one."""
@ -1026,12 +1137,15 @@ class MCPServerManager:
)
auth_type = server_config.get("auth_type", None)
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
or self._obo_needs_endpoint_discovery(
auth_type,
server_config.get("token_exchange_endpoint"),
server_config.get("token_url"),
manual_token_url,
)
):
mcp_oauth_metadata = await self._descovery_metadata(
@ -1041,20 +1155,29 @@ class MCPServerManager:
else:
mcp_oauth_metadata = None
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
)
# 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
# entra_obo fail-closed scope precondition and POSTs an empty scope to the IdP.
resolved_scopes = self._extract_scopes(server_config.get("scopes")) or (
mcp_oauth_metadata.scopes if mcp_oauth_metadata else None
gated_oauth_metadata.scopes if gated_oauth_metadata else None
)
resolved_authorization_url = server_config.get("authorization_url") or (
mcp_oauth_metadata.authorization_url if mcp_oauth_metadata else None
resolved_authorization_url = manual_authorization_url or (
gated_oauth_metadata.authorization_url if gated_oauth_metadata else None
)
resolved_token_url = server_config.get("token_url") or (
mcp_oauth_metadata.token_url if mcp_oauth_metadata else None
)
resolved_registration_url = server_config.get("registration_url") or (
mcp_oauth_metadata.registration_url if mcp_oauth_metadata else None
resolved_token_url = manual_token_url or (gated_oauth_metadata.token_url if gated_oauth_metadata else None)
resolved_registration_url = manual_registration_url or (
gated_oauth_metadata.registration_url if gated_oauth_metadata else None
)
config_oauth2_flow = server_config.get("oauth2_flow", None)
@ -1447,13 +1570,17 @@ class MCPServerManager:
auth_type = cast(MCPAuthType, mcp_server.auth_type)
server_url = mcp_server.url
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)
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 mcp_server.authorization_url)
(auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES and not has_all_upstream_oauth_fields)
or self._obo_needs_endpoint_discovery(
auth_type,
mcp_server.token_exchange_endpoint
or (credentials_dict.get("token_exchange_endpoint") if credentials_dict else None),
mcp_server.token_url,
manual_token_url,
)
)
mcp_oauth_metadata = (
@ -1467,12 +1594,22 @@ class MCPServerManager:
if needs_discovery and mcp_oauth_metadata is None:
verbose_logger.warning(
"MCP OAuth discovery yielded no metadata for server %s (%s); "
"OAuth endpoints stay unresolved until a rebuild succeeds",
"OAuth endpoints/scopes stay unresolved until a rebuild succeeds",
mcp_server.server_id,
server_url,
)
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
)
resolved_scopes = scopes or (mcp_oauth_metadata.scopes if mcp_oauth_metadata else None)
resolved_scopes = scopes or (gated_oauth_metadata.scopes if gated_oauth_metadata else None)
new_server = MCPServer(
server_id=mcp_server.server_id,
@ -1492,9 +1629,9 @@ 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,
authorization_url=mcp_server.authorization_url or getattr(mcp_oauth_metadata, "authorization_url", None),
token_url=mcp_server.token_url or getattr(mcp_oauth_metadata, "token_url", None),
registration_url=mcp_server.registration_url or getattr(mcp_oauth_metadata, "registration_url", None),
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),
token_endpoint_auth_method=(
credentials_dict.get("token_endpoint_auth_method") if credentials_dict else None
),
@ -1545,16 +1682,16 @@ class MCPServerManager:
await self._persist_discovered_obo_token_url(
server_id=mcp_server.server_id,
auth_type=auth_type,
existing_token_url=mcp_server.token_url,
existing_token_url=manual_token_url,
discovered_token_url=new_server.token_url,
)
await self._persist_discovered_oauth_endpoints(
server_id=mcp_server.server_id,
auth_type=auth_type,
existing_authorization_url=mcp_server.authorization_url,
existing_token_url=mcp_server.token_url,
existing_authorization_url=manual_authorization_url,
existing_token_url=manual_token_url,
existing_scopes=scopes,
metadata=mcp_oauth_metadata,
metadata=gated_oauth_metadata,
)
return new_server

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

@ -17,6 +17,12 @@ MCPInfo = Dict[str, Any]
class MCPOAuthMetadata(BaseModel):
scopes: Optional[List[str]] = None
"""Resource-driven scopes for the authorization request: the RFC 9728 protected-resource
``scopes_supported``, or the ``scope`` from the WWW-Authenticate 401 challenge when the resource
supplied one, else the authorization server's ``scopes_supported``. This is the scope value a
client requests per the MCP authorization spec Scope Selection Strategy; scope minimization and
inflation control are the authorization server's and user's job at consent (RFC 6749 §3.3), not
the client's."""
authorization_url: Optional[str] = None
token_url: Optional[str] = None
registration_url: Optional[str] = None

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

@ -356,6 +356,86 @@ class TestMCPServerManager:
assert server.oauth2_flow == "authorization_code"
assert server.needs_user_oauth_token is True
@pytest.mark.asyncio
async def test_load_servers_from_config_rejects_uncorroborated_endpoints_but_keeps_resource_scopes(self):
"""A yaml server with a manual authorization_url has the same config-time mix-up exposure as a
DB row: a document advertising a different authorize endpoint has its token_url rejected. The
resource-driven scopes are kept, because scope selection is resource-driven (MCP Scope
Selection Strategy) and scope inflation is bounded by the authorization server at consent, not
by dropping scopes when an endpoint mismatches."""
manager = MCPServerManager()
metadata = MCPOAuthMetadata(
authorization_url="https://attacker.example.com/authorize",
token_url="https://attacker.example.com/token",
scopes=["read", "admin"],
)
config = self._oauth2_config(
oauth2_flow="authorization_code",
authorization_url="https://idp.example.com/authorize",
token_url=None,
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
await manager.load_servers_from_config(config)
server = next(iter(manager.config_mcp_servers.values()))
assert server.authorization_url == "https://idp.example.com/authorize"
assert server.token_url is None
assert server.scopes == ["read", "admin"]
@pytest.mark.asyncio
async def test_load_servers_from_config_fills_token_url_when_metadata_corroborates_manual_authorization_url(self):
"""Corroborated metadata keeps the self-heal on the config path: when the discovered document
advertises the same authorize endpoint the admin pinned, its token_url fills the blank field
and scopes come through resource-driven (the discovered document's resource-preferred scopes),
not the authorization server's own capability list."""
manager = MCPServerManager()
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
scopes=["read", "admin"],
)
config = self._oauth2_config(
oauth2_flow="authorization_code",
authorization_url="https://idp.example.com/authorize/",
token_url=None,
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
await manager.load_servers_from_config(config)
server = next(iter(manager.config_mcp_servers.values()))
assert server.token_url == "https://idp.example.com/token"
assert server.scopes == ["read", "admin"]
@pytest.mark.asyncio
@pytest.mark.parametrize("blank_authorization_url", ["", " "])
async def test_load_servers_from_config_blank_authorization_url_is_not_a_pin(self, blank_authorization_url):
"""A blank authorization_url — empty or whitespace-only — is not a trust anchor, so discovery
backfills the whole set (authorize endpoint, token_url, and its resource-preferred scopes)
from the same chain, exactly as if the field had been omitted. The merge and the corroboration
gate must agree that blank means unpinned; a whitespace value that the merge kept for redirects
while the gate treated as unpinned would strand a broken half-discovered config."""
manager = MCPServerManager()
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
scopes=["read"],
)
config = self._oauth2_config(
oauth2_flow="authorization_code",
authorization_url=blank_authorization_url,
token_url=None,
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
await manager.load_servers_from_config(config)
server = next(iter(manager.config_mcp_servers.values()))
assert server.authorization_url == "https://idp.example.com/authorize"
assert server.token_url == "https://idp.example.com/token"
assert server.scopes == ["read"]
@pytest.mark.asyncio
async def test_load_servers_from_config_non_oauth2_needs_no_flow(self):
manager = MCPServerManager()
@ -1026,7 +1106,6 @@ class TestMCPServerManager:
"""The gateway's relayed authorize flow (used by the browser-only Authorize) needs the
upstream's authorization_url on the registry entry, and these rows never persist one, so
the DB build must discover it the same way oauth2 rows do."""
from types import SimpleNamespace
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
@ -1053,6 +1132,181 @@ class TestMCPServerManager:
assert built.authorization_url == "https://idp.example.com/authorize"
assert built.token_url == "https://idp.example.com/token"
@pytest.mark.asyncio
async def test_build_from_table_backfills_resource_driven_scopes_for_pinned_authorization_url(self):
"""When authorization_url is admin-pinned and corroborated, scopes backfill as the
resource-driven value (the WWW-Authenticate challenge scope, else the RFC 9728
protected-resource scopes_supported), per the MCP authorization spec Scope Selection Strategy.
The client does not restrict scopes to the authorization server's own scopes_supported; scope
minimization and inflation control are the authorization server's and user's job at consent
(RFC 6749 §3.3)."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="manual-auth-url-1",
alias="manual_auth_url",
description="manual authorization_url, blank scopes",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://idp.example.com/authorize",
created_at=datetime.now(),
updated_at=datetime.now(),
)
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
scopes=["read", "admin"],
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)) as mock_discovery:
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
mock_discovery.assert_awaited_once()
assert built.authorization_url == "https://idp.example.com/authorize"
assert built.token_url == "https://idp.example.com/token"
assert built.scopes == ["read", "admin"]
@pytest.mark.asyncio
async def test_build_from_table_whitespace_authorization_url_is_not_a_pin(self):
"""A whitespace-only authorization_url on the row must not be kept for redirects while the
gate treats it as unpinned. It is normalized to unpinned everywhere, so the built server
takes the discovered authorize endpoint, token_url, and scopes as one consistent group
rather than serving the whitespace value with half-discovered fields."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="whitespace-auth-url",
alias="whitespace_auth_url",
description="whitespace authorization_url is not a pin",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url=" ",
created_at=datetime.now(),
updated_at=datetime.now(),
)
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
scopes=["read"],
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
assert built.authorization_url == "https://idp.example.com/authorize"
assert built.token_url == "https://idp.example.com/token"
assert built.scopes == ["read"]
@pytest.mark.asyncio
async def test_build_from_table_fills_endpoints_when_metadata_corroborates_manual_authorization_url(self):
"""A discovered token_url is only trusted next to a manual authorization_url when the same
metadata document advertises that authorize endpoint, and the comparison must tolerate
formatting-only differences (host case, trailing slash, query params like ?prompt=consent)
so hand-copied URLs still self-heal."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="manual-auth-url-2",
alias="manual_auth_url_match",
description="manual authorization_url matching discovery, blank token_url",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://IDP.example.com/authorize/?prompt=consent",
created_at=datetime.now(),
updated_at=datetime.now(),
)
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
registration_url="https://idp.example.com/register",
scopes=["read"],
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)):
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
assert built.authorization_url == "https://IDP.example.com/authorize/?prompt=consent"
assert built.token_url == "https://idp.example.com/token"
assert built.registration_url == "https://idp.example.com/register"
assert built.scopes == ["read"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"advertised_authorization_url",
["https://attacker.example.com/authorize", None],
)
async def test_build_from_table_rejects_uncorroborated_endpoints_but_keeps_resource_scopes(
self, advertised_authorization_url
):
"""Resource-rooted discovery lets a compromised upstream advertise its own authorization
server. With a manual authorization_url pinned, a document that does not corroborate it has
its token_url and registration_url dropped: accepting them would send the code, client secret,
and PKCE verifier to the attacker (config-time RFC 9700 mix-up). The resource-driven scopes
are kept, because scope selection is resource-driven (MCP Scope Selection Strategy) and scope
inflation is bounded by the authorization server at consent (RFC 6749 §3.3), not by dropping
scopes on an endpoint mismatch. Both the in-memory merge and the persisted metadata drop only
the uncorroborated endpoints."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="manual-auth-url-3",
alias="manual_auth_url_mismatch",
description="manual authorization_url, hostile discovery document",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://idp.example.com/authorize",
created_at=datetime.now(),
updated_at=datetime.now(),
)
metadata = MCPOAuthMetadata(
authorization_url=advertised_authorization_url,
token_url="https://attacker.example.com/token",
registration_url="https://attacker.example.com/register",
scopes=["read", "admin"],
)
with (
patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=metadata)),
patch.object(manager, "_persist_discovered_oauth_endpoints", new=AsyncMock()) as mock_persist,
):
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
assert built.authorization_url == "https://idp.example.com/authorize"
assert built.token_url is None
assert built.registration_url is None
assert built.scopes == ["read", "admin"]
persisted_metadata = mock_persist.await_args.kwargs["metadata"]
assert persisted_metadata.token_url is None
assert persisted_metadata.registration_url is None
assert persisted_metadata.scopes == ["read", "admin"]
@pytest.mark.asyncio
async def test_build_from_table_skips_discovery_when_all_upstream_oauth_fields_present(self):
"""A fully hand-configured server (authorization_url, token_url, and scopes all set) has
nothing left for discovery to fill, so the build must not fetch upstream metadata."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="fully-manual-1",
alias="fully_manual",
description="all upstream oauth fields set by the admin",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://idp.example.com/manual-authorize",
token_url="https://idp.example.com/manual-token",
credentials={"scopes": ["calendar.read"]},
created_at=datetime.now(),
updated_at=datetime.now(),
)
with patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=None)) as mock_discovery:
built = await manager.build_mcp_server_from_table(row, credentials_are_encrypted=False)
mock_discovery.assert_not_awaited()
assert built.authorization_url == "https://idp.example.com/manual-authorize"
assert built.token_url == "https://idp.example.com/manual-token"
assert built.scopes == ["calendar.read"]
async def _capture_subject_token(self, call) -> Optional[str]:
"""Run a manager method (via ``call(manager)``) and return the subject_token it threaded
into ``_create_mcp_client``."""
@ -2127,6 +2381,46 @@ class TestMCPServerManager:
assert result.scopes == ["api://some-scope/.default"]
assert result.from_origin_fallback is False
@pytest.mark.asyncio
async def test_descovery_metadata_scopes_are_resource_driven(self):
"""The effective `scopes` are resource-driven: the RFC 9728 protected-resource advertisement
(or WWW-Authenticate challenge) overrides the authorization server's own scopes_supported. This
is the MCP Scope Selection Strategy: the client requests what the resource needs, not the AS's
full capability list."""
manager = MCPServerManager()
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
authorization_server_metadata = MCPOAuthMetadata(
scopes=["as.read", "as.write"],
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
)
with (
patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
),
patch.object(
manager,
"_attempt_well_known_discovery",
AsyncMock(return_value=(["https://idp.example.com"], ["resource.only"])),
),
patch.object(
manager,
"_fetch_authorization_server_metadata",
AsyncMock(return_value=authorization_server_metadata),
),
):
result = await manager._descovery_metadata("https://up.example.com/mcp")
assert result is not None
assert result.scopes == ["resource.only"]
@pytest.mark.asyncio
async def test_fetch_single_authorization_server_metadata_supports_azure_issuer_path(
self,
@ -2252,6 +2546,10 @@ class TestMCPServerManager:
@pytest.mark.asyncio
async def test_load_servers_from_config_overrides_discovery_metadata(self):
"""Config values win per field. The discovered token_url/registration_url do NOT fill the
blanks here: the document advertises a different authorization_endpoint than the manually
configured one, so combining its endpoints with the pinned authorize URL would be the
config-time mix-up the discovery gate exists to prevent."""
manager = MCPServerManager()
discovered_metadata = MCPOAuthMetadata(
@ -2285,8 +2583,8 @@ class TestMCPServerManager:
server = next(iter(manager.config_mcp_servers.values()))
assert server.scopes == ["config"] # config overrides discovery
assert server.authorization_url == "https://config.example.com/auth"
assert server.token_url == "https://discovered.example.com/token"
assert server.registration_url == "https://discovered.example.com/register"
assert server.token_url is None
assert server.registration_url is None
@pytest.mark.asyncio
async def test_load_servers_from_config_filters_blank_scopes(self):
@ -5093,6 +5391,102 @@ class TestMCPServerTimestamps:
_carry_forward_resolved_oauth_endpoints(new_server=explicit, previous_server=previous)
assert explicit.authorization_url == "https://configured.example.com/auth"
def test_carry_forward_does_not_revive_token_url_across_authorization_url_change(self):
"""Carry-forward is a non-manual endpoint source, so it obeys the same trust rule as
discovery: a previous token_url/registration_url belongs to the previous authorization
server, so it must not be pinned to a NEW authorization_url the admin re-pointed to. Without
this, re-pointing authorize to server B while the same MCP url keeps serving A's token
endpoint recreates the RFC 9700 mix-up, durably, and the discovery gate alone cannot catch
it because the stale endpoint comes from the registry, not from discovery."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
_carry_forward_resolved_oauth_endpoints,
)
previous = MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://idp-a.example.com/authorize",
token_url="https://idp-a.example.com/token",
registration_url="https://idp-a.example.com/register",
)
repointed = MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://idp-b.example.com/authorize",
)
_carry_forward_resolved_oauth_endpoints(new_server=repointed, previous_server=previous)
assert repointed.authorization_url == "https://idp-b.example.com/authorize"
assert repointed.token_url is None
assert repointed.registration_url is None
def test_carry_forward_restores_endpoints_when_authorization_url_unchanged(self):
"""The last-known-good path still works: a rebuild whose discovery blipped (no authorize
endpoint) adopts the previous authorize endpoint AND its token endpoint together as a
consistent group, and a rebuild that re-pins the same authorize endpoint (formatting aside)
keeps carrying the corroborated token endpoint."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
_carry_forward_resolved_oauth_endpoints,
)
def previous() -> MCPServer:
return MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
registration_url="https://idp.example.com/register",
)
blipped = MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url=None,
)
_carry_forward_resolved_oauth_endpoints(new_server=blipped, previous_server=previous())
assert blipped.authorization_url == "https://idp.example.com/authorize"
assert blipped.token_url == "https://idp.example.com/token"
assert blipped.registration_url == "https://idp.example.com/register"
same_authorize = MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
authorization_url="https://IDP.example.com:443/authorize/",
)
_carry_forward_resolved_oauth_endpoints(new_server=same_authorize, previous_server=previous())
assert same_authorize.token_url == "https://idp.example.com/token"
assert same_authorize.registration_url == "https://idp.example.com/register"
def test_normalized_authorize_endpoint_treats_default_port_and_slash_as_identity(self):
"""The corroboration check must not fail on formatting-only differences an IdP legitimately
emits: default port, trailing slash, host case, and query string are not identity, but a
non-default port is."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
_normalized_authorize_endpoint,
)
canonical = _normalized_authorize_endpoint("https://idp.example.com/authorize")
assert _normalized_authorize_endpoint("https://idp.example.com:443/authorize") == canonical
assert _normalized_authorize_endpoint("https://IDP.example.com/authorize/") == canonical
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_build_mcp_server_table_preserves_timestamps(self):
"""_build_mcp_server_table must use the MCPServer's stored timestamps, not datetime.now()."""
manager = MCPServerManager()