mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
Filter MCP access groups in dropdown for non-admin users
Add get_allowed_mcp_access_groups_for_user() which returns only the access groups visible to a user via their teams and key. Filter the GET /v1/mcp/access_groups response for non-admins so the UI dropdown only shows access groups they can actually assign. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
2868796980
commit
45124c904e
3 changed files with 156 additions and 2 deletions
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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],
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue