From 0ef0b709fafc81e2b260b7ddc961b9261f9723e0 Mon Sep 17 00:00:00 2001 From: user <70670632+stuxf@users.noreply.github.com> Date: Sun, 10 May 2026 15:23:27 +0000 Subject: [PATCH] fix(mcp): apply IP gate to server-id lookup on OAuth endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _get_cached_temporary_mcp_server_or_404 (used by /server/oauth/{server_id}/{authorize,token,register}) used to do: server = manager.get_mcp_server_by_id(server_id) \ or manager.get_mcp_server_by_name(server_id, client_ip=client_ip) The id-lookup branch returned servers without applying IP gating, so an external caller hitting any of those OAuth endpoints with the UUID of an internal-only server bypassed the IP restriction — the name-lookup fallback was the only one gated. Apply _is_server_accessible_from_ip(server, client_ip) to the id-lookup result before falling through. Add a regression test that mocks the gate to deny external IPs and asserts a 404 when an external request hits the UUID of an internal-only server. --- .../mcp_management_endpoints.py | 22 +++++-- .../test_mcp_management_endpoints.py | 61 +++++++++++++++++++ 2 files changed, 78 insertions(+), 5 deletions(-) diff --git a/litellm/proxy/management_endpoints/mcp_management_endpoints.py b/litellm/proxy/management_endpoints/mcp_management_endpoints.py index abce1597bab..b4d2427707d 100644 --- a/litellm/proxy/management_endpoints/mcp_management_endpoints.py +++ b/litellm/proxy/management_endpoints/mcp_management_endpoints.py @@ -1569,11 +1569,23 @@ if MCP_AVAILABLE: if request else INTERNAL_REQUEST ) - server = global_mcp_server_manager.get_mcp_server_by_id( - server_id - ) or global_mcp_server_manager.get_mcp_server_by_name( - server_id, client_ip=client_ip - ) + # get_mcp_server_by_id alone does not apply IP gating, so an + # external caller hitting /server/oauth/{server_id}/{authorize, + # token,register} with the UUID of an internal-only server would + # bypass the IP restriction. Apply the gate to the id-lookup + # result before falling back to the name lookup (which gates). + server = global_mcp_server_manager.get_mcp_server_by_id(server_id) + if ( + server is not None + and not global_mcp_server_manager._is_server_accessible_from_ip( + server, client_ip + ) + ): + server = None + if server is None: + server = global_mcp_server_manager.get_mcp_server_by_name( + server_id, client_ip=client_ip + ) if server is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, 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 f0909afcbf6..aa28d4212f5 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 @@ -1365,6 +1365,67 @@ class TestTemporaryMCPSessionEndpoints: assert exc_info.value.status_code == 404 + @pytest.mark.asyncio + async def test_get_cached_temporary_mcp_server_id_lookup_applies_ip_gate(self): + """ + /server/oauth/{server_id}/{authorize,token,register} resolves the + server via _get_cached_temporary_mcp_server_or_404, which used to do + get_mcp_server_by_id(server_id) OR get_mcp_server_by_name(...). The + id-lookup branch did not apply IP gating, letting an external caller + hit an internal-only server by UUID. The fix gates the id-lookup + result before falling back to name lookup. + """ + from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( + _InternalRequest, + ) + from litellm.proxy.management_endpoints.mcp_management_endpoints import ( + _get_cached_temporary_mcp_server_or_404, + ) + + internal_only_server = generate_mock_mcp_server_config_record( + server_id="internal-only", name="Internal Only" + ) + internal_only_server.available_on_public_internet = False + admin_auth = generate_mock_user_api_key_auth( + user_role=LitellmUserRoles.PROXY_ADMIN, + ) + + mock_manager = MagicMock() + mock_manager.get_mcp_server_by_id.return_value = internal_only_server + mock_manager.get_mcp_server_by_name.return_value = None + + # External IP: gate denies, both id and name lookups should miss → 404. + def _gate(server, client_ip): + if isinstance(client_ip, _InternalRequest): + return True + return False # external IP, server is internal-only + + mock_manager._is_server_accessible_from_ip.side_effect = _gate + external_request = _make_mock_request(ip="8.8.8.8") + + with ( + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.get_cached_temporary_mcp_server", + return_value=None, + ), + patch( + "litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager", + mock_manager, + ), + ): + with pytest.raises(HTTPException) as exc_info: + await _get_cached_temporary_mcp_server_or_404( + "internal-only", admin_auth, request=external_request + ) + + assert ( + exc_info.value.status_code == 404 + ), "External caller must NOT be able to reach internal-only server by UUID" + # Gate must have been consulted on the id-lookup result. + mock_manager._is_server_accessible_from_ip.assert_any_call( + internal_only_server, "8.8.8.8" + ) + @pytest.mark.asyncio async def test_get_cached_temporary_mcp_server_non_admin_denied(self): """Non-admin without access to the server gets 403, not the server."""