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

Merge pull request #33450 from BerriAI/litellm_mcp_issuer_anchored_discovery

Browse source 
feat(mcp): issuer-anchored OAuth discovery (RFC 8414 §3.3) to close the authorization-server mix-up
This commit is contained in:
tin-berri 2026-07-16 11:55:35 -07:00 • committed by GitHub
commit db800152c0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 983 additions and 56 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

36
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",
@ -60,6 +61,13 @@ _AUTH_FLOW_SCOPED_FIELDS: frozenset = frozenset(
}
)
def _blank_to_none(value: Optional[str]) -> Optional[str]:
if not isinstance(value, str):
return None
return value.strip() or None
# Token-exchange settings with dedicated columns that also exist on
# ``MCPCredentials`` as a legacy shape (rows and REST callers that predate the
# columns). Every write lifts blob values into the columns and strips them from
@ -697,13 +705,15 @@ async def update_mcp_server(
# of being reset to a schema default (transport=sse, allow_all_keys=False...).
data_dict = _prepare_mcp_server_data(data, exclude_unset=True, fields_set=fields_set)
# Pre-fetch existing record once if we need it for auth_type or credential logic
# Pre-fetch existing record once if we need it for auth_type, url, or credential logic
existing = None
has_credentials = "credentials" in data_dict and data_dict["credentials"] is not None
# An explicit token-exchange column write (set or clear) also migrates the
# legacy blob copies below, so the existing row is needed for those updates.
explicit_te_write = bool(_TOKEN_EXCHANGE_COLUMN_FIELDS & data_dict.keys())
if data.auth_type or has_credentials or explicit_te_write:
url_provided = "url" in data_dict and data_dict["url"] is not None
issuer_provided = "issuer" in data_dict
if data.auth_type or has_credentials or explicit_te_write or url_provided or issuer_provided:
existing = await MCPServerRepository(prisma_client).table.find_unique(where={"server_id": data.server_id})
auth_type_changed = bool(
@ -711,13 +721,30 @@ async def update_mcp_server(
and existing
and _credential_auth_class(existing.auth_type) != _credential_auth_class(data.auth_type)
)
# A url change re-points the server at a potentially different upstream, so any discovered or
# trust-on-first-use OAuth endpoints/issuer belong to the old upstream and must re-discover.
url_changed = bool(url_provided and existing and existing.url != data_dict["url"])
old_issuer = _blank_to_none(getattr(existing, "issuer", None)) if existing else None
issuer_changed = bool(
issuer_provided and old_issuer is not None and _blank_to_none(data_dict.get("issuer")) != old_issuer
)
# Clear stale credentials when auth_type changes but no new credentials provided
if auth_type_changed and "credentials" not in data_dict:
data_dict["credentials"] = None
if auth_type_changed:
data_dict.update({field: None for field in _AUTH_FLOW_SCOPED_FIELDS if field not in data_dict})
if auth_type_changed or url_changed or issuer_changed:
# Clear each auth-flow-scoped field that the caller either omitted (partial update) or
# resubmitted unchanged. The edit form re-sends every field, so a stale issuer/endpoint
# belonging to the old upstream would otherwise survive a url/auth_type change and win in the
# resolution merge; only a genuinely new submitted value is kept.
data_dict.update(
{
field: None
for field in _AUTH_FLOW_SCOPED_FIELDS
if field not in data_dict or data_dict[field] == getattr(existing, field, None)
}
)
# An explicit column write that does not touch credentials must still migrate
# the row's legacy blob copies: lift values for columns the caller left
@ -1181,6 +1208,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),

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

@ -201,6 +201,38 @@ def _blank_to_none(value: str | None) -> str | None:
return value.strip() or None
def _uses_issuer_anchor(manual_issuer: str | None, is_discovery_auth_type: bool) -> bool:
"""Whether the endpoints are authoritatively anchored to an admin-pinned issuer (RFC 8414 §3.3).
This is the trust/provenance property, distinct from whether the ``issuer`` field is merely
populated: a trust-on-first-use discovered issuer sets ``issuer`` for token identity but is NOT
anchored, so its endpoints stay resource-rooted. Anchoring holds only when the issuer was pinned
(present on the row/config) on a discovery auth type. Every consumer of "is this anchored" reads
this one definition, so the answer cannot diverge across build paths.
"""
return _blank_to_none(manual_issuer) is not None and is_discovery_auth_type
def _endpoints_yield_to_issuer(
issuer: str | None,
is_discovery_auth_type: bool,
authorization_url: str | None,
token_url: str | None,
registration_url: str | None,
) -> tuple[str | None, str | None, str | None]:
"""The single rule that makes an admin-configured ``issuer`` the sole authoritative endpoint
source (RFC 8414 §3.3): when it is set for a discovery auth type, the stored/manual
``authorization_url``/``token_url``/``registration_url`` do not apply. They neither anchor nor
short-circuit discovery, never override the issuer document in the merge, and never substitute for
it when the issuer fetch fails (fail-closed). Returns the endpoint values that remain in force,
i.e. all ``None`` when issuer-anchored, else the inputs unchanged. Called at every resolution site
so the invariant holds in one place instead of being re-derived per merge.
"""
if issuer is not None and is_discovery_auth_type:
return None, None, None
return authorization_url, token_url, registration_url
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
@ -217,6 +249,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,
@ -260,11 +303,27 @@ def _carry_forward_resolved_oauth_endpoints(new_server: MCPServer, previous_serv
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.
When the server is issuer-anchored (``issuer_is_anchored`` -- a pinned issuer on a discovery auth
type), the endpoints come solely from the §3.3-validated issuer document, so carry-forward is
skipped entirely for its endpoints: a failed issuer fetch leaves them ``None`` and must stay
``None`` (fail-closed), never resurrected from the previous registry entry. A merely discovered
(trust-on-first-use) issuer is NOT anchored -- ``issuer`` is set for token identity but the
endpoints are resource-rooted, so they still carry forward as last-known-good, gated by the
corroboration check below like any other resource-rooted server. Scopes stay resource-driven and
can carry either way.
"""
if previous_server is None:
return
if previous_server.url != new_server.url or previous_server.auth_type != new_server.auth_type:
return
if new_server.issuer_is_anchored:
# Endpoints come solely from the §3.3-validated issuer document; a failed fetch stays
# fail-closed and must not be resurrected from the previous entry. Only the resource-driven
# scopes carry as last-known-good.
if not new_server.scopes and previous_server.scopes:
new_server.scopes = previous_server.scopes
return
may_carry = _endpoints_corroborate_authorization_url(
previous_server.authorization_url, new_server.authorization_url
)
@ -1137,34 +1196,48 @@ 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 = _uses_issuer_anchor(manual_issuer, is_discovery_auth_type)
manual_authorization_url, manual_token_url, manual_registration_url = _endpoints_yield_to_issuer(
manual_issuer,
is_discovery_auth_type,
manual_authorization_url,
manual_token_url,
manual_registration_url,
)
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, server_url)
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
@ -1179,6 +1252,12 @@ class MCPServerManager:
resolved_registration_url = manual_registration_url or (
gated_oauth_metadata.registration_url if gated_oauth_metadata else None
)
discovered_issuer = (
gated_oauth_metadata.discovered_issuer
if gated_oauth_metadata and not gated_oauth_metadata.from_origin_fallback
else None
)
effective_issuer = manual_issuer or discovered_issuer
config_oauth2_flow = server_config.get("oauth2_flow", None)
if auth_type == MCPAuth.oauth2 and config_oauth2_flow not in (
@ -1227,6 +1306,8 @@ class MCPServerManager:
client_secret=server_config.get("client_secret", None),
oauth2_flow=self._explicit_oauth2_flow(config_oauth2_flow),
scopes=resolved_scopes,
issuer=effective_issuer,
issuer_is_anchored=use_issuer_anchor,
authorization_url=resolved_authorization_url,
token_url=resolved_token_url,
registration_url=resolved_registration_url,
@ -1487,6 +1568,52 @@ class MCPServerManager:
decrypt_global_env_var_values(env_vars_list)
return env_vars_list
async def _resolve_table_oauth_metadata(
self,
*,
mcp_server: LiteLLM_MCPServerTable,
auth_type: MCPAuthType,
server_url: Optional[str],
manual_issuer: Optional[str],
manual_authorization_url: Optional[str],
manual_token_url: Optional[str],
is_discovery_auth_type: bool,
use_issuer_anchor: bool,
scopes: Optional[list[str]],
token_exchange_endpoint: Optional[str],
) -> Optional[MCPOAuthMetadata]:
has_all_upstream_oauth_fields = bool(manual_authorization_url and manual_token_url and scopes)
needs_discovery = bool(server_url) and (
(is_discovery_auth_type and not has_all_upstream_oauth_fields)
or self._obo_needs_endpoint_discovery(auth_type, token_exchange_endpoint, manual_token_url)
)
if not needs_discovery:
mcp_oauth_metadata: Optional[MCPOAuthMetadata] = None
elif use_issuer_anchor and manual_issuer is not None:
mcp_oauth_metadata = await self._fetch_issuer_anchored_oauth_metadata(manual_issuer, server_url)
else:
mcp_oauth_metadata = await self._descovery_metadata(
server_url=server_url, # type: ignore[arg-type]
allow_origin_fallback=is_discovery_auth_type,
)
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,
)
if use_issuer_anchor:
return mcp_oauth_metadata
if is_discovery_auth_type:
return _restrict_discovery_to_corroborated_authorization_server(
mcp_oauth_metadata,
manual_authorization_url,
mcp_server.server_id,
bool(getattr(mcp_server, "dcr_bridge", None)),
)
return mcp_oauth_metadata
async def build_mcp_server_from_table(
self,
mcp_server: LiteLLM_MCPServerTable,
@ -1570,46 +1697,38 @@ 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)
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)
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),
manual_token_url,
)
is_discovery_auth_type = auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
use_issuer_anchor = _uses_issuer_anchor(manual_issuer, is_discovery_auth_type)
manual_authorization_url, manual_token_url, manual_registration_url = _endpoints_yield_to_issuer(
manual_issuer, is_discovery_auth_type, manual_authorization_url, manual_token_url, manual_registration_url
)
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,
)
if needs_discovery
else None
token_exchange_endpoint = mcp_server.token_exchange_endpoint or (
credentials_dict.get("token_exchange_endpoint") if credentials_dict else None
)
if needs_discovery 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(
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
gated_oauth_metadata = await self._resolve_table_oauth_metadata(
mcp_server=mcp_server,
auth_type=auth_type,
server_url=server_url,
manual_issuer=manual_issuer,
manual_authorization_url=manual_authorization_url,
manual_token_url=manual_token_url,
is_discovery_auth_type=is_discovery_auth_type,
use_issuer_anchor=use_issuer_anchor,
scopes=scopes,
token_exchange_endpoint=token_exchange_endpoint,
)
resolved_scopes = scopes or (gated_oauth_metadata.scopes if gated_oauth_metadata else None)
discovered_issuer = (
gated_oauth_metadata.discovered_issuer
if gated_oauth_metadata and not gated_oauth_metadata.from_origin_fallback
else None
)
effective_issuer = manual_issuer or discovered_issuer
new_server = MCPServer(
server_id=mcp_server.server_id,
@ -1629,6 +1748,8 @@ 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=effective_issuer,
issuer_is_anchored=use_issuer_anchor,
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),
@ -1688,10 +1809,12 @@ class MCPServerManager:
await self._persist_discovered_oauth_endpoints(
server_id=mcp_server.server_id,
auth_type=auth_type,
existing_issuer=manual_issuer,
existing_authorization_url=manual_authorization_url,
existing_token_url=manual_token_url,
existing_scopes=scopes,
metadata=gated_oauth_metadata,
is_issuer_anchored=use_issuer_anchor,
)
return new_server
@ -1735,10 +1858,12 @@ class MCPServerManager:
*,
server_id: str,
auth_type: MCPAuthType | None,
existing_issuer: str | None,
existing_authorization_url: str | None,
existing_token_url: str | None,
existing_scopes: list[str] | None,
metadata: MCPOAuthMetadata | None,
is_issuer_anchored: bool = False,
) -> None:
"""Write freshly discovered OAuth endpoints back onto the DB row.
@ -1752,19 +1877,37 @@ class MCPServerManager:
because ``_dcr_bridge_relays_client_registration`` keys off that column. Best-effort: a
failed write re-discovers on the next build. Scopes go through ``update_mcp_server`` so
they merge into the credentials blob without touching the stored client credentials.
For an issuer-anchored server (``is_issuer_anchored``) the endpoints are re-derived from the
§3.3-validated issuer document on every build, so they are NOT persisted into the endpoint
columns: persisting them would make the next build see populated endpoints and treat them as
authoritative stored values, defeating the "endpoints come solely from the issuer" invariant.
Only the resource-driven scopes are persisted for such servers.
"""
if auth_type not in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES:
return
if metadata is None or metadata.from_origin_fallback:
return
issuer_update = (
{"issuer": metadata.discovered_issuer} if metadata.discovered_issuer and not existing_issuer else {}
)
authorization_url_update = (
{"authorization_url": metadata.authorization_url}
if metadata.authorization_url and not existing_authorization_url
if metadata.authorization_url and not existing_authorization_url and not is_issuer_anchored
else {}
)
token_url_update = (
{"token_url": metadata.token_url}
if metadata.token_url and not existing_token_url and not is_issuer_anchored
else {}
)
token_url_update = {"token_url": metadata.token_url} if metadata.token_url and not existing_token_url else {}
scopes_update = {"credentials": {"scopes": metadata.scopes}} if metadata.scopes and not existing_scopes else {}
updates: dict[str, object] = {**authorization_url_update, **token_url_update, **scopes_update}
updates: dict[str, object] = {
**issuer_update,
**authorization_url_update,
**token_url_update,
**scopes_update,
}
if not updates:
return
from litellm.proxy._experimental.mcp_server.db import ( # noqa: PLC0415 # db.py imports this module at load
@ -3337,8 +3480,41 @@ class MCPServerManager:
return metadata
return None
async def _fetch_issuer_anchored_oauth_metadata(
self, issuer: str, server_url: Optional[str]
) -> Optional[MCPOAuthMetadata]:
"""RFC 8414 issuer-anchored discovery for the OAuth endpoints, with resource-driven scopes.
Fetch authorization-server metadata from the admin-configured issuer's own origin and adopt
its ``token_endpoint``/``registration_endpoint`` 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 advertises, the endpoints 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 endpoint fetch is treated
as same-authority and is not subject to the resource-scoped SSRF shortcut.
Scopes are NOT taken from the issuer document. Per the MCP authorization spec Scope Selection
Strategy and RFC 9728, the scopes a client requests are resource-driven (the WWW-Authenticate
challenge or the protected-resource ``scopes_supported``), so the resource's advertised scopes
are fetched separately and used; the resource can influence only the requested scope, which
the authorization server and user consent bound (RFC 6749 §3.3), never the token endpoint.
"""
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 stay unresolved until a rebuild succeeds",
issuer,
)
return None
resource_metadata = (
await self._descovery_metadata(server_url, allow_origin_fallback=False) if server_url else None
)
resource_scopes = resource_metadata.scopes if resource_metadata else None
return metadata.model_copy(update={"scopes": resource_scopes})
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,20 +3558,33 @@ 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"),
token_url=data.get("token_endpoint"),
registration_url=data.get("registration_endpoint"),
discovered_issuer=claimed_issuer if isinstance(claimed_issuer, str) and claimed_issuer else None,
)
if any(
@ -5116,6 +5305,7 @@ class MCPServerManager:
command=getattr(server, "command", None),
args=getattr(server, "args", None) or [],
env=getattr(server, "env", None) or {},
issuer=server.issuer,
authorization_url=server.authorization_url,
token_url=server.token_url,
registration_url=server.registration_url,
@ -5225,6 +5415,7 @@ class MCPServerManager:
command=getattr(server, "command", None),
args=getattr(server, "args", None) or [],
env=getattr(server, "env", None) or {},
issuer=server.issuer,
authorization_url=server.authorization_url,
token_url=server.token_url,
registration_url=server.registration_url,

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

3
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -536,6 +536,7 @@ if MCP_AVAILABLE:
sanitized.env = {}
sanitized.command = None
sanitized.args = []
sanitized.issuer = None
sanitized.authorization_url = None
sanitized.token_url = None
sanitized.registration_url = None
@ -581,6 +582,7 @@ if MCP_AVAILABLE:
sanitized.teams = []
sanitized.env_vars = None
sanitized.issuer = None
sanitized.authorization_url = None
sanitized.token_url = None
sanitized.registration_url = None
@ -686,6 +688,7 @@ if MCP_AVAILABLE:
command=payload.command,
args=payload.args,
env=payload.env,
issuer=payload.issuer,
authorization_url=payload.authorization_url,
token_url=payload.token_url,
registration_url=payload.registration_url,

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?

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

@ -26,6 +26,11 @@ class MCPOAuthMetadata(BaseModel):
authorization_url: Optional[str] = None
token_url: Optional[str] = None
registration_url: Optional[str] = None
discovered_issuer: Optional[str] = None
"""The ``issuer`` the authorization-server metadata document self-attests (RFC 8414). Persisted
trust-on-first-use as the server's ``issuer`` when none is configured, so that later rebuilds
anchor discovery on it (RFC 8414 §3.3) and a subsequently compromised resource cannot re-point
it. Never overwrites an admin-configured issuer."""
from_origin_fallback: bool = False
"""True when the metadata came from guessing the resource origin as its authorization
server rather than from an RFC 9728/8414-advertised document. Guessed endpoints are
@ -60,6 +65,8 @@ class MCPServer(BaseModel):
# OAuth-specific fields
client_id: Optional[str] = None
client_secret: Optional[str] = None
issuer: Optional[str] = None
issuer_is_anchored: bool = False
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?

160
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_partial_update.py
View file

@ -203,6 +203,7 @@ async def test_auth_type_switch_clears_stale_flow_scoped_fields():
data_dict = await _run_update_with_existing(data, existing_auth_type="oauth2")
for stale_field in (
"issuer",
"authorization_url",
"token_url",
"registration_url",
@ -217,6 +218,165 @@ async def test_auth_type_switch_clears_stale_flow_scoped_fields():
assert _credentials_cleared(data_dict["credentials"])
@pytest.mark.asyncio
async def test_url_change_clears_stale_discovered_oauth_fields():
"""Re-pointing the server url at a potentially different upstream must clear the discovered or
trust-on-first-use OAuth issuer and endpoints, so the new upstream re-discovers instead of
anchoring on the previous upstream's issuer (RFC 8414 §3.3 against a stale anchor)."""
mock_prisma = _mock_prisma()
existing = MagicMock()
existing.auth_type = "oauth2"
existing.url = "https://old.example.com/mcp"
existing.credentials = None
mock_prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=existing)
data = UpdateMCPServerRequest(server_id="my-test-server", url="https://new.example.com/mcp")
await update_mcp_server(mock_prisma, data, "test-user")
data_dict = mock_prisma.db.litellm_mcpservertable.update.call_args[1]["data"]
assert data_dict["url"] == "https://new.example.com/mcp"
for stale_field in ("issuer", "authorization_url", "token_url", "registration_url"):
assert data_dict[stale_field] is None, f"{stale_field} must be cleared on url change"
@pytest.mark.asyncio
async def test_url_change_clears_stale_oauth_fields_even_when_resubmitted_unchanged():
"""The edit form re-sends every field, so a URL change arrives WITH the previous upstream's issuer
and endpoints in the payload. Those resubmitted-unchanged values are stale and must still clear
(otherwise they survive the url change and win in the resolution merge). A genuinely new value the
caller changed in the same submit is kept."""
mock_prisma = _mock_prisma()
existing = MagicMock()
existing.auth_type = "oauth2"
existing.url = "https://old.example.com/mcp"
existing.credentials = None
existing.issuer = "https://old-idp.example.com"
existing.token_url = "https://old-idp.example.com/token"
existing.authorization_url = "https://old-idp.example.com/authorize"
mock_prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=existing)
data = UpdateMCPServerRequest(
server_id="my-test-server",
url="https://new.example.com/mcp",
issuer="https://old-idp.example.com", # resubmitted unchanged -> stale, must clear
token_url="https://old-idp.example.com/token", # resubmitted unchanged -> stale, must clear
authorization_url="https://new-idp.example.com/authorize", # genuinely changed -> kept
)
await update_mcp_server(mock_prisma, data, "test-user")
data_dict = mock_prisma.db.litellm_mcpservertable.update.call_args[1]["data"]
assert data_dict["issuer"] is None
assert data_dict["token_url"] is None
assert data_dict["authorization_url"] == "https://new-idp.example.com/authorize"
@pytest.mark.asyncio
async def test_clearing_pinned_issuer_clears_stale_oauth_endpoints():
"""Clearing a previously pinned issuer must not revive the endpoints resolved under it. Under an
issuer anchor the endpoints come solely from the issuer document and are not persisted, but a row
that was resource-rooted before the pin can still hold stale authorization_url/token_url; clearing
the anchor without clearing those would let them win the resolution merge and be posted to without
fresh discovery (RFC 8414 §3.3 provenance)."""
mock_prisma = _mock_prisma()
existing = MagicMock()
existing.auth_type = "oauth2"
existing.url = "https://same.example.com/mcp"
existing.credentials = None
existing.issuer = "https://pinned-idp.example.com"
existing.token_url = "https://pinned-idp.example.com/token"
existing.authorization_url = "https://pinned-idp.example.com/authorize"
mock_prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=existing)
data = UpdateMCPServerRequest(
server_id="my-test-server",
issuer="", # admin clears the anchor; url and auth_type unchanged
token_url="https://pinned-idp.example.com/token",
authorization_url="https://pinned-idp.example.com/authorize",
)
await update_mcp_server(mock_prisma, data, "test-user")
data_dict = mock_prisma.db.litellm_mcpservertable.update.call_args[1]["data"]
assert data_dict["token_url"] is None
assert data_dict["authorization_url"] is None
@pytest.mark.asyncio
async def test_repointing_pinned_issuer_clears_stale_endpoints_keeps_new_issuer():
"""Re-pointing the issuer to a different authorization server invalidates the old issuer's
endpoints while keeping the new issuer the admin submitted."""
mock_prisma = _mock_prisma()
existing = MagicMock()
existing.auth_type = "oauth2"
existing.url = "https://same.example.com/mcp"
existing.credentials = None
existing.issuer = "https://old-idp.example.com"
existing.token_url = "https://old-idp.example.com/token"
existing.authorization_url = "https://old-idp.example.com/authorize"
mock_prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=existing)
data = UpdateMCPServerRequest(
server_id="my-test-server",
issuer="https://new-idp.example.com",
token_url="https://old-idp.example.com/token", # resubmitted stale -> must clear
authorization_url="https://old-idp.example.com/authorize", # resubmitted stale -> must clear
)
await update_mcp_server(mock_prisma, data, "test-user")
data_dict = mock_prisma.db.litellm_mcpservertable.update.call_args[1]["data"]
assert data_dict["issuer"] == "https://new-idp.example.com"
assert data_dict["token_url"] is None
assert data_dict["authorization_url"] is None
@pytest.mark.asyncio
async def test_establishing_issuer_first_time_preserves_discovered_fields():
"""Establishing an issuer for the first time (None -> X), which is exactly what the trust-on-first-use
discovery write-back does, must NOT clear the endpoints or oauth2_flow it discovered in the same
write. Only an issuer that was already pinned and is now changed or cleared invalidates its
endpoints, so the discovery persist cannot wipe the fields it just resolved."""
mock_prisma = _mock_prisma()
existing = MagicMock()
existing.auth_type = "oauth2"
existing.url = "https://same.example.com/mcp"
existing.credentials = None
existing.issuer = None
mock_prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=existing)
data = UpdateMCPServerRequest(
server_id="my-test-server",
issuer="https://discovered-idp.example.com",
authorization_url="https://discovered-idp.example.com/authorize",
token_url="https://discovered-idp.example.com/token",
oauth2_flow="authorization_code",
)
await update_mcp_server(mock_prisma, data, "mcp_oauth_discovery")
data_dict = mock_prisma.db.litellm_mcpservertable.update.call_args[1]["data"]
assert data_dict["issuer"] == "https://discovered-idp.example.com"
assert data_dict["authorization_url"] == "https://discovered-idp.example.com/authorize"
assert data_dict["token_url"] == "https://discovered-idp.example.com/token"
assert data_dict.get("oauth2_flow") == "authorization_code"
@pytest.mark.asyncio
async def test_unchanged_url_does_not_clear_discovered_oauth_fields():
"""A partial update that resends the same url (or omits it) must not clear the discovered OAuth
fields, so a routine save does not force needless re-discovery."""
mock_prisma = _mock_prisma()
existing = MagicMock()
existing.auth_type = "oauth2"
existing.url = "https://same.example.com/mcp"
existing.credentials = None
mock_prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=existing)
data = UpdateMCPServerRequest(server_id="my-test-server", url="https://same.example.com/mcp")
await update_mcp_server(mock_prisma, data, "test-user")
data_dict = mock_prisma.db.litellm_mcpservertable.update.call_args[1]["data"]
for preserved_field in ("issuer", "authorization_url", "token_url", "registration_url"):
assert preserved_field not in data_dict, f"{preserved_field} must not be cleared when url is unchanged"
@pytest.mark.asyncio
async def test_auth_type_switch_keeps_explicitly_provided_flow_fields():
"""Fields explicitly provided alongside the auth_type switch must survive it."""

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

@ -1166,6 +1166,67 @@ class TestMCPServerManager:
assert built.token_url == "https://idp.example.com/token"
assert built.scopes == ["read", "admin"]
@pytest.mark.asyncio
async def test_build_from_table_reflects_discovered_issuer_trust_on_first_use(self):
"""An unpinned server resolves endpoints resource-rooted on first discovery and records the
discovered issuer trust-on-first-use. The returned in-memory server must carry that discovered
issuer so the registry matches what gets persisted to the row; otherwise the OAuth token
identity (which includes issuer) differs between this build and the next rebuild, forcing a
spurious re-auth. Endpoints and issuer come from the same authorization-server document, so
they are consistent."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="tofu-issuer-1",
alias="tofu_issuer",
description="unpinned, discovers its issuer",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
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"],
discovered_issuer="https://idp.example.com",
)
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.issuer == "https://idp.example.com"
assert built.issuer_is_anchored is False
assert built.authorization_url == "https://idp.example.com/authorize"
@pytest.mark.asyncio
async def test_build_from_table_origin_fallback_issuer_is_not_reflected(self):
"""An origin-fallback discovery is a guess that is deliberately never persisted, so the built
server must not claim an issuer the row will not hold; otherwise in-memory and DB would
disagree in the opposite direction."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="origin-fallback-1",
alias="origin_fallback",
description="unpinned, origin-fallback discovery",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
created_at=datetime.now(),
updated_at=datetime.now(),
)
metadata = MCPOAuthMetadata(
authorization_url="https://up.example.com/authorize",
token_url="https://up.example.com/token",
discovered_issuer="https://up.example.com",
from_origin_fallback=True,
)
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.issuer is None
@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
@ -1230,6 +1291,159 @@ class TestMCPServerManager:
assert built.registration_url == "https://idp.example.com/register"
assert built.scopes == ["read"]
@pytest.mark.asyncio
async def test_build_from_table_uses_issuer_anchored_endpoints_when_issuer_configured(self):
"""When an admin configures an issuer, the build takes its endpoints from the issuer-anchored
fetch (RFC 8414 §3.3) rather than the resource-rooted corroboration path. The build path does
not call _descovery_metadata directly; the issuer-anchored helper is responsible for combining
issuer endpoints with resource-driven scopes internally, and is invoked with the server url so
it can fetch those scopes."""
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(),
)
resolved = 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"],
)
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=resolved)) 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", "https://up.example.com/mcp")
resource_rooted.assert_not_awaited()
assert built.issuer == "https://idp.example.com"
assert built.issuer_is_anchored is True
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_fetch_issuer_anchored_metadata_takes_endpoints_from_issuer_scopes_from_resource(self):
"""The issuer-anchored helper adopts token_endpoint/registration_endpoint from the pinned
issuer's own §3.3-validated document, but the scopes are resource-driven: it fetches the
resource's advertised scopes and uses those, not the issuer document's scopes_supported. This
keeps endpoint trust anchored on the issuer while scope selection stays resource-driven per the
MCP Scope Selection Strategy."""
manager = MCPServerManager()
issuer_document = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
registration_url="https://idp.example.com/register",
scopes=["as.everything"],
)
resource_document = MCPOAuthMetadata(scopes=["resource.read"])
with (
patch.object(
manager, "_fetch_single_authorization_server_metadata", new=AsyncMock(return_value=issuer_document)
) as issuer_fetch,
patch.object(manager, "_descovery_metadata", new=AsyncMock(return_value=resource_document)) as resource_fetch,
):
result = await manager._fetch_issuer_anchored_oauth_metadata(
"https://idp.example.com", "https://up.example.com/mcp"
)
issuer_fetch.assert_awaited_once_with(
"https://idp.example.com", "https://idp.example.com", require_issuer="https://idp.example.com"
)
resource_fetch.assert_awaited_once()
assert result is not None
assert result.token_url == "https://idp.example.com/token"
assert result.registration_url == "https://idp.example.com/register"
assert result.scopes == ["resource.read"]
@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
async def test_build_from_table_issuer_anchor_overrides_stored_endpoints_even_when_populated(self):
"""When an issuer is pinned, the endpoints come SOLELY from the §3.3-validated issuer document
and win over any stored/manual endpoint values, even a fully-populated row. Otherwise an
attacker who controls a stored token endpoint keeps receiving codes/secrets after an admin
pins a trusted issuer: `needs_discovery` must not short-circuit on populated fields, and the
issuer's endpoints must override the stored ones."""
manager = MCPServerManager()
row = LiteLLM_MCPServerTable(
server_id="issuer-anchored-populated",
alias="issuer_anchored_populated",
description="issuer set, but stale/hostile endpoints already stored",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
issuer="https://idp.example.com",
authorization_url="https://attacker.example.com/authorize",
token_url="https://attacker.example.com/steal",
credentials={"scopes": ["stale"]},
created_at=datetime.now(),
updated_at=datetime.now(),
)
issuer_resolved = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
scopes=["read"],
)
with (
patch.object(
manager, "_fetch_issuer_anchored_oauth_metadata", new=AsyncMock(return_value=issuer_resolved)
) as anchored,
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)
anchored.assert_awaited_once_with("https://idp.example.com", "https://up.example.com/mcp")
assert built.authorization_url == "https://idp.example.com/authorize"
assert built.token_url == "https://idp.example.com/token"
assert built.token_url != "https://attacker.example.com/steal"
# The issuer-anchored endpoints are never persisted into the endpoint columns, so a later
# build cannot treat them as authoritative stored values.
assert mock_persist.await_args.kwargs["is_issuer_anchored"] is True
@pytest.mark.asyncio
@pytest.mark.parametrize(
"advertised_authorization_url",
@ -2462,6 +2676,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,
@ -2489,6 +2775,37 @@ class TestMCPServerManager:
assert result.authorization_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/authorize"
assert result.token_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/token"
@pytest.mark.asyncio
async def test_azure_heuristic_reachable_under_require_issuer(self):
"""Under issuer-anchored discovery (require_issuer set), an Entra issuer whose OIDC document
cannot be fetched still gets the deterministic Azure endpoint construction. The heuristic
derives the endpoints from the pinned issuer's own tenant URL, so it is authoritative-by-
construction and safe under require_issuer; only a non-Entra issuer stays fail-closed (None)."""
manager = MCPServerManager()
issuer = "https://login.microsoftonline.com/test-tenant-id/v2.0"
request = httpx.Request("GET", issuer)
response_obj = httpx.Response(status_code=404, request=request)
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock(
side_effect=httpx.HTTPStatusError("not found", request=request, response=response_obj)
)
mock_client = MagicMock()
mock_client.get = AsyncMock(return_value=mock_response)
with patch(
"litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
return_value=mock_client,
):
azure = await manager._fetch_single_authorization_server_metadata(issuer, issuer, require_issuer=issuer)
non_entra = await manager._fetch_single_authorization_server_metadata(
"https://idp.example.com", "https://idp.example.com", require_issuer="https://idp.example.com"
)
assert azure is not None
assert azure.token_url == "https://login.microsoftonline.com/test-tenant-id/oauth2/v2.0/token"
assert non_entra is None
@pytest.mark.asyncio
async def test_descovery_metadata_falls_back_to_origin_when_no_auth_servers(self):
manager = MCPServerManager()
@ -5194,6 +5511,7 @@ class TestMCPServerTimestamps:
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.api_key,
existing_issuer=None,
existing_authorization_url=None,
existing_token_url=None,
existing_scopes=None,
@ -5202,6 +5520,7 @@ class TestMCPServerTimestamps:
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer=None,
existing_authorization_url=None,
existing_token_url=None,
existing_scopes=None,
@ -5210,6 +5529,7 @@ class TestMCPServerTimestamps:
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer=None,
existing_authorization_url=None,
existing_token_url=None,
existing_scopes=None,
@ -5218,6 +5538,7 @@ class TestMCPServerTimestamps:
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer=None,
existing_authorization_url="https://configured.example.com/authorize",
existing_token_url="https://configured.example.com/token",
existing_scopes=["configured"],
@ -5243,6 +5564,7 @@ class TestMCPServerTimestamps:
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer=None,
existing_authorization_url=None,
existing_token_url="https://configured.example.com/token",
existing_scopes=None,
@ -5259,6 +5581,81 @@ class TestMCPServerTimestamps:
assert persisted.credentials == {"scopes": ["s1"]}
assert "token_url" not in persisted.fields_set()
@pytest.mark.asyncio
async def test_persist_discovered_oauth_endpoints_writes_discovered_issuer_trust_on_first_use(self):
"""A server with no configured issuer records the discovered issuer trust-on-first-use, so the
next rebuild anchors discovery on it (RFC 8414 §3.3) instead of re-trusting the resource. When
an issuer is already set (admin-typed or a prior discovery), it is never overwritten."""
manager = MCPServerManager()
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
discovered_issuer="https://idp.example.com",
)
update_mcp_server_mock = AsyncMock()
with (
patch("litellm.proxy._experimental.mcp_server.db.update_mcp_server", new=update_mcp_server_mock),
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
):
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer=None,
existing_authorization_url=None,
existing_token_url=None,
existing_scopes=None,
metadata=metadata,
)
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer="https://admin-configured.example.com",
existing_authorization_url="https://admin-configured.example.com/authorize",
existing_token_url="https://admin-configured.example.com/token",
existing_scopes=["cfg"],
metadata=metadata,
)
assert update_mcp_server_mock.await_count == 1
persisted = update_mcp_server_mock.call_args.kwargs["data"]
assert persisted.issuer == "https://idp.example.com"
@pytest.mark.asyncio
async def test_persist_discovered_oauth_endpoints_does_not_persist_endpoints_for_issuer_anchored(self):
"""For an issuer-anchored server the endpoints are re-derived from the §3.3-validated issuer
document every build, so they must NOT be written into the endpoint columns: persisting them
would make the next build see populated endpoints and treat them as authoritative stored
values, defeating the issuer-only invariant. Only the resource-driven scopes are persisted."""
manager = MCPServerManager()
metadata = MCPOAuthMetadata(
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
scopes=["read"],
)
update_mcp_server_mock = AsyncMock()
with (
patch("litellm.proxy._experimental.mcp_server.db.update_mcp_server", new=update_mcp_server_mock),
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
):
await manager._persist_discovered_oauth_endpoints(
server_id="s",
auth_type=MCPAuth.oauth2,
existing_issuer="https://idp.example.com",
existing_authorization_url=None,
existing_token_url=None,
existing_scopes=None,
metadata=metadata,
is_issuer_anchored=True,
)
update_mcp_server_mock.assert_awaited_once()
persisted = update_mcp_server_mock.call_args.kwargs["data"]
assert "authorization_url" not in persisted.fields_set()
assert "token_url" not in persisted.fields_set()
assert persisted.credentials == {"scopes": ["read"]}
@pytest.mark.asyncio
async def test_build_mcp_server_from_table_skips_persistence_for_temporary_servers(self):
"""The session endpoint builds temporary servers whose server_id has no DB row; with
@ -5473,6 +5870,88 @@ class TestMCPServerTimestamps:
assert same_authorize.token_url == "https://idp.example.com/token"
assert same_authorize.registration_url == "https://idp.example.com/register"
def test_carry_forward_does_not_restore_endpoints_for_issuer_anchored_server(self):
"""When the server is issuer-anchored the endpoints come solely from the §3.3-validated issuer
document, so a failed issuer fetch (token_url None) must stay fail-closed. Carry-forward must
NOT resurrect the previous registry entry's token endpoint, or the very attacker-controlled
endpoint the issuer anchor distrusts would keep being served across rebuilds. Resource-driven
scopes still carry as last-known-good. Anchoring is keyed on the explicit issuer_is_anchored
flag, not on issuer truthiness, so a discovered issuer does not trip this fail-closed branch."""
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,
issuer="https://idp.example.com",
issuer_is_anchored=True,
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
registration_url="https://idp.example.com/register",
scopes=["read"],
)
failed_rebuild = MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
issuer="https://idp.example.com",
issuer_is_anchored=True,
)
_carry_forward_resolved_oauth_endpoints(new_server=failed_rebuild, previous_server=previous)
assert failed_rebuild.authorization_url is None
assert failed_rebuild.token_url is None
assert failed_rebuild.registration_url is None
assert failed_rebuild.scopes == ["read"]
def test_carry_forward_restores_endpoints_for_discovered_issuer_not_anchored(self):
"""A server that merely DISCOVERED its issuer trust-on-first-use is not anchored: issuer is set
for token identity but the endpoints are resource-rooted, so on a transient discovery blip they
must still carry forward as last-known-good, the same as any resource-rooted server. This is the
regression the explicit issuer_is_anchored flag prevents: keying fail-closed on issuer truthiness
alone would drop the working endpoints the moment the server learned its issuer."""
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,
issuer="https://idp.example.com",
issuer_is_anchored=False,
authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
registration_url="https://idp.example.com/register",
scopes=["read"],
)
blipped_rebuild = MCPServer(
server_id="s1",
name="s1",
url="https://up.example.com/mcp",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
issuer="https://idp.example.com",
issuer_is_anchored=False,
authorization_url=None,
)
_carry_forward_resolved_oauth_endpoints(new_server=blipped_rebuild, previous_server=previous)
assert blipped_rebuild.authorization_url == "https://idp.example.com/authorize"
assert blipped_rebuild.token_url == "https://idp.example.com/token"
assert blipped_rebuild.registration_url == "https://idp.example.com/register"
assert blipped_rebuild.scopes == ["read"]
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
@ -5487,6 +5966,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()

11
ui/litellm-dashboard/src/app/(dashboard)/mcp-servers/_components/OAuthFormFields.tsx
View file

@ -167,6 +167,17 @@ const OAuthFormFields: React.FC<OAuthFormFieldsProps> = ({
>
<Select mode="tags" tokenSeparators={[","]} placeholder="Add scopes" className="rounded-lg" size="large" />
</Form.Item>
<Form.Item
label={
<FieldLabel
label="Issuer (optional)"
tooltip="OAuth 2.0 authorization server issuer (RFC 8414). Auto-discovered from the upstream on first connect; set it explicitly to pin the trust anchor so token and scope discovery is fetched from and validated against this issuer (RFC 8414 §3.3) instead of anything the resource advertises."
/>
}
name="issuer"
>
<TextInput placeholder="https://issuer.example.com" className={fieldClassName} />
</Form.Item>
<Form.Item
label={
<FieldLabel

3
ui/litellm-dashboard/src/app/(dashboard)/mcp-servers/_components/create_mcp_server.tsx
View file

@ -210,6 +210,7 @@ const CreateMCPServer: React.FC<CreateMCPServerProps> = ({
credentials: isClientForwardedTokenMode(values.auth_type)
? preservedDeclaredAppCredentials(values.credentials)
: { ...((values.credentials as Record<string, unknown> | undefined) ?? {}), ...(dcrClientRef.current ?? {}) },
issuer: values.issuer,
authorization_url: values.authorization_url,
token_url: values.token_url,
registration_url: values.registration_url,
@ -759,7 +760,7 @@ const CreateMCPServer: React.FC<CreateMCPServerProps> = ({
if ("credentials" in changedValues) {
setAppMayNotMatchUpstream(false);
} else {
const upstreamChanged = ["url", "spec_path", "authorization_url", "token_url", "registration_url"].some(
const upstreamChanged = ["url", "spec_path", "issuer", "authorization_url", "token_url", "registration_url"].some(
(key) => key in changedValues,
);
const hasDeclaredApp = preservedDeclaredAppCredentials(form.getFieldValue("credentials")) !== undefined;

24
ui/litellm-dashboard/src/app/(dashboard)/mcp-servers/_components/mcp_server_edit.tsx
View file

@ -117,6 +117,7 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
const currentAuthType = Form.useWatch("auth_type", form);
const currentStaticHeaders = Form.useWatch("static_headers", form);
const currentCredentials = Form.useWatch("credentials", form);
const currentIssuer = Form.useWatch("issuer", form);
const currentAuthorizationUrl = Form.useWatch("authorization_url", form);
const currentTokenUrl = Form.useWatch("token_url", form);
const currentRegistrationUrl = Form.useWatch("registration_url", form);
@ -471,7 +472,7 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
if ("credentials" in changedValues) {
setAppMayNotMatchUpstream(false);
} else {
const upstreamChanged = ["url", "spec_path", "authorization_url", "token_url", "registration_url"].some(
const upstreamChanged = ["url", "spec_path", "issuer", "authorization_url", "token_url", "registration_url"].some(
(key) => key in changedValues,
);
const hasDeclaredApp = preservedDeclaredAppCredentials(form.getFieldValue("credentials")) !== undefined;
@ -517,6 +518,7 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
transport: rawTransport === TRANSPORT.OPENAPI ? TRANSPORT.HTTP : rawTransport,
auth_type: AUTH_TYPE.OAUTH2,
oauth2_flow: MCP_OAUTH2_FLOW_INTERACTIVE,
issuer: values.issuer,
authorization_url: values.authorization_url,
token_url: values.token_url,
registration_url: values.registration_url,
@ -636,6 +638,7 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
spec_path: undefined,
auth_type: undefined,
credentials: undefined,
issuer: undefined,
authorization_url: undefined,
token_url: undefined,
registration_url: undefined,
@ -845,7 +848,7 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
stdio_config: undefined,
env_json: undefined,
...(mcpServer.auth_type === AUTH_TYPE.OAUTH2 && restValues.auth_type !== AUTH_TYPE.OAUTH2
? { authorization_url: null, token_url: null, registration_url: null }
? { issuer: null, authorization_url: null, token_url: null, registration_url: null }
: {}),
...(mcpServer.auth_type === AUTH_TYPE.OAUTH2_TOKEN_EXCHANGE &&
restValues.auth_type !== AUTH_TYPE.OAUTH2_TOKEN_EXCHANGE
@ -1301,6 +1304,22 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
size="large"
/>
</Form.Item>
<Form.Item
label={
<span className="text-sm font-medium text-gray-700 flex items-center">
Issuer (optional)
<Tooltip title="OAuth 2.0 authorization server issuer (RFC 8414). Auto-discovered on first connect; set it explicitly to pin the trust anchor so token and scope discovery is fetched from and validated against this issuer (RFC 8414 §3.3) instead of anything the resource advertises.">
<InfoCircleOutlined className="ml-2 text-blue-400 hover:text-blue-600 cursor-help" />
</Tooltip>
</span>
}
name="issuer"
>
<Input
placeholder="https://issuer.example.com"
className="rounded-lg border-gray-300 focus:border-blue-500 focus:ring-blue-500"
/>
</Form.Item>
<Form.Item
label={
<span className="text-sm font-medium text-gray-700 flex items-center">
@ -1588,6 +1607,7 @@ const MCPServerEdit: React.FC<MCPServerEditProps> = ({
oauthFlowTypeValue ?? oauth2FlowToFormValue(mcpServer.oauth2_flow) ?? OAUTH_FLOW.INTERACTIVE,
static_headers: currentStaticHeaders ?? mcpServer.static_headers,
credentials: currentCredentials,
issuer: currentIssuer ?? mcpServer.issuer,
authorization_url: currentAuthorizationUrl ?? mcpServer.authorization_url,
token_url: currentTokenUrl ?? mcpServer.token_url,
registration_url: currentRegistrationUrl ?? mcpServer.registration_url,

2
ui/litellm-dashboard/src/components/mcp_tools/types.tsx
View file

@ -81,6 +81,7 @@ export const getOAuthAuthorizationIdentity = (values: Record<string, unknown>):
client_id: credentials.client_id ?? null,
client_secret: credentials.client_secret ?? null,
scopes: credentials.scopes ?? null,
issuer: values.issuer ?? null,
authorization_url: values.authorization_url ?? null,
token_url: values.token_url ?? null,
registration_url: values.registration_url ?? null,
@ -341,6 +342,7 @@ export interface MCPServer {
transport?: string | null;
auth_type?: string | null;
oauth2_flow?: string | null;
issuer?: string | null;
authorization_url?: string | null;
token_url?: string | null;
registration_url?: string | null;