Source code for azure.identity._credentials.shared_cache

# ------------------------------------
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
# ------------------------------------
from .. import CredentialUnavailableError
from .._constants import AZURE_CLI_CLIENT_ID
from .._internal import AadClient, wrap_exceptions
from .._internal.shared_token_cache import NO_TOKEN, SharedTokenCacheBase

try:
    from typing import TYPE_CHECKING
except ImportError:
    TYPE_CHECKING = False

if TYPE_CHECKING:
    # pylint:disable=unused-import,ungrouped-imports
    from typing import Any, Mapping
    from azure.core.credentials import AccessToken
    from .._internal import AadClientBase


[docs]class SharedTokenCacheCredential(SharedTokenCacheBase): """Authenticates using tokens in the local cache shared between Microsoft applications. :param str username: Username (typically an email address) of the user to authenticate as. This is used when the local cache contains tokens for multiple identities. :keyword str authority: Authority of an Azure Active Directory endpoint, for example 'login.microsoftonline.com', the authority for Azure Public Cloud (which is the default). :class:`~azure.identity.KnownAuthorities` defines authorities for other clouds. :keyword str tenant_id: an Azure Active Directory tenant ID. Used to select an account when the cache contains tokens for multiple identities. """
[docs] @wrap_exceptions def get_token(self, *scopes, **kwargs): # pylint:disable=unused-argument # type (*str, **Any) -> AccessToken """Get an access token for `scopes` from the shared cache. If no access token is cached, attempt to acquire one using a cached refresh token. .. note:: This method is called by Azure SDK clients. It isn't intended for use in application code. :param str scopes: desired scopes for the token :rtype: :class:`azure.core.credentials.AccessToken` :raises ~azure.identity.CredentialUnavailableError: the cache is unavailable or contains insufficient user information :raises ~azure.core.exceptions.ClientAuthenticationError: authentication failed. The error's ``message`` attribute gives a reason. Any error response from Azure Active Directory is available as the error's ``response`` attribute. """ if not self._client: raise CredentialUnavailableError(message="Shared token cache unavailable") account = self._get_account(self._username, self._tenant_id) # try each refresh token, returning the first access token acquired for refresh_token in self._get_refresh_tokens(account): token = self._client.obtain_token_by_refresh_token(refresh_token, scopes) return token raise CredentialUnavailableError(message=NO_TOKEN.format(account.get("username")))
def _get_auth_client(self, **kwargs): # type: (**Any) -> AadClientBase return AadClient(tenant_id="common", client_id=AZURE_CLI_CLIENT_ID, **kwargs)