[Identity] Improve WorkloadIdentityCredential/DAC diagnosability (Azure#42346)

* [Identity] Improve WorkloadIdentityCredentia/DAC diagnosability

The goal here is to have WorkloadIdentityCredential be reported in the
the final error message as an attempted credential.

A substitute credential for credentials that fail to construct has been
added and used in the WorkloadIdentityCredential case. This will allow
the reason WorkloadIdentityCredential can't be used to be propagated to
the user.

Signed-off-by: Paul Van Eck <[email protected]>

* Apply suggestions from code review

Co-authored-by: Copilot <[email protected]>

---------

Signed-off-by: Paul Van Eck <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: pvaneck

sdk/identity/azure-identity/CHANGELOG.md

@@ -13,6 +13,7 @@
 ### Other Changes
 
 - `ManagedIdentityCredential` now retries IMDS 410 status responses for at least 70 seconds total duration as required by [Azure IMDS documentation](https://learn.microsoft.com/azure/virtual-machines/instance-metadata-service?tabs=windows#errors-and-debugging).  ([#42330](https://github.com/Azure/azure-sdk-for-python/pull/42330))
+- Improved `DefaultAzureCredential` diagnostics when `WorkloadIdentityCredential` initialization fails. If DAC fails to find a successful credential in the chain, the reason `WorkloadIdentityCredential` failed will be included in the error message. ([#42346](https://github.com/Azure/azure-sdk-for-python/pull/42346))
 
 ## 1.24.0b1 (2025-07-17)
 

sdk/identity/azure-identity/azure/identity/_credentials/chained.py

@@ -23,10 +23,16 @@
 def _get_error_message(history):
     attempts = []
     for credential, error in history:
+        # Check if credential has a custom name (for DACErrorReporter instances)
+        if hasattr(credential, "_credential_name"):
+            credential_name = credential._credential_name  # pylint: disable=protected-access
+        else:
+            credential_name = credential.__class__.__name__
+
         if error:
-            attempts.append("{}: {}".format(credential.__class__.__name__, error))
+            attempts.append("{}: {}".format(credential_name, error))
         else:
-            attempts.append(credential.__class__.__name__)
+            attempts.append(credential_name)
     return """
 Attempted credentials:\n\t{}""".format(
         "\n\t".join(attempts)

sdk/identity/azure-identity/azure/identity/_credentials/default.py

@@ -6,7 +6,14 @@
 import os
 from typing import List, Any, Optional, cast
 
-from azure.core.credentials import AccessToken, AccessTokenInfo, TokenRequestOptions, SupportsTokenInfo, TokenCredential
+from azure.core.credentials import (
+    AccessToken,
+    AccessTokenInfo,
+    TokenRequestOptions,
+    SupportsTokenInfo,
+    TokenCredential,
+)
+from .. import CredentialUnavailableError
 from .._constants import EnvironmentVariables
 from .._internal.utils import get_default_authority, normalize_authority, within_dac, process_credential_exclusions
 from .azure_powershell import AzurePowerShellCredential
@@ -24,6 +31,32 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+class FailedDACCredential:
+    """This acts as a substitute for a credential that has failed to initialize in the DAC chain.
+
+    This allows instantiation errors to be reported in ChainTokenCredential if all token requests fail.
+    """
+
+    def __init__(self, credential_name: str, error: str) -> None:
+        self._error = error
+        self._credential_name = credential_name
+
+    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+        raise CredentialUnavailableError(self._error)
+
+    def get_token_info(self, *scopes, options: Optional[TokenRequestOptions] = None, **kwargs: Any) -> AccessTokenInfo:
+        raise CredentialUnavailableError(self._error)
+
+    def __enter__(self) -> "FailedDACCredential":
+        return self
+
+    def __exit__(self, *args: Any) -> None:
+        pass
+
+    def close(self) -> None:
+        pass
+
+
 class DefaultAzureCredential(ChainedTokenCredential):
     """A credential capable of handling most Azure SDK authentication scenarios. For more information, See
     `Usage guidance for DefaultAzureCredential
@@ -216,16 +249,17 @@ def __init__(self, **kwargs: Any) -> None:  # pylint: disable=too-many-statement
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, _within_dac=True, **kwargs))
         if not exclude_workload_identity_credential:
-            if all(os.environ.get(var) for var in EnvironmentVariables.WORKLOAD_IDENTITY_VARS):
-                client_id = workload_identity_client_id
+            try:
                 credentials.append(
                     WorkloadIdentityCredential(
-                        client_id=cast(str, client_id),
+                        client_id=cast(str, workload_identity_client_id),
                         tenant_id=workload_identity_tenant_id,
-                        token_file_path=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
+                        token_file_path=os.environ.get(EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE),
                         **kwargs,
                     )
                 )
+            except ValueError as ex:
+                credentials.append(FailedDACCredential("WorkloadIdentityCredential", error=str(ex)))
         if not exclude_managed_identity_credential:
             credentials.append(
                 ManagedIdentityCredential(

sdk/identity/azure-identity/azure/identity/_credentials/workload_identity.py

@@ -11,6 +11,13 @@
 from .._constants import EnvironmentVariables
 
 
+WORKLOAD_CONFIG_ERROR = (
+    "WorkloadIdentityCredential authentication unavailable. The workload options are not fully "
+    "configured. See the troubleshooting guide for more information: "
+    "https://aka.ms/azsdk/python/identity/workloadidentitycredential"
+)
+
+
 class TokenFileMixin:
 
     _token_file_path: str
@@ -72,21 +79,25 @@ def __init__(
         tenant_id = tenant_id or os.environ.get(EnvironmentVariables.AZURE_TENANT_ID)
         client_id = client_id or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
         token_file_path = token_file_path or os.environ.get(EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE)
+
+        missing_args = []
         if not tenant_id:
-            raise ValueError(
-                "'tenant_id' is required. Please pass it in or set the "
-                f"{EnvironmentVariables.AZURE_TENANT_ID} environment variable"
-            )
+            missing_args.append("'tenant_id'")
         if not client_id:
-            raise ValueError(
-                "'client_id' is required. Please pass it in or set the "
-                f"{EnvironmentVariables.AZURE_CLIENT_ID} environment variable"
-            )
+            missing_args.append("'client_id'")
         if not token_file_path:
-            raise ValueError(
-                "'token_file_path' is required. Please pass it in or set the "
-                f"{EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE} environment variable"
-            )
+            missing_args.append("'token_file_path'")
+
+        if missing_args:
+            missing_args_str = ", ".join(missing_args)
+            error_message = f"{WORKLOAD_CONFIG_ERROR}. Missing required arguments: {missing_args_str}."
+            raise ValueError(error_message)
+
+        # Type assertions since we've validated these are not None
+        assert tenant_id is not None
+        assert client_id is not None
+        assert token_file_path is not None
+
         self._token_file_path = token_file_path
         super(WorkloadIdentityCredential, self).__init__(
             tenant_id=tenant_id,

sdk/identity/azure-identity/azure/identity/aio/_credentials/default.py

@@ -8,6 +8,7 @@
 
 from azure.core.credentials import AccessToken, AccessTokenInfo, TokenRequestOptions
 from azure.core.credentials_async import AsyncTokenCredential, AsyncSupportsTokenInfo
+from ... import CredentialUnavailableError
 from ..._constants import EnvironmentVariables
 from ..._internal import get_default_authority, normalize_authority, within_dac, process_credential_exclusions
 from .azure_cli import AzureCliCredential
@@ -24,6 +25,34 @@
 _LOGGER = logging.getLogger(__name__)
 
 
+class AsyncFailedDACCredential:
+    """Async version of FailedDACCredential for use in async credential chains.
+
+    This acts as a substitute for an async credential that has failed to initialize in the DAC chain.
+    """
+
+    def __init__(self, credential_name: str, error: str) -> None:
+        self._error = error
+        self._credential_name = credential_name
+
+    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+        raise CredentialUnavailableError(self._error)
+
+    async def get_token_info(
+        self, *scopes, options: Optional[TokenRequestOptions] = None, **kwargs: Any
+    ) -> AccessTokenInfo:
+        raise CredentialUnavailableError(self._error)
+
+    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
+        pass
+
+    async def __aenter__(self) -> "AsyncFailedDACCredential":
+        return self
+
+    async def close(self) -> None:
+        pass
+
+
 class DefaultAzureCredential(ChainedTokenCredential):
     """A credential capable of handling most Azure SDK authentication scenarios. See
     https://aka.ms/azsdk/python/identity/credential-chains#usage-guidance-for-defaultazurecredential.
@@ -180,16 +209,17 @@ def __init__(self, **kwargs: Any) -> None:  # pylint: disable=too-many-statement
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, _within_dac=True, **kwargs))
         if not exclude_workload_identity_credential:
-            if all(os.environ.get(var) for var in EnvironmentVariables.WORKLOAD_IDENTITY_VARS):
-                client_id = workload_identity_client_id
+            try:
                 credentials.append(
                     WorkloadIdentityCredential(
-                        client_id=cast(str, client_id),
+                        client_id=cast(str, workload_identity_client_id),
                         tenant_id=workload_identity_tenant_id,
-                        token_file_path=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
+                        token_file_path=os.environ.get(EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE),
                         **kwargs,
                     )
                 )
+            except ValueError as ex:
+                credentials.append(AsyncFailedDACCredential("WorkloadIdentityCredential", error=str(ex)))
         if not exclude_managed_identity_credential:
             credentials.append(
                 ManagedIdentityCredential(

sdk/identity/azure-identity/azure/identity/aio/_credentials/workload_identity.py

@@ -5,7 +5,7 @@
 import os
 from typing import Any, Optional
 from .client_assertion import ClientAssertionCredential
-from ..._credentials.workload_identity import TokenFileMixin
+from ..._credentials.workload_identity import TokenFileMixin, WORKLOAD_CONFIG_ERROR
 from ..._constants import EnvironmentVariables
 
 
@@ -52,21 +52,25 @@ def __init__(
         tenant_id = tenant_id or os.environ.get(EnvironmentVariables.AZURE_TENANT_ID)
         client_id = client_id or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
         token_file_path = token_file_path or os.environ.get(EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE)
+
+        missing_args = []
         if not tenant_id:
-            raise ValueError(
-                "'tenant_id' is required. Please pass it in or set the "
-                f"{EnvironmentVariables.AZURE_TENANT_ID} environment variable"
-            )
+            missing_args.append("'tenant_id'")
         if not client_id:
-            raise ValueError(
-                "'client_id' is required. Please pass it in or set the "
-                f"{EnvironmentVariables.AZURE_CLIENT_ID} environment variable"
-            )
+            missing_args.append("'client_id'")
         if not token_file_path:
-            raise ValueError(
-                "'token_file_path' is required. Please pass it in or set the "
-                f"{EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE} environment variable"
-            )
+            missing_args.append("'token_file_path'")
+
+        if missing_args:
+            missing_args_str = ", ".join(missing_args)
+            error_message = f"{WORKLOAD_CONFIG_ERROR}. Missing required arguments: {missing_args_str}."
+            raise ValueError(error_message)
+
+        # Type assertions since we've validated these are not None
+        assert tenant_id is not None
+        assert client_id is not None
+        assert token_file_path is not None
+
         self._token_file_path = token_file_path
         super().__init__(
             tenant_id=tenant_id,

sdk/identity/azure-identity/tests/test_default.py

@@ -6,6 +6,7 @@
 import sys
 
 from azure.core.credentials import AccessToken, AccessTokenInfo
+from azure.core.exceptions import ClientAuthenticationError
 from azure.identity import (
     AzureCliCredential,
     AzureDeveloperCliCredential,
@@ -504,3 +505,68 @@ def test_broker_credential_requirements_not_installed():
         broker_cred = broker_credentials[0]
         with pytest.raises(CredentialUnavailableError) as exc_info:
             broker_cred.get_token_info("https://management.azure.com/.default")
+
+
+def test_failed_dac_credential_error_reporting():
+    """Test that FailedDACCredential properly reports initialization errors"""
+    from azure.identity._credentials.default import FailedDACCredential
+
+    credential_name = "WorkloadIdentityCredential"
+    error_message = "Failed to initialize: missing required environment variable AZURE_FEDERATED_TOKEN_FILE"
+
+    failed_credential = FailedDACCredential(credential_name, error_message)
+
+    # Test get_token raises CredentialUnavailableError with the original error
+    with pytest.raises(CredentialUnavailableError) as exc_info:
+        failed_credential.get_token("https://management.azure.com/.default")
+
+    assert str(exc_info.value) == error_message
+
+    # Test get_token_info raises CredentialUnavailableError with the original error
+    with pytest.raises(CredentialUnavailableError) as exc_info:
+        failed_credential.get_token_info("https://management.azure.com/.default")
+
+    assert str(exc_info.value) == error_message
+
+    # Test context manager support
+    with failed_credential:
+        pass  # Should not raise during context entry/exit
+
+    # Test close method
+    failed_credential.close()  # Should not raise
+
+
+def test_failed_dac_credential_in_chain():
+    """Test that FailedDACCredential errors are properly reported when DefaultAzureCredential fails"""
+    from azure.identity._credentials.default import FailedDACCredential
+
+    # Create a mock successful credential to ensure the chain doesn't fail immediately
+    successful_credential = Mock(
+        spec_set=["get_token", "get_token_info"],
+        get_token=Mock(return_value=AccessToken("***", 42)),
+        get_token_info=Mock(return_value=AccessTokenInfo("***", 42)),
+    )
+
+    # Create a DefaultAzureCredential and replace its credentials with a failed credential and successful one
+    credential = DefaultAzureCredential()
+    failed_cred = FailedDACCredential("WorkloadIdentityCredential", "initialization error")
+    credential.credentials = (failed_cred, successful_credential)
+
+    # The chain should succeed using the successful credential
+    token = credential.get_token("https://management.azure.com/.default")
+    assert token.token == "***"
+    assert token.expires_on == 42
+
+    # Test with only failed credentials to ensure error propagation
+    credential_all_failed = DefaultAzureCredential()
+    failed_cred1 = FailedDACCredential("WorkloadIdentityCredential", "workload identity error")
+    failed_cred2 = FailedDACCredential("TestCredential", "test credential error")
+    credential_all_failed.credentials = (failed_cred1, failed_cred2)
+
+    # Should raise an error that includes both credential errors
+    with pytest.raises(ClientAuthenticationError) as exc_info:
+        credential_all_failed.get_token("https://management.azure.com/.default")
+
+    # The error should mention the failed credentials
+    error_str = str(exc_info.value)
+    assert "workload identity error" in error_str or "test credential error" in error_str