[Identity] Log user-assigned MI IDs (Azure#39621)

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

Author: pvaneck

sdk/identity/azure-identity/CHANGELOG.md

@@ -25,6 +25,7 @@
 ### Other Changes
 
 - `AzureCliCredential` and `AzureDeveloperCliCredential` will now call their corresponding executables directly instead of going through the shell. ([#38606](https://github.com/Azure/azure-sdk-for-python/pull/38606))
+- `ManagedIdentityCredential` will now log the configured user-assigned identity if one is set. ([#39621](https://github.com/Azure/azure-sdk-for-python/pull/39621))
 
 ## 1.19.0 (2024-10-08)
 

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

@@ -4,7 +4,7 @@
 # ------------------------------------
 import logging
 import os
-from typing import Optional, Any, Mapping, cast
+from typing import Optional, Any, Mapping, cast, Tuple
 
 from azure.core.credentials import AccessToken, AccessTokenInfo, TokenRequestOptions, TokenCredential, SupportsTokenInfo
 from .. import CredentialUnavailableError
@@ -15,20 +15,34 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-def validate_identity_config(client_id: Optional[str], identity_config: Optional[Mapping[str, str]]) -> None:
+def validate_identity_config(
+    client_id: Optional[str], identity_config: Optional[Mapping[str, str]]
+) -> Optional[Tuple[str, str]]:
     if identity_config:
+        valid_keys = {"object_id", "resource_id", "client_id"}
         if client_id:
-            if any(key in identity_config for key in ("object_id", "resource_id", "client_id")):
+            if any(key in identity_config for key in valid_keys):
                 raise ValueError(
-                    "identity_config must not contain 'object_id', 'resource_id', or 'client_id' when 'client_id' is "
-                    "provided as a keyword argument."
+                    "When 'client_id' is provided as a keyword argument, 'identity_config' must not contain any of the "
+                    f"following keys: {', '.join(valid_keys)}"
                 )
-        # Only one of these keys should be present if one is present.
-        valid_keys = {"object_id", "resource_id", "client_id"}
-        if len(identity_config.keys() & valid_keys) > 1:
-            raise ValueError(
-                f"identity_config must not contain more than one of the following keys: {', '.join(valid_keys)}"
-            )
+            return "client_id", client_id
+
+        # Only one of the valid keys should be present if one is present.
+        result = None
+        for key in valid_keys:
+            if key in identity_config:
+                if result:
+                    raise ValueError(
+                        "identity_config must not contain more than one of the following keys: "
+                        f"{', '.join(valid_keys)}"
+                    )
+                result = key, identity_config[key]
+        return result
+
+    if client_id:
+        return "client_id", client_id
+    return None
 
 
 class ManagedIdentityCredential:
@@ -59,51 +73,58 @@ class ManagedIdentityCredential:
     def __init__(
         self, *, client_id: Optional[str] = None, identity_config: Optional[Mapping[str, str]] = None, **kwargs: Any
     ) -> None:
-        validate_identity_config(client_id, identity_config)
+        user_identity_info = validate_identity_config(client_id, identity_config)
         self._credential: Optional[SupportsTokenInfo] = None
         exclude_workload_identity = kwargs.pop("_exclude_workload_identity_credential", False)
+        managed_identity_type = None
+
         if os.environ.get(EnvironmentVariables.IDENTITY_ENDPOINT):
             if os.environ.get(EnvironmentVariables.IDENTITY_HEADER):
                 if os.environ.get(EnvironmentVariables.IDENTITY_SERVER_THUMBPRINT):
-                    _LOGGER.info("%s will use Service Fabric managed identity", self.__class__.__name__)
+                    managed_identity_type = "Service Fabric managed identity"
                     from .service_fabric import ServiceFabricCredential
 
                     self._credential = ServiceFabricCredential(
                         client_id=client_id, identity_config=identity_config, **kwargs
                     )
                 else:
-                    _LOGGER.info("%s will use App Service managed identity", self.__class__.__name__)
+                    managed_identity_type = "App Service managed identity"
                     from .app_service import AppServiceCredential
 
                     self._credential = AppServiceCredential(
                         client_id=client_id, identity_config=identity_config, **kwargs
                     )
             elif os.environ.get(EnvironmentVariables.IMDS_ENDPOINT):
-                _LOGGER.info("%s will use Azure Arc managed identity", self.__class__.__name__)
+                managed_identity_type = "Azure Arc managed identity"
                 from .azure_arc import AzureArcCredential
 
                 self._credential = AzureArcCredential(client_id=client_id, identity_config=identity_config, **kwargs)
         elif os.environ.get(EnvironmentVariables.MSI_ENDPOINT):
             if os.environ.get(EnvironmentVariables.MSI_SECRET):
-                _LOGGER.info("%s will use Azure ML managed identity", self.__class__.__name__)
+                managed_identity_type = "Azure ML managed identity"
                 from .azure_ml import AzureMLCredential
 
                 self._credential = AzureMLCredential(client_id=client_id, identity_config=identity_config, **kwargs)
             else:
-                _LOGGER.info("%s will use Cloud Shell managed identity", self.__class__.__name__)
+                managed_identity_type = "Cloud Shell managed identity"
                 from .cloud_shell import CloudShellCredential
 
                 self._credential = CloudShellCredential(client_id=client_id, identity_config=identity_config, **kwargs)
         elif (
             all(os.environ.get(var) for var in EnvironmentVariables.WORKLOAD_IDENTITY_VARS)
             and not exclude_workload_identity
         ):
-            _LOGGER.info("%s will use workload identity", self.__class__.__name__)
             from .workload_identity import WorkloadIdentityCredential
 
             workload_client_id = client_id or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
             if not workload_client_id:
-                raise ValueError('Configure the environment with a client ID or pass a value for "client_id" argument')
+                raise ValueError(
+                    "Workload identity was selected but no client ID was provided. "
+                    'Configure the environment with a client ID or pass a value for "client_id" argument'
+                )
+
+            managed_identity_type = "workload identity"
+            user_identity_info = ("client_id", workload_client_id)
 
             self._credential = WorkloadIdentityCredential(
                 tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
@@ -112,11 +133,17 @@ def __init__(
                 **kwargs,
             )
         else:
+            managed_identity_type = "IMDS"
             from .imds import ImdsCredential
 
-            _LOGGER.info("%s will use IMDS", self.__class__.__name__)
             self._credential = ImdsCredential(client_id=client_id, identity_config=identity_config, **kwargs)
 
+        if managed_identity_type:
+            log_msg = f"{self.__class__.__name__} will use {managed_identity_type}"
+            if user_identity_info:
+                log_msg += f" with {user_identity_info[0]}: {user_identity_info[1]}"
+            _LOGGER.info(log_msg)
+
     def __enter__(self) -> "ManagedIdentityCredential":
         if self._credential:
             self._credential.__enter__()  # type: ignore

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

@@ -46,65 +46,76 @@ class ManagedIdentityCredential(AsyncContextManager):
     def __init__(
         self, *, client_id: Optional[str] = None, identity_config: Optional[Mapping[str, str]] = None, **kwargs: Any
     ) -> None:
-        validate_identity_config(client_id, identity_config)
+        user_identity_info = validate_identity_config(client_id, identity_config)
         self._credential: Optional[AsyncSupportsTokenInfo] = None
         exclude_workload_identity = kwargs.pop("_exclude_workload_identity_credential", False)
-
+        managed_identity_type = None
         if os.environ.get(EnvironmentVariables.IDENTITY_ENDPOINT):
             if os.environ.get(EnvironmentVariables.IDENTITY_HEADER):
                 if os.environ.get(EnvironmentVariables.IDENTITY_SERVER_THUMBPRINT):
-                    _LOGGER.info("%s will use Service Fabric managed identity", self.__class__.__name__)
+                    managed_identity_type = "Service Fabric managed identity"
                     from .service_fabric import ServiceFabricCredential
 
                     self._credential = ServiceFabricCredential(
                         client_id=client_id, identity_config=identity_config, **kwargs
                     )
                 else:
-                    _LOGGER.info("%s will use App Service managed identity", self.__class__.__name__)
+                    managed_identity_type = "App Service managed identity"
                     from .app_service import AppServiceCredential
 
                     self._credential = AppServiceCredential(
                         client_id=client_id, identity_config=identity_config, **kwargs
                     )
             elif os.environ.get(EnvironmentVariables.IMDS_ENDPOINT):
-                _LOGGER.info("%s will use Azure Arc managed identity", self.__class__.__name__)
+                managed_identity_type = "Azure Arc managed identity"
                 from .azure_arc import AzureArcCredential
 
                 self._credential = AzureArcCredential(client_id=client_id, identity_config=identity_config, **kwargs)
         elif os.environ.get(EnvironmentVariables.MSI_ENDPOINT):
             if os.environ.get(EnvironmentVariables.MSI_SECRET):
-                _LOGGER.info("%s will use Azure ML managed identity", self.__class__.__name__)
+                managed_identity_type = "Azure ML managed identity"
                 from .azure_ml import AzureMLCredential
 
                 self._credential = AzureMLCredential(client_id=client_id, identity_config=identity_config, **kwargs)
             else:
-                _LOGGER.info("%s will use Cloud Shell managed identity", self.__class__.__name__)
+                managed_identity_type = "Cloud Shell managed identity"
                 from .cloud_shell import CloudShellCredential
 
                 self._credential = CloudShellCredential(client_id=client_id, identity_config=identity_config, **kwargs)
         elif (
             all(os.environ.get(var) for var in EnvironmentVariables.WORKLOAD_IDENTITY_VARS)
             and not exclude_workload_identity
         ):
-            _LOGGER.info("%s will use workload identity", self.__class__.__name__)
             from .workload_identity import WorkloadIdentityCredential
 
             workload_client_id = client_id or os.environ.get(EnvironmentVariables.AZURE_CLIENT_ID)
             if not workload_client_id:
-                raise ValueError('Configure the environment with a client ID or pass a value for "client_id" argument')
+                raise ValueError(
+                    "Workload identity was selected but no client ID was provided. "
+                    'Configure the environment with a client ID or pass a value for "client_id" argument'
+                )
+
+            managed_identity_type = "workload identity"
+            user_identity_info = ("client_id", workload_client_id)
 
             self._credential = WorkloadIdentityCredential(
                 tenant_id=os.environ[EnvironmentVariables.AZURE_TENANT_ID],
                 client_id=workload_client_id,
                 token_file_path=os.environ[EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE],
-                **kwargs
+                **kwargs,
             )
         else:
+            managed_identity_type = "IMDS"
             from .imds import ImdsCredential
 
-            _LOGGER.info("%s will use IMDS", self.__class__.__name__)
             self._credential = ImdsCredential(client_id=client_id, identity_config=identity_config, **kwargs)
 
+        if managed_identity_type:
+            log_msg = f"{self.__class__.__name__} will use {managed_identity_type}"
+            if user_identity_info:
+                log_msg += f" with {user_identity_info[0]}: {user_identity_info[1]}"
+            _LOGGER.info(log_msg)
+
     async def __aenter__(self) -> "ManagedIdentityCredential":
         if self._credential:
             await self._credential.__aenter__()

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

@@ -4,11 +4,13 @@
 # ------------------------------------
 from itertools import product
 import time
+import logging
 from unittest import mock
 
 from azure.identity import ManagedIdentityCredential, CredentialUnavailableError
 from azure.identity._constants import EnvironmentVariables
 from azure.identity._credentials.imds import IMDS_AUTHORITY, IMDS_TOKEN_PATH
+from azure.identity._credentials.managed_identity import validate_identity_config
 from azure.identity._internal.user_agent import USER_AGENT
 from azure.identity._internal import within_credential_chain
 import pytest
@@ -1003,6 +1005,20 @@ def test_validate_identity_config():
         ManagedIdentityCredential(identity_config={"object_id": "bar", "client_id": "foo"})
 
 
+def test_validate_identity_config_output():
+    output = validate_identity_config(None, {"client_id": "foo"})
+    assert output == ("client_id", "foo")
+
+    output = validate_identity_config("foo", None)
+    assert output == ("client_id", "foo")
+
+    output = validate_identity_config(None, {"object_id": "bar"})
+    assert output == ("object_id", "bar")
+
+    output = validate_identity_config(None, {"resource_id": "biz"})
+    assert output == ("resource_id", "biz")
+
+
 def test_validate_cloud_shell_credential():
     with mock.patch.dict(
         MANAGED_IDENTITY_ENVIRON, {EnvironmentVariables.MSI_ENDPOINT: "https://localhost"}, clear=True
@@ -1016,3 +1032,41 @@ def test_validate_cloud_shell_credential():
             ManagedIdentityCredential(identity_config={"object_id": "foo"})
         with pytest.raises(ValueError):
             ManagedIdentityCredential(identity_config={"resource_id": "foo"})
+
+
+def test_log(caplog):
+    with caplog.at_level(logging.INFO, logger="azure.identity._credentials.managed_identity"):
+        ManagedIdentityCredential()
+        assert "ManagedIdentityCredential will use IMDS" in caplog.text
+
+        caplog.clear()
+        with mock.patch.dict(
+            MANAGED_IDENTITY_ENVIRON,
+            {
+                EnvironmentVariables.IDENTITY_ENDPOINT: "new_endpoint",
+                EnvironmentVariables.IDENTITY_HEADER: "new_secret",
+                EnvironmentVariables.MSI_ENDPOINT: "endpoint",
+                EnvironmentVariables.MSI_SECRET: "secret",
+            },
+            clear=True,
+        ):
+            ManagedIdentityCredential()
+            assert "App Service managed identity" in caplog.text
+
+            caplog.clear()
+            ManagedIdentityCredential(client_id="foo")
+            assert "App Service managed identity with client_id: foo" in caplog.text
+
+            caplog.clear()
+            ManagedIdentityCredential(identity_config={"object_id": "bar"})
+            assert "App Service managed identity with object_id: bar" in caplog.text
+
+        caplog.clear()
+        mock_environ = {
+            EnvironmentVariables.AZURE_AUTHORITY_HOST: "authority",
+            EnvironmentVariables.AZURE_TENANT_ID: "tenant",
+            EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: "token_file",
+        }
+        with mock.patch.dict("os.environ", mock_environ, clear=True):
+            ManagedIdentityCredential(client_id="foo")
+            assert "workload identity with client_id: foo" in caplog.text

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

@@ -5,6 +5,7 @@
 from itertools import product
 import os
 import time
+import logging
 from unittest import mock
 
 from azure.core.exceptions import ClientAuthenticationError
@@ -1301,3 +1302,41 @@ def test_validate_cloud_shell_credential():
             ManagedIdentityCredential(identity_config={"object_id": "foo"})
         with pytest.raises(ValueError):
             ManagedIdentityCredential(identity_config={"resource_id": "foo"})
+
+
+def test_log(caplog):
+    with caplog.at_level(logging.INFO, logger="azure.identity.aio._credentials.managed_identity"):
+        ManagedIdentityCredential()
+        assert "ManagedIdentityCredential will use IMDS" in caplog.text
+
+        caplog.clear()
+        with mock.patch.dict(
+            MANAGED_IDENTITY_ENVIRON,
+            {
+                EnvironmentVariables.IDENTITY_ENDPOINT: "new_endpoint",
+                EnvironmentVariables.IDENTITY_HEADER: "new_secret",
+                EnvironmentVariables.MSI_ENDPOINT: "endpoint",
+                EnvironmentVariables.MSI_SECRET: "secret",
+            },
+            clear=True,
+        ):
+            ManagedIdentityCredential()
+            assert "App Service managed identity" in caplog.text
+
+            caplog.clear()
+            ManagedIdentityCredential(client_id="foo")
+            assert "App Service managed identity with client_id: foo" in caplog.text
+
+            caplog.clear()
+            ManagedIdentityCredential(identity_config={"object_id": "bar"})
+            assert "App Service managed identity with object_id: bar" in caplog.text
+
+        caplog.clear()
+        mock_environ = {
+            EnvironmentVariables.AZURE_AUTHORITY_HOST: "authority",
+            EnvironmentVariables.AZURE_TENANT_ID: "tenant",
+            EnvironmentVariables.AZURE_FEDERATED_TOKEN_FILE: "token_file",
+        }
+        with mock.patch.dict("os.environ", mock_environ, clear=True):
+            ManagedIdentityCredential(client_id="foo")
+            assert "workload identity with client_id: foo" in caplog.text