diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py index 9c510a568e6..173351ac8c5 100644 --- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py +++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py @@ -477,6 +477,40 @@ if MCP_AVAILABLE: allowed_routes = getattr(user_api_key_dict, "allowed_routes", None) return isinstance(allowed_routes, list) and len(allowed_routes) > 0 + def _sanitize_mcp_server_for_non_admin( + mcp_server: LiteLLM_MCPServerTable, + ) -> LiteLLM_MCPServerTable: + """Strip credential-bearing fields for non-admin viewers. + + Non-admin users may legitimately need to discover MCP servers + their team has access to (so they can pick one in the UI), but + they must never see fields that can carry bearer tokens or + upstream API keys. ``_redact_mcp_credentials`` already clears + the explicit ``credentials`` field; this layers on top to catch + the URL+headers+env vectors that the virtual-key sanitizer also + strips. Reset values match each field's declared default on + ``LiteLLM_MCPServerTable`` (``None`` for Optional fields, + ``[]``/``{}`` for required list/dict fields). + """ + sanitized = _redact_mcp_credentials(mcp_server) + # URL is the highest-impact vector: many MCP integrations embed + # the upstream API key directly in the path. spec_path can carry + # similar tokens in the OpenAPI spec URL. + sanitized.url = None + sanitized.spec_path = None + sanitized.static_headers = None + sanitized.extra_headers = [] + sanitized.env = {} + sanitized.authorization_url = None + sanitized.token_url = None + sanitized.registration_url = None + return sanitized + + def _sanitize_mcp_server_list_for_non_admin( + mcp_servers: Iterable[LiteLLM_MCPServerTable], + ) -> List[LiteLLM_MCPServerTable]: + return [_sanitize_mcp_server_for_non_admin(s) for s in mcp_servers] + def _sanitize_mcp_server_for_virtual_key( mcp_server: LiteLLM_MCPServerTable, ) -> LiteLLM_MCPServerTable: @@ -926,6 +960,12 @@ if MCP_AVAILABLE: if is_restricted_virtual_key: return _sanitize_mcp_server_list_for_virtual_key(redacted_mcp_servers) + # Non-admin authenticated users may see the server inventory but + # not credential-bearing fields like `url` (often contains bearer + # tokens) or headers/env (often contain Authorization). + if not _user_has_admin_view(user_api_key_dict): + return _sanitize_mcp_server_list_for_non_admin(redacted_mcp_servers) + return redacted_mcp_servers @router.get( @@ -1293,6 +1333,8 @@ if MCP_AVAILABLE: redacted = _redact_mcp_credentials(mcp_server) if is_restricted_virtual_key: return _sanitize_mcp_server_for_virtual_key(redacted) + if not _user_has_admin_view(user_api_key_dict): + return _sanitize_mcp_server_for_non_admin(redacted) return redacted @router.post( 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 821e2002906..ea8ac7730ef 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 @@ -626,14 +626,16 @@ class TestListMCPServers: assert "config_server_allowed" in server_ids assert "config_server_not_allowed" not in server_ids - # Check server details + # Check server details — non-admin viewers must not see the + # raw `url` (it can carry bearer tokens for many MCP + # integrations). Identity fields stay so the UI can list + # the server. for server in result: + assert server.url is None if server.server_id == "db_server_allowed": assert server.alias == "Allowed Gmail MCP" - assert server.url == "https://gmail-mcp.example.com/mcp" elif server.server_id == "config_server_allowed": assert server.alias == "Allowed Zapier MCP" - assert server.url == "https://actions.zapier.com/mcp/sse" @pytest.mark.asyncio async def test_admin_user_with_object_permission_respects_mcp_servers(self): @@ -2897,3 +2899,162 @@ async def test_list_mcp_user_credentials_batch_server_fetch(): assert result[0].alias == "My Server" # expires_at should always be the raw timestamp (not set to None when expired) assert result[0].expires_at == "2099-01-01T00:00:00+00:00" + + +# --------------------------------------------------------------------------- +# VERIA-8: non-admin viewers must not see the raw MCP server URL +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_list_mcp_servers_non_admin_url_redacted(): + """A standard authenticated user (no admin role, not a restricted + virtual key) used to receive the raw `url` field, which can contain + bearer tokens like `https://actions.zapier.com/mcp//sse`. + They must now get the credential-bearing fields stripped.""" + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + fetch_all_mcp_servers, + ) + + user = UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + user_id="alice", + api_key="sk-alice", + # NOT allowed_routes — a normal authenticated user. + ) + + server = generate_mock_mcp_server_db_record( + server_id="zapier-1", + alias="Zapier", + url="https://actions.zapier.com/mcp/SUPER-SECRET-TOKEN/sse", + ) + server.static_headers = {"Authorization": "Bearer SUPER-SECRET-TOKEN"} + server.env = {"API_KEY": "another-secret"} + server.extra_headers = ["Authorization"] + server.authorization_url = "https://oauth.example.com/authorize?token=foo" + server.token_url = "https://oauth.example.com/token" + server.registration_url = "https://oauth.example.com/register" + + mock_manager = MagicMock() + mock_manager.get_all_mcp_servers_unfiltered = AsyncMock(return_value=[server]) + mock_manager.get_all_allowed_mcp_servers = AsyncMock(return_value=[server]) + + with ( + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints._get_user_mcp_management_mode", + return_value="view_all", + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager", + mock_manager, + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw", + return_value=MagicMock(), + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.build_effective_auth_contexts", + AsyncMock(return_value=[user]), + ), + ): + result = await fetch_all_mcp_servers(user_api_key_dict=user) + + assert len(result) == 1 + s = result[0] + # Identity fields stay so the UI can list the server. + assert s.server_id == "zapier-1" + assert s.alias == "Zapier" + # Credential-bearing fields must all be cleared. + assert s.url is None + assert s.static_headers is None + assert s.env == {} + assert s.extra_headers == [] + assert s.authorization_url is None + assert s.token_url is None + assert s.registration_url is None + + +@pytest.mark.asyncio +async def test_list_mcp_servers_admin_keeps_url(): + """Proxy admins must continue to see the raw URL — the redaction + only applies to non-admin viewers.""" + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + fetch_all_mcp_servers, + ) + + admin = UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + user_id="root", + api_key="sk-admin", + ) + + server = generate_mock_mcp_server_db_record( + server_id="zapier-1", + alias="Zapier", + url="https://actions.zapier.com/mcp/SUPER-SECRET-TOKEN/sse", + ) + server.static_headers = {"Authorization": "Bearer SUPER-SECRET-TOKEN"} + + mock_manager = MagicMock() + mock_manager.get_all_mcp_servers_unfiltered = AsyncMock(return_value=[server]) + mock_manager.get_all_allowed_mcp_servers = AsyncMock(return_value=[server]) + + with ( + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints._get_user_mcp_management_mode", + return_value="view_all", + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager", + mock_manager, + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw", + return_value=MagicMock(), + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.build_effective_auth_contexts", + AsyncMock(return_value=[admin]), + ), + ): + result = await fetch_all_mcp_servers(user_api_key_dict=admin) + + assert len(result) == 1 + assert result[0].url == "https://actions.zapier.com/mcp/SUPER-SECRET-TOKEN/sse" + assert result[0].static_headers == {"Authorization": "Bearer SUPER-SECRET-TOKEN"} + + +def test_sanitize_mcp_server_for_non_admin_clears_credential_fields(): + """Direct unit test on the helper for fast feedback.""" + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + _sanitize_mcp_server_for_non_admin, + ) + + server = generate_mock_mcp_server_db_record( + url="https://example.com/mcp/SUPER-SECRET/sse", + ) + server.credentials = {"auth_value": "secret"} + server.static_headers = {"Authorization": "Bearer x"} + server.spec_path = "https://example.com/specs/openapi-with-token.yaml" + server.env = {"API_KEY": "y"} + server.extra_headers = ["Authorization"] + server.authorization_url = "https://idp/authorize" + server.token_url = "https://idp/token" + server.registration_url = "https://idp/register" + + sanitized = _sanitize_mcp_server_for_non_admin(server) + + assert sanitized.credentials is None + assert sanitized.url is None + assert sanitized.spec_path is None + assert sanitized.static_headers is None + assert sanitized.env == {} + assert sanitized.extra_headers == [] + assert sanitized.authorization_url is None + assert sanitized.token_url is None + assert sanitized.registration_url is None + + # Identity / metadata fields are preserved so the UI can list the + # server without exposing secrets. + assert sanitized.server_id == server.server_id + assert sanitized.alias == server.alias diff --git a/ui/litellm-dashboard/src/components/mcp_tools/mcp_server_view.tsx b/ui/litellm-dashboard/src/components/mcp_tools/mcp_server_view.tsx index c70d188bdbb..f9a3d57e952 100644 --- a/ui/litellm-dashboard/src/components/mcp_tools/mcp_server_view.tsx +++ b/ui/litellm-dashboard/src/components/mcp_tools/mcp_server_view.tsx @@ -147,7 +147,12 @@ export const MCPServerView: React.FC = ({ {renderUrlWithToggle(mcpServer.url, showFullUrl)} - {hasToken && ( + {/* Only proxy admins may reveal the raw URL — non-admins + receive a sanitized server object from the backend + with `url=null`, but hide the toggle anyway as + defense-in-depth in case the URL ever leaks back + into the response. */} + {hasToken && isProxyAdmin && (