fix(security): enforce per-user token isolation in gateway proxy

CRITICAL: build_gateway_auth_headers() only used gateway-level default
credentials, ignoring per-user tokens entirely. When User A stored a
per-user API key for a gateway, User B (without their own key) would
use User A's gateway-level credentials — a cross-user token leak.

Add resolve_gateway_auth_headers() that checks per-user credentials
(UserGatewayCredential, then OAuthToken) before falling back to
gateway defaults.

Update all 6 call sites:
- streamablehttp_transport: _proxy_list_tools, _proxy_list_resources,
  _proxy_read_resource
- tool_service: invoke_tool_direct
- resource_service: direct_proxy resource read
- prompt_service: _fetch_gateway_prompt_result

Author: Olivier Gintrand

mcpgateway/services/prompt_service.py

@@ -58,7 +58,7 @@
 from mcpgateway.services.structured_logger import get_structured_logger
 from mcpgateway.services.team_management_service import TeamManagementService
 from mcpgateway.utils.create_slug import slugify
-from mcpgateway.utils.gateway_access import build_gateway_auth_headers
+from mcpgateway.utils.gateway_access import build_gateway_auth_headers, resolve_gateway_auth_headers
 from mcpgateway.utils.metrics_common import build_top_performers
 from mcpgateway.utils.pagination import unified_paginate
 from mcpgateway.utils.services_auth import decode_auth
@@ -320,7 +320,10 @@ async def _fetch_gateway_prompt_result(self, prompt: DbPrompt, arguments: Option
             raise PromptError(f"Prompt '{prompt.name}' is gateway-backed but missing gateway metadata")
 
         gateway_url = str(gateway.url)
-        headers = build_gateway_auth_headers(gateway)
+        # Resolve per-user credentials (falls back to gateway defaults)
+        from mcpgateway.db import SessionLocal  # pylint: disable=import-outside-toplevel
+        with SessionLocal() as db:
+            headers = await resolve_gateway_auth_headers(gateway, app_user_email=user_identity, db=db)
         auth_query_params_decrypted: Optional[Dict[str, str]] = None
 
         if getattr(gateway, "auth_type", None) == "query_param" and getattr(gateway, "auth_query_params", None):

mcpgateway/services/resource_service.py

@@ -71,7 +71,7 @@
 from mcpgateway.services.oauth_manager import OAuthManager
 from mcpgateway.services.observability_service import current_trace_id, ObservabilityService
 from mcpgateway.services.structured_logger import get_structured_logger
-from mcpgateway.utils.gateway_access import build_gateway_auth_headers, check_gateway_access
+from mcpgateway.utils.gateway_access import build_gateway_auth_headers, check_gateway_access, resolve_gateway_auth_headers
 from mcpgateway.utils.metrics_common import build_top_performers
 from mcpgateway.utils.pagination import unified_paginate
 from mcpgateway.utils.services_auth import decode_auth
@@ -2306,8 +2306,8 @@ async def read_resource(
 
                             gateway = resource_db.gateway
 
-                            # Prepare headers with gateway auth
-                            headers = build_gateway_auth_headers(gateway)
+                            # Prepare headers with per-user credentials (falls back to gateway defaults)
+                            headers = await resolve_gateway_auth_headers(gateway, app_user_email=user, db=db)
 
                             # Use MCP SDK to connect and read resource
                             async with streamablehttp_client(url=gateway.url, headers=headers, timeout=settings.mcpgateway_direct_proxy_timeout) as (read_stream, write_stream, _get_session_id):

mcpgateway/services/tool_service.py

@@ -88,7 +88,7 @@
 from mcpgateway.utils.correlation_id import get_correlation_id
 from mcpgateway.utils.create_slug import slugify
 from mcpgateway.utils.display_name import generate_display_name
-from mcpgateway.utils.gateway_access import build_gateway_auth_headers, check_gateway_access, extract_gateway_id_from_headers
+from mcpgateway.utils.gateway_access import build_gateway_auth_headers, check_gateway_access, extract_gateway_id_from_headers, resolve_gateway_auth_headers
 from mcpgateway.utils.metrics_common import build_top_performers
 from mcpgateway.utils.pagination import decode_cursor, encode_cursor, unified_paginate
 from mcpgateway.utils.passthrough_headers import compute_passthrough_headers_cached
@@ -2985,8 +2985,8 @@ async def invoke_tool_direct(
             if not await check_gateway_access(db, gateway, user_email, token_teams):
                 raise ToolNotFoundError(f"Tool not found: {name}")
 
-            # Prepare headers with gateway auth
-            headers = build_gateway_auth_headers(gateway)
+            # Prepare headers with per-user credentials (falls back to gateway defaults)
+            headers = await resolve_gateway_auth_headers(gateway, app_user_email=user_email, db=db)
 
             # Forward passthrough headers if configured
             if gateway.passthrough_headers and request_headers:

mcpgateway/transports/streamablehttp_transport.py

@@ -78,7 +78,7 @@
 from mcpgateway.services.resource_service import ResourceService
 from mcpgateway.services.tool_service import ToolService
 from mcpgateway.transports.redis_event_store import RedisEventStore
-from mcpgateway.utils.gateway_access import build_gateway_auth_headers, check_gateway_access, extract_gateway_id_from_headers, GATEWAY_ID_HEADER
+from mcpgateway.utils.gateway_access import build_gateway_auth_headers, check_gateway_access, extract_gateway_id_from_headers, GATEWAY_ID_HEADER, resolve_gateway_auth_headers
 from mcpgateway.utils.internal_http import internal_loopback_base_url, internal_loopback_verify
 from mcpgateway.utils.log_sanitizer import sanitize_for_log
 from mcpgateway.utils.orjson_response import ORJSONResponse
@@ -1157,21 +1157,23 @@ async def _validate_streamable_session_access(
     return False, HTTP_403_FORBIDDEN, "Session owner metadata unavailable"
 
 
-async def _proxy_list_tools_to_gateway(gateway: Any, request_headers: dict, user_context: dict, meta: Optional[Any] = None) -> List[types.Tool]:  # pylint: disable=unused-argument
+async def _proxy_list_tools_to_gateway(gateway: Any, request_headers: dict, user_context: dict, meta: Optional[Any] = None) -> List[types.Tool]:
     """Proxy tools/list request directly to remote MCP gateway using MCP SDK.
 
     Args:
         gateway: Gateway ORM instance
         request_headers: Request headers from client
-        user_context: User context (not used - _meta comes from MCP SDK)
+        user_context: User context dict with email for per-user credential lookup
         meta: Request metadata (_meta) from the original request
 
     Returns:
         List of Tool objects from remote server
     """
     try:
-        # Prepare headers with gateway auth
-        headers = build_gateway_auth_headers(gateway)
+        # Prepare headers with per-user credentials (falls back to gateway defaults)
+        user_email = user_context.get("email") if user_context else None
+        with SessionLocal() as db:
+            headers = await resolve_gateway_auth_headers(gateway, app_user_email=user_email, db=db)
 
         # Forward passthrough headers using shared utility (includes X-Upstream-Authorization rename)
         if request_headers:
@@ -1209,21 +1211,23 @@ async def _proxy_list_tools_to_gateway(gateway: Any, request_headers: dict, user
         return []
 
 
-async def _proxy_list_resources_to_gateway(gateway: Any, request_headers: dict, user_context: dict, meta: Optional[Any] = None) -> List[types.Resource]:  # pylint: disable=unused-argument
+async def _proxy_list_resources_to_gateway(gateway: Any, request_headers: dict, user_context: dict, meta: Optional[Any] = None) -> List[types.Resource]:
     """Proxy resources/list request directly to remote MCP gateway using MCP SDK.
 
     Args:
         gateway: Gateway ORM instance
         request_headers: Request headers from client
-        user_context: User context (not used - _meta comes from MCP SDK)
+        user_context: User context dict with email for per-user credential lookup
         meta: Request metadata (_meta) from the original request
 
     Returns:
         List of Resource objects from remote server
     """
     try:
-        # Prepare headers with gateway auth
-        headers = build_gateway_auth_headers(gateway)
+        # Prepare headers with per-user credentials (falls back to gateway defaults)
+        user_email = user_context.get("email") if user_context else None
+        with SessionLocal() as db:
+            headers = await resolve_gateway_auth_headers(gateway, app_user_email=user_email, db=db)
 
         # Forward passthrough headers using shared utility (includes X-Upstream-Authorization rename)
         if request_headers:
@@ -1267,21 +1271,23 @@ async def _proxy_list_resources_to_gateway(gateway: Any, request_headers: dict,
         return []
 
 
-async def _proxy_read_resource_to_gateway(gateway: Any, resource_uri: str, user_context: dict, meta: Optional[Any] = None) -> List[Any]:  # pylint: disable=unused-argument
+async def _proxy_read_resource_to_gateway(gateway: Any, resource_uri: str, user_context: dict, meta: Optional[Any] = None) -> List[Any]:
     """Proxy resources/read request directly to remote MCP gateway using MCP SDK.
 
     Args:
         gateway: Gateway ORM instance
         resource_uri: URI of the resource to read
-        user_context: User context (not used - auth comes from gateway config)
+        user_context: User context dict with email for per-user credential lookup
         meta: Request metadata (_meta) from the original request
 
     Returns:
         List of content objects (TextResourceContents or BlobResourceContents) from remote server
     """
     try:
-        # Prepare headers with gateway auth
-        headers = build_gateway_auth_headers(gateway)
+        # Prepare headers with per-user credentials (falls back to gateway defaults)
+        user_email = user_context.get("email") if user_context else None
+        with SessionLocal() as db:
+            headers = await resolve_gateway_auth_headers(gateway, app_user_email=user_email, db=db)
 
         # Get request headers
         request_headers = request_headers_var.get()

mcpgateway/utils/gateway_access.py

@@ -10,6 +10,7 @@
 """
 
 # Standard
+import logging
 from typing import Dict, List, Optional
 
 # Third-Party
@@ -19,6 +20,8 @@
 from mcpgateway.db import Gateway as DbGateway
 from mcpgateway.utils.services_auth import decode_auth
 
+logger = logging.getLogger(__name__)
+
 # Header name used by clients to target a specific gateway for direct_proxy mode.
 # Defined once here to avoid string literal repetition across the codebase.
 GATEWAY_ID_HEADER = "X-Context-Forge-Gateway-Id"
@@ -159,3 +162,71 @@ def build_gateway_auth_headers(gateway: DbGateway) -> Dict[str, str]:
                 headers["Authorization"] = auth_header
 
     return headers
+
+
+async def resolve_gateway_auth_headers(
+    gateway: DbGateway,
+    app_user_email: Optional[str] = None,
+    db: Optional[Session] = None,
+) -> Dict[str, str]:
+    """Resolve auth headers for gateway requests, preferring per-user credentials.
+
+    Security-critical: ensures per-user credential isolation. When a user has
+    stored personal credentials (API key, bearer token, basic auth) or OAuth
+    tokens for a gateway, those MUST be used instead of the gateway defaults.
+
+    Lookup order:
+        1. Per-user personal credential (UserGatewayCredential table)
+        2. Per-user OAuth token (OAuthToken table)
+        3. Gateway-level default credentials (gateway.auth_value)
+
+    Args:
+        gateway: Gateway ORM object with auth configuration.
+        app_user_email: Email of the requesting user (None skips per-user lookup).
+        db: Database session for per-user credential lookup.
+
+    Returns:
+        Dictionary of HTTP headers with Authorization header.
+    """
+    if app_user_email and db:
+        # 1. Check per-user personal credentials (API keys, bearer tokens, basic auth)
+        try:
+            from mcpgateway.services.credential_storage_service import CredentialStorageService  # pylint: disable=import-outside-toplevel
+
+            cred_service = CredentialStorageService(db)
+            record = await cred_service.get_credential_record(gateway.id, app_user_email)
+            if record:
+                decrypted = await cred_service.get_credential(gateway.id, app_user_email)
+                if decrypted:
+                    if record.credential_type in ("bearer_token", "api_key"):
+                        logger.debug(
+                            "Using per-user %s credential for gateway %s, user %s",
+                            record.credential_type, gateway.id, app_user_email,
+                        )
+                        return {"Authorization": f"Bearer {decrypted}"}
+                    elif record.credential_type == "basic_auth":
+                        logger.debug(
+                            "Using per-user basic_auth credential for gateway %s, user %s",
+                            gateway.id, app_user_email,
+                        )
+                        return {"Authorization": f"Basic {decrypted}"}
+        except Exception:
+            logger.exception("Error looking up per-user credential for gateway %s", gateway.id)
+
+        # 2. Check per-user OAuth tokens
+        try:
+            from mcpgateway.services.token_storage_service import TokenStorageService  # pylint: disable=import-outside-toplevel
+
+            token_service = TokenStorageService(db)
+            oauth_token = await token_service.get_user_token(gateway.id, app_user_email)
+            if oauth_token:
+                logger.debug(
+                    "Using per-user OAuth token for gateway %s, user %s",
+                    gateway.id, app_user_email,
+                )
+                return {"Authorization": f"Bearer {oauth_token}"}
+        except Exception:
+            logger.exception("Error looking up per-user OAuth token for gateway %s", gateway.id)
+
+    # 3. Fall back to gateway-level default credentials
+    return build_gateway_auth_headers(gateway)