Added WorkloadIdentityCredential

sdk/identity/azure-identity/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features Added
 
 - Added `AzureDeveloperCredential` for Azure Developer CLI. ([#27916](https://github.com/Azure/azure-sdk-for-python/pull/27916))
+- Added `WorkloadIdentityCredential` for Workload Identity Federation on Kubernetes
 
 ### Breaking Changes
 

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

@@ -19,6 +19,7 @@
 from .user_password import UsernamePasswordCredential
 from .vscode import VisualStudioCodeCredential
 from .client_assertion import ClientAssertionCredential
+from .workload_identity import WorkloadIdentityCredential
 
 
 __all__ = [
@@ -39,5 +40,6 @@
     "SharedTokenCacheCredential",
     "AzureCliCredential",
     "UsernamePasswordCredential",
+    "WorkloadIdentityCredential",
     "VisualStudioCodeCredential",
 ]

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

@@ -18,6 +18,7 @@
 from .azure_cli import AzureCliCredential
 from .azd_cli import AzureDeveloperCliCredential
 from .vscode import VisualStudioCodeCredential
+from .workload_identity import WorkloadIdentityCredential
 
 if TYPE_CHECKING:
     from azure.core.credentials import TokenCredential
@@ -33,12 +34,15 @@ class DefaultAzureCredential(ChainedTokenCredential):
 
     1. A service principal configured by environment variables. See :class:`~azure.identity.EnvironmentCredential` for
        more details.
-    2. An Azure managed identity. See :class:`~azure.identity.ManagedIdentityCredential` for more details.
-    3. On Windows only: a user who has signed in with a Microsoft application, such as Visual Studio. If multiple
+    2. WorkloadIdentityCredential if environment variable configuration is set by the Azure workload
+       identity webhook.
+    3. An Azure managed identity. See :class:`~azure.identity.ManagedIdentityCredential` for more details.
+    4. The identity currently logged in to the Azure Developer CLI.
+    5. On Windows only: a user who has signed in with a Microsoft application, such as Visual Studio. If multiple
        identities are in the cache, then the value of  the environment variable ``AZURE_USERNAME`` is used to select
        which identity to use. See :class:`~azure.identity.SharedTokenCacheCredential` for more details.
-    4. The identity currently logged in to the Azure CLI.
-    5. The identity currently logged in to Azure PowerShell.
+    6. The identity currently logged in to the Azure CLI.
+    7. The identity currently logged in to Azure PowerShell.
 
     This default behavior is configurable with keyword arguments.
 
@@ -119,6 +123,7 @@ def __init__(self, **kwargs: Any) -> None:
         credentials = []  # type: List[TokenCredential]
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, **kwargs))
+        credentials.append(WorkloadIdentityCredential(client_id=managed_identity_client_id, **kwargs))
         if not exclude_managed_identity_credential:
             credentials.append(ManagedIdentityCredential(client_id=managed_identity_client_id, **kwargs))
         if not exclude_azd_cli_credential:

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

@@ -21,7 +21,7 @@ def __init__(
 
     def get_service_account_token(self) -> str:
         now = int(time.time())
-        if now - self._last_read_time > 300:
+        if now - self._last_read_time > 600:
             with open(self._token_file_path) as f:
                 self._jwt = f.read()
             self._last_read_time = now

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

@@ -0,0 +1,66 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import Any
+import logging
+from azure.core.credentials import AccessToken
+from .token_exchange import TokenExchangeCredential
+from .._internal.decorators import log_get_token
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class WorkloadIdentityCredential:
+    """WorkloadIdentityCredential supports Azure workload identity on Kubernetes.
+    See https://learn.microsoft.com/azure/aks/workload-identity-overview for more information
+
+    :param str tenant_id: ID of the application's Azure Active Directory tenant. Also called its "directory" ID.
+    :param str client_id: the client ID of an Azure AD app registration.
+    :param str file: The path to a file containing a Kubernetes service account token that authenticates the identity.
+    """
+
+    def __init__(
+            self,
+            tenant_id: str,
+            client_id: str,
+            file: str,
+            **kwargs: Any
+    ) -> None:
+        kwargs.pop("token_file_path", None)
+        self._credential = TokenExchangeCredential(
+            tenant_id=tenant_id,
+            client_id=client_id,
+            token_file_path=file,
+            **kwargs
+        )
+
+    def __enter__(self):
+        if self._credential:
+            self._credential.__enter__()
+        return self
+
+    def __exit__(self, *args):
+        if self._credential:
+            self._credential.__exit__(*args)
+
+    def close(self) -> None:
+        """Close the credential's transport session."""
+        self.__exit__()
+
+    @log_get_token("WorkloadIdentityCredential")
+    def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+        """Request an access token for `scopes`.
+
+        This method is called automatically by Azure SDK clients.
+
+        :param str scopes: desired scopes for the access token. This method requires at least one scope.
+            For more information about scopes, see
+            https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str tenant_id: optional tenant to include in the token request.
+
+        :rtype: :class:`azure.core.credentials.AccessToken`
+
+        :raises ~azure.identity.CredentialUnavailableError: environment variable configuration is incomplete
+        """
+        return self._credential.get_token(*scopes, **kwargs)

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

@@ -16,6 +16,7 @@
 from .azd_cli import AzureDeveloperCliCredential
 from .vscode import VisualStudioCodeCredential
 from .client_assertion import ClientAssertionCredential
+from .workload_identity import WorkloadIdentityCredential
 
 
 __all__ = [
@@ -33,4 +34,5 @@
     "SharedTokenCacheCredential",
     "VisualStudioCodeCredential",
     "ClientAssertionCredential",
+    "WorkloadIdentityCredential",
 ]

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

@@ -17,6 +17,7 @@
 from .managed_identity import ManagedIdentityCredential
 from .shared_cache import SharedTokenCacheCredential
 from .vscode import VisualStudioCodeCredential
+from .workload_identity import WorkloadIdentityCredential
 
 if TYPE_CHECKING:
     from azure.core.credentials_async import AsyncTokenCredential
@@ -32,12 +33,15 @@ class DefaultAzureCredential(ChainedTokenCredential):
 
     1. A service principal configured by environment variables. See :class:`~azure.identity.aio.EnvironmentCredential`
        for more details.
-    2. An Azure managed identity. See :class:`~azure.identity.aio.ManagedIdentityCredential` for more details.
-    3. On Windows only: a user who has signed in with a Microsoft application, such as Visual Studio. If multiple
+    2. WorkloadIdentityCredential if environment variable configuration is set by the Azure workload
+       identity webhook.
+    3. An Azure managed identity. See :class:`~azure.identity.aio.ManagedIdentityCredential` for more details.
+    4. The identity currently logged in to the Azure Developer CLI.
+    5. On Windows only: a user who has signed in with a Microsoft application, such as Visual Studio. If multiple
        identities are in the cache, then the value of  the environment variable ``AZURE_USERNAME`` is used to select
        which identity to use. See :class:`~azure.identity.aio.SharedTokenCacheCredential` for more details.
-    4. The identity currently logged in to the Azure CLI.
-    5. The identity currently logged in to Azure PowerShell.
+    6. The identity currently logged in to the Azure CLI.
+    7. The identity currently logged in to Azure PowerShell.
 
     This default behavior is configurable with keyword arguments.
 
@@ -109,6 +113,7 @@ def __init__(self, **kwargs: Any) -> None:
         credentials = []  # type: List[AsyncTokenCredential]
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, **kwargs))
+        credentials.append(WorkloadIdentityCredential(client_id=managed_identity_client_id, **kwargs))
         if not exclude_managed_identity_credential:
             credentials.append(ManagedIdentityCredential(client_id=managed_identity_client_id, **kwargs))
         if not exclude_azd_cli_credential:

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

@@ -0,0 +1,56 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import Any
+import logging
+from azure.core.credentials import AccessToken
+from .token_exchange import TokenExchangeCredential
+from .._internal.decorators import log_get_token_async
+from .._internal import AsyncContextManager
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class WorkloadIdentityCredential(AsyncContextManager):
+    """WorkloadIdentityCredential supports Azure workload identity on Kubernetes.
+    See https://learn.microsoft.com/azure/aks/workload-identity-overview for more information
+
+    :param str tenant_id: ID of the application's Azure Active Directory tenant. Also called its "directory" ID.
+    :param str client_id: the client ID of an Azure AD app registration.
+    :param str file: The path to a file containing a Kubernetes service account token that authenticates the identity.
+    """
+    def __init__(self, tenant_id: str, client_id: str, file: str, **kwargs: Any) -> None:
+        kwargs.pop("token_file_path", None)
+        self._credential = TokenExchangeCredential(
+            tenant_id=tenant_id,
+            client_id=client_id,
+            token_file_path=file,
+            **kwargs
+        )
+
+    async def __aenter__(self):
+        if self._credential:
+            await self._credential.__aenter__()
+        return self
+
+    async def close(self) -> None:
+        """Close the credential's transport session."""
+
+        if self._credential:
+            await self._credential.__aexit__()
+
+    @log_get_token_async
+    async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
+        """Asynchronously request an access token for `scopes`.
+
+        This method is called automatically by Azure SDK clients.
+
+        :param str scopes: desired scopes for the access token. This method requires at least one scope.
+            For more information about scopes, see
+            https://learn.microsoft.com/azure/active-directory/develop/scopes-oidc.
+        :keyword str tenant_id: optional tenant to include in the token request.
+        :rtype: :class:`azure.core.credentials.AccessToken`
+        :raises ~azure.identity.CredentialUnavailableError: environment variable configuration is incomplete
+        """
+        return await self._credential.get_token(*scopes, **kwargs)

sdk/identity/azure-identity/images/mermaidjs/DefaultAzureCredentialAuthFlow.md

@@ -5,15 +5,13 @@
 %% 2. Run command: mmdc -i DefaultAzureCredentialAuthFlow.md -o DefaultAzureCredentialAuthFlow.svg
 
 flowchart LR;
-    A(Environment):::deployed ==> B(Managed Identity):::deployed ==> C(Azure CLI):::developer ==> D(Azure PowerShell):::developer ==> E(Interactive browser):::interactive;
+    A(Environment):::deployed ==> B(Workload Identity):::deployed ==> C(Managed Identity):::deployed ==> D(Azure Developer CLI):::developer ==> E(Azure CLI):::developer ==> F(Azure PowerShell):::developer ==> G(Interactive browser):::interactive;
 
     subgraph CREDENTIAL TYPES;
         direction LR;
         Deployed(Deployed service):::deployed ==> Developer(Developer):::developer ==> Interactive(Interactive developer):::interactive;
 
         %% Hide links between boxes in the legend by setting width to 0. The integers after "linkStyle" represent link indices.
-        linkStyle 4 stroke-width:0px;
-        linkStyle 5 stroke-width:0px;
     end;
 
     %% Define styles for credential type boxes
@@ -22,9 +20,9 @@ flowchart LR;
     classDef interactive fill:#A5A5A5, stroke:#828282;
 
     %% Add API ref links to credential type boxes
-    click A "https://docs.microsoft.com/python/api/azure-identity/azure.identity.environmentcredential?view=azure-python" _blank;
-    click B "https://docs.microsoft.com/python/api/azure-identity/azure.identity.managedidentitycredential?view=azure-python" _blank;
-    click C "https://docs.microsoft.com/python/api/azure-identity/azure.identity.azureclicredential?view=azure-python" _blank;
-    click D "https://docs.microsoft.com/python/api/azure-identity/azure.identity.azurepowershellcredential?view=azure-python" _blank;
-    click E "https://docs.microsoft.com/python/api/azure-identity/azure.identity.interactivebrowsercredential?view=azure-python" _blank;
+    click A "https://learn.microsoft.com/python/api/azure-identity/azure.identity.environmentcredential?view=azure-python" _blank;
+    click C "https://learn.microsoft.com/python/api/azure-identity/azure.identity.managedidentitycredential?view=azure-python" _blank;
+    click E "https://learn.microsoft.com/python/api/azure-identity/azure.identity.azureclicredential?view=azure-python" _blank;
+    click F "https://learn.microsoft.com/python/api/azure-identity/azure.identity.azurepowershellcredential?view=azure-python" _blank;
+    click G "https://learn.microsoft.com/python/api/azure-identity/azure.identity.interactivebrowsercredential?view=azure-python" _blank;
 ```