track1-auth

Author: jiasli

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

@@ -385,16 +385,13 @@ def logout_all(self):
         identity.logout_all_users()
         identity.logout_all_service_principal()
 
-    def get_login_credentials(self, resource=None, subscription_id=None, aux_subscriptions=None, aux_tenants=None):
+    def get_login_credentials(self, subscription_id=None, aux_subscriptions=None, aux_tenants=None):
         """Get a CredentialAdaptor instance to be used with both Track 1 and Track 2 SDKs.
 
-        :param resource: The resource ID to acquire an access token. Only provide it for Track 1 SDKs.
         :param subscription_id:
         :param aux_subscriptions:
         :param aux_tenants:
         """
-        resource = resource or self.cli_ctx.cloud.endpoints.active_directory_resource_id
-
         if aux_tenants and aux_subscriptions:
             raise CLIError("Please specify only one of aux_subscriptions and aux_tenants, not both")
 
@@ -407,17 +404,21 @@ def get_login_credentials(self, resource=None, subscription_id=None, aux_subscri
             from .auth.msal_credentials import CloudShellCredential
             from azure.cli.core.auth.credential_adaptor import CredentialAdaptor
             # The credential must be wrapped by CredentialAdaptor so that it can work with Track 1 SDKs.
-            cred = CredentialAdaptor(CloudShellCredential(), resource=resource)
+            cred = CredentialAdaptor(CloudShellCredential())
 
         elif managed_identity_type:
             # managed identity
             if _on_azure_arc():
                 from .auth.msal_credentials import ManagedIdentityCredential
                 from azure.cli.core.auth.credential_adaptor import CredentialAdaptor
                 # The credential must be wrapped by CredentialAdaptor so that it can work with Track 1 SDKs.
-                cred = CredentialAdaptor(ManagedIdentityCredential(), resource=resource)
+                cred = CredentialAdaptor(ManagedIdentityCredential())
             else:
-                cred = MsiAccountTypes.msi_auth_factory(managed_identity_type, managed_identity_id, resource)
+                # The resource is merely used by msrestazure to get the first access token.
+                # It is not actually used in an API invocation.
+                cred = MsiAccountTypes.msi_auth_factory(
+                    managed_identity_type, managed_identity_id,
+                    self.cli_ctx.cloud.endpoints.active_directory_resource_id)
 
         else:
             # user and service principal
@@ -436,9 +437,7 @@ def get_login_credentials(self, resource=None, subscription_id=None, aux_subscri
             for external_tenant in external_tenants:
                 external_credentials.append(self._create_credential(account, tenant_id=external_tenant))
             from azure.cli.core.auth.credential_adaptor import CredentialAdaptor
-            cred = CredentialAdaptor(credential,
-                                     auxiliary_credentials=external_credentials,
-                                     resource=resource)
+            cred = CredentialAdaptor(credential, auxiliary_credentials=external_credentials)
 
         return (cred,
                 str(account[_SUBSCRIPTION_ID]),

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

@@ -13,59 +13,33 @@
 
 
 class CredentialAdaptor:
-    def __init__(self, credential, resource=None, auxiliary_credentials=None):
-        """
-        Adaptor to both
-        - Track 1: msrest.authentication.Authentication, which exposes signed_session
-        - Track 2: azure.core.credentials.TokenCredential, which exposes get_token
+    def __init__(self, credential, auxiliary_credentials=None):
+        """Cross-tenant credential adaptor. It takes a main credential and auxiliary credentials.
+
+        It implements Track 2 SDK's azure.core.credentials.TokenCredential by exposing get_token.
 
         :param credential: Main credential from .msal_authentication
-        :param resource: AAD resource for Track 1 only
         :param auxiliary_credentials: Credentials from .msal_authentication for cross tenant authentication.
             Details about cross tenant authentication:
             https://learn.microsoft.com/en-us/azure/azure-resource-manager/management/authenticate-multi-tenant
         """
 
         self._credential = credential
         self._auxiliary_credentials = auxiliary_credentials
-        self._resource = resource
-
-    def _get_token(self, scopes=None, **kwargs):
-        external_tenant_tokens = []
-        # If scopes is not provided, use CLI-managed resource
-        scopes = scopes or resource_to_scopes(self._resource)
-        try:
-            token = self._credential.get_token(*scopes, **kwargs)
-            if self._auxiliary_credentials:
-                external_tenant_tokens = [cred.get_token(*scopes) for cred in self._auxiliary_credentials]
-            return token, external_tenant_tokens
-        except requests.exceptions.SSLError as err:
-            from azure.cli.core.util import SSLERROR_TEMPLATE
-            raise CLIError(SSLERROR_TEMPLATE.format(str(err)))
-
-    def signed_session(self, session=None):
-        logger.debug("CredentialAdaptor.signed_session")
-        session = session or requests.Session()
-        token, external_tenant_tokens = self._get_token()
-        header = "{} {}".format('Bearer', token.token)
-        session.headers['Authorization'] = header
-        if external_tenant_tokens:
-            aux_tokens = ';'.join(['{} {}'.format('Bearer', tokens2.token) for tokens2 in external_tenant_tokens])
-            session.headers['x-ms-authorization-auxiliary'] = aux_tokens
-        return session
 
     def get_token(self, *scopes, **kwargs):
+        """Get an access token from the main credential."""
         logger.debug("CredentialAdaptor.get_token: scopes=%r, kwargs=%r", scopes, kwargs)
 
         # Discard unsupported kwargs: tenant_id, enable_cae
         filtered_kwargs = {}
         if 'data' in kwargs:
             filtered_kwargs['data'] = kwargs['data']
 
-        token, _ = self._get_token(scopes, **filtered_kwargs)
-        return token
+        return self._credential.get_token(scopes, **filtered_kwargs)
 
     def get_auxiliary_tokens(self, *scopes, **kwargs):
+        """Get access tokens from auxiliary credentials."""
         # To test cross-tenant authentication, see https://github.com/Azure/azure-cli/issues/16691
         if self._auxiliary_credentials:
             return [cred.get_token(*scopes, **kwargs) for cred in self._auxiliary_credentials]

src/azure-cli-core/azure/cli/core/auth/tests/test_credential_adaptor.py

@@ -213,7 +213,6 @@ def _get_mgmt_service_client(cli_ctx,
                              subscription_id=None,
                              api_version=None,
                              base_url_bound=True,
-                             resource=None,
                              sdk_profile=None,
                              aux_subscriptions=None,
                              aux_tenants=None,
@@ -222,10 +221,6 @@ def _get_mgmt_service_client(cli_ctx,
     from azure.cli.core._profile import Profile
     logger.debug('Getting management service client client_type=%s', client_type.__name__)
 
-    # Track 1 SDK doesn't maintain the `resource`. The `resource` of the token is the one passed to
-    # get_login_credentials.
-    resource = resource or cli_ctx.cloud.endpoints.active_directory_resource_id
-
     if credential:
         # Use a custom credential
         if not subscription_id:
@@ -234,7 +229,7 @@ def _get_mgmt_service_client(cli_ctx,
         # Get a credential for the current `az login` context
         profile = Profile(cli_ctx=cli_ctx)
         credential, subscription_id, _ = profile.get_login_credentials(
-            subscription_id=subscription_id, resource=resource,
+            subscription_id=subscription_id,
             aux_subscriptions=aux_subscriptions, aux_tenants=aux_tenants)
 
     client_kwargs = {}