fix(auth): honour TRUST_PROXY_AUTH for virtual MCP-server HTTP calls (/servers/<id>/mcp)

mcpgateway/transports/streamablehttp_transport.py

@@ -740,12 +740,21 @@ async def streamable_http_auth(scope: Any, receive: Any, send: Any) -> bool:
 
     headers = Headers(scope=scope)
     authorization = headers.get("authorization")
+    proxy_user = headers.get(settings.proxy_user_header) if settings.trust_proxy_auth else None
 
-    token = None
+    # Determine authentication strategy based on settings
+    if not settings.mcp_client_auth_enabled and settings.trust_proxy_auth:
+        # Client auth disabled → allow proxy header
+        if proxy_user:
+            return True  # Trusted proxy supplied user
+
+    # --- Standard JWT authentication flow (client auth enabled) ---
+    token: str | None = None
     if authorization:
         scheme, credentials = get_authorization_scheme_param(authorization)
         if scheme.lower() == "bearer" and credentials:
             token = credentials
+
     try:
         if token is None:
             raise Exception()

tests/unit/mcpgateway/utils/test_proxy_auth.py

@@ -9,7 +9,6 @@
 Tests the new MCP_CLIENT_AUTH_ENABLED and proxy authentication features.
 """
 
-import asyncio
 import pytest
 from unittest.mock import Mock, patch, AsyncMock
 from fastapi import HTTPException, Request
@@ -240,3 +239,50 @@ async def test_websocket_with_proxy_auth(self):
 
             # Should accept connection with proxy auth
             websocket.accept.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_streamable_http_auth_with_proxy_header(self):
+        """streamable_http_auth should allow request when proxy header present and auth disabled."""
+        # from types import SimpleNamespace
+        # from starlette.datastructures import Headers
+        from mcpgateway.transports.streamablehttp_transport import streamable_http_auth
+
+        # Build ASGI scope
+        scope = {
+            "type": "http",
+            "path": "/servers/123/mcp",
+            "headers": [(b"x-authenticated-user", b"proxy-user")],
+        }
+        # Patch settings dynamically
+        with patch("mcpgateway.transports.streamablehttp_transport.settings") as mock_settings:
+            mock_settings.mcp_client_auth_enabled = False
+            mock_settings.trust_proxy_auth = True
+            mock_settings.proxy_user_header = "X-Authenticated-User"
+            mock_settings.jwt_secret_key = "secret"
+            mock_settings.jwt_algorithm = "HS256"
+            mock_settings.auth_required = False
+
+            allowed = await streamable_http_auth(scope, AsyncMock(), AsyncMock())
+            assert allowed is True
+
+    @pytest.mark.asyncio
+    async def test_streamable_http_auth_no_header_denied_when_required(self):
+        """Should deny when proxy header missing and auth_required true."""
+        from mcpgateway.transports.streamablehttp_transport import streamable_http_auth
+        scope = {
+            "type": "http",
+            "path": "/servers/123/mcp",
+            "headers": [],
+        }
+        with patch("mcpgateway.transports.streamablehttp_transport.settings") as mock_settings:
+            mock_settings.mcp_client_auth_enabled = False
+            mock_settings.trust_proxy_auth = True
+            mock_settings.proxy_user_header = "X-Authenticated-User"
+            mock_settings.jwt_secret_key = "secret"
+            mock_settings.jwt_algorithm = "HS256"
+            mock_settings.auth_required = True
+            send = AsyncMock()
+            ok = await streamable_http_auth(scope, AsyncMock(), send)
+            # When denied, function returns False and send called with 401 response
+            assert ok is False
+            assert any(isinstance(call.args[0], dict) and call.args[0].get("status") == 401 for call in send.mock_calls)