fix(agentgateway): pin MCP tool URLs to gateway origin before attaching the Bearer token

src/sap_cloud_sdk/agentgateway/_customer.py

@@ -52,6 +52,7 @@
     MCPToolFilter,
 )
 from sap_cloud_sdk.agentgateway._token_cache import _TokenCache
+from sap_cloud_sdk.agentgateway._url import _require_same_origin, _validate_mcp_id
 from sap_cloud_sdk.agentgateway.exceptions import (
     AgentGatewaySDKError,
     AgentGatewayServerError,
@@ -627,7 +628,13 @@ def _build_mcp_url(gateway_url: str, ord_id: str, gt_id: str) -> str:
 
     Returns:
         Full MCP server URL.
+
+    Raises:
+        AgentGatewaySDKError: If ord_id or gt_id is empty or contains characters
+            that are not allowed in a URL path segment.
     """
+    _validate_mcp_id(ord_id, "ord_id")
+    _validate_mcp_id(gt_id, "global_tenant_id")
     # Gateway URL may or may not include /v1/mcp
     if "/v1/mcp" in gateway_url:
         return f"{gateway_url}/{ord_id}/{gt_id}"
@@ -783,6 +790,7 @@ async def get_mcp_tools_customer(
 
     async def _guarded(dep: IntegrationDependency) -> list[MCPTool]:
         url = _build_mcp_url(credentials.gateway_url, dep.ord_id, dep.global_tenant_id)
+        _require_same_origin(url, credentials.gateway_url)
         logger.debug(
             "Discovering tools from %s (ord_id=%s, gt_id=%s)",
             url,
@@ -824,6 +832,7 @@ async def call_mcp_tool_customer(
     tool: MCPTool,
     auth_token: str,
     timeout: float,
+    gateway_url: str,
     **kwargs,
 ) -> str:
     """Invoke an MCP tool using customer flow.
@@ -835,11 +844,17 @@ async def call_mcp_tool_customer(
         tool: MCPTool to invoke.
         auth_token: Pre-fetched raw access token for authentication.
         timeout: HTTP timeout in seconds for the MCP server call.
+        gateway_url: Trusted gateway origin; tool.url is pinned to it before the
+            token is attached.
         **kwargs: Tool input parameters.
 
     Returns:
         Tool execution result as string.
+
+    Raises:
+        AgentGatewaySDKError: If tool.url is not pinned to the gateway origin.
     """
+    _require_same_origin(tool.url, gateway_url)
     logger.info("Calling tool '%s' on server '%s'", tool.name, tool.server_name)
 
     async with httpx.AsyncClient(

src/sap_cloud_sdk/agentgateway/_lob.py

@@ -52,6 +52,7 @@
     MCPToolFilter,
 )
 from sap_cloud_sdk.agentgateway._token_cache import _GatewayUrlCache, _TokenCache
+from sap_cloud_sdk.agentgateway._url import _require_same_origin
 from sap_cloud_sdk.agentgateway.exceptions import (
     AgentGatewaySDKError,
     AgentGatewayServerError,
@@ -364,18 +365,31 @@ def _log_mcp_server_error(fragment_name: str, exc: BaseException) -> None:
 
 
 async def list_server_tools(
-    dest_url: str, auth_token: str, fragment_name: str, timeout: float
+    dest_url: str,
+    auth_token: str,
+    fragment_name: str,
+    timeout: float,
+    gateway_url: str,
+    allowed_tool_hosts: tuple[str, ...] = (),
 ) -> list[MCPTool]:
     """List tools from a single MCP server.
 
     Args:
         dest_url: MCP endpoint URL.
         auth_token: Raw access token for the request.
         fragment_name: Fragment name for reference.
+        timeout: HTTP timeout in seconds for the MCP server call.
+        gateway_url: Trusted gateway origin; dest_url is pinned to it before the
+            token is attached.
+        allowed_tool_hosts: Additional permitted hosts (LoB escape hatch).
 
     Returns:
         List of MCPTool objects from this server.
+
+    Raises:
+        AgentGatewaySDKError: If dest_url is not pinned to the gateway origin.
     """
+    _require_same_origin(dest_url, gateway_url, extra_allowed_hosts=allowed_tool_hosts)
     async with httpx.AsyncClient(
         headers={
             "Authorization": f"Bearer {auth_token}",
@@ -415,8 +429,10 @@ async def get_mcp_tools_lob(
     tenant_subdomain: str,
     system_token: str,
     timeout: float,
+    gateway_url: str,
     filter: MCPToolFilter | None = None,
     max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS,
+    allowed_tool_hosts: tuple[str, ...] = (),
 ) -> list[MCPTool]:
     """List all MCP tools using LoB flow (destination-based).
 
@@ -426,10 +442,13 @@ async def get_mcp_tools_lob(
         tenant_subdomain: Tenant subdomain for multi-tenant lookup.
         system_token: Pre-fetched raw system token (from get_system_auth).
         timeout: HTTP timeout in seconds for MCP server calls.
+        gateway_url: Trusted gateway origin; each fragment URL is pinned to it
+            before the token is attached.
         filter: Optional MCPToolFilter narrowing results by tool name or ORD ID.
             If None or empty, all tools are included.
         max_concurrent_tasks: Maximum number of fragment fetches that run
             concurrently. Defaults to 15.
+        allowed_tool_hosts: Additional permitted hosts (LoB escape hatch).
 
     Returns:
         List of MCPTool objects from all MCP servers.
@@ -471,6 +490,17 @@ async def get_mcp_tools_lob(
                 "Fragment '%s' has no URL property — skipping", fragment_name
             )
             continue
+        try:
+            _require_same_origin(
+                mcp_url, gateway_url, extra_allowed_hosts=allowed_tool_hosts
+            )
+        except AgentGatewaySDKError as exc:
+            logger.warning(
+                "Fragment '%s' URL is not pinned to the gateway origin — skipping: %s",
+                fragment_name,
+                exc,
+            )
+            continue
         tasks.append((fragment_name, mcp_url))
 
     # Fetch all fragments concurrently; isolate per-fragment failures
@@ -479,7 +509,12 @@ async def get_mcp_tools_lob(
     async def _guarded_mcp(mcp_url: str, fragment_name: str) -> list[MCPTool]:
         async with semaphore:
             return await list_server_tools(
-                mcp_url, system_token, fragment_name, timeout
+                mcp_url,
+                system_token,
+                fragment_name,
+                timeout,
+                gateway_url,
+                allowed_tool_hosts,
             )
 
     results = await asyncio.gather(
@@ -515,6 +550,8 @@ async def call_mcp_tool_lob(
     tool: MCPTool,
     user_auth_token: str,
     timeout: float,
+    gateway_url: str,
+    allowed_tool_hosts: tuple[str, ...] = (),
     **kwargs,
 ) -> str:
     """Invoke an MCP tool using LoB flow (destination-based).
@@ -525,11 +562,18 @@ async def call_mcp_tool_lob(
         tool: MCPTool object (from list_mcp_tools).
         user_auth_token: Pre-fetched raw user token (from get_user_auth).
         timeout: HTTP timeout in seconds for the MCP server call.
+        gateway_url: Trusted gateway origin; tool.url is pinned to it before the
+            token is attached.
+        allowed_tool_hosts: Additional permitted hosts (LoB escape hatch).
         **kwargs: Tool input parameters.
 
     Returns:
         Tool execution result as string.
+
+    Raises:
+        AgentGatewaySDKError: If tool.url is not pinned to the gateway origin.
     """
+    _require_same_origin(tool.url, gateway_url, extra_allowed_hosts=allowed_tool_hosts)
     async with httpx.AsyncClient(
         headers={
             "Authorization": f"Bearer {user_auth_token}",

src/sap_cloud_sdk/agentgateway/_url.py

@@ -0,0 +1,117 @@
+"""URL validation helpers for the Agent Gateway module.
+
+These guard the MCP tool-invocation path against bearer-token exfiltration:
+the authenticated ``httpx`` client attaches a user/system-scoped token as a
+connection-wide header, so the destination URL must be pinned to the trusted
+gateway origin *before* that client is built. Any off-origin or non-HTTPS
+destination is rejected here, so a poisoned tool/fragment URL can never
+redirect the token to an attacker-controlled host.
+"""
+
+import re
+from urllib.parse import urlparse
+
+from sap_cloud_sdk.agentgateway.exceptions import AgentGatewaySDKError
+
+# ORD IDs and global tenant IDs are interpolated into the MCP URL path. They
+# legitimately contain dots and colons (e.g. "sap.mcpbuilder:apiResource:cost-center:v1"
+# or a UUID), so the stricter DMS id pattern cannot be reused. This allow-list
+# still forbids every URL path metacharacter — "/", "\", "?", "#", "%", "@",
+# whitespace — so a value can neither add path segments nor alter the origin.
+_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9._:-]+$")
+
+
+def _validate_mcp_id(value: str, param_name: str) -> str:
+    """Validate an ID that is interpolated into the MCP URL path.
+
+    Args:
+        value: The ``ord_id`` or ``global_tenant_id`` value.
+        param_name: Name of the parameter, used in the error message.
+
+    Returns:
+        The validated value, unchanged (the gateway routes on the raw ``:``/``.``
+        characters, so the value is not percent-encoded).
+
+    Raises:
+        AgentGatewaySDKError: If *value* is empty/whitespace, contains a URL path
+            metacharacter, or is a ``.``/``..`` traversal segment.
+    """
+    if not value or not value.strip():
+        raise AgentGatewaySDKError(f"{param_name} must not be empty")
+    if value in (".", ".."):
+        raise AgentGatewaySDKError(
+            f"{param_name} must not be a path traversal segment: {value!r}"
+        )
+    if not _SAFE_ID_RE.match(value):
+        raise AgentGatewaySDKError(
+            f"{param_name} contains characters not allowed in a URL path segment: {value!r}"
+        )
+    return value
+
+
+_DEFAULT_PORTS: dict[str, int] = {"https": 443, "http": 80}
+
+
+def _origin_host(
+    parsed_host: str | None, parsed_port: int | None, scheme: str = ""
+) -> str:
+    """Render a ``host[:port]`` key for comparison (host lower-cased).
+
+    The port is omitted when it equals the scheme default (443 for https, 80
+    for http), so ``https://host:443/`` and ``https://host/`` compare equal.
+    """
+    host = (parsed_host or "").lower()
+    if parsed_port is not None and parsed_port != _DEFAULT_PORTS.get(scheme):
+        return f"{host}:{parsed_port}"
+    return host
+
+
+def _require_same_origin(
+    url: str,
+    allowed_origin: str,
+    *,
+    extra_allowed_hosts: tuple[str, ...] = (),
+) -> None:
+    """Pin an MCP destination URL to the trusted gateway origin.
+
+    Call this before building the authenticated ``httpx`` client, so the bearer
+    token is never attached to a request bound for an untrusted host.
+
+    Args:
+        url: The MCP tool/fragment destination URL to validate.
+        allowed_origin: The trusted gateway base URL (from ``AuthResult``/
+            credentials). Its host (and port) define the permitted origin.
+        extra_allowed_hosts: Additional ``host`` or ``host:port`` values that are
+            also permitted (the LoB escape hatch from ``ClientConfig``). Matching
+            is case-insensitive.
+
+    Raises:
+        AgentGatewaySDKError: If *url* is not HTTPS, embeds credentials
+            (userinfo), or its host does not match the allowed origin or an
+            allow-listed host.
+    """
+    parsed = urlparse(url)
+
+    if parsed.scheme != "https":
+        raise AgentGatewaySDKError(
+            f"MCP tool URL must use https, got scheme {parsed.scheme!r} in {url!r}"
+        )
+
+    if parsed.username or parsed.password:
+        raise AgentGatewaySDKError("MCP tool URL must not embed credentials (userinfo)")
+
+    url_host = _origin_host(parsed.hostname, parsed.port, parsed.scheme)
+    if not url_host:
+        raise AgentGatewaySDKError(f"MCP tool URL has no host: {url!r}")
+
+    origin_parsed = urlparse(allowed_origin)
+    origin_host = _origin_host(
+        origin_parsed.hostname, origin_parsed.port, origin_parsed.scheme
+    )
+
+    allowed = {origin_host} | {h.lower() for h in extra_allowed_hosts}
+    if url_host not in allowed:
+        raise AgentGatewaySDKError(
+            f"MCP tool URL host {url_host!r} is not the trusted gateway origin "
+            f"{origin_host!r}"
+        )

src/sap_cloud_sdk/agentgateway/agw_client.py

@@ -464,8 +464,10 @@ async def list_mcp_tools(
                 tenant,
                 auth.access_token,
                 self._config.timeout,
+                auth.gateway_url,
                 filter=filter,
                 max_concurrent_tasks=self._config.max_concurrent_tasks,
+                allowed_tool_hosts=self._config.allowed_tool_hosts,
             )
 
         except AgentGatewaySDKError:
@@ -647,23 +649,36 @@ async def call_mcp_tool(
                 )
 
                 return await call_mcp_tool_customer(
-                    tool, auth.access_token, self._config.timeout, **kwargs
+                    tool,
+                    auth.access_token,
+                    self._config.timeout,
+                    auth.gateway_url,
+                    **kwargs,
                 )
 
             # Check for transparent mode
             if detect_transparent_credentials():
                 logger.debug(_LOG_TRANSPARENT_MODE)
 
                 return await call_mcp_tool_customer(
-                    tool, auth.access_token, self._config.timeout, **kwargs
+                    tool,
+                    auth.access_token,
+                    self._config.timeout,
+                    auth.gateway_url,
+                    **kwargs,
                 )
 
             if not user_token:
                 raise AgentGatewaySDKError(
                     "user_token is required for LoB tool invocation."
                 )
             return await call_mcp_tool_lob(
-                tool, auth.access_token, self._config.timeout, **kwargs
+                tool,
+                auth.access_token,
+                self._config.timeout,
+                auth.gateway_url,
+                self._config.allowed_tool_hosts,
+                **kwargs,
             )
 
         except AgentGatewaySDKError:

src/sap_cloud_sdk/agentgateway/config.py

@@ -26,6 +26,12 @@ class ClientConfig:
         max_concurrent_tasks: Maximum number of MCP and Agent fetches (MCP tool
             discovery or A2A agent card fetches) that run concurrently.
             Defaults to 15.
+        allowed_tool_hosts: Additional hostnames (``host`` or ``host:port``),
+            beyond the gateway origin, permitted as MCP tool/fragment
+            destinations in the LoB flow. Tool URLs are otherwise pinned to the
+            gateway origin before the bearer token is attached. Default empty
+            (strict pin); add a host here only if a legitimate LoB destination
+            is served from a different origin than the discovered gateway URL.
     """
 
     timeout: float = DEFAULT_TIMEOUT_SECONDS
@@ -34,6 +40,7 @@ class ClientConfig:
     max_system_token_cache_size: int = DEFAULT_MAX_SYSTEM_TOKEN_CACHE_SIZE
     max_user_token_cache_size: int = DEFAULT_MAX_USER_TOKEN_CACHE_SIZE
     max_concurrent_tasks: int = DEFAULT_MAX_CONCURRENT_TASKS
+    allowed_tool_hosts: tuple[str, ...] = ()
 
     def __post_init__(self) -> None:
         if self.token_expiry_buffer_seconds >= self.fallback_token_ttl_seconds: