mirror of
https://github.com/tiennm99/litellm.git
synced 2026-08-23 18:25:22 +00:00
Fix cross-team MCP server info disclosure and restricted key bypass
The GET /v1/mcp/server endpoint allowed any authenticated user to pass an arbitrary team_id and enumerate another team's MCP server config. Restricted virtual keys could also use the team_id param to bypass their access limitations. Add team membership check for non-admins and block restricted keys from using the team_id filter. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
yuneng-jiang
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
c362ae5095
commit
860cb17571
@@ -687,7 +687,43 @@ if MCP_AVAILABLE:
|
||||
user_api_key_dict
|
||||
)
|
||||
if team_id is not None and isinstance(team_id, str) and team_id.strip():
|
||||
redacted_mcp_servers = await _get_team_scoped_mcp_server_list(team_id.strip())
|
||||
# Restricted virtual keys must not use the team_id filter to
|
||||
# bypass their own access limitations.
|
||||
if is_restricted_virtual_key:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Restricted virtual keys cannot query team-scoped MCP servers.",
|
||||
)
|
||||
|
||||
# Only proxy admins may query another team's MCP servers.
|
||||
# Non-admins must belong to the requested team.
|
||||
sanitized_team_id = team_id.strip()
|
||||
is_admin = _user_has_admin_view(user_api_key_dict)
|
||||
if not is_admin:
|
||||
from litellm.proxy.auth.auth_checks import get_team_object
|
||||
from litellm.proxy.proxy_server import (
|
||||
prisma_client,
|
||||
user_api_key_cache,
|
||||
)
|
||||
|
||||
team_obj = await get_team_object(
|
||||
team_id=sanitized_team_id,
|
||||
prisma_client=prisma_client,
|
||||
user_api_key_cache=user_api_key_cache,
|
||||
check_db_only=True,
|
||||
)
|
||||
user_in_team = any(
|
||||
m.user_id is not None
|
||||
and m.user_id == user_api_key_dict.user_id
|
||||
for m in team_obj.members_with_roles
|
||||
)
|
||||
if not user_in_team:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="You do not have permission to view MCP servers for this team.",
|
||||
)
|
||||
|
||||
redacted_mcp_servers = await _get_team_scoped_mcp_server_list(sanitized_team_id)
|
||||
else:
|
||||
user_mcp_management_mode = _get_user_mcp_management_mode()
|
||||
|
||||
|
||||
@@ -797,6 +797,157 @@ class TestListMCPServers:
|
||||
assert result.status == "healthy"
|
||||
|
||||
|
||||
class TestTeamScopedMCPServerAccess:
|
||||
"""Tests for cross-team information disclosure and restricted key bypass fixes."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_member_cannot_query_foreign_team(self):
|
||||
"""Non-admin user who is NOT a member of the target team should get 403."""
|
||||
from litellm.proxy._types import Member
|
||||
|
||||
mock_user_auth = generate_mock_user_api_key_auth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER,
|
||||
user_id="attacker_user",
|
||||
)
|
||||
|
||||
# Team with a different member
|
||||
mock_team_obj = MagicMock()
|
||||
mock_team_obj.members_with_roles = [
|
||||
Member(user_id="legitimate_user", role="admin"),
|
||||
]
|
||||
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints._user_has_admin_view",
|
||||
return_value=False,
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.auth.auth_checks.get_team_object",
|
||||
AsyncMock(return_value=mock_team_obj),
|
||||
),
|
||||
):
|
||||
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
|
||||
fetch_all_mcp_servers,
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await fetch_all_mcp_servers(
|
||||
user_api_key_dict=mock_user_auth, team_id="foreign-team-id"
|
||||
)
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "permission" in str(exc_info.value.detail).lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_team_member_can_query_own_team(self):
|
||||
"""User who IS a member of the team should be able to query it."""
|
||||
from litellm.proxy._types import Member
|
||||
|
||||
mock_user_auth = generate_mock_user_api_key_auth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER,
|
||||
user_id="team_member",
|
||||
)
|
||||
|
||||
mock_team_obj = MagicMock()
|
||||
mock_team_obj.members_with_roles = [
|
||||
Member(user_id="team_member", role="user"),
|
||||
]
|
||||
mock_team_obj.object_permission = MagicMock(mcp_servers=["server-1"])
|
||||
|
||||
mock_server = generate_mock_mcp_server_config_record(
|
||||
server_id="server-1", name="Team Server"
|
||||
)
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.get_mcp_server_by_id = MagicMock(return_value=mock_server)
|
||||
mock_manager._build_mcp_server_table = MagicMock(
|
||||
return_value=generate_mock_mcp_server_db_record(server_id="server-1")
|
||||
)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints._user_has_admin_view",
|
||||
return_value=False,
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.auth.auth_checks.get_team_object",
|
||||
AsyncMock(return_value=mock_team_obj),
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_team_scoped_mcp_server_list",
|
||||
AsyncMock(
|
||||
return_value=[
|
||||
generate_mock_mcp_server_db_record(server_id="server-1")
|
||||
]
|
||||
),
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager",
|
||||
mock_manager,
|
||||
),
|
||||
):
|
||||
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
|
||||
fetch_all_mcp_servers,
|
||||
)
|
||||
|
||||
result = await fetch_all_mcp_servers(
|
||||
user_api_key_dict=mock_user_auth, team_id="my-team-id"
|
||||
)
|
||||
assert len(result) == 1
|
||||
assert result[0].server_id == "server-1"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_can_query_any_team(self):
|
||||
"""Proxy admins should be able to query any team's MCP servers."""
|
||||
mock_user_auth = generate_mock_user_api_key_auth(
|
||||
user_role=LitellmUserRoles.PROXY_ADMIN,
|
||||
user_id="admin_user",
|
||||
)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints._user_has_admin_view",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints._get_team_scoped_mcp_server_list",
|
||||
AsyncMock(
|
||||
return_value=[
|
||||
generate_mock_mcp_server_db_record(server_id="server-1")
|
||||
]
|
||||
),
|
||||
),
|
||||
):
|
||||
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
|
||||
fetch_all_mcp_servers,
|
||||
)
|
||||
|
||||
# Admin should NOT need to be a team member
|
||||
result = await fetch_all_mcp_servers(
|
||||
user_api_key_dict=mock_user_auth, team_id="any-team-id"
|
||||
)
|
||||
assert len(result) == 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_restricted_virtual_key_cannot_use_team_id_filter(self):
|
||||
"""Restricted virtual keys must not bypass access limits via team_id."""
|
||||
mock_user_auth = UserAPIKeyAuth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER,
|
||||
user_id="vkey_user",
|
||||
api_key="sk-restricted",
|
||||
allowed_routes=["mcp_routes"],
|
||||
)
|
||||
|
||||
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
|
||||
fetch_all_mcp_servers,
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await fetch_all_mcp_servers(
|
||||
user_api_key_dict=mock_user_auth, team_id="some-team"
|
||||
)
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "Restricted virtual key" in str(exc_info.value.detail)
|
||||
|
||||
|
||||
class TestTemporaryMCPSessionEndpoints:
|
||||
def test_inherit_credentials_from_existing_server(self):
|
||||
payload = NewMCPServerRequest(
|
||||
|
||||
Reference in New Issue
Block a user