[azure-core] Withhold key/SAS credentials on cross-domain redirect

sdk/core/azure-core/CHANGELOG.md

@@ -7,6 +7,10 @@
 - Added `ODataV4Format.as_dict()` to convert parsed OData V4 errors, including nested details and model-valued inner errors, to JSON-compatible dictionaries for generated model serialization, preserving `None` values and empty collections.
 - Added `azure-deprecating` to the default allowed headers list in `HttpLoggingPolicy`, so deprecation notification headers are logged without redaction.
 
+### Bugs Fixed
+
+- `AzureKeyCredentialPolicy` and `AzureSasCredentialPolicy` no longer re-attach their credential (the custom key header, or the SAS query signature) after a cross-domain redirect. Previously the redirect credential cleanup only removed the `Authorization` header family, so a custom-named key header or SAS was re-sent to the redirect target host. The credential is now withheld when `RedirectPolicy` signals an insecure domain change, matching the protection already in place for `Authorization`.
+
 ## 1.41.0 (2026-05-07)
 
 ### Features Added

sdk/core/azure-core/azure/core/pipeline/policies/_authentication.py

@@ -312,6 +312,16 @@ def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None:
         :param request: The request to be modified before sending.
         :type request: ~azure.core.pipeline.PipelineRequest
         """
+        # Do not send the key credential to a cross-domain redirect target: the request is now
+        # bound for a different authority than the one the credential was provisioned for.
+        # "insecure_domain_change" is set by RedirectPolicy on a cross-domain redirect and is the
+        # same signal SensitiveHeaderCleanupPolicy uses to strip the Authorization header family.
+        # Because this credential uses a custom header name, SensitiveHeaderCleanupPolicy does not
+        # know to remove it, so the header set on the pre-redirect request must be removed here
+        # (and not re-added) on the redirected send.
+        if request.context.get("insecure_domain_change", False):
+            request.http_request.headers.pop(self._name, None)
+            return
         request.http_request.headers[self._name] = f"{self._prefix}{self._credential.key}"
 
 
@@ -339,6 +349,11 @@ def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None:
         :param request: The request to be modified before sending.
         :type request: ~azure.core.pipeline.PipelineRequest
         """
+        # Do not re-attach the SAS after a cross-domain redirect; the signature would otherwise be
+        # appended to a URL on a different authority than the one it was issued for. See the note in
+        # AzureKeyCredentialPolicy.on_request about the "insecure_domain_change" signal.
+        if request.context.get("insecure_domain_change", False):
+            return
         url = request.http_request.url
         query = request.http_request.query
         signature = self._credential.signature

sdk/core/azure-core/tests/test_authentication.py

@@ -1220,3 +1220,97 @@ def send(self, request, **kwargs):
     response = pipeline.run(HttpRequest("GET", "https://legitimate.azure.com"))
     assert response.http_response.status_code == 200
     assert transport._request_count == 3
+
+
+def test_key_credential_withheld_after_cross_domain_redirect():
+    """A custom-header key credential (AzureKeyCredentialPolicy) must not be re-sent to the redirect
+    target after a cross-domain redirect. SensitiveHeaderCleanupPolicy only strips the Authorization
+    header family, so the credential policy itself must withhold its custom-named header once
+    RedirectPolicy sets insecure_domain_change."""
+
+    class MockTransport(HttpTransport):
+        def __init__(self):
+            self._request_count = 0
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            pass
+
+        def close(self):
+            pass
+
+        def open(self):
+            pass
+
+        def send(self, request, **kwargs):
+            self._request_count += 1
+            if self._request_count == 1:
+                # First request to the original domain — key header present
+                assert request.headers.get("api-key") == "secret-key"
+                response = Response()
+                response.status_code = 307
+                response.headers["location"] = "https://redirect-target.example.invalid"
+                return response
+            # Second request after cross-domain redirect — key header must be withheld
+            assert not request.headers.get("api-key"), "api-key must not be sent to a cross-domain redirect target"
+            response = Response()
+            response.status_code = 200
+            return response
+
+    credential = AzureKeyCredential("secret-key")
+    auth_policy = AzureKeyCredentialPolicy(credential, "api-key")
+    redirect_policy = RedirectPolicy()
+    header_clean_up_policy = SensitiveHeaderCleanupPolicy()
+    transport = MockTransport()
+    pipeline = Pipeline(
+        transport=transport,
+        policies=[redirect_policy, auth_policy, header_clean_up_policy],
+    )
+    response = pipeline.run(HttpRequest("GET", "https://legitimate.azure.com"))
+    assert response.http_response.status_code == 200
+    assert transport._request_count == 2
+
+
+def test_sas_credential_withheld_after_cross_domain_redirect():
+    """AzureSasCredentialPolicy must not re-append the SAS to the URL after a cross-domain redirect;
+    the signature would otherwise be sent to a different authority than it was issued for."""
+
+    class MockTransport(HttpTransport):
+        def __init__(self):
+            self._request_count = 0
+
+        def __exit__(self, exc_type, exc_val, exc_tb):
+            pass
+
+        def close(self):
+            pass
+
+        def open(self):
+            pass
+
+        def send(self, request, **kwargs):
+            self._request_count += 1
+            if self._request_count == 1:
+                # First request to the original domain — SAS appended to the query
+                assert "sig=secret-sas" in request.url
+                response = Response()
+                response.status_code = 307
+                response.headers["location"] = "https://redirect-target.example.invalid/path"
+                return response
+            # Second request after cross-domain redirect — SAS must not be appended
+            assert "sig=secret-sas" not in request.url, "SAS must not be sent to a cross-domain redirect target"
+            response = Response()
+            response.status_code = 200
+            return response
+
+    credential = AzureSasCredential("sig=secret-sas")
+    auth_policy = AzureSasCredentialPolicy(credential)
+    redirect_policy = RedirectPolicy()
+    header_clean_up_policy = SensitiveHeaderCleanupPolicy()
+    transport = MockTransport()
+    pipeline = Pipeline(
+        transport=transport,
+        policies=[redirect_policy, auth_policy, header_clean_up_policy],
+    )
+    response = pipeline.run(HttpRequest("GET", "https://legitimate.azure.com"))
+    assert response.http_response.status_code == 200
+    assert transport._request_count == 2