Repository: Azure/azure-cli

Issue:
{Core} AdalAuthentication doesn't honor `scopes` for Track 2 SDK
**Describe the bug**

In Track 2 SDK, `scopes` (`resource`) is managed by SDK, instead of Azure CLI. For example, in [`AzureAppConfigurationClient`](https://github.com/Azure/azure-sdk-for-python/blob/022bc80f2bc645d8d82d647620376fcea2229522/sdk/appconfiguration/azure-appconfiguration/azure/appconfiguration/_azure_appconfiguration_client.py#L128):

```py
if aad_mode:
    scope = base_url.strip("/") + "/.default"
```

But in `get_token`, `scopes` is not honored (discarded), resulting in getting a token for a wrong `scopes` (ARM `https://management.core.windows.net/` by default):

https://github.com/Azure/azure-cli/blob/c2b9a2c5cfe34b108b52b49ac50c6802ac4d729f/src/azure-cli-core/azure/cli/core/adal_authentication.py#L61-L67


Discussion:
yonzhan: Core
jiasli: Unfortunately, due an incorrect `credential_scopes` managing logic in existing SDKs (https://github.com/Azure/azure-sdk-for-python/issues/12947), even though the bug has been fixed in https://github.com/Azure/autorest.python/pull/745, merging this PR will break all commands using **Track 2 SDKs with sovereign clouds**, as `credential_scopes` will be passed to `scopes` as 

```
['https://management.azure.com/.default', 'https://management.core.chinacloudapi.cn/.default']
```

`'https://management.core.chinacloudapi.cn/.default'` will be discarded.

Patch:
diff --git a/src/azure-cli-core/azure/cli/core/_profile.py b/src/azure-cli-core/azure/cli/core/_profile.py
index 3dee2b626..79e08e5c9 100644
--- a/src/azure-cli-core/azure/cli/core/_profile.py
+++ b/src/azure-cli-core/azure/cli/core/_profile.py
@@ -559,26 +559,35 @@ class Profile:
                     external_tenants_info.append(sub[_TENANT_ID])
 
         if identity_type is None:
-            def _retrieve_token():
+            def _retrieve_token(sdk_resource=None):
+                # When called by
+                #   - Track 1 SDK, use `resource` specified by CLI
+                #   - Track 2 SDK, use `sdk_resource` specified by SDK and ignore `resource` specified by CLI
+                token_resource = sdk_resource or resource
+                logger.debug("Retrieving token from ADAL for resource %r", token_resource)
+
                 if in_cloud_console() and account[_USER_ENTITY].get(_CLOUD_SHELL_ID):
-                    return self._get_token_from_cloud_shell(resource)
+                    return self._get_token_from_cloud_shell(token_resource)
                 if user_type == _USER:
                     return self._creds_cache.retrieve_token_for_user(username_or_sp_id,
-                                                                     account[_TENANT_ID], resource)
+                                                                     account[_TENANT_ID], token_resource)
                 use_cert_sn_issuer = account[_USER_ENTITY].get(_SERVICE_PRINCIPAL_CERT_SN_ISSUER_AUTH)
-                return self._creds_cache.retrieve_token_for_service_principal(username_or_sp_id, resource,
+                return self._creds_cache.retrieve_token_for_service_principal(username_or_sp_id, token_resource,
                                                                               account[_TENANT_ID],
                                                                               use_cert_sn_issuer)
 
-            def _retrieve_tokens_from_external_tenants():
+            def _retrieve_tokens_from_external_tenants(sdk_resource=None):
+                token_resource = sdk_resource or resource
+                logger.debug("Retrieving token from ADAL for external tenants and resource %r", token_resource)
+
                 external_tokens = []
                 for sub_tenant_id in external_tenants_info:
                     if user_type == _USER:
                         external_tokens.append(self._creds_cache.retrieve_token_for_user(
-                            username_or_sp_id, sub_tenant_id, resource))
+                            username_or_sp_id, sub_tenant_id, token_resource))
                     else:
                         external_tokens.append(self._creds_cache.retrieve_token_for_service_principal(
-                            username_or_sp_id, resource, sub_tenant_id, resource))
+                            username_or_sp_id, token_resource, sub_tenant_id, token_resource))
                 return external_tokens
 
             from azure.cli.core.adal_authentication import AdalAuthentication
@@ -621,6 +630,8 @@ class Profile:
         return username_or_sp_id, sp_secret, None, str(account[_TENANT_ID])
 
     def get_raw_token(self, resource=None, subscription=None, tenant=None):
+        logger.debug("Profile.get_raw_token invoked with resource=%r, subscription=%r, tenant=%r",
+                     resource, subscription, tenant)
         if subscription and tenant:
             raise CLIError("Please specify only one of subscription and tenant, not both")
         account = self.get_subscription(subscription)
diff --git a/src/azure-cli-core/azure/cli/core/adal_authentication.py b/src/azure-cli-core/azure/cli/core/adal_authentication.py
index 7ff90a1bd..a4496b05d 100644
--- a/src/azure-cli-core/azure/cli/core/adal_authentication.py
+++ b/src/azure-cli-core/azure/cli/core/adal_authentication.py
@@ -10,9 +10,12 @@ import adal
 from msrest.authentication import Authentication
 from msrestazure.azure_active_directory import MSIAuthentication
 from azure.core.credentials import AccessToken
-from azure.cli.core.util import in_cloud_console
+from azure.cli.core.util import in_cloud_console, scopes_to_resource
 
 from knack.util import CLIError
+from knack.log import get_logger
+
+logger = get_logger(__name__)
 
 
 class AdalAuthentication(Authentication):  # pylint: disable=too-few-public-methods
@@ -21,12 +24,15 @@ class AdalAuthentication(Authentication):  # pylint: disable=too-few-public-meth
         self._token_retriever = token_retriever
         self._external_tenant_token_retriever = external_tenant_token_retriever
 
-    def _get_token(self):
+    def _get_token(self, sdk_resource=None):
+        """
+        :param sdk_resource: `resource` converted from Track 2 SDK's `scopes`
+        """
         external_tenant_tokens = None
         try:
-            scheme, token, full_token = self._token_retriever()
+            scheme, token, full_token = self._token_retriever(sdk_resource)
             if self._external_tenant_token_retriever:
-                external_tenant_tokens = self._external_tenant_token_retriever()
+                external_tenant_tokens = self._external_tenant_token_retriever(sdk_resource)
         except CLIError as err:
             if in_cloud_console():
                 AdalAuthentication._log_hostname()
@@ -60,7 +66,16 @@ class AdalAuthentication(Authentication):  # pylint: disable=too-few-public-meth
 
     # This method is exposed for Azure Core.
     def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
-        _, token, full_token, _ = self._get_token()
+        logger.debug("AdalAuthentication.get_token invoked by Track 2 SDK with scopes=%s", scopes)
+
+        # Deal with an old Track 2 SDK issue where the default credential_scopes is extended with
+        # custom credential_scopes. Instead, credential_scopes should be replaced by custom credential_scopes.
+        # https://github.com/Azure/azure-sdk-for-python/issues/12947
+        # We simply remove the first one if there are multiple scopes provided.
+        if len(scopes) > 1:
+            scopes = scopes[1:]
+
+        _, token, full_token, _ = self._get_token(scopes_to_resource(scopes))
         try:
             return AccessToken(token, int(full_token['expiresIn'] + time.time()))
         except KeyError:  # needed to deal with differing unserialized MSI token payload
@@ -68,6 +83,7 @@ class AdalAuthentication(Authentication):  # pylint: disable=too-few-public-meth
 
     # This method is exposed for msrest.
     def signed_session(self, session=None):  # pylint: disable=arguments-differ
+        logger.debug("AdalAuthentication.signed_session invoked by Track 1 SDK")
         session = session or super(AdalAuthentication, self).signed_session()
 
         scheme, token, _, external_tenant_tokens = self._get_token()
@@ -82,8 +98,6 @@ class AdalAuthentication(Authentication):  # pylint: disable=too-few-public-meth
     @staticmethod
     def _log_hostname():
         import socket
-        from knack.log import get_logger
-        logger = get_logger(__name__)
         logger.warning("A Cloud Shell credential problem occurred. When you report the issue with the error "
                        "below, please mention the hostname '%s'", socket.gethostname())
 
@@ -91,13 +105,13 @@ class AdalAuthentication(Authentication):  # pylint: disable=too-few-public-meth
 class MSIAuthenticationWrapper(MSIAuthentication):
     # This method is exposed for Azure Core. Add *scopes, **kwargs to fit azure.core requirement
     def get_token(self, *scopes, **kwargs):  # pylint:disable=unused-argument
+        logger.debug("MSIAuthenticationWrapper.get_token invoked by Track 2 SDK with scopes=%s", scopes)
+        self.resource = scopes_to_resource(scopes)
         self.set_token()
         return AccessToken(self.token['access_token'], int(self.token['expires_on']))
 
     def set_token(self):
         import traceback
-        from knack.log import get_logger
-        logger = get_logger(__name__)
         from azure.cli.core.azclierror import AzureConnectionError, AzureResponseError
         try:
             super(MSIAuthenticationWrapper, self).set_token()
@@ -117,3 +131,7 @@ class MSIAuthenticationWrapper(MSIAuthentication):
                          traceback.format_exc())
             raise AzureConnectionError('MSI endpoint is not responding. Please make sure MSI is configured correctly.\n'
                                        'Error detail: {}'.format(str(err)))
+
+    def signed_session(self, session=None):
+        logger.debug("MSIAuthenticationWrapper.signed_session invoked by Track 1 SDK")
+        super().signed_session(session)
diff --git a/src/azure-cli-core/azure/cli/core/commands/client_factory.py b/src/azure-cli-core/azure/cli/core/commands/client_factory.py
index 8a9472469..c8c4b5aa1 100644
--- a/src/azure-cli-core/azure/cli/core/commands/client_factory.py
+++ b/src/azure-cli-core/azure/cli/core/commands/client_factory.py
@@ -150,6 +150,7 @@ def _get_mgmt_service_client(cli_ctx,
                              aux_tenants=None,
                              **kwargs):
     from azure.cli.core._profile import Profile
+    from azure.cli.core.util import resource_to_scopes
     logger.debug('Getting management service client client_type=%s', client_type.__name__)
     resource = resource or cli_ctx.cloud.endpoints.active_directory_resource_id
     profile = Profile(cli_ctx=cli_ctx)
@@ -169,6 +170,7 @@ def _get_mgmt_service_client(cli_ctx,
 
     if is_track2(client_type):
         client_kwargs.update(configure_common_settings_track2(cli_ctx))
+        client_kwargs['credential_scopes'] = resource_to_scopes(resource)
 
     if subscription_bound:
         client = client_type(cred, subscription_id, **client_kwargs)
diff --git a/src/azure-cli-core/azure/cli/core/util.py b/src/azure-cli-core/azure/cli/core/util.py
index 283f7e9bb..e85f32be8 100644
--- a/src/azure-cli-core/azure/cli/core/util.py
+++ b/src/azure-cli-core/azure/cli/core/util.py
@@ -1165,3 +1165,36 @@ def handle_version_update():
             refresh_known_clouds()
     except Exception as ex:  # pylint: disable=broad-except
         logger.warning(ex)
+
+
+def resource_to_scopes(resource):
+    """Convert the ADAL resource ID to MSAL scopes by appending the /.default suffix and return a list.
+    For example: 'https://management.core.windows.net/' -> ['https://management.core.windows.net/.default']
+    :param resource: The ADAL resource ID
+    :return: A list of scopes
+    """
+    if 'datalake' in resource or 'batch' in resource or 'database' in resource:
+        # For datalake, batch and database, the slash must be doubled due to service issue, like
+        #   https://datalake.azure.net//.default
+        # TODO: This should be fixed on the service side.
+        scope = resource + '/.default'
+    else:
+        scope = resource.rstrip('/') + '/.default'
+    return [scope]
+
+
+def scopes_to_resource(scopes):
+    """Convert MSAL scopes to ADAL resource by stripping the /.default suffix and return a str.
+    For example: ['https://management.core.windows.net/.default'] -> 'https://management.core.windows.net/'
+
+    :param scopes: The MSAL scopes. It can be a list or tuple of string
+    :return: The ADAL resource
+    :rtype: str
+    """
+    scope = scopes[0]
+    if scope.endswith(".default"):
+        scope = scope[:-len(".default")]
+
+    # Trim extra ending slashes. https://datalake.azure.net// -> https://datalake.azure.net/
+    scope = scope.rstrip('/') + '/'
+    return scope
diff --git a/src/azure-cli/azure/cli/command_modules/storage/_validators.py b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
index 458ac3d23..861ef2671 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators.py
@@ -129,8 +129,7 @@ def validate_client_parameters(cmd, namespace):
             if is_storagev2(prefix):
                 from azure.cli.core._profile import Profile
                 profile = Profile(cli_ctx=cmd.cli_ctx)
-                n.token_credential, _, _ = profile.get_login_credentials(
-                    resource="https://storage.azure.com", subscription_id=n._subscription)
+                n.token_credential, _, _ = profile.get_login_credentials(subscription_id=n._subscription)
             # Otherwise, we will assume it is in track1 and keep previous token updater
             else:
                 n.token_credential = _create_token_credential(cmd.cli_ctx)
diff --git a/src/azure-cli/azure/cli/command_modules/storage/_validators_azure_stack.py b/src/azure-cli/azure/cli/command_modules/storage/_validators_azure_stack.py
index 9c79e5f4a..c9ab00d3b 100644
--- a/src/azure-cli/azure/cli/command_modules/storage/_validators_azure_stack.py
+++ b/src/azure-cli/azure/cli/command_modules/storage/_validators_azure_stack.py
@@ -128,8 +128,7 @@ def validate_client_parameters(cmd, namespace):
             if is_storagev2(prefix):
                 from azure.cli.core._profile import Profile
                 profile = Profile(cli_ctx=cmd.cli_ctx)
-                n.token_credential, _, _ = profile.get_login_credentials(
-                    resource="https://storage.azure.com", subscription_id=n._subscription)
+                n.token_credential, _, _ = profile.get_login_credentials(subscription_id=n._subscription)
             # Otherwise, we will assume it is in track1 and keep previous token updater
             else:
                 n.token_credential = _create_token_credential(cmd.cli_ctx)


Test patch:
diff --git a/src/azure-cli-core/azure/cli/core/tests/test_util.py b/src/azure-cli-core/azure/cli/core/tests/test_util.py
index 821a35370..84a57be1e 100644
--- a/src/azure-cli-core/azure/cli/core/tests/test_util.py
+++ b/src/azure-cli-core/azure/cli/core/tests/test_util.py
@@ -377,6 +377,29 @@ class TestUtils(unittest.TestCase):
             request = send_mock.call_args.args[1]
             self.assertEqual(request.headers['User-Agent'], get_az_rest_user_agent() + ' env-ua ARG-UA')
 
+    def test_scopes_to_resource(self):
+        from azure.cli.core.util import scopes_to_resource
+        # scopes as a list
+        self.assertEqual(scopes_to_resource(['https://management.core.windows.net/.default']),
+                         'https://management.core.windows.net/')
+        # scopes as a tuple
+        self.assertEqual(scopes_to_resource(('https://storage.azure.com/.default',)),
+                         'https://storage.azure.com/')
+
+        # Double slashes are reduced
+        self.assertEqual(scopes_to_resource(['https://datalake.azure.net//.default']),
+                         'https://datalake.azure.net/')
+
+    def test_resource_to_scopes(self):
+        from azure.cli.core.util import resource_to_scopes
+        # resource converted to a scopes list
+        self.assertEqual(resource_to_scopes('https://management.core.windows.net/'),
+                         ['https://management.core.windows.net/.default'])
+
+        # Use double slashes for certain services
+        self.assertEqual(resource_to_scopes('https://datalake.azure.net/'),
+                         ['https://datalake.azure.net//.default'])
+
 
 class TestBase64ToHex(unittest.TestCase):