Isolate authentication for multi-endpoint exports

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

@@ -8,6 +8,13 @@
 
 ### Bugs Fixed
 
+- Isolate multi-endpoint Microsoft Entra authentication from the host connection
+  string's audience. Configure routed audiences with
+  `APPLICATIONINSIGHTS_MULTI_ENDPOINT_AAD_AUDIENCE`; the default is the public cloud.
+  Reject custom authentication policies in routing mode and refuse credential-bearing,
+  port-changing, or untrusted routed redirects. Preserve the complete path of accepted
+  routed redirects. Single-endpoint behavior is unchanged.
+
 ### Other Changes
 
 ## 1.0.0b59 (2026-11-05)

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

@@ -182,7 +182,10 @@ policy chain for each destination. `SansIOHTTPPolicy` entries remain supported.
 
 Items are grouped by normalized endpoint and authentication mode. Set
 `microsoft.ingestion_auth` to `"aad"` to use the exporter's configured credential;
-other values use key-only authentication. AAD groups without a credential fail.
+the value is trimmed and case-insensitive. Other values use key-only authentication
+and do not receive the exporter's token, even when a credential is configured.
+AAD groups without a credential fail without being sent or stored; other groups
+continue exporting.
 Span events follow their parent span's destination, instrumentation key, and
 authentication mode, not routing attributes on the events themselves.
 Metrics use routing attributes on each data point, not the resource or
@@ -191,6 +194,30 @@ Cloud-role names come from the item's `microsoft.multi_endpoint_cloud_role` attr
 missing, blank, or non-string values use `unknown_service`. Span events inherit their
 parent span's role name. Cloud-role instances remain resource-derived.
 
+#### Microsoft Entra authentication
+
+1. Pass a token credential, such as `DefaultAzureCredential`, through `credential`,
+   or configure managed identity using `APPLICATIONINSIGHTS_AUTHENTICATION_STRING`.
+2. Set `microsoft.ingestion_auth` to `"aad"` on telemetry that requires Microsoft
+   Entra authentication, including resources with local authentication disabled.
+3. Assign the identity the **Monitoring Metrics Publisher** role on each destination resource.
+
+The default audience is Azure public cloud. For another cloud, set the audience
+before constructing the exporter, for example:
+
+```powershell
+$env:APPLICATIONINSIGHTS_MULTI_ENDPOINT_AAD_AUDIENCE = "https://monitor.azure.cn/"
+```
+
+All AAD destinations within an exporter share its credential and audience. Use
+separately configured exporters when identities or clouds differ. The host connection
+string's `AadAudience` does not apply to routed telemetry. Custom
+`authentication_policy` is not supported in routing mode.
+
+> **Security:** Use only trusted routing destinations. Custom headers, policies,
+> hooks, transports, or sessions may send credentials to any routed destination,
+> including key-only routes.
+
 Retry storage is separate from the host storage, partitioned by endpoint and
 authentication mode, and shares the configured storage capacity across routed
 partitions. Pending items preserve their original instrumentation keys and are

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

@@ -37,6 +37,7 @@
 
 _APPLICATIONINSIGHTS_METRICS_TO_LOGANALYTICS_ENABLED = "APPLICATIONINSIGHTS_METRICS_TO_LOGANALYTICS_ENABLED"
 _APPLICATIONINSIGHTS_AUTHENTICATION_STRING = "APPLICATIONINSIGHTS_AUTHENTICATION_STRING"
+_APPLICATIONINSIGHTS_MULTI_ENDPOINT_AAD_AUDIENCE = "APPLICATIONINSIGHTS_MULTI_ENDPOINT_AAD_AUDIENCE"
 
 # RPs
 

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

@@ -10,7 +10,10 @@
 
 from opentelemetry.util.types import Attributes
 
+from azure.core.pipeline import PipelineRequest
+from azure.core.pipeline.policies import SansIOHTTPPolicy
 from azure.monitor.opentelemetry.exporter._constants import (
+    _ALLOWED_REDIRECT_DOMAIN_SUFFIXES,
     _MICROSOFT_INGESTION_AUTH,
     _MICROSOFT_INGESTION_ENDPOINT,
     _MICROSOFT_INSTRUMENTATION_KEY,
@@ -22,6 +25,17 @@
 _logger = logging.getLogger(__name__)
 
 
+class _RoutedRedirectPolicy(SansIOHTTPPolicy):
+    """Preserve the complete validated redirect URL instead of appending the generated operation path."""
+
+    def __init__(self) -> None:
+        self.redirect_url: str | None = None
+
+    def on_request(self, request: PipelineRequest) -> None:
+        if self.redirect_url is not None:
+            request.http_request.url = self.redirect_url
+
+
 @dataclass(frozen=True)
 class _Route:
     """Validated per-item routing metadata, separate from its transport grouping key."""
@@ -126,6 +140,31 @@ def _normalize_endpoint_host(host: str) -> str:
     return host.lower() if host.isascii() else host
 
 
+def _is_trusted_routed_redirect(current_endpoint: str, redirect_endpoint: str) -> bool:
+    """Permit same-origin HTTPS redirects or default-port redirects within one trusted suffix.
+
+    :param str current_endpoint: The current ingestion endpoint.
+    :param str redirect_endpoint: The server's redirect destination.
+    :return: Whether the destination is safe for the routed pipeline.
+    :rtype: bool
+    """
+    try:
+        current = urlsplit(_get_ingestion_endpoint({_MICROSOFT_INGESTION_ENDPOINT: current_endpoint}))
+        redirect = urlsplit(_get_ingestion_endpoint({_MICROSOFT_INGESTION_ENDPOINT: redirect_endpoint}))
+    except ValueError:
+        return False
+    current_host, redirect_host = current.hostname, redirect.hostname
+    if current_host is None or redirect_host is None:
+        return False
+    if current_host == redirect_host:
+        return current.port == redirect.port
+    if current.port is not None or redirect.port is not None:
+        return False
+    return any(
+        current_host.endswith(suffix) and redirect_host.endswith(suffix) for suffix in _ALLOWED_REDIRECT_DOMAIN_SUFFIXES
+    )
+
+
 def _get_cloud_role(attributes: Attributes) -> str:
     """Resolve a per-item cloud role without falling back to the host resource.
 

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

@@ -40,6 +40,7 @@
     _AZURE_MONITOR_DISTRO_VERSION_ARG,
     _APPLICATIONINSIGHTS_AUTHENTICATION_STRING,
     _APPLICATIONINSIGHTS_EXPERIMENTAL_FEATURES,
+    _APPLICATIONINSIGHTS_MULTI_ENDPOINT_AAD_AUDIENCE,
     _INVALID_STATUS_CODES,
     _REACHED_INGESTION_STATUS_CODES,
     _REDIRECT_STATUS_CODES,
@@ -59,6 +60,7 @@
 from azure.monitor.opentelemetry.exporter._connection_string_parser import (
     ConnectionStringParser,
 )
+from azure.monitor.opentelemetry.exporter._routing import _is_trusted_routed_redirect, _RoutedRedirectPolicy
 from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage
 from azure.monitor.opentelemetry.exporter._configuration._state import (
     get_configuration_manager,
@@ -145,6 +147,7 @@ def __init__(self, **kwargs: Any) -> None:
         :rtype: None
         """
         route: Optional[tuple[str, bool]] = kwargs.pop("_route", None)
+        routed_aad_audience: Optional[str] = kwargs.pop("_routed_aad_audience", None)
         connection_string_parser: Optional[ConnectionStringParser] = kwargs.pop("_connection_string_parser", None)
         self._enable_multi_endpoint_routing = (
             self._is_experimental_feature_enabled("multi_endpoint_routing")
@@ -184,12 +187,20 @@ def __init__(self, **kwargs: Any) -> None:
             # We use the credential on a regular exporter or customer sdkStats exporter
             self._credential = _get_authentication_credential(**kwargs)
         self._consecutive_redirects = 0  # To prevent circular redirects
+        self._routed_redirect_policy = _RoutedRedirectPolicy()
         self._disable_offline_storage = kwargs.get("disable_offline_storage", False)
         self._connection_string = parsed_connection_string._connection_string
         self._endpoint = route[0].rstrip("/") if route is not None else parsed_connection_string.endpoint
         self._region = parsed_connection_string.region
         self._instrumentation_key = parsed_connection_string.instrumentation_key
-        self._aad_audience = parsed_connection_string.aad_audience
+        self._routed_aad_audience = ""
+        if self._enable_multi_endpoint_routing:
+            self._routed_aad_audience = (
+                routed_aad_audience
+                if routed_aad_audience is not None
+                else os.environ.get(_APPLICATIONINSIGHTS_MULTI_ENDPOINT_AAD_AUDIENCE, "").strip()
+            )
+        self._aad_audience = self._routed_aad_audience if route is not None else parsed_connection_string.aad_audience
         self._storage_maintenance_period = kwargs.get(
             "storage_maintenance_period", 60
         )  # Maintenance interval in seconds.
@@ -244,6 +255,8 @@ def __init__(self, **kwargs: Any) -> None:
             # Explicitly disabling to avoid infinite loop of Span creation when data is exported
             # DistributedTracingPolicy(**kwargs),
         ]
+        if route is not None:
+            policies.insert(0, self._routed_redirect_policy)
 
         # Exclude HttpLoggingPolicy for the sdkstats exporter so its HTTP
         # traffic does not appear in the user's logs.
@@ -330,6 +343,7 @@ def _get_route_exporter(self, endpoint: str, is_aad: bool) -> tuple["BaseExporte
             "connection_string": self._connection_string,
             "credential": self._credential if is_aad else None,
             "authentication_policy": None,
+            "_routed_aad_audience": self._routed_aad_audience,
             "max_envelopes_per_second": 0,
         }
         if temporary:
@@ -722,12 +736,12 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal
                             redirect_has_headers = False
                         if redirect_has_headers and url.scheme and url.netloc:  # pylint: disable=E0606
                             current_url = urlparse(self.client._config.host)  # pylint: disable=W0212
-                            is_insecure_routed_redirect = self._route is not None and url.scheme.lower() != "https"
-                            is_cross_origin_redirect = not self._is_same_registered_domain(
+                            is_insecure_routed_redirect = self._route is not None and not _is_trusted_routed_redirect(
+                                self.client._config.host, location
+                            )
+                            is_cross_origin_redirect = self._route is None and not self._is_same_registered_domain(
                                 current_url.netloc, url.netloc
                             )
-                            # Routed endpoints are HTTPS-only. Also refuse cross-origin
-                            # redirects so auth is never attached for a foreign host.
                             if is_insecure_routed_redirect or is_cross_origin_redirect:
                                 if not self._is_stats_exporter():
                                     if self._should_collect_customer_sdkstats():
@@ -736,14 +750,18 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal
                                             DropCode.CLIENT_EXCEPTION,
                                             _exception_categories.CLIENT_EXCEPTION.value,
                                         )
-                                    logger.error(
-                                        "Refusing %s redirect to %s://%s.",
-                                        "non-HTTPS routed" if is_insecure_routed_redirect else "cross-origin",
-                                        url.scheme,
-                                        url.netloc,
-                                    )
+                                    if is_insecure_routed_redirect:
+                                        logger.error("Refusing untrusted routed redirect.")
+                                    else:
+                                        logger.error(
+                                            "Refusing cross-origin redirect to %s://%s.",
+                                            url.scheme,
+                                            url.netloc,
+                                        )
                                 result = ExportResult.FAILED_NOT_RETRYABLE
                             else:
+                                if self._route is not None:
+                                    self._routed_redirect_policy.redirect_url = location
                                 # Change the host to the new redirected host
                                 self.client._config.host = "{}://{}".format(
                                     url.scheme, url.netloc
@@ -1134,6 +1152,11 @@ def _get_authentication_credential(
 
 
 def _validate_routing_policy_options(options: Dict[str, Any]) -> None:
+    if options.get("authentication_policy") is not None:
+        raise ValueError(
+            "Custom authentication policies are not supported with multi-endpoint routing. "
+            "Pass a token credential and use microsoft.ingestion_auth instead."
+        )
     for name in _ROUTING_REUSED_NAMED_POLICY_OPTIONS:
         if isinstance(options.get(name), HTTPPolicy):
             raise ValueError(