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
 - Added support to use "TryAutoDetect" as the value for `AZURE_REGIONAL_AUTHORITY_NAME` to enable auto detecting the appropriate authority ([#526](https://github.com/AzureAD/microsoft-authentication-library-for-python/issues/526))
 
 ## 1.13.0b1 (2023-01-10)

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

@@ -25,6 +25,7 @@
     SharedTokenCacheCredential,
     UsernamePasswordCredential,
     VisualStudioCodeCredential,
+    WorkloadIdentityCredential,
 )
 from ._persistent_cache import TokenCachePersistenceOptions
 
@@ -53,6 +54,7 @@
     "TokenCachePersistenceOptions",
     "UsernamePasswordCredential",
     "VisualStudioCodeCredential",
+    "WorkloadIdentityCredential",
 ]
 
 from ._version import VERSION

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

@@ -4,7 +4,7 @@
 # ------------------------------------
 import logging
 import os
-from typing import List, TYPE_CHECKING, Any
+from typing import List, TYPE_CHECKING, Any, cast
 
 from azure.core.credentials import AccessToken
 from .._constants import EnvironmentVariables
@@ -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.
 
@@ -76,7 +80,7 @@ class DefaultAzureCredential(ChainedTokenCredential):
         Directory work or school accounts.
     """
 
-    def __init__(self, **kwargs: Any) -> None:
+    def __init__(self, **kwargs: Any) -> None:  # pylint: disable=too-many-statements
         if "tenant_id" in kwargs:
             raise TypeError("'tenant_id' is not supported in DefaultAzureCredential.")
 
@@ -119,6 +123,13 @@ def __init__(self, **kwargs: Any) -> None:
         credentials = []  # type: List[TokenCredential]
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, **kwargs))
+        if all(os.environ.get(var) for var in EnvironmentVariables.TOKEN_EXCHANGE_VARS):
+            client_id = managed_identity_client_id or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
+            credentials.append(WorkloadIdentityCredential(
+                client_id=cast(str, client_id),
+                tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
+                file=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
+                **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/managed_identity.py

@@ -64,17 +64,17 @@ def __init__(self, **kwargs: Any) -> None:
 
                 self._credential = CloudShellCredential(**kwargs)
         elif all(os.environ.get(var) for var in EnvironmentVariables.TOKEN_EXCHANGE_VARS):
-            _LOGGER.info("%s will use token exchange", self.__class__.__name__)
-            from .token_exchange import TokenExchangeCredential
+            _LOGGER.info("%s will use workload identity", self.__class__.__name__)
+            from .workload_identity import WorkloadIdentityCredential
 
             client_id = kwargs.pop("client_id", None) or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
             if not client_id:
                 raise ValueError('Configure the environment with a client ID or pass a value for "client_id" argument')
 
-            self._credential = TokenExchangeCredential(
+            self._credential = WorkloadIdentityCredential(
                 tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
                 client_id=client_id,
-                token_file_path=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
+                file=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
                 **kwargs
             )
         else:

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

@@ -11,35 +11,43 @@
 class TokenFileMixin:
     def __init__(
             self,
-            token_file_path: str,
+            file: str,
             **_: Any
     ) -> None:
         super(TokenFileMixin, self).__init__()
         self._jwt = ""
         self._last_read_time = 0
-        self._token_file_path = token_file_path
+        self._file = file
 
     def get_service_account_token(self) -> str:
         now = int(time.time())
-        if now - self._last_read_time > 300:
-            with open(self._token_file_path) as f:
+        if now - self._last_read_time > 600:
+            with open(self._file) as f:
                 self._jwt = f.read()
             self._last_read_time = now
         return self._jwt
 
 
-class TokenExchangeCredential(ClientAssertionCredential, TokenFileMixin):
+class WorkloadIdentityCredential(ClientAssertionCredential, TokenFileMixin):
+    """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,
-            token_file_path: str,
+            file: str,
             **kwargs: Any
     ) -> None:
-        super(TokenExchangeCredential, self).__init__(
+        super(WorkloadIdentityCredential, self).__init__(
             tenant_id=tenant_id,
             client_id=client_id,
             func=self.get_service_account_token,
-            token_file_path=token_file_path,
+            file=file,
             **kwargs
         )

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

@@ -19,6 +19,7 @@
     SharedTokenCacheCredential,
     VisualStudioCodeCredential,
     ClientAssertionCredential,
+    WorkloadIdentityCredential,
 )
 
 
@@ -37,4 +38,5 @@
     "SharedTokenCacheCredential",
     "VisualStudioCodeCredential",
     "ClientAssertionCredential",
+    "WorkloadIdentityCredential",
 ]

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

@@ -4,7 +4,7 @@
 # ------------------------------------
 import logging
 import os
-from typing import List, TYPE_CHECKING, Any
+from typing import List, TYPE_CHECKING, Any, cast
 
 from azure.core.credentials import AccessToken
 from ..._constants import EnvironmentVariables
@@ -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,13 @@ def __init__(self, **kwargs: Any) -> None:
         credentials = []  # type: List[AsyncTokenCredential]
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, **kwargs))
+        if all(os.environ.get(var) for var in EnvironmentVariables.TOKEN_EXCHANGE_VARS):
+            client_id = managed_identity_client_id or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
+            credentials.append(WorkloadIdentityCredential(
+                client_id=cast(str, client_id),
+                tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
+                file=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
+                **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/managed_identity.py

@@ -71,17 +71,17 @@ def __init__(self, **kwargs: Any) -> None:
 
                 self._credential = CloudShellCredential(**kwargs)
         elif all(os.environ.get(var) for var in EnvironmentVariables.TOKEN_EXCHANGE_VARS):
-            _LOGGER.info("%s will use token exchange", self.__class__.__name__)
-            from .token_exchange import TokenExchangeCredential
+            _LOGGER.info("%s will use workload identity", self.__class__.__name__)
+            from .workload_identity import WorkloadIdentityCredential
 
             client_id = kwargs.pop("client_id", None) or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
             if not client_id:
                 raise ValueError('Configure the environment with a client ID or pass a value for "client_id" argument')
 
-            self._credential = TokenExchangeCredential(
+            self._credential = WorkloadIdentityCredential(
                 tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
                 client_id=client_id,
-                token_file_path=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
+                file=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
                 **kwargs
             )
         else:

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

@@ -0,0 +1,25 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import Any
+from .client_assertion import ClientAssertionCredential
+from ..._credentials.workload_identity import TokenFileMixin
+
+
+class WorkloadIdentityCredential(ClientAssertionCredential, TokenFileMixin):
+    """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:
+        super().__init__(
+            tenant_id=tenant_id,
+            client_id=client_id,
+            func=self.get_service_account_token,
+            file=file,
+            **kwargs
+        )

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

@@ -5,15 +5,15 @@
 %% 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;
+        linkStyle 6 stroke-width:0px;
+        linkStyle 7 stroke-width:0px;
     end;
 
     %% Define styles for credential type boxes
@@ -22,9 +22,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;
 ```