cloud-shell-auth

Author: jiasli

src/azure-cli-core/azure/cli/core/_profile.py

@@ -284,17 +284,16 @@ def login_with_managed_identity(self, identity_id=None, allow_no_subscriptions=N
 
     def login_in_cloud_shell(self):
         import jwt
-        from azure.cli.core.auth.adal_authentication import MSIAuthenticationWrapper
+        from .auth.msal_credentials import CloudShellCredential
 
-        msi_creds = MSIAuthenticationWrapper(resource=self.cli_ctx.cloud.endpoints.active_directory_resource_id)
-        token_entry = msi_creds.token
-        token = token_entry['access_token']
-        logger.info('MSI: token was retrieved. Now trying to initialize local accounts...')
+        cred = CloudShellCredential()
+        token = cred.get_token(*self._arm_scope).token
+        logger.info('Cloud Shell token was retrieved. Now trying to initialize local accounts...')
         decode = jwt.decode(token, algorithms=['RS256'], options={"verify_signature": False})
         tenant = decode['tid']
 
         subscription_finder = SubscriptionFinder(self.cli_ctx)
-        subscriptions = subscription_finder.find_using_specific_tenant(tenant, msi_creds)
+        subscriptions = subscription_finder.find_using_specific_tenant(tenant, cred)
         if not subscriptions:
             raise CLIError('No subscriptions were found in the cloud shell')
         user = decode.get('unique_name', 'N/A')
@@ -351,11 +350,19 @@ def get_login_credentials(self, resource=None, client_id=None, subscription_id=N
 
         managed_identity_type, managed_identity_id = Profile._try_parse_msi_account_name(account)
 
-        # Cloud Shell is just a system assignment managed identity
         if in_cloud_console() and account[_USER_ENTITY].get(_CLOUD_SHELL_ID):
-            managed_identity_type = MsiAccountTypes.system_assigned
+            # Cloud Shell
+            from .auth.msal_credentials import CloudShellCredential
+            from azure.cli.core.auth.credential_adaptor import CredentialAdaptor
+            cs_cred = CloudShellCredential()
+            # The cloud shell credential must be wrapped by CredentialAdaptor so that it can work with Track 1 SDKs.
+            cred = CredentialAdaptor(cs_cred, resource=resource)
 
-        if managed_identity_type is None:
+        elif managed_identity_type:
+            # managed identity
+            cred = MsiAccountTypes.msi_auth_factory(managed_identity_type, managed_identity_id, resource)
+
+        else:
             # user and service principal
             external_tenants = []
             if aux_tenants:
@@ -375,9 +382,7 @@ def get_login_credentials(self, resource=None, client_id=None, subscription_id=N
             cred = CredentialAdaptor(credential,
                                      auxiliary_credentials=external_credentials,
                                      resource=resource)
-        else:
-            # managed identity
-            cred = MsiAccountTypes.msi_auth_factory(managed_identity_type, managed_identity_id, resource)
+
         return (cred,
                 str(account[_SUBSCRIPTION_ID]),
                 str(account[_TENANT_ID]))
@@ -397,27 +402,27 @@ def get_raw_token(self, resource=None, scopes=None, subscription=None, tenant=No
 
         account = self.get_subscription(subscription)
 
-        identity_type, identity_id = Profile._try_parse_msi_account_name(account)
-        if identity_type:
+        managed_identity_type, managed_identity_id = Profile._try_parse_msi_account_name(account)
+
+        if in_cloud_console() and account[_USER_ENTITY].get(_CLOUD_SHELL_ID):
+            # Cloud Shell
+            if tenant:
+                raise CLIError("Tenant shouldn't be specified for Cloud Shell account")
+            from .auth.msal_credentials import CloudShellCredential
+            cred = CloudShellCredential()
+
+        elif managed_identity_type:
             # managed identity
             if tenant:
                 raise CLIError("Tenant shouldn't be specified for managed identity account")
             from .auth.util import scopes_to_resource
-            msi_creds = MsiAccountTypes.msi_auth_factory(identity_type, identity_id,
-                                                         scopes_to_resource(scopes))
-            sdk_token = msi_creds.get_token(*scopes)
-        elif in_cloud_console() and account[_USER_ENTITY].get(_CLOUD_SHELL_ID):
-            # Cloud Shell, which is just a system-assigned managed identity.
-            if tenant:
-                raise CLIError("Tenant shouldn't be specified for Cloud Shell account")
-            from .auth.util import scopes_to_resource
-            msi_creds = MsiAccountTypes.msi_auth_factory(MsiAccountTypes.system_assigned, identity_id,
-                                                         scopes_to_resource(scopes))
-            sdk_token = msi_creds.get_token(*scopes)
+            cred = MsiAccountTypes.msi_auth_factory(managed_identity_type, managed_identity_id,
+                                                    scopes_to_resource(scopes))
+
         else:
-            credential = self._create_credential(account, tenant)
-            sdk_token = credential.get_token(*scopes)
+            cred = self._create_credential(account, tenant)
 
+        sdk_token = cred.get_token(*scopes)
         # Convert epoch int 'expires_on' to datetime string 'expiresOn' for backward compatibility
         # WARNING: expiresOn is deprecated and will be removed in future release.
         import datetime
@@ -429,11 +434,11 @@ def get_raw_token(self, resource=None, scopes=None, subscription=None, tenant=No
             'expiresOn': expiresOn  # datetime string, like "2020-11-12 13:50:47.114324"
         }
 
-        # (tokenType, accessToken, tokenEntry)
-        creds = 'Bearer', sdk_token.token, token_entry
+        # Build a tuple of (token_type, token, token_entry)
+        token_tuple = 'Bearer', sdk_token.token, token_entry
 
-        # (cred, subscription, tenant)
-        return (creds,
+        # Return a tuple of (token_tuple, subscription, tenant)
+        return (token_tuple,
                 None if tenant else str(account[_SUBSCRIPTION_ID]),
                 str(tenant if tenant else account[_TENANT_ID]))
 

src/azure-cli-core/azure/cli/core/auth/adal_authentication.py

@@ -24,7 +24,7 @@ def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
                 # Use MSAL to get VM SSH certificate
                 import msal
                 from .util import check_result, build_sdk_access_token
-                from .identity import AZURE_CLI_CLIENT_ID
+                from .constants import AZURE_CLI_CLIENT_ID
                 app = msal.PublicClientApplication(
                     AZURE_CLI_CLIENT_ID,  # Use a real client_id, so that cache would work
                     # TODO: This PoC does not currently maintain a token cache;

src/azure-cli-core/azure/cli/core/auth/constants.py

@@ -0,0 +1,6 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+AZURE_CLI_CLIENT_ID = '04b07795-8ddb-461a-bbee-02f9e1bf7b46'

src/azure-cli-core/azure/cli/core/auth/identity.py

@@ -13,12 +13,11 @@
 from knack.util import CLIError
 from msal import PublicClientApplication, ConfidentialClientApplication
 
+from .constants import AZURE_CLI_CLIENT_ID
 from .msal_credentials import UserCredential, ServicePrincipalCredential
 from .persistence import load_persisted_token_cache, file_extensions, load_secret_store
 from .util import check_result
 
-AZURE_CLI_CLIENT_ID = '04b07795-8ddb-461a-bbee-02f9e1bf7b46'
-
 # Service principal entry properties. Names are taken from OAuth 2.0 client credentials flow parameters:
 # https://learn.microsoft.com/en-us/entra/identity-platform/v2-oauth2-client-creds-grant-flow
 _TENANT = 'tenant'

src/azure-cli-core/azure/cli/core/auth/msal_credentials.py

@@ -20,6 +20,7 @@
 from knack.util import CLIError
 from msal import PublicClientApplication, ConfidentialClientApplication
 
+from .constants import AZURE_CLI_CLIENT_ID
 from .util import check_result, build_sdk_access_token
 
 logger = get_logger(__name__)
@@ -108,3 +109,25 @@ def get_token(self, *scopes, **kwargs):
         result = self._msal_app.acquire_token_for_client(list(scopes), **kwargs)
         check_result(result)
         return build_sdk_access_token(result)
+
+
+class CloudShellCredential:  # pylint: disable=too-few-public-methods
+    # Cloud Shell acts as a "broker" to obtain access token for the user account, so even though it uses
+    # managed identity protocol, it returns a user token.
+    # That's why MSAL uses acquire_token_interactive to retrieve an access token in Cloud Shell.
+    # See https://github.com/Azure/azure-cli/pull/29637
+
+    def __init__(self):
+        self._msal_app = PublicClientApplication(
+            AZURE_CLI_CLIENT_ID,  # Use a real client_id, so that cache would work
+            # TODO: We currently don't maintain an MSAL token cache as Cloud Shell already has its own token cache.
+            #   Ideally we should also use an MSAL token cache.
+            #   token_cache=...
+        )
+
+    def get_token(self, *scopes, **kwargs):
+        logger.debug("CloudShellCredential.get_token: scopes=%r, kwargs=%r", scopes, kwargs)
+        # kwargs is already sanitized by CredentialAdaptor, so it can be safely passed to MSAL
+        result = self._msal_app.acquire_token_interactive(list(scopes), prompt="none", **kwargs)
+        check_result(result, scopes=scopes)
+        return build_sdk_access_token(result)