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,
     )
 

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

@@ -12,6 +12,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
@@ -22,6 +23,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.
@@ -230,32 +232,80 @@ 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.
 
         :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
@@ -650,13 +700,27 @@ 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
             if 'client_assertion' in client_credential:
                 client_assertion = client_credential['client_assertion']
+                logger.debug("CCA authenticated by assertion specified in client_credential")
             else:
                 headers = {}
                 if 'public_certificate' in client_credential:
@@ -671,14 +735,16 @@ def _build_client(self, client_credential, authority, skip_regional_client=False
                         _str2bytes(client_credential["passphrase"]),
                         backend=default_backend(),  # It was a required param until 2020
                         )
-                assertion = JwtAssertionCreator(
+                assertion_creator = JwtAssertionCreator(
                     unencrypted_private_key, algorithm="RS256",
                     sha1_thumbprint=client_credential.get("thumbprint"), headers=headers)
-                client_assertion = assertion.create_regenerative_assertion(
+                client_assertion = assertion_creator.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

@@ -137,6 +137,17 @@ def __init__(
         self, managed_identity, *, http_client, token_cache=None, http_cache=None):
         """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 dict managed_identity:
             It accepts an instance of :class:`SystemAssignedManagedIdentity`
             or :class:`UserAssignedManagedIdentity`.

sample/.env.sample.entra-id

@@ -11,8 +11,16 @@ AUTHORITY=<authority url>
 # The following variables are required for the app to run.
 CLIENT_ID=<client id>
 
-# Leave it empty if you are using a public client which has no client secret.
+
+# Leave it empty if your app does not have a client secret
 CLIENT_SECRET=<client secret>
+# Leave it intact if you do not intend to use client credential.
+# One of the usage here is to configure 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=null
 
 # Multiple scopes can be added into the same line, separated by a space.
 # Here we use a Microsoft Graph API as an example

sample/confidential_client_secret_sample.py

@@ -43,7 +43,8 @@
     os.getenv('CLIENT_ID'),
     authority=os.getenv('AUTHORITY'),  # For Entra ID or External ID
     oidc_authority=os.getenv('OIDC_AUTHORITY'),  # For External ID with custom domain
-    client_credential=os.getenv('CLIENT_SECRET'),
+    client_credential=os.getenv('CLIENT_SECRET') or json.loads(
+        os.getenv("CLIENT_CREDENTIAL_JSON")),
     token_cache=global_token_cache,  # Let this app (re)use an existing token cache.
         # If absent, ClientApplication will create its own empty token cache
     )