diff --git a/tests/integration/mcp/test_mcp_management.py b/tests/integration/mcp/test_mcp_management.py index bde18840d7d..4dfadbbce35 100644 --- a/tests/integration/mcp/test_mcp_management.py +++ b/tests/integration/mcp/test_mcp_management.py @@ -2,6 +2,7 @@ import uuid from pathlib import Path from typing import Final +import pytest import yaml from integration._support.client import Gateway, eventually from integration._support.mcp import ( @@ -16,9 +17,17 @@ from integration._support.mcp import ( ) from integration._support.process import owned_proxy +from litellm.models.user import LiteLLM_UserTable +from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken + ADD: Final = {"a": 4, "b": 5} +def _dashboard_ui_session_token(user_id: str) -> str: + user: Final = LiteLLM_UserTable(user_id=user_id, user_role="internal_user", models=[]) + return ExperimentalUIJWTToken.get_experimental_ui_login_jwt_auth_token(user) + + def _servers(gateway: Gateway, key: str | None = None) -> dict[str, dict[str, object]]: response: Final = gateway.client.get("/v1/mcp/server", headers={"x-litellm-api-key": key or gateway.key}) assert response.status_code == 200, response.text @@ -285,3 +294,57 @@ def test_config_declared_server_behaves_like_database_server_but_is_read_only(ga assert declared_id in _servers(candidate) assert call_tool(candidate, key, declared_id, declared_names["add"], ADD).status_code == 200 assert len(tool_calls(declared_peer.drain())) == 1 and tool_calls(database_peer.drain()) == () + + +@pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(AssertionError, match="Team-granted server detail access should succeed"), + reason="LIT-3974 A: team-granted detail access", +) +def test_team_granted_database_server_detail_is_available_to_team_key(gateway: Gateway) -> None: + with mcp_peer() as peer, gateway.scenario() as scenario: + alias: Final = "lit3974_team_" + uuid.uuid4().hex[:8] + server_id: Final = register_mcp(scenario, peer, alias) + team_id: Final = scenario.team(object_permission={"mcp_servers": [server_id]}) + key: Final = scenario.key(team_id=team_id) + + response: Final = gateway.request("GET", f"/v1/mcp/server/{server_id}", key=key) + + assert response.status_code == 200, f"Team-granted server detail access should succeed: {response.text}" + assert response.json()["server_id"] == server_id, response.text + assert response.json()["alias"] == alias, response.text + + +@pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(AssertionError, match="Team-granted server detail access should succeed"), + reason="LIT-3974 A: team-granted detail access", +) +def test_ui_session_lists_and_fetches_team_granted_config_server( + gateway: Gateway, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LITELLM_SALT_KEY", "sk-integration-salt") + with mcp_peer() as peer, gateway.scenario() as scenario: + alias: Final = "lit3974_config_" + uuid.uuid4().hex[:8] + server_id: Final = "lit3974-" + uuid.uuid4().hex[:12] + team_id: Final = scenario.team(object_permission={"mcp_servers": [server_id]}) + user_id: Final = scenario.user(user_role="internal_user", teams=[team_id]) + config: Final = yaml.safe_load((Path(__file__).resolve().parents[1] / "proxy_config.yaml").read_text()) + config["mcp_servers"] = {alias: {**peer.registration(), "alias": alias, "server_id": server_id}} + config_path: Final = tmp_path / "lit3974-mcp.yaml" + config_path.write_text(yaml.safe_dump(config)) + + with owned_proxy(gateway, tmp_path, {}, config=config_path) as candidate: + token: Final = _dashboard_ui_session_token(user_id) + headers: Final = {"Authorization": f"Bearer {token}"} + listed: Final = candidate.client.get("/v1/mcp/server", headers=headers) + assert listed.status_code == 200, listed.text + assert [server["server_id"] for server in listed.json()] == [server_id], listed.text + + detail: Final = candidate.client.get(f"/v1/mcp/server/{server_id}", headers=headers) + + assert detail.status_code == 200, f"Team-granted server detail access should succeed: {detail.text}" + assert detail.json()["server_id"] == server_id, detail.text + assert detail.json()["alias"] == alias, detail.text diff --git a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py index 557e753a76f..0b81e6c9080 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py @@ -1,26 +1,35 @@ +import asyncio import os import sys import types import json import logging -from contextlib import ExitStack +from collections.abc import Iterator, Mapping +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass, field from datetime import datetime, timedelta from types import SimpleNamespace from typing import Final, List, Optional, cast from unittest.mock import AsyncMock, MagicMock, patch import pytest +from pydantic import BaseModel from respx import MockRouter from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient from litellm._uuid import uuid +from litellm.constants import UI_SESSION_TOKEN_TEAM_ID +from litellm.models.access_group import LiteLLM_AccessGroupTable +from litellm.models.organization import LiteLLM_OrganizationTable +from litellm.models.team import LiteLLM_TeamTable +from litellm.models.user import LiteLLM_UserTable from litellm.proxy.management_endpoints import ( mcp_management_endpoints as mgmt_endpoints, ) - from litellm.proxy._types import ( + LiteLLM_ObjectPermissionTable, LiteLLM_MCPServerTable, LitellmUserRoles, MCPTransport, @@ -29,6 +38,7 @@ from litellm.proxy._types import ( UpdateMCPServerRequest, UserAPIKeyAuth, ) +from litellm.proxy._experimental.mcp_server.mcp_server_manager import MCPServerConfig, MCPServerManager from litellm.types.mcp import MCPAuth, MCPCredentials from litellm.types.mcp_server.mcp_server_manager import MCPServer @@ -1342,6 +1352,7 @@ class TestListMCPServers: mock_manager = MagicMock() mock_manager.add_server = AsyncMock() + mock_manager.get_allowed_mcp_servers = AsyncMock(return_value=["env-server"]) mock_manager.health_check_server = AsyncMock(return_value=mock_health_result) mock_user_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.INTERNAL_USER) @@ -1356,7 +1367,11 @@ class TestListMCPServers: mock_manager, ), patch( - "litellm.proxy.management_endpoints.mcp_management_endpoints.get_all_mcp_servers_for_user", + "litellm.proxy._experimental.mcp_server.db.get_mcp_servers_by_verificationtoken", + AsyncMock(return_value=["env-server"]), + ), + patch( + "litellm.proxy._experimental.mcp_server.db.get_mcp_servers", AsyncMock(return_value=[generate_mock_mcp_server_db_record(server_id="env-server")]), ), patch( @@ -2299,9 +2314,10 @@ class TestTemporaryMCPSessionEndpoints: mock_manager, ), patch( - "litellm.proxy.management_endpoints.mcp_management_endpoints.build_effective_auth_contexts", + "litellm.proxy._experimental.mcp_server.ui_session_utils.build_effective_auth_contexts", AsyncMock(return_value=[non_admin]), - ), + ) as effective_contexts, + patch.object(mgmt_endpoints, "build_effective_auth_contexts", effective_contexts), ): with pytest.raises(HTTPException) as exc_info: await _get_cached_temporary_mcp_server_or_404("server-x", non_admin) @@ -2323,6 +2339,7 @@ class TestTemporaryMCPSessionEndpoints: mock_manager = MagicMock() mock_manager.get_mcp_server_by_id.return_value = registry_server mock_manager.get_mcp_server_by_name.return_value = None + mock_manager._build_mcp_server_table.return_value = generate_mock_mcp_server_db_record(server_id="server-x") mock_manager.get_allowed_mcp_servers = AsyncMock(return_value=["server-x"]) with ( @@ -2368,6 +2385,7 @@ class TestTemporaryMCPSessionEndpoints: mock_manager = MagicMock() mock_manager.get_mcp_server_by_id.return_value = registry_server mock_manager.get_mcp_server_by_name.return_value = None + mock_manager._build_mcp_server_table.return_value = generate_mock_mcp_server_db_record(server_id="server-x") def allowed_for(auth): return ["server-x"] if auth.team_id == "team-with-mcp-grant" else [] @@ -2384,9 +2402,10 @@ class TestTemporaryMCPSessionEndpoints: mock_manager, ), patch( - "litellm.proxy.management_endpoints.mcp_management_endpoints.build_effective_auth_contexts", + "litellm.proxy._experimental.mcp_server.ui_session_utils.build_effective_auth_contexts", AsyncMock(return_value=[ui_session_auth, team_context]), - ), + ) as effective_contexts, + patch.object(mgmt_endpoints, "build_effective_auth_contexts", effective_contexts), ): result = await _get_cached_temporary_mcp_server_or_404("server-x", ui_session_auth) @@ -5597,7 +5616,7 @@ async def test_list_mcp_user_credentials_batch_server_fetch(): ), ): result = await list_mcp_user_credentials( - user_api_key_dict=_make_user_auth(user_id), + user_api_key_dict=generate_mock_user_api_key_auth(user_role=LitellmUserRoles.PROXY_ADMIN, user_id=user_id), ) batch_mock.assert_called_once() @@ -7903,10 +7922,13 @@ class TestGetMcpToolsWireShape: @pytest.mark.asyncio -@pytest.mark.parametrize("role,expected_status", [ - (LitellmUserRoles.PROXY_ADMIN, 404), - (LitellmUserRoles.INTERNAL_USER, 403), -]) +@pytest.mark.parametrize( + "role,expected_status", + [ + (LitellmUserRoles.PROXY_ADMIN, 404), + (LitellmUserRoles.INTERNAL_USER, 403), + ], +) async def test_config_server_edit_preserves_api_contract_without_creating_rows(role, expected_status): from litellm.proxy._experimental.mcp_server.mcp_server_manager import MCPServerManager @@ -7929,9 +7951,7 @@ async def test_config_server_edit_preserves_api_contract_without_creating_rows(r assert exc.value.status_code == expected_status if role == LitellmUserRoles.PROXY_ADMIN: - assert exc.value.detail == { - "error": f"MCP Server not found, passed server_id={server.server_id}" - } + assert exc.value.detail == {"error": f"MCP Server not found, passed server_id={server.server_id}"} prisma.db.litellm_mcpservertable.update.assert_awaited_once() else: prisma.db.litellm_mcpservertable.update.assert_not_awaited() @@ -8166,3 +8186,2396 @@ class TestDuplicateIdentifierRejection: assert [entry.name for entry in result.skipped] == ["fresh"] assert "fresh" in result.skipped[0].reason assert result.imported == () + + +@dataclass(frozen=True) +class _ResolutionEffects: + byok_store: AsyncMock = field(default_factory=AsyncMock) + oauth_store: AsyncMock = field(default_factory=AsyncMock) + env_merge: AsyncMock = field(default_factory=lambda: AsyncMock(return_value={"LIT3974_TOKEN": "lit3974-secret"})) + env_delete: AsyncMock = field(default_factory=AsyncMock) + byok_invalidate: AsyncMock = field(default_factory=AsyncMock) + oauth_invalidate: AsyncMock = field(default_factory=AsyncMock) + env_invalidate: MagicMock = field(default_factory=MagicMock) + + @contextmanager + def patch(self, manager: MCPServerManager) -> Iterator[None]: + with ( + patch.object(mgmt_endpoints, "store_user_credential", self.byok_store), + patch.object(mgmt_endpoints, "store_user_oauth_credential", self.oauth_store), + patch.object(mgmt_endpoints, "merge_user_env_vars", self.env_merge), + patch.object(mgmt_endpoints, "delete_user_env_vars", self.env_delete), + patch.object(manager, "invalidate_user_oauth_token_cache", self.oauth_invalidate), + patch("litellm.proxy._experimental.mcp_server.server._invalidate_byok_cred_cache", self.byok_invalidate), + patch( + "litellm.proxy._experimental.mcp_server.mcp_server_manager.invalidate_user_env_vars_cache", + self.env_invalidate, + ), + ): + yield + + def assert_no_writes(self) -> None: + self.byok_store.assert_not_awaited() + self.oauth_store.assert_not_awaited() + self.env_merge.assert_not_awaited() + self.env_delete.assert_not_awaited() + self.byok_invalidate.assert_not_awaited() + self.oauth_invalidate.assert_not_awaited() + self.env_invalidate.assert_not_called() + + +def _mock_mcp_resolution_prisma_client( + server: LiteLLM_MCPServerTable, + key_permission: LiteLLM_ObjectPermissionTable, + team: LiteLLM_TeamTable, + user: LiteLLM_UserTable | None = None, + organization: LiteLLM_OrganizationTable | None = None, + access_group: LiteLLM_AccessGroupTable | None = None, + object_permission: LiteLLM_ObjectPermissionTable | None = None, +) -> MagicMock: + prisma: Final = MagicMock() + prisma.db.litellm_verificationtoken.find_unique = AsyncMock( + return_value=SimpleNamespace(object_permission=key_permission) + ) + + def matches_server_filter(name: str, condition: object) -> bool: + if name == "submitted_by": + return server.submitted_by == condition + if name == "server_id": + if isinstance(condition, str): + return server.server_id == condition + if isinstance(condition, Mapping) and set(condition) == {"in"}: + return server.server_id in condition["in"] + if name == "mcp_access_groups" and isinstance(condition, Mapping) and set(condition) == {"hasSome"}: + return bool(set(server.mcp_access_groups).intersection(condition["hasSome"])) + raise AssertionError(f"Unsupported MCP fixture filter: {name}={condition!r}") + + def find_many_side_effect(**kwargs: object) -> list[LiteLLM_MCPServerTable]: + where: Final = kwargs.get("where", {}) + assert isinstance(where, Mapping) + return [server] if all(matches_server_filter(name, condition) for name, condition in where.items()) else [] + + def unique_lookup(row: BaseModel | None, identity: str) -> AsyncMock: + def find_unique(**kwargs: object) -> BaseModel | None: + return row if row is not None and kwargs.get("where") == {identity: getattr(row, identity)} else None + + return AsyncMock(side_effect=find_unique) + + prisma.db.litellm_mcpservertable.find_many = AsyncMock(side_effect=find_many_side_effect) + prisma.db.litellm_mcpservertable.find_unique = unique_lookup(server, "server_id") + prisma.db.litellm_teamtable.find_unique = unique_lookup(team, "team_id") + prisma.db.litellm_usertable.find_unique = unique_lookup(user, "user_id") + prisma.db.litellm_organizationtable.find_unique = unique_lookup(organization, "organization_id") + prisma.db.litellm_accessgrouptable.find_unique = unique_lookup(access_group, "access_group_id") + prisma.db.litellm_objectpermissiontable.find_unique = unique_lookup(object_permission, "object_permission_id") + return prisma + + +def _mock_mcp_resolution_cache() -> MagicMock: + cache: Final = MagicMock() + cache.async_get_cache = AsyncMock(return_value=None) + cache.async_set_cache = AsyncMock() + return cache + + +class TestMCPServerResolutionRegressions: + @pytest.mark.asyncio + @pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, check=lambda error: error.status_code == 403 and "permission" in str(error.detail) + ), + reason="LIT-3974 change A: detail authorization includes a server granted to the caller's team", + ) + async def test_team_granted_database_server_is_visible_to_virtual_key(self) -> None: + server_id: Final = "lit3974-team-db" + team_id: Final = "lit3974-team" + server: Final = generate_mock_mcp_server_db_record( + server_id=server_id, + alias="Team server", + url="http://127.0.0.1:1/mcp", + ) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-key-permission", + mcp_servers=[], + ) + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-team-permission", + mcp_servers=[server_id], + ), + ) + prisma: Final = _mock_mcp_resolution_prisma_client(server, key_permission, team) + manager: Final = MCPServerManager() + auth: Final = UserAPIKeyAuth( + api_key="lit3974-key", + user_id="lit3974-user", + team_id=team_id, + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=key_permission, + ) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=server)), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + try: + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + except HTTPException as exc: + logging.warning("db_runtime/team_grant: HTTP %s detail=%r", exc.status_code, exc.detail) + raise + + assert result.server_id == server_id, "team-granted DB server detail must resolve for the team's key" + assert result.alias == "Team server", "detail must identify the granted DB server" + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "case_name,key_server_ids,team_server_ids,org_server_ids", + [ + ("key-team-intersection", ["lit3974-target"], ["lit3974-other"], None), + ("key-opt-out", ["no-mcp-servers", "lit3974-target"], ["lit3974-target"], None), + ("org-ceiling", ["lit3974-target"], ["lit3974-target"], ["lit3974-other"]), + ], + ) + @pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(pytest.fail.Exception, match="DID NOT RAISE"), + reason="LIT-3974 change A: detail authorization enforces key, team, and organization ceilings", + ) + async def test_database_server_detail_obeys_authz_intersection( + self, + case_name: str, + key_server_ids: list[str], + team_server_ids: list[str], + org_server_ids: list[str] | None, + ) -> None: + server_id: Final = "lit3974-target" + team_id: Final = "lit3974-team" + organization_id: Final = "lit3974-organization" if org_server_ids is not None else None + server: Final = generate_mock_mcp_server_db_record( + server_id=server_id, + alias="Target server", + url="http://127.0.0.1:1/mcp", + ) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974-key-permission-{case_name}", + mcp_servers=key_server_ids, + ) + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974-team-permission-{case_name}", + mcp_servers=team_server_ids, + ), + organization_id=organization_id, + ) + organization: Final = ( + LiteLLM_OrganizationTable( + organization_id=organization_id, + organization_alias="LIT-3974", + budget_id="lit3974-budget", + created_by="lit3974-test", + updated_by="lit3974-test", + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974-org-permission-{case_name}", + mcp_servers=org_server_ids, + ), + object_permission_id=f"lit3974-org-permission-{case_name}", + ) + if org_server_ids is not None + else None + ) + prisma: Final = _mock_mcp_resolution_prisma_client(server, key_permission, team, organization=organization) + manager: Final = MCPServerManager() + health_check: Final = AsyncMock() + add_server: Final = AsyncMock() + auth: Final = UserAPIKeyAuth( + api_key=f"lit3974-key-{case_name}", + user_id="lit3974-user", + team_id=team_id, + org_id=organization_id, + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=key_permission, + ) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=server)), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "health_check_server", health_check), + patch.object(manager, "add_server", add_server), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + with pytest.raises(HTTPException) as exc_info: + await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + + assert exc_info.value.status_code == 403, f"{case_name}: narrowed detail access must return 403" + assert exc_info.value.detail == { + "error": ( + f"User does not have permission to view mcp server with id {server_id}. " + "You can only view mcp servers that you have access to." + ) + }, f"{case_name}: authorization denial body" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "case_name,key_server_ids,team_server_ids,org_server_ids", + [ + pytest.param( + "key-team-intersection-control", + ["lit3974-target"], + ["lit3974-target"], + None, + id="key-team-intersection-control", + ), + pytest.param( + "key-opt-out-control", + ["lit3974-target"], + ["lit3974-target"], + None, + id="key-opt-out-control", + ), + pytest.param( + "org-ceiling-control", + ["lit3974-target"], + ["lit3974-target"], + ["lit3974-target"], + id="org-ceiling-control", + ), + ], + ) + async def test_database_server_detail_intersection_controls( + self, + case_name: str, + key_server_ids: list[str], + team_server_ids: list[str], + org_server_ids: list[str] | None, + ) -> None: + server_id: Final = "lit3974-target" + team_id: Final = "lit3974-team" + organization_id: Final = "lit3974-organization" if org_server_ids is not None else None + server: Final = generate_mock_mcp_server_db_record( + server_id=server_id, + alias="Target server", + url="http://127.0.0.1:1/mcp", + ) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974-key-permission-{case_name}", + mcp_servers=key_server_ids, + ) + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974-team-permission-{case_name}", + mcp_servers=team_server_ids, + ), + organization_id=organization_id, + ) + organization: Final = ( + LiteLLM_OrganizationTable( + organization_id=organization_id, + organization_alias="LIT-3974", + budget_id="lit3974-budget", + created_by="lit3974-test", + updated_by="lit3974-test", + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974-org-permission-{case_name}", + mcp_servers=org_server_ids, + ), + ) + if org_server_ids is not None + else None + ) + prisma: Final = _mock_mcp_resolution_prisma_client(server, key_permission, team, organization=organization) + manager: Final = MCPServerManager() + health_check: Final = AsyncMock() + add_server: Final = AsyncMock() + auth: Final = UserAPIKeyAuth( + api_key=f"lit3974-key-{case_name}", + user_id="lit3974-user", + team_id=team_id, + org_id=organization_id, + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=key_permission, + ) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=server)), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "health_check_server", health_check), + patch.object(manager, "add_server", add_server), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + + assert result.server_id == server_id + assert result.alias == "Target server" + + @pytest.mark.asyncio + @pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, check=lambda error: error.status_code == 403 and "permission" in str(error.detail) + ), + reason="LIT-3974 change A: dashboard detail authorization resolves team grants for config servers", + ) + async def test_ui_session_team_grant_resolves_config_server_detail(self) -> None: + server_id: Final = "lit3974-config-server" + team_id: Final = "lit3974-ui-team" + user_id: Final = "lit3974-ui-user" + server: Final = generate_mock_mcp_server_db_record(server_id=server_id) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-ui-key-permission", + mcp_servers=[], + ) + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-ui-team-permission", + mcp_servers=[server_id], + ), + ) + user: Final = LiteLLM_UserTable( + user_id=user_id, + teams=[team_id], + user_role=LitellmUserRoles.INTERNAL_USER, + ) + prisma: Final = _mock_mcp_resolution_prisma_client(server, key_permission, team, user=user) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + await manager.load_servers_from_config( + { + "config_server": { + "server_id": server_id, + "alias": "Config_server", + "url": "https://config.example.com/mcp", + "transport": "http", + "auth_type": MCPAuth.oauth2, + "oauth2_flow": "authorization_code", + "authorization_url": "https://oauth.example.com/authorize", + "token_url": "https://oauth.example.com/token", + } + } + ) + auth: Final = UserAPIKeyAuth( + user_id=user_id, + team_id=UI_SESSION_TOKEN_TEAM_ID, + user_role=LitellmUserRoles.INTERNAL_USER, + ) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=None)), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + try: + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + except HTTPException as exc: + logging.warning("config/ui_session_team_grant: HTTP %s detail=%r", exc.status_code, exc.detail) + raise + + assert result.server_id == server_id, "UI session team grant must resolve the config server" + assert result.alias == "Config_server", "config detail must retain its display alias" + + @pytest.mark.asyncio + @pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(pytest.fail.Exception, match="DID NOT RAISE"), + reason="LIT-3974 change B: creation rejects an identifier already owned by a config server", + ) + async def test_create_rejects_config_server_identifier_collision(self) -> None: + server_id: Final = "lit3974-config-collision" + prisma: Final = _mock_mcp_resolution_prisma_client( + generate_mock_mcp_server_db_record(server_id=server_id), + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974-key", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974-team"), + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + await manager.load_servers_from_config( + { + "config_server": { + "server_id": server_id, + "alias": "config_server", + "url": "http://127.0.0.1:1/mcp", + "transport": "http", + "auth_type": MCPAuth.oauth2, + "oauth2_flow": "authorization_code", + "authorization_url": "https://oauth.example.com/authorize", + "token_url": "https://oauth.example.com/token", + } + } + ) + payload: Final = NewMCPServerRequest( + server_id=server_id, + alias="duplicate", + url="https://new.example.com/mcp", + transport=MCPTransport.http, + ) + created: Final = generate_mock_mcp_server_db_record(server_id=server_id, alias="duplicate") + create_server: Final = AsyncMock(return_value=created) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=None)), + patch.object(mgmt_endpoints, "create_mcp_server_if_identifier_free", create_server), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + ): + with pytest.raises(HTTPException) as exc_info: + await mgmt_endpoints.add_mcp_server( + payload=payload, + user_api_key_dict=generate_mock_user_api_key_auth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="lit3974-admin", + ), + ) + + assert exc_info.value.status_code == 400, "config-server identifier collision must be a client error" + assert exc_info.value.detail == { + "error": f"MCP Server with id {server_id} already exists. Cannot create another." + }, "config-server collision response body" + create_server.assert_not_awaited() + + @pytest.mark.asyncio + async def test_alias_lookup_authorizes_the_resolved_canonical_server_id(self) -> None: + allowed_id: Final = "lit3974-allowed-config" + denied_id: Final = "lit3974-denied-config" + prisma: Final = _mock_mcp_resolution_prisma_client( + generate_mock_mcp_server_db_record(server_id=denied_id), + LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-alias-permission", + mcp_servers=[allowed_id], + ), + LiteLLM_TeamTable(team_id="lit3974-alias-team"), + ) + manager: Final = MCPServerManager() + await manager.load_servers_from_config( + { + "allowed_server": { + "server_id": allowed_id, + "alias": "allowed_alias", + "url": "https://allowed.example.com/mcp", + "transport": "http", + "auth_type": MCPAuth.oauth2, + "oauth2_flow": "authorization_code", + "authorization_url": "https://oauth.example.com/authorize", + "token_url": "https://oauth.example.com/token", + }, + "denied_server": { + "server_id": denied_id, + "alias": "denied_alias", + "url": "https://denied.example.com/mcp", + "transport": "http", + "auth_type": MCPAuth.oauth2, + "oauth2_flow": "authorization_code", + "authorization_url": "https://oauth.example.com/authorize", + "token_url": "https://oauth.example.com/token", + }, + } + ) + auth: Final = UserAPIKeyAuth( + api_key="lit3974-alias-key", + user_id="lit3974-alias-user", + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-alias-permission", + mcp_servers=[allowed_id], + ), + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=None)), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + with pytest.raises(HTTPException) as exc_info: + await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id="denied_alias", + user_api_key_dict=auth, + ) + + assert exc_info.value.status_code == 403, "alias resolution must not widen canonical-id authorization" + assert exc_info.value.detail == { + "error": ( + "User does not have permission to view mcp server with id denied_alias. " + "You can only view mcp servers that you have access to." + ) + }, "alias denial response body" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + + @pytest.mark.asyncio + async def test_alias_lookup_allows_when_canonical_id_is_granted(self) -> None: + server_id: Final = "lit3974-granted-alias-config" + prisma: Final = _mock_mcp_resolution_prisma_client( + generate_mock_mcp_server_db_record(server_id=server_id), + LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-granted-alias-permission", + mcp_servers=[server_id], + ), + LiteLLM_TeamTable(team_id="lit3974-granted-alias-team"), + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + existing_tasks: Final = asyncio.all_tasks() + with MockRouter(assert_all_called=False) as httpx_mock: + await manager.load_servers_from_config( + { + "granted_alias_server": { + "server_id": server_id, + "alias": "granted_alias", + "url": "https://granted.example.com/mcp", + "transport": "http", + } + } + ) + startup_tasks: Final = tuple(task for task in asyncio.all_tasks() if task not in existing_tasks) + for task in startup_tasks: + task.cancel() + await asyncio.gather(*startup_tasks, return_exceptions=True) + assert httpx_mock.calls.call_count == 0 + auth: Final = UserAPIKeyAuth( + api_key="lit3974-granted-alias-key", + user_id="lit3974-granted-alias-user", + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974-granted-alias-permission", + mcp_servers=[server_id], + ), + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock(return_value=generate_mock_mcp_server_db_record(server_id=server_id)) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "get_mcp_server", AsyncMock(return_value=None)), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id="granted_alias", + user_api_key_dict=auth, + ) + + assert result.server_id == server_id + assert result.alias == "granted_alias" + add_server.assert_not_awaited() + health_check.assert_awaited_once() + + +class TestMCPServerResolutionCharacterization: + @pytest.mark.asyncio + @pytest.mark.parametrize("caller", ["denied", "admin"]) + @pytest.mark.parametrize( + "approval_status,registered", + [ + ("pending_review", False), + ("rejected", False), + ("draft", False), + ("pending_review", True), + ("rejected", True), + ("draft", True), + (None, False), + ("active", False), + ], + ) + async def test_catalog_view_does_not_expose_hidden_database_details( + self, caller: str, approval_status: str | None, registered: bool + ) -> None: + server_id: Final = "lit3974_hidden_submission" + prisma, manager, auth = await self._resolution_case("db_runtime", caller, server_id) + hidden: Final = generate_mock_mcp_server_db_record(server_id=server_id).model_copy( + update={ + "approval_status": approval_status, + "submitted_by": "another-user", + "review_notes": "private submission review", + } + ) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=hidden) + if not registered: + manager.config_mcp_servers = {} + health: Final = AsyncMock(return_value=hidden) + add: Final = AsyncMock() + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add), + patch.object(manager, "health_check_server", health), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {"user_mcp_management_mode": "view_all"}), + ): + listed: Final = await mgmt_endpoints.fetch_all_mcp_servers(auth, team_id=None) + assert (server_id in {item.server_id for item in listed}) is registered + if caller != "admin": + with pytest.raises(HTTPException) as error: + await mgmt_endpoints.fetch_mcp_server(_make_mock_request(), server_id, auth) + assert error.value.status_code == 403 + add.assert_not_awaited() + health.assert_not_awaited() + return + detail: Final = await mgmt_endpoints.fetch_mcp_server(_make_mock_request(), server_id, auth) + assert detail.server_id == server_id + assert detail.submitted_by == "another-user" + assert detail.review_notes == "private submission review" + + @pytest.mark.asyncio + async def test_credential_metadata_resolves_permissions_once_for_multiple_servers(self) -> None: + first_id: Final = "lit3974_first_credential" + second_id: Final = "lit3974_second_credential" + prisma, manager, caller = await self._resolution_case("db_runtime", "allowed", first_id) + ids: Final = (first_id, second_id) + rows: Final = tuple( + generate_mock_mcp_server_db_record(server_id=sid, alias=f"alias-{sid}") for sid in ids + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=list(rows)) + auth: Final = caller.model_copy( + update={"object_permission": LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_multiple_credentials", mcp_servers=list(ids) + )} + ) + manager.config_mcp_servers = { + **manager.config_mcp_servers, + second_id: generate_mock_mcp_server_config_record(server_id=second_id), + } + permissions: Final = AsyncMock(wraps=manager.get_allowed_mcp_servers) + credentials: Final = [{"server_id": sid, "expires_at": None, "connected_at": None} for sid in ids] + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "get_allowed_mcp_servers", permissions), + patch.object(mgmt_endpoints, "list_user_oauth_credentials", AsyncMock(return_value=credentials)), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {}), + ): + result: Final = await mgmt_endpoints.list_mcp_user_credentials(auth) + assert [item.server_id for item in result] == list(ids) + assert [item.alias for item in result] == [row.alias for row in rows] + assert all(item.has_credential for item in result) + assert permissions.await_count <= 1, "credential count must not multiply permission resolution" + + @pytest.mark.asyncio + @pytest.mark.parametrize("source", ["db_runtime", "config"]) + @pytest.mark.parametrize( + "mode,restricted,allowed", + [ + pytest.param( + "view_all", + False, + True, + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(AssertionError, match="view_all detail denied"), + reason="LIT-3974 A: view_all permits redacted catalog detail", + ), + ), + ("view_all", True, False), + ("restricted", False, False), + ], + ) + async def test_detail_obeys_catalog_visibility( + self, + source: str, + mode: str, + restricted: bool, + allowed: bool, + ) -> None: + server_id: Final = "lit3974_visibility" + prisma, manager, caller = await self._resolution_case(source, "denied", server_id) + auth: Final = caller.model_copy(update={"allowed_routes": ["mcp_routes"] if restricted else []}) + health: Final = AsyncMock(return_value=generate_mock_mcp_server_db_record(server_id=server_id)) + add: Final = AsyncMock() + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add), + patch.object(manager, "health_check_server", health), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {"user_mcp_management_mode": mode}), + ): + listed: Final = await mgmt_endpoints.fetch_all_mcp_servers(auth, team_id=None) + assert (server_id in {item.server_id for item in listed}) is allowed + if not allowed: + with pytest.raises(HTTPException) as error: + await mgmt_endpoints.fetch_mcp_server(_make_mock_request(), server_id, auth) + assert error.value.status_code == 403 + add.assert_not_awaited() + health.assert_not_awaited() + return + try: + detail: Final = await mgmt_endpoints.fetch_mcp_server(_make_mock_request(), server_id, auth) + except HTTPException as error: + if error.status_code != 403: + raise + raise AssertionError("view_all detail denied") from error + assert detail.server_id == server_id + assert detail.credentials is None + assert detail.url is None + assert detail.static_headers is None + assert detail.env_vars is None + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "source,caller,visible", + [ + ("db_runtime", "allowed", True), + ("db_runtime", "admin", True), + pytest.param( + "db_runtime", + "denied", + False, + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(AssertionError, match="credential metadata visibility"), + reason="LIT-3974 C: revoked grants hide DB metadata without removing credentials", + ), + ), + pytest.param( + "config", + "allowed", + True, + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(AssertionError, match="credential metadata visibility"), + reason="LIT-3974 C: authorized config credential metadata", + ), + ), + pytest.param( + "config", + "admin", + True, + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc(AssertionError, match="credential metadata visibility"), + reason="LIT-3974 C: admin config credential metadata", + ), + ), + ("config", "denied", False), + ("missing", "allowed", False), + ("missing", "denied", False), + ("missing", "admin", False), + ], + ) + async def test_credential_metadata_requires_current_access( + self, + source: str, + caller: str, + visible: bool, + ) -> None: + server_id: Final = "lit3974_credential_metadata" + prisma, manager, auth = await self._resolution_case(source, caller, server_id) + credential: Final = {"server_id": server_id, "expires_at": None, "connected_at": None} + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(mgmt_endpoints, "list_user_oauth_credentials", AsyncMock(return_value=[credential])), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {}), + ): + result: Final = await mgmt_endpoints.list_mcp_user_credentials(auth) + assert len(result) == 1 + assert result[0].server_id == server_id + assert result[0].has_credential is True + assert result[0].expires_at is None + assert result[0].connected_at is None + assert result[0].server_name == (f"lit3974_{source}_server" if visible else None), ( + "credential metadata visibility" + ) + assert result[0].alias == ("lit3974_alias" if visible else None), "credential metadata visibility" + + async def _load_registry_config( + self, + manager: MCPServerManager, + config: dict[str, MCPServerConfig], + ) -> None: + existing_tasks: Final = asyncio.all_tasks() + with MockRouter(assert_all_called=False) as httpx_mock: + await manager.load_servers_from_config(config) + startup_tasks: Final = tuple(task for task in asyncio.all_tasks() if task not in existing_tasks) + for task in startup_tasks: + task.cancel() + await asyncio.gather(*startup_tasks, return_exceptions=True) + assert httpx_mock.calls.call_count == 0, "registry setup must not make upstream HTTP calls" + + async def _resolution_case( + self, + source: str, + caller: str, + server_id: str, + *, + is_byok: bool = False, + ) -> tuple[MagicMock, MCPServerManager, UserAPIKeyAuth]: + team_id: Final = "lit3974_resolution_team" + user_id: Final = f"lit3974_{caller}_user" + db_server: Final = generate_mock_mcp_server_db_record( + server_id=server_id, + alias="lit3974_alias", + ).model_copy( + update={ + "server_name": f"lit3974_{source}_server", + "is_byok": is_byok, + "env_vars": [ + { + "name": "LIT3974_TOKEN", + "value": "", + "scope": "user", + "description": "MCP credential", + } + ], + "static_headers": {"Authorization": "Bearer ${LIT3974_TOKEN}"}, + } + ) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_{caller}_permission", + mcp_servers=[server_id] if caller == "allowed" else [], + ) + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_resolution_team_permission", + mcp_servers=[server_id] if caller == "ui_allowed" else [], + ), + ) + user: Final = ( + LiteLLM_UserTable( + user_id=user_id, + teams=[team_id], + user_role=LitellmUserRoles.INTERNAL_USER, + ) + if caller == "ui_allowed" + else None + ) + prisma: Final = _mock_mcp_resolution_prisma_client(db_server, key_permission, team, user=user) + if source != "db_runtime": + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + + manager: Final = MCPServerManager() + if source in ("db_runtime", "config"): + await self._load_registry_config( + manager, + { + f"lit3974_{source}_server": { + "server_id": server_id, + "alias": "lit3974_alias", + "url": "https://mcp.example.com/server", + "transport": "http", + "is_byok": is_byok, + "env_vars": [ + { + "name": "LIT3974_TOKEN", + "value": "", + "scope": "user", + "description": "MCP credential", + } + ], + "static_headers": {"Authorization": "Bearer ${LIT3974_TOKEN}"}, + } + }, + ) + + auth: Final = UserAPIKeyAuth( + api_key=f"lit3974_{caller}_key", + user_id=user_id, + team_id=UI_SESSION_TOKEN_TEAM_ID if caller == "ui_allowed" else None, + user_role=(LitellmUserRoles.PROXY_ADMIN if caller == "admin" else LitellmUserRoles.INTERNAL_USER), + object_permission=key_permission if caller != "ui_allowed" else None, + ) + return prisma, manager, auth + + async def _detail_grant_case( + self, + source: str, + grant_route: str, + server_id: str, + ) -> tuple[MagicMock, MCPServerManager, UserAPIKeyAuth]: + team_id: Final = UI_SESSION_TOKEN_TEAM_ID if grant_route == "direct user object_permission" else "lit3974_team" + user_id: Final = "lit3974_direct_user" + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_{grant_route}_key_permission", + mcp_servers=None, + ) + route_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_{grant_route}_permission", + mcp_servers=[server_id], + ) + organization_id: Final = "lit3974_grant_organization" if grant_route == "org object_permission" else None + organization: Final = ( + LiteLLM_OrganizationTable( + organization_id=organization_id, + organization_alias="LIT-3974", + budget_id="lit3974-budget", + created_by="lit3974-test", + updated_by="lit3974-test", + object_permission=route_permission, + object_permission_id=route_permission.object_permission_id, + ) + if organization_id is not None + else None + ) + user: Final = ( + LiteLLM_UserTable( + user_id=user_id, + teams=[], + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission_id=route_permission.object_permission_id, + object_permission=route_permission, + ) + if grant_route == "direct user object_permission" + else None + ) + access_group: Final = ( + LiteLLM_AccessGroupTable( + access_group_id="lit3974_access_group", + access_group_name="LIT3974", + access_mcp_server_ids=[server_id], + ) + if grant_route == "access-group" + else None + ) + server: Final = generate_mock_mcp_server_db_record(server_id=server_id, alias="lit3974_grant").model_copy( + update={"allow_all_keys": grant_route == "allow_all_keys"} + ) + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_empty_team_permission", + mcp_servers=[], + ), + ) + prisma: Final = _mock_mcp_resolution_prisma_client( + server, + key_permission, + team, + user=user, + organization=organization, + access_group=access_group, + object_permission=route_permission if user is not None or organization is not None else None, + ) + if source != "db_runtime": + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + f"lit3974_{source}_grant": { + "server_id": server_id, + "alias": "lit3974_grant", + "url": "https://grant.example.com/mcp", + "transport": "http", + "allow_all_keys": grant_route == "allow_all_keys", + } + }, + ) + auth: Final = UserAPIKeyAuth( + api_key=None if user is not None else f"lit3974_{grant_route}_key", + user_id=user_id, + team_id=team_id if user is not None else None, + org_id=organization_id, + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=None if user is not None else key_permission, + access_group_ids=["lit3974_access_group"] if access_group is not None else None, + ) + return prisma, manager, auth + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "grant_route,identity_field,foreign_identity", + [ + ("org object_permission", "org_id", "lit3974_foreign_org"), + ("direct user object_permission", "user_id", "lit3974_foreign_user"), + ("access-group", "access_group_ids", ["lit3974_foreign_group"]), + ], + ) + async def test_grants_do_not_cross_caller_identities( + self, grant_route: str, identity_field: str, foreign_identity: str | list[str] + ) -> None: + server_id: Final = "lit3974_identity_isolation" + prisma, manager, auth = await self._detail_grant_case("config", grant_route, server_id) + foreign_auth: Final = auth.model_copy(update={identity_field: foreign_identity}) + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager", manager), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {}), + ): + permitted: Final = await mgmt_endpoints.fetch_all_mcp_servers(auth, team_id=None) + denied: Final = await mgmt_endpoints.fetch_all_mcp_servers(foreign_auth, team_id=None) + assert server_id in {server.server_id for server in permitted} + assert server_id not in {server.server_id for server in denied} + + @staticmethod + def _resolution_error(source: str, caller: str, server_id: str) -> tuple[int, dict[str, str]] | None: + if source == "missing": + if caller == "admin": + return 404, {"error": f"MCP Server {server_id} not found"} + return ( + 403, + { + "error": ( + f"User does not have permission to access mcp server with id {server_id}. " + "You can only manage mcp servers that you have access to." + ) + }, + ) + if caller == "denied": + return ( + 403, + { + "error": ( + f"User does not have permission to access mcp server with id {server_id}. " + "You can only manage mcp servers that you have access to." + ) + }, + ) + return None + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "operation,source,caller", + [ + ("byok_store", "db_runtime", "admin"), + ("byok_store", "db_runtime", "allowed"), + ("byok_store", "db_runtime", "denied"), + ("byok_store", "config", "allowed"), + ("byok_store", "config", "denied"), + ("byok_store", "missing", "admin"), + ("byok_store", "missing", "allowed"), + ("byok_store", "missing", "denied"), + ("byok_store", "config", "ui_allowed"), + ("oauth_store", "db_runtime", "admin"), + ("oauth_store", "db_runtime", "allowed"), + ("oauth_store", "db_runtime", "denied"), + ("oauth_store", "missing", "admin"), + ("oauth_store", "missing", "allowed"), + ("oauth_store", "missing", "denied"), + ("oauth_store", "config", "ui_allowed"), + ("env_get", "db_runtime", "admin"), + ("env_get", "db_runtime", "allowed"), + ("env_get", "db_runtime", "denied"), + ("env_get", "config", "admin"), + ("env_get", "config", "allowed"), + ("env_get", "config", "denied"), + ("env_get", "missing", "allowed"), + ("env_get", "config", "ui_allowed"), + ("env_store", "db_runtime", "admin"), + ("env_store", "db_runtime", "allowed"), + ("env_store", "db_runtime", "denied"), + ("env_store", "config", "admin"), + ("env_store", "config", "denied"), + ("env_store", "missing", "allowed"), + ("env_store", "missing", "denied"), + ("env_store", "config", "ui_allowed"), + ("env_clear", "db_runtime", "admin"), + ("env_clear", "db_runtime", "allowed"), + ("env_clear", "db_runtime", "denied"), + ("env_clear", "config", "admin"), + ("env_clear", "config", "allowed"), + ("env_clear", "config", "denied"), + ("env_clear", "missing", "allowed"), + ("env_clear", "missing", "denied"), + ("env_clear", "config", "ui_allowed"), + ], + ) + async def test_credential_and_env_var_resolution_cells( + self, + operation: str, + source: str, + caller: str, + ) -> None: + effects: Final = _ResolutionEffects() + server_id: Final = f"lit3974_{operation}_{source}" + prisma, manager, auth = await self._resolution_case( + source, + caller, + server_id, + is_byok=operation == "byok_store", + ) + + oauth_read: Final = AsyncMock(return_value={"expires_at": "2099-01-01T00:00:00+00:00"}) + env_read: Final = AsyncMock(return_value={}) + + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + expected_error: Final = self._resolution_error(source, caller, server_id) + + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch( + "litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager", + manager, + ), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch.object(mgmt_endpoints, "get_user_oauth_credential", oauth_read), + patch.object(mgmt_endpoints, "get_user_env_vars", env_read), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + if expected_error is not None: + with pytest.raises(HTTPException) as exc_info: + await self._call_credential_or_env_operation(operation, server_id, auth) + + assert exc_info.value.status_code == expected_error[0], f"{operation}/{source}/{caller}: status" + assert exc_info.value.detail == expected_error[1], f"{operation}/{source}/{caller}: full detail body" + effects.assert_no_writes() + add_server.assert_not_awaited() + health_check.assert_not_awaited() + assert httpx_mock.calls.call_count == 0, f"{operation}/{source}/{caller}: no upstream HTTP" + return + + if operation == "byok_store" and source == "config": + with pytest.raises(HTTPException) as exc_info: + await self._call_credential_or_env_operation(operation, server_id, auth) + + assert exc_info.value.status_code == 400, f"{operation}/{source}/{caller}: status" + assert exc_info.value.detail == {"error": "This MCP server does not support BYOK credentials"}, ( + f"{operation}/{source}/{caller}: full detail body" + ) + effects.assert_no_writes() + add_server.assert_not_awaited() + health_check.assert_not_awaited() + assert httpx_mock.calls.call_count == 0, f"{operation}/{source}/{caller}: no upstream HTTP" + return + + result: Final = await self._call_credential_or_env_operation(operation, server_id, auth) + + if operation == "byok_store": + assert result.model_dump() == {"server_id": server_id, "has_credential": True} + effects.byok_store.assert_awaited_once() + effects.byok_invalidate.assert_awaited_once_with(auth.user_id, server_id) + elif operation == "oauth_store": + assert result.model_dump() == { + "server_id": server_id, + "has_credential": True, + "expires_at": "2099-01-01T00:00:00+00:00", + "is_expired": False, + "connected_at": None, + } + effects.oauth_store.assert_awaited_once() + effects.oauth_invalidate.assert_awaited_once_with(auth.user_id, server_id) + elif operation == "env_get": + assert result.model_dump() == { + "server_id": server_id, + "server_name": f"lit3974_{source}_server", + "alias": "lit3974_alias", + "required": [{"name": "LIT3974_TOKEN", "description": "MCP credential", "is_set": False}], + "missing_count": 1, + "setup_url": f"/ui/mcp-servers?fill_env_vars={server_id}", + } + env_read.assert_awaited_once_with(prisma, auth.user_id, server_id) + elif operation == "env_store": + assert result.model_dump() == { + "server_id": server_id, + "server_name": f"lit3974_{source}_server", + "alias": "lit3974_alias", + "required": [{"name": "LIT3974_TOKEN", "description": "MCP credential", "is_set": True}], + "missing_count": 0, + "setup_url": f"/ui/mcp-servers?fill_env_vars={server_id}", + } + effects.env_merge.assert_awaited_once() + effects.env_invalidate.assert_called_once_with(auth.user_id, server_id) + else: + assert result.model_dump() == { + "server_id": server_id, + "server_name": f"lit3974_{source}_server", + "alias": "lit3974_alias", + "required": [{"name": "LIT3974_TOKEN", "description": "MCP credential", "is_set": False}], + "missing_count": 1, + "setup_url": f"/ui/mcp-servers?fill_env_vars={server_id}", + } + effects.env_delete.assert_awaited_once_with(prisma, auth.user_id, server_id) + effects.env_invalidate.assert_called_once_with(auth.user_id, server_id) + + async def _call_credential_or_env_operation( + self, + operation: str, + server_id: str, + auth: UserAPIKeyAuth, + ) -> MCPUserCredentialResponse | mgmt_endpoints.MCPOAuthUserCredentialStatus | mgmt_endpoints.MCPUserEnvVarsStatus: + if operation == "byok_store": + return await mgmt_endpoints.store_mcp_user_credential( + server_id=server_id, + payload=mgmt_endpoints.MCPUserCredentialRequest(credential="lit3974-secret"), + user_api_key_dict=auth, + ) + if operation == "oauth_store": + return await mgmt_endpoints.store_mcp_oauth_user_credential( + server_id=server_id, + payload=mgmt_endpoints.MCPOAuthUserCredentialRequest( + access_token="lit3974-token", + expires_in=3600, + ), + user_api_key_dict=auth, + ) + if operation == "env_get": + return await mgmt_endpoints.get_mcp_user_env_vars( + server_id=server_id, + user_api_key_dict=auth, + ) + if operation == "env_store": + return await mgmt_endpoints.store_mcp_user_env_vars( + server_id=server_id, + payload=mgmt_endpoints.MCPUserEnvVarsRequest(values={"LIT3974_TOKEN": "lit3974-secret"}), + user_api_key_dict=auth, + ) + return await mgmt_endpoints.clear_mcp_user_env_vars( + server_id=server_id, + user_api_key_dict=auth, + ) + + @pytest.mark.asyncio + async def test_oauth_credential_status_does_not_resolve_server_access(self) -> None: + server_id: Final = "lit3974_missing_oauth_status" + prisma, manager, auth = await self._resolution_case("missing", "denied", server_id) + oauth_read: Final = AsyncMock(return_value=None) + oauth_invalidate: Final = AsyncMock() + + with ( + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "invalidate_user_oauth_token_cache", oauth_invalidate), + patch.object(mgmt_endpoints, "get_user_oauth_credential", oauth_read), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + result: Final = await mgmt_endpoints.get_mcp_oauth_user_credential_status( + server_id=server_id, + user_api_key_dict=auth, + ) + + assert result.model_dump() == { + "server_id": server_id, + "has_credential": False, + "expires_at": None, + "is_expired": False, + "connected_at": None, + } + oauth_read.assert_awaited_once_with(prisma, auth.user_id, server_id) + oauth_invalidate.assert_not_awaited() + assert httpx_mock.calls.call_count == 0 + + @pytest.mark.asyncio + async def test_non_admin_deletes_own_oauth_credential_for_missing_server(self) -> None: + server_id: Final = "lit3974_removed_oauth_server" + user_id: Final = "lit3974_oauth_owner" + prisma: Final = _mock_mcp_resolution_prisma_client( + generate_mock_mcp_server_db_record(server_id=server_id), + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974_delete_key", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974_delete_team"), + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + auth: Final = UserAPIKeyAuth( + api_key="lit3974_oauth_owner_key", + user_id=user_id, + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_delete_key", + mcp_servers=[], + ), + ) + credential_read: Final = AsyncMock(return_value={"type": "oauth2", "access_token": "lit3974-token"}) + delete_credential: Final = AsyncMock() + invalidate: Final = AsyncMock() + + with ( + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch( + "litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager", + manager, + ), + patch.object(manager, "invalidate_user_oauth_token_cache", invalidate), + patch.object(mgmt_endpoints, "get_user_oauth_credential", credential_read), + patch.object(mgmt_endpoints, "delete_user_credential", delete_credential), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + result: Final = await mgmt_endpoints.delete_mcp_oauth_user_credential( + server_id=server_id, + user_api_key_dict=auth, + ) + + assert result.model_dump() == { + "server_id": server_id, + "has_credential": False, + "expires_at": None, + "is_expired": False, + "connected_at": None, + } + credential_read.assert_awaited_once_with(prisma, user_id, server_id) + delete_credential.assert_awaited_once_with(prisma, user_id, server_id) + invalidate.assert_awaited_once_with(user_id, server_id) + assert httpx_mock.calls.call_count == 0 + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "server_id,exists,role,expected_status", + [ + ("lit3974_duplicate", True, LitellmUserRoles.PROXY_ADMIN, 400), + ("lit3974_new", False, LitellmUserRoles.PROXY_ADMIN, 200), + ("all-team-mcpservers", False, LitellmUserRoles.PROXY_ADMIN, 400), + ("all-proxy-mcpservers", False, LitellmUserRoles.PROXY_ADMIN, 400), + ("lit3974_new", False, LitellmUserRoles.INTERNAL_USER, 403), + ], + ) + async def test_create_checks_identifier_before_side_effects( + self, + server_id: str, + exists: bool, + role: LitellmUserRoles, + expected_status: int, + ) -> None: + server: Final = generate_mock_mcp_server_db_record(server_id=server_id) + prisma: Final = _mock_mcp_resolution_prisma_client( + server, + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974_create_key", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974_create_team"), + ) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=server if exists else None) + manager: Final = MCPServerManager() + create_server: Final = AsyncMock(return_value=server) + add_server: Final = AsyncMock() + reload_servers: Final = AsyncMock() + payload: Final = NewMCPServerRequest( + server_id=server_id, + alias="lit3974_create", + url="https://mcp.example.com/create", + transport=MCPTransport.http, + ) + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "reload_servers_from_database", reload_servers), + patch.object(mgmt_endpoints, "create_mcp_server_if_identifier_free", create_server), + ): + operation: Final = mgmt_endpoints.add_mcp_server( + payload=payload, + user_api_key_dict=generate_mock_user_api_key_auth(user_role=role), + ) + if expected_status == 200: + result: Final = await operation + assert result.server_id == server_id + create_server.assert_awaited_once() + add_server.assert_awaited_once_with(server) + reload_servers.assert_awaited_once() + return + with pytest.raises(HTTPException) as error: + await operation + assert error.value.status_code == expected_status + assert error.value.detail == { + "error": ( + "User does not have permission to create mcp servers. You can only create mcp servers if you are a PROXY_ADMIN." + if expected_status == 403 + else f"MCP Server with id {server_id} already exists. Cannot create another." + if exists + else f"MCP Server with id {server_id} is special and cannot be used." + ) + } + create_server.assert_not_awaited() + add_server.assert_not_awaited() + reload_servers.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.parametrize("case", ["no-user", "empty", "missing-id"]) + async def test_credential_list_boundaries_do_not_resolve_servers(self, case: str) -> None: + prisma: Final = MagicMock() + rows: Final = AsyncMock(return_value=[{}] if case == "missing-id" else []) + batch: Final = AsyncMock(return_value=[]) + manager: Final = MCPServerManager() + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "list_user_oauth_credentials", rows), + patch.object(mgmt_endpoints, "get_mcp_servers", batch), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "get_mcp_server_by_id") as lookup, + ): + auth: Final = _make_user_auth("" if case == "no-user" else "lit3974_list_user") + if case == "no-user": + with pytest.raises(HTTPException) as error: + await mgmt_endpoints.list_mcp_user_credentials(auth) + assert error.value.status_code == 400 + assert error.value.detail == {"error": "User ID not found in token"} + rows.assert_not_awaited() + else: + assert await mgmt_endpoints.list_mcp_user_credentials(auth) == [] + lookup.assert_not_called() + if case == "missing-id": + batch.assert_awaited_once_with(prisma, []) + else: + batch.assert_not_awaited() + + @pytest.mark.asyncio + async def test_user_credential_list_keeps_entry_for_missing_server_in_one_batch(self) -> None: + missing_server_id: Final = "lit3974_list_missing_server" + manager: Final = MCPServerManager() + prisma_client: Final = MagicMock() + credential_rows: Final = [ + { + "server_id": missing_server_id, + "expires_at": None, + "connected_at": None, + }, + ] + list_credentials: Final = AsyncMock(return_value=credential_rows) + get_servers: Final = AsyncMock(return_value=[]) + get_single_server: Final = AsyncMock() + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma_client), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(mgmt_endpoints, "list_user_oauth_credentials", list_credentials), + patch.object(mgmt_endpoints, "get_mcp_servers", get_servers), + patch.object(mgmt_endpoints, "get_mcp_server", get_single_server), + ): + result: Final = await mgmt_endpoints.list_mcp_user_credentials( + user_api_key_dict=generate_mock_user_api_key_auth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="lit3974_list_user", + ) + ) + + assert [item.model_dump() for item in result] == [ + { + "server_id": missing_server_id, + "server_name": None, + "alias": None, + "credential_type": "oauth2", + "has_credential": True, + "expires_at": None, + "connected_at": None, + }, + ] + get_servers.assert_awaited_once_with(prisma_client, [missing_server_id]) + get_single_server.assert_not_awaited() + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "source,caller,expected_status", + [ + ("db_runtime", "admin", 200), + ("db_runtime", "view_only", 200), + ("db_runtime", "allowed", 200), + ("db_runtime", "denied", 403), + ("config", "view_only", 200), + ("config", "ui_key_allowed", 200), + ("config", "ui_denied", 403), + ("missing", "admin", 404), + ("missing", "view_only", 404), + ("missing", "allowed", 404), + ("missing", "denied", 404), + ], + ) + async def test_fetch_mcp_server_resolution_cells( + self, + source: str, + caller: str, + expected_status: int, + ) -> None: + effects: Final = _ResolutionEffects() + server_id: Final = f"lit3974_{source}_detail" + server: Final = generate_mock_mcp_server_db_record(server_id=server_id, alias="LIT3974 detail") + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_{source}_{caller}_permission", + mcp_servers=[server_id] if caller in ("allowed", "ui_key_allowed") else [], + ) + team: Final = LiteLLM_TeamTable( + team_id="lit3974_detail_team", + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_detail_team_permission", + mcp_servers=[], + ), + ) + user: Final = ( + LiteLLM_UserTable( + user_id="lit3974_detail_user", + teams=[], + user_role=LitellmUserRoles.INTERNAL_USER, + ) + if caller in ("ui_denied", "ui_key_allowed") + else None + ) + prisma: Final = _mock_mcp_resolution_prisma_client(server, key_permission, team, user=user) + if source != "db_runtime": + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + + manager: Final = MCPServerManager() + if source in ("db_runtime", "config"): + await self._load_registry_config( + manager, + { + "lit3974_detail_server": { + "server_id": server_id, + "alias": "LIT3974 detail", + "url": "https://detail.example.com/mcp", + "transport": "http", + } + }, + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock(return_value=server) + + auth: Final = UserAPIKeyAuth( + api_key=f"lit3974_{source}_{caller}_key", + user_id="lit3974_detail_user", + team_id=UI_SESSION_TOKEN_TEAM_ID if caller in ("ui_denied", "ui_key_allowed") else None, + user_role=( + LitellmUserRoles.PROXY_ADMIN + if caller == "admin" + else LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY + if caller == "view_only" + else LitellmUserRoles.INTERNAL_USER + ), + object_permission=key_permission, + ) + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + if expected_status in (403, 404): + with pytest.raises(HTTPException) as exc_info: + await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + + expected_detail: Final = ( + {"error": f"MCP Server with id {server_id} not found"} + if expected_status == 404 + else { + "error": ( + f"User does not have permission to view mcp server with id {server_id}. " + "You can only view mcp servers that you have access to." + ) + } + ) + assert exc_info.value.status_code == expected_status, f"{source}/{caller}: detail status" + assert exc_info.value.detail == expected_detail, f"{source}/{caller}: complete detail body" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0, f"{source}/{caller}: no upstream HTTP" + return + + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + + assert result.server_id == server_id, f"{source}/{caller}: resolved server id" + assert result.alias == "LIT3974 detail", f"{source}/{caller}: resolved display alias" + if source == "db_runtime": + add_server.assert_awaited_once() + else: + add_server.assert_not_awaited() + health_check.assert_awaited_once_with(server_id) + + @pytest.mark.asyncio + async def test_fetch_config_alias_filters_external_client_ip(self) -> None: + effects: Final = _ResolutionEffects() + server_id: Final = "lit3974_private_config" + prisma: Final = _mock_mcp_resolution_prisma_client( + generate_mock_mcp_server_db_record(server_id=server_id), + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974_private_key", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974_private_team"), + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + "lit3974_private_server": { + "server_id": server_id, + "alias": "private_alias", + "url": "https://private.example.com/mcp", + "transport": "http", + "available_on_public_internet": False, + } + }, + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + + auth: Final = UserAPIKeyAuth( + api_key="lit3974_private_key", + user_id="lit3974_private_user", + user_role=LitellmUserRoles.PROXY_ADMIN, + ) + + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + ): + with pytest.raises(HTTPException) as exc_info: + await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(ip="203.0.113.25"), + server_id="private_alias", + user_api_key_dict=auth, + ) + + assert exc_info.value.status_code == 404, "config alias hidden from an external client IP" + assert exc_info.value.detail == {"error": "MCP Server with id private_alias not found"}, ( + "complete IP-filtered alias lookup detail" + ) + add_server.assert_not_awaited() + health_check.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0 + + @pytest.mark.asyncio + async def test_fetch_db_runtime_ignores_external_client_ip(self) -> None: + server_id: Final = "lit3974_private_db" + server: Final = generate_mock_mcp_server_db_record(server_id=server_id, alias="Private DB") + prisma: Final = _mock_mcp_resolution_prisma_client( + server, + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974_private_db_key", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974_private_db_team"), + ) + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + "lit3974_private_db_server": { + "server_id": server_id, + "alias": "Private DB", + "url": "https://private.example.com/mcp", + "transport": "http", + "available_on_public_internet": False, + } + }, + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock(return_value=server) + auth: Final = UserAPIKeyAuth( + api_key="lit3974_private_db_admin", + user_id="lit3974_private_db_admin", + user_role=LitellmUserRoles.PROXY_ADMIN, + ) + + with ( + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + ): + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(ip="203.0.113.25"), + server_id=server_id, + user_api_key_dict=auth, + ) + + assert result.server_id == server_id, "DB detail lookup is not filtered by the client IP" + assert result.alias == "Private DB", "DB detail response retains its alias" + add_server.assert_awaited_once() + health_check.assert_awaited_once_with(server_id) + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "source,caller,expected_status", + [ + ("temp_mem", "allowed", 403), + ("temp_draft", "admin", 200), + ("temp_draft", "allowed", 403), + ("temp_draft", "denied", 403), + ("temp_redis", "admin", 200), + ("temp_redis", "allowed", 403), + ("temp_redis", "denied", 403), + ("config", "admin", 200), + ("db_only", "admin", 404), + ("db_only", "allowed", 404), + ("db_only", "denied", 404), + ("missing", "allowed", 404), + ("missing", "denied", 404), + ], + ) + async def test_temporary_oauth_resolution_source_and_caller_cells( + self, + source: str, + caller: str, + expected_status: int, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + effects: Final = _ResolutionEffects() + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + _cache_temporary_mcp_server_in_redis, + _get_cached_temporary_mcp_server_or_404, + _TemporaryMCPServerEntry, + ) + + monkeypatch.setenv("LITELLM_SALT_KEY", "lit3974-test-salt-key") + server_id: Final = f"lit3974_{source}_oauth" + temp_server: Final = generate_mock_mcp_server_config_record(server_id=server_id) + db_server: Final = generate_mock_mcp_server_db_record(server_id=server_id) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_{source}_{caller}_oauth_permission", + mcp_servers=[server_id] if caller == "allowed" else [], + ) + team: Final = LiteLLM_TeamTable(team_id=f"lit3974_{source}_oauth_team") + prisma: Final = _mock_mcp_resolution_prisma_client(db_server, key_permission, team) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock( + return_value=db_server if source == "db_only" else None + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock( + return_value=[db_server.model_copy(update={"approval_status": "draft"})] if source == "temp_draft" else [] + ) + manager: Final = MCPServerManager() + config: Final = ( + { + "lit3974_oauth_config": { + "server_id": server_id, + "alias": "LIT3974 OAuth", + "url": "https://oauth.example.com/mcp", + "transport": "http", + } + } + if source == "config" + else { + "lit3974_oauth_unrelated": { + "server_id": "lit3974_unrelated_oauth", + "url": "https://unrelated.example.com/mcp", + "transport": "http", + } + } + ) + await self._load_registry_config(manager, config) + auth: Final = UserAPIKeyAuth( + api_key=f"lit3974_{source}_{caller}_oauth_key", + user_id="lit3974_oauth_user", + user_role=(LitellmUserRoles.PROXY_ADMIN if caller == "admin" else LitellmUserRoles.INTERNAL_USER), + object_permission=key_permission, + ) + cache_backend: Final = SimpleNamespace( + async_get_cache=AsyncMock(return_value=None), + async_set_cache=AsyncMock(), + ) + original_cache: Final = mgmt_endpoints.litellm.cache + mgmt_endpoints.litellm.cache = SimpleNamespace(cache=cache_backend) + memory_cache: Final = ( + { + server_id: _TemporaryMCPServerEntry( + server=temp_server, + expires_at=datetime.utcnow() + timedelta(seconds=300), + ) + } + if source == "temp_mem" + else {} + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + + try: + if source == "temp_redis": + await _cache_temporary_mcp_server_in_redis(temp_server, ttl_seconds=300) + cache_backend.async_get_cache = AsyncMock( + return_value=cache_backend.async_set_cache.await_args.kwargs["value"] + ) + + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "_temporary_mcp_servers", memory_cache), + patch.object(mgmt_endpoints, "_get_prisma_client_or_none", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + if expected_status in (403, 404): + with pytest.raises(HTTPException) as exc_info: + await _get_cached_temporary_mcp_server_or_404( + server_id, + auth, + request=_make_mock_request(), + ) + + expected_detail: Final = ( + {"error": f"MCP server {server_id} not found"} + if expected_status == 404 + else {"error": f"Access denied to MCP server {server_id}"} + ) + assert exc_info.value.status_code == expected_status, f"{source}/{caller}: OAuth resolution status" + assert exc_info.value.detail == expected_detail, f"{source}/{caller}: complete OAuth detail body" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0, f"{source}/{caller}: no upstream HTTP" + else: + resolved: Final = await _get_cached_temporary_mcp_server_or_404( + server_id, + auth, + request=_make_mock_request(), + ) + expected_alias: Final = ( + db_server.alias + if source == "temp_draft" + else "LIT3974 OAuth" + if source == "config" + else temp_server.alias + ) + assert resolved.server_id == server_id, f"{source}/{caller}: resolved OAuth server" + assert resolved.alias == expected_alias, f"{source}/{caller}: resolved OAuth display name" + finally: + mgmt_endpoints.litellm.cache = original_cache + + @pytest.mark.asyncio + async def test_temporary_oauth_id_and_name_lookup_keep_distinct_ip_behavior(self) -> None: + effects: Final = _ResolutionEffects() + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + _get_cached_temporary_mcp_server_or_404, + _TemporaryMCPServerEntry, + ) + + server_id: Final = "lit3974_private_oauth" + prisma: Final = _mock_mcp_resolution_prisma_client( + generate_mock_mcp_server_db_record(server_id=server_id), + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974_private_oauth_permission", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974_private_oauth_team"), + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + "lit3974_private_oauth": { + "server_id": server_id, + "alias": "private_oauth_alias", + "url": "https://private.example.com/mcp", + "transport": "http", + "available_on_public_internet": False, + } + }, + ) + entry: Final = _TemporaryMCPServerEntry( + server=generate_mock_mcp_server_config_record(server_id="lit3974_unused_temp"), + expires_at=datetime.utcnow() + timedelta(seconds=300), + ) + auth: Final = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) + request: Final = _make_mock_request(ip="203.0.113.25") + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + + original_cache: Final = mgmt_endpoints.litellm.cache + mgmt_endpoints.litellm.cache = SimpleNamespace( + cache=SimpleNamespace(async_get_cache=AsyncMock(return_value=None)) + ) + try: + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "_temporary_mcp_servers", {entry.server.server_id: entry}), + patch.object(mgmt_endpoints, "_get_prisma_client_or_none", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + ): + resolved: Final = await _get_cached_temporary_mcp_server_or_404( + server_id, + auth, + request=request, + ) + assert resolved.server_id == server_id, "registry ID lookup omits client-IP filtering" + with pytest.raises(HTTPException) as exc_info: + await _get_cached_temporary_mcp_server_or_404( + "private_oauth_alias", + auth, + request=request, + ) + finally: + mgmt_endpoints.litellm.cache = original_cache + + assert exc_info.value.status_code == 404, "registry name lookup filters an external client IP" + assert exc_info.value.detail == {"error": "MCP server private_oauth_alias not found"}, ( + "complete OAuth alias IP-filter detail" + ) + add_server.assert_not_awaited() + health_check.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0 + + @pytest.mark.asyncio + @pytest.mark.parametrize("endpoint", ["authorize", "token", "register"], ids=["authorize", "token", "register"]) + @pytest.mark.parametrize( + "source,expected_status", + [("config_denied", 403), ("missing", 404)], + ids=["existing-but-denied", "missing"], + ) + async def test_oauth_endpoints_reject_denied_and_missing_servers_before_upstream( + self, + endpoint: str, + source: str, + expected_status: int, + ) -> None: + effects: Final = _ResolutionEffects() + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + mcp_authorize, + mcp_register, + mcp_token, + ) + + server_id: Final = f"lit3974_{source}_oauth_endpoint" + db_server: Final = generate_mock_mcp_server_db_record(server_id=server_id) + prisma: Final = _mock_mcp_resolution_prisma_client( + db_server, + LiteLLM_ObjectPermissionTable(object_permission_id="lit3974_oauth_endpoint_key", mcp_servers=[]), + LiteLLM_TeamTable(team_id="lit3974_oauth_endpoint_team"), + ) + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + "lit3974_oauth_endpoint_server": { + "server_id": server_id, + "alias": "LIT3974 OAuth endpoint", + "url": "https://oauth.example.com/mcp", + "transport": "http", + } + } + if source == "config_denied" + else { + "lit3974_oauth_endpoint_unrelated": { + "server_id": "lit3974_unrelated_oauth_endpoint", + "url": "https://unrelated.example.com/mcp", + "transport": "http", + } + }, + ) + auth: Final = UserAPIKeyAuth( + api_key="lit3974_oauth_endpoint_key", + user_id="lit3974_oauth_endpoint_user", + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_oauth_endpoint_key_permission", + mcp_servers=[], + ), + ) + cache_backend: Final = SimpleNamespace(async_get_cache=AsyncMock(return_value=None)) + original_cache: Final = mgmt_endpoints.litellm.cache + mgmt_endpoints.litellm.cache = SimpleNamespace(cache=cache_backend) + upstream_authorize: Final = AsyncMock() + upstream_token: Final = AsyncMock() + upstream_register: Final = AsyncMock() + + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + request: Final = _make_mock_request() + try: + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "_temporary_mcp_servers", {}), + patch.object(mgmt_endpoints, "_get_prisma_client_or_none", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch.object(mgmt_endpoints, "authorize_with_server", upstream_authorize), + patch.object(mgmt_endpoints, "exchange_token_with_server", upstream_token), + patch.object(mgmt_endpoints, "register_client_with_server", upstream_register), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + if endpoint == "authorize": + operation = mcp_authorize( + request=request, + server_id=server_id, + user_api_key_dict=auth, + client_id="lit3974-client", + redirect_uri="https://client.example.com/callback", + ) + elif endpoint == "token": + operation = mcp_token( + request=request, + server_id=server_id, + user_api_key_dict=auth, + grant_type="authorization_code", + ) + else: + operation = mcp_register( + request=request, + server_id=server_id, + user_api_key_dict=auth, + ) + with pytest.raises(HTTPException) as exc_info: + await operation + + expected_detail: Final = ( + {"error": f"Access denied to MCP server {server_id}"} + if expected_status == 403 + else {"error": f"MCP server {server_id} not found"} + ) + assert exc_info.value.status_code == expected_status, f"{endpoint}/{source}: OAuth status" + assert exc_info.value.detail == expected_detail, f"{endpoint}/{source}: complete OAuth detail body" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + upstream_authorize.assert_not_awaited() + upstream_token.assert_not_awaited() + upstream_register.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0, f"{endpoint}/{source}: no upstream HTTP" + finally: + mgmt_endpoints.litellm.cache = original_cache + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "source,grant_route", + [ + pytest.param( + "db_runtime", + "org object_permission", + id="db-runtime-org-object-permission", + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, + check=lambda error: error.status_code == 403 and "permission" in str(error.detail), + ), + reason="LIT-3974 change A: detail authorization includes org object_permission grants", + ), + ), + pytest.param("config", "org object_permission", id="config-org-object-permission"), + pytest.param( + "db_runtime", + "direct user object_permission", + id="db-runtime-direct-user-permission", + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, + check=lambda error: error.status_code == 403 and "permission" in str(error.detail), + ), + reason="LIT-3974 change A: detail authorization includes direct user object_permission grants", + ), + ), + pytest.param( + "config", + "direct user object_permission", + id="config-direct-user-permission", + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, + check=lambda error: error.status_code == 403 and "permission" in str(error.detail), + ), + reason="LIT-3974 change A: detail authorization includes direct user object_permission grants", + ), + ), + pytest.param( + "db_runtime", + "allow_all_keys", + id="db-runtime-allow-all-keys", + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, + check=lambda error: error.status_code == 403 and "permission" in str(error.detail), + ), + reason="LIT-3974 change A: detail authorization includes allow_all_keys grants", + ), + ), + pytest.param("config", "allow_all_keys", id="config-allow-all-keys"), + pytest.param( + "db_runtime", + "access-group", + id="db-runtime-access-group", + marks=pytest.mark.xfail( + strict=True, + raises=pytest.RaisesExc( + HTTPException, + check=lambda error: error.status_code == 403 and "permission" in str(error.detail), + ), + reason="LIT-3974 change A: detail authorization includes access-group grants", + ), + ), + pytest.param("config", "access-group", id="config-access-group"), + ], + ) + async def test_fetch_mcp_server_widening_grant_routes(self, source: str, grant_route: str) -> None: + effects: Final = _ResolutionEffects() + server_id: Final = f"lit3974_{source}_{grant_route.replace(' ', '_')}" + prisma, manager, auth = await self._detail_grant_case(source, grant_route, server_id) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock( + return_value=generate_mock_mcp_server_db_record(server_id=server_id, alias="lit3974_grant") + ) + + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {}), + ): + if grant_route == "direct user object_permission": + effective_contexts: Final = await mgmt_endpoints.build_effective_auth_contexts(auth) + admitted_context: Final = next( + (context for context in effective_contexts if getattr(context, "mcp_admitted_user_subject", False)), + None, + ) + assert admitted_context is not None, "direct user permission must resolve an admitted context" + assert server_id in await manager.get_allowed_mcp_servers(admitted_context) + else: + assert server_id in await manager.get_allowed_mcp_servers(auth), ( + f"{source}/{grant_route}: real grant resolution must include the server" + ) + + def assert_detail_denial(exc: HTTPException) -> None: + logging.warning( + "%s/%s: HTTP %s detail=%r", + source, + grant_route, + exc.status_code, + exc.detail, + ) + assert exc.status_code == 403, f"{source}/{grant_route}: detail denial status" + assert exc.detail == { + "error": ( + f"User does not have permission to view mcp server with id {server_id}. " + "You can only view mcp servers that you have access to." + ) + }, f"{source}/{grant_route}: complete detail denial body" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0 + + try: + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + except HTTPException as exc: + assert_detail_denial(exc) + raise + + assert result.server_id == server_id, f"{source}/{grant_route}: detail server ID" + assert result.alias == "lit3974_grant", f"{source}/{grant_route}: detail alias" + if source == "db_runtime": + add_server.assert_awaited_once() + else: + add_server.assert_not_awaited() + health_check.assert_awaited_once() + assert httpx_mock.calls.call_count == 0 + + @pytest.mark.asyncio + async def test_fetch_mcp_server_allows_restricted_key_with_granted_database_server(self) -> None: + server_id: Final = "lit3974_restricted_detail" + server: Final = generate_mock_mcp_server_db_record(server_id=server_id, alias="restricted_detail").model_copy( + update={ + "credentials": {"auth_value": "top-secret"}, + "static_headers": {"Authorization": "Bearer top-secret"}, + } + ) + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id="lit3974_restricted_detail_permission", + mcp_servers=[server_id], + ) + prisma: Final = _mock_mcp_resolution_prisma_client( + server, + key_permission, + LiteLLM_TeamTable(team_id="lit3974_restricted_detail_team"), + ) + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + "lit3974_restricted_detail": { + "server_id": server_id, + "alias": "restricted_detail", + "url": "https://restricted.example.com/mcp", + "transport": "http", + } + }, + ) + auth: Final = UserAPIKeyAuth( + api_key="lit3974_restricted_detail_key", + user_id="lit3974_restricted_detail_user", + user_role=LitellmUserRoles.INTERNAL_USER, + allowed_routes=["mcp_routes"], + object_permission=key_permission, + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock(return_value=server) + + with ( + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + ): + result: Final = await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + + assert result.server_id == server_id + assert result.alias == "restricted_detail" + assert result.credentials is None + assert result.url is None + assert result.static_headers is None + assert result.env_vars is None + assert result.env == {} + assert result.command is None + assert result.args == [] + assert result.extra_headers == [] + assert result.allowed_tools == [] + assert result.mcp_access_groups == [] + assert result.teams == [] + add_server.assert_awaited_once() + health_check.assert_awaited_once() + assert httpx_mock.calls.call_count == 0 + + @pytest.mark.asyncio + @pytest.mark.parametrize("source", ["db_runtime", "config"], ids=["db-runtime", "config"]) + async def test_fetch_mcp_server_denies_key_without_explicit_mcp_access_when_required(self, source: str) -> None: + effects: Final = _ResolutionEffects() + server_id: Final = f"lit3974_require_key_access_{source}" + key_permission: Final = LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_require_key_access_{source}_permission", + mcp_servers=None, + ) + server: Final = generate_mock_mcp_server_db_record(server_id=server_id, alias="Team-only server") + team_id: Final = f"lit3974_require_key_access_{source}_team" + team: Final = LiteLLM_TeamTable( + team_id=team_id, + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id=f"lit3974_require_key_access_{source}_team_permission", + mcp_servers=[server_id], + ), + ) + prisma: Final = _mock_mcp_resolution_prisma_client(server, key_permission, team) + if source == "config": + prisma.db.litellm_mcpservertable.find_many = AsyncMock(return_value=[]) + prisma.db.litellm_mcpservertable.find_unique = AsyncMock(return_value=None) + manager: Final = MCPServerManager() + await self._load_registry_config( + manager, + { + f"lit3974_{source}_require_key_access": { + "server_id": server_id, + "alias": "Team-only server", + "url": "https://team-only.example.com/mcp", + "transport": "http", + } + }, + ) + auth: Final = UserAPIKeyAuth( + api_key=f"lit3974_require_key_access_{source}_key", + user_id=f"lit3974_require_key_access_{source}_user", + team_id=team_id, + user_role=LitellmUserRoles.INTERNAL_USER, + object_permission=key_permission, + ) + add_server: Final = AsyncMock() + health_check: Final = AsyncMock() + + with ( + effects.patch(manager), + MockRouter(assert_all_called=False) as httpx_mock, + patch.object(mgmt_endpoints, "get_prisma_client_or_throw", return_value=prisma), + patch.object(mgmt_endpoints, "global_mcp_server_manager", manager), + patch.object(manager, "add_server", add_server), + patch.object(manager, "health_check_server", health_check), + patch("litellm.proxy.proxy_server.prisma_client", prisma), + patch("litellm.proxy.proxy_server.user_api_key_cache", _mock_mcp_resolution_cache()), + patch("litellm.proxy.proxy_server.general_settings", {"require_key_mcp_access_defined": True}), + ): + with pytest.raises(HTTPException) as exc_info: + await mgmt_endpoints.fetch_mcp_server( + request=_make_mock_request(), + server_id=server_id, + user_api_key_dict=auth, + ) + + assert exc_info.value.status_code == 403 + assert exc_info.value.detail == { + "error": ( + f"User does not have permission to view mcp server with id {server_id}. " + "You can only view mcp servers that you have access to." + ) + }, f"{source}: complete detail denial body with require_key_mcp_access_defined" + add_server.assert_not_awaited() + health_check.assert_not_awaited() + effects.assert_no_writes() + assert httpx_mock.calls.call_count == 0