diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py index 8c4d4e7937e..f23dd75a753 100644 --- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py +++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py @@ -455,11 +455,15 @@ if MCP_AVAILABLE: user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), ): """ - Get all available MCP access groups from the database AND config + Get MCP access groups available to the user. Non-admins only see groups + they have access to via their teams; admins see all groups. """ from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( global_mcp_server_manager, ) + from litellm.proxy.management_helpers.object_permission_utils import ( + get_allowed_mcp_access_groups_for_user, + ) from litellm.proxy.proxy_server import prisma_client access_groups = set() @@ -482,6 +486,14 @@ if MCP_AVAILABLE: except Exception as e: verbose_proxy_logger.debug(f"Error getting MCP access groups: {e}") + # Filter for non-admins: only return groups the user has access to + allowed_groups = await get_allowed_mcp_access_groups_for_user( + user_api_key_dict=user_api_key_dict, + prisma_client=prisma_client, + ) + if allowed_groups is not None: + access_groups = access_groups & allowed_groups + # Convert to sorted list access_groups_list = sorted(list(access_groups)) return {"access_groups": access_groups_list} diff --git a/litellm/proxy/management_helpers/object_permission_utils.py b/litellm/proxy/management_helpers/object_permission_utils.py index 069509a8fd3..d49f62183da 100644 --- a/litellm/proxy/management_helpers/object_permission_utils.py +++ b/litellm/proxy/management_helpers/object_permission_utils.py @@ -5,7 +5,7 @@ organizations, teams, and keys. import json from litellm._uuid import uuid -from typing import Any, Dict, List, Optional, Union +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union from fastapi import HTTPException @@ -13,6 +13,9 @@ from litellm._logging import verbose_proxy_logger from litellm.proxy._types import LiteLLM_TeamTableCachedObj from litellm.proxy.utils import PrismaClient from litellm.litellm_core_utils.safe_json_dumps import safe_dumps + +if TYPE_CHECKING: + from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth @@ -183,6 +186,60 @@ async def _set_object_permission( return data_json +async def get_allowed_mcp_access_groups_for_user( + user_api_key_dict: "UserAPIKeyAuth", + prisma_client: Optional[PrismaClient], +) -> Optional[Set[str]]: + """ + Return the set of MCP access group IDs visible to the user (via their teams or key). + Returns None for admins, meaning they can see all access groups. + + Used to filter the /v1/mcp/access_groups response for non-admin users so the + UI only shows access groups the user can actually assign. + """ + from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view + + if _user_has_admin_view(user_api_key_dict): + return None # Admins see everything + + try: + from litellm.proxy._experimental.mcp_server.ui_session_utils import ( + build_effective_auth_contexts, + ) + from litellm.proxy.auth.auth_checks import get_team_object + from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache + except ImportError: + return set() + + if prisma_client is None or user_api_key_cache is None: + return set() + + allowed_groups: Set[str] = set() + + auth_contexts = await build_effective_auth_contexts(user_api_key_dict) + for auth_context in auth_contexts: + if auth_context.team_id: + try: + team_obj = await get_team_object( + team_id=auth_context.team_id, + prisma_client=prisma_client, + user_api_key_cache=user_api_key_cache, + parent_otel_span=getattr(user_api_key_dict, "parent_otel_span", None), + proxy_logging_obj=proxy_logging_obj, + ) + except HTTPException: + continue + if team_obj and team_obj.object_permission: + groups = team_obj.object_permission.mcp_access_groups or [] + allowed_groups.update(groups) + + if user_api_key_dict.object_permission: + key_groups = user_api_key_dict.object_permission.mcp_access_groups or [] + allowed_groups.update(key_groups) + + return allowed_groups + + async def validate_key_mcp_servers_against_team( object_permission: Optional[Union[Dict, Any]], team_obj: Optional[LiteLLM_TeamTableCachedObj], diff --git a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py index 13cc3983c20..a7e8c0cd7ef 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_key_management_endpoints.py @@ -6621,3 +6621,88 @@ async def test_mcp_validation_key_update_rejects_disallowed_server(monkeypatch): exc = exc_info.value # The 403 HTTPException is re-raised as a ProxyException assert "403" in str(exc) or (hasattr(exc, "code") and str(exc.code) == "403") + + +@pytest.mark.asyncio +async def test_get_allowed_mcp_access_groups_for_user_non_admin(): + """Non-admin user only gets access groups from their teams, not all groups.""" + from unittest.mock import AsyncMock, MagicMock, patch + + from litellm.proxy._types import ( + LiteLLM_ObjectPermissionTable, + LiteLLM_TeamTableCachedObj, + LitellmUserRoles, + ) + from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth + from litellm.proxy.management_helpers.object_permission_utils import ( + get_allowed_mcp_access_groups_for_user, + ) + + user = UserAPIKeyAuth( + user_role=LitellmUserRoles.INTERNAL_USER, + api_key="sk-user", + user_id="u1", + team_id="team-1", + ) + + team_obj = LiteLLM_TeamTableCachedObj( + team_id="team-1", + object_permission=LiteLLM_ObjectPermissionTable( + object_permission_id="perm-1", + mcp_access_groups=["group-a", "group-b"], + ), + ) + + mock_prisma = MagicMock() + mock_cache = MagicMock() + + async def mock_get_team(**kwargs): + return team_obj + + with ( + patch( + "litellm.proxy._experimental.mcp_server.ui_session_utils.build_effective_auth_contexts", + AsyncMock(return_value=[user]), + ), + patch( + "litellm.proxy.auth.auth_checks.get_team_object", + mock_get_team, + ), + patch( + "litellm.proxy.proxy_server.user_api_key_cache", + mock_cache, + ), + patch( + "litellm.proxy.proxy_server.proxy_logging_obj", + MagicMock(), + ), + ): + result = await get_allowed_mcp_access_groups_for_user( + user_api_key_dict=user, + prisma_client=mock_prisma, + ) + + assert result == {"group-a", "group-b"} + + +@pytest.mark.asyncio +async def test_get_allowed_mcp_access_groups_for_user_admin_returns_none(): + """Admin users get None (meaning all groups are allowed) from the helper.""" + from litellm.proxy._types import LitellmUserRoles + from litellm.proxy.auth.user_api_key_auth import UserAPIKeyAuth + from litellm.proxy.management_helpers.object_permission_utils import ( + get_allowed_mcp_access_groups_for_user, + ) + + admin = UserAPIKeyAuth( + user_role=LitellmUserRoles.PROXY_ADMIN, + api_key="sk-admin", + user_id="admin-1", + ) + + result = await get_allowed_mcp_access_groups_for_user( + user_api_key_dict=admin, + prisma_client=MagicMock(), + ) + + assert result is None