Fix the redirect auth issues (#47265)

* Fix the redirect auth issues

* Add CHANGELOG

* Address feedback

* Fix format

Author: rads-1996

sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md

@@ -9,7 +9,8 @@
 - Use `APPLICATIONINSIGHTS_PYTHON_ATTACHTYPE` environment variable in `_is_attach_enabled` to
   reliably detect successful auto-instrumentation attach, with fallback to legacy path-based detection
   ([#46955](https://github.com/Azure/azure-sdk-for-python/pull/46955))
-
+- Safeguard URL redirection issues
+  ([#47265](https://github.com/Azure/azure-sdk-for-python/pull/47265))
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py

@@ -49,6 +49,17 @@
     308,  # Permanent redirect
 )
 
+_ALLOWED_REDIRECT_DOMAIN_SUFFIXES = (
+    ".livediagnostics.monitor.azure.com",
+    ".monitor.azure.com",
+    ".services.visualstudio.com",
+    ".applicationinsights.azure.com",
+    ".monitor.azure.us",
+    ".applicationinsights.azure.us",
+    ".monitor.azure.cn",
+    ".applicationinsights.azure.cn",
+)
+
 _RETRYABLE_STATUS_CODES = (
     401,  # Unauthorized
     403,  # Forbidden

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_policy.py

@@ -8,6 +8,9 @@
 
 from azure.core.pipeline import PipelineResponse, policies
 
+from azure.monitor.opentelemetry.exporter._constants import (
+    _ALLOWED_REDIRECT_DOMAIN_SUFFIXES,
+)
 from azure.monitor.opentelemetry.exporter._quickpulse._constants import (
     _QUICKPULSE_REDIRECT_HEADER_NAME,
 )
@@ -17,19 +20,6 @@
 
 _logger = logging.getLogger(__name__)
 
-# Allowed domain suffixes for QuickPulse redirect targets.
-# Only redirects to these trusted Azure Monitor domains are accepted.
-_ALLOWED_REDIRECT_DOMAIN_SUFFIXES = (
-    ".livediagnostics.monitor.azure.com",
-    ".monitor.azure.com",
-    ".services.visualstudio.com",
-    ".applicationinsights.azure.com",
-    ".monitor.azure.us",
-    ".applicationinsights.azure.us",
-    ".monitor.azure.cn",
-    ".applicationinsights.azure.cn",
-)
-
 
 def _is_redirect_target_allowed(netloc: str) -> bool:
     """Validate that the redirect target host belongs to a known Azure Monitor domain.

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py

@@ -28,6 +28,7 @@
     TelemetryItem,
 )
 from azure.monitor.opentelemetry.exporter._constants import (
+    _ALLOWED_REDIRECT_DOMAIN_SUFFIXES,
     _AZURE_MONITOR_DISTRO_VERSION_ARG,
     _APPLICATIONINSIGHTS_AUTHENTICATION_STRING,
     _INVALID_STATUS_CODES,
@@ -458,10 +459,32 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal
                         else:
                             redirect_has_headers = False
                         if redirect_has_headers and url.scheme and url.netloc:  # pylint: disable=E0606
-                            # Change the host to the new redirected host
-                            self.client._config.host = "{}://{}".format(url.scheme, url.netloc)  # pylint: disable=W0212
-                            # Attempt to export again
-                            result = self._transmit(envelopes, _skip_rate_limit=True)
+                            current_url = urlparse(self.client._config.host)  # pylint: disable=W0212
+                            # Refuse cross-origin redirects so an attacker-controlled
+                            # `Location` header cannot cause the auth policy to attach
+                            # a freshly-signed Authorization header for a foreign host
+                            # on the recursive _transmit call.
+                            if not self._is_same_registered_domain(current_url.netloc, url.netloc):
+                                if not self._is_stats_exporter():
+                                    if self._should_collect_customer_sdkstats():
+                                        track_dropped_items(
+                                            envelopes,
+                                            DropCode.CLIENT_EXCEPTION,
+                                            _exception_categories.CLIENT_EXCEPTION.value,
+                                        )
+                                    logger.error(
+                                        "Refusing cross-origin redirect to %s://%s.",
+                                        url.scheme,
+                                        url.netloc,
+                                    )
+                                result = ExportResult.FAILED_NOT_RETRYABLE
+                            else:
+                                # Change the host to the new redirected host
+                                self.client._config.host = "{}://{}".format(
+                                    url.scheme, url.netloc
+                                )  # pylint: disable=W0212
+                                # Attempt to export again
+                                result = self._transmit(envelopes, _skip_rate_limit=True)
                         else:
                             if not self._is_stats_exporter():
                                 if self._should_collect_customer_sdkstats():
@@ -611,6 +634,46 @@ def _is_stats_exporter(self):
     def _is_customer_sdkstats_exporter(self):
         return getattr(self, "_is_customer_sdkstats", False)
 
+    def _is_same_registered_domain(self, current_netloc: str, redirect_netloc: str) -> bool:
+        """Return True if the redirect target is safe to follow.
+
+        Used to gate redirects so an attacker-controlled ``Location`` header
+        cannot cause the exporter (and its credential-bearing pipeline) to
+        send telemetry and the Authorization header to an unrelated host.
+
+        A redirect is permitted only when the target equals the currently
+        configured host exactly, or when both the current host and the
+        redirect target are under one of the known Azure Monitor ingestion
+        host suffixes (see ``_ALLOWED_REDIRECT_DOMAIN_SUFFIXES``). Customers
+        with a custom (non-Azure) ingestion host will therefore not have
+        server-issued cross-host redirects followed; such deployments should
+        configure their proxy to terminate redirects locally.
+
+        :param str current_netloc: The netloc of the currently-configured ingestion endpoint.
+        :param str redirect_netloc: The netloc of the redirect target from the server's location header.
+        :return: True if the redirect target is considered safe to follow.
+        :rtype: bool
+        """
+
+        def _host(netloc: str) -> str:
+            return netloc.split("@")[-1].split(":")[0].lower().rstrip(".")
+
+        if not current_netloc or not redirect_netloc:
+            return False
+        current_host = _host(current_netloc)
+        redirect_host = _host(redirect_netloc)
+        if not current_host or not redirect_host:
+            return False
+        # Exact host match is always safe.
+        if current_host == redirect_host:
+            return True
+        # Otherwise both hosts must live under the same trusted Azure Monitor
+        # ingestion suffix.
+        for suffix in _ALLOWED_REDIRECT_DOMAIN_SUFFIXES:
+            if current_host.endswith(suffix) and redirect_host.endswith(suffix):
+                return True
+        return False
+
 
 def _is_invalid_code(response_code: Optional[int]) -> bool:
     """Determine if response is a invalid response.

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py

@@ -747,7 +747,11 @@ def test_transmit_http_error_not_retryable(self):
     def test_transmit_http_error_redirect(self):
         response = HttpResponse(None, None)
         response.status_code = 307
-        response.headers = {"location": "https://example.com"}
+        # Redirect target whose host differs from the default ingestion host
+        # (`dc.services.visualstudio.com`) only in the leftmost DNS label, with
+        # the same number of labels and a 3-label shared suffix, so the
+        # cross-origin redirect guard permits it.
+        response.headers = {"location": "https://westus.services.visualstudio.com"}
         prev_redirects = self._base.client._config.redirect_policy.max_redirects
         self._base.client._config.redirect_policy.max_redirects = 2
         prev_host = self._base.client._config.host
@@ -757,10 +761,57 @@ def test_transmit_http_error_redirect(self):
             result = self._base._transmit(self._envelopes_to_export)
             self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE)
             self.assertEqual(post.call_count, 2)
-            self.assertEqual(self._base.client._config.host, "https://example.com")
+            self.assertEqual(
+                self._base.client._config.host,
+                "https://westus.services.visualstudio.com",
+            )
         self._base.client._config.redirect_policy.max_redirects = prev_redirects
         self._base.client._config.host = prev_host
 
+    def test_transmit_http_error_redirect_refuses_cross_origin(self):
+        """A redirect to a different registered domain must be refused so the
+        auth policy does not attach a bearer token for a foreign host on the
+        recursive _transmit call."""
+        response = HttpResponse(None, None)
+        response.status_code = 307
+        response.headers = {"location": "https://attacker.example.com"}
+        prev_host = self._base.client._config.host
+        error = HttpResponseError(response=response)
+        with mock.patch.object(AzureMonitorClient, "track") as post:
+            post.side_effect = error
+            result = self._base._transmit(self._envelopes_to_export)
+            self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE)
+            self.assertEqual(post.call_count, 1)
+            self.assertEqual(self._base.client._config.host, prev_host)
+
+    def test_is_same_registered_domain(self):
+        same = self._base._is_same_registered_domain
+        # Same host (exact match) is always safe, even when not under a
+        # trusted ingestion suffix (e.g. customer-configured custom host).
+        self.assertTrue(same("westus-0.in.applicationinsights.azure.com", "westus-0.in.applicationinsights.azure.com"))
+        self.assertTrue(same("custom-ingestion.example.invalid", "custom-ingestion.example.invalid"))
+        # Both hosts under the same trusted Azure Monitor ingestion suffix
+        # are permitted -- this is the cross-region case.
+        self.assertTrue(same("westus-0.in.applicationinsights.azure.com", "eastus-0.in.applicationinsights.azure.com"))
+        self.assertTrue(same("dc.services.visualstudio.com", "westus.services.visualstudio.com"))
+        self.assertTrue(same("foo.applicationinsights.azure.us", "bar.applicationinsights.azure.us"))
+        # Different registered domain entirely is rejected.
+        self.assertFalse(same("westus-0.in.applicationinsights.azure.com", "attacker.com"))
+        self.assertFalse(same("foo.example.com", "foo.example.org"))
+        # Sibling subdomains under an untrusted parent are rejected -- this
+        # is the cross-origin-leak PoC scenario.
+        self.assertFalse(same("legit-ingestion.example.invalid", "attacker.example.invalid"))
+        self.assertFalse(same("foo.azure.com", "bar.azure.com"))
+        # A trusted host cannot be redirected to a host under an untrusted
+        # suffix (and vice versa).
+        self.assertFalse(same("dc.services.visualstudio.com", "attacker.example.invalid"))
+        self.assertFalse(same("legit-ingestion.example.invalid", "dc.services.visualstudio.com"))
+        # Mixing two different trusted suffixes is rejected.
+        self.assertFalse(same("dc.services.visualstudio.com", "westus-0.in.applicationinsights.azure.com"))
+        # Empty inputs are treated as not-same.
+        self.assertFalse(same("", "applicationinsights.azure.com"))
+        self.assertFalse(same("applicationinsights.azure.com", ""))
+
     def test_transmit_http_error_redirect_missing_headers(self):
         response = HttpResponse(None, None)
         response.status_code = 307

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_rate_limiter.py

@@ -262,7 +262,10 @@ def test_redirect_does_not_double_consume_rate_limiter(self):
 
         # First call raises a 307 redirect, second call succeeds
         mock_response = mock.Mock()
-        mock_response.headers = {"location": "https://redirected.example.com/v2/track"}
+        # Redirect target differing from the default ingestion host
+        # (`dc.services.visualstudio.com`) only in the leftmost DNS label,
+        # so the cross-origin redirect guard permits it.
+        mock_response.headers = {"location": "https://westus.services.visualstudio.com/v2/track"}
         redirect_error = HttpResponseError(
             message="Temporary Redirect",
             response=mock_response,