[Storage] Strip sensitive auth info on cross-domain redirect

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client.py

@@ -55,6 +55,7 @@
     StorageLoggingPolicy,
     StorageRequestHook,
     StorageResponseHook,
+    StorageSensitiveHeaderCleanupPolicy,
 )
 from .request_handlers import serialize_batch_body, _get_batch_request_delimiter
 from .response_handlers import PartialBatchErrorException, process_storage_error
@@ -331,6 +332,7 @@ def _create_pipeline(
             StorageResponseHook(**kwargs),
             DistributedTracingPolicy(**kwargs),
             HttpLoggingPolicy(**kwargs),
+            StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/base_client_async.py

@@ -39,6 +39,7 @@
     StorageHeadersPolicy,
     StorageHosts,
     StorageRequestHook,
+    StorageSensitiveHeaderCleanupPolicy,
 )
 from .policies_async import (
     AsyncStorageBearerTokenCredentialPolicy,
@@ -145,6 +146,7 @@ def _create_pipeline(
             AsyncStorageResponseHook(**kwargs),
             DistributedTracingPolicy(**kwargs),
             HttpLoggingPolicy(**kwargs),
+            StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore

sdk/storage/azure-storage-blob/azure/storage/blob/_shared/policies.py

@@ -11,7 +11,7 @@
 import uuid
 from io import BytesIO, SEEK_SET, UnsupportedOperation
 from time import time
-from typing import Any, Dict, Optional, TYPE_CHECKING, Union
+from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
 from urllib.parse import (
     parse_qsl,
     urlencode,
@@ -843,3 +843,65 @@ def on_challenge(self, request: "PipelineRequest", response: "PipelineResponse")
         self.authorize_request(request, scope, tenant_id=challenge.tenant_id)
 
         return True
+
+
+class StorageSensitiveHeaderCleanupPolicy(SansIOHTTPPolicy):
+    """A simple policy that cleans up sensitive headers
+
+    :keyword list[str] blocked_redirect_headers: The headers to clean up when redirecting to another domain.
+    :keyword bool disable_redirect_cleanup: Opt out cleaning up sensitive headers when redirecting to another domain.
+    """
+
+    DEFAULT_SENSITIVE_HEADERS = {
+        "Authorization",
+        "x-ms-authorization-auxiliary",
+        "x-ms-copy-source",
+        "x-ms-copy-source-authorization",
+        "x-ms-rename-source",
+    }
+
+    DEFAULT_SENSITIVE_QUERY_PARAMS = {"sig"}
+
+    def __init__(
+        self,  # pylint: disable=unused-argument
+        *,
+        blocked_redirect_headers: Optional[List[str]] = None,
+        blocked_redirect_query_params: Optional[List[str]] = None,
+        disable_redirect_cleanup: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        self._disable_redirect_cleanup = disable_redirect_cleanup
+        self._blocked_redirect_headers = (
+            StorageSensitiveHeaderCleanupPolicy.DEFAULT_SENSITIVE_HEADERS
+            if blocked_redirect_headers is None
+            else blocked_redirect_headers
+        )
+        self._blocked_query_params = (
+            StorageSensitiveHeaderCleanupPolicy.DEFAULT_SENSITIVE_QUERY_PARAMS
+            if blocked_redirect_query_params is None
+            else blocked_redirect_query_params
+        )
+
+    def on_request(self, request: "PipelineRequest") -> None:
+        """This is executed before sending the request to the next policy.
+
+        :param request: The PipelineRequest object.
+        :type request: ~azure.core.pipeline.PipelineRequest
+        """
+        # "insecure_domain_change" is used to indicate that a redirect
+        # has occurred to a different domain. This tells the SensitiveHeaderCleanupPolicy
+        # to clean up sensitive headers.
+        insecure_domain_change = request.context.get("insecure_domain_change", False)
+        if not self._disable_redirect_cleanup and insecure_domain_change:
+            # Clean up request query parameters
+            parsed = urlparse(request.http_request.url)
+            kept = [
+                pair
+                for pair in parsed.query.split("&")
+                if pair and pair.split("=", 1)[0] not in self._blocked_query_params
+            ]
+            request.http_request.url = urlunparse(parsed._replace(query="&".join(kept)))
+
+            # Clean up request headers
+            for header in self._blocked_redirect_headers:
+                request.http_request.headers.pop(header, None)

sdk/storage/azure-storage-blob/tests/test_sensitive_redirect.py

@@ -0,0 +1,63 @@
+from azure.core.pipeline import PipelineRequest, PipelineContext
+from azure.core.rest import HttpRequest
+from azure.storage.blob._shared.policies import StorageSensitiveHeaderCleanupPolicy
+
+
+def make_request(url, headers=None):
+    http_request = HttpRequest("GET", url, headers=headers or {})
+    context = PipelineContext(None)
+    return PipelineRequest(http_request, context)
+
+
+class TestStorageSensitiveHeaderCleanup:
+    SENSITIVE_HEADERS = {
+        "Authorization": "Bearer token",
+        "x-ms-authorization-auxiliary": "Bearer aux-token",
+        "x-ms-copy-source": "https://acct.blob.core.windows.net/c/src?sig=SECRET",
+        "x-ms-copy-source-authorization": "Bearer copy-token",
+        "x-ms-rename-source": "/c/old",
+    }
+
+    def test_storage_sensitive_cleanup_on_redirect(self):
+        headers = dict(self.SENSITIVE_HEADERS)
+        headers["x-ms-meta-keep"] = "ok"
+        request = make_request(
+            "https://acct.blob.core.windows.net/c/b?comp=block&sv=2026-04-06&sig=SECRET",
+            headers=headers,
+        )
+        request.context["insecure_domain_change"] = True
+
+        StorageSensitiveHeaderCleanupPolicy().on_request(request)
+
+        assert "sig=" not in request.http_request.url
+        assert "sv=2026-04-06" in request.http_request.url
+        assert "comp=block" in request.http_request.url
+
+        for header in self.SENSITIVE_HEADERS:
+            assert header not in request.http_request.headers
+
+        assert request.http_request.headers["x-ms-meta-keep"] == "ok"
+
+    def test_no_cleanup_when_no_redirect(self):
+        request = make_request(
+            "https://acct.blob.core.windows.net/c/b?sig=SECRET",
+            headers=dict(self.SENSITIVE_HEADERS),
+        )
+        StorageSensitiveHeaderCleanupPolicy().on_request(request)
+
+        assert "sig=SECRET" in request.http_request.url
+        for header, value in self.SENSITIVE_HEADERS.items():
+            assert request.http_request.headers[header] == value
+
+    def test_no_cleanup_when_disabled(self):
+        request = make_request(
+            "https://acct.blob.core.windows.net/c/b?sig=SECRET",
+            headers=dict(self.SENSITIVE_HEADERS),
+        )
+        request.context["insecure_domain_change"] = True
+
+        StorageSensitiveHeaderCleanupPolicy(disable_redirect_cleanup=True).on_request(request)
+
+        assert "sig=SECRET" in request.http_request.url
+        for header, value in self.SENSITIVE_HEADERS.items():
+            assert request.http_request.headers[header] == value

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/base_client.py

@@ -55,6 +55,7 @@
     StorageLoggingPolicy,
     StorageRequestHook,
     StorageResponseHook,
+    StorageSensitiveHeaderCleanupPolicy,
 )
 from .request_handlers import serialize_batch_body, _get_batch_request_delimiter
 from .response_handlers import PartialBatchErrorException, process_storage_error
@@ -333,6 +334,7 @@ def _create_pipeline(
             StorageResponseHook(**kwargs),
             DistributedTracingPolicy(**kwargs),
             HttpLoggingPolicy(**kwargs),
+            StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/base_client_async.py

@@ -39,6 +39,7 @@
     StorageHeadersPolicy,
     StorageHosts,
     StorageRequestHook,
+    StorageSensitiveHeaderCleanupPolicy,
 )
 from .policies_async import (
     AsyncStorageBearerTokenCredentialPolicy,
@@ -145,6 +146,7 @@ def _create_pipeline(
             AsyncStorageResponseHook(**kwargs),
             DistributedTracingPolicy(**kwargs),
             HttpLoggingPolicy(**kwargs),
+            StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore

sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/policies.py

@@ -11,7 +11,7 @@
 import uuid
 from io import BytesIO, SEEK_SET, UnsupportedOperation
 from time import time
-from typing import Any, Dict, Optional, TYPE_CHECKING, Union
+from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
 from urllib.parse import (
     parse_qsl,
     urlencode,
@@ -843,3 +843,65 @@ def on_challenge(self, request: "PipelineRequest", response: "PipelineResponse")
         self.authorize_request(request, scope, tenant_id=challenge.tenant_id)
 
         return True
+
+
+class StorageSensitiveHeaderCleanupPolicy(SansIOHTTPPolicy):
+    """A simple policy that cleans up sensitive headers
+
+    :keyword list[str] blocked_redirect_headers: The headers to clean up when redirecting to another domain.
+    :keyword bool disable_redirect_cleanup: Opt out cleaning up sensitive headers when redirecting to another domain.
+    """
+
+    DEFAULT_SENSITIVE_HEADERS = {
+        "Authorization",
+        "x-ms-authorization-auxiliary",
+        "x-ms-copy-source",
+        "x-ms-copy-source-authorization",
+        "x-ms-rename-source",
+    }
+
+    DEFAULT_SENSITIVE_QUERY_PARAMS = {"sig"}
+
+    def __init__(
+        self,  # pylint: disable=unused-argument
+        *,
+        blocked_redirect_headers: Optional[List[str]] = None,
+        blocked_redirect_query_params: Optional[List[str]] = None,
+        disable_redirect_cleanup: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        self._disable_redirect_cleanup = disable_redirect_cleanup
+        self._blocked_redirect_headers = (
+            StorageSensitiveHeaderCleanupPolicy.DEFAULT_SENSITIVE_HEADERS
+            if blocked_redirect_headers is None
+            else blocked_redirect_headers
+        )
+        self._blocked_query_params = (
+            StorageSensitiveHeaderCleanupPolicy.DEFAULT_SENSITIVE_QUERY_PARAMS
+            if blocked_redirect_query_params is None
+            else blocked_redirect_query_params
+        )
+
+    def on_request(self, request: "PipelineRequest") -> None:
+        """This is executed before sending the request to the next policy.
+
+        :param request: The PipelineRequest object.
+        :type request: ~azure.core.pipeline.PipelineRequest
+        """
+        # "insecure_domain_change" is used to indicate that a redirect
+        # has occurred to a different domain. This tells the SensitiveHeaderCleanupPolicy
+        # to clean up sensitive headers.
+        insecure_domain_change = request.context.get("insecure_domain_change", False)
+        if not self._disable_redirect_cleanup and insecure_domain_change:
+            # Clean up request query parameters
+            parsed = urlparse(request.http_request.url)
+            kept = [
+                pair
+                for pair in parsed.query.split("&")
+                if pair and pair.split("=", 1)[0] not in self._blocked_query_params
+            ]
+            request.http_request.url = urlunparse(parsed._replace(query="&".join(kept)))
+
+            # Clean up request headers
+            for header in self._blocked_redirect_headers:
+                request.http_request.headers.pop(header, None)

sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/base_client.py

@@ -55,6 +55,7 @@
     StorageLoggingPolicy,
     StorageRequestHook,
     StorageResponseHook,
+    StorageSensitiveHeaderCleanupPolicy,
 )
 from .request_handlers import serialize_batch_body, _get_batch_request_delimiter
 from .response_handlers import PartialBatchErrorException, process_storage_error
@@ -333,6 +334,7 @@ def _create_pipeline(
             StorageResponseHook(**kwargs),
             DistributedTracingPolicy(**kwargs),
             HttpLoggingPolicy(**kwargs),
+            StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore

sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/base_client_async.py

@@ -39,6 +39,7 @@
     StorageHeadersPolicy,
     StorageHosts,
     StorageRequestHook,
+    StorageSensitiveHeaderCleanupPolicy,
 )
 from .policies_async import (
     AsyncStorageBearerTokenCredentialPolicy,
@@ -145,6 +146,7 @@ def _create_pipeline(
             AsyncStorageResponseHook(**kwargs),
             DistributedTracingPolicy(**kwargs),
             HttpLoggingPolicy(**kwargs),
+            StorageSensitiveHeaderCleanupPolicy(**kwargs),
         ]
         if kwargs.get("_additional_pipeline_policies"):
             policies = policies + kwargs.get("_additional_pipeline_policies")  # type: ignore

sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/policies.py

@@ -11,7 +11,7 @@
 import uuid
 from io import BytesIO, SEEK_SET, UnsupportedOperation
 from time import time
-from typing import Any, Dict, Optional, TYPE_CHECKING, Union
+from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union
 from urllib.parse import (
     parse_qsl,
     urlencode,
@@ -843,3 +843,65 @@ def on_challenge(self, request: "PipelineRequest", response: "PipelineResponse")
         self.authorize_request(request, scope, tenant_id=challenge.tenant_id)
 
         return True
+
+
+class StorageSensitiveHeaderCleanupPolicy(SansIOHTTPPolicy):
+    """A simple policy that cleans up sensitive headers
+
+    :keyword list[str] blocked_redirect_headers: The headers to clean up when redirecting to another domain.
+    :keyword bool disable_redirect_cleanup: Opt out cleaning up sensitive headers when redirecting to another domain.
+    """
+
+    DEFAULT_SENSITIVE_HEADERS = {
+        "Authorization",
+        "x-ms-authorization-auxiliary",
+        "x-ms-copy-source",
+        "x-ms-copy-source-authorization",
+        "x-ms-rename-source",
+    }
+
+    DEFAULT_SENSITIVE_QUERY_PARAMS = {"sig"}
+
+    def __init__(
+        self,  # pylint: disable=unused-argument
+        *,
+        blocked_redirect_headers: Optional[List[str]] = None,
+        blocked_redirect_query_params: Optional[List[str]] = None,
+        disable_redirect_cleanup: bool = False,
+        **kwargs: Any,
+    ) -> None:
+        self._disable_redirect_cleanup = disable_redirect_cleanup
+        self._blocked_redirect_headers = (
+            StorageSensitiveHeaderCleanupPolicy.DEFAULT_SENSITIVE_HEADERS
+            if blocked_redirect_headers is None
+            else blocked_redirect_headers
+        )
+        self._blocked_query_params = (
+            StorageSensitiveHeaderCleanupPolicy.DEFAULT_SENSITIVE_QUERY_PARAMS
+            if blocked_redirect_query_params is None
+            else blocked_redirect_query_params
+        )
+
+    def on_request(self, request: "PipelineRequest") -> None:
+        """This is executed before sending the request to the next policy.
+
+        :param request: The PipelineRequest object.
+        :type request: ~azure.core.pipeline.PipelineRequest
+        """
+        # "insecure_domain_change" is used to indicate that a redirect
+        # has occurred to a different domain. This tells the SensitiveHeaderCleanupPolicy
+        # to clean up sensitive headers.
+        insecure_domain_change = request.context.get("insecure_domain_change", False)
+        if not self._disable_redirect_cleanup and insecure_domain_change:
+            # Clean up request query parameters
+            parsed = urlparse(request.http_request.url)
+            kept = [
+                pair
+                for pair in parsed.query.split("&")
+                if pair and pair.split("=", 1)[0] not in self._blocked_query_params
+            ]
+            request.http_request.url = urlunparse(parsed._replace(query="&".join(kept)))
+
+            # Clean up request headers
+            for header in self._blocked_redirect_headers:
+                request.http_request.headers.pop(header, None)