diff --git a/google/auth/_default.py b/google/auth/_default.py index 63009dfb8..27985727c 100644 --- a/google/auth/_default.py +++ b/google/auth/_default.py @@ -277,6 +277,30 @@ def _get_explicit_environ_credentials(quota_project_id=None): return None, None +def _get_temporary_access_token_environ(): + """Gets credentials from the GOOGLE_TEMPORARY_ACCESS_TOKEN environment + variable.""" + + from google.oauth2 import credentials + + token = os.environ.get(environment_vars.TEMPORARY_ACCESS_TOKEN) + + _LOGGER.debug("Checking %s for temporary access token as part of auth process...") + + if token is not None: + try: + credentials = credentials.Credentials.from_temporary_access_token(token) + except ValueError as caught_exc: + msg = "Failed to load valid temporary access token" + new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) + raise new_exc from caught_exc + + return credentials, None + + else: + return None, None + + def _get_gae_credentials(): """Gets Google App Engine App Identity credentials and project ID.""" # If not GAE gen1, prefer the metadata service even if the GAE APIs are @@ -647,6 +671,7 @@ def default(scopes=None, request=None, quota_project_id=None, default_scopes=Non # with_scopes_if_required() below will ensure scopes/default scopes are # safely set on the returned credentials since requires_scopes will # guard against setting scopes on user credentials. + lambda: _get_temporary_access_token_environ(), lambda: _get_explicit_environ_credentials(quota_project_id=quota_project_id), lambda: _get_gcloud_sdk_credentials(quota_project_id=quota_project_id), _get_gae_credentials, diff --git a/google/auth/_default_async.py b/google/auth/_default_async.py index 2e53e2088..15848baa5 100644 --- a/google/auth/_default_async.py +++ b/google/auth/_default_async.py @@ -126,6 +126,28 @@ def _get_gcloud_sdk_credentials(quota_project_id=None): return credentials, project_id +def _get_temporary_access_token_environ(): + """Gets credentials from the GOOGLE_TEMPORARY_ACCESS_TOKEN environment + variable.""" + + from google.oauth2 import credentials + + token = os.environ.get(environment_vars.TEMPORARY_ACCESS_TOKEN) + + if token is not None: + try: + credentials = credentials.Credentials.from_temporary_access_token(token) + except ValueError as caught_exc: + msg = "Failed to load valid temporary access token" + new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) + raise new_exc from caught_exc + + return credentials, None + + else: + return None, None + + def _get_explicit_environ_credentials(quota_project_id=None): """Gets credentials from the GOOGLE_APPLICATION_CREDENTIALS environment variable.""" @@ -255,6 +277,7 @@ def default_async(scopes=None, request=None, quota_project_id=None): ) checkers = ( + lambda: _get_temporary_access_token_environ(), lambda: _get_explicit_environ_credentials(quota_project_id=quota_project_id), lambda: _get_gcloud_sdk_credentials(quota_project_id=quota_project_id), _get_gae_credentials, diff --git a/google/auth/environment_vars.py b/google/auth/environment_vars.py index 81f31571e..e5e540f6b 100644 --- a/google/auth/environment_vars.py +++ b/google/auth/environment_vars.py @@ -37,6 +37,9 @@ """Environment variable defining the location of Google application default credentials.""" +TEMPORARY_ACCESS_TOKEN = "GOOGLE_TEMPORARY_ACCESS_TOKEN" +"""Environment variable defining access token value. This could be generated using print-access-token cli command""" + # The environment variable name which can replace ~/.config if set. CLOUD_SDK_CONFIG_DIR = "CLOUDSDK_CONFIG" """Environment variable defines the location of Google Cloud SDK's config diff --git a/google/oauth2/credentials.py b/google/oauth2/credentials.py index c239beed1..26ee82531 100644 --- a/google/oauth2/credentials.py +++ b/google/oauth2/credentials.py @@ -31,11 +31,12 @@ .. _rfc6749 section 4.1: https://tools.ietf.org/html/rfc6749#section-4.1 """ -from datetime import datetime +from datetime import datetime, timedelta import io import json import logging import warnings +import requests from google.auth import _cloud_sdk from google.auth import _helpers @@ -461,6 +462,52 @@ def refresh(self, request): ) ) + @classmethod + def from_temporary_access_token(cls, token, scopes=None): + """Creates a Credentials instance from temporary access token info. + + Args: + token (Mapping[str, str]): The authorized token in Google + format. + scopes (Sequence[str]): Optional list of scopes to include in the + credentials. + + Returns: + google.oauth2.credentials.Credentials: The constructed + credentials. + + Raises: + ValueError: If the info is not in the expected format. + """ + + token_validation_response = requests.get( + "https://www.googleapis.com/oauth2/v1/tokeninfo", + params={"access_token": token}, + ) + info = token_validation_response.json() + if info.get("error"): + raise ValueError( + "Authorized access token was not in the expected format, error {}".format( + info.get("error") + ) + ) + + # access token expiry (datetime obj); auto-expire if not saved + expiry = datetime.now() + timedelta(seconds=info["expires_in"]) + + # process scopes, which needs to be a seq + if scopes is None and "scope" in info: + scopes = info.get("scope") + if isinstance(scopes, str): + scopes = scopes.split(" ") + + return cls( + token=token, + token_uri=_GOOGLE_OAUTH2_TOKEN_ENDPOINT, # always overrides + scopes=scopes, + expiry=expiry, + ) + @classmethod def from_authorized_user_info(cls, info, scopes=None): """Creates a Credentials instance from parsed authorized user info. diff --git a/tests/oauth2/test_credentials.py b/tests/oauth2/test_credentials.py index 216641946..b7ea29166 100644 --- a/tests/oauth2/test_credentials.py +++ b/tests/oauth2/test_credentials.py @@ -26,6 +26,7 @@ from google.auth import transport from google.auth.credentials import TokenState from google.oauth2 import credentials +from google.auth import environment_vars DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data") @@ -811,6 +812,13 @@ def test_with_token_uri(self): assert creds_with_new_token_uri._token_uri == new_token_uri + def test_with_temporary_access_token(self): + token = os.environ[environment_vars.TEMPORARY_ACCESS_TOKEN] + + creds = credentials.Credentials.from_temporary_access_token(token) + + assert creds.expiry is not None + def test_from_authorized_user_info(self): info = AUTH_USER_INFO.copy()