CCA federated by managed identity

Author: rayluo

msal/__init__.py

@@ -35,7 +35,7 @@
 from .token_cache import TokenCache, SerializableTokenCache
 from .auth_scheme import PopAuthScheme
 from .managed_identity import (
-    SystemAssignedManagedIdentity, UserAssignedManagedIdentity,
+    ManagedIdentity, SystemAssignedManagedIdentity, UserAssignedManagedIdentity,
     ManagedIdentityClient,
     ManagedIdentityError,
     ArcPlatformNotSupportedError,

msal/__main__.py

@@ -11,6 +11,8 @@
     shiv -e msal.__main__._main -o msaltest-on-os-name.pyz .
 """
 import base64, getpass, json, logging, sys, os, atexit, msal
+#import http.client as http_client
+#http_client.HTTPConnection.debuglevel = 1  # Show http request/response on the wire
 
 _token_cache_filename = "msal_cache.bin"
 global_cache = msal.SerializableTokenCache()
@@ -263,9 +265,11 @@ def _main():
         {"client_id": "95de633a-083e-42f5-b444-a4295d8e9314", "name": "Whiteboard Services (Non MSA-PT app. Accepts AAD & MSA accounts.)"},
         {
             "client_id": os.getenv("CLIENT_ID"),
-            "client_secret": os.getenv("CLIENT_SECRET"),
-            "name": "A confidential client app (CCA) whose settings are defined "
-                "in environment variables CLIENT_ID and CLIENT_SECRET",
+            "client_secret": os.getenv("CLIENT_SECRET", msal.SystemAssignedManagedIdentity()),
+            "name": "A confidential client app (CCA) whose "
+                "(1) client_id is defined in environment variables CLIENT_ID "
+                "(2) client_secret is either in env var CLIENT_SECRET (if any) "
+                "or federated by system-assigned managed identity (if applicable)",
         },
         ],
         option_renderer=lambda a: a["name"],

msal/application.py

@@ -8,6 +8,7 @@
 import os
 
 from .oauth2cli import Client, JwtAssertionCreator
+from .oauth2cli.assertion import AutoRefresher
 from .oauth2cli.oidc import decode_part
 from .authority import Authority, WORLD_WIDE
 from .mex import send_request as mex_send_request
@@ -18,6 +19,7 @@
 from .region import _detect_region
 from .throttled_http_client import ThrottledHttpClient
 from .cloudshell import _is_running_in_cloud_shell
+from .managed_identity import ManagedIdentity, ManagedIdentityClient
 
 
 # The __init__.py will import this. Not the other way around.
@@ -249,29 +251,76 @@ def __init__(
             The thumbprint is available in your app's registration in Azure Portal.
             Alternatively, you can `calculate the thumbprint <https://github.com/Azure/azure-sdk-for-python/blob/07d10639d7e47f4852eaeb74aef5d569db499d6e/sdk/identity/azure-identity/azure/identity/_credentials/certificate.py#L94-L97>`_.
 
-            *Added in version 0.5.0*:
-            public_certificate (optional) is public key certificate
-            which will be sent through 'x5c' JWT header only for
-            subject name and issuer authentication to support cert auto rolls.
-
-            Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
-            "the certificate containing
-            the public key corresponding to the key used to digitally sign the
-            JWS MUST be the first certificate.  This MAY be followed by
-            additional certificates, with each subsequent certificate being the
-            one used to certify the previous one."
-            However, your certificate's issuer may use a different order.
-            So, if your attempt ends up with an error AADSTS700027 -
-            "The provided signature value did not match the expected signature value",
-            you may try use only the leaf cert (in PEM/str format) instead.
-
-            *Added in version 1.13.0*:
-            It can also be a completely pre-signed assertion that you've assembled yourself.
-            Simply pass a container containing only the key "client_assertion", like this::
+            .. admonition:: Using ``public_certificate`` to support Subject Name/Issuer Auth
 
-                {
-                    "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
-                }
+                *Added in version 0.5.0*:
+                public_certificate (optional) is public key certificate
+                which will be sent through 'x5c' JWT header only for
+                subject name and issuer authentication to support cert auto rolls.
+
+                Per `specs <https://tools.ietf.org/html/rfc7515#section-4.1.6>`_,
+                "the certificate containing
+                the public key corresponding to the key used to digitally sign the
+                JWS MUST be the first certificate.  This MAY be followed by
+                additional certificates, with each subsequent certificate being the
+                one used to certify the previous one."
+                However, your certificate's issuer may use a different order.
+                So, if your attempt ends up with an error AADSTS700027 -
+                "The provided signature value did not match the expected signature value",
+                you may try use only the leaf cert (in PEM/str format) instead.
+
+            .. admonition:: Supporting raw assertion obtained from elsewhere
+
+                *Added in version 1.13.0*:
+                It can also be a completely pre-signed assertion that you've assembled yourself.
+                Simply pass a container containing only the key "client_assertion", like this::
+
+                    {
+                        "client_assertion": "...a JWT with claims aud, exp, iss, jti, nbf, and sub..."
+                    }
+
+            .. admonition:: Supporting workload identity federated by Managed Identity
+
+                *Added in version 1.29.0*:
+                A confidential client app can authenticate via a managed identity.
+                This is known as "federated identity credential (FIC)" or
+                `"Workload identity federation" <https://learn.microsoft.com/entra/workload-id/workload-identity-federation>`_.
+
+                Once you setup the federation, the following declarative API
+                takes care of the managed identity token acquisition for you.
+                You just need to assign ``client_credential``
+                with an instance of :py:class:`msal.SystemAssignedManagedIdentity`
+                or :py:class:`msal.UserAssignedManagedIdentity`, for example::
+
+                    app = msal.ConfidentialClientApplication(
+                        "my_client_id",
+                        client_credential=msal.SystemAssignedManagedIdentity(),
+                        ...)
+
+                or their equivalent ``dict`` representation, such as::
+
+                    app = msal.ConfidentialClientApplication(
+                        "my_client_id",
+                        client_credential={
+                            "ManagedIdentityIdType": "SystemAssigned",
+                            "Id": None},
+                        ...)
+
+                The second example above also imples that you can
+                load the ``dict`` from its equivalent ``json`` representation,
+                which could in turn be read from an ENV VAR, for instance::
+
+                    ## Supposed this is one of your ENV VAR
+                    # CRED={"ManagedIdentityIdType": "SystemAssigned", "Id": null}
+                    ## Now you can read it like this
+                    app = msal.ConfidentialClientApplication(
+                        "my_client_id",
+                        client_credential=json.loads(os.getenv("CRED")),
+                        ...)
+
+                This way, your same Confidential Client Application implementation
+                can be configured to use either client secret or managed identity,
+                without code change. Write once, run anywhere with managed identity.
 
             .. admonition:: Supporting reading client cerficates from PFX files
 
@@ -289,6 +338,7 @@ def __init__(
 
         :type client_credential: Union[dict, str]
 
+
         :param dict client_claims:
             *Added in version 0.5.0*:
             It is a dictionary of extra claims that would be signed by
@@ -691,12 +741,29 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
         if self.app_version:
             default_headers['x-app-ver'] = self.app_version
         default_body = {"client_info": 1}
-        if isinstance(client_credential, dict):
+        if ManagedIdentity.is_managed_identity(client_credential):
+            # Federated Identity Credential (FIC), a.k.a. workload identity
+            # https://learn.microsoft.com/entra/workload-id/workload-identity-federation
+            client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT
+            client_assertion = AutoRefresher(
+                lambda: ManagedIdentityClient(
+                    client_credential, self.http_client,
+                    ).acquire_token_for_client(
+                        resource="api://AzureADTokenExchange",
+                    ).get("access_token"),
+                expires_in=3600,  # Managed Identity token expires in 1 hour
+                )
+            logger.debug("CCA federated by Managed Identity specified via client_credential")
+        elif isinstance(client_credential, dict):
+            assert (("private_key" in client_credential
+                    and "thumbprint" in client_credential) or
+                    "client_assertion" in client_credential)
             client_assertion_type = Client.CLIENT_ASSERTION_TYPE_JWT
             # Use client_credential.get("...") rather than "..." in client_credential
             # so that we can ignore an empty string came from an empty ENV VAR.
             if client_credential.get("client_assertion"):
                 client_assertion = client_credential['client_assertion']
+                logger.debug("CCA authenticated by assertion specified in client_credential")
             else:
                 headers = {}
                 if client_credential.get('public_certificate'):
@@ -726,8 +793,10 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
                 client_assertion = assertion.create_regenerative_assertion(
                     audience=authority.token_endpoint, issuer=self.client_id,
                     additional_claims=self.client_claims or {})
+                logger.debug("CCA authenticated by certificate: {...}")
         else:
             default_body['client_secret'] = client_credential
+            logger.debug("CCA authenticated by secret: ******")
         central_configuration = {
             "authorization_endpoint": authority.authorization_endpoint,
             "token_endpoint": authority.token_endpoint,

msal/managed_identity.py

@@ -161,6 +161,17 @@ def __init__(
     ):
         """Create a managed identity client.
 
+        .. note::
+            You do not have to work with Managed Identity and this class directly.
+
+            A better approach is to use
+            `Workload identity federation <https://learn.microsoft.com/entra/workload-id/workload-identity-federation>`_.
+            Specifically, you can use MSAL's
+            :class:`msal.ConfidentialClientApplication` and feed
+            its :paramref:`msal.ClientApplication.client_credential` parameter
+            with an instance of :class:`msal.SystemAssignedManagedIdentity`
+            or :class:`msal.UserAssignedManagedIdentity`.
+
         :param managed_identity:
             It accepts an instance of :class:`SystemAssignedManagedIdentity`
             or :class:`UserAssignedManagedIdentity`.

sample/.env.sample.entra-id

@@ -15,7 +15,13 @@ CLIENT_ID=<client id>
 #CLIENT_SECRET=<client secret>
 
 # Configure this if you are using a confidential client which has a client credential.
-# Example value: {"private_key_pfx_path": "/path/to/your.pfx"}
+# Example value for using a PFX file: {"private_key_pfx_path": "/path/to/your.pfx"}
+#
+# More examples here for configuring your app federated by a managed identity
+# e.g. {"ManagedIdentityIdType": "SystemAssigned", "Id": null}
+# or {"ManagedIdentityIdType": "ClientId", "Id": "foo"}
+# or {"ManagedIdentityIdType": "ResourceId", "Id": "foo"}
+# or {"ManagedIdentityIdType": "ObjectId", "Id": "foo"}
 CLIENT_CREDENTIAL_JSON=<client credential json>
 
 # Multiple scopes can be added into the same line, separated by a space.