fix(mcp): discover missing OAuth scopes and token_url when authorization_url is set manually

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -186,33 +186,117 @@ def validate_tool_name(name: str) -> _ToolNameValidationResult:  # type: ignore[
 )
 
 
+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 trusted_authorization_url is None:
+        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.
 
     A rebuild wholesale-replaces the registry entry, so without this a transient upstream outage
     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`` 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 token endpoint.
     """
     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 = _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_endpoints 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_endpoints 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:
         new_server.scopes = previous_server.scopes
 
 
+def _gate_discovered_endpoints_against_manual_authorization_url(
+    metadata: MCPOAuthMetadata | None,
+    manual_authorization_url: str | None,
+    server_identifier: str,
+    is_dcr_bridge: bool,
+) -> MCPOAuthMetadata | None:
+    """Apply :func:`_endpoints_corroborate_authorization_url` to freshly discovered metadata, the
+    other non-manual endpoint source. Drops the discovered ``token_url``/``registration_url`` (never
+    the scopes, which carry no credentials) when they cannot be vouched for by the pinned authorize
+    endpoint, logging why so an intentional mismatch can be resolved by setting Token URL by hand.
+    """
+    if metadata is None or (not metadata.token_url and not metadata.registration_url):
+        return metadata
+    if _endpoints_corroborate_authorization_url(metadata.authorization_url, manual_authorization_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 go to endpoints vouched for by 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) if manual_authorization_url else "<absent>",
+        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."""
@@ -1041,20 +1125,31 @@ async def load_servers_from_config(
             else:
                 mcp_oauth_metadata = None
 
+            gated_oauth_metadata = (
+                _gate_discovered_endpoints_against_manual_authorization_url(
+                    mcp_oauth_metadata,
+                    server_config.get("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
+                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
+                gated_oauth_metadata.token_url if gated_oauth_metadata else None
             )
             resolved_registration_url = server_config.get("registration_url") or (
-                mcp_oauth_metadata.registration_url if mcp_oauth_metadata else None
+                gated_oauth_metadata.registration_url if gated_oauth_metadata else None
             )
 
             config_oauth2_flow = server_config.get("oauth2_flow", None)
@@ -1447,8 +1542,9 @@ async def build_mcp_server_from_table(
 
         auth_type = cast(MCPAuthType, mcp_server.auth_type)
         server_url = mcp_server.url
+        has_all_upstream_oauth_fields = bool(mcp_server.authorization_url and mcp_server.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
@@ -1467,12 +1563,22 @@ async def build_mcp_server_from_table(
         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 = (
+            _gate_discovered_endpoints_against_manual_authorization_url(
+                mcp_oauth_metadata,
+                mcp_server.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 +1598,9 @@ async def build_mcp_server_from_table(
             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=mcp_server.authorization_url or getattr(gated_oauth_metadata, "authorization_url", None),
+            token_url=mcp_server.token_url or getattr(gated_oauth_metadata, "token_url", None),
+            registration_url=mcp_server.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
             ),
@@ -1554,7 +1660,7 @@ async def build_mcp_server_from_table(
                 existing_authorization_url=mcp_server.authorization_url,
                 existing_token_url=mcp_server.token_url,
                 existing_scopes=scopes,
-                metadata=mcp_oauth_metadata,
+                metadata=gated_oauth_metadata,
             )
         return new_server