[Identity] Improve WorkloadIdentityCredential/DAC diagnosability

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