From 9c8659e28d8ab119ef5e914cdca58644ee882b83 Mon Sep 17 00:00:00 2001 From: yuneng-jiang Date: Tue, 17 Mar 2026 12:53:26 -0700 Subject: [PATCH] Fix duplicate test class, dead code, stale comment, and orphaned comment - Remove duplicate TestTeamScopedMCPServerAccess class (second definition at line 1248 silently shadowed the first, making those 4 tests unreachable) - Remove duplicate _redact_mcp_credentials_list call (immediate overwrite) - Update stale comment on fetch_all_mcp_servers to reflect that allow_all_keys servers are no longer included in the team-scoped response - Remove orphaned "Set Management Endpoint Metadata Fields" placeholder comment Co-Authored-By: Claude Sonnet 4.6 --- .../key_management_endpoints.py | 2 - .../mcp_management_endpoints.py | 6 +- .../test_mcp_management_endpoints.py | 151 ------------------ 3 files changed, 1 insertion(+), 158 deletions(-) diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index ff4d861b1b2..19ecd483ed1 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -2107,8 +2107,6 @@ async def update_key_fn( llm_router=llm_router, ) - # Set Management Endpoint Metadata Fields - # Validate MCP servers in object_permission against the effective team if data.object_permission is not None: effective_team_obj = team_obj diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py index 22cd6fd141e..4fe8ab3e377 100644 --- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py +++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py @@ -709,7 +709,7 @@ if MCP_AVAILABLE: ``` """ - # If team_id is provided, return team-scoped servers + allow_all_keys servers + # If team_id is provided, return only team-scoped servers (allow_all_keys servers are globally accessible at request time) is_restricted_virtual_key = _is_restricted_virtual_key_request( user_api_key_dict ) @@ -778,10 +778,6 @@ if MCP_AVAILABLE: aggregated_servers.values() ) - redacted_mcp_servers = _redact_mcp_credentials_list( - aggregated_servers.values() - ) - # augment the mcp servers with public status if litellm.public_mcp_servers is not None: for server in redacted_mcp_servers: diff --git a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py index fe07c29306c..eeaeb498328 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py @@ -1245,157 +1245,6 @@ class TestTeamScopedMCPServerAccess: assert "Restricted virtual key" in str(exc_info.value.detail) -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(