Proof-of-Concept: MI via CCA

docs/index.rst

@@ -130,3 +130,24 @@ See `SerializableTokenCache` for example.
 
 .. autoclass:: msal.SerializableTokenCache
    :members:
+
+
+Managed Identity
+================
+MSAL supports
+`Managed Identity <https://learn.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/overview>`_.
+
+You can create one of these two kinds of managed identity configuration objects:
+
+.. autoclass:: msal.SystemAssignedManagedIdentity
+   :members:
+
+.. autoclass:: msal.UserAssignedManagedIdentity
+   :members:
+
+And then feed the configuration object into a :class:`ManagedIdentityClient` object.
+
+.. autoclass:: msal.ManagedIdentityClient
+   :members:
+
+   .. automethod:: __init__

msal/__init__.py

@@ -33,4 +33,8 @@
     )
 from .oauth2cli.oidc import Prompt
 from .token_cache import TokenCache, SerializableTokenCache
+from .managed_identity import (
+    SystemAssignedManagedIdentity, UserAssignedManagedIdentity,
+    ManagedIdentityClient,
+    )
 

msal/application.py

@@ -22,6 +22,7 @@
 from .region import _detect_region
 from .throttled_http_client import ThrottledHttpClient
 from .cloudshell import _is_running_in_cloud_shell
+from .imds import ManagedIdentityClient, ManagedIdentity, _scope_to_resource
 
 
 # The __init__.py will import this. Not the other way around.
@@ -2021,6 +2022,14 @@ def acquire_token_for_client(self, scopes, claims_challenge=None, **kwargs):
             - an error response would contain "error" and usually "error_description".
         """
         # TBD: force_refresh behavior
+        if ManagedIdentity.is_managed_identity(self.client_id):
+            if len(scopes) != 1:
+                raise ValueError("Managed Identity supports only one scope/resource")
+            if claims_challenge:
+                raise ValueError("Managed Identity does not support claims_challenge")
+            return ManagedIdentityClient(
+                self.http_client, self.client_id, self.token_cache
+                ).acquire_token(_scope_to_resource(scopes[0]))
         if self.authority.tenant.lower() in ["common", "organizations"]:
             warnings.warn(
                 "Using /common or /organizations authority "