fix(destination): do not forward secret headers across HTTP redirects

CHANGELOG.md

@@ -0,0 +1,25 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+Format: [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
+
+## [Unreleased]
+
+### Security
+
+- **`DestinationHttpClient` no longer follows HTTP redirects by default.**
+  Destination credentials (auth tokens, API keys, `x-sap-*` headers) pre-baked into the
+  session are now never forwarded to a redirect target implicitly. This prevents credential
+  leakage to a different origin. This is a **behavioral change**: callers that previously
+  relied on transparent redirect following will now receive the 3xx response directly.
+
+  To opt in to same-origin redirect following:
+  ```python
+  http = DestinationHttpClient(dest, redirect_mode=RedirectMode.SAME_ORIGIN)
+  ```
+  `RedirectMode` is now exported from `sap_cloud_sdk.destination`.
+
+- `HttpClient` (internal SAP service client used by Destination, AI Core, and others) also
+  stops following redirects.
+- Passing `allow_redirects=True` in `**kwargs` to `DestinationHttpClient.request()` is now
+  silently ignored. Redirect behavior is controlled exclusively via `redirect_mode`.

src/sap_cloud_sdk/core/protocol/http/client.py

@@ -95,6 +95,7 @@ def _execute(
             method_str,
             f"{base_url}{path}",
             timeout=self._timeout,
+            allow_redirects=False,
             **kwargs,
         )
 

src/sap_cloud_sdk/destination/__init__.py

@@ -52,7 +52,10 @@
     _make_config_factory,
 )
 from sap_cloud_sdk.core.protocol.http import HttpClient, XsuaaAuthProvider
-from sap_cloud_sdk.destination._destination_http_client import DestinationHttpClient
+from sap_cloud_sdk.destination._destination_http_client import (
+    DestinationHttpClient,
+    RedirectMode,
+)
 from sap_cloud_sdk.destination.client import DestinationClient
 from sap_cloud_sdk.destination.fragment_client import FragmentClient
 from sap_cloud_sdk.destination.certificate_client import CertificateClient
@@ -260,6 +263,7 @@ def create_certificate_client(
     "LocalDevFragmentClient",
     "LocalDevCertificateClient",
     "DestinationHttpClient",
+    "RedirectMode",
     # Exceptions
     "DestinationError",
     "ClientCreationError",

src/sap_cloud_sdk/destination/_destination_http_client.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import ssl
+from enum import Enum
 from typing import Any, Dict, Optional
+from urllib.parse import urljoin, urlparse
 
 import requests
 from requests import Response
@@ -12,6 +14,35 @@
 from sap_cloud_sdk.destination._cert_loader import build_client_cert_context
 from sap_cloud_sdk.destination._models import Destination, DestinationType
 
+_REDIRECT_STATUS_CODES = frozenset({301, 302, 303, 307, 308})
+_MAX_REDIRECT_HOPS = 5
+
+
+class RedirectMode(Enum):
+    """Controls whether ``DestinationHttpClient`` follows HTTP redirects.
+
+    Attributes:
+        NONE: Never follow redirects (default). The 3xx response is returned
+            as-is. Destination credentials are never forwarded beyond the
+            configured origin.
+        SAME_ORIGIN: Follow redirects only when the ``Location`` URL resolves to
+            the same scheme, host, and port as the destination URL. Cross-origin
+            redirects, port changes, and HTTPS-to-HTTP downgrades are rejected
+            (the redirect response is returned). Hop count is bounded at 5.
+    """
+
+    NONE = "none"
+    SAME_ORIGIN = "same_origin"
+
+
+def _parse_origin(url: str) -> tuple[str, str, int]:
+    """Return the normalized (scheme, host, effective_port) triple for *url*."""
+    p = urlparse(url)
+    scheme = p.scheme.lower()
+    host = (p.hostname or "").lower()
+    port = p.port if p.port is not None else (443 if scheme == "https" else 80)
+    return (scheme, host, port)
+
 
 class _ClientCertAdapter(HTTPAdapter):
     """requests HTTPAdapter that injects a stdlib SSLContext for mTLS."""
@@ -36,13 +67,23 @@ class DestinationHttpClient:
     sap-language), URL.headers.* properties, and auth tokens. Certificates from the
     destination's certificate list are mounted into the session.
 
+    Redirects are **not** followed by default (``RedirectMode.NONE``). This
+    prevents destination credentials pre-baked in the session from being
+    forwarded to a different origin. Use ``redirect_mode=RedirectMode.SAME_ORIGIN``
+    to allow same-origin relative redirects.
+
     Use as a context manager to ensure the underlying session is closed:
 
         with DestinationHttpClient(dest) as http:
             response = http.request("GET", "/api/resource")
     """
 
-    def __init__(self, destination: Destination) -> None:
+    def __init__(
+        self,
+        destination: Destination,
+        *,
+        redirect_mode: RedirectMode = RedirectMode.NONE,
+    ) -> None:
         if destination.type != DestinationType.HTTP:
             raise ValueError(
                 f"DestinationHttpClient only supports HTTP destinations, got: {destination.type}"
@@ -51,6 +92,8 @@ def __init__(self, destination: Destination) -> None:
         self._session = requests.Session()
         self._session.headers.update(destination.get_headers())
         self._base_url = destination.url.rstrip("/") if destination.url else ""
+        self._redirect_mode = redirect_mode
+        self._origin = _parse_origin(self._base_url)
 
         ssl_ctx = build_client_cert_context(destination)
         if ssl_ctx is not None:
@@ -68,27 +111,90 @@ def request(
     ) -> Response:
         """Send an HTTP request to the target system.
 
+        Redirects are not followed by default. Construct with
+        ``redirect_mode=RedirectMode.SAME_ORIGIN`` to follow same-origin
+        relative redirects. Passing ``allow_redirects`` in ``**kwargs`` has no
+        effect — it is silently removed to prevent accidental credential leakage.
+
         Args:
             method: HTTP verb (GET, POST, PUT, PATCH, DELETE).
             path: Path relative to the destination URL.
             params: Optional query parameters.
             json: Optional JSON body.
             headers: Optional additional headers merged on top of pre-baked ones.
-            **kwargs: Passed through to requests.Session.request.
+            **kwargs: Passed through to requests.Session.request (except
+                ``allow_redirects``, which is always controlled by the
+                ``redirect_mode`` set at construction time).
 
         Returns:
             requests.Response from the target system.
         """
         url = f"{self._base_url}/{path.lstrip('/')}" if path else self._base_url
+        kwargs.pop("allow_redirects", None)
+        if self._redirect_mode is RedirectMode.SAME_ORIGIN:
+            return self._request_following_same_origin(
+                method.upper(), url, params=params, json=json, headers=headers, **kwargs
+            )
         return self._session.request(
             method=method.upper(),
             url=url,
             params=params,
             json=json,
             headers=headers,
+            allow_redirects=False,
             **kwargs,
         )
 
+    def _request_following_same_origin(
+        self,
+        method: str,
+        start_url: str,
+        *,
+        params: Optional[Dict[str, Any]],
+        json: Optional[Any],
+        headers: Optional[Dict[str, str]],
+        **kwargs: Any,
+    ) -> Response:
+        """Follow redirects only when the Location resolves to the same origin.
+
+        Uses a manual hop loop so that ALL credential headers baked into the
+        session are protected — not only Authorization (which requests strips
+        natively). Every hop is validated before the next request is issued.
+        """
+        current_url = start_url
+        for _ in range(_MAX_REDIRECT_HOPS):
+            resp = self._session.request(
+                method=method,
+                url=current_url,
+                params=params,
+                json=json,
+                headers=headers,
+                allow_redirects=False,
+                **kwargs,
+            )
+            if resp.status_code not in _REDIRECT_STATUS_CODES:
+                return resp
+            location = resp.headers.get("Location")
+            if not location:
+                return resp
+            next_url = urljoin(current_url, location)
+            np = urlparse(next_url)
+            # Reject userinfo (e.g. http://user:pass@host/) or malformed authority
+            if "@" in np.netloc or not np.hostname:
+                return resp
+            # Reject HTTPS → HTTP downgrade (defense-in-depth on top of origin check)
+            if urlparse(current_url).scheme == "https" and np.scheme == "http":
+                return resp
+            # Reject any change in scheme, host, or effective port
+            if _parse_origin(next_url) != self._origin:
+                return resp
+            # 303 See Other — always convert to GET and drop the request body
+            if resp.status_code == 303:
+                method, json = "GET", None
+            params = None  # Location already encodes the resolved query string
+            current_url = next_url
+        return resp  # max hops exhausted — return the last redirect response
+
     def __enter__(self) -> "DestinationHttpClient":
         return self
 

src/sap_cloud_sdk/destination/user-guide.md

@@ -471,6 +471,32 @@ response = http.request("GET", "/api/resource", headers={"X-Correlation-ID": "ab
 
 Per-request headers are merged on top of the pre-baked session headers.
 
+### Redirect behavior (security boundary)
+
+`DestinationHttpClient` **does not follow HTTP redirects by default.** If the target system
+responds with a 3xx, the redirect response is returned as-is. This is intentional: credentials
+pre-baked into the session (auth tokens, API keys, `x-sap-*` headers) must never be forwarded
+to a different origin implicitly.
+
+To follow same-host relative redirects, construct the client with `redirect_mode=RedirectMode.SAME_ORIGIN`:
+
+```python
+from sap_cloud_sdk.destination import create_client, DestinationHttpClient, RedirectMode
+
+dest = client.get_destination("my-erp")
+http = DestinationHttpClient(dest, redirect_mode=RedirectMode.SAME_ORIGIN)
+response = http.request("GET", "/api/resource")
+```
+
+`SAME_ORIGIN` follows a redirect only when the `Location` URL resolves to the same scheme,
+host, and port as the destination URL. Cross-origin redirects, port changes, HTTPS-to-HTTP
+downgrades, and `Location` values with userinfo are silently rejected (the redirect response is
+returned). Redirect chains are bounded at 5 hops.
+
+If the target system has permanently moved to a new URL, update the destination in BTP rather
+than relying on redirect following. Passing `allow_redirects=True` in `**kwargs` has no effect
+— it is silently removed.
+
 ### Using `get_headers()` directly
 
 If you manage your own HTTP client, use `dest.get_headers()` to get all derived headers as a plain dict:

tests/core/unit/test_http_client.py

@@ -64,6 +64,12 @@ def test_401_without_auth_provider_is_not_retried(self):
         assert result.status_code == 401
         assert mock_req.call_count == 1
 
+    def test_execute_sets_allow_redirects_false(self):
+        """Redirects must never be followed — credentials could leak to another origin."""
+        client, auth, session = _make_client_with_mock_session()
+        client.request(HttpMethod.GET, "/path")
+        assert session.request.call_args.kwargs.get("allow_redirects") is False
+
     def test_200_response_not_retried(self):
         client, auth, session = _make_client_with_mock_session()
         result = client.request(HttpMethod.GET, "/data")