fix(mcp): enforce the issuer trust anchor at every endpoint adoption site

When an admin pins an issuer, RFC 8414 section 3.3 makes that issuer the sole
authoritative source of the authorization and token endpoints, so a compromised
or misconfigured upstream cannot smuggle a token endpoint by echoing the pinned
authorize URL. The first cut enforced that only on the database build path; the
carry-forward, persistence, config-load, serialization and sanitization paths
could still restore or emit upstream-derived endpoints for an issuer-anchored
server, which is the class of gap the review flagged.

Every site now routes through one predicate. _endpoints_yield_to_issuer returns
all-None whenever the issuer is the anchor, so both build paths,
has_all_upstream_oauth_fields, needs_discovery and the endpoint merge defer to
the issuer. _carry_forward_resolved_oauth_endpoints carries only scopes for an
issuer-anchored server and fails closed on endpoints.
_persist_discovered_oauth_endpoints skips endpoint writes under the anchor. The
two table serializers round-trip the issuer and both non-admin sanitizers redact
it. Scope selection stays resource-driven per the MCP authorization spec:
_fetch_issuer_anchored_oauth_metadata takes endpoints from the issuer document
and scopes from the resource document.

The OAuth metadata resolution and corroboration gating for the database build
path move into _resolve_table_oauth_metadata so build_mcp_server_from_table
stays within the cyclomatic-complexity budget without changing behavior.

Regression tests pin the invariant at each site: the issuer overrides stored
endpoints even when they are populated, carry-forward does not restore endpoints
under the anchor, persistence does not write endpoints under the anchor, a url or
auth_type change clears stale issuer-scoped fields even when resubmitted
unchanged, the Azure heuristic stays reachable under a required issuer, and
anchored metadata takes endpoints from the issuer while scopes come from the
resource

litellm/proxy/_experimental/mcp_server/db.py

@@ -722,7 +722,17 @@ async def update_mcp_server(
         data_dict["credentials"] = None
 
     if auth_type_changed or url_changed:
-        data_dict.update({field: None for field in _AUTH_FLOW_SCOPED_FIELDS if field not in data_dict})
+        # 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

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -201,6 +201,26 @@ def _blank_to_none(value: str | None) -> str | None:
     return value.strip() or None
 
 
+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
@@ -271,11 +291,23 @@ 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 an ``issuer`` is configured the endpoints must 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. Scopes stay resource-driven and can still carry.
     """
     if previous_server is None:
         return
     if previous_server.url != new_server.url or previous_server.auth_type != new_server.auth_type:
         return
+    if _blank_to_none(new_server.issuer):
+        # 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
     )
@@ -1154,6 +1186,13 @@ class MCPServerManager:
             manual_registration_url = _blank_to_none(server_config.get("registration_url"))
             is_discovery_auth_type = auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
             use_issuer_anchor = manual_issuer is not None and is_discovery_auth_type
+            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(
@@ -1506,6 +1545,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,
@@ -1595,46 +1680,24 @@ class MCPServerManager:
         manual_registration_url = _blank_to_none(mcp_server.registration_url)
         is_discovery_auth_type = auth_type in _UPSTREAM_OAUTH_DISCOVERY_AUTH_TYPES
         use_issuer_anchor = manual_issuer is not None and is_discovery_auth_type
-        has_all_upstream_oauth_fields = bool(manual_authorization_url and manual_token_url and scopes)
-        needs_discovery = bool(server_url) and (
-            (is_discovery_auth_type 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,
-            )
+        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
+        )
+        token_exchange_endpoint = mcp_server.token_exchange_endpoint or (
+            credentials_dict.get("token_exchange_endpoint") if credentials_dict else None
+        )
+        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,
         )
-        if not needs_discovery:
-            mcp_oauth_metadata = None
-        elif manual_issuer is not None and is_discovery_auth_type:
-            mcp_oauth_metadata = await self._fetch_issuer_anchored_oauth_metadata(
-                manual_issuer,
-                server_url,  # type: ignore[arg-type]
-            )
-        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:
-            gated_oauth_metadata = mcp_oauth_metadata
-        elif is_discovery_auth_type:
-            gated_oauth_metadata = _restrict_discovery_to_corroborated_authorization_server(
-                mcp_oauth_metadata,
-                manual_authorization_url,
-                mcp_server.server_id,
-                bool(getattr(mcp_server, "dcr_bridge", None)),
-            )
-        else:
-            gated_oauth_metadata = mcp_oauth_metadata
 
         resolved_scopes = scopes or (gated_oauth_metadata.scopes if gated_oauth_metadata else None)
 
@@ -1721,6 +1784,7 @@ class MCPServerManager:
                 existing_token_url=manual_token_url,
                 existing_scopes=scopes,
                 metadata=gated_oauth_metadata,
+                is_issuer_anchored=use_issuer_anchor,
             )
         return new_server
 
@@ -1769,6 +1833,7 @@ class MCPServerManager:
         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.
 
@@ -1782,6 +1847,12 @@ 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
@@ -1792,10 +1863,14 @@ class MCPServerManager:
         )
         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] = {
             **issuer_update,
@@ -3375,7 +3450,9 @@ class MCPServerManager:
                 return metadata
         return None
 
-    async def _fetch_issuer_anchored_oauth_metadata(self, issuer: str, server_url: str) -> Optional[MCPOAuthMetadata]:
+    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
@@ -3400,7 +3477,9 @@ class MCPServerManager:
                 issuer,
             )
             return None
-        resource_metadata = await self._descovery_metadata(server_url, allow_origin_fallback=False)
+        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})
 
@@ -3488,8 +3567,6 @@ class MCPServerManager:
             ):
                 return metadata
 
-        if require_issuer is not None:
-            return None
         return self._build_azure_authorization_server_metadata(parsed)
 
     @staticmethod
@@ -5198,6 +5275,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,
@@ -5307,6 +5385,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,

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -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,

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_partial_update.py

@@ -239,6 +239,37 @@ async def test_url_change_clears_stale_discovered_oauth_fields():
         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_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

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -1338,6 +1338,50 @@ class TestMCPServerManager:
         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",
@@ -2669,6 +2713,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()
@@ -5484,6 +5559,41 @@ class TestMCPServerTimestamps:
         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
@@ -5698,6 +5808,44 @@ 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 an issuer is configured 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."""
+        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",
+            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",
+        )
+
+        _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_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