Add managed_identity_client_id argument to DefaultAzureCredential

sdk/identity/azure-identity/CHANGELOG.md

@@ -5,6 +5,9 @@
 - Application authentication APIs from 1.4.0b7
 - `ManagedIdentityCredential` supports the latest version of App Service
   ([#11346](https://github.com/Azure/azure-sdk-for-python/issues/11346))
+- `DefaultAzureCredential` allows specifying the client ID of a user-assigned
+  managed identity via keyword argument `managed_identity_client_id`
+  ([#12991](https://github.com/Azure/azure-sdk-for-python/issues/12991)) 
 
 ## 1.4.0 (2020-08-10)
 ### Added

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

@@ -62,6 +62,8 @@ class DefaultAzureCredential(ChainedTokenCredential):
     :keyword str interactive_browser_tenant_id: Tenant ID to use when authenticating a user through
         :class:`~azure.identity.InteractiveBrowserCredential`. Defaults to the value of environment variable
         AZURE_TENANT_ID, if any. If unspecified, users will authenticate in their home tenants.
+    :keyword str managed_identity_client_id: The client ID of a user-assigned managed identity. Defaults to the value
+        of the environment variable AZURE_CLIENT_ID, if any. If not specified, a system-assigned identity will be used.
     :keyword str shared_cache_username: Preferred username for :class:`~azure.identity.SharedTokenCacheCredential`.
         Defaults to the value of environment variable AZURE_USERNAME, if any.
     :keyword str shared_cache_tenant_id: Preferred tenant for :class:`~azure.identity.SharedTokenCacheCredential`.
@@ -79,6 +81,10 @@ def __init__(self, **kwargs):
             "interactive_browser_tenant_id", os.environ.get(EnvironmentVariables.AZURE_TENANT_ID)
         )
 
+        managed_identity_client_id = kwargs.pop(
+            "managed_identity_client_id", os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
+        )
+
         shared_cache_username = kwargs.pop("shared_cache_username", os.environ.get(EnvironmentVariables.AZURE_USERNAME))
         shared_cache_tenant_id = kwargs.pop(
             "shared_cache_tenant_id", os.environ.get(EnvironmentVariables.AZURE_TENANT_ID)
@@ -99,9 +105,7 @@ def __init__(self, **kwargs):
         if not exclude_environment_credential:
             credentials.append(EnvironmentCredential(authority=authority, **kwargs))
         if not exclude_managed_identity_credential:
-            credentials.append(
-                ManagedIdentityCredential(client_id=os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID), **kwargs)
-            )
+            credentials.append(ManagedIdentityCredential(client_id=managed_identity_client_id, **kwargs))
         if not exclude_shared_token_cache_credential and SharedTokenCacheCredential.supported():
             try:
                 # username and/or tenant_id are only required when the cache contains tokens for multiple identities

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

@@ -50,6 +50,8 @@ class DefaultAzureCredential(ChainedTokenCredential):
         Defaults to **False**.
     :keyword bool exclude_shared_token_cache_credential: Whether to exclude the shared token cache. Defaults to
         **False**.
+    :keyword str managed_identity_client_id: The client ID of a user-assigned managed identity. Defaults to the value
+        of the environment variable AZURE_CLIENT_ID, if any. If not specified, a system-assigned identity will be used.
     :keyword str shared_cache_username: Preferred username for :class:`~azure.identity.SharedTokenCacheCredential`.
         Defaults to the value of environment variable AZURE_USERNAME, if any.
     :keyword str shared_cache_tenant_id: Preferred tenant for :class:`~azure.identity.SharedTokenCacheCredential`.
@@ -67,6 +69,10 @@ def __init__(self, **kwargs: "Any") -> None:
             "shared_cache_tenant_id", os.environ.get(EnvironmentVariables.AZURE_TENANT_ID)
         )
 
+        managed_identity_client_id = kwargs.pop(
+            "managed_identity_client_id", os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
+        )
+
         vscode_tenant_id = kwargs.pop(
             "visual_studio_code_tenant_id", os.environ.get(EnvironmentVariables.AZURE_TENANT_ID)
         )
@@ -82,7 +88,7 @@ def __init__(self, **kwargs: "Any") -> None:
             credentials.append(EnvironmentCredential(authority=authority, **kwargs))
         if not exclude_managed_identity_credential:
             credentials.append(
-                ManagedIdentityCredential(client_id=os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID), **kwargs)
+                ManagedIdentityCredential(client_id=managed_identity_client_id, **kwargs)
             )
         if not exclude_shared_token_cache_credential and SharedTokenCacheCredential.supported():
             try:

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

@@ -265,20 +265,27 @@ def test_default_credential_shared_cache_use(mock_credential):
 
 
 def test_managed_identity_client_id():
-    """The credential should initialize ManagedIdentityCredential with the value of AZURE_CLIENT_ID"""
+    """the credential should accept a user-assigned managed identity's client ID by kwarg or environment variable"""
 
-    expected_client_id = "the-client"
-    with patch.dict(os.environ, {EnvironmentVariables.AZURE_CLIENT_ID: expected_client_id}, clear=True):
-        with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
-            DefaultAzureCredential()
+    expected_args = {"client_id": "the-client"}
 
-    mock_credential.assert_called_once_with(client_id=expected_client_id)
+    with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
+        DefaultAzureCredential(managed_identity_client_id=expected_args["client_id"])
+    mock_credential.assert_called_once_with(**expected_args)
 
-    with patch.dict(os.environ, {}, clear=True):
+    # client id can also be specified in $AZURE_CLIENT_ID
+    with patch.dict(os.environ, {EnvironmentVariables.AZURE_CLIENT_ID: expected_args["client_id"]}, clear=True):
         with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
             DefaultAzureCredential()
+    mock_credential.assert_called_once_with(**expected_args)
 
-    mock_credential.assert_called_once_with(client_id=None)
+    # keyword argument should override environment variable
+    with patch.dict(
+        os.environ, {EnvironmentVariables.AZURE_CLIENT_ID: "not-" + expected_args["client_id"]}, clear=True
+    ):
+        with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
+            DefaultAzureCredential(managed_identity_client_id=expected_args["client_id"])
+    mock_credential.assert_called_once_with(**expected_args)
 
 
 def get_credential_for_shared_cache_test(expected_refresh_token, expected_access_token, cache, **kwargs):

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

@@ -253,20 +253,27 @@ async def test_default_credential_shared_cache_use():
 
 
 def test_managed_identity_client_id():
-    """The credential should initialize ManagedIdentityCredential with the value of AZURE_CLIENT_ID"""
+    """the credential should accept a user-assigned managed identity's client ID by kwarg or environment variable"""
 
-    expected_client_id = "the-client"
-    with patch.dict(os.environ, {EnvironmentVariables.AZURE_CLIENT_ID: expected_client_id}, clear=True):
-        with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
-            DefaultAzureCredential()
+    expected_args = {"client_id": "the client"}
 
-    mock_credential.assert_called_once_with(client_id=expected_client_id)
+    with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
+        DefaultAzureCredential(managed_identity_client_id=expected_args["client_id"])
+    mock_credential.assert_called_once_with(**expected_args)
 
-    with patch.dict(os.environ, {}, clear=True):
+    # client id can also be specified in $AZURE_CLIENT_ID
+    with patch.dict(os.environ, {EnvironmentVariables.AZURE_CLIENT_ID: expected_args["client_id"]}, clear=True):
         with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
             DefaultAzureCredential()
+    mock_credential.assert_called_once_with(**expected_args)
 
-    mock_credential.assert_called_once_with(client_id=None)
+    # keyword argument should override environment variable
+    with patch.dict(
+        os.environ, {EnvironmentVariables.AZURE_CLIENT_ID: "not-" + expected_args["client_id"]}, clear=True
+    ):
+        with patch(DefaultAzureCredential.__module__ + ".ManagedIdentityCredential") as mock_credential:
+            DefaultAzureCredential(managed_identity_client_id=expected_args["client_id"])
+    mock_credential.assert_called_once_with(**expected_args)
 
 
 def get_credential_for_shared_cache_test(expected_refresh_token, expected_access_token, cache, **kwargs):