fix(direct_proxy): address review feedback for security and consistency

crivetimihai · crivetimihai · commit abca2c677941 · 2026-02-13T18:44:41.000Z
- call_tool: add feature-flag gate (settings.mcpgateway_direct_proxy_enabled)
  matching list_tools/list_resources/read_resource
- call_tool: return error on proxy failure instead of falling through to
  cache mode (fail-closed)
- call_tool: sanitize error message to avoid leaking exception details
- read_resource: return empty string directly on access denial instead of
  raising HTTPException that gets swallowed by outer handler
- read_resource: remove _meta from session.read_resource() call (MCP SDK
  only accepts uri parameter)
- GatewayCreate: change gateway_mode from Optional[str] to str with
  before-validator defaulting None to 'cache' (prevents DB integrity error)
- admin: expose direct_proxy_timeout in Connection Timeouts section
- list_tools docstring: fix refresh_strategy -&gt; gateway_mode
- Fix jq test assertions to match TextContent return type

Signed-off-by: Mihai Criveti &lt;crivetimihai@gmail.com&gt;
diff --git a/mcpgateway/admin.py b/mcpgateway/admin.py
@@ -1406,6 +1406,7 @@ def mask_sensitive(value: Any, key: str) -> Any:
         },
         "Connection Timeouts": {
             "federation_timeout": settings.federation_timeout,  # Gateway/server HTTP request timeout
+            "mcpgateway_direct_proxy_timeout": settings.mcpgateway_direct_proxy_timeout,
         },
         "Transport": {
             "transport_type": settings.transport_type,
diff --git a/mcpgateway/schemas.py b/mcpgateway/schemas.py
@@ -2549,9 +2549,20 @@ class GatewayCreate(BaseModel):
     refresh_interval_seconds: Optional[int] = Field(None, ge=60, description="Per-gateway refresh interval in seconds (minimum 60); uses global default if not set")
 
     # Gateway mode configuration
-    gateway_mode: Optional[str] = Field(
-        default="cache", description="Gateway mode: 'cache' (database caching, default) or 'direct_proxy' (pass-through mode with no caching)", pattern="^(cache|direct_proxy)$"
-    )
+    gateway_mode: str = Field(default="cache", description="Gateway mode: 'cache' (database caching, default) or 'direct_proxy' (pass-through mode with no caching)", pattern="^(cache|direct_proxy)$")
+
+    @field_validator("gateway_mode", mode="before")
+    @classmethod
+    def default_gateway_mode(cls, v: Optional[str]) -> str:
+        """Default gateway_mode to 'cache' when None is provided.
+
+        Args:
+            v: Gateway mode value (may be None).
+
+        Returns:
+            The validated gateway mode string, defaulting to 'cache'.
+        """
+        return v if v is not None else "cache"
 
     @field_validator("tags")
     @classmethod
diff --git a/mcpgateway/services/resource_service.py b/mcpgateway/services/resource_service.py
@@ -2201,13 +2201,8 @@ async def read_resource(
                                     # Skip session initialize for stateless servers
                                     # await session.initialize()
 
-                                    # Read resource with _meta if provided
-                                    read_params = {"uri": uri}
-                                    if meta_data:
-                                        read_params["_meta"] = meta_data
-                                        logger.debug(f"Forwarding _meta to remote gateway: {meta_data}")
-
-                                    result = await session.read_resource(**read_params)
+                                    # Note: MCP SDK read_resource() only accepts uri; _meta is not supported
+                                    result = await session.read_resource(uri=uri)
 
                                     # Convert MCP result to MCP-compliant content models
                                     # result.contents is a list of TextResourceContents or BlobResourceContents
diff --git a/mcpgateway/transports/streamablehttp_transport.py b/mcpgateway/transports/streamablehttp_transport.py
@@ -701,7 +701,7 @@ async def call_tool(name: str, arguments: dict) -> Union[
                 from mcpgateway.db import Gateway as DbGateway  # pylint: disable=import-outside-toplevel
 
                 gateway = check_db.execute(select(DbGateway).where(DbGateway.id == gateway_id_from_header)).scalar_one_or_none()
-                if gateway and getattr(gateway, "gateway_mode", "cache") == "direct_proxy":
+                if gateway and getattr(gateway, "gateway_mode", "cache") == "direct_proxy" and settings.mcpgateway_direct_proxy_enabled:
                     # SECURITY: Check gateway access before allowing direct proxy
                     if not await check_gateway_access(check_db, gateway, user_email, token_teams):
                         logger.warning(f"Access denied to gateway {gateway_id_from_header} in direct_proxy mode for user {user_email}")
@@ -722,7 +722,7 @@ async def call_tool(name: str, arguments: dict) -> Union[
                     )
         except Exception as e:
             logger.error(f"Direct proxy mode failed for gateway {gateway_id_from_header}: {e}")
-            # Fall through to normal mode on error
+            return types.CallToolResult(content=[types.TextContent(type="text", text="Direct proxy tool invocation failed")], isError=True)
 
     # Normal mode: use standard tool invocation with normalization
     app_user_email = get_user_email_from_context()  # Keep for OAuth token selection
@@ -952,7 +952,7 @@ async def list_tools() -> List[types.Tool]:
     """
     Lists all tools available to the MCP Server.
 
-    Supports two modes based on gateway's refresh_strategy:
+    Supports two modes based on gateway's gateway_mode:
     - 'cache': Returns tools from database (default behavior)
     - 'direct_proxy': Proxies the request directly to the remote MCP server
 
@@ -1278,9 +1278,6 @@ async def read_resource(resource_uri: str) -> Union[str, bytes]:
         Union[str, bytes]: The content of the resource as text or binary data.
         Returns empty string on failure or if no content is found.
 
-    Raises:
-        HTTPException: If access is denied to the gateway in direct_proxy mode.
-
     Logs exceptions if any errors occur during reading.
 
     Examples:
@@ -1326,7 +1323,6 @@ async def read_resource(resource_uri: str) -> Union[str, bytes]:
             # If X-Context-Forge-Gateway-Id is provided, check if that gateway is in direct_proxy mode
             if gateway_id:
                 # Third-Party
-                from fastapi import HTTPException  # pylint: disable=import-outside-toplevel
                 from sqlalchemy import select  # pylint: disable=import-outside-toplevel
 
                 # First-Party
@@ -1337,7 +1333,7 @@ async def read_resource(resource_uri: str) -> Union[str, bytes]:
                     # SECURITY: Check gateway access before allowing direct proxy
                     if not await check_gateway_access(db, gateway, user_email, token_teams):
                         logger.warning(f"Access denied to gateway {gateway_id} in direct_proxy mode for user {user_email}")
-                        raise HTTPException(status_code=404, detail="Resource not found")
+                        return ""
 
                     # Direct proxy mode: forward request to remote MCP server
                     # Get _meta from request context if available
diff --git a/tests/unit/mcpgateway/services/test_resource_service.py b/tests/unit/mcpgateway/services/test_resource_service.py
@@ -5822,7 +5822,7 @@ async def mock_streamable_client_error(*_args, **_kwargs):
 
     @pytest.mark.asyncio
     async def test_read_resource_direct_proxy_with_meta(self, resource_service, mock_direct_proxy_resource):
-        """meta_data is forwarded to session.read_resource as _meta."""
+        """meta_data is accepted but not forwarded to session.read_resource (SDK doesn't support _meta)."""
         from contextlib import asynccontextmanager
 
         db = self._make_mock_db(mock_direct_proxy_resource)
@@ -5860,9 +5860,9 @@ async def mock_streamable_client(*_args, **_kwargs):
                 meta_data=meta,
             )
 
+        # MCP SDK read_resource() only accepts uri; _meta is not forwarded
         session_mock.read_resource.assert_awaited_once_with(
             uri="http://example.com/dp-resource",
-            _meta={"request_id": "trace-abc-123"},
         )
 
     @pytest.mark.asyncio
diff --git a/tests/unit/mcpgateway/services/test_tool_service.py b/tests/unit/mcpgateway/services/test_tool_service.py
@@ -199,14 +199,14 @@ def all(self):
         monkeypatch.setattr("mcpgateway.services.tool_service._compile_jq_filter", lambda _f: DummyProgram())
 
         result = extract_using_jq({"a": 1}, ".a")
-        assert result == "Error applying jsonpath filter"
+        assert result == [TextContent(type="text", text="Error applying jsonpath filter")]
 
     def test_extract_using_jq_returns_exception_message(self, monkeypatch):
-        """Exceptions during jq execution should return message string."""
+        """Exceptions during jq execution should return list with TextContent error."""
         monkeypatch.setattr("mcpgateway.services.tool_service._compile_jq_filter", lambda _f: (_ for _ in ()).throw(RuntimeError("boom")))
 
         result = extract_using_jq({"a": 1}, ".a")
-        assert result == "Error applying jsonpath filter: boom"
+        assert result == [TextContent(type="text", text="Error applying jsonpath filter: boom")]
 
     def test_tool_service_plugin_env_override(self, monkeypatch):
         """PLUGINS_ENABLED env flag should override settings."""
diff --git a/tests/unit/mcpgateway/transports/test_streamablehttp_transport.py b/tests/unit/mcpgateway/transports/test_streamablehttp_transport.py
@@ -6797,10 +6797,8 @@ class MockContent:
         assert result == b"Binary data"
 
     @pytest.mark.asyncio
-    async def test_read_resource_direct_proxy_access_denied_raises_404(self):
-        """Test read_resource raises HTTPException 404 when gateway access is denied."""
-        from fastapi import HTTPException
-
+    async def test_read_resource_direct_proxy_access_denied_returns_empty(self):
+        """Test read_resource returns empty string when gateway access is denied."""
         mock_gateway = MagicMock()
         mock_gateway.id = "gw-direct"
         mock_gateway.gateway_mode = "direct_proxy"
@@ -6818,11 +6816,9 @@ async def mock_get_db():
 
         with patch("mcpgateway.transports.streamablehttp_transport.get_db", mock_get_db):
             with patch("mcpgateway.transports.streamablehttp_transport.check_gateway_access", return_value=False):
-                # The HTTPException is raised but caught by outer try/except and logged
-                # So we expect empty string return, not an exception
                 result = await tr.read_resource("file:///test.txt")
 
-        # Access denied returns empty string (exception is caught and logged)
+        # Access denied returns empty string directly (no exception raised)
         assert result == ""
 
     @pytest.mark.asyncio
@@ -6859,6 +6855,12 @@ async def mock_get_db():
 class TestCallToolDirectProxy:
     """Test direct_proxy mode in the call_tool handler."""
 
+    @pytest.fixture(autouse=True)
+    def enable_direct_proxy(self):
+        """Enable direct_proxy feature flag for all tests in this class."""
+        with patch.object(tr.settings, "mcpgateway_direct_proxy_enabled", True):
+            yield
+
     @pytest.mark.asyncio
     async def test_call_tool_direct_proxy_success(self):
         """Test call_tool returns CallToolResult from invoke_tool_direct when
@@ -6938,8 +6940,10 @@ async def mock_get_db():
         assert result.content[0].text == "Tool not found: secret_tool"
 
     @pytest.mark.asyncio
-    async def test_call_tool_direct_proxy_exception_falls_through(self):
-        """Test call_tool falls through to normal mode when invoke_tool_direct raises."""
+    async def test_call_tool_direct_proxy_exception_returns_error(self):
+        """Test call_tool returns error when invoke_tool_direct raises (no fallback to cache mode)."""
+        from mcp import types as mcp_types
+
         mock_gateway = MagicMock()
         mock_gateway.id = "gw-direct"
         mock_gateway.gateway_mode = "direct_proxy"
@@ -6956,20 +6960,6 @@ async def mock_get_db():
         # invoke_tool_direct raises an exception
         mock_invoke_direct = AsyncMock(side_effect=RuntimeError("connection failed"))
 
-        # Normal mode invoke_tool returns a result with content
-        # Attributes must be explicit (not auto-created by MagicMock) so the
-        # content normalization code in call_tool works correctly.
-        mock_content_item = MagicMock(spec=[])
-        mock_content_item.type = "text"
-        mock_content_item.text = "normal result"
-        mock_content_item.annotations = None
-        mock_content_item.meta = None
-        mock_content_item.size = None
-        normal_result = MagicMock(spec=[])
-        normal_result.content = [mock_content_item]
-        normal_result.structuredContent = None
-        mock_invoke_normal = AsyncMock(return_value=normal_result)
-
         tr.server_id_var.set("server-123")
         tr.request_headers_var.set({"x-context-forge-gateway-id": "gw-direct"})
         tr.user_context_var.set({"email": "user@test.com", "teams": ["team1"], "is_admin": False})
@@ -6978,15 +6968,14 @@ async def mock_get_db():
             with patch("mcpgateway.transports.streamablehttp_transport.extract_gateway_id_from_headers", return_value="gw-direct"):
                 with patch("mcpgateway.transports.streamablehttp_transport.check_gateway_access", new_callable=AsyncMock, return_value=True):
                     with patch.object(tr.tool_service, "invoke_tool_direct", mock_invoke_direct):
-                        with patch.object(tr.tool_service, "invoke_tool", mock_invoke_normal):
-                            with patch("mcpgateway.transports.streamablehttp_transport.settings") as mock_settings:
-                                mock_settings.mcpgateway_session_affinity_enabled = False
-                                result = await tr.call_tool("my_tool", {"arg": "value"})
+                        result = await tr.call_tool("my_tool", {"arg": "value"})
 
         # invoke_tool_direct was called and raised
         mock_invoke_direct.assert_awaited_once()
-        # Normal mode invoke_tool was called as fallback
-        mock_invoke_normal.assert_awaited_once()
+        # Should return error result, NOT fall through to normal mode
+        assert isinstance(result, mcp_types.CallToolResult)
+        assert result.isError is True
+        assert result.content[0].text == "Direct proxy tool invocation failed"
 
     @pytest.mark.asyncio
     async def test_call_tool_direct_proxy_gateway_not_direct_proxy_falls_through(self):
@@ -7029,3 +7018,49 @@ async def mock_get_db():
 
         # Normal mode invoke_tool was called since gateway is not direct_proxy
         mock_invoke_normal.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_call_tool_direct_proxy_feature_disabled_falls_through(self):
+        """Test call_tool falls through to normal mode when feature flag is disabled."""
+        mock_gateway = MagicMock()
+        mock_gateway.id = "gw-direct"
+        mock_gateway.gateway_mode = "direct_proxy"
+
+        mock_db = MagicMock()
+        mock_db.execute = MagicMock(
+            return_value=MagicMock(scalar_one_or_none=MagicMock(return_value=mock_gateway))
+        )
+
+        @asynccontextmanager
+        async def mock_get_db():
+            yield mock_db
+
+        mock_content_item = MagicMock(spec=[])
+        mock_content_item.type = "text"
+        mock_content_item.text = "cache result"
+        mock_content_item.annotations = None
+        mock_content_item.meta = None
+        mock_content_item.size = None
+        normal_result = MagicMock(spec=[])
+        normal_result.content = [mock_content_item]
+        normal_result.structuredContent = None
+        mock_invoke_normal = AsyncMock(return_value=normal_result)
+        mock_invoke_direct = AsyncMock()
+
+        tr.server_id_var.set("server-123")
+        tr.request_headers_var.set({"x-context-forge-gateway-id": "gw-direct"})
+        tr.user_context_var.set({"email": "user@test.com", "teams": ["team1"], "is_admin": False})
+
+        with patch("mcpgateway.transports.streamablehttp_transport.get_db", mock_get_db):
+            with patch("mcpgateway.transports.streamablehttp_transport.extract_gateway_id_from_headers", return_value="gw-direct"):
+                with patch.object(tr.tool_service, "invoke_tool_direct", mock_invoke_direct):
+                    with patch.object(tr.tool_service, "invoke_tool", mock_invoke_normal):
+                        with patch("mcpgateway.transports.streamablehttp_transport.settings") as mock_settings:
+                            mock_settings.mcpgateway_direct_proxy_enabled = False
+                            mock_settings.mcpgateway_session_affinity_enabled = False
+                            result = await tr.call_tool("my_tool", {"arg": "value"})
+
+        # Direct proxy was NOT called since feature flag is disabled
+        mock_invoke_direct.assert_not_awaited()
+        # Normal mode was used instead
+        mock_invoke_normal.assert_awaited_once()
