PoC: Managed Identity for Azure VM

Author: Ray Luo

msal/application.py

@@ -2001,6 +2001,21 @@ 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 self.client_credential is None:
+            from .imds import _scope_to_resource, _obtain_token
+            response = _obtain_token(
+                self.http_client,
+                " ".join(map(_scope_to_resource, scopes)),
+                client_id=self.client_id,  # None for system-assigned, GUID for user-assigned
+                )
+            if "error" not in response:
+                self.token_cache.add(dict(
+                    client_id=self.client_id,
+                    scope=response["scope"].split() if "scope" in response else scopes,
+                    token_endpoint=self.authority.token_endpoint,
+                    response=response.copy(),
+                    ))
+            return response
         if self.authority.tenant.lower() in ["common", "organizations"]:
             warnings.warn(
                 "Using /common or /organizations authority "

msal/imds.py

@@ -0,0 +1,46 @@
+# Copyright (c) Microsoft Corporation.
+# All rights reserved.
+#
+# This code is licensed under the MIT License.
+import json
+import logging
+try:  # Python 2
+    from urlparse import urlparse
+except:  # Python 3
+    from urllib.parse import urlparse
+
+logger = logging.getLogger(__name__)
+
+def _scope_to_resource(scope):  # This is an experimental reasonable-effort approach
+    u = urlparse(scope)
+    if u.scheme:
+        return "{}://{}".format(u.scheme, u.netloc)
+    return scope  # There is no much else we can do here
+
+def _obtain_token(http_client, resource, client_id=None):
+    # Based on https://docs.microsoft.com/en-us/azure/active-directory/managed-identities-azure-resources/how-to-use-vm-token#get-a-token-using-http
+    params = {
+        "api-version": "2018-02-01",
+        "resource": resource,
+        }
+    if client_id:
+        params["client_id"] = client_id
+    resp = http_client.get(
+        "http://169.254.169.254/metadata/identity/oauth2/token",
+        params=params,
+        headers={"Metadata": "true"},
+        )
+    try:
+        payload = json.loads(resp.text)
+        if payload.get("access_token") and payload.get("expires_in"):
+            return {  # Normalizing the payload into OAuth2 format
+                "access_token": payload["access_token"],
+                "expires_in": int(payload["expires_in"]),
+                "resource": payload.get("resource"),
+                "token_type": payload.get("token_type", "Bearer"),
+                }
+        return payload  # Typically an error
+    except ValueError:
+        logger.debug("IMDS emits unexpected payload: %s", resp.text)
+        raise
+

tests/msaltest.py

@@ -47,8 +47,11 @@ def _input_scopes():
         raise ValueError("SSH Cert scope shall be tested by its dedicated functions")
     return scopes
 
-def _select_account(app):
+def _select_account(app, show_confidential_app_placeholder=False):
     accounts = app.get_accounts()
+    if show_confidential_app_placeholder and isinstance(
+            app, msal.ConfidentialClientApplication):
+        accounts.insert(0, {"username": "This Client"})
     if accounts:
         return _select_options(
             accounts,
@@ -60,11 +63,11 @@ def _select_account(app):
 
 def acquire_token_silent(app):
     """acquire_token_silent() - with an account already signed into MSAL Python."""
-    account = _select_account(app)
+    account = _select_account(app, show_confidential_app_placeholder=True)
     if account:
         pprint.pprint(app.acquire_token_silent(
             _input_scopes(),
-            account=account,
+            account=account if "home_account_id" in account else None,
             force_refresh=_input_boolean("Bypass MSAL Python's token cache?"),
             ))
 
@@ -138,6 +141,10 @@ def remove_account(app):
         app.remove_account(account)
         print('Account "{}" and/or its token(s) are signed out from MSAL Python'.format(account["username"]))
 
+def acquire_token_for_client(app):
+    """acquire_token_for_client() - Only for confidential client"""
+    pprint.pprint(app.acquire_token_for_client(_input_scopes()))
+
 def exit(app):
     """Exit"""
     bug_link = (
@@ -154,13 +161,12 @@ def main():
         {"client_id": AZURE_CLI, "name": "Azure CLI (Correctly configured for MSA-PT)"},
         {"client_id": VISUAL_STUDIO, "name": "Visual Studio (Correctly configured for MSA-PT)"},
         {"client_id": "95de633a-083e-42f5-b444-a4295d8e9314", "name": "Whiteboard Services (Non MSA-PT app. Accepts AAD & MSA accounts.)"},
+        {"client_id": None, "client_secret": None, "name": "System-assigned Managed Identity (Only works when running inside a supported environment, such as Azure VM)"},
         ],
         option_renderer=lambda a: a["name"],
         header="Impersonate this app (or you can type in the client_id of your own app)",
         accept_nonempty_string=True)
-    app = msal.PublicClientApplication(
-        chosen_app["client_id"] if isinstance(chosen_app, dict) else chosen_app,
-        authority=_select_options([
+    authority = _select_options([
             "https://login.microsoftonline.com/common",
             "https://login.microsoftonline.com/organizations",
             "https://login.microsoftonline.com/microsoft.onmicrosoft.com",
@@ -169,21 +175,33 @@ def main():
             ],
             header="Input authority (Note that MSA-PT apps would NOT use the /common authority)",
             accept_nonempty_string=True,
-            ),
-        allow_broker=_input_boolean("Allow broker? (Azure CLI currently only supports @microsoft.com accounts when enabling broker)"),
-        )
+            )
+    if isinstance(chosen_app, dict) and "client_secret" in chosen_app:
+        app = msal.ConfidentialClientApplication(
+            chosen_app["client_id"],
+            client_credential=chosen_app["client_secret"],
+            authority=authority,
+            )
+    else:
+        app = msal.PublicClientApplication(
+            chosen_app["client_id"] if isinstance(chosen_app, dict) else chosen_app,
+            authority=authority,
+            allow_broker=_input_boolean("Allow broker? (Azure CLI currently only supports @microsoft.com accounts when enabling broker)"),
+            )
     if _input_boolean("Enable MSAL Python's DEBUG log?"):
         logging.basicConfig(level=logging.DEBUG)
     while True:
-        func = _select_options([
+        func = _select_options(list(filter(None, [
             acquire_token_silent,
             acquire_token_interactive,
             acquire_token_by_username_password,
             acquire_ssh_cert_silently,
             acquire_ssh_cert_interactive,
             remove_account,
+            acquire_token_for_client if isinstance(
+                app, msal.ConfidentialClientApplication) else None,
             exit,
-            ], option_renderer=lambda f: f.__doc__, header="MSAL Python APIs:")
+            ])), option_renderer=lambda f: f.__doc__, header="MSAL Python APIs:")
         try:
             func(app)
         except ValueError as e: