auth

Author: jiasli

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

@@ -11,6 +11,7 @@
 from azure.cli.core._session import ACCOUNT
 from azure.cli.core.azclierror import AuthenticationError
 from azure.cli.core.cloud import get_active_cloud, set_cloud_subscription
+from azure.cli.core.auth.credential_adaptor import CredentialAdaptor
 from azure.cli.core.util import in_cloud_console, can_launch_browser, is_github_codespaces
 from knack.log import get_logger
 from knack.util import CLIError
@@ -477,7 +478,7 @@ def get_raw_token(self, resource=None, scopes=None, subscription=None, tenant=No
         else:
             cred = self._create_credential(account, tenant_id=tenant)
 
-        sdk_token = cred.get_token(*scopes)
+        sdk_token = CredentialAdaptor(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
@@ -856,7 +857,6 @@ def find_using_common_tenant(self, username, credential=None):
             specific_tenant_credential = identity.get_user_credential(username)
 
             try:
-
                 subscriptions = self.find_using_specific_tenant(tenant_id, specific_tenant_credential,
                                                                 tenant_id_description=t)
             except AuthenticationError as ex:
@@ -927,9 +927,11 @@ def _create_subscription_client(self, credential):
             raise CLIInternalError("Unable to get '{}' in profile '{}'"
                                    .format(ResourceType.MGMT_RESOURCE_SUBSCRIPTIONS, self.cli_ctx.cloud.profile))
         api_version = get_api_version(self.cli_ctx, ResourceType.MGMT_RESOURCE_SUBSCRIPTIONS)
-        client_kwargs = _prepare_mgmt_client_kwargs_track2(self.cli_ctx, credential)
 
-        client = client_type(credential, api_version=api_version,
+        sdk_credential = CredentialAdaptor(credential)
+        client_kwargs = _prepare_mgmt_client_kwargs_track2(self.cli_ctx, sdk_credential)
+
+        client = client_type(sdk_credential, api_version=api_version,
                              base_url=self.cli_ctx.cloud.endpoints.resource_manager,
                              **client_kwargs)
         return client

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

@@ -4,19 +4,19 @@
 # --------------------------------------------------------------------------------------------
 
 from knack.log import get_logger
+from .util import build_sdk_access_token
 
 logger = get_logger(__name__)
 
 
 class CredentialAdaptor:
     def __init__(self, credential, auxiliary_credentials=None):
-        """Cross-tenant credential adaptor. It takes a main credential and auxiliary credentials.
-
+        """Credential adaptor between MSAL credential and SDK credential.
         It implements Track 2 SDK's azure.core.credentials.TokenCredential by exposing get_token.
 
-        :param credential: Main credential from .msal_authentication
-        :param auxiliary_credentials: Credentials from .msal_authentication for cross tenant authentication.
-            Details about cross tenant authentication:
+        :param credential: MSAL credential from ._msal_credentials
+        :param auxiliary_credentials: MSAL credentials for cross-tenant authentication.
+            Details about cross-tenant authentication:
             https://learn.microsoft.com/en-us/azure/azure-resource-manager/management/authenticate-multi-tenant
         """
 
@@ -32,11 +32,12 @@ def get_token(self, *scopes, **kwargs):
         if 'data' in kwargs:
             filtered_kwargs['data'] = kwargs['data']
 
-        return self._credential.get_token(*scopes, **filtered_kwargs)
+        return build_sdk_access_token(self._credential.acquire_token(list(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]
+            return [build_sdk_access_token(cred.acquire_token(list(scopes), **kwargs))
+                    for cred in self._auxiliary_credentials]
         return None

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

@@ -192,9 +192,8 @@ def login_with_service_principal(self, client_id, credential, scopes):
         """
         sp_auth = ServicePrincipalAuth.build_from_credential(self.tenant_id, client_id, credential)
         client_credential = sp_auth.get_msal_client_credential()
-        cca = ConfidentialClientApplication(client_id, client_credential=client_credential, **self._msal_app_kwargs)
-        result = cca.acquire_token_for_client(scopes)
-        check_result(result)
+        cred = ServicePrincipalCredential(client_id, client_credential, **self._msal_app_kwargs)
+        cred.acquire_token(scopes)
 
         # Only persist the service principal after a successful login
         entry = sp_auth.get_entry_to_persist()

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

@@ -4,16 +4,7 @@
 # --------------------------------------------------------------------------------------------
 
 """
-Credentials defined in this module are alternative implementations of credentials provided by Azure Identity.
-
-These credentials implement azure.core.credentials.TokenCredential by exposing `get_token` method for Track 2
-SDK invocation.
-
-If you want to implement your own credential, the credential must also expose `get_token` method.
-
-`get_token` method takes `scopes` as positional arguments and other optional `kwargs`, such as `claims`, `data`.
-The return value should be a named tuple containing two elements: token (str), expires_on (int). You may simply use
-azure.cli.core.auth.util.AccessToken to build the return value. See below credentials as examples.
+Credentials to acquire tokens from MSAL.
 """
 
 from knack.log import get_logger
@@ -22,15 +13,15 @@
                   ManagedIdentityClient, SystemAssignedManagedIdentity)
 
 from .constants import AZURE_CLI_CLIENT_ID
-from .util import check_result, build_sdk_access_token
+from .util import check_result
 
 logger = get_logger(__name__)
 
 
 class UserCredential:  # pylint: disable=too-few-public-methods
 
     def __init__(self, client_id, username, **kwargs):
-        """User credential implementing get_token interface.
+        """User credential wrapping msal.application.PublicClientApplication
 
         :param client_id: Client ID of the CLI.
         :param username: The username for user credential.
@@ -52,14 +43,15 @@ def __init__(self, client_id, username, **kwargs):
 
         self._account = accounts[0]
 
-    def get_token(self, *scopes, claims=None, **kwargs):
-        # scopes = ['https://pas.windows.net/CheckMyAccess/Linux/.default']
-        logger.debug("UserCredential.get_token: scopes=%r, claims=%r, kwargs=%r", scopes, claims, kwargs)
+    def acquire_token(self, scopes, claims=None, **kwargs):
+        # scopes must be a list.
+        # For acquiring SSH certificate, scopes is ['https://pas.windows.net/CheckMyAccess/Linux/.default']
+        logger.debug("UserCredential.acquire_token: scopes=%r, claims=%r, kwargs=%r", scopes, claims, kwargs)
 
         if claims:
             logger.warning('Acquiring new access token silently for tenant %s with claims challenge: %s',
                            self._msal_app.authority.tenant, claims)
-        result = self._msal_app.acquire_token_silent_with_error(list(scopes), self._account, claims_challenge=claims,
+        result = self._msal_app.acquire_token_silent_with_error(scopes, self._account, claims_challenge=claims,
                                                                 **kwargs)
 
         from azure.cli.core.azclierror import AuthenticationError
@@ -82,7 +74,7 @@ def get_token(self, *scopes, claims=None, **kwargs):
                 success_template, error_template = read_response_templates()
 
                 result = self._msal_app.acquire_token_interactive(
-                    list(scopes), login_hint=self._account['username'],
+                    scopes, login_hint=self._account['username'],
                     port=8400 if self._msal_app.authority.is_adfs else None,
                     success_template=success_template, error_template=error_template, **kwargs)
                 check_result(result)
@@ -91,25 +83,25 @@ def get_token(self, *scopes, claims=None, **kwargs):
             # launch browser, but show the error message and `az login` command instead.
             else:
                 raise
-        return build_sdk_access_token(result)
+        return result
 
 
 class ServicePrincipalCredential:  # pylint: disable=too-few-public-methods
 
     def __init__(self, client_id, client_credential, **kwargs):
-        """Service principal credential implementing get_token interface.
+        """Service principal credential wrapping msal.application.ConfidentialClientApplication.
 
         :param client_id: The service principal's client ID.
         :param client_credential: client_credential that will be passed to MSAL.
         """
         self._msal_app = ConfidentialClientApplication(client_id, client_credential, **kwargs)
 
-    def get_token(self, *scopes, **kwargs):
-        logger.debug("ServicePrincipalCredential.get_token: scopes=%r, kwargs=%r", scopes, kwargs)
-
-        result = self._msal_app.acquire_token_for_client(list(scopes), **kwargs)
+    def acquire_token(self, scopes, **kwargs):
+        # scopes must be a list
+        logger.debug("ServicePrincipalCredential.acquire_token: scopes=%r, kwargs=%r", scopes, kwargs)
+        result = self._msal_app.acquire_token_for_client(scopes, **kwargs)
         check_result(result)
-        return build_sdk_access_token(result)
+        return result
 
 
 class CloudShellCredential:  # pylint: disable=too-few-public-methods
@@ -126,12 +118,12 @@ def __init__(self):
             #   token_cache=...
         )
 
-    def get_token(self, *scopes, **kwargs):
-        logger.debug("CloudShellCredential.get_token: scopes=%r, kwargs=%r", scopes, kwargs)
+    def acquire_token(self, scopes, **kwargs):
+        logger.debug("CloudShellCredential.acquire_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)
+        result = self._msal_app.acquire_token_interactive(scopes, prompt="none", **kwargs)
         check_result(result, scopes=scopes)
-        return build_sdk_access_token(result)
+        return result
 
 
 class ManagedIdentityCredential:  # pylint: disable=too-few-public-methods
@@ -143,10 +135,10 @@ def __init__(self):
         import requests
         self._msal_client = ManagedIdentityClient(SystemAssignedManagedIdentity(), http_client=requests.Session())
 
-    def get_token(self, *scopes, **kwargs):
-        logger.debug("ManagedIdentityCredential.get_token: scopes=%r, kwargs=%r", scopes, kwargs)
+    def acquire_token(self, scopes, **kwargs):
+        logger.debug("ManagedIdentityCredential.acquire_token: scopes=%r, kwargs=%r", scopes, kwargs)
 
         from .util import scopes_to_resource
         result = self._msal_client.acquire_token_for_client(resource=scopes_to_resource(scopes))
         check_result(result)
-        return build_sdk_access_token(result)
+        return result