scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 353

fix(mcp): scope /v1/mcp/access_groups to caller accessible servers (#46912)

Browse source 
- Restrict GET /v1/mcp/access_groups to accessible servers for non-admin callers
- Preserve full catalog view for admin callers
- Add unit tests verifying admin vs non-admin access group scoping
This commit is contained in:
Praising Harris Ratnam 2026-10-18 21:12:36 +05:30
parent ef1cb19351
commit 3e7121706b
2 changed files with 93 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -1112,7 +1112,8 @@ 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 available MCP access groups from the database AND config.
Admins receive the full catalog; non-admin callers are scoped to servers they have access to.
"""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
@ -1121,22 +1122,30 @@ if MCP_AVAILABLE:
access_groups: Final = set()
# Get from config-loaded servers
for server in global_mcp_server_manager.config_mcp_servers.values():
if server.access_groups:
access_groups.update(server.access_groups)
if user_api_key_has_admin_view(user_api_key_dict):
# Get from config-loaded servers
for server in global_mcp_server_manager.config_mcp_servers.values():
if server.access_groups:
access_groups.update(server.access_groups)
# Get from DB
if prisma_client is not None:
try:
mcp_servers: Final[Sequence[prisma_models.LiteLLM_MCPServerTable]] = await MCPServerRepository(
prisma_client
).table.find_many()
for server in mcp_servers:
if hasattr(server, "mcp_access_groups") and server.mcp_access_groups:
access_groups.update(server.mcp_access_groups)
except Exception as e:
verbose_proxy_logger.debug("Error getting MCP access groups: %s", e)
# Get from DB
if prisma_client is not None:
try:
mcp_servers: Final[Sequence[prisma_models.LiteLLM_MCPServerTable]] = await MCPServerRepository(
prisma_client
).table.find_many()
for server in mcp_servers:
if hasattr(server, "mcp_access_groups") and server.mcp_access_groups:
access_groups.update(server.mcp_access_groups)
except Exception as e:
verbose_proxy_logger.debug("Error getting MCP access groups: %s", e)
else:
accessible_servers: Final = await _resolve_accessible_mcp_servers(user_api_key_dict)
for server in accessible_servers:
if hasattr(server, "mcp_access_groups") and server.mcp_access_groups:
access_groups.update(server.mcp_access_groups)
if hasattr(server, "access_groups") and server.access_groups:
access_groups.update(server.access_groups)
# Convert to sorted list
access_groups_list: Final = sorted(list(access_groups))

68
tests/unit/proxy/management_endpoints/test_mcp_management_endpoints.py
View file

@ -11704,3 +11704,71 @@ async def test_mcp_register_application_type_reaches_upstream_or_is_rejected(
assert posted["application_type"] == application_type
finally:
mgmt_endpoints._temporary_mcp_servers.pop(server.server_id, None)
@pytest.mark.asyncio
async def test_get_mcp_access_groups_admin_sees_all_groups() -> None:
admin_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.PROXY_ADMIN)
config_server = MagicMock()
config_server.access_groups = ["config-group-1"]
db_server = MagicMock()
db_server.mcp_access_groups = ["db-group-1"]
mock_manager = MagicMock()
mock_manager.config_mcp_servers = {"server-1": config_server}
mock_repo = MagicMock()
mock_repo.table.find_many = AsyncMock(return_value=[db_server])
with (
patch("litellm.proxy.management_endpoints.mcp_management_endpoints.user_api_key_has_admin_view", return_value=True),
patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager", mock_manager),
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.MCPServerRepository",
return_value=mock_repo,
),
):
result = await mgmt_endpoints.get_mcp_access_groups(user_api_key_dict=admin_auth)
assert result == {"access_groups": ["config-group-1", "db-group-1"]}
@pytest.mark.asyncio
async def test_get_mcp_access_groups_scoped_for_non_admin() -> None:
non_admin_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.INTERNAL_USER)
accessible_server = MagicMock()
accessible_server.mcp_access_groups = ["allowed-group"]
accessible_server.access_groups = None
with (
patch("litellm.proxy.management_endpoints.mcp_management_endpoints.user_api_key_has_admin_view", return_value=False),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._resolve_accessible_mcp_servers",
AsyncMock(return_value=[accessible_server]),
) as mock_resolve,
):
result = await mgmt_endpoints.get_mcp_access_groups(user_api_key_dict=non_admin_auth)
mock_resolve.assert_awaited_once_with(non_admin_auth)
assert result == {"access_groups": ["allowed-group"]}
@pytest.mark.asyncio
async def test_get_mcp_access_groups_non_admin_empty_when_no_accessible_servers() -> None:
non_admin_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.INTERNAL_USER)
with (
patch("litellm.proxy.management_endpoints.mcp_management_endpoints.user_api_key_has_admin_view", return_value=False),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._resolve_accessible_mcp_servers",
AsyncMock(return_value=[]),
),
):
result = await mgmt_endpoints.get_mcp_access_groups(user_api_key_dict=non_admin_auth)
assert result == {"access_groups": []}