Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -992,42 +992,78 @@ async def _get_allowed_mcp_servers_for_team(
"""
Get allowed MCP servers for a team.

Note: object_permission is automatically loaded by get_team_object() in main auth flow.
Unions two sources:
- Legacy team.object_permission (mcp_servers, mcp_access_groups,
mcp_tool_permissions).
- Unified team.access_group_ids → access_group.access_mcp_server_ids.
Mirrors the model-side pattern in can_team_access_model — the group
is already attached to the team, so the team relationship is itself
the gate (no assigned_team_ids check needed here).
"""
try:
# Get team object permission (already loaded in main auth flow)
object_permissions = await MCPRequestHandler._get_team_object_permission(
user_api_key_auth
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.proxy.auth.auth_checks import (
_get_mcp_server_ids_from_access_groups,
get_team_object,
)
from litellm.proxy.proxy_server import (
prisma_client,
proxy_logging_obj,
user_api_key_cache,
)

if object_permissions is None:
if (
user_api_key_auth is None
or not user_api_key_auth.team_id
or prisma_client is None
):
return []

# Permission entries may be server_ids OR names/aliases — expand to ids.
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
team_obj: Optional[LiteLLM_TeamTable] = await get_team_object(
team_id=user_api_key_auth.team_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=user_api_key_auth.parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
if team_obj is None:
return []

team_access_group_servers = await _get_mcp_server_ids_from_access_groups(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Medium: Missing access-group assignment check

A team admin can write access_group_ids on their team through /team/update; this now grants every MCP server in those groups without confirming the access group’s assigned_team_ids includes user_api_key_auth.team_id. Filter these groups through the access-group assignment metadata before adding access_mcp_server_ids, or restrict team-level access_group_ids changes to proxy-admin-managed assignment paths.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pre-existing. model-side already reads team.access_group_ids without an assigned_team_ids check (auth_checks.py:3146-3162). root cause is /team/update permitting the write.

access_group_ids=team_obj.access_group_ids or [],
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
proxy_logging_obj=proxy_logging_obj,
)
Comment thread
ryan-crabbe-berri marked this conversation as resolved.

object_permissions = team_obj.object_permission
if object_permissions is None:
return list(set(team_access_group_servers))

direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
object_permissions.mcp_servers or []
)

# Get MCP servers from access groups
access_group_servers = (
legacy_access_group_servers = (
await MCPRequestHandler._get_mcp_servers_from_access_groups(
object_permissions.mcp_access_groups or []
)
)

# servers referenced in tool permissions should also be accessible
tool_perm_servers = list(
global_mcp_server_manager.expand_tool_permissions(
object_permissions.mcp_tool_permissions
).keys()
)

# Combine all lists
all_servers = direct_mcp_servers + access_group_servers + tool_perm_servers
all_servers = (
direct_mcp_servers
+ legacy_access_group_servers
+ tool_perm_servers
+ team_access_group_servers
)
return list(set(all_servers))
except Exception as e:
verbose_logger.warning(
Expand Down
Loading
Loading