fix(passthrough): stop coercing an unset gateway allowlist to [] in the tool cache payload

CHANGELOG.md

@@ -8,6 +8,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- **Global passthrough header inheritance** ([#6963](https://github.com/IBM/mcp-context-forge/issues/6963)) - MCP gateways with an unset allowlist inherit the global allowlist. An explicit empty list still blocks forwarding.
+
 ## [1.1.0] - 2026-10-19 - Modern MCP Protocol, New Redesigned Web UI, Federated Completions, and Catalog Pagination
 
 ### Overview

mcpgateway/services/tool_service.py

@@ -1415,7 +1415,8 @@ def _build_tool_cache_payload(self, tool: DbTool, gateway: Optional[DbGateway])
                 "slug": gateway.slug,
                 "transport": gateway.transport,
                 "capabilities": gateway.capabilities or {},
-                "passthrough_headers": gateway.passthrough_headers or [],
+                # None inherits the global allowlist; [] explicitly blocks forwarding.
+                "passthrough_headers": gateway.passthrough_headers,
                 "auth_type": gateway.auth_type,
                 "ca_certificate": getattr(gateway, "ca_certificate", None),
                 "ca_certificate_sig": getattr(gateway, "ca_certificate_sig", None),

tests/live_gateway/e2e/test_candidate_credential_handshake.py

@@ -141,7 +141,13 @@ def log_message(self, format, *args):  # noqa: A002
 # isolated_gateway fixture  (mirrors test_rest_non_json_output_schema_e2e.py)
 # ---------------------------------------------------------------------------
 @pytest.fixture
-def isolated_gateway(tmp_path: Path) -> Iterator[_GatewayFixture]:
+def gateway_environment() -> dict[str, str]:
+    """Return additional settings for an isolated gateway subprocess."""
+    return {}
+
+
+@pytest.fixture
+def isolated_gateway(tmp_path: Path, gateway_environment: dict[str, str]) -> Iterator[_GatewayFixture]:
     """Start a private, single-tenant gateway subprocess for this test module.
 
     Uses SQLite + a random port so no compose stack is required.
@@ -188,6 +194,8 @@ def isolated_gateway(tmp_path: Path) -> Iterator[_GatewayFixture]:
         "PYTHONUNBUFFERED": "1",
     }
 
+    env.update(gateway_environment)
+
     token = make_test_jwt(
         "admin@example.com",
         is_admin=True,

tests/live_gateway/mcp/test_gateway_header_inheritance.py

@@ -0,0 +1,121 @@
+# -*- coding: utf-8 -*-
+"""Location: ./tests/live_gateway/mcp/test_gateway_header_inheritance.py
+Copyright contributors to the MCP-CONTEXT-FORGE project
+SPDX-License-Identifier: Apache-2.0
+
+Verify gateway header inheritance through real MCP tool calls.
+"""
+
+# Standard
+from http.server import BaseHTTPRequestHandler
+import json
+import threading
+from typing import Any
+
+# Third-Party
+import pytest
+
+# Local
+from tests.helpers.auth import make_test_jwt
+from tests.live_gateway.e2e import test_candidate_credential_handshake as handshake_fixtures
+from tests.live_gateway.plugins._helpers import wait_for_gateway_tools
+
+pytestmark = pytest.mark.e2e
+isolated_gateway = handshake_fixtures.isolated_gateway
+
+
+@pytest.fixture(params=[True, False], ids=["enabled", "disabled"])
+def gateway_environment(request, tmp_path):
+    """Configure global forwarding independently of each gateway allowlist."""
+    return {
+        "TMPDIR": str(tmp_path),
+        "ENABLE_HEADER_PASSTHROUGH": str(request.param).lower(),
+        "PASSTHROUGH_HEADERS_SOURCE": "env",
+        "DEFAULT_PASSTHROUGH_HEADERS": '["X-Tenant-Id"]',
+        "MCP_CLIENT_CONNECT_MODE": "legacy",
+        "GATEWAY_ASYNC_LIFECYCLE_ENABLED": "false",
+    }
+
+
+@pytest.fixture
+def upstream():
+    """Serve a real MCP tool that reports the headers it receives."""
+    calls = []
+
+    class Handler(BaseHTTPRequestHandler):
+        """Implement the MCP methods needed for registration and invocation."""
+
+        def do_POST(self):  # noqa: N802
+            """Handle MCP requests and record tool invocation headers."""
+            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            method = body["method"]
+            if "id" not in body:
+                self.send_response(202)
+                self.end_headers()
+                return
+            result: dict[str, Any]
+            if method == "initialize":
+                result = {"protocolVersion": "2025-11-25", "capabilities": {"tools": {}}, "serverInfo": {"name": "echo", "version": "1"}}
+            elif method == "tools/list":
+                result = {"tools": [{"name": "echo", "description": "Echo headers", "inputSchema": {"type": "object", "properties": {}}}]}
+            elif method == "tools/call":
+                headers = {key.lower(): value for key, value in self.headers.items()}
+                calls.append(headers)
+                result = {"content": [{"type": "text", "text": json.dumps(headers)}], "isError": False}
+            else:
+                result = {}
+            response = json.dumps({"jsonrpc": "2.0", "id": body["id"], "result": result}).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(response)))
+            self.end_headers()
+            self.wfile.write(response)
+
+        def do_GET(self):  # noqa: N802
+            """Reject optional SSE listening on this stateless upstream."""
+            self.send_response(405)
+            self.end_headers()
+
+        def log_message(self, format, *args):
+            """Suppress fixture HTTP logs."""
+
+    server = handshake_fixtures._DualStackThreadingHTTPServer(("::", 0), Handler)
+    worker = threading.Thread(target=server.serve_forever, daemon=True)
+    worker.start()
+    try:
+        yield f"http://localhost:{server.server_port}/mcp", calls
+    finally:
+        server.shutdown()
+        server.server_close()
+        worker.join(timeout=5)
+
+
+@pytest.mark.parametrize("allowlist", [None, [], ["X-Other"]], ids=["inherited", "empty", "explicit"])
+def test_gateway_header_inheritance(isolated_gateway, gateway_environment, upstream, allowlist):
+    """Preserve allowlist semantics across cold and warm tool lookups."""
+    client, _, _, signing_key = isolated_gateway
+    url, calls = upstream
+    response = client.post("/gateways", json={"name": "header-echo", "url": url, "transport": "STREAMABLEHTTP", "passthrough_headers": allowlist})
+    assert response.status_code in (200, 201), response.text
+    tool = wait_for_gateway_tools(client, response.json()["id"])[0]
+    payload = {"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": tool["name"], "arguments": {}}}
+    headers = {"X-Tenant-Id": "tenant-1", "X-Other": "other", "X-Unlisted": "blocked"}
+    enabled = gateway_environment["ENABLE_HEADER_PASSTHROUGH"] == "true"
+    expected = (allowlist if allowlist is not None else ["X-Tenant-Id"]) if enabled else []
+    for _ in range(2):
+        response = client.post("/rpc", json=payload, headers=headers)
+        assert response.status_code == 200, response.text
+        assert "error" not in response.json(), response.text
+        assert not response.json()["result"].get("isError"), response.text
+        forwarded = json.loads(response.json()["result"]["content"][0]["text"])
+        for header, value in headers.items():
+            assert forwarded.get(header.lower()) == (value if header in expected else None)
+    assert len(calls) == 2
+    for authorization in ("", "Bearer " + make_test_jwt("admin@example.com", is_admin=True, teams=None, secret=signing_key, scopes={"permissions": ["tools.read"]})):
+        response = client.post("/rpc", json=payload, headers={**headers, "Authorization": authorization})
+        if authorization:
+            assert response.status_code == 200, response.text
+            assert response.json()["error"]["code"] == -32003, response.text
+        else:
+            assert response.status_code == 401, response.text
+        assert len(calls) == 2

tests/unit/mcpgateway/services/test_tool_service.py

@@ -5283,6 +5283,73 @@ def mock_passthrough(req_headers, base_headers, allowed_headers, gateway_auth_ty
         session_mock.initialize.assert_not_awaited()
         session_mock.call_tool.assert_awaited_once()
 
+    @pytest.mark.parametrize(
+        "gateway_allowlist, expected_forwarded, expected_dropped",
+        [
+            # Unset per-gateway allowlist: the global allowlist applies.
+            (None, ["X-Tenant-Id"], ["X-Other"]),
+            # Explicit empty per-gateway allowlist: still an override that forwards nothing.
+            ([], [], ["X-Tenant-Id", "X-Other"]),
+            # Explicit per-gateway allowlist: replaces the global one.
+            (["X-Other"], ["X-Other"], ["X-Tenant-Id"]),
+        ],
+    )
+    async def test_invoke_tool_mcp_applies_global_passthrough_when_gateway_allowlist_unset(
+        self, tool_service, mock_tool, mock_gateway, test_db, gateway_allowlist, expected_forwarded, expected_dropped
+    ):
+        """Global passthrough headers reach an MCP gateway tool unless the gateway sets its own allowlist.
+
+        Regression: the tool cache payload coerced an unset ``gateway.passthrough_headers`` to ``[]``,
+        which ``compute_passthrough_headers_cached`` reads as an explicit empty per-gateway allowlist,
+        so the global allowlist was never applied to tools served by an MCP gateway.
+        """
+        mock_tool.integration_type = "MCP"
+        mock_tool.request_type = "sse"
+        mock_gateway.auth_value = None
+        mock_gateway.passthrough_headers = gateway_allowlist
+
+        mock_scalar1 = Mock()
+        mock_scalar1.scalar_one_or_none.return_value = mock_tool
+        mock_scalar1.scalars.return_value = mock_scalar1
+        mock_scalar1.all.return_value = [mock_tool]
+        mock_scalar2 = Mock()
+        mock_scalar2.scalar_one_or_none.return_value = mock_gateway
+        mock_scalar3 = Mock()
+        mock_scalar3.scalar_one_or_none.return_value = mock_gateway
+        test_db.execute = Mock(side_effect=[mock_scalar1, mock_scalar2, mock_scalar3])
+
+        session_mock = AsyncMock()
+        session_mock.initialize = AsyncMock()
+        session_mock.call_tool = AsyncMock(return_value=ToolResult(content=[TextContent(type="text", text="ok")]))
+        session_mock.session.call_tool = session_mock.call_tool
+        client_session_cm = AsyncMock()
+        client_session_cm.__aenter__.return_value = session_mock
+        client_session_cm.__aexit__.return_value = AsyncMock()
+        mcp_proxy_client_mock = Mock(return_value=client_session_cm)
+
+        request_headers = {"X-Tenant-Id": "tenant-1", "X-Other": "other", "Authorization": "Bearer test"}
+
+        # First-Party
+        import mcpgateway.utils.passthrough_headers as passthrough_headers_module
+
+        with (
+            patch("mcpgateway.services.tool_service.mcp_proxy_client", mcp_proxy_client_mock),
+            patch("mcpgateway.services.tool_service.decode_auth", return_value={}),
+            patch("mcpgateway.services.tool_service.extract_using_jq", side_effect=lambda data, _filt: data),
+            # The real compute_passthrough_headers_cached runs; only its configuration is pinned.
+            patch("mcpgateway.services.tool_service.global_config_cache.get_passthrough_headers", return_value=["X-Tenant-Id"]),
+            patch.object(passthrough_headers_module.settings, "enable_header_passthrough", True),
+            patch.object(passthrough_headers_module.settings, "enable_overwrite_base_headers", False),
+        ):
+            await tool_service.invoke_tool(test_db, "test_tool", {"param": "value"}, request_headers=request_headers)
+
+        session_mock.call_tool.assert_awaited_once()
+        forwarded = mcp_proxy_client_mock.call_args.kwargs["headers"]
+        for header in expected_forwarded:
+            assert forwarded.get(header) == request_headers[header], f"{header} not forwarded: {forwarded}"
+        for header in expected_dropped:
+            assert header not in forwarded, f"{header} forwarded unexpectedly: {forwarded}"
+
     @pytest.mark.parametrize("gateway_id", ["f90870e0-bbbf-4e3c-80c5-69d72a1e6e21", None])
     @pytest.mark.parametrize("reuse_context", [False, True])
     async def test_invoke_tool_with_plugin_post_invoke_success(self, tool_service, mock_tool, mock_global_config_obj, test_db, gateway_id, reuse_context):
@@ -9051,7 +9118,12 @@ def test_build_tool_cache_payload_and_pydantic_helpers(self):
         assert payload["tool"]["header_mapping"] == {}
         assert "auth_value" not in payload["tool"]
         assert "oauth_config" not in payload["tool"]
-        assert payload["gateway"]["passthrough_headers"] == []
+        # None must survive: compute_passthrough_headers_cached() treats any list as a per-gateway
+        # override of the global allowlist, so an unset value coerced to [] would disable global
+        # passthrough headers for every gateway tool.
+        assert payload["gateway"]["passthrough_headers"] is None
+        gateway.passthrough_headers = ["X-Tenant-Id"]
+        assert service._build_tool_cache_payload(tool, gateway)["gateway"]["passthrough_headers"] == ["X-Tenant-Id"]
         # auth_value is now included in gateway cache payload (required by Gateway Pydantic model)
         assert payload["gateway"]["auth_value"] == "secret"
         assert "oauth_config" not in payload["gateway"]