Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2025-04-14 17:36
📄 __init__.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Google Auth Library for Python.""" import logging import sys import warnings from google.auth import version as google_auth_version from google.auth._default import ( default, load_credentials_from_dict, load_credentials_from_file, ) __version__ = google_auth_version.__version__ __all__ = ["default", "load_credentials_from_file", "load_credentials_from_dict"] class Python37DeprecationWarning(DeprecationWarning): # pragma: NO COVER """ Deprecation warning raised when Python 3.7 runtime is detected. Python 3.7 support will be dropped after January 1, 2024. """ pass # Checks if the current runtime is Python 3.7. if sys.version_info.major == 3 and sys.version_info.minor == 7: # pragma: NO COVER message = ( "After January 1, 2024, new releases of this library will drop support " "for Python 3.7." ) warnings.warn(message, Python37DeprecationWarning) # Set default logging handler to avoid "No handler found" warnings. logging.getLogger(__name__).addHandler(logging.NullHandler())	1.6 KB	2025-04-14 17:36
📄 _cloud_sdk.py # Copyright 2015 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Helpers for reading the Google Cloud SDK's configuration.""" import os import subprocess from google.auth import _helpers from google.auth import environment_vars from google.auth import exceptions # The ~/.config subdirectory containing gcloud credentials. _CONFIG_DIRECTORY = "gcloud" # Windows systems store config at %APPDATA%\gcloud _WINDOWS_CONFIG_ROOT_ENV_VAR = "APPDATA" # The name of the file in the Cloud SDK config that contains default # credentials. _CREDENTIALS_FILENAME = "application_default_credentials.json" # The name of the Cloud SDK shell script _CLOUD_SDK_POSIX_COMMAND = "gcloud" _CLOUD_SDK_WINDOWS_COMMAND = "gcloud.cmd" # The command to get the Cloud SDK configuration _CLOUD_SDK_CONFIG_GET_PROJECT_COMMAND = ("config", "get", "project") # The command to get google user access token _CLOUD_SDK_USER_ACCESS_TOKEN_COMMAND = ("auth", "print-access-token") # Cloud SDK's application-default client ID CLOUD_SDK_CLIENT_ID = ( "764086051850-6qr4p6gpi6hn506pt8ejuq83di341hur.apps.googleusercontent.com" ) def get_config_path(): """Returns the absolute path the the Cloud SDK's configuration directory. Returns: str: The Cloud SDK config path. """ # If the path is explicitly set, return that. try: return os.environ[environment_vars.CLOUD_SDK_CONFIG_DIR] except KeyError: pass # Non-windows systems store this at ~/.config/gcloud if os.name != "nt": return os.path.join(os.path.expanduser("~"), ".config", _CONFIG_DIRECTORY) # Windows systems store config at %APPDATA%\gcloud else: try: return os.path.join( os.environ[_WINDOWS_CONFIG_ROOT_ENV_VAR], _CONFIG_DIRECTORY ) except KeyError: # This should never happen unless someone is really # messing with things, but we'll cover the case anyway. drive = os.environ.get("SystemDrive", "C:") return os.path.join(drive, "\\", _CONFIG_DIRECTORY) def get_application_default_credentials_path(): """Gets the path to the application default credentials file. The path may or may not exist. Returns: str: The full path to application default credentials. """ config_path = get_config_path() return os.path.join(config_path, _CREDENTIALS_FILENAME) def _run_subprocess_ignore_stderr(command): """ Return subprocess.check_output with the given command and ignores stderr.""" with open(os.devnull, "w") as devnull: output = subprocess.check_output(command, stderr=devnull) return output def get_project_id(): """Gets the project ID from the Cloud SDK. Returns: Optional[str]: The project ID. """ if os.name == "nt": command = _CLOUD_SDK_WINDOWS_COMMAND else: command = _CLOUD_SDK_POSIX_COMMAND try: # Ignore the stderr coming from gcloud, so it won't be mixed into the output. # https://github.com/googleapis/google-auth-library-python/issues/673 project = _run_subprocess_ignore_stderr( (command,) + _CLOUD_SDK_CONFIG_GET_PROJECT_COMMAND ) # Turn bytes into a string and remove "\n" project = _helpers.from_bytes(project).strip() return project if project else None except (subprocess.CalledProcessError, OSError, IOError): return None def get_auth_access_token(account=None): """Load user access token with the ``gcloud auth print-access-token`` command. Args: account (Optional[str]): Account to get the access token for. If not specified, the current active account will be used. Returns: str: The user access token. Raises: google.auth.exceptions.UserAccessTokenError: if failed to get access token from gcloud. """ if os.name == "nt": command = _CLOUD_SDK_WINDOWS_COMMAND else: command = _CLOUD_SDK_POSIX_COMMAND try: if account: command = ( (command,) + _CLOUD_SDK_USER_ACCESS_TOKEN_COMMAND + ("--account=" + account,) ) else: command = (command,) + _CLOUD_SDK_USER_ACCESS_TOKEN_COMMAND access_token = subprocess.check_output(command, stderr=subprocess.STDOUT) # remove the trailing "\n" return access_token.decode("utf-8").strip() except (subprocess.CalledProcessError, OSError, IOError) as caught_exc: new_exc = exceptions.UserAccessTokenError( "Failed to obtain access token", caught_exc ) raise new_exc from caught_exc	5.09 KB	2025-04-14 17:36
📄 _credentials_async.py # Copyright 2020 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Interfaces for credentials.""" import abc import inspect from google.auth import credentials class Credentials(credentials.Credentials, metaclass=abc.ABCMeta): """Async inherited credentials class from google.auth.credentials. The added functionality is the before_request call which requires async/await syntax. All credentials have a :attr:`token` that is used for authentication and may also optionally set an :attr:`expiry` to indicate when the token will no longer be valid. Most credentials will be :attr:`invalid` until :meth:`refresh` is called. Credentials can do this automatically before the first HTTP request in :meth:`before_request`. Although the token and expiration will change as the credentials are :meth:`refreshed <refresh>` and used, credentials should be considered immutable. Various credentials will accept configuration such as private keys, scopes, and other options. These options are not changeable after construction. Some classes will provide mechanisms to copy the credentials with modifications such as :meth:`ScopedCredentials.with_scopes`. """ async def before_request(self, request, method, url, headers): """Performs credential-specific before request logic. Refreshes the credentials if necessary, then calls :meth:`apply` to apply the token to the authentication header. Args: request (google.auth.transport.Request): The object used to make HTTP requests. method (str): The request's HTTP method or the RPC method being invoked. url (str): The request's URI or the RPC service's URI. headers (Mapping): The request's headers. """ # pylint: disable=unused-argument # (Subclasses may use these arguments to ascertain information about # the http request.) if not self.valid: if inspect.iscoroutinefunction(self.refresh): await self.refresh(request) else: self.refresh(request) self.apply(headers) class CredentialsWithQuotaProject(credentials.CredentialsWithQuotaProject): """Abstract base for credentials supporting ``with_quota_project`` factory""" class AnonymousCredentials(credentials.AnonymousCredentials, Credentials): """Credentials that do not provide any authentication information. These are useful in the case of services that support anonymous access or local service emulators that do not use credentials. This class inherits from the sync anonymous credentials file, but is kept if async credentials is initialized and we would like anonymous credentials. """ class ReadOnlyScoped(credentials.ReadOnlyScoped, metaclass=abc.ABCMeta): """Interface for credentials whose scopes can be queried. OAuth 2.0-based credentials allow limiting access using scopes as described in `RFC6749 Section 3.3`_. If a credential class implements this interface then the credentials either use scopes in their implementation. Some credentials require scopes in order to obtain a token. You can check if scoping is necessary with :attr:`requires_scopes`:: if credentials.requires_scopes: # Scoping is required. credentials = _credentials_async.with_scopes(scopes=['one', 'two']) Credentials that require scopes must either be constructed with scopes:: credentials = SomeScopedCredentials(scopes=['one', 'two']) Or must copy an existing instance using :meth:`with_scopes`:: scoped_credentials = _credentials_async.with_scopes(scopes=['one', 'two']) Some credentials have scopes but do not allow or require scopes to be set, these credentials can be used as-is. .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 """ class Scoped(credentials.Scoped): """Interface for credentials whose scopes can be replaced while copying. OAuth 2.0-based credentials allow limiting access using scopes as described in `RFC6749 Section 3.3`_. If a credential class implements this interface then the credentials either use scopes in their implementation. Some credentials require scopes in order to obtain a token. You can check if scoping is necessary with :attr:`requires_scopes`:: if credentials.requires_scopes: # Scoping is required. credentials = _credentials_async.create_scoped(['one', 'two']) Credentials that require scopes must either be constructed with scopes:: credentials = SomeScopedCredentials(scopes=['one', 'two']) Or must copy an existing instance using :meth:`with_scopes`:: scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) Some credentials have scopes but do not allow or require scopes to be set, these credentials can be used as-is. .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 """ def with_scopes_if_required(credentials, scopes): """Creates a copy of the credentials with scopes if scoping is required. This helper function is useful when you do not know (or care to know) the specific type of credentials you are using (such as when you use :func:`google.auth.default`). This function will call :meth:`Scoped.with_scopes` if the credentials are scoped credentials and if the credentials require scoping. Otherwise, it will return the credentials as-is. Args: credentials (google.auth.credentials.Credentials): The credentials to scope if necessary. scopes (Sequence[str]): The list of scopes to use. Returns: google.auth._credentials_async.Credentials: Either a new set of scoped credentials, or the passed in credentials instance if no scoping was required. """ if isinstance(credentials, Scoped) and credentials.requires_scopes: return credentials.with_scopes(scopes) else: return credentials class Signing(credentials.Signing, metaclass=abc.ABCMeta): """Interface for credentials that can cryptographically sign messages."""	6.64 KB	2025-04-14 17:36
📄 _credentials_base.py # Copyright 2024 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Interface for base credentials.""" import abc from google.auth import _helpers class _BaseCredentials(metaclass=abc.ABCMeta): """Base class for all credentials. All credentials have a :attr:`token` that is used for authentication and may also optionally set an :attr:`expiry` to indicate when the token will no longer be valid. Most credentials will be :attr:`invalid` until :meth:`refresh` is called. Credentials can do this automatically before the first HTTP request in :meth:`before_request`. Although the token and expiration will change as the credentials are :meth:`refreshed <refresh>` and used, credentials should be considered immutable. Various credentials will accept configuration such as private keys, scopes, and other options. These options are not changeable after construction. Some classes will provide mechanisms to copy the credentials with modifications such as :meth:`ScopedCredentials.with_scopes`. Attributes: token (Optional[str]): The bearer token that can be used in HTTP headers to make authenticated requests. """ def __init__(self): self.token = None @abc.abstractmethod def refresh(self, request): """Refreshes the access token. Args: request (google.auth.transport.Request): The object used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If the credentials could not be refreshed. """ # pylint: disable=missing-raises-doc # (pylint doesn't recognize that this is abstract) raise NotImplementedError("Refresh must be implemented") def _apply(self, headers, token=None): """Apply the token to the authentication header. Args: headers (Mapping): The HTTP request headers. token (Optional[str]): If specified, overrides the current access token. """ headers["authorization"] = "Bearer {}".format( _helpers.from_bytes(token or self.token) )	2.63 KB	2025-04-14 17:36
📄 _default.py # Copyright 2015 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Application default credentials. Implements application default credentials and project ID detection. """ import io import json import logging import os import warnings from google.auth import environment_vars from google.auth import exceptions import google.auth.transport._http_client _LOGGER = logging.getLogger(__name__) # Valid types accepted for file-based credentials. _AUTHORIZED_USER_TYPE = "authorized_user" _SERVICE_ACCOUNT_TYPE = "service_account" _EXTERNAL_ACCOUNT_TYPE = "external_account" _EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE = "external_account_authorized_user" _IMPERSONATED_SERVICE_ACCOUNT_TYPE = "impersonated_service_account" _GDCH_SERVICE_ACCOUNT_TYPE = "gdch_service_account" _VALID_TYPES = ( _AUTHORIZED_USER_TYPE, _SERVICE_ACCOUNT_TYPE, _EXTERNAL_ACCOUNT_TYPE, _EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE, _IMPERSONATED_SERVICE_ACCOUNT_TYPE, _GDCH_SERVICE_ACCOUNT_TYPE, ) # Help message when no credentials can be found. _CLOUD_SDK_MISSING_CREDENTIALS = """\ Your default credentials were not found. To set up Application Default Credentials, \ see https://cloud.google.com/docs/authentication/external/set-up-adc for more information.\ """ # Warning when using Cloud SDK user credentials _CLOUD_SDK_CREDENTIALS_WARNING = """\ Your application has authenticated using end user credentials from Google \ Cloud SDK without a quota project. You might receive a "quota exceeded" \ or "API not enabled" error. See the following page for troubleshooting: \ https://cloud.google.com/docs/authentication/adc-troubleshooting/user-creds. \ """ # The subject token type used for AWS external_account credentials. _AWS_SUBJECT_TOKEN_TYPE = "urn:ietf:params:aws:token-type:aws4_request" def _warn_about_problematic_credentials(credentials): """Determines if the credentials are problematic. Credentials from the Cloud SDK that are associated with Cloud SDK's project are problematic because they may not have APIs enabled and have limited quota. If this is the case, warn about it. """ from google.auth import _cloud_sdk if credentials.client_id == _cloud_sdk.CLOUD_SDK_CLIENT_ID: warnings.warn(_CLOUD_SDK_CREDENTIALS_WARNING) def load_credentials_from_file( filename, scopes=None, default_scopes=None, quota_project_id=None, request=None ): """Loads Google credentials from a file. The credentials file must be a service account key, stored authorized user credentials, external account credentials, or impersonated service account credentials. .. warning:: Important: If you accept a credential configuration (credential JSON/File/Stream) from an external source for authentication to Google Cloud Platform, you must validate it before providing it to any Google API or client library. Providing an unvalidated credential configuration to Google APIs or libraries can compromise the security of your systems and data. For more information, refer to `Validate credential configurations from external sources`_. .. _Validate credential configurations from external sources: https://cloud.google.com/docs/authentication/external/externally-sourced-credentials Args: filename (str): The full path to the credentials file. scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If specified, the credentials will automatically be scoped if necessary default_scopes (Optional[Sequence[str]]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. quota_project_id (Optional[str]): The project ID used for quota and billing. request (Optional[google.auth.transport.Request]): An object used to make HTTP requests. This is used to determine the associated project ID for a workload identity pool resource (external account credentials). If not specified, then it will use a google.auth.transport.requests.Request client to make requests. Returns: Tuple[google.auth.credentials.Credentials, Optional[str]]: Loaded credentials and the project ID. Authorized user credentials do not have the project ID information. External account credentials project IDs may not always be determined. Raises: google.auth.exceptions.DefaultCredentialsError: if the file is in the wrong format or is missing. """ if not os.path.exists(filename): raise exceptions.DefaultCredentialsError( "File {} was not found.".format(filename) ) with io.open(filename, "r") as file_obj: try: info = json.load(file_obj) except ValueError as caught_exc: new_exc = exceptions.DefaultCredentialsError( "File {} is not a valid json file.".format(filename), caught_exc ) raise new_exc from caught_exc return _load_credentials_from_info( filename, info, scopes, default_scopes, quota_project_id, request ) def load_credentials_from_dict( info, scopes=None, default_scopes=None, quota_project_id=None, request=None ): """Loads Google credentials from a dict. The credentials file must be a service account key, stored authorized user credentials, external account credentials, or impersonated service account credentials. .. warning:: Important: If you accept a credential configuration (credential JSON/File/Stream) from an external source for authentication to Google Cloud Platform, you must validate it before providing it to any Google API or client library. Providing an unvalidated credential configuration to Google APIs or libraries can compromise the security of your systems and data. For more information, refer to `Validate credential configurations from external sources`_. .. _Validate credential configurations from external sources: https://cloud.google.com/docs/authentication/external/externally-sourced-credentials Args: info (Dict[str, Any]): A dict object containing the credentials scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If specified, the credentials will automatically be scoped if necessary default_scopes (Optional[Sequence[str]]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. quota_project_id (Optional[str]): The project ID used for quota and billing. request (Optional[google.auth.transport.Request]): An object used to make HTTP requests. This is used to determine the associated project ID for a workload identity pool resource (external account credentials). If not specified, then it will use a google.auth.transport.requests.Request client to make requests. Returns: Tuple[google.auth.credentials.Credentials, Optional[str]]: Loaded credentials and the project ID. Authorized user credentials do not have the project ID information. External account credentials project IDs may not always be determined. Raises: google.auth.exceptions.DefaultCredentialsError: if the file is in the wrong format or is missing. """ if not isinstance(info, dict): raise exceptions.DefaultCredentialsError( "info object was of type {} but dict type was expected.".format(type(info)) ) return _load_credentials_from_info( "dict object", info, scopes, default_scopes, quota_project_id, request ) def _load_credentials_from_info( filename, info, scopes, default_scopes, quota_project_id, request ): from google.auth.credentials import CredentialsWithQuotaProject credential_type = info.get("type") if credential_type == _AUTHORIZED_USER_TYPE: credentials, project_id = _get_authorized_user_credentials( filename, info, scopes ) elif credential_type == _SERVICE_ACCOUNT_TYPE: credentials, project_id = _get_service_account_credentials( filename, info, scopes, default_scopes ) elif credential_type == _EXTERNAL_ACCOUNT_TYPE: credentials, project_id = _get_external_account_credentials( info, filename, scopes=scopes, default_scopes=default_scopes, request=request, ) elif credential_type == _EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE: credentials, project_id = _get_external_account_authorized_user_credentials( filename, info, request ) elif credential_type == _IMPERSONATED_SERVICE_ACCOUNT_TYPE: credentials, project_id = _get_impersonated_service_account_credentials( filename, info, scopes ) elif credential_type == _GDCH_SERVICE_ACCOUNT_TYPE: credentials, project_id = _get_gdch_service_account_credentials(filename, info) else: raise exceptions.DefaultCredentialsError( "The file {file} does not have a valid type. " "Type is {type}, expected one of {valid_types}.".format( file=filename, type=credential_type, valid_types=_VALID_TYPES ) ) if isinstance(credentials, CredentialsWithQuotaProject): credentials = _apply_quota_project_id(credentials, quota_project_id) return credentials, project_id def _get_gcloud_sdk_credentials(quota_project_id=None): """Gets the credentials and project ID from the Cloud SDK.""" from google.auth import _cloud_sdk _LOGGER.debug("Checking Cloud SDK credentials as part of auth process...") # Check if application default credentials exist. credentials_filename = _cloud_sdk.get_application_default_credentials_path() if not os.path.isfile(credentials_filename): _LOGGER.debug("Cloud SDK credentials not found on disk; not using them") return None, None credentials, project_id = load_credentials_from_file( credentials_filename, quota_project_id=quota_project_id ) credentials._cred_file_path = credentials_filename if not project_id: project_id = _cloud_sdk.get_project_id() return credentials, project_id def _get_explicit_environ_credentials(quota_project_id=None): """Gets credentials from the GOOGLE_APPLICATION_CREDENTIALS environment variable.""" from google.auth import _cloud_sdk cloud_sdk_adc_path = _cloud_sdk.get_application_default_credentials_path() explicit_file = os.environ.get(environment_vars.CREDENTIALS) _LOGGER.debug( "Checking %s for explicit credentials as part of auth process...", explicit_file ) if explicit_file is not None and explicit_file == cloud_sdk_adc_path: # Cloud sdk flow calls gcloud to fetch project id, so if the explicit # file path is cloud sdk credentials path, then we should fall back # to cloud sdk flow, otherwise project id cannot be obtained. _LOGGER.debug( "Explicit credentials path %s is the same as Cloud SDK credentials path, fall back to Cloud SDK credentials flow...", explicit_file, ) return _get_gcloud_sdk_credentials(quota_project_id=quota_project_id) if explicit_file is not None: credentials, project_id = load_credentials_from_file( os.environ[environment_vars.CREDENTIALS], quota_project_id=quota_project_id ) credentials._cred_file_path = f"{explicit_file} file via the GOOGLE_APPLICATION_CREDENTIALS environment variable" return credentials, project_id 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 # available as per https://google.aip.dev/auth/4115. if os.environ.get(environment_vars.LEGACY_APPENGINE_RUNTIME) != "python27": return None, None # While this library is normally bundled with app_engine, there are # some cases where it's not available, so we tolerate ImportError. try: _LOGGER.debug("Checking for App Engine runtime as part of auth process...") import google.auth.app_engine as app_engine except ImportError: _LOGGER.warning("Import of App Engine auth library failed.") return None, None try: credentials = app_engine.Credentials() project_id = app_engine.get_project_id() return credentials, project_id except EnvironmentError: _LOGGER.debug( "No App Engine library was found so cannot authentication via App Engine Identity Credentials." ) return None, None def _get_gce_credentials(request=None, quota_project_id=None): """Gets credentials and project ID from the GCE Metadata Service.""" # Ping requires a transport, but we want application default credentials # to require no arguments. So, we'll use the _http_client transport which # uses http.client. This is only acceptable because the metadata server # doesn't do SSL and never requires proxies. # While this library is normally bundled with compute_engine, there are # some cases where it's not available, so we tolerate ImportError. try: from google.auth import compute_engine from google.auth.compute_engine import _metadata except ImportError: _LOGGER.warning("Import of Compute Engine auth library failed.") return None, None if request is None: request = google.auth.transport._http_client.Request() if _metadata.is_on_gce(request=request): # Get the project ID. try: project_id = _metadata.get_project_id(request=request) except exceptions.TransportError: project_id = None cred = compute_engine.Credentials() cred = _apply_quota_project_id(cred, quota_project_id) return cred, project_id else: _LOGGER.warning( "Authentication failed using Compute Engine authentication due to unavailable metadata server." ) return None, None def _get_external_account_credentials( info, filename, scopes=None, default_scopes=None, request=None ): """Loads external account Credentials from the parsed external account info. The credentials information must correspond to a supported external account credentials. Args: info (Mapping[str, str]): The external account info in Google format. filename (str): The full path to the credentials file. scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If specified, the credentials will automatically be scoped if necessary. default_scopes (Optional[Sequence[str]]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. request (Optional[google.auth.transport.Request]): An object used to make HTTP requests. This is used to determine the associated project ID for a workload identity pool resource (external account credentials). If not specified, then it will use a google.auth.transport.requests.Request client to make requests. Returns: Tuple[google.auth.credentials.Credentials, Optional[str]]: Loaded credentials and the project ID. External account credentials project IDs may not always be determined. Raises: google.auth.exceptions.DefaultCredentialsError: if the info dictionary is in the wrong format or is missing required information. """ # There are currently 3 types of external_account credentials. if info.get("subject_token_type") == _AWS_SUBJECT_TOKEN_TYPE: # Check if configuration corresponds to an AWS credentials. from google.auth import aws credentials = aws.Credentials.from_info( info, scopes=scopes, default_scopes=default_scopes ) elif ( info.get("credential_source") is not None and info.get("credential_source").get("executable") is not None ): from google.auth import pluggable credentials = pluggable.Credentials.from_info( info, scopes=scopes, default_scopes=default_scopes ) else: try: # Check if configuration corresponds to an Identity Pool credentials. from google.auth import identity_pool credentials = identity_pool.Credentials.from_info( info, scopes=scopes, default_scopes=default_scopes ) except ValueError: # If the configuration is invalid or does not correspond to any # supported external_account credentials, raise an error. raise exceptions.DefaultCredentialsError( "Failed to load external account credentials from {}".format(filename) ) if request is None: import google.auth.transport.requests request = google.auth.transport.requests.Request() return credentials, credentials.get_project_id(request=request) def _get_external_account_authorized_user_credentials( filename, info, scopes=None, default_scopes=None, request=None ): try: from google.auth import external_account_authorized_user credentials = external_account_authorized_user.Credentials.from_info(info) except ValueError: raise exceptions.DefaultCredentialsError( "Failed to load external account authorized user credentials from {}".format( filename ) ) return credentials, None def _get_authorized_user_credentials(filename, info, scopes=None): from google.oauth2 import credentials try: credentials = credentials.Credentials.from_authorized_user_info( info, scopes=scopes ) except ValueError as caught_exc: msg = "Failed to load authorized user credentials from {}".format(filename) new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) raise new_exc from caught_exc return credentials, None def _get_service_account_credentials(filename, info, scopes=None, default_scopes=None): from google.oauth2 import service_account try: credentials = service_account.Credentials.from_service_account_info( info, scopes=scopes, default_scopes=default_scopes ) except ValueError as caught_exc: msg = "Failed to load service account credentials from {}".format(filename) new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) raise new_exc from caught_exc return credentials, info.get("project_id") def _get_impersonated_service_account_credentials(filename, info, scopes): from google.auth import impersonated_credentials try: credentials = impersonated_credentials.Credentials.from_impersonated_service_account_info( info, scopes=scopes ) except ValueError as caught_exc: msg = "Failed to load impersonated service account credentials from {}".format( filename ) new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) raise new_exc from caught_exc return credentials, None def _get_gdch_service_account_credentials(filename, info): from google.oauth2 import gdch_credentials try: credentials = gdch_credentials.ServiceAccountCredentials.from_service_account_info( info ) except ValueError as caught_exc: msg = "Failed to load GDCH service account credentials from {}".format(filename) new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) raise new_exc from caught_exc return credentials, info.get("project") def get_api_key_credentials(key): """Return credentials with the given API key.""" from google.auth import api_key return api_key.Credentials(key) def _apply_quota_project_id(credentials, quota_project_id): if quota_project_id: credentials = credentials.with_quota_project(quota_project_id) else: credentials = credentials.with_quota_project_from_environment() from google.oauth2 import credentials as authorized_user_credentials if isinstance(credentials, authorized_user_credentials.Credentials) and ( not credentials.quota_project_id ): _warn_about_problematic_credentials(credentials) return credentials def default(scopes=None, request=None, quota_project_id=None, default_scopes=None): """Gets the default credentials for the current environment. `Application Default Credentials`_ provides an easy way to obtain credentials to call Google APIs for server-to-server or local applications. This function acquires credentials from the environment in the following order: 1. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set to the path of a valid service account JSON private key file, then it is loaded and returned. The project ID returned is the project ID defined in the service account file if available (some older files do not contain project ID information). If the environment variable is set to the path of a valid external account JSON configuration file (workload identity federation), then the configuration file is used to determine and retrieve the external credentials from the current environment (AWS, Azure, etc). These will then be exchanged for Google access tokens via the Google STS endpoint. The project ID returned in this case is the one corresponding to the underlying workload identity pool resource if determinable. If the environment variable is set to the path of a valid GDCH service account JSON file (`Google Distributed Cloud Hosted`_), then a GDCH credential will be returned. The project ID returned is the project specified in the JSON file. 2. If the `Google Cloud SDK`_ is installed and has application default credentials set they are loaded and returned. To enable application default credentials with the Cloud SDK run:: gcloud auth application-default login If the Cloud SDK has an active project, the project ID is returned. The active project can be set using:: gcloud config set project 3. If the application is running in the `App Engine standard environment`_ (first generation) then the credentials and project ID from the `App Identity Service`_ are used. 4. If the application is running in `Compute Engine`_ or `Cloud Run`_ or the `App Engine flexible environment`_ or the `App Engine standard environment`_ (second generation) then the credentials and project ID are obtained from the `Metadata Service`_. 5. If no credentials are found, :class:`~google.auth.exceptions.DefaultCredentialsError` will be raised. .. _Application Default Credentials: https://developers.google.com\ /identity/protocols/application-default-credentials .. _Google Cloud SDK: https://cloud.google.com/sdk .. _App Engine standard environment: https://cloud.google.com/appengine .. _App Identity Service: https://cloud.google.com/appengine/docs/python\ /appidentity/ .. _Compute Engine: https://cloud.google.com/compute .. _App Engine flexible environment: https://cloud.google.com\ /appengine/flexible .. _Metadata Service: https://cloud.google.com/compute/docs\ /storing-retrieving-metadata .. _Cloud Run: https://cloud.google.com/run .. _Google Distributed Cloud Hosted: https://cloud.google.com/blog/topics\ /hybrid-cloud/announcing-google-distributed-cloud-edge-and-hosted Example:: import google.auth credentials, project_id = google.auth.default() Args: scopes (Sequence[str]): The list of scopes for the credentials. If specified, the credentials will automatically be scoped if necessary. request (Optional[google.auth.transport.Request]): An object used to make HTTP requests. This is used to either detect whether the application is running on Compute Engine or to determine the associated project ID for a workload identity pool resource (external account credentials). If not specified, then it will either use the standard library http client to make requests for Compute Engine credentials or a google.auth.transport.requests.Request client for external account credentials. quota_project_id (Optional[str]): The project ID used for quota and billing. default_scopes (Optional[Sequence[str]]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. Returns: Tuple[~google.auth.credentials.Credentials, Optional[str]]: the current environment's credentials and project ID. Project ID may be None, which indicates that the Project ID could not be ascertained from the environment. Raises: ~google.auth.exceptions.DefaultCredentialsError: If no credentials were found, or if the credentials found were invalid. """ from google.auth.credentials import with_scopes_if_required from google.auth.credentials import CredentialsWithQuotaProject explicit_project_id = os.environ.get( environment_vars.PROJECT, os.environ.get(environment_vars.LEGACY_PROJECT) ) checkers = ( # Avoid passing scopes here to prevent passing scopes to user credentials. # 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_explicit_environ_credentials(quota_project_id=quota_project_id), lambda: _get_gcloud_sdk_credentials(quota_project_id=quota_project_id), _get_gae_credentials, lambda: _get_gce_credentials(request, quota_project_id=quota_project_id), ) for checker in checkers: credentials, project_id = checker() if credentials is not None: credentials = with_scopes_if_required( credentials, scopes, default_scopes=default_scopes ) effective_project_id = explicit_project_id or project_id # For external account credentials, scopes are required to determine # the project ID. Try to get the project ID again if not yet # determined. if not effective_project_id and callable( getattr(credentials, "get_project_id", None) ): if request is None: import google.auth.transport.requests request = google.auth.transport.requests.Request() effective_project_id = credentials.get_project_id(request=request) if quota_project_id and isinstance( credentials, CredentialsWithQuotaProject ): credentials = credentials.with_quota_project(quota_project_id) if not effective_project_id: _LOGGER.warning( "No project ID could be determined. Consider running " "`gcloud config set project` or setting the %s " "environment variable", environment_vars.PROJECT, ) return credentials, effective_project_id raise exceptions.DefaultCredentialsError(_CLOUD_SDK_MISSING_CREDENTIALS)	28.01 KB	2025-04-14 17:36
📄 _default_async.py # Copyright 2020 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Application default credentials. Implements application default credentials and project ID detection. """ import io import json import os from google.auth import _default from google.auth import environment_vars from google.auth import exceptions def load_credentials_from_file(filename, scopes=None, quota_project_id=None): """Loads Google credentials from a file. The credentials file must be a service account key or stored authorized user credentials. Args: filename (str): The full path to the credentials file. scopes (Optional[Sequence[str]]): The list of scopes for the credentials. If specified, the credentials will automatically be scoped if necessary quota_project_id (Optional[str]): The project ID used for quota and billing. Returns: Tuple[google.auth.credentials.Credentials, Optional[str]]: Loaded credentials and the project ID. Authorized user credentials do not have the project ID information. Raises: google.auth.exceptions.DefaultCredentialsError: if the file is in the wrong format or is missing. """ if not os.path.exists(filename): raise exceptions.DefaultCredentialsError( "File {} was not found.".format(filename) ) with io.open(filename, "r") as file_obj: try: info = json.load(file_obj) except ValueError as caught_exc: new_exc = exceptions.DefaultCredentialsError( "File {} is not a valid json file.".format(filename), caught_exc ) raise new_exc from caught_exc # The type key should indicate that the file is either a service account # credentials file or an authorized user credentials file. credential_type = info.get("type") if credential_type == _default._AUTHORIZED_USER_TYPE: from google.oauth2 import _credentials_async as credentials try: credentials = credentials.Credentials.from_authorized_user_info( info, scopes=scopes ) except ValueError as caught_exc: msg = "Failed to load authorized user credentials from {}".format(filename) new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) raise new_exc from caught_exc if quota_project_id: credentials = credentials.with_quota_project(quota_project_id) if not credentials.quota_project_id: _default._warn_about_problematic_credentials(credentials) return credentials, None elif credential_type == _default._SERVICE_ACCOUNT_TYPE: from google.oauth2 import _service_account_async as service_account try: credentials = service_account.Credentials.from_service_account_info( info, scopes=scopes ).with_quota_project(quota_project_id) except ValueError as caught_exc: msg = "Failed to load service account credentials from {}".format(filename) new_exc = exceptions.DefaultCredentialsError(msg, caught_exc) raise new_exc from caught_exc return credentials, info.get("project_id") else: raise exceptions.DefaultCredentialsError( "The file {file} does not have a valid type. " "Type is {type}, expected one of {valid_types}.".format( file=filename, type=credential_type, valid_types=_default._VALID_TYPES ) ) def _get_gcloud_sdk_credentials(quota_project_id=None): """Gets the credentials and project ID from the Cloud SDK.""" from google.auth import _cloud_sdk # Check if application default credentials exist. credentials_filename = _cloud_sdk.get_application_default_credentials_path() if not os.path.isfile(credentials_filename): return None, None credentials, project_id = load_credentials_from_file( credentials_filename, quota_project_id=quota_project_id ) if not project_id: project_id = _cloud_sdk.get_project_id() return credentials, project_id def _get_explicit_environ_credentials(quota_project_id=None): """Gets credentials from the GOOGLE_APPLICATION_CREDENTIALS environment variable.""" from google.auth import _cloud_sdk cloud_sdk_adc_path = _cloud_sdk.get_application_default_credentials_path() explicit_file = os.environ.get(environment_vars.CREDENTIALS) if explicit_file is not None and explicit_file == cloud_sdk_adc_path: # Cloud sdk flow calls gcloud to fetch project id, so if the explicit # file path is cloud sdk credentials path, then we should fall back # to cloud sdk flow, otherwise project id cannot be obtained. return _get_gcloud_sdk_credentials(quota_project_id=quota_project_id) if explicit_file is not None: credentials, project_id = load_credentials_from_file( os.environ[environment_vars.CREDENTIALS], quota_project_id=quota_project_id ) return credentials, project_id else: return None, None def _get_gae_credentials(): """Gets Google App Engine App Identity credentials and project ID.""" # While this library is normally bundled with app_engine, there are # some cases where it's not available, so we tolerate ImportError. return _default._get_gae_credentials() def _get_gce_credentials(request=None): """Gets credentials and project ID from the GCE Metadata Service.""" # Ping requires a transport, but we want application default credentials # to require no arguments. So, we'll use the _http_client transport which # uses http.client. This is only acceptable because the metadata server # doesn't do SSL and never requires proxies. # While this library is normally bundled with compute_engine, there are # some cases where it's not available, so we tolerate ImportError. return _default._get_gce_credentials(request) def default_async(scopes=None, request=None, quota_project_id=None): """Gets the default credentials for the current environment. `Application Default Credentials`_ provides an easy way to obtain credentials to call Google APIs for server-to-server or local applications. This function acquires credentials from the environment in the following order: 1. If the environment variable ``GOOGLE_APPLICATION_CREDENTIALS`` is set to the path of a valid service account JSON private key file, then it is loaded and returned. The project ID returned is the project ID defined in the service account file if available (some older files do not contain project ID information). 2. If the `Google Cloud SDK`_ is installed and has application default credentials set they are loaded and returned. To enable application default credentials with the Cloud SDK run:: gcloud auth application-default login If the Cloud SDK has an active project, the project ID is returned. The active project can be set using:: gcloud config set project 3. If the application is running in the `App Engine standard environment`_ (first generation) then the credentials and project ID from the `App Identity Service`_ are used. 4. If the application is running in `Compute Engine`_ or `Cloud Run`_ or the `App Engine flexible environment`_ or the `App Engine standard environment`_ (second generation) then the credentials and project ID are obtained from the `Metadata Service`_. 5. If no credentials are found, :class:`~google.auth.exceptions.DefaultCredentialsError` will be raised. .. _Application Default Credentials: https://developers.google.com\ /identity/protocols/application-default-credentials .. _Google Cloud SDK: https://cloud.google.com/sdk .. _App Engine standard environment: https://cloud.google.com/appengine .. _App Identity Service: https://cloud.google.com/appengine/docs/python\ /appidentity/ .. _Compute Engine: https://cloud.google.com/compute .. _App Engine flexible environment: https://cloud.google.com\ /appengine/flexible .. _Metadata Service: https://cloud.google.com/compute/docs\ /storing-retrieving-metadata .. _Cloud Run: https://cloud.google.com/run Example:: import google.auth credentials, project_id = google.auth.default() Args: scopes (Sequence[str]): The list of scopes for the credentials. If specified, the credentials will automatically be scoped if necessary. request (google.auth.transport.Request): An object used to make HTTP requests. This is used to detect whether the application is running on Compute Engine. If not specified, then it will use the standard library http client to make requests. quota_project_id (Optional[str]): The project ID used for quota and billing. Returns: Tuple[~google.auth.credentials.Credentials, Optional[str]]: the current environment's credentials and project ID. Project ID may be None, which indicates that the Project ID could not be ascertained from the environment. Raises: ~google.auth.exceptions.DefaultCredentialsError: If no credentials were found, or if the credentials found were invalid. """ from google.auth._credentials_async import with_scopes_if_required from google.auth.credentials import CredentialsWithQuotaProject explicit_project_id = os.environ.get( environment_vars.PROJECT, os.environ.get(environment_vars.LEGACY_PROJECT) ) checkers = ( 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, lambda: _get_gce_credentials(request), ) for checker in checkers: credentials, project_id = checker() if credentials is not None: credentials = with_scopes_if_required(credentials, scopes) if quota_project_id and isinstance( credentials, CredentialsWithQuotaProject ): credentials = credentials.with_quota_project(quota_project_id) effective_project_id = explicit_project_id or project_id if not effective_project_id: _default._LOGGER.warning( "No project ID could be determined. Consider running " "`gcloud config set project` or setting the %s " "environment variable", environment_vars.PROJECT, ) return credentials, effective_project_id raise exceptions.DefaultCredentialsError(_default._CLOUD_SDK_MISSING_CREDENTIALS)	11.3 KB	2025-04-14 17:36
📄 _exponential_backoff.py # Copyright 2022 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import asyncio import random import time from google.auth import exceptions # The default amount of retry attempts _DEFAULT_RETRY_TOTAL_ATTEMPTS = 3 # The default initial backoff period (1.0 second). _DEFAULT_INITIAL_INTERVAL_SECONDS = 1.0 # The default randomization factor (0.1 which results in a random period ranging # between 10% below and 10% above the retry interval). _DEFAULT_RANDOMIZATION_FACTOR = 0.1 # The default multiplier value (2 which is 100% increase per back off). _DEFAULT_MULTIPLIER = 2.0 """Exponential Backoff Utility This is a private module that implements the exponential back off algorithm. It can be used as a utility for code that needs to retry on failure, for example an HTTP request. """ class _BaseExponentialBackoff: """An exponential backoff iterator base class. Args: total_attempts Optional[int]: The maximum amount of retries that should happen. The default value is 3 attempts. initial_wait_seconds Optional[int]: The amount of time to sleep in the first backoff. This parameter should be in seconds. The default value is 1 second. randomization_factor Optional[float]: The amount of jitter that should be in each backoff. For example, a value of 0.1 will introduce a jitter range of 10% to the current backoff period. The default value is 0.1. multiplier Optional[float]: The backoff multipler. This adjusts how much each backoff will increase. For example a value of 2.0 leads to a 200% backoff on each attempt. If the initial_wait is 1.0 it would look like this sequence [1.0, 2.0, 4.0, 8.0]. The default value is 2.0. """ def __init__( self, total_attempts=_DEFAULT_RETRY_TOTAL_ATTEMPTS, initial_wait_seconds=_DEFAULT_INITIAL_INTERVAL_SECONDS, randomization_factor=_DEFAULT_RANDOMIZATION_FACTOR, multiplier=_DEFAULT_MULTIPLIER, ): if total_attempts < 1: raise exceptions.InvalidValue( f"total_attempts must be greater than or equal to 1 but was {total_attempts}" ) self._total_attempts = total_attempts self._initial_wait_seconds = initial_wait_seconds self._current_wait_in_seconds = self._initial_wait_seconds self._randomization_factor = randomization_factor self._multiplier = multiplier self._backoff_count = 0 @property def total_attempts(self): """The total amount of backoff attempts that will be made.""" return self._total_attempts @property def backoff_count(self): """The current amount of backoff attempts that have been made.""" return self._backoff_count def _reset(self): self._backoff_count = 0 self._current_wait_in_seconds = self._initial_wait_seconds def _calculate_jitter(self): jitter_variance = self._current_wait_in_seconds * self._randomization_factor jitter = random.uniform( self._current_wait_in_seconds - jitter_variance, self._current_wait_in_seconds + jitter_variance, ) return jitter class ExponentialBackoff(_BaseExponentialBackoff): """An exponential backoff iterator. This can be used in a for loop to perform requests with exponential backoff. """ def __init__(self, args, kwargs): super(ExponentialBackoff, self).__init__(args, *kwargs) def __iter__(self): self._reset() return self def __next__(self): if self._backoff_count >= self._total_attempts: raise StopIteration self._backoff_count += 1 if self._backoff_count <= 1: return self._backoff_count jitter = self._calculate_jitter() time.sleep(jitter) self._current_wait_in_seconds = self._multiplier return self._backoff_count class AsyncExponentialBackoff(_BaseExponentialBackoff): """An async exponential backoff iterator. This can be used in a for loop to perform async requests with exponential backoff. """ def __init__(self, args, kwargs): super(AsyncExponentialBackoff, self).__init__(args, *kwargs) def __aiter__(self): self._reset() return self async def __anext__(self): if self._backoff_count >= self._total_attempts: raise StopAsyncIteration self._backoff_count += 1 if self._backoff_count <= 1: return self._backoff_count jitter = self._calculate_jitter() await asyncio.sleep(jitter) self._current_wait_in_seconds = self._multiplier return self._backoff_count	5.25 KB	2025-04-14 17:36
📄 _helpers.py # Copyright 2015 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Helper functions for commonly used utilities.""" import base64 import calendar import datetime from email.message import Message import sys import urllib from google.auth import exceptions # The smallest MDS cache used by this library stores tokens until 4 minutes from # expiry. REFRESH_THRESHOLD = datetime.timedelta(minutes=3, seconds=45) def copy_docstring(source_class): """Decorator that copies a method's docstring from another class. Args: source_class (type): The class that has the documented method. Returns: Callable: A decorator that will copy the docstring of the same named method in the source class to the decorated method. """ def decorator(method): """Decorator implementation. Args: method (Callable): The method to copy the docstring to. Returns: Callable: the same method passed in with an updated docstring. Raises: google.auth.exceptions.InvalidOperation: if the method already has a docstring. """ if method.__doc__: raise exceptions.InvalidOperation("Method already has a docstring.") source_method = getattr(source_class, method.__name__) method.__doc__ = source_method.__doc__ return method return decorator def parse_content_type(header_value): """Parse a 'content-type' header value to get just the plain media-type (without parameters). This is done using the class Message from email.message as suggested in PEP 594 (because the cgi is now deprecated and will be removed in python 3.13, see https://peps.python.org/pep-0594/#cgi). Args: header_value (str): The value of a 'content-type' header as a string. Returns: str: A string with just the lowercase media-type from the parsed 'content-type' header. If the provided content-type is not parsable, returns 'text/plain', the default value for textual files. """ m = Message() m["content-type"] = header_value return ( m.get_content_type() ) # Despite the name, actually returns just the media-type def utcnow(): """Returns the current UTC datetime. Returns: datetime: The current time in UTC. """ # We used datetime.utcnow() before, since it's deprecated from python 3.12, # we are using datetime.now(timezone.utc) now. "utcnow()" is offset-native # (no timezone info), but "now()" is offset-aware (with timezone info). # This will cause datetime comparison problem. For backward compatibility, # we need to remove the timezone info. now = datetime.datetime.now(datetime.timezone.utc) now = now.replace(tzinfo=None) return now def datetime_to_secs(value): """Convert a datetime object to the number of seconds since the UNIX epoch. Args: value (datetime): The datetime to convert. Returns: int: The number of seconds since the UNIX epoch. """ return calendar.timegm(value.utctimetuple()) def to_bytes(value, encoding="utf-8"): """Converts a string value to bytes, if necessary. Args: value (Union[str, bytes]): The value to be converted. encoding (str): The encoding to use to convert unicode to bytes. Defaults to "utf-8". Returns: bytes: The original value converted to bytes (if unicode) or as passed in if it started out as bytes. Raises: google.auth.exceptions.InvalidValue: If the value could not be converted to bytes. """ result = value.encode(encoding) if isinstance(value, str) else value if isinstance(result, bytes): return result else: raise exceptions.InvalidValue( "{0!r} could not be converted to bytes".format(value) ) def from_bytes(value): """Converts bytes to a string value, if necessary. Args: value (Union[str, bytes]): The value to be converted. Returns: str: The original value converted to unicode (if bytes) or as passed in if it started out as unicode. Raises: google.auth.exceptions.InvalidValue: If the value could not be converted to unicode. """ result = value.decode("utf-8") if isinstance(value, bytes) else value if isinstance(result, str): return result else: raise exceptions.InvalidValue( "{0!r} could not be converted to unicode".format(value) ) def update_query(url, params, remove=None): """Updates a URL's query parameters. Replaces any current values if they are already present in the URL. Args: url (str): The URL to update. params (Mapping[str, str]): A mapping of query parameter keys to values. remove (Sequence[str]): Parameters to remove from the query string. Returns: str: The URL with updated query parameters. Examples: >>> url = 'http://example.com?a=1' >>> update_query(url, {'a': '2'}) http://example.com?a=2 >>> update_query(url, {'b': '3'}) http://example.com?a=1&b=3 >> update_query(url, {'b': '3'}, remove=['a']) http://example.com?b=3 """ if remove is None: remove = [] # Split the URL into parts. parts = urllib.parse.urlparse(url) # Parse the query string. query_params = urllib.parse.parse_qs(parts.query) # Update the query parameters with the new parameters. query_params.update(params) # Remove any values specified in remove. query_params = { key: value for key, value in query_params.items() if key not in remove } # Re-encoded the query string. new_query = urllib.parse.urlencode(query_params, doseq=True) # Unsplit the url. new_parts = parts._replace(query=new_query) return urllib.parse.urlunparse(new_parts) def scopes_to_string(scopes): """Converts scope value to a string suitable for sending to OAuth 2.0 authorization servers. Args: scopes (Sequence[str]): The sequence of scopes to convert. Returns: str: The scopes formatted as a single string. """ return " ".join(scopes) def string_to_scopes(scopes): """Converts stringifed scopes value to a list. Args: scopes (Union[Sequence, str]): The string of space-separated scopes to convert. Returns: Sequence(str): The separated scopes. """ if not scopes: return [] return scopes.split(" ") def padded_urlsafe_b64decode(value): """Decodes base64 strings lacking padding characters. Google infrastructure tends to omit the base64 padding characters. Args: value (Union[str, bytes]): The encoded value. Returns: bytes: The decoded value """ b64string = to_bytes(value) padded = b64string + b"=" * (-len(b64string) % 4) return base64.urlsafe_b64decode(padded) def unpadded_urlsafe_b64encode(value): """Encodes base64 strings removing any padding characters. `rfc 7515`_ defines Base64url to NOT include any padding characters, but the stdlib doesn't do that by default. _rfc7515: https://tools.ietf.org/html/rfc7515#page-6 Args: value (Union[str\|bytes]): The bytes-like value to encode Returns: Union[str\|bytes]: The encoded value """ return base64.urlsafe_b64encode(value).rstrip(b"=") def is_python_3(): """Check if the Python interpreter is Python 2 or 3. Returns: bool: True if the Python interpreter is Python 3 and False otherwise. """ return sys.version_info > (3, 0)	8.04 KB	2025-04-14 17:36
📄 _jwt_async.py # Copyright 2020 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """JSON Web Tokens Provides support for creating (encoding) and verifying (decoding) JWTs, especially JWTs generated and consumed by Google infrastructure. See `rfc7519`_ for more details on JWTs. To encode a JWT use :func:`encode`:: from google.auth import crypt from google.auth import jwt_async signer = crypt.Signer(private_key) payload = {'some': 'payload'} encoded = jwt_async.encode(signer, payload) To decode a JWT and verify claims use :func:`decode`:: claims = jwt_async.decode(encoded, certs=public_certs) You can also skip verification:: claims = jwt_async.decode(encoded, verify=False) .. _rfc7519: https://tools.ietf.org/html/rfc7519 NOTE: This async support is experimental and marked internal. This surface may change in minor releases. """ from google.auth import _credentials_async from google.auth import jwt def encode(signer, payload, header=None, key_id=None): """Make a signed JWT. Args: signer (google.auth.crypt.Signer): The signer used to sign the JWT. payload (Mapping[str, str]): The JWT payload. header (Mapping[str, str]): Additional JWT header payload. key_id (str): The key id to add to the JWT header. If the signer has a key id it will be used as the default. If this is specified it will override the signer's key id. Returns: bytes: The encoded JWT. """ return jwt.encode(signer, payload, header, key_id) def decode(token, certs=None, verify=True, audience=None): """Decode and verify a JWT. Args: token (str): The encoded JWT. certs (Union[str, bytes, Mapping[str, Union[str, bytes]]]): The certificate used to validate the JWT signature. If bytes or string, it must the the public key certificate in PEM format. If a mapping, it must be a mapping of key IDs to public key certificates in PEM format. The mapping must contain the same key ID that's specified in the token's header. verify (bool): Whether to perform signature and claim validation. Verification is done by default. audience (str): The audience claim, 'aud', that this JWT should contain. If None then the JWT's 'aud' parameter is not verified. Returns: Mapping[str, str]: The deserialized JSON payload in the JWT. Raises: ValueError: if any verification checks failed. """ return jwt.decode(token, certs, verify, audience) class Credentials( jwt.Credentials, _credentials_async.Signing, _credentials_async.Credentials ): """Credentials that use a JWT as the bearer token. These credentials require an "audience" claim. This claim identifies the intended recipient of the bearer token. The constructor arguments determine the claims for the JWT that is sent with requests. Usually, you'll construct these credentials with one of the helper constructors as shown in the next section. To create JWT credentials using a Google service account private key JSON file:: audience = 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher' credentials = jwt_async.Credentials.from_service_account_file( 'service-account.json', audience=audience) If you already have the service account file loaded and parsed:: service_account_info = json.load(open('service_account.json')) credentials = jwt_async.Credentials.from_service_account_info( service_account_info, audience=audience) Both helper methods pass on arguments to the constructor, so you can specify the JWT claims:: credentials = jwt_async.Credentials.from_service_account_file( 'service-account.json', audience=audience, additional_claims={'meta': 'data'}) You can also construct the credentials directly if you have a :class:`~google.auth.crypt.Signer` instance:: credentials = jwt_async.Credentials( signer, issuer='your-issuer', subject='your-subject', audience=audience) The claims are considered immutable. If you want to modify the claims, you can easily create another instance using :meth:`with_claims`:: new_audience = ( 'https://pubsub.googleapis.com/google.pubsub.v1.Subscriber') new_credentials = credentials.with_claims(audience=new_audience) """ class OnDemandCredentials( jwt.OnDemandCredentials, _credentials_async.Signing, _credentials_async.Credentials ): """On-demand JWT credentials. Like :class:`Credentials`, this class uses a JWT as the bearer token for authentication. However, this class does not require the audience at construction time. Instead, it will generate a new token on-demand for each request using the request URI as the audience. It caches tokens so that multiple requests to the same URI do not incur the overhead of generating a new token every time. This behavior is especially useful for `gRPC`_ clients. A gRPC service may have multiple audience and gRPC clients may not know all of the audiences required for accessing a particular service. With these credentials, no knowledge of the audiences is required ahead of time. .. _grpc: http://www.grpc.io/ """	5.83 KB	2025-04-14 17:36
📄 _oauth2client.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Helpers for transitioning from oauth2client to google-auth. .. warning:: This module is private as it is intended to assist first-party downstream clients with the transition from oauth2client to google-auth. """ from __future__ import absolute_import from google.auth import _helpers import google.auth.app_engine import google.auth.compute_engine import google.oauth2.credentials import google.oauth2.service_account try: import oauth2client.client # type: ignore import oauth2client.contrib.gce # type: ignore import oauth2client.service_account # type: ignore except ImportError as caught_exc: raise ImportError("oauth2client is not installed.") from caught_exc try: import oauth2client.contrib.appengine # type: ignore _HAS_APPENGINE = True except ImportError: _HAS_APPENGINE = False _CONVERT_ERROR_TMPL = "Unable to convert {} to a google-auth credentials class." def _convert_oauth2_credentials(credentials): """Converts to :class:`google.oauth2.credentials.Credentials`. Args: credentials (Union[oauth2client.client.OAuth2Credentials, oauth2client.client.GoogleCredentials]): The credentials to convert. Returns: google.oauth2.credentials.Credentials: The converted credentials. """ new_credentials = google.oauth2.credentials.Credentials( token=credentials.access_token, refresh_token=credentials.refresh_token, token_uri=credentials.token_uri, client_id=credentials.client_id, client_secret=credentials.client_secret, scopes=credentials.scopes, ) new_credentials._expires = credentials.token_expiry return new_credentials def _convert_service_account_credentials(credentials): """Converts to :class:`google.oauth2.service_account.Credentials`. Args: credentials (Union[ oauth2client.service_account.ServiceAccountCredentials, oauth2client.service_account._JWTAccessCredentials]): The credentials to convert. Returns: google.oauth2.service_account.Credentials: The converted credentials. """ info = credentials.serialization_data.copy() info["token_uri"] = credentials.token_uri return google.oauth2.service_account.Credentials.from_service_account_info(info) def _convert_gce_app_assertion_credentials(credentials): """Converts to :class:`google.auth.compute_engine.Credentials`. Args: credentials (oauth2client.contrib.gce.AppAssertionCredentials): The credentials to convert. Returns: google.oauth2.service_account.Credentials: The converted credentials. """ return google.auth.compute_engine.Credentials( service_account_email=credentials.service_account_email ) def _convert_appengine_app_assertion_credentials(credentials): """Converts to :class:`google.auth.app_engine.Credentials`. Args: credentials (oauth2client.contrib.app_engine.AppAssertionCredentials): The credentials to convert. Returns: google.oauth2.service_account.Credentials: The converted credentials. """ # pylint: disable=invalid-name return google.auth.app_engine.Credentials( scopes=_helpers.string_to_scopes(credentials.scope), service_account_id=credentials.service_account_id, ) _CLASS_CONVERSION_MAP = { oauth2client.client.OAuth2Credentials: _convert_oauth2_credentials, oauth2client.client.GoogleCredentials: _convert_oauth2_credentials, oauth2client.service_account.ServiceAccountCredentials: _convert_service_account_credentials, oauth2client.service_account._JWTAccessCredentials: _convert_service_account_credentials, oauth2client.contrib.gce.AppAssertionCredentials: _convert_gce_app_assertion_credentials, } if _HAS_APPENGINE: _CLASS_CONVERSION_MAP[ oauth2client.contrib.appengine.AppAssertionCredentials ] = _convert_appengine_app_assertion_credentials def convert(credentials): """Convert oauth2client credentials to google-auth credentials. This class converts: - :class:`oauth2client.client.OAuth2Credentials` to :class:`google.oauth2.credentials.Credentials`. - :class:`oauth2client.client.GoogleCredentials` to :class:`google.oauth2.credentials.Credentials`. - :class:`oauth2client.service_account.ServiceAccountCredentials` to :class:`google.oauth2.service_account.Credentials`. - :class:`oauth2client.service_account._JWTAccessCredentials` to :class:`google.oauth2.service_account.Credentials`. - :class:`oauth2client.contrib.gce.AppAssertionCredentials` to :class:`google.auth.compute_engine.Credentials`. - :class:`oauth2client.contrib.appengine.AppAssertionCredentials` to :class:`google.auth.app_engine.Credentials`. Returns: google.auth.credentials.Credentials: The converted credentials. Raises: ValueError: If the credentials could not be converted. """ credentials_class = type(credentials) try: return _CLASS_CONVERSION_MAP[credentials_class](credentials) except KeyError as caught_exc: new_exc = ValueError(_CONVERT_ERROR_TMPL.format(credentials_class)) raise new_exc from caught_exc	5.72 KB	2025-04-14 17:36
📄 _refresh_worker.py # Copyright 2023 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. import copy import logging import threading import google.auth.exceptions as e _LOGGER = logging.getLogger(__name__) class RefreshThreadManager: """ Organizes exactly one background job that refresh a token. """ def __init__(self): """Initializes the manager.""" self._worker = None self._lock = threading.Lock() # protects access to worker threads. def start_refresh(self, cred, request): """Starts a refresh thread for the given credentials. The credentials are refreshed using the request parameter. request and cred MUST not be None Returns True if a background refresh was kicked off. False otherwise. Args: cred: A credentials object. request: A request object. Returns: bool """ if cred is None or request is None: raise e.InvalidValue( "Unable to start refresh. cred and request must be valid and instantiated objects." ) with self._lock: if self._worker is not None and self._worker._error_info is not None: return False if self._worker is None or not self._worker.is_alive(): # pragma: NO COVER self._worker = RefreshThread(cred=cred, request=copy.deepcopy(request)) self._worker.start() return True def clear_error(self): """ Removes any errors that were stored from previous background refreshes. """ with self._lock: if self._worker: self._worker._error_info = None def __getstate__(self): """Pickle helper that serializes the _lock attribute.""" state = self.__dict__.copy() state["_lock"] = None return state def __setstate__(self, state): """Pickle helper that deserializes the _lock attribute.""" state["_lock"] = threading.Lock() self.__dict__.update(state) class RefreshThread(threading.Thread): """ Thread that refreshes credentials. """ def __init__(self, cred, request, kwargs): """Initializes the thread. Args: cred: A Credential object to refresh. request: A Request object used to perform a credential refresh. kwargs: Additional keyword arguments. """ super().__init__(**kwargs) self._cred = cred self._request = request self._error_info = None def run(self): """ Perform the credential refresh. """ try: self._cred.refresh(self._request) except Exception as err: # pragma: NO COVER _LOGGER.error(f"Background refresh failed due to: {err}") self._error_info = err	3.3 KB	2025-04-14 17:36
📄 _service_account_info.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Helper functions for loading data from a Google service account file.""" import io import json from google.auth import crypt from google.auth import exceptions def from_dict(data, require=None, use_rsa_signer=True): """Validates a dictionary containing Google service account data. Creates and returns a :class:`google.auth.crypt.Signer` instance from the private key specified in the data. Args: data (Mapping[str, str]): The service account data require (Sequence[str]): List of keys required to be present in the info. use_rsa_signer (Optional[bool]): Whether to use RSA signer or EC signer. We use RSA signer by default. Returns: google.auth.crypt.Signer: A signer created from the private key in the service account file. Raises: MalformedError: if the data was in the wrong format, or if one of the required keys is missing. """ keys_needed = set(require if require is not None else []) missing = keys_needed.difference(data.keys()) if missing: raise exceptions.MalformedError( "Service account info was not in the expected format, missing " "fields {}.".format(", ".join(missing)) ) # Create a signer. if use_rsa_signer: signer = crypt.RSASigner.from_service_account_info(data) else: signer = crypt.ES256Signer.from_service_account_info(data) return signer def from_filename(filename, require=None, use_rsa_signer=True): """Reads a Google service account JSON file and returns its parsed info. Args: filename (str): The path to the service account .json file. require (Sequence[str]): List of keys required to be present in the info. use_rsa_signer (Optional[bool]): Whether to use RSA signer or EC signer. We use RSA signer by default. Returns: Tuple[ Mapping[str, str], google.auth.crypt.Signer ]: The verified info and a signer instance. """ with io.open(filename, "r", encoding="utf-8") as json_file: data = json.load(json_file) return data, from_dict(data, require=require, use_rsa_signer=use_rsa_signer)	2.75 KB	2025-04-14 17:36
📁 aio	-	2025-04-14 17:36
📄 api_key.py # Copyright 2022 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Google API key support. This module provides authentication using the `API key`_. .. _API key: https://cloud.google.com/docs/authentication/api-keys/ """ from google.auth import _helpers from google.auth import credentials from google.auth import exceptions class Credentials(credentials.Credentials): """API key credentials. These credentials use API key to provide authorization to applications. """ def __init__(self, token): """ Args: token (str): API key string Raises: ValueError: If the provided API key is not a non-empty string. """ super(Credentials, self).__init__() if not token: raise exceptions.InvalidValue("Token must be a non-empty API key string") self.token = token @property def expired(self): return False @property def valid(self): return True @_helpers.copy_docstring(credentials.Credentials) def refresh(self, request): return def apply(self, headers, token=None): """Apply the API key token to the x-goog-api-key header. Args: headers (Mapping): The HTTP request headers. token (Optional[str]): If specified, overrides the current access token. """ headers["x-goog-api-key"] = token or self.token def before_request(self, request, method, url, headers): """Performs credential-specific before request logic. Refreshes the credentials if necessary, then calls :meth:`apply` to apply the token to the x-goog-api-key header. Args: request (google.auth.transport.Request): The object used to make HTTP requests. method (str): The request's HTTP method or the RPC method being invoked. url (str): The request's URI or the RPC service's URI. headers (Mapping): The request's headers. """ self.apply(headers)	2.52 KB	2025-04-14 17:36
📄 app_engine.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Google App Engine standard environment support. This module provides authentication and signing for applications running on App Engine in the standard environment using the `App Identity API`_. .. _App Identity API: https://cloud.google.com/appengine/docs/python/appidentity/ """ import datetime from google.auth import _helpers from google.auth import credentials from google.auth import crypt from google.auth import exceptions # pytype: disable=import-error try: from google.appengine.api import app_identity # type: ignore except ImportError: app_identity = None # type: ignore # pytype: enable=import-error class Signer(crypt.Signer): """Signs messages using the App Engine App Identity service. This can be used in place of :class:`google.auth.crypt.Signer` when running in the App Engine standard environment. """ @property def key_id(self): """Optional[str]: The key ID used to identify this private key. .. warning:: This is always ``None``. The key ID used by App Engine can not be reliably determined ahead of time. """ return None @_helpers.copy_docstring(crypt.Signer) def sign(self, message): message = _helpers.to_bytes(message) _, signature = app_identity.sign_blob(message) return signature def get_project_id(): """Gets the project ID for the current App Engine application. Returns: str: The project ID Raises: google.auth.exceptions.OSError: If the App Engine APIs are unavailable. """ # pylint: disable=missing-raises-doc # Pylint rightfully thinks google.auth.exceptions.OSError is OSError, but doesn't # realize it's a valid alias. if app_identity is None: raise exceptions.OSError("The App Engine APIs are not available.") return app_identity.get_application_id() class Credentials( credentials.Scoped, credentials.Signing, credentials.CredentialsWithQuotaProject ): """App Engine standard environment credentials. These credentials use the App Engine App Identity API to obtain access tokens. """ def __init__( self, scopes=None, default_scopes=None, service_account_id=None, quota_project_id=None, ): """ Args: scopes (Sequence[str]): Scopes to request from the App Identity API. default_scopes (Sequence[str]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. service_account_id (str): The service account ID passed into :func:`google.appengine.api.app_identity.get_access_token`. If not specified, the default application service account ID will be used. quota_project_id (Optional[str]): The project ID used for quota and billing. Raises: google.auth.exceptions.OSError: If the App Engine APIs are unavailable. """ # pylint: disable=missing-raises-doc # Pylint rightfully thinks google.auth.exceptions.OSError is OSError, but doesn't # realize it's a valid alias. if app_identity is None: raise exceptions.OSError("The App Engine APIs are not available.") super(Credentials, self).__init__() self._scopes = scopes self._default_scopes = default_scopes self._service_account_id = service_account_id self._signer = Signer() self._quota_project_id = quota_project_id @_helpers.copy_docstring(credentials.Credentials) def refresh(self, request): scopes = self._scopes if self._scopes is not None else self._default_scopes # pylint: disable=unused-argument token, ttl = app_identity.get_access_token(scopes, self._service_account_id) expiry = datetime.datetime.utcfromtimestamp(ttl) self.token, self.expiry = token, expiry @property def service_account_email(self): """The service account email.""" if self._service_account_id is None: self._service_account_id = app_identity.get_service_account_name() return self._service_account_id @property def requires_scopes(self): """Checks if the credentials requires scopes. Returns: bool: True if there are no scopes set otherwise False. """ return not self._scopes and not self._default_scopes @_helpers.copy_docstring(credentials.Scoped) def with_scopes(self, scopes, default_scopes=None): return self.__class__( scopes=scopes, default_scopes=default_scopes, service_account_id=self._service_account_id, quota_project_id=self.quota_project_id, ) @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): return self.__class__( scopes=self._scopes, service_account_id=self._service_account_id, quota_project_id=quota_project_id, ) @_helpers.copy_docstring(credentials.Signing) def sign_bytes(self, message): return self._signer.sign(message) @property # type: ignore @_helpers.copy_docstring(credentials.Signing) def signer_email(self): return self.service_account_email @property # type: ignore @_helpers.copy_docstring(credentials.Signing) def signer(self): return self._signer	5.98 KB	2025-04-14 17:36
📄 aws.py # Copyright 2020 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """AWS Credentials and AWS Signature V4 Request Signer. This module provides credentials to access Google Cloud resources from Amazon Web Services (AWS) workloads. These credentials are recommended over the use of service account credentials in AWS as they do not involve the management of long-live service account private keys. AWS Credentials are initialized using external_account arguments which are typically loaded from the external credentials JSON file. This module also provides a definition for an abstract AWS security credentials supplier. This supplier can be implemented to return valid AWS security credentials and an AWS region and used to create AWS credentials. The credentials will then call the supplier instead of using pre-defined methods such as calling the EC2 metadata endpoints. This module also provides a basic implementation of the `AWS Signature Version 4`_ request signing algorithm. AWS Credentials use serialized signed requests to the `AWS STS GetCallerIdentity`_ API that can be exchanged for Google access tokens via the GCP STS endpoint. .. _AWS Signature Version 4: https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html .. _AWS STS GetCallerIdentity: https://docs.aws.amazon.com/STS/latest/APIReference/API_GetCallerIdentity.html """ import abc from dataclasses import dataclass import hashlib import hmac import http.client as http_client import json import os import posixpath import re from typing import Optional import urllib from urllib.parse import urljoin from google.auth import _helpers from google.auth import environment_vars from google.auth import exceptions from google.auth import external_account # AWS Signature Version 4 signing algorithm identifier. _AWS_ALGORITHM = "AWS4-HMAC-SHA256" # The termination string for the AWS credential scope value as defined in # https://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html _AWS_REQUEST_TYPE = "aws4_request" # The AWS authorization header name for the security session token if available. _AWS_SECURITY_TOKEN_HEADER = "x-amz-security-token" # The AWS authorization header name for the auto-generated date. _AWS_DATE_HEADER = "x-amz-date" # The default AWS regional credential verification URL. _DEFAULT_AWS_REGIONAL_CREDENTIAL_VERIFICATION_URL = ( "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15" ) # IMDSV2 session token lifetime. This is set to a low value because the session token is used immediately. _IMDSV2_SESSION_TOKEN_TTL_SECONDS = "300" class RequestSigner(object): """Implements an AWS request signer based on the AWS Signature Version 4 signing process. https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html """ def __init__(self, region_name): """Instantiates an AWS request signer used to compute authenticated signed requests to AWS APIs based on the AWS Signature Version 4 signing process. Args: region_name (str): The AWS region to use. """ self._region_name = region_name def get_request_options( self, aws_security_credentials, url, method, request_payload="", additional_headers={}, ): """Generates the signed request for the provided HTTP request for calling an AWS API. This follows the steps described at: https://docs.aws.amazon.com/general/latest/gr/sigv4_signing.html Args: aws_security_credentials (AWSSecurityCredentials): The AWS security credentials. url (str): The AWS service URL containing the canonical URI and query string. method (str): The HTTP method used to call this API. request_payload (Optional[str]): The optional request payload if available. additional_headers (Optional[Mapping[str, str]]): The optional additional headers needed for the requested AWS API. Returns: Mapping[str, str]: The AWS signed request dictionary object. """ additional_headers = additional_headers or {} uri = urllib.parse.urlparse(url) # Normalize the URL path. This is needed for the canonical_uri. # os.path.normpath can't be used since it normalizes "/" paths # to "\\" in Windows OS. normalized_uri = urllib.parse.urlparse( urljoin(url, posixpath.normpath(uri.path)) ) # Validate provided URL. if not uri.hostname or uri.scheme != "https": raise exceptions.InvalidResource("Invalid AWS service URL") header_map = _generate_authentication_header_map( host=uri.hostname, canonical_uri=normalized_uri.path or "/", canonical_querystring=_get_canonical_querystring(uri.query), method=method, region=self._region_name, aws_security_credentials=aws_security_credentials, request_payload=request_payload, additional_headers=additional_headers, ) headers = { "Authorization": header_map.get("authorization_header"), "host": uri.hostname, } # Add x-amz-date if available. if "amz_date" in header_map: headers[_AWS_DATE_HEADER] = header_map.get("amz_date") # Append additional optional headers, eg. X-Amz-Target, Content-Type, etc. for key in additional_headers: headers[key] = additional_headers[key] # Add session token if available. if aws_security_credentials.session_token is not None: headers[_AWS_SECURITY_TOKEN_HEADER] = aws_security_credentials.session_token signed_request = {"url": url, "method": method, "headers": headers} if request_payload: signed_request["data"] = request_payload return signed_request def _get_canonical_querystring(query): """Generates the canonical query string given a raw query string. Logic is based on https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html Args: query (str): The raw query string. Returns: str: The canonical query string. """ # Parse raw query string. querystring = urllib.parse.parse_qs(query) querystring_encoded_map = {} for key in querystring: quote_key = urllib.parse.quote(key, safe="-_.~") # URI encode key. querystring_encoded_map[quote_key] = [] for item in querystring[key]: # For each key, URI encode all values for that key. querystring_encoded_map[quote_key].append( urllib.parse.quote(item, safe="-_.~") ) # Sort values for each key. querystring_encoded_map[quote_key].sort() # Sort keys. sorted_keys = list(querystring_encoded_map.keys()) sorted_keys.sort() # Reconstruct the query string. Preserve keys with multiple values. querystring_encoded_pairs = [] for key in sorted_keys: for item in querystring_encoded_map[key]: querystring_encoded_pairs.append("{}={}".format(key, item)) return "&".join(querystring_encoded_pairs) def _sign(key, msg): """Creates the HMAC-SHA256 hash of the provided message using the provided key. Args: key (str): The HMAC-SHA256 key to use. msg (str): The message to hash. Returns: str: The computed hash bytes. """ return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest() def _get_signing_key(key, date_stamp, region_name, service_name): """Calculates the signing key used to calculate the signature for AWS Signature Version 4 based on: https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html Args: key (str): The AWS secret access key. date_stamp (str): The '%Y%m%d' date format. region_name (str): The AWS region. service_name (str): The AWS service name, eg. sts. Returns: str: The signing key bytes. """ k_date = _sign(("AWS4" + key).encode("utf-8"), date_stamp) k_region = _sign(k_date, region_name) k_service = _sign(k_region, service_name) k_signing = _sign(k_service, "aws4_request") return k_signing def _generate_authentication_header_map( host, canonical_uri, canonical_querystring, method, region, aws_security_credentials, request_payload="", additional_headers={}, ): """Generates the authentication header map needed for generating the AWS Signature Version 4 signed request. Args: host (str): The AWS service URL hostname. canonical_uri (str): The AWS service URL path name. canonical_querystring (str): The AWS service URL query string. method (str): The HTTP method used to call this API. region (str): The AWS region. aws_security_credentials (AWSSecurityCredentials): The AWS security credentials. request_payload (Optional[str]): The optional request payload if available. additional_headers (Optional[Mapping[str, str]]): The optional additional headers needed for the requested AWS API. Returns: Mapping[str, str]: The AWS authentication header dictionary object. This contains the x-amz-date and authorization header information. """ # iam.amazonaws.com host => iam service. # sts.us-east-2.amazonaws.com host => sts service. service_name = host.split(".")[0] current_time = _helpers.utcnow() amz_date = current_time.strftime("%Y%m%dT%H%M%SZ") date_stamp = current_time.strftime("%Y%m%d") # Change all additional headers to be lower case. full_headers = {} for key in additional_headers: full_headers[key.lower()] = additional_headers[key] # Add AWS session token if available. if aws_security_credentials.session_token is not None: full_headers[ _AWS_SECURITY_TOKEN_HEADER ] = aws_security_credentials.session_token # Required headers full_headers["host"] = host # Do not use generated x-amz-date if the date header is provided. # Previously the date was not fixed with x-amz- and could be provided # manually. # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-header-value-trim.req if "date" not in full_headers: full_headers[_AWS_DATE_HEADER] = amz_date # Header keys need to be sorted alphabetically. canonical_headers = "" header_keys = list(full_headers.keys()) header_keys.sort() for key in header_keys: canonical_headers = "{}{}:{}\n".format( canonical_headers, key, full_headers[key] ) signed_headers = ";".join(header_keys) payload_hash = hashlib.sha256((request_payload or "").encode("utf-8")).hexdigest() # https://docs.aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html canonical_request = "{}\n{}\n{}\n{}\n{}\n{}".format( method, canonical_uri, canonical_querystring, canonical_headers, signed_headers, payload_hash, ) credential_scope = "{}/{}/{}/{}".format( date_stamp, region, service_name, _AWS_REQUEST_TYPE ) # https://docs.aws.amazon.com/general/latest/gr/sigv4-create-string-to-sign.html string_to_sign = "{}\n{}\n{}\n{}".format( _AWS_ALGORITHM, amz_date, credential_scope, hashlib.sha256(canonical_request.encode("utf-8")).hexdigest(), ) # https://docs.aws.amazon.com/general/latest/gr/sigv4-calculate-signature.html signing_key = _get_signing_key( aws_security_credentials.secret_access_key, date_stamp, region, service_name ) signature = hmac.new( signing_key, string_to_sign.encode("utf-8"), hashlib.sha256 ).hexdigest() # https://docs.aws.amazon.com/general/latest/gr/sigv4-add-signature-to-request.html authorization_header = "{} Credential={}/{}, SignedHeaders={}, Signature={}".format( _AWS_ALGORITHM, aws_security_credentials.access_key_id, credential_scope, signed_headers, signature, ) authentication_header = {"authorization_header": authorization_header} # Do not use generated x-amz-date if the date header is provided. if "date" not in full_headers: authentication_header["amz_date"] = amz_date return authentication_header @dataclass class AwsSecurityCredentials: """A class that models AWS security credentials with an optional session token. Attributes: access_key_id (str): The AWS security credentials access key id. secret_access_key (str): The AWS security credentials secret access key. session_token (Optional[str]): The optional AWS security credentials session token. This should be set when using temporary credentials. """ access_key_id: str secret_access_key: str session_token: Optional[str] = None class AwsSecurityCredentialsSupplier(metaclass=abc.ABCMeta): """Base class for AWS security credential suppliers. This can be implemented with custom logic to retrieve AWS security credentials to exchange for a Google Cloud access token. The AWS external account credential does not cache the AWS security credentials, so caching logic should be added in the implementation. """ @abc.abstractmethod def get_aws_security_credentials(self, context, request): """Returns the AWS security credentials for the requested context. .. warning: This is not cached by the calling Google credential, so caching logic should be implemented in the supplier. Args: context (google.auth.externalaccount.SupplierContext): The context object containing information about the requested audience and subject token type. request (google.auth.transport.Request): The object used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If an error is encountered during security credential retrieval logic. Returns: AwsSecurityCredentials: The requested AWS security credentials. """ raise NotImplementedError("") @abc.abstractmethod def get_aws_region(self, context, request): """Returns the AWS region for the requested context. Args: context (google.auth.externalaccount.SupplierContext): The context object containing information about the requested audience and subject token type. request (google.auth.transport.Request): The object used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If an error is encountered during region retrieval logic. Returns: str: The AWS region. """ raise NotImplementedError("") class _DefaultAwsSecurityCredentialsSupplier(AwsSecurityCredentialsSupplier): """Default implementation of AWS security credentials supplier. Supports retrieving credentials and region via EC2 metadata endpoints and environment variables. """ def __init__(self, credential_source): self._region_url = credential_source.get("region_url") self._security_credentials_url = credential_source.get("url") self._imdsv2_session_token_url = credential_source.get( "imdsv2_session_token_url" ) @_helpers.copy_docstring(AwsSecurityCredentialsSupplier) def get_aws_security_credentials(self, context, request): # Check environment variables for permanent credentials first. # https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html env_aws_access_key_id = os.environ.get(environment_vars.AWS_ACCESS_KEY_ID) env_aws_secret_access_key = os.environ.get( environment_vars.AWS_SECRET_ACCESS_KEY ) # This is normally not available for permanent credentials. env_aws_session_token = os.environ.get(environment_vars.AWS_SESSION_TOKEN) if env_aws_access_key_id and env_aws_secret_access_key: return AwsSecurityCredentials( env_aws_access_key_id, env_aws_secret_access_key, env_aws_session_token ) imdsv2_session_token = self._get_imdsv2_session_token(request) role_name = self._get_metadata_role_name(request, imdsv2_session_token) # Get security credentials. credentials = self._get_metadata_security_credentials( request, role_name, imdsv2_session_token ) return AwsSecurityCredentials( credentials.get("AccessKeyId"), credentials.get("SecretAccessKey"), credentials.get("Token"), ) @_helpers.copy_docstring(AwsSecurityCredentialsSupplier) def get_aws_region(self, context, request): # The AWS metadata server is not available in some AWS environments # such as AWS lambda. Instead, it is available via environment # variable. env_aws_region = os.environ.get(environment_vars.AWS_REGION) if env_aws_region is not None: return env_aws_region env_aws_region = os.environ.get(environment_vars.AWS_DEFAULT_REGION) if env_aws_region is not None: return env_aws_region if not self._region_url: raise exceptions.RefreshError("Unable to determine AWS region") headers = None imdsv2_session_token = self._get_imdsv2_session_token(request) if imdsv2_session_token is not None: headers = {"X-aws-ec2-metadata-token": imdsv2_session_token} response = request(url=self._region_url, method="GET", headers=headers) # Support both string and bytes type response.data. response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) if response.status != http_client.OK: raise exceptions.RefreshError( "Unable to retrieve AWS region: {}".format(response_body) ) # This endpoint will return the region in format: us-east-2b. # Only the us-east-2 part should be used. return response_body[:-1] def _get_imdsv2_session_token(self, request): if request is not None and self._imdsv2_session_token_url is not None: headers = { "X-aws-ec2-metadata-token-ttl-seconds": _IMDSV2_SESSION_TOKEN_TTL_SECONDS } imdsv2_session_token_response = request( url=self._imdsv2_session_token_url, method="PUT", headers=headers ) if imdsv2_session_token_response.status != http_client.OK: raise exceptions.RefreshError( "Unable to retrieve AWS Session Token: {}".format( imdsv2_session_token_response.data ) ) return imdsv2_session_token_response.data else: return None def _get_metadata_security_credentials( self, request, role_name, imdsv2_session_token ): """Retrieves the AWS security credentials required for signing AWS requests from the AWS metadata server. Args: request (google.auth.transport.Request): A callable used to make HTTP requests. role_name (str): The AWS role name required by the AWS metadata server security_credentials endpoint in order to return the credentials. imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a header in the requests to AWS metadata endpoint. Returns: Mapping[str, str]: The AWS metadata server security credentials response. Raises: google.auth.exceptions.RefreshError: If an error occurs while retrieving the AWS security credentials. """ headers = {"Content-Type": "application/json"} if imdsv2_session_token is not None: headers["X-aws-ec2-metadata-token"] = imdsv2_session_token response = request( url="{}/{}".format(self._security_credentials_url, role_name), method="GET", headers=headers, ) # support both string and bytes type response.data response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) if response.status != http_client.OK: raise exceptions.RefreshError( "Unable to retrieve AWS security credentials: {}".format(response_body) ) credentials_response = json.loads(response_body) return credentials_response def _get_metadata_role_name(self, request, imdsv2_session_token): """Retrieves the AWS role currently attached to the current AWS workload by querying the AWS metadata server. This is needed for the AWS metadata server security credentials endpoint in order to retrieve the AWS security credentials needed to sign requests to AWS APIs. Args: request (google.auth.transport.Request): A callable used to make HTTP requests. imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a header in the requests to AWS metadata endpoint. Returns: str: The AWS role name. Raises: google.auth.exceptions.RefreshError: If an error occurs while retrieving the AWS role name. """ if self._security_credentials_url is None: raise exceptions.RefreshError( "Unable to determine the AWS metadata server security credentials endpoint" ) headers = None if imdsv2_session_token is not None: headers = {"X-aws-ec2-metadata-token": imdsv2_session_token} response = request( url=self._security_credentials_url, method="GET", headers=headers ) # support both string and bytes type response.data response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) if response.status != http_client.OK: raise exceptions.RefreshError( "Unable to retrieve AWS role name {}".format(response_body) ) return response_body class Credentials(external_account.Credentials): """AWS external account credentials. This is used to exchange serialized AWS signature v4 signed requests to AWS STS GetCallerIdentity service for Google access tokens. """ def __init__( self, audience, subject_token_type, token_url=external_account._DEFAULT_TOKEN_URL, credential_source=None, aws_security_credentials_supplier=None, args, kwargs ): """Instantiates an AWS workload external account credentials object. Args: audience (str): The STS audience field. subject_token_type (str): The subject token type based on the Oauth2.0 token exchange spec. Expected values include:: “urn:ietf:params:aws:token-type:aws4_request” token_url (Optional [str]): The STS endpoint URL. If not provided, will default to "https://sts.googleapis.com/v1/token". credential_source (Optional [Mapping]): The credential source dictionary used to provide instructions on how to retrieve external credential to be exchanged for Google access tokens. Either a credential source or an AWS security credentials supplier must be provided. Example credential_source for AWS credential:: { "environment_id": "aws1", "regional_cred_verification_url": "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15", "region_url": "http://169.254.169.254/latest/meta-data/placement/availability-zone", "url": "http://169.254.169.254/latest/meta-data/iam/security-credentials", imdsv2_session_token_url": "http://169.254.169.254/latest/api/token" } aws_security_credentials_supplier (Optional [AwsSecurityCredentialsSupplier]): Optional AWS security credentials supplier. This will be called to supply valid AWS security credentails which will then be exchanged for Google access tokens. Either an AWS security credentials supplier or a credential source must be provided. args (List): Optional positional arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method. kwargs (Mapping): Optional keyword arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method. Raises: google.auth.exceptions.RefreshError: If an error is encountered during access token retrieval logic. ValueError: For invalid parameters. .. note:: Typically one of the helper constructors :meth:`from_file` or :meth:`from_info` are used instead of calling the constructor directly. """ super(Credentials, self).__init__( audience=audience, subject_token_type=subject_token_type, token_url=token_url, credential_source=credential_source, args, kwargs ) if credential_source is None and aws_security_credentials_supplier is None: raise exceptions.InvalidValue( "A valid credential source or AWS security credentials supplier must be provided." ) if ( credential_source is not None and aws_security_credentials_supplier is not None ): raise exceptions.InvalidValue( "AWS credential cannot have both a credential source and an AWS security credentials supplier." ) if aws_security_credentials_supplier: self._aws_security_credentials_supplier = aws_security_credentials_supplier # The regional cred verification URL would normally be provided through the credential source. So set it to the default one here. self._cred_verification_url = ( _DEFAULT_AWS_REGIONAL_CREDENTIAL_VERIFICATION_URL ) else: environment_id = credential_source.get("environment_id") or "" self._aws_security_credentials_supplier = _DefaultAwsSecurityCredentialsSupplier( credential_source ) self._cred_verification_url = credential_source.get( "regional_cred_verification_url" ) # Get the environment ID, i.e. "aws1". Currently, only one version supported (1). matches = re.match(r"^(aws)([\d]+)$", environment_id) if matches: env_id, env_version = matches.groups() else: env_id, env_version = (None, None) if env_id != "aws" or self._cred_verification_url is None: raise exceptions.InvalidResource( "No valid AWS 'credential_source' provided" ) elif env_version is None or int(env_version) != 1: raise exceptions.InvalidValue( "aws version '{}' is not supported in the current build.".format( env_version ) ) self._target_resource = audience self._request_signer = None def retrieve_subject_token(self, request): """Retrieves the subject token using the credential_source object. The subject token is a serialized `AWS GetCallerIdentity signed request`_. The logic is summarized as: Retrieve the AWS region from the AWS_REGION or AWS_DEFAULT_REGION environment variable or from the AWS metadata server availability-zone if not found in the environment variable. Check AWS credentials in environment variables. If not found, retrieve from the AWS metadata server security-credentials endpoint. When retrieving AWS credentials from the metadata server security-credentials endpoint, the AWS role needs to be determined by calling the security-credentials endpoint without any argument. Then the credentials can be retrieved via: security-credentials/role_name Generate the signed request to AWS STS GetCallerIdentity action. Inject x-goog-cloud-target-resource into header and serialize the signed request. This will be the subject-token to pass to GCP STS. .. _AWS GetCallerIdentity signed request: https://cloud.google.com/iam/docs/access-resources-aws#exchange-token Args: request (google.auth.transport.Request): A callable used to make HTTP requests. Returns: str: The retrieved subject token. """ # Initialize the request signer if not yet initialized after determining # the current AWS region. if self._request_signer is None: self._region = self._aws_security_credentials_supplier.get_aws_region( self._supplier_context, request ) self._request_signer = RequestSigner(self._region) # Retrieve the AWS security credentials needed to generate the signed # request. aws_security_credentials = self._aws_security_credentials_supplier.get_aws_security_credentials( self._supplier_context, request ) # Generate the signed request to AWS STS GetCallerIdentity API. # Use the required regional endpoint. Otherwise, the request will fail. request_options = self._request_signer.get_request_options( aws_security_credentials, self._cred_verification_url.replace("{region}", self._region), "POST", ) # The GCP STS endpoint expects the headers to be formatted as: # [ # {key: 'x-amz-date', value: '...'}, # {key: 'Authorization', value: '...'}, # ... # ] # And then serialized as: # quote(json.dumps({ # url: '...', # method: 'POST', # headers: [{key: 'x-amz-date', value: '...'}, ...] # })) request_headers = request_options.get("headers") # The full, canonical resource name of the workload identity pool # provider, with or without the HTTPS prefix. # Including this header as part of the signature is recommended to # ensure data integrity. request_headers["x-goog-cloud-target-resource"] = self._target_resource # Serialize AWS signed request. aws_signed_req = {} aws_signed_req["url"] = request_options.get("url") aws_signed_req["method"] = request_options.get("method") aws_signed_req["headers"] = [] # Reformat header to GCP STS expected format. for key in request_headers.keys(): aws_signed_req["headers"].append( {"key": key, "value": request_headers[key]} ) return urllib.parse.quote( json.dumps(aws_signed_req, separators=(",", ":"), sort_keys=True) ) def _create_default_metrics_options(self): metrics_options = super(Credentials, self)._create_default_metrics_options() metrics_options["source"] = "aws" if self._has_custom_supplier(): metrics_options["source"] = "programmatic" return metrics_options def _has_custom_supplier(self): return self._credential_source is None def _constructor_args(self): args = super(Credentials, self)._constructor_args() # If a custom supplier was used, append it to the args dict. if self._has_custom_supplier(): args.update( { "aws_security_credentials_supplier": self._aws_security_credentials_supplier } ) return args @classmethod def from_info(cls, info, kwargs): """Creates an AWS Credentials instance from parsed external account info. Args: info (Mapping[str, str]): The AWS external account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.aws.Credentials: The constructed credentials. Raises: ValueError: For invalid parameters. """ aws_security_credentials_supplier = info.get( "aws_security_credentials_supplier" ) kwargs.update( {"aws_security_credentials_supplier": aws_security_credentials_supplier} ) return super(Credentials, cls).from_info(info, kwargs) @classmethod def from_file(cls, filename, kwargs): """Creates an AWS Credentials instance from an external account json file. Args: filename (str): The path to the AWS external account json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.aws.Credentials: The constructed credentials. """ return super(Credentials, cls).from_file(filename, **kwargs)	33.76 KB	2025-04-14 17:36
📁 compute_engine	-	2025-04-14 17:36
📄 credentials.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Interfaces for credentials.""" import abc from enum import Enum import os from google.auth import _helpers, environment_vars from google.auth import exceptions from google.auth import metrics from google.auth._credentials_base import _BaseCredentials from google.auth._refresh_worker import RefreshThreadManager DEFAULT_UNIVERSE_DOMAIN = "googleapis.com" class Credentials(_BaseCredentials): """Base class for all credentials. All credentials have a :attr:`token` that is used for authentication and may also optionally set an :attr:`expiry` to indicate when the token will no longer be valid. Most credentials will be :attr:`invalid` until :meth:`refresh` is called. Credentials can do this automatically before the first HTTP request in :meth:`before_request`. Although the token and expiration will change as the credentials are :meth:`refreshed <refresh>` and used, credentials should be considered immutable. Various credentials will accept configuration such as private keys, scopes, and other options. These options are not changeable after construction. Some classes will provide mechanisms to copy the credentials with modifications such as :meth:`ScopedCredentials.with_scopes`. """ def __init__(self): super(Credentials, self).__init__() self.expiry = None """Optional[datetime]: When the token expires and is no longer valid. If this is None, the token is assumed to never expire.""" self._quota_project_id = None """Optional[str]: Project to use for quota and billing purposes.""" self._trust_boundary = None """Optional[dict]: Cache of a trust boundary response which has a list of allowed regions and an encoded string representation of credentials trust boundary.""" self._universe_domain = DEFAULT_UNIVERSE_DOMAIN """Optional[str]: The universe domain value, default is googleapis.com """ self._use_non_blocking_refresh = False self._refresh_worker = RefreshThreadManager() @property def expired(self): """Checks if the credentials are expired. Note that credentials can be invalid but not expired because Credentials with :attr:`expiry` set to None is considered to never expire. .. deprecated:: v2.24.0 Prefer checking :attr:`token_state` instead. """ if not self.expiry: return False # Remove some threshold from expiry to err on the side of reporting # expiration early so that we avoid the 401-refresh-retry loop. skewed_expiry = self.expiry - _helpers.REFRESH_THRESHOLD return _helpers.utcnow() >= skewed_expiry @property def valid(self): """Checks the validity of the credentials. This is True if the credentials have a :attr:`token` and the token is not :attr:`expired`. .. deprecated:: v2.24.0 Prefer checking :attr:`token_state` instead. """ return self.token is not None and not self.expired @property def token_state(self): """ See `:obj:`TokenState` """ if self.token is None: return TokenState.INVALID # Credentials that can't expire are always treated as fresh. if self.expiry is None: return TokenState.FRESH expired = _helpers.utcnow() >= self.expiry if expired: return TokenState.INVALID is_stale = _helpers.utcnow() >= (self.expiry - _helpers.REFRESH_THRESHOLD) if is_stale: return TokenState.STALE return TokenState.FRESH @property def quota_project_id(self): """Project to use for quota and billing purposes.""" return self._quota_project_id @property def universe_domain(self): """The universe domain value.""" return self._universe_domain def get_cred_info(self): """The credential information JSON. The credential information will be added to auth related error messages by client library. Returns: Mapping[str, str]: The credential information JSON. """ return None @abc.abstractmethod def refresh(self, request): """Refreshes the access token. Args: request (google.auth.transport.Request): The object used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If the credentials could not be refreshed. """ # pylint: disable=missing-raises-doc # (pylint doesn't recognize that this is abstract) raise NotImplementedError("Refresh must be implemented") def _metric_header_for_usage(self): """The x-goog-api-client header for token usage metric. This header will be added to the API service requests in before_request method. For example, "cred-type/sa-jwt" means service account self signed jwt access token is used in the API service request authorization header. Children credentials classes need to override this method to provide the header value, if the token usage metric is needed. Returns: str: The x-goog-api-client header value. """ return None def apply(self, headers, token=None): """Apply the token to the authentication header. Args: headers (Mapping): The HTTP request headers. token (Optional[str]): If specified, overrides the current access token. """ self._apply(headers, token=token) """Trust boundary value will be a cached value from global lookup. The response of trust boundary will be a list of regions and a hex encoded representation. An example of global lookup response: { "locations": [ "us-central1", "us-east1", "europe-west1", "asia-east1" ] "encoded_locations": "0xA30" } """ if self._trust_boundary is not None: headers["x-allowed-locations"] = self._trust_boundary["encoded_locations"] if self.quota_project_id: headers["x-goog-user-project"] = self.quota_project_id def _blocking_refresh(self, request): if not self.valid: self.refresh(request) def _non_blocking_refresh(self, request): use_blocking_refresh_fallback = False if self.token_state == TokenState.STALE: use_blocking_refresh_fallback = not self._refresh_worker.start_refresh( self, request ) if self.token_state == TokenState.INVALID or use_blocking_refresh_fallback: self.refresh(request) # If the blocking refresh succeeds then we can clear the error info # on the background refresh worker, and perform refreshes in a # background thread. self._refresh_worker.clear_error() def before_request(self, request, method, url, headers): """Performs credential-specific before request logic. Refreshes the credentials if necessary, then calls :meth:`apply` to apply the token to the authentication header. Args: request (google.auth.transport.Request): The object used to make HTTP requests. method (str): The request's HTTP method or the RPC method being invoked. url (str): The request's URI or the RPC service's URI. headers (Mapping): The request's headers. """ # pylint: disable=unused-argument # (Subclasses may use these arguments to ascertain information about # the http request.) if self._use_non_blocking_refresh: self._non_blocking_refresh(request) else: self._blocking_refresh(request) metrics.add_metric_header(headers, self._metric_header_for_usage()) self.apply(headers) def with_non_blocking_refresh(self): self._use_non_blocking_refresh = True class CredentialsWithQuotaProject(Credentials): """Abstract base for credentials supporting ``with_quota_project`` factory""" def with_quota_project(self, quota_project_id): """Returns a copy of these credentials with a modified quota project. Args: quota_project_id (str): The project to use for quota and billing purposes Returns: google.auth.credentials.Credentials: A new credentials instance. """ raise NotImplementedError("This credential does not support quota project.") def with_quota_project_from_environment(self): quota_from_env = os.environ.get(environment_vars.GOOGLE_CLOUD_QUOTA_PROJECT) if quota_from_env: return self.with_quota_project(quota_from_env) return self class CredentialsWithTokenUri(Credentials): """Abstract base for credentials supporting ``with_token_uri`` factory""" def with_token_uri(self, token_uri): """Returns a copy of these credentials with a modified token uri. Args: token_uri (str): The uri to use for fetching/exchanging tokens Returns: google.auth.credentials.Credentials: A new credentials instance. """ raise NotImplementedError("This credential does not use token uri.") class CredentialsWithUniverseDomain(Credentials): """Abstract base for credentials supporting ``with_universe_domain`` factory""" def with_universe_domain(self, universe_domain): """Returns a copy of these credentials with a modified universe domain. Args: universe_domain (str): The universe domain to use Returns: google.auth.credentials.Credentials: A new credentials instance. """ raise NotImplementedError( "This credential does not support with_universe_domain." ) class AnonymousCredentials(Credentials): """Credentials that do not provide any authentication information. These are useful in the case of services that support anonymous access or local service emulators that do not use credentials. """ @property def expired(self): """Returns `False`, anonymous credentials never expire.""" return False @property def valid(self): """Returns `True`, anonymous credentials are always valid.""" return True def refresh(self, request): """Raises :class:``InvalidOperation``, anonymous credentials cannot be refreshed.""" raise exceptions.InvalidOperation("Anonymous credentials cannot be refreshed.") def apply(self, headers, token=None): """Anonymous credentials do nothing to the request. The optional ``token`` argument is not supported. Raises: google.auth.exceptions.InvalidValue: If a token was specified. """ if token is not None: raise exceptions.InvalidValue("Anonymous credentials don't support tokens.") def before_request(self, request, method, url, headers): """Anonymous credentials do nothing to the request.""" class ReadOnlyScoped(metaclass=abc.ABCMeta): """Interface for credentials whose scopes can be queried. OAuth 2.0-based credentials allow limiting access using scopes as described in `RFC6749 Section 3.3`_. If a credential class implements this interface then the credentials either use scopes in their implementation. Some credentials require scopes in order to obtain a token. You can check if scoping is necessary with :attr:`requires_scopes`:: if credentials.requires_scopes: # Scoping is required. credentials = credentials.with_scopes(scopes=['one', 'two']) Credentials that require scopes must either be constructed with scopes:: credentials = SomeScopedCredentials(scopes=['one', 'two']) Or must copy an existing instance using :meth:`with_scopes`:: scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) Some credentials have scopes but do not allow or require scopes to be set, these credentials can be used as-is. .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 """ def __init__(self): super(ReadOnlyScoped, self).__init__() self._scopes = None self._default_scopes = None @property def scopes(self): """Sequence[str]: the credentials' current set of scopes.""" return self._scopes @property def default_scopes(self): """Sequence[str]: the credentials' current set of default scopes.""" return self._default_scopes @abc.abstractproperty def requires_scopes(self): """True if these credentials require scopes to obtain an access token. """ return False def has_scopes(self, scopes): """Checks if the credentials have the given scopes. .. warning: This method is not guaranteed to be accurate if the credentials are :attr:`~Credentials.invalid`. Args: scopes (Sequence[str]): The list of scopes to check. Returns: bool: True if the credentials have the given scopes. """ credential_scopes = ( self._scopes if self._scopes is not None else self._default_scopes ) return set(scopes).issubset(set(credential_scopes or [])) class Scoped(ReadOnlyScoped): """Interface for credentials whose scopes can be replaced while copying. OAuth 2.0-based credentials allow limiting access using scopes as described in `RFC6749 Section 3.3`_. If a credential class implements this interface then the credentials either use scopes in their implementation. Some credentials require scopes in order to obtain a token. You can check if scoping is necessary with :attr:`requires_scopes`:: if credentials.requires_scopes: # Scoping is required. credentials = credentials.create_scoped(['one', 'two']) Credentials that require scopes must either be constructed with scopes:: credentials = SomeScopedCredentials(scopes=['one', 'two']) Or must copy an existing instance using :meth:`with_scopes`:: scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) Some credentials have scopes but do not allow or require scopes to be set, these credentials can be used as-is. .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 """ @abc.abstractmethod def with_scopes(self, scopes, default_scopes=None): """Create a copy of these credentials with the specified scopes. Args: scopes (Sequence[str]): The list of scopes to attach to the current credentials. Raises: NotImplementedError: If the credentials' scopes can not be changed. This can be avoided by checking :attr:`requires_scopes` before calling this method. """ raise NotImplementedError("This class does not require scoping.") def with_scopes_if_required(credentials, scopes, default_scopes=None): """Creates a copy of the credentials with scopes if scoping is required. This helper function is useful when you do not know (or care to know) the specific type of credentials you are using (such as when you use :func:`google.auth.default`). This function will call :meth:`Scoped.with_scopes` if the credentials are scoped credentials and if the credentials require scoping. Otherwise, it will return the credentials as-is. Args: credentials (google.auth.credentials.Credentials): The credentials to scope if necessary. scopes (Sequence[str]): The list of scopes to use. default_scopes (Sequence[str]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. Returns: google.auth.credentials.Credentials: Either a new set of scoped credentials, or the passed in credentials instance if no scoping was required. """ if isinstance(credentials, Scoped) and credentials.requires_scopes: return credentials.with_scopes(scopes, default_scopes=default_scopes) else: return credentials class Signing(metaclass=abc.ABCMeta): """Interface for credentials that can cryptographically sign messages.""" @abc.abstractmethod def sign_bytes(self, message): """Signs the given message. Args: message (bytes): The message to sign. Returns: bytes: The message's cryptographic signature. """ # pylint: disable=missing-raises-doc,redundant-returns-doc # (pylint doesn't recognize that this is abstract) raise NotImplementedError("Sign bytes must be implemented.") @abc.abstractproperty def signer_email(self): """Optional[str]: An email address that identifies the signer.""" # pylint: disable=missing-raises-doc # (pylint doesn't recognize that this is abstract) raise NotImplementedError("Signer email must be implemented.") @abc.abstractproperty def signer(self): """google.auth.crypt.Signer: The signer used to sign bytes.""" # pylint: disable=missing-raises-doc # (pylint doesn't recognize that this is abstract) raise NotImplementedError("Signer must be implemented.") class TokenState(Enum): """ Tracks the state of a token. FRESH: The token is valid. It is not expired or close to expired, or the token has no expiry. STALE: The token is close to expired, and should be refreshed. The token can be used normally. INVALID: The token is expired or invalid. The token cannot be used for a normal operation. """ FRESH = 1 STALE = 2 INVALID = 3	18.26 KB	2025-04-14 17:36
📁 crypt	-	2025-06-25 08:26
📄 downscoped.py # Copyright 2021 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Downscoping with Credential Access Boundaries This module provides the ability to downscope credentials using `Downscoping with Credential Access Boundaries`_. This is useful to restrict the Identity and Access Management (IAM) permissions that a short-lived credential can use. To downscope permissions of a source credential, a Credential Access Boundary that specifies which resources the new credential can access, as well as an upper bound on the permissions that are available on each resource, has to be defined. A downscoped credential can then be instantiated using the source credential and the Credential Access Boundary. The common pattern of usage is to have a token broker with elevated access generate these downscoped credentials from higher access source credentials and pass the downscoped short-lived access tokens to a token consumer via some secure authenticated channel for limited access to Google Cloud Storage resources. For example, a token broker can be set up on a server in a private network. Various workloads (token consumers) in the same network will send authenticated requests to that broker for downscoped tokens to access or modify specific google cloud storage buckets. The broker will instantiate downscoped credentials instances that can be used to generate short lived downscoped access tokens that can be passed to the token consumer. These downscoped access tokens can be injected by the consumer into google.oauth2.Credentials and used to initialize a storage client instance to access Google Cloud Storage resources with restricted access. Note: Only Cloud Storage supports Credential Access Boundaries. Other Google Cloud services do not support this feature. .. _Downscoping with Credential Access Boundaries: https://cloud.google.com/iam/docs/downscoping-short-lived-credentials """ import datetime from google.auth import _helpers from google.auth import credentials from google.auth import exceptions from google.oauth2 import sts # The maximum number of access boundary rules a Credential Access Boundary can # contain. _MAX_ACCESS_BOUNDARY_RULES_COUNT = 10 # The token exchange grant_type used for exchanging credentials. _STS_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" # The token exchange requested_token_type. This is always an access_token. _STS_REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" # The STS token URL used to exchanged a short lived access token for a downscoped one. _STS_TOKEN_URL_PATTERN = "https://sts.{}/v1/token" # The subject token type to use when exchanging a short lived access token for a # downscoped token. _STS_SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" class CredentialAccessBoundary(object): """Defines a Credential Access Boundary which contains a list of access boundary rules. Each rule contains information on the resource that the rule applies to, the upper bound of the permissions that are available on that resource and an optional condition to further restrict permissions. """ def __init__(self, rules=[]): """Instantiates a Credential Access Boundary. A Credential Access Boundary can contain up to 10 access boundary rules. Args: rules (Sequence[google.auth.downscoped.AccessBoundaryRule]): The list of access boundary rules limiting the access that a downscoped credential will have. Raises: InvalidType: If any of the rules are not a valid type. InvalidValue: If the provided rules exceed the maximum allowed. """ self.rules = rules @property def rules(self): """Returns the list of access boundary rules defined on the Credential Access Boundary. Returns: Tuple[google.auth.downscoped.AccessBoundaryRule, ...]: The list of access boundary rules defined on the Credential Access Boundary. These are returned as an immutable tuple to prevent modification. """ return tuple(self._rules) @rules.setter def rules(self, value): """Updates the current rules on the Credential Access Boundary. This will overwrite the existing set of rules. Args: value (Sequence[google.auth.downscoped.AccessBoundaryRule]): The list of access boundary rules limiting the access that a downscoped credential will have. Raises: InvalidType: If any of the rules are not a valid type. InvalidValue: If the provided rules exceed the maximum allowed. """ if len(value) > _MAX_ACCESS_BOUNDARY_RULES_COUNT: raise exceptions.InvalidValue( "Credential access boundary rules can have a maximum of {} rules.".format( _MAX_ACCESS_BOUNDARY_RULES_COUNT ) ) for access_boundary_rule in value: if not isinstance(access_boundary_rule, AccessBoundaryRule): raise exceptions.InvalidType( "List of rules provided do not contain a valid 'google.auth.downscoped.AccessBoundaryRule'." ) # Make a copy of the original list. self._rules = list(value) def add_rule(self, rule): """Adds a single access boundary rule to the existing rules. Args: rule (google.auth.downscoped.AccessBoundaryRule): The access boundary rule, limiting the access that a downscoped credential will have, to be added to the existing rules. Raises: InvalidType: If any of the rules are not a valid type. InvalidValue: If the provided rules exceed the maximum allowed. """ if len(self.rules) == _MAX_ACCESS_BOUNDARY_RULES_COUNT: raise exceptions.InvalidValue( "Credential access boundary rules can have a maximum of {} rules.".format( _MAX_ACCESS_BOUNDARY_RULES_COUNT ) ) if not isinstance(rule, AccessBoundaryRule): raise exceptions.InvalidType( "The provided rule does not contain a valid 'google.auth.downscoped.AccessBoundaryRule'." ) self._rules.append(rule) def to_json(self): """Generates the dictionary representation of the Credential Access Boundary. This uses the format expected by the Security Token Service API as documented in `Defining a Credential Access Boundary`_. .. _Defining a Credential Access Boundary: https://cloud.google.com/iam/docs/downscoping-short-lived-credentials#define-boundary Returns: Mapping: Credential Access Boundary Rule represented in a dictionary object. """ rules = [] for access_boundary_rule in self.rules: rules.append(access_boundary_rule.to_json()) return {"accessBoundary": {"accessBoundaryRules": rules}} class AccessBoundaryRule(object): """Defines an access boundary rule which contains information on the resource that the rule applies to, the upper bound of the permissions that are available on that resource and an optional condition to further restrict permissions. """ def __init__( self, available_resource, available_permissions, availability_condition=None ): """Instantiates a single access boundary rule. Args: available_resource (str): The full resource name of the Cloud Storage bucket that the rule applies to. Use the format "//storage.googleapis.com/projects/_/buckets/bucket-name". available_permissions (Sequence[str]): A list defining the upper bound that the downscoped token will have on the available permissions for the resource. Each value is the identifier for an IAM predefined role or custom role, with the prefix "inRole:". For example: "inRole:roles/storage.objectViewer". Only the permissions in these roles will be available. availability_condition (Optional[google.auth.downscoped.AvailabilityCondition]): Optional condition that restricts the availability of permissions to specific Cloud Storage objects. Raises: InvalidType: If any of the parameters are not of the expected types. InvalidValue: If any of the parameters are not of the expected values. """ self.available_resource = available_resource self.available_permissions = available_permissions self.availability_condition = availability_condition @property def available_resource(self): """Returns the current available resource. Returns: str: The current available resource. """ return self._available_resource @available_resource.setter def available_resource(self, value): """Updates the current available resource. Args: value (str): The updated value of the available resource. Raises: google.auth.exceptions.InvalidType: If the value is not a string. """ if not isinstance(value, str): raise exceptions.InvalidType( "The provided available_resource is not a string." ) self._available_resource = value @property def available_permissions(self): """Returns the current available permissions. Returns: Tuple[str, ...]: The current available permissions. These are returned as an immutable tuple to prevent modification. """ return tuple(self._available_permissions) @available_permissions.setter def available_permissions(self, value): """Updates the current available permissions. Args: value (Sequence[str]): The updated value of the available permissions. Raises: InvalidType: If the value is not a list of strings. InvalidValue: If the value is not valid. """ for available_permission in value: if not isinstance(available_permission, str): raise exceptions.InvalidType( "Provided available_permissions are not a list of strings." ) if available_permission.find("inRole:") != 0: raise exceptions.InvalidValue( "available_permissions must be prefixed with 'inRole:'." ) # Make a copy of the original list. self._available_permissions = list(value) @property def availability_condition(self): """Returns the current availability condition. Returns: Optional[google.auth.downscoped.AvailabilityCondition]: The current availability condition. """ return self._availability_condition @availability_condition.setter def availability_condition(self, value): """Updates the current availability condition. Args: value (Optional[google.auth.downscoped.AvailabilityCondition]): The updated value of the availability condition. Raises: google.auth.exceptions.InvalidType: If the value is not of type google.auth.downscoped.AvailabilityCondition or None. """ if not isinstance(value, AvailabilityCondition) and value is not None: raise exceptions.InvalidType( "The provided availability_condition is not a 'google.auth.downscoped.AvailabilityCondition' or None." ) self._availability_condition = value def to_json(self): """Generates the dictionary representation of the access boundary rule. This uses the format expected by the Security Token Service API as documented in `Defining a Credential Access Boundary`_. .. _Defining a Credential Access Boundary: https://cloud.google.com/iam/docs/downscoping-short-lived-credentials#define-boundary Returns: Mapping: The access boundary rule represented in a dictionary object. """ json = { "availablePermissions": list(self.available_permissions), "availableResource": self.available_resource, } if self.availability_condition: json["availabilityCondition"] = self.availability_condition.to_json() return json class AvailabilityCondition(object): """An optional condition that can be used as part of a Credential Access Boundary to further restrict permissions.""" def __init__(self, expression, title=None, description=None): """Instantiates an availability condition using the provided expression and optional title or description. Args: expression (str): A condition expression that specifies the Cloud Storage objects where permissions are available. For example, this expression makes permissions available for objects whose name starts with "customer-a": "resource.name.startsWith('projects/_/buckets/example-bucket/objects/customer-a')" title (Optional[str]): An optional short string that identifies the purpose of the condition. description (Optional[str]): Optional details about the purpose of the condition. Raises: InvalidType: If any of the parameters are not of the expected types. InvalidValue: If any of the parameters are not of the expected values. """ self.expression = expression self.title = title self.description = description @property def expression(self): """Returns the current condition expression. Returns: str: The current conditon expression. """ return self._expression @expression.setter def expression(self, value): """Updates the current condition expression. Args: value (str): The updated value of the condition expression. Raises: google.auth.exceptions.InvalidType: If the value is not of type string. """ if not isinstance(value, str): raise exceptions.InvalidType("The provided expression is not a string.") self._expression = value @property def title(self): """Returns the current title. Returns: Optional[str]: The current title. """ return self._title @title.setter def title(self, value): """Updates the current title. Args: value (Optional[str]): The updated value of the title. Raises: google.auth.exceptions.InvalidType: If the value is not of type string or None. """ if not isinstance(value, str) and value is not None: raise exceptions.InvalidType("The provided title is not a string or None.") self._title = value @property def description(self): """Returns the current description. Returns: Optional[str]: The current description. """ return self._description @description.setter def description(self, value): """Updates the current description. Args: value (Optional[str]): The updated value of the description. Raises: google.auth.exceptions.InvalidType: If the value is not of type string or None. """ if not isinstance(value, str) and value is not None: raise exceptions.InvalidType( "The provided description is not a string or None." ) self._description = value def to_json(self): """Generates the dictionary representation of the availability condition. This uses the format expected by the Security Token Service API as documented in `Defining a Credential Access Boundary`_. .. _Defining a Credential Access Boundary: https://cloud.google.com/iam/docs/downscoping-short-lived-credentials#define-boundary Returns: Mapping[str, str]: The availability condition represented in a dictionary object. """ json = {"expression": self.expression} if self.title: json["title"] = self.title if self.description: json["description"] = self.description return json class Credentials(credentials.CredentialsWithQuotaProject): """Defines a set of Google credentials that are downscoped from an existing set of Google OAuth2 credentials. This is useful to restrict the Identity and Access Management (IAM) permissions that a short-lived credential can use. The common pattern of usage is to have a token broker with elevated access generate these downscoped credentials from higher access source credentials and pass the downscoped short-lived access tokens to a token consumer via some secure authenticated channel for limited access to Google Cloud Storage resources. """ def __init__( self, source_credentials, credential_access_boundary, quota_project_id=None, universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN, ): """Instantiates a downscoped credentials object using the provided source credentials and credential access boundary rules. To downscope permissions of a source credential, a Credential Access Boundary that specifies which resources the new credential can access, as well as an upper bound on the permissions that are available on each resource, has to be defined. A downscoped credential can then be instantiated using the source credential and the Credential Access Boundary. Args: source_credentials (google.auth.credentials.Credentials): The source credentials to be downscoped based on the provided Credential Access Boundary rules. credential_access_boundary (google.auth.downscoped.CredentialAccessBoundary): The Credential Access Boundary which contains a list of access boundary rules. Each rule contains information on the resource that the rule applies to, the upper bound of the permissions that are available on that resource and an optional condition to further restrict permissions. quota_project_id (Optional[str]): The optional quota project ID. universe_domain (Optional[str]): The universe domain value, default is googleapis.com Raises: google.auth.exceptions.RefreshError: If the source credentials return an error on token refresh. google.auth.exceptions.OAuthError: If the STS token exchange endpoint returned an error during downscoped token generation. """ super(Credentials, self).__init__() self._source_credentials = source_credentials self._credential_access_boundary = credential_access_boundary self._quota_project_id = quota_project_id self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN self._sts_client = sts.Client( _STS_TOKEN_URL_PATTERN.format(self.universe_domain) ) @_helpers.copy_docstring(credentials.Credentials) def refresh(self, request): # Generate an access token from the source credentials. self._source_credentials.refresh(request) now = _helpers.utcnow() # Exchange the access token for a downscoped access token. response_data = self._sts_client.exchange_token( request=request, grant_type=_STS_GRANT_TYPE, subject_token=self._source_credentials.token, subject_token_type=_STS_SUBJECT_TOKEN_TYPE, requested_token_type=_STS_REQUESTED_TOKEN_TYPE, additional_options=self._credential_access_boundary.to_json(), ) self.token = response_data.get("access_token") # For downscoping CAB flow, the STS endpoint may not return the expiration # field for some flows. The generated downscoped token should always have # the same expiration time as the source credentials. When no expires_in # field is returned in the response, we can just get the expiration time # from the source credentials. if response_data.get("expires_in"): lifetime = datetime.timedelta(seconds=response_data.get("expires_in")) self.expiry = now + lifetime else: self.expiry = self._source_credentials.expiry @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): return self.__class__( self._source_credentials, self._credential_access_boundary, quota_project_id=quota_project_id, )	21.28 KB	2025-04-14 17:36
📄 environment_vars.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Environment variables used by :mod:`google.auth`.""" PROJECT = "GOOGLE_CLOUD_PROJECT" """Environment variable defining default project. This used by :func:`google.auth.default` to explicitly set a project ID. This environment variable is also used by the Google Cloud Python Library. """ LEGACY_PROJECT = "GCLOUD_PROJECT" """Previously used environment variable defining the default project. This environment variable is used instead of the current one in some situations (such as Google App Engine). """ GOOGLE_CLOUD_QUOTA_PROJECT = "GOOGLE_CLOUD_QUOTA_PROJECT" """Environment variable defining the project to be used for quota and billing.""" CREDENTIALS = "GOOGLE_APPLICATION_CREDENTIALS" """Environment variable defining the location of Google application default credentials.""" # 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 files.""" # These two variables allow for customization of the addresses used when # contacting the GCE metadata service. GCE_METADATA_HOST = "GCE_METADATA_HOST" """Environment variable providing an alternate hostname or host:port to be used for GCE metadata requests. This environment variable was originally named GCE_METADATA_ROOT. The system will check this environemnt variable first; should there be no value present, the system will fall back to the old variable. """ GCE_METADATA_ROOT = "GCE_METADATA_ROOT" """Old environment variable for GCE_METADATA_HOST.""" GCE_METADATA_IP = "GCE_METADATA_IP" """Environment variable providing an alternate ip:port to be used for ip-only GCE metadata requests.""" GOOGLE_API_USE_CLIENT_CERTIFICATE = "GOOGLE_API_USE_CLIENT_CERTIFICATE" """Environment variable controlling whether to use client certificate or not. The default value is false. Users have to explicitly set this value to true in order to use client certificate to establish a mutual TLS channel.""" LEGACY_APPENGINE_RUNTIME = "APPENGINE_RUNTIME" """Gen1 environment variable defining the App Engine Runtime. Used to distinguish between GAE gen1 and GAE gen2+. """ # AWS environment variables used with AWS workload identity pools to retrieve # AWS security credentials and the AWS region needed to create a serialized # signed requests to the AWS STS GetCalledIdentity API that can be exchanged # for a Google access tokens via the GCP STS endpoint. # When not available the AWS metadata server is used to retrieve these values. AWS_ACCESS_KEY_ID = "AWS_ACCESS_KEY_ID" AWS_SECRET_ACCESS_KEY = "AWS_SECRET_ACCESS_KEY" AWS_SESSION_TOKEN = "AWS_SESSION_TOKEN" AWS_REGION = "AWS_REGION" AWS_DEFAULT_REGION = "AWS_DEFAULT_REGION"	3.22 KB	2025-04-14 17:36
📄 exceptions.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Exceptions used in the google.auth package.""" class GoogleAuthError(Exception): """Base class for all google.auth errors.""" def __init__(self, args, kwargs): super(GoogleAuthError, self).__init__(args) retryable = kwargs.get("retryable", False) self._retryable = retryable @property def retryable(self): return self._retryable class TransportError(GoogleAuthError): """Used to indicate an error occurred during an HTTP request.""" class RefreshError(GoogleAuthError): """Used to indicate that an refreshing the credentials' access token failed.""" class UserAccessTokenError(GoogleAuthError): """Used to indicate ``gcloud auth print-access-token`` command failed.""" class DefaultCredentialsError(GoogleAuthError): """Used to indicate that acquiring default credentials failed.""" class MutualTLSChannelError(GoogleAuthError): """Used to indicate that mutual TLS channel creation is failed, or mutual TLS channel credentials is missing or invalid.""" class ClientCertError(GoogleAuthError): """Used to indicate that client certificate is missing or invalid.""" @property def retryable(self): return False class OAuthError(GoogleAuthError): """Used to indicate an error occurred during an OAuth related HTTP request.""" class ReauthFailError(RefreshError): """An exception for when reauth failed.""" def __init__(self, message=None, kwargs): super(ReauthFailError, self).__init__( "Reauthentication failed. {0}".format(message), kwargs ) class ReauthSamlChallengeFailError(ReauthFailError): """An exception for SAML reauth challenge failures.""" class MalformedError(DefaultCredentialsError, ValueError): """An exception for malformed data.""" class InvalidResource(DefaultCredentialsError, ValueError): """An exception for URL error.""" class InvalidOperation(DefaultCredentialsError, ValueError): """An exception for invalid operation.""" class InvalidValue(DefaultCredentialsError, ValueError): """Used to wrap general ValueError of python.""" class InvalidType(DefaultCredentialsError, TypeError): """Used to wrap general TypeError of python.""" class OSError(DefaultCredentialsError, EnvironmentError): """Used to wrap EnvironmentError(OSError after python3.3).""" class TimeoutError(GoogleAuthError): """Used to indicate a timeout error occurred during an HTTP request.""" class ResponseError(GoogleAuthError): """Used to indicate an error occurred when reading an HTTP response."""	3.12 KB	2025-04-14 17:36
📄 external_account.py # Copyright 2020 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """External Account Credentials. This module provides credentials that exchange workload identity pool external credentials for Google access tokens. This facilitates accessing Google Cloud Platform resources from on-prem and non-Google Cloud platforms (e.g. AWS, Microsoft Azure, OIDC identity providers), using native credentials retrieved from the current environment without the need to copy, save and manage long-lived service account credentials. Specifically, this is intended to use access tokens acquired using the GCP STS token exchange endpoint following the `OAuth 2.0 Token Exchange`_ spec. .. _OAuth 2.0 Token Exchange: https://tools.ietf.org/html/rfc8693 """ import abc import copy from dataclasses import dataclass import datetime import functools import io import json import re from google.auth import _helpers from google.auth import credentials from google.auth import exceptions from google.auth import impersonated_credentials from google.auth import metrics from google.oauth2 import sts from google.oauth2 import utils # External account JSON type identifier. _EXTERNAL_ACCOUNT_JSON_TYPE = "external_account" # The token exchange grant_type used for exchanging credentials. _STS_GRANT_TYPE = "urn:ietf:params:oauth:grant-type:token-exchange" # The token exchange requested_token_type. This is always an access_token. _STS_REQUESTED_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" # Cloud resource manager URL used to retrieve project information. _CLOUD_RESOURCE_MANAGER = "https://cloudresourcemanager.googleapis.com/v1/projects/" # Default Google sts token url. _DEFAULT_TOKEN_URL = "https://sts.{universe_domain}/v1/token" @dataclass class SupplierContext: """A context class that contains information about the requested third party credential that is passed to AWS security credential and subject token suppliers. Attributes: subject_token_type (str): The requested subject token type based on the Oauth2.0 token exchange spec. Expected values include:: “urn:ietf:params:oauth:token-type:jwt” “urn:ietf:params:oauth:token-type:id-token” “urn:ietf:params:oauth:token-type:saml2” “urn:ietf:params:aws:token-type:aws4_request” audience (str): The requested audience for the subject token. """ subject_token_type: str audience: str class Credentials( credentials.Scoped, credentials.CredentialsWithQuotaProject, credentials.CredentialsWithTokenUri, metaclass=abc.ABCMeta, ): """Base class for all external account credentials. This is used to instantiate Credentials for exchanging external account credentials for Google access token and authorizing requests to Google APIs. The base class implements the common logic for exchanging external account credentials for Google access tokens. """ def __init__( self, audience, subject_token_type, token_url, credential_source, service_account_impersonation_url=None, service_account_impersonation_options=None, client_id=None, client_secret=None, token_info_url=None, quota_project_id=None, scopes=None, default_scopes=None, workforce_pool_user_project=None, universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN, trust_boundary=None, ): """Instantiates an external account credentials object. Args: audience (str): The STS audience field. subject_token_type (str): The subject token type based on the Oauth2.0 token exchange spec. Expected values include:: “urn:ietf:params:oauth:token-type:jwt” “urn:ietf:params:oauth:token-type:id-token” “urn:ietf:params:oauth:token-type:saml2” “urn:ietf:params:aws:token-type:aws4_request” token_url (str): The STS endpoint URL. credential_source (Mapping): The credential source dictionary. service_account_impersonation_url (Optional[str]): The optional service account impersonation generateAccessToken URL. client_id (Optional[str]): The optional client ID. client_secret (Optional[str]): The optional client secret. token_info_url (str): The optional STS endpoint URL for token introspection. quota_project_id (Optional[str]): The optional quota project ID. scopes (Optional[Sequence[str]]): Optional scopes to request during the authorization grant. default_scopes (Optional[Sequence[str]]): Default scopes passed by a Google client library. Use 'scopes' for user-defined scopes. workforce_pool_user_project (Optona[str]): The optional workforce pool user project number when the credential corresponds to a workforce pool and not a workload identity pool. The underlying principal must still have serviceusage.services.use IAM permission to use the project for billing/quota. universe_domain (str): The universe domain. The default universe domain is googleapis.com. trust_boundary (str): String representation of trust boundary meta. Raises: google.auth.exceptions.RefreshError: If the generateAccessToken endpoint returned an error. """ super(Credentials, self).__init__() self._audience = audience self._subject_token_type = subject_token_type self._universe_domain = universe_domain self._token_url = token_url if self._token_url == _DEFAULT_TOKEN_URL: self._token_url = self._token_url.replace( "{universe_domain}", self._universe_domain ) self._token_info_url = token_info_url self._credential_source = credential_source self._service_account_impersonation_url = service_account_impersonation_url self._service_account_impersonation_options = ( service_account_impersonation_options or {} ) self._client_id = client_id self._client_secret = client_secret self._quota_project_id = quota_project_id self._scopes = scopes self._default_scopes = default_scopes self._workforce_pool_user_project = workforce_pool_user_project self._trust_boundary = { "locations": [], "encoded_locations": "0x0", } # expose a placeholder trust boundary value. if self._client_id: self._client_auth = utils.ClientAuthentication( utils.ClientAuthType.basic, self._client_id, self._client_secret ) else: self._client_auth = None self._sts_client = sts.Client(self._token_url, self._client_auth) self._metrics_options = self._create_default_metrics_options() self._impersonated_credentials = None self._project_id = None self._supplier_context = SupplierContext( self._subject_token_type, self._audience ) self._cred_file_path = None if not self.is_workforce_pool and self._workforce_pool_user_project: # Workload identity pools do not support workforce pool user projects. raise exceptions.InvalidValue( "workforce_pool_user_project should not be set for non-workforce pool " "credentials" ) @property def info(self): """Generates the dictionary representation of the current credentials. Returns: Mapping: The dictionary representation of the credentials. This is the reverse of "from_info" defined on the subclasses of this class. It is useful for serializing the current credentials so it can deserialized later. """ config_info = self._constructor_args() config_info.update( type=_EXTERNAL_ACCOUNT_JSON_TYPE, service_account_impersonation=config_info.pop( "service_account_impersonation_options", None ), ) config_info.pop("scopes", None) config_info.pop("default_scopes", None) return {key: value for key, value in config_info.items() if value is not None} def _constructor_args(self): args = { "audience": self._audience, "subject_token_type": self._subject_token_type, "token_url": self._token_url, "token_info_url": self._token_info_url, "service_account_impersonation_url": self._service_account_impersonation_url, "service_account_impersonation_options": copy.deepcopy( self._service_account_impersonation_options ) or None, "credential_source": copy.deepcopy(self._credential_source), "quota_project_id": self._quota_project_id, "client_id": self._client_id, "client_secret": self._client_secret, "workforce_pool_user_project": self._workforce_pool_user_project, "scopes": self._scopes, "default_scopes": self._default_scopes, "universe_domain": self._universe_domain, } if not self.is_workforce_pool: args.pop("workforce_pool_user_project") return args @property def service_account_email(self): """Returns the service account email if service account impersonation is used. Returns: Optional[str]: The service account email if impersonation is used. Otherwise None is returned. """ if self._service_account_impersonation_url: # Parse email from URL. The formal looks as follows: # https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/name@project-id.iam.gserviceaccount.com:generateAccessToken url = self._service_account_impersonation_url start_index = url.rfind("/") end_index = url.find(":generateAccessToken") if start_index != -1 and end_index != -1 and start_index < end_index: start_index = start_index + 1 return url[start_index:end_index] return None @property def is_user(self): """Returns whether the credentials represent a user (True) or workload (False). Workloads behave similarly to service accounts. Currently workloads will use service account impersonation but will eventually not require impersonation. As a result, this property is more reliable than the service account email property in determining if the credentials represent a user or workload. Returns: bool: True if the credentials represent a user. False if they represent a workload. """ # If service account impersonation is used, the credentials will always represent a # service account. if self._service_account_impersonation_url: return False return self.is_workforce_pool @property def is_workforce_pool(self): """Returns whether the credentials represent a workforce pool (True) or workload (False) based on the credentials' audience. This will also return True for impersonated workforce pool credentials. Returns: bool: True if the credentials represent a workforce pool. False if they represent a workload. """ # Workforce pools representing users have the following audience format: # //iam.googleapis.com/locations/$location/workforcePools/$poolId/providers/$providerId p = re.compile(r"//iam\.googleapis\.com/locations/[^/]+/workforcePools/") return p.match(self._audience or "") is not None @property def requires_scopes(self): """Checks if the credentials requires scopes. Returns: bool: True if there are no scopes set otherwise False. """ return not self._scopes and not self._default_scopes @property def project_number(self): """Optional[str]: The project number corresponding to the workload identity pool.""" # STS audience pattern: # //iam.googleapis.com/projects/$PROJECT_NUMBER/locations/... components = self._audience.split("/") try: project_index = components.index("projects") if project_index + 1 < len(components): return components[project_index + 1] or None except ValueError: return None @property def token_info_url(self): """Optional[str]: The STS token introspection endpoint.""" return self._token_info_url @_helpers.copy_docstring(credentials.Credentials) def get_cred_info(self): if self._cred_file_path: cred_info_json = { "credential_source": self._cred_file_path, "credential_type": "external account credentials", } if self.service_account_email: cred_info_json["principal"] = self.service_account_email return cred_info_json return None @_helpers.copy_docstring(credentials.Scoped) def with_scopes(self, scopes, default_scopes=None): kwargs = self._constructor_args() kwargs.update(scopes=scopes, default_scopes=default_scopes) scoped = self.__class__(kwargs) scoped._cred_file_path = self._cred_file_path scoped._metrics_options = self._metrics_options return scoped @abc.abstractmethod def retrieve_subject_token(self, request): """Retrieves the subject token using the credential_source object. Args: request (google.auth.transport.Request): A callable used to make HTTP requests. Returns: str: The retrieved subject token. """ # pylint: disable=missing-raises-doc # (pylint doesn't recognize that this is abstract) raise NotImplementedError("retrieve_subject_token must be implemented") def get_project_id(self, request): """Retrieves the project ID corresponding to the workload identity or workforce pool. For workforce pool credentials, it returns the project ID corresponding to the workforce_pool_user_project. When not determinable, None is returned. This is introduced to support the current pattern of using the Auth library: credentials, project_id = google.auth.default() The resource may not have permission (resourcemanager.projects.get) to call this API or the required scopes may not be selected: https://cloud.google.com/resource-manager/reference/rest/v1/projects/get#authorization-scopes Args: request (google.auth.transport.Request): A callable used to make HTTP requests. Returns: Optional[str]: The project ID corresponding to the workload identity pool or workforce pool if determinable. """ if self._project_id: # If already retrieved, return the cached project ID value. return self._project_id scopes = self._scopes if self._scopes is not None else self._default_scopes # Scopes are required in order to retrieve a valid access token. project_number = self.project_number or self._workforce_pool_user_project if project_number and scopes: headers = {} url = _CLOUD_RESOURCE_MANAGER + project_number self.before_request(request, "GET", url, headers) response = request(url=url, method="GET", headers=headers) response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) response_data = json.loads(response_body) if response.status == 200: # Cache result as this field is immutable. self._project_id = response_data.get("projectId") return self._project_id return None @_helpers.copy_docstring(credentials.Credentials) def refresh(self, request): scopes = self._scopes if self._scopes is not None else self._default_scopes # Inject client certificate into request. if self._mtls_required(): request = functools.partial( request, cert=self._get_mtls_cert_and_key_paths() ) if self._should_initialize_impersonated_credentials(): self._impersonated_credentials = self._initialize_impersonated_credentials() if self._impersonated_credentials: self._impersonated_credentials.refresh(request) self.token = self._impersonated_credentials.token self.expiry = self._impersonated_credentials.expiry else: now = _helpers.utcnow() additional_options = None # Do not pass workforce_pool_user_project when client authentication # is used. The client ID is sufficient for determining the user project. if self._workforce_pool_user_project and not self._client_id: additional_options = {"userProject": self._workforce_pool_user_project} additional_headers = { metrics.API_CLIENT_HEADER: metrics.byoid_metrics_header( self._metrics_options ) } response_data = self._sts_client.exchange_token( request=request, grant_type=_STS_GRANT_TYPE, subject_token=self.retrieve_subject_token(request), subject_token_type=self._subject_token_type, audience=self._audience, scopes=scopes, requested_token_type=_STS_REQUESTED_TOKEN_TYPE, additional_options=additional_options, additional_headers=additional_headers, ) self.token = response_data.get("access_token") expires_in = response_data.get("expires_in") # Some services do not respect the OAUTH2.0 RFC and send expires_in as a # JSON String. if isinstance(expires_in, str): expires_in = int(expires_in) lifetime = datetime.timedelta(seconds=expires_in) self.expiry = now + lifetime def _make_copy(self): kwargs = self._constructor_args() new_cred = self.__class__(kwargs) new_cred._cred_file_path = self._cred_file_path new_cred._metrics_options = self._metrics_options return new_cred @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): # Return copy of instance with the provided quota project ID. cred = self._make_copy() cred._quota_project_id = quota_project_id return cred @_helpers.copy_docstring(credentials.CredentialsWithTokenUri) def with_token_uri(self, token_uri): cred = self._make_copy() cred._token_url = token_uri return cred @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain) def with_universe_domain(self, universe_domain): cred = self._make_copy() cred._universe_domain = universe_domain return cred def _should_initialize_impersonated_credentials(self): return ( self._service_account_impersonation_url is not None and self._impersonated_credentials is None ) def _initialize_impersonated_credentials(self): """Generates an impersonated credentials. For more details, see `projects.serviceAccounts.generateAccessToken`_. .. _projects.serviceAccounts.generateAccessToken: https://cloud.google.com/iam/docs/reference/credentials/rest/v1/projects.serviceAccounts/generateAccessToken Returns: impersonated_credentials.Credential: The impersonated credentials object. Raises: google.auth.exceptions.RefreshError: If the generateAccessToken endpoint returned an error. """ # Return copy of instance with no service account impersonation. kwargs = self._constructor_args() kwargs.update( service_account_impersonation_url=None, service_account_impersonation_options={}, ) source_credentials = self.__class__(kwargs) source_credentials._metrics_options = self._metrics_options # Determine target_principal. target_principal = self.service_account_email if not target_principal: raise exceptions.RefreshError( "Unable to determine target principal from service account impersonation URL." ) scopes = self._scopes if self._scopes is not None else self._default_scopes # Initialize and return impersonated credentials. return impersonated_credentials.Credentials( source_credentials=source_credentials, target_principal=target_principal, target_scopes=scopes, quota_project_id=self._quota_project_id, iam_endpoint_override=self._service_account_impersonation_url, lifetime=self._service_account_impersonation_options.get( "token_lifetime_seconds" ), ) def _create_default_metrics_options(self): metrics_options = {} if self._service_account_impersonation_url: metrics_options["sa-impersonation"] = "true" else: metrics_options["sa-impersonation"] = "false" if self._service_account_impersonation_options.get("token_lifetime_seconds"): metrics_options["config-lifetime"] = "true" else: metrics_options["config-lifetime"] = "false" return metrics_options def _mtls_required(self): """Returns a boolean representing whether the current credential is configured for mTLS and should add a certificate to the outgoing calls to the sts and service account impersonation endpoint. Returns: bool: True if the credential is configured for mTLS, False if it is not. """ return False def _get_mtls_cert_and_key_paths(self): """Gets the file locations for a certificate and private key file to be used for configuring mTLS for the sts and service account impersonation calls. Currently only expected to return a value when using X509 workload identity federation. Returns: Tuple[str, str]: The cert and key file locations as strings in a tuple. Raises: NotImplementedError: When the current credential is not configured for mTLS. """ raise NotImplementedError( "_get_mtls_cert_and_key_location must be implemented." ) @classmethod def from_info(cls, info, kwargs): """Creates a Credentials instance from parsed external account info. Args: info (Mapping[str, str]): The external account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.identity_pool.Credentials: The constructed credentials. Raises: InvalidValue: For invalid parameters. """ return cls( audience=info.get("audience"), subject_token_type=info.get("subject_token_type"), token_url=info.get("token_url"), token_info_url=info.get("token_info_url"), service_account_impersonation_url=info.get( "service_account_impersonation_url" ), service_account_impersonation_options=info.get( "service_account_impersonation" ) or {}, client_id=info.get("client_id"), client_secret=info.get("client_secret"), credential_source=info.get("credential_source"), quota_project_id=info.get("quota_project_id"), workforce_pool_user_project=info.get("workforce_pool_user_project"), universe_domain=info.get( "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN ), kwargs ) @classmethod def from_file(cls, filename, kwargs): """Creates a Credentials instance from an external account json file. Args: filename (str): The path to the external account json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.identity_pool.Credentials: The constructed credentials. """ with io.open(filename, "r", encoding="utf-8") as json_file: data = json.load(json_file) return cls.from_info(data, **kwargs)	25.37 KB	2025-04-14 17:36
📄 external_account_authorized_user.py # Copyright 2022 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """External Account Authorized User Credentials. This module provides credentials based on OAuth 2.0 access and refresh tokens. These credentials usually access resources on behalf of a user (resource owner). Specifically, these are sourced using external identities via Workforce Identity Federation. Obtaining the initial access and refresh token can be done through the Google Cloud CLI. Example credential: { "type": "external_account_authorized_user", "audience": "//iam.googleapis.com/locations/global/workforcePools/$WORKFORCE_POOL_ID/providers/$PROVIDER_ID", "refresh_token": "refreshToken", "token_url": "https://sts.googleapis.com/v1/oauth/token", "token_info_url": "https://sts.googleapis.com/v1/instrospect", "client_id": "clientId", "client_secret": "clientSecret" } """ import datetime import io import json from google.auth import _helpers from google.auth import credentials from google.auth import exceptions from google.oauth2 import sts from google.oauth2 import utils _EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE = "external_account_authorized_user" class Credentials( credentials.CredentialsWithQuotaProject, credentials.ReadOnlyScoped, credentials.CredentialsWithTokenUri, ): """Credentials for External Account Authorized Users. This is used to instantiate Credentials for exchanging refresh tokens from authorized users for Google access token and authorizing requests to Google APIs. The credentials are considered immutable. If you want to modify the quota project, use `with_quota_project` and if you want to modify the token uri, use `with_token_uri`. """ def __init__( self, token=None, expiry=None, refresh_token=None, audience=None, client_id=None, client_secret=None, token_url=None, token_info_url=None, revoke_url=None, scopes=None, quota_project_id=None, universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN, ): """Instantiates a external account authorized user credentials object. Args: token (str): The OAuth 2.0 access token. Can be None if refresh information is provided. expiry (datetime.datetime): The optional expiration datetime of the OAuth 2.0 access token. refresh_token (str): The optional OAuth 2.0 refresh token. If specified, credentials can be refreshed. audience (str): The optional STS audience which contains the resource name for the workforce pool and the provider identifier in that pool. client_id (str): The OAuth 2.0 client ID. Must be specified for refresh, can be left as None if the token can not be refreshed. client_secret (str): The OAuth 2.0 client secret. Must be specified for refresh, can be left as None if the token can not be refreshed. token_url (str): The optional STS token exchange endpoint for refresh. Must be specified for refresh, can be left as None if the token can not be refreshed. token_info_url (str): The optional STS endpoint URL for token introspection. revoke_url (str): The optional STS endpoint URL for revoking tokens. quota_project_id (str): The optional project ID used for quota and billing. This project may be different from the project used to create the credentials. universe_domain (Optional[str]): The universe domain. The default value is googleapis.com. Returns: google.auth.external_account_authorized_user.Credentials: The constructed credentials. """ super(Credentials, self).__init__() self.token = token self.expiry = expiry self._audience = audience self._refresh_token = refresh_token self._token_url = token_url self._token_info_url = token_info_url self._client_id = client_id self._client_secret = client_secret self._revoke_url = revoke_url self._quota_project_id = quota_project_id self._scopes = scopes self._universe_domain = universe_domain or credentials.DEFAULT_UNIVERSE_DOMAIN self._cred_file_path = None if not self.valid and not self.can_refresh: raise exceptions.InvalidOperation( "Token should be created with fields to make it valid (`token` and " "`expiry`), or fields to allow it to refresh (`refresh_token`, " "`token_url`, `client_id`, `client_secret`)." ) self._client_auth = None if self._client_id: self._client_auth = utils.ClientAuthentication( utils.ClientAuthType.basic, self._client_id, self._client_secret ) self._sts_client = sts.Client(self._token_url, self._client_auth) @property def info(self): """Generates the serializable dictionary representation of the current credentials. Returns: Mapping: The dictionary representation of the credentials. This is the reverse of the "from_info" method defined in this class. It is useful for serializing the current credentials so it can deserialized later. """ config_info = self.constructor_args() config_info.update(type=_EXTERNAL_ACCOUNT_AUTHORIZED_USER_JSON_TYPE) if config_info["expiry"]: config_info["expiry"] = config_info["expiry"].isoformat() + "Z" return {key: value for key, value in config_info.items() if value is not None} def constructor_args(self): return { "audience": self._audience, "refresh_token": self._refresh_token, "token_url": self._token_url, "token_info_url": self._token_info_url, "client_id": self._client_id, "client_secret": self._client_secret, "token": self.token, "expiry": self.expiry, "revoke_url": self._revoke_url, "scopes": self._scopes, "quota_project_id": self._quota_project_id, "universe_domain": self._universe_domain, } @property def scopes(self): """Optional[str]: The OAuth 2.0 permission scopes.""" return self._scopes @property def requires_scopes(self): """ False: OAuth 2.0 credentials have their scopes set when the initial token is requested and can not be changed.""" return False @property def client_id(self): """Optional[str]: The OAuth 2.0 client ID.""" return self._client_id @property def client_secret(self): """Optional[str]: The OAuth 2.0 client secret.""" return self._client_secret @property def audience(self): """Optional[str]: The STS audience which contains the resource name for the workforce pool and the provider identifier in that pool.""" return self._audience @property def refresh_token(self): """Optional[str]: The OAuth 2.0 refresh token.""" return self._refresh_token @property def token_url(self): """Optional[str]: The STS token exchange endpoint for refresh.""" return self._token_url @property def token_info_url(self): """Optional[str]: The STS endpoint for token info.""" return self._token_info_url @property def revoke_url(self): """Optional[str]: The STS endpoint for token revocation.""" return self._revoke_url @property def is_user(self): """ True: This credential always represents a user.""" return True @property def can_refresh(self): return all( (self._refresh_token, self._token_url, self._client_id, self._client_secret) ) def get_project_id(self, request=None): """Retrieves the project ID corresponding to the workload identity or workforce pool. For workforce pool credentials, it returns the project ID corresponding to the workforce_pool_user_project. When not determinable, None is returned. Args: request (google.auth.transport.requests.Request): Request object. Unused here, but passed from _default.default(). Return: str: project ID is not determinable for this credential type so it returns None """ return None def to_json(self, strip=None): """Utility function that creates a JSON representation of this credential. Args: strip (Sequence[str]): Optional list of members to exclude from the generated JSON. Returns: str: A JSON representation of this instance. When converted into a dictionary, it can be passed to from_info() to create a new instance. """ strip = strip if strip else [] return json.dumps({k: v for (k, v) in self.info.items() if k not in strip}) def refresh(self, request): """Refreshes the access token. Args: request (google.auth.transport.Request): The object used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If the credentials could not be refreshed. """ if not self.can_refresh: raise exceptions.RefreshError( "The credentials do not contain the necessary fields need to " "refresh the access token. You must specify refresh_token, " "token_url, client_id, and client_secret." ) now = _helpers.utcnow() response_data = self._make_sts_request(request) self.token = response_data.get("access_token") lifetime = datetime.timedelta(seconds=response_data.get("expires_in")) self.expiry = now + lifetime if "refresh_token" in response_data: self._refresh_token = response_data["refresh_token"] def _make_sts_request(self, request): return self._sts_client.refresh_token(request, self._refresh_token) @_helpers.copy_docstring(credentials.Credentials) def get_cred_info(self): if self._cred_file_path: return { "credential_source": self._cred_file_path, "credential_type": "external account authorized user credentials", } return None def _make_copy(self): kwargs = self.constructor_args() cred = self.__class__(kwargs) cred._cred_file_path = self._cred_file_path return cred @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): cred = self._make_copy() cred._quota_project_id = quota_project_id return cred @_helpers.copy_docstring(credentials.CredentialsWithTokenUri) def with_token_uri(self, token_uri): cred = self._make_copy() cred._token_url = token_uri return cred @_helpers.copy_docstring(credentials.CredentialsWithUniverseDomain) def with_universe_domain(self, universe_domain): cred = self._make_copy() cred._universe_domain = universe_domain return cred @classmethod def from_info(cls, info, kwargs): """Creates a Credentials instance from parsed external account info. Args: info (Mapping[str, str]): The external account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.external_account_authorized_user.Credentials: The constructed credentials. Raises: ValueError: For invalid parameters. """ expiry = info.get("expiry") if expiry: expiry = datetime.datetime.strptime( expiry.rstrip("Z").split(".")[0], "%Y-%m-%dT%H:%M:%S" ) return cls( audience=info.get("audience"), refresh_token=info.get("refresh_token"), token_url=info.get("token_url"), token_info_url=info.get("token_info_url"), client_id=info.get("client_id"), client_secret=info.get("client_secret"), token=info.get("token"), expiry=expiry, revoke_url=info.get("revoke_url"), quota_project_id=info.get("quota_project_id"), scopes=info.get("scopes"), universe_domain=info.get( "universe_domain", credentials.DEFAULT_UNIVERSE_DOMAIN ), kwargs ) @classmethod def from_file(cls, filename, kwargs): """Creates a Credentials instance from an external account json file. Args: filename (str): The path to the external account json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.external_account_authorized_user.Credentials: The constructed credentials. """ with io.open(filename, "r", encoding="utf-8") as json_file: data = json.load(json_file) return cls.from_info(data, **kwargs)	13.66 KB	2025-04-14 17:36
📄 iam.py # Copyright 2017 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Tools for using the Google `Cloud Identity and Access Management (IAM) API`_'s auth-related functionality. .. _Cloud Identity and Access Management (IAM) API: https://cloud.google.com/iam/docs/ """ import base64 import http.client as http_client import json from google.auth import _exponential_backoff from google.auth import _helpers from google.auth import credentials from google.auth import crypt from google.auth import exceptions IAM_RETRY_CODES = { http_client.INTERNAL_SERVER_ERROR, http_client.BAD_GATEWAY, http_client.SERVICE_UNAVAILABLE, http_client.GATEWAY_TIMEOUT, } _IAM_SCOPE = ["https://www.googleapis.com/auth/iam"] _IAM_ENDPOINT = ( "https://iamcredentials.googleapis.com/v1/projects/-" + "/serviceAccounts/{}:generateAccessToken" ) _IAM_SIGN_ENDPOINT = ( "https://iamcredentials.googleapis.com/v1/projects/-" + "/serviceAccounts/{}:signBlob" ) _IAM_SIGNJWT_ENDPOINT = ( "https://iamcredentials.googleapis.com/v1/projects/-" + "/serviceAccounts/{}:signJwt" ) _IAM_IDTOKEN_ENDPOINT = ( "https://iamcredentials.googleapis.com/v1/" + "projects/-/serviceAccounts/{}:generateIdToken" ) class Signer(crypt.Signer): """Signs messages using the IAM `signBlob API`_. This is useful when you need to sign bytes but do not have access to the credential's private key file. .. _signBlob API: https://cloud.google.com/iam/reference/rest/v1/projects.serviceAccounts /signBlob """ def __init__(self, request, credentials, service_account_email): """ Args: request (google.auth.transport.Request): The object used to make HTTP requests. credentials (google.auth.credentials.Credentials): The credentials that will be used to authenticate the request to the IAM API. The credentials must have of one the following scopes: - https://www.googleapis.com/auth/iam - https://www.googleapis.com/auth/cloud-platform service_account_email (str): The service account email identifying which service account to use to sign bytes. Often, this can be the same as the service account email in the given credentials. """ self._request = request self._credentials = credentials self._service_account_email = service_account_email def _make_signing_request(self, message): """Makes a request to the API signBlob API.""" message = _helpers.to_bytes(message) method = "POST" url = _IAM_SIGN_ENDPOINT.replace( credentials.DEFAULT_UNIVERSE_DOMAIN, self._credentials.universe_domain ).format(self._service_account_email) headers = {"Content-Type": "application/json"} body = json.dumps( {"payload": base64.b64encode(message).decode("utf-8")} ).encode("utf-8") retries = _exponential_backoff.ExponentialBackoff() for _ in retries: self._credentials.before_request(self._request, method, url, headers) response = self._request(url=url, method=method, body=body, headers=headers) if response.status in IAM_RETRY_CODES: continue if response.status != http_client.OK: raise exceptions.TransportError( "Error calling the IAM signBlob API: {}".format(response.data) ) return json.loads(response.data.decode("utf-8")) raise exceptions.TransportError("exhausted signBlob endpoint retries") @property def key_id(self): """Optional[str]: The key ID used to identify this private key. .. warning:: This is always ``None``. The key ID used by IAM can not be reliably determined ahead of time. """ return None @_helpers.copy_docstring(crypt.Signer) def sign(self, message): response = self._make_signing_request(message) return base64.b64decode(response["signedBlob"])	4.56 KB	2025-04-14 17:36
📄 identity_pool.py # Copyright 2020 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Identity Pool Credentials. This module provides credentials to access Google Cloud resources from on-prem or non-Google Cloud platforms which support external credentials (e.g. OIDC ID tokens) retrieved from local file locations or local servers. This includes Microsoft Azure and OIDC identity providers (e.g. K8s workloads registered with Hub with Hub workload identity enabled). These credentials are recommended over the use of service account credentials in on-prem/non-Google Cloud platforms as they do not involve the management of long-live service account private keys. Identity Pool Credentials are initialized using external_account arguments which are typically loaded from an external credentials file or an external credentials URL. This module also provides a definition for an abstract subject token supplier. This supplier can be implemented to return a valid OIDC or SAML2.0 subject token and used to create Identity Pool credentials. The credentials will then call the supplier instead of using pre-defined methods such as reading a local file or calling a URL. """ try: from collections.abc import Mapping # Python 2.7 compatibility except ImportError: # pragma: NO COVER from collections import Mapping # type: ignore import abc import base64 import json import os from typing import NamedTuple from google.auth import _helpers from google.auth import exceptions from google.auth import external_account from google.auth.transport import _mtls_helper class SubjectTokenSupplier(metaclass=abc.ABCMeta): """Base class for subject token suppliers. This can be implemented with custom logic to retrieve a subject token to exchange for a Google Cloud access token when using Workload or Workforce Identity Federation. The identity pool credential does not cache the subject token, so caching logic should be added in the implementation. """ @abc.abstractmethod def get_subject_token(self, context, request): """Returns the requested subject token. The subject token must be valid. .. warning: This is not cached by the calling Google credential, so caching logic should be implemented in the supplier. Args: context (google.auth.externalaccount.SupplierContext): The context object containing information about the requested audience and subject token type. request (google.auth.transport.Request): The object used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If an error is encountered during subject token retrieval logic. Returns: str: The requested subject token string. """ raise NotImplementedError("") class _TokenContent(NamedTuple): """Models the token content response from file and url internal suppliers. Attributes: content (str): The string content of the file or URL response. location (str): The location the content was retrieved from. This will either be a file location or a URL. """ content: str location: str class _FileSupplier(SubjectTokenSupplier): """ Internal implementation of subject token supplier which supports reading a subject token from a file.""" def __init__(self, path, format_type, subject_token_field_name): self._path = path self._format_type = format_type self._subject_token_field_name = subject_token_field_name @_helpers.copy_docstring(SubjectTokenSupplier) def get_subject_token(self, context, request): if not os.path.exists(self._path): raise exceptions.RefreshError("File '{}' was not found.".format(self._path)) with open(self._path, "r", encoding="utf-8") as file_obj: token_content = _TokenContent(file_obj.read(), self._path) return _parse_token_data( token_content, self._format_type, self._subject_token_field_name ) class _UrlSupplier(SubjectTokenSupplier): """ Internal implementation of subject token supplier which supports retrieving a subject token by calling a URL endpoint.""" def __init__(self, url, format_type, subject_token_field_name, headers): self._url = url self._format_type = format_type self._subject_token_field_name = subject_token_field_name self._headers = headers @_helpers.copy_docstring(SubjectTokenSupplier) def get_subject_token(self, context, request): response = request(url=self._url, method="GET", headers=self._headers) # support both string and bytes type response.data response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) if response.status != 200: raise exceptions.RefreshError( "Unable to retrieve Identity Pool subject token", response_body ) token_content = _TokenContent(response_body, self._url) return _parse_token_data( token_content, self._format_type, self._subject_token_field_name ) class _X509Supplier(SubjectTokenSupplier): """Internal supplier for X509 workload credentials. This class is used internally and always returns an empty string as the subject token.""" def __init__(self, trust_chain_path, leaf_cert_callback): self._trust_chain_path = trust_chain_path self._leaf_cert_callback = leaf_cert_callback @_helpers.copy_docstring(SubjectTokenSupplier) def get_subject_token(self, context, request): # Import OpennSSL inline because it is an extra import only required by customers # using mTLS. from OpenSSL import crypto leaf_cert = crypto.load_certificate( crypto.FILETYPE_PEM, self._leaf_cert_callback() ) trust_chain = self._read_trust_chain() cert_chain = [] cert_chain.append(_X509Supplier._encode_cert(leaf_cert)) if trust_chain is None or len(trust_chain) == 0: return json.dumps(cert_chain) # Append the first cert if it is not the leaf cert. first_cert = _X509Supplier._encode_cert(trust_chain[0]) if first_cert != cert_chain[0]: cert_chain.append(first_cert) for i in range(1, len(trust_chain)): encoded = _X509Supplier._encode_cert(trust_chain[i]) # Check if the current cert is the leaf cert and raise an exception if it is. if encoded == cert_chain[0]: raise exceptions.RefreshError( "The leaf certificate must be at the top of the trust chain file" ) else: cert_chain.append(encoded) return json.dumps(cert_chain) def _read_trust_chain(self): # Import OpennSSL inline because it is an extra import only required by customers # using mTLS. from OpenSSL import crypto certificate_trust_chain = [] # If no trust chain path was provided, return an empty list. if self._trust_chain_path is None or self._trust_chain_path == "": return certificate_trust_chain try: # Open the trust chain file. with open(self._trust_chain_path, "rb") as f: trust_chain_data = f.read() # Split PEM data into individual certificates. cert_blocks = trust_chain_data.split(b"-----BEGIN CERTIFICATE-----") for cert_block in cert_blocks: # Skip empty blocks. if cert_block.strip(): cert_data = b"-----BEGIN CERTIFICATE-----" + cert_block try: # Load each certificate and add it to the trust chain. cert = crypto.load_certificate( crypto.FILETYPE_PEM, cert_data ) certificate_trust_chain.append(cert) except Exception as e: raise exceptions.RefreshError( "Error loading PEM certificates from the trust chain file '{}'".format( self._trust_chain_path ) ) from e return certificate_trust_chain except FileNotFoundError: raise exceptions.RefreshError( "Trust chain file '{}' was not found.".format(self._trust_chain_path) ) def _encode_cert(cert): # Import OpennSSL inline because it is an extra import only required by customers # using mTLS. from OpenSSL import crypto return base64.b64encode( crypto.dump_certificate(crypto.FILETYPE_ASN1, cert) ).decode("utf-8") def _parse_token_data(token_content, format_type="text", subject_token_field_name=None): if format_type == "text": token = token_content.content else: try: # Parse file content as JSON. response_data = json.loads(token_content.content) # Get the subject_token. token = response_data[subject_token_field_name] except (KeyError, ValueError): raise exceptions.RefreshError( "Unable to parse subject_token from JSON file '{}' using key '{}'".format( token_content.location, subject_token_field_name ) ) if not token: raise exceptions.RefreshError( "Missing subject_token in the credential_source file" ) return token class Credentials(external_account.Credentials): """External account credentials sourced from files and URLs.""" def __init__( self, audience, subject_token_type, token_url=external_account._DEFAULT_TOKEN_URL, credential_source=None, subject_token_supplier=None, args, kwargs ): """Instantiates an external account credentials object from a file/URL. Args: audience (str): The STS audience field. subject_token_type (str): The subject token type based on the Oauth2.0 token exchange spec. Expected values include:: “urn:ietf:params:oauth:token-type:jwt” “urn:ietf:params:oauth:token-type:id-token” “urn:ietf:params:oauth:token-type:saml2” token_url (Optional [str]): The STS endpoint URL. If not provided, will default to "https://sts.googleapis.com/v1/token". credential_source (Optional [Mapping]): The credential source dictionary used to provide instructions on how to retrieve external credential to be exchanged for Google access tokens. Either a credential source or a subject token supplier must be provided. Example credential_source for url-sourced credential:: { "url": "http://www.example.com", "format": { "type": "json", "subject_token_field_name": "access_token", }, "headers": {"foo": "bar"}, } Example credential_source for file-sourced credential:: { "file": "/path/to/token/file.txt" } subject_token_supplier (Optional [SubjectTokenSupplier]): Optional subject token supplier. This will be called to supply a valid subject token which will then be exchanged for Google access tokens. Either a subject token supplier or a credential source must be provided. args (List): Optional positional arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method. kwargs (Mapping): Optional keyword arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method. Raises: google.auth.exceptions.RefreshError: If an error is encountered during access token retrieval logic. ValueError: For invalid parameters. .. note:: Typically one of the helper constructors :meth:`from_file` or :meth:`from_info` are used instead of calling the constructor directly. """ super(Credentials, self).__init__( audience=audience, subject_token_type=subject_token_type, token_url=token_url, credential_source=credential_source, args, kwargs ) if credential_source is None and subject_token_supplier is None: raise exceptions.InvalidValue( "A valid credential source or a subject token supplier must be provided." ) if credential_source is not None and subject_token_supplier is not None: raise exceptions.InvalidValue( "Identity pool credential cannot have both a credential source and a subject token supplier." ) if subject_token_supplier is not None: self._subject_token_supplier = subject_token_supplier self._credential_source_file = None self._credential_source_url = None self._credential_source_certificate = None else: if not isinstance(credential_source, Mapping): self._credential_source_executable = None raise exceptions.MalformedError( "Invalid credential_source. The credential_source is not a dict." ) self._credential_source_file = credential_source.get("file") self._credential_source_url = credential_source.get("url") self._credential_source_certificate = credential_source.get("certificate") # environment_id is only supported in AWS or dedicated future external # account credentials. if "environment_id" in credential_source: raise exceptions.MalformedError( "Invalid Identity Pool credential_source field 'environment_id'" ) # check that only one of file, url, or certificate are provided. self._validate_single_source() if self._credential_source_certificate: self._validate_certificate_config() else: self._validate_file_or_url_config(credential_source) if self._credential_source_file: self._subject_token_supplier = _FileSupplier( self._credential_source_file, self._credential_source_format_type, self._credential_source_field_name, ) elif self._credential_source_url: self._subject_token_supplier = _UrlSupplier( self._credential_source_url, self._credential_source_format_type, self._credential_source_field_name, self._credential_source_headers, ) else: # self._credential_source_certificate self._subject_token_supplier = _X509Supplier( self._trust_chain_path, self._get_cert_bytes ) @_helpers.copy_docstring(external_account.Credentials) def retrieve_subject_token(self, request): return self._subject_token_supplier.get_subject_token( self._supplier_context, request ) def _get_mtls_cert_and_key_paths(self): if self._credential_source_certificate is None: raise exceptions.RefreshError( 'The credential is not configured to use mtls requests. The credential should include a "certificate" section in the credential source.' ) else: return _mtls_helper._get_workload_cert_and_key_paths( self._certificate_config_location ) def _get_cert_bytes(self): cert_path, _ = self._get_mtls_cert_and_key_paths() return _mtls_helper._read_cert_file(cert_path) def _mtls_required(self): return self._credential_source_certificate is not None def _create_default_metrics_options(self): metrics_options = super(Credentials, self)._create_default_metrics_options() # Check that credential source is a dict before checking for credential type. This check needs to be done # here because the external_account credential constructor needs to pass the metrics options to the # impersonated credential object before the identity_pool credentials are validated. if isinstance(self._credential_source, Mapping): if self._credential_source.get("file"): metrics_options["source"] = "file" elif self._credential_source.get("url"): metrics_options["source"] = "url" else: metrics_options["source"] = "x509" else: metrics_options["source"] = "programmatic" return metrics_options def _has_custom_supplier(self): return self._credential_source is None def _constructor_args(self): args = super(Credentials, self)._constructor_args() # If a custom supplier was used, append it to the args dict. if self._has_custom_supplier(): args.update({"subject_token_supplier": self._subject_token_supplier}) return args def _validate_certificate_config(self): self._certificate_config_location = self._credential_source_certificate.get( "certificate_config_location" ) use_default = self._credential_source_certificate.get( "use_default_certificate_config" ) self._trust_chain_path = self._credential_source_certificate.get( "trust_chain_path" ) if self._certificate_config_location and use_default: raise exceptions.MalformedError( "Invalid certificate configuration, certificate_config_location cannot be specified when use_default_certificate_config = true." ) if not self._certificate_config_location and not use_default: raise exceptions.MalformedError( "Invalid certificate configuration, use_default_certificate_config should be true if no certificate_config_location is provided." ) def _validate_file_or_url_config(self, credential_source): self._credential_source_headers = credential_source.get("headers") credential_source_format = credential_source.get("format", {}) # Get credential_source format type. When not provided, this # defaults to text. self._credential_source_format_type = ( credential_source_format.get("type") or "text" ) if self._credential_source_format_type not in ["text", "json"]: raise exceptions.MalformedError( "Invalid credential_source format '{}'".format( self._credential_source_format_type ) ) # For JSON types, get the required subject_token field name. if self._credential_source_format_type == "json": self._credential_source_field_name = credential_source_format.get( "subject_token_field_name" ) if self._credential_source_field_name is None: raise exceptions.MalformedError( "Missing subject_token_field_name for JSON credential_source format" ) else: self._credential_source_field_name = None def _validate_single_source(self): credential_sources = [ self._credential_source_file, self._credential_source_url, self._credential_source_certificate, ] valid_credential_sources = list( filter(lambda source: source is not None, credential_sources) ) if len(valid_credential_sources) > 1: raise exceptions.MalformedError( "Ambiguous credential_source. 'file', 'url', and 'certificate' are mutually exclusive.." ) if len(valid_credential_sources) != 1: raise exceptions.MalformedError( "Missing credential_source. A 'file', 'url', or 'certificate' must be provided." ) @classmethod def from_info(cls, info, kwargs): """Creates an Identity Pool Credentials instance from parsed external account info. Args: info (Mapping[str, str]): The Identity Pool external account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.identity_pool.Credentials: The constructed credentials. Raises: ValueError: For invalid parameters. """ subject_token_supplier = info.get("subject_token_supplier") kwargs.update({"subject_token_supplier": subject_token_supplier}) return super(Credentials, cls).from_info(info, kwargs) @classmethod def from_file(cls, filename, kwargs): """Creates an IdentityPool Credentials instance from an external account json file. Args: filename (str): The path to the IdentityPool external account json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.identity_pool.Credentials: The constructed credentials. """ return super(Credentials, cls).from_file(filename, **kwargs)	22.03 KB	2025-04-14 17:36
📄 impersonated_credentials.py # Copyright 2018 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Google Cloud Impersonated credentials. This module provides authentication for applications where local credentials impersonates a remote service account using `IAM Credentials API`_. This class can be used to impersonate a service account as long as the original Credential object has the "Service Account Token Creator" role on the target service account. .. _IAM Credentials API: https://cloud.google.com/iam/credentials/reference/rest/ """ import base64 import copy from datetime import datetime import http.client as http_client import json from google.auth import _exponential_backoff from google.auth import _helpers from google.auth import credentials from google.auth import exceptions from google.auth import iam from google.auth import jwt from google.auth import metrics from google.oauth2 import _client _REFRESH_ERROR = "Unable to acquire impersonated credentials" _DEFAULT_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds _GOOGLE_OAUTH2_TOKEN_ENDPOINT = "https://oauth2.googleapis.com/token" _SOURCE_CREDENTIAL_AUTHORIZED_USER_TYPE = "authorized_user" _SOURCE_CREDENTIAL_SERVICE_ACCOUNT_TYPE = "service_account" _SOURCE_CREDENTIAL_EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE = ( "external_account_authorized_user" ) def _make_iam_token_request( request, principal, headers, body, universe_domain=credentials.DEFAULT_UNIVERSE_DOMAIN, iam_endpoint_override=None, ): """Makes a request to the Google Cloud IAM service for an access token. Args: request (Request): The Request object to use. principal (str): The principal to request an access token for. headers (Mapping[str, str]): Map of headers to transmit. body (Mapping[str, str]): JSON Payload body for the iamcredentials API call. iam_endpoint_override (Optiona[str]): The full IAM endpoint override with the target_principal embedded. This is useful when supporting impersonation with regional endpoints. Raises: google.auth.exceptions.TransportError: Raised if there is an underlying HTTP connection error google.auth.exceptions.RefreshError: Raised if the impersonated credentials are not available. Common reasons are `iamcredentials.googleapis.com` is not enabled or the `Service Account Token Creator` is not assigned """ iam_endpoint = iam_endpoint_override or iam._IAM_ENDPOINT.replace( credentials.DEFAULT_UNIVERSE_DOMAIN, universe_domain ).format(principal) body = json.dumps(body).encode("utf-8") response = request(url=iam_endpoint, method="POST", headers=headers, body=body) # support both string and bytes type response.data response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) if response.status != http_client.OK: raise exceptions.RefreshError(_REFRESH_ERROR, response_body) try: token_response = json.loads(response_body) token = token_response["accessToken"] expiry = datetime.strptime(token_response["expireTime"], "%Y-%m-%dT%H:%M:%SZ") return token, expiry except (KeyError, ValueError) as caught_exc: new_exc = exceptions.RefreshError( "{}: No access token or invalid expiration in response.".format( _REFRESH_ERROR ), response_body, ) raise new_exc from caught_exc class Credentials( credentials.Scoped, credentials.CredentialsWithQuotaProject, credentials.Signing ): """This module defines impersonated credentials which are essentially impersonated identities. Impersonated Credentials allows credentials issued to a user or service account to impersonate another. The target service account must grant the originating credential principal the `Service Account Token Creator`_ IAM role: For more information about Token Creator IAM role and IAMCredentials API, see `Creating Short-Lived Service Account Credentials`_. .. _Service Account Token Creator: https://cloud.google.com/iam/docs/service-accounts#the_service_account_token_creator_role .. _Creating Short-Lived Service Account Credentials: https://cloud.google.com/iam/docs/creating-short-lived-service-account-credentials Usage: First grant source_credentials the `Service Account Token Creator` role on the target account to impersonate. In this example, the service account represented by svc_account.json has the token creator role on `impersonated-account@_project_.iam.gserviceaccount.com`. Enable the IAMCredentials API on the source project: `gcloud services enable iamcredentials.googleapis.com`. Initialize a source credential which does not have access to list bucket:: from google.oauth2 import service_account target_scopes = [ 'https://www.googleapis.com/auth/devstorage.read_only'] source_credentials = ( service_account.Credentials.from_service_account_file( '/path/to/svc_account.json', scopes=target_scopes)) Now use the source credentials to acquire credentials to impersonate another service account:: from google.auth import impersonated_credentials target_credentials = impersonated_credentials.Credentials( source_credentials=source_credentials, target_principal='impersonated-account@_project_.iam.gserviceaccount.com', target_scopes = target_scopes, lifetime=500) Resource access is granted:: client = storage.Client(credentials=target_credentials) buckets = client.list_buckets(project='your_project') for bucket in buckets: print(bucket.name) """ def __init__( self, source_credentials, target_principal, target_scopes, delegates=None, subject=None, lifetime=_DEFAULT_TOKEN_LIFETIME_SECS, quota_project_id=None, iam_endpoint_override=None, ): """ Args: source_credentials (google.auth.Credentials): The source credential used as to acquire the impersonated credentials. target_principal (str): The service account to impersonate. target_scopes (Sequence[str]): Scopes to request during the authorization grant. delegates (Sequence[str]): The chained list of delegates required to grant the final access_token. If set, the sequence of identities must have "Service Account Token Creator" capability granted to the prceeding identity. For example, if set to [serviceAccountB, serviceAccountC], the source_credential must have the Token Creator role on serviceAccountB. serviceAccountB must have the Token Creator on serviceAccountC. Finally, C must have Token Creator on target_principal. If left unset, source_credential must have that role on target_principal. lifetime (int): Number of seconds the delegated credential should be valid for (upto 3600). quota_project_id (Optional[str]): The project ID used for quota and billing. This project may be different from the project used to create the credentials. iam_endpoint_override (Optional[str]): The full IAM endpoint override with the target_principal embedded. This is useful when supporting impersonation with regional endpoints. subject (Optional[str]): sub field of a JWT. This field should only be set if you wish to impersonate as a user. This feature is useful when using domain wide delegation. """ super(Credentials, self).__init__() self._source_credentials = copy.copy(source_credentials) # Service account source credentials must have the _IAM_SCOPE # added to refresh correctly. User credentials cannot have # their original scopes modified. if isinstance(self._source_credentials, credentials.Scoped): self._source_credentials = self._source_credentials.with_scopes( iam._IAM_SCOPE ) # If the source credential is service account and self signed jwt # is needed, we need to create a jwt credential inside it if ( hasattr(self._source_credentials, "_create_self_signed_jwt") and self._source_credentials._always_use_jwt_access ): self._source_credentials._create_self_signed_jwt(None) self._universe_domain = source_credentials.universe_domain self._target_principal = target_principal self._target_scopes = target_scopes self._delegates = delegates self._subject = subject self._lifetime = lifetime or _DEFAULT_TOKEN_LIFETIME_SECS self.token = None self.expiry = _helpers.utcnow() self._quota_project_id = quota_project_id self._iam_endpoint_override = iam_endpoint_override self._cred_file_path = None def _metric_header_for_usage(self): return metrics.CRED_TYPE_SA_IMPERSONATE @_helpers.copy_docstring(credentials.Credentials) def refresh(self, request): self._update_token(request) def _update_token(self, request): """Updates credentials with a new access_token representing the impersonated account. Args: request (google.auth.transport.requests.Request): Request object to use for refreshing credentials. """ # Refresh our source credentials if it is not valid. if ( self._source_credentials.token_state == credentials.TokenState.STALE or self._source_credentials.token_state == credentials.TokenState.INVALID ): self._source_credentials.refresh(request) body = { "delegates": self._delegates, "scope": self._target_scopes, "lifetime": str(self._lifetime) + "s", } headers = { "Content-Type": "application/json", metrics.API_CLIENT_HEADER: metrics.token_request_access_token_impersonate(), } # Apply the source credentials authentication info. self._source_credentials.apply(headers) # If a subject is specified a domain-wide delegation auth-flow is initiated # to impersonate as the provided subject (user). if self._subject: if self.universe_domain != credentials.DEFAULT_UNIVERSE_DOMAIN: raise exceptions.GoogleAuthError( "Domain-wide delegation is not supported in universes other " + "than googleapis.com" ) now = _helpers.utcnow() payload = { "iss": self._target_principal, "scope": _helpers.scopes_to_string(self._target_scopes or ()), "sub": self._subject, "aud": _GOOGLE_OAUTH2_TOKEN_ENDPOINT, "iat": _helpers.datetime_to_secs(now), "exp": _helpers.datetime_to_secs(now) + _DEFAULT_TOKEN_LIFETIME_SECS, } assertion = _sign_jwt_request( request=request, principal=self._target_principal, headers=headers, payload=payload, delegates=self._delegates, ) self.token, self.expiry, _ = _client.jwt_grant( request, _GOOGLE_OAUTH2_TOKEN_ENDPOINT, assertion ) return self.token, self.expiry = _make_iam_token_request( request=request, principal=self._target_principal, headers=headers, body=body, universe_domain=self.universe_domain, iam_endpoint_override=self._iam_endpoint_override, ) def sign_bytes(self, message): from google.auth.transport.requests import AuthorizedSession iam_sign_endpoint = iam._IAM_SIGN_ENDPOINT.replace( credentials.DEFAULT_UNIVERSE_DOMAIN, self.universe_domain ).format(self._target_principal) body = { "payload": base64.b64encode(message).decode("utf-8"), "delegates": self._delegates, } headers = {"Content-Type": "application/json"} authed_session = AuthorizedSession(self._source_credentials) try: retries = _exponential_backoff.ExponentialBackoff() for _ in retries: response = authed_session.post( url=iam_sign_endpoint, headers=headers, json=body ) if response.status_code in iam.IAM_RETRY_CODES: continue if response.status_code != http_client.OK: raise exceptions.TransportError( "Error calling sign_bytes: {}".format(response.json()) ) return base64.b64decode(response.json()["signedBlob"]) finally: authed_session.close() raise exceptions.TransportError("exhausted signBlob endpoint retries") @property def signer_email(self): return self._target_principal @property def service_account_email(self): return self._target_principal @property def signer(self): return self @property def requires_scopes(self): return not self._target_scopes @_helpers.copy_docstring(credentials.Credentials) def get_cred_info(self): if self._cred_file_path: return { "credential_source": self._cred_file_path, "credential_type": "impersonated credentials", "principal": self._target_principal, } return None def _make_copy(self): cred = self.__class__( self._source_credentials, target_principal=self._target_principal, target_scopes=self._target_scopes, delegates=self._delegates, lifetime=self._lifetime, quota_project_id=self._quota_project_id, iam_endpoint_override=self._iam_endpoint_override, ) cred._cred_file_path = self._cred_file_path return cred @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): cred = self._make_copy() cred._quota_project_id = quota_project_id return cred @_helpers.copy_docstring(credentials.Scoped) def with_scopes(self, scopes, default_scopes=None): cred = self._make_copy() cred._target_scopes = scopes or default_scopes return cred @classmethod def from_impersonated_service_account_info(cls, info, scopes=None): """Creates a Credentials instance from parsed impersonated service account credentials info. Args: info (Mapping[str, str]): The impersonated service account credentials info in Google format. scopes (Sequence[str]): Optional list of scopes to include in the credentials. Returns: google.oauth2.credentials.Credentials: The constructed credentials. Raises: InvalidType: If the info["source_credentials"] are not a supported impersonation type InvalidValue: If the info["service_account_impersonation_url"] is not in the expected format. ValueError: If the info is not in the expected format. """ source_credentials_info = info.get("source_credentials") source_credentials_type = source_credentials_info.get("type") if source_credentials_type == _SOURCE_CREDENTIAL_AUTHORIZED_USER_TYPE: from google.oauth2 import credentials source_credentials = credentials.Credentials.from_authorized_user_info( source_credentials_info ) elif source_credentials_type == _SOURCE_CREDENTIAL_SERVICE_ACCOUNT_TYPE: from google.oauth2 import service_account source_credentials = service_account.Credentials.from_service_account_info( source_credentials_info ) elif ( source_credentials_type == _SOURCE_CREDENTIAL_EXTERNAL_ACCOUNT_AUTHORIZED_USER_TYPE ): from google.auth import external_account_authorized_user source_credentials = external_account_authorized_user.Credentials.from_info( source_credentials_info ) else: raise exceptions.InvalidType( "source credential of type {} is not supported.".format( source_credentials_type ) ) impersonation_url = info.get("service_account_impersonation_url") start_index = impersonation_url.rfind("/") end_index = impersonation_url.find(":generateAccessToken") if start_index == -1 or end_index == -1 or start_index > end_index: raise exceptions.InvalidValue( "Cannot extract target principal from {}".format(impersonation_url) ) target_principal = impersonation_url[start_index + 1 : end_index] delegates = info.get("delegates") quota_project_id = info.get("quota_project_id") return cls( source_credentials, target_principal, scopes, delegates, quota_project_id=quota_project_id, ) class IDTokenCredentials(credentials.CredentialsWithQuotaProject): """Open ID Connect ID Token-based service account credentials. """ def __init__( self, target_credentials, target_audience=None, include_email=False, quota_project_id=None, ): """ Args: target_credentials (google.auth.Credentials): The target credential used as to acquire the id tokens for. target_audience (string): Audience to issue the token for. include_email (bool): Include email in IdToken quota_project_id (Optional[str]): The project ID used for quota and billing. """ super(IDTokenCredentials, self).__init__() if not isinstance(target_credentials, Credentials): raise exceptions.GoogleAuthError( "Provided Credential must be " "impersonated_credentials" ) self._target_credentials = target_credentials self._target_audience = target_audience self._include_email = include_email self._quota_project_id = quota_project_id def from_credentials(self, target_credentials, target_audience=None): return self.__class__( target_credentials=target_credentials, target_audience=target_audience, include_email=self._include_email, quota_project_id=self._quota_project_id, ) def with_target_audience(self, target_audience): return self.__class__( target_credentials=self._target_credentials, target_audience=target_audience, include_email=self._include_email, quota_project_id=self._quota_project_id, ) def with_include_email(self, include_email): return self.__class__( target_credentials=self._target_credentials, target_audience=self._target_audience, include_email=include_email, quota_project_id=self._quota_project_id, ) @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): return self.__class__( target_credentials=self._target_credentials, target_audience=self._target_audience, include_email=self._include_email, quota_project_id=quota_project_id, ) @_helpers.copy_docstring(credentials.Credentials) def refresh(self, request): from google.auth.transport.requests import AuthorizedSession iam_sign_endpoint = iam._IAM_IDTOKEN_ENDPOINT.replace( credentials.DEFAULT_UNIVERSE_DOMAIN, self._target_credentials.universe_domain, ).format(self._target_credentials.signer_email) body = { "audience": self._target_audience, "delegates": self._target_credentials._delegates, "includeEmail": self._include_email, } headers = { "Content-Type": "application/json", metrics.API_CLIENT_HEADER: metrics.token_request_id_token_impersonate(), } authed_session = AuthorizedSession( self._target_credentials._source_credentials, auth_request=request ) try: response = authed_session.post( url=iam_sign_endpoint, headers=headers, data=json.dumps(body).encode("utf-8"), ) finally: authed_session.close() if response.status_code != http_client.OK: raise exceptions.RefreshError( "Error getting ID token: {}".format(response.json()) ) id_token = response.json()["token"] self.token = id_token self.expiry = datetime.utcfromtimestamp( jwt.decode(id_token, verify=False)["exp"] ) def _sign_jwt_request(request, principal, headers, payload, delegates=[]): """Makes a request to the Google Cloud IAM service to sign a JWT using a service account's system-managed private key. Args: request (Request): The Request object to use. principal (str): The principal to request an access token for. headers (Mapping[str, str]): Map of headers to transmit. payload (Mapping[str, str]): The JWT payload to sign. Must be a serialized JSON object that contains a JWT Claims Set. delegates (Sequence[str]): The chained list of delegates required to grant the final access_token. If set, the sequence of identities must have "Service Account Token Creator" capability granted to the prceeding identity. For example, if set to [serviceAccountB, serviceAccountC], the source_credential must have the Token Creator role on serviceAccountB. serviceAccountB must have the Token Creator on serviceAccountC. Finally, C must have Token Creator on target_principal. If left unset, source_credential must have that role on target_principal. Raises: google.auth.exceptions.TransportError: Raised if there is an underlying HTTP connection error google.auth.exceptions.RefreshError: Raised if the impersonated credentials are not available. Common reasons are `iamcredentials.googleapis.com` is not enabled or the `Service Account Token Creator` is not assigned """ iam_endpoint = iam._IAM_SIGNJWT_ENDPOINT.format(principal) body = {"delegates": delegates, "payload": json.dumps(payload)} body = json.dumps(body).encode("utf-8") response = request(url=iam_endpoint, method="POST", headers=headers, body=body) # support both string and bytes type response.data response_body = ( response.data.decode("utf-8") if hasattr(response.data, "decode") else response.data ) if response.status != http_client.OK: raise exceptions.RefreshError(_REFRESH_ERROR, response_body) try: jwt_response = json.loads(response_body) signed_jwt = jwt_response["signedJwt"] return signed_jwt except (KeyError, ValueError) as caught_exc: new_exc = exceptions.RefreshError( "{}: No signed JWT in response.".format(_REFRESH_ERROR), response_body ) raise new_exc from caught_exc	24.39 KB	2025-04-14 17:36
📄 jwt.py # Copyright 2016 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """JSON Web Tokens Provides support for creating (encoding) and verifying (decoding) JWTs, especially JWTs generated and consumed by Google infrastructure. See `rfc7519`_ for more details on JWTs. To encode a JWT use :func:`encode`:: from google.auth import crypt from google.auth import jwt signer = crypt.Signer(private_key) payload = {'some': 'payload'} encoded = jwt.encode(signer, payload) To decode a JWT and verify claims use :func:`decode`:: claims = jwt.decode(encoded, certs=public_certs) You can also skip verification:: claims = jwt.decode(encoded, verify=False) .. _rfc7519: https://tools.ietf.org/html/rfc7519 """ try: from collections.abc import Mapping # Python 2.7 compatibility except ImportError: # pragma: NO COVER from collections import Mapping # type: ignore import copy import datetime import json import urllib import cachetools from google.auth import _helpers from google.auth import _service_account_info from google.auth import crypt from google.auth import exceptions import google.auth.credentials try: from google.auth.crypt import es256 except ImportError: # pragma: NO COVER es256 = None # type: ignore _DEFAULT_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds _DEFAULT_MAX_CACHE_SIZE = 10 _ALGORITHM_TO_VERIFIER_CLASS = {"RS256": crypt.RSAVerifier} _CRYPTOGRAPHY_BASED_ALGORITHMS = frozenset(["ES256"]) if es256 is not None: # pragma: NO COVER _ALGORITHM_TO_VERIFIER_CLASS["ES256"] = es256.ES256Verifier # type: ignore def encode(signer, payload, header=None, key_id=None): """Make a signed JWT. Args: signer (google.auth.crypt.Signer): The signer used to sign the JWT. payload (Mapping[str, str]): The JWT payload. header (Mapping[str, str]): Additional JWT header payload. key_id (str): The key id to add to the JWT header. If the signer has a key id it will be used as the default. If this is specified it will override the signer's key id. Returns: bytes: The encoded JWT. """ if header is None: header = {} if key_id is None: key_id = signer.key_id header.update({"typ": "JWT"}) if "alg" not in header: if es256 is not None and isinstance(signer, es256.ES256Signer): header.update({"alg": "ES256"}) else: header.update({"alg": "RS256"}) if key_id is not None: header["kid"] = key_id segments = [ _helpers.unpadded_urlsafe_b64encode(json.dumps(header).encode("utf-8")), _helpers.unpadded_urlsafe_b64encode(json.dumps(payload).encode("utf-8")), ] signing_input = b".".join(segments) signature = signer.sign(signing_input) segments.append(_helpers.unpadded_urlsafe_b64encode(signature)) return b".".join(segments) def _decode_jwt_segment(encoded_section): """Decodes a single JWT segment.""" section_bytes = _helpers.padded_urlsafe_b64decode(encoded_section) try: return json.loads(section_bytes.decode("utf-8")) except ValueError as caught_exc: new_exc = exceptions.MalformedError( "Can't parse segment: {0}".format(section_bytes) ) raise new_exc from caught_exc def _unverified_decode(token): """Decodes a token and does no verification. Args: token (Union[str, bytes]): The encoded JWT. Returns: Tuple[Mapping, Mapping, str, str]: header, payload, signed_section, and signature. Raises: google.auth.exceptions.MalformedError: if there are an incorrect amount of segments in the token or segments of the wrong type. """ token = _helpers.to_bytes(token) if token.count(b".") != 2: raise exceptions.MalformedError( "Wrong number of segments in token: {0}".format(token) ) encoded_header, encoded_payload, signature = token.split(b".") signed_section = encoded_header + b"." + encoded_payload signature = _helpers.padded_urlsafe_b64decode(signature) # Parse segments header = _decode_jwt_segment(encoded_header) payload = _decode_jwt_segment(encoded_payload) if not isinstance(header, Mapping): raise exceptions.MalformedError( "Header segment should be a JSON object: {0}".format(encoded_header) ) if not isinstance(payload, Mapping): raise exceptions.MalformedError( "Payload segment should be a JSON object: {0}".format(encoded_payload) ) return header, payload, signed_section, signature def decode_header(token): """Return the decoded header of a token. No verification is done. This is useful to extract the key id from the header in order to acquire the appropriate certificate to verify the token. Args: token (Union[str, bytes]): the encoded JWT. Returns: Mapping: The decoded JWT header. """ header, _, _, _ = _unverified_decode(token) return header def _verify_iat_and_exp(payload, clock_skew_in_seconds=0): """Verifies the ``iat`` (Issued At) and ``exp`` (Expires) claims in a token payload. Args: payload (Mapping[str, str]): The JWT payload. clock_skew_in_seconds (int): The clock skew used for `iat` and `exp` validation. Raises: google.auth.exceptions.InvalidValue: if value validation failed. google.auth.exceptions.MalformedError: if schema validation failed. """ now = _helpers.datetime_to_secs(_helpers.utcnow()) # Make sure the iat and exp claims are present. for key in ("iat", "exp"): if key not in payload: raise exceptions.MalformedError( "Token does not contain required claim {}".format(key) ) # Make sure the token wasn't issued in the future. iat = payload["iat"] # Err on the side of accepting a token that is slightly early to account # for clock skew. earliest = iat - clock_skew_in_seconds if now < earliest: raise exceptions.InvalidValue( "Token used too early, {} < {}. Check that your computer's clock is set correctly.".format( now, iat ) ) # Make sure the token wasn't issued in the past. exp = payload["exp"] # Err on the side of accepting a token that is slightly out of date # to account for clow skew. latest = exp + clock_skew_in_seconds if latest < now: raise exceptions.InvalidValue("Token expired, {} < {}".format(latest, now)) def decode(token, certs=None, verify=True, audience=None, clock_skew_in_seconds=0): """Decode and verify a JWT. Args: token (str): The encoded JWT. certs (Union[str, bytes, Mapping[str, Union[str, bytes]]]): The certificate used to validate the JWT signature. If bytes or string, it must the the public key certificate in PEM format. If a mapping, it must be a mapping of key IDs to public key certificates in PEM format. The mapping must contain the same key ID that's specified in the token's header. verify (bool): Whether to perform signature and claim validation. Verification is done by default. audience (str or list): The audience claim, 'aud', that this JWT should contain. Or a list of audience claims. If None then the JWT's 'aud' parameter is not verified. clock_skew_in_seconds (int): The clock skew used for `iat` and `exp` validation. Returns: Mapping[str, str]: The deserialized JSON payload in the JWT. Raises: google.auth.exceptions.InvalidValue: if value validation failed. google.auth.exceptions.MalformedError: if schema validation failed. """ header, payload, signed_section, signature = _unverified_decode(token) if not verify: return payload # Pluck the key id and algorithm from the header and make sure we have # a verifier that can support it. key_alg = header.get("alg") key_id = header.get("kid") try: verifier_cls = _ALGORITHM_TO_VERIFIER_CLASS[key_alg] except KeyError as exc: if key_alg in _CRYPTOGRAPHY_BASED_ALGORITHMS: raise exceptions.InvalidValue( "The key algorithm {} requires the cryptography package to be installed.".format( key_alg ) ) from exc else: raise exceptions.InvalidValue( "Unsupported signature algorithm {}".format(key_alg) ) from exc # If certs is specified as a dictionary of key IDs to certificates, then # use the certificate identified by the key ID in the token header. if isinstance(certs, Mapping): if key_id: if key_id not in certs: raise exceptions.MalformedError( "Certificate for key id {} not found.".format(key_id) ) certs_to_check = [certs[key_id]] # If there's no key id in the header, check against all of the certs. else: certs_to_check = certs.values() else: certs_to_check = certs # Verify that the signature matches the message. if not crypt.verify_signature( signed_section, signature, certs_to_check, verifier_cls ): raise exceptions.MalformedError("Could not verify token signature.") # Verify the issued at and created times in the payload. _verify_iat_and_exp(payload, clock_skew_in_seconds) # Check audience. if audience is not None: claim_audience = payload.get("aud") if isinstance(audience, str): audience = [audience] if claim_audience not in audience: raise exceptions.InvalidValue( "Token has wrong audience {}, expected one of {}".format( claim_audience, audience ) ) return payload class Credentials( google.auth.credentials.Signing, google.auth.credentials.CredentialsWithQuotaProject ): """Credentials that use a JWT as the bearer token. These credentials require an "audience" claim. This claim identifies the intended recipient of the bearer token. The constructor arguments determine the claims for the JWT that is sent with requests. Usually, you'll construct these credentials with one of the helper constructors as shown in the next section. To create JWT credentials using a Google service account private key JSON file:: audience = 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher' credentials = jwt.Credentials.from_service_account_file( 'service-account.json', audience=audience) If you already have the service account file loaded and parsed:: service_account_info = json.load(open('service_account.json')) credentials = jwt.Credentials.from_service_account_info( service_account_info, audience=audience) Both helper methods pass on arguments to the constructor, so you can specify the JWT claims:: credentials = jwt.Credentials.from_service_account_file( 'service-account.json', audience=audience, additional_claims={'meta': 'data'}) You can also construct the credentials directly if you have a :class:`~google.auth.crypt.Signer` instance:: credentials = jwt.Credentials( signer, issuer='your-issuer', subject='your-subject', audience=audience) The claims are considered immutable. If you want to modify the claims, you can easily create another instance using :meth:`with_claims`:: new_audience = ( 'https://pubsub.googleapis.com/google.pubsub.v1.Subscriber') new_credentials = credentials.with_claims(audience=new_audience) """ def __init__( self, signer, issuer, subject, audience, additional_claims=None, token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS, quota_project_id=None, ): """ Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. issuer (str): The `iss` claim. subject (str): The `sub` claim. audience (str): the `aud` claim. The intended audience for the credentials. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. token_lifetime (int): The amount of time in seconds for which the token is valid. Defaults to 1 hour. quota_project_id (Optional[str]): The project ID used for quota and billing. """ super(Credentials, self).__init__() self._signer = signer self._issuer = issuer self._subject = subject self._audience = audience self._token_lifetime = token_lifetime self._quota_project_id = quota_project_id if additional_claims is None: additional_claims = {} self._additional_claims = additional_claims @classmethod def _from_signer_and_info(cls, signer, info, kwargs): """Creates a Credentials instance from a signer and service account info. Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. info (Mapping[str, str]): The service account info. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: The constructed credentials. Raises: google.auth.exceptions.MalformedError: If the info is not in the expected format. """ kwargs.setdefault("subject", info["client_email"]) kwargs.setdefault("issuer", info["client_email"]) return cls(signer, kwargs) @classmethod def from_service_account_info(cls, info, kwargs): """Creates an Credentials instance from a dictionary. Args: info (Mapping[str, str]): The service account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: The constructed credentials. Raises: google.auth.exceptions.MalformedError: If the info is not in the expected format. """ signer = _service_account_info.from_dict(info, require=["client_email"]) return cls._from_signer_and_info(signer, info, kwargs) @classmethod def from_service_account_file(cls, filename, kwargs): """Creates a Credentials instance from a service account .json file in Google format. Args: filename (str): The path to the service account .json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: The constructed credentials. """ info, signer = _service_account_info.from_filename( filename, require=["client_email"] ) return cls._from_signer_and_info(signer, info, kwargs) @classmethod def from_signing_credentials(cls, credentials, audience, kwargs): """Creates a new :class:`google.auth.jwt.Credentials` instance from an existing :class:`google.auth.credentials.Signing` instance. The new instance will use the same signer as the existing instance and will use the existing instance's signer email as the issuer and subject by default. Example:: svc_creds = service_account.Credentials.from_service_account_file( 'service_account.json') audience = ( 'https://pubsub.googleapis.com/google.pubsub.v1.Publisher') jwt_creds = jwt.Credentials.from_signing_credentials( svc_creds, audience=audience) Args: credentials (google.auth.credentials.Signing): The credentials to use to construct the new credentials. audience (str): the `aud` claim. The intended audience for the credentials. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: A new Credentials instance. """ kwargs.setdefault("issuer", credentials.signer_email) kwargs.setdefault("subject", credentials.signer_email) return cls(credentials.signer, audience=audience, kwargs) def with_claims( self, issuer=None, subject=None, audience=None, additional_claims=None ): """Returns a copy of these credentials with modified claims. Args: issuer (str): The `iss` claim. If unspecified the current issuer claim will be used. subject (str): The `sub` claim. If unspecified the current subject claim will be used. audience (str): the `aud` claim. If unspecified the current audience claim will be used. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. This will be merged with the current additional claims. Returns: google.auth.jwt.Credentials: A new credentials instance. """ new_additional_claims = copy.deepcopy(self._additional_claims) new_additional_claims.update(additional_claims or {}) return self.__class__( self._signer, issuer=issuer if issuer is not None else self._issuer, subject=subject if subject is not None else self._subject, audience=audience if audience is not None else self._audience, additional_claims=new_additional_claims, quota_project_id=self._quota_project_id, ) @_helpers.copy_docstring(google.auth.credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): return self.__class__( self._signer, issuer=self._issuer, subject=self._subject, audience=self._audience, additional_claims=self._additional_claims, quota_project_id=quota_project_id, ) def _make_jwt(self): """Make a signed JWT. Returns: Tuple[bytes, datetime]: The encoded JWT and the expiration. """ now = _helpers.utcnow() lifetime = datetime.timedelta(seconds=self._token_lifetime) expiry = now + lifetime payload = { "iss": self._issuer, "sub": self._subject, "iat": _helpers.datetime_to_secs(now), "exp": _helpers.datetime_to_secs(expiry), } if self._audience: payload["aud"] = self._audience payload.update(self._additional_claims) jwt = encode(self._signer, payload) return jwt, expiry def refresh(self, request): """Refreshes the access token. Args: request (Any): Unused. """ # pylint: disable=unused-argument # (pylint doesn't correctly recognize overridden methods.) self.token, self.expiry = self._make_jwt() @_helpers.copy_docstring(google.auth.credentials.Signing) def sign_bytes(self, message): return self._signer.sign(message) @property # type: ignore @_helpers.copy_docstring(google.auth.credentials.Signing) def signer_email(self): return self._issuer @property # type: ignore @_helpers.copy_docstring(google.auth.credentials.Signing) def signer(self): return self._signer @property # type: ignore def additional_claims(self): """ Additional claims the JWT object was created with.""" return self._additional_claims class OnDemandCredentials( google.auth.credentials.Signing, google.auth.credentials.CredentialsWithQuotaProject ): """On-demand JWT credentials. Like :class:`Credentials`, this class uses a JWT as the bearer token for authentication. However, this class does not require the audience at construction time. Instead, it will generate a new token on-demand for each request using the request URI as the audience. It caches tokens so that multiple requests to the same URI do not incur the overhead of generating a new token every time. This behavior is especially useful for `gRPC`_ clients. A gRPC service may have multiple audience and gRPC clients may not know all of the audiences required for accessing a particular service. With these credentials, no knowledge of the audiences is required ahead of time. .. _grpc: http://www.grpc.io/ """ def __init__( self, signer, issuer, subject, additional_claims=None, token_lifetime=_DEFAULT_TOKEN_LIFETIME_SECS, max_cache_size=_DEFAULT_MAX_CACHE_SIZE, quota_project_id=None, ): """ Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. issuer (str): The `iss` claim. subject (str): The `sub` claim. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. token_lifetime (int): The amount of time in seconds for which the token is valid. Defaults to 1 hour. max_cache_size (int): The maximum number of JWT tokens to keep in cache. Tokens are cached using :class:`cachetools.LRUCache`. quota_project_id (Optional[str]): The project ID used for quota and billing. """ super(OnDemandCredentials, self).__init__() self._signer = signer self._issuer = issuer self._subject = subject self._token_lifetime = token_lifetime self._quota_project_id = quota_project_id if additional_claims is None: additional_claims = {} self._additional_claims = additional_claims self._cache = cachetools.LRUCache(maxsize=max_cache_size) @classmethod def _from_signer_and_info(cls, signer, info, kwargs): """Creates an OnDemandCredentials instance from a signer and service account info. Args: signer (google.auth.crypt.Signer): The signer used to sign JWTs. info (Mapping[str, str]): The service account info. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.OnDemandCredentials: The constructed credentials. Raises: google.auth.exceptions.MalformedError: If the info is not in the expected format. """ kwargs.setdefault("subject", info["client_email"]) kwargs.setdefault("issuer", info["client_email"]) return cls(signer, kwargs) @classmethod def from_service_account_info(cls, info, kwargs): """Creates an OnDemandCredentials instance from a dictionary. Args: info (Mapping[str, str]): The service account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.OnDemandCredentials: The constructed credentials. Raises: google.auth.exceptions.MalformedError: If the info is not in the expected format. """ signer = _service_account_info.from_dict(info, require=["client_email"]) return cls._from_signer_and_info(signer, info, kwargs) @classmethod def from_service_account_file(cls, filename, kwargs): """Creates an OnDemandCredentials instance from a service account .json file in Google format. Args: filename (str): The path to the service account .json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.OnDemandCredentials: The constructed credentials. """ info, signer = _service_account_info.from_filename( filename, require=["client_email"] ) return cls._from_signer_and_info(signer, info, kwargs) @classmethod def from_signing_credentials(cls, credentials, kwargs): """Creates a new :class:`google.auth.jwt.OnDemandCredentials` instance from an existing :class:`google.auth.credentials.Signing` instance. The new instance will use the same signer as the existing instance and will use the existing instance's signer email as the issuer and subject by default. Example:: svc_creds = service_account.Credentials.from_service_account_file( 'service_account.json') jwt_creds = jwt.OnDemandCredentials.from_signing_credentials( svc_creds) Args: credentials (google.auth.credentials.Signing): The credentials to use to construct the new credentials. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.jwt.Credentials: A new Credentials instance. """ kwargs.setdefault("issuer", credentials.signer_email) kwargs.setdefault("subject", credentials.signer_email) return cls(credentials.signer, kwargs) def with_claims(self, issuer=None, subject=None, additional_claims=None): """Returns a copy of these credentials with modified claims. Args: issuer (str): The `iss` claim. If unspecified the current issuer claim will be used. subject (str): The `sub` claim. If unspecified the current subject claim will be used. additional_claims (Mapping[str, str]): Any additional claims for the JWT payload. This will be merged with the current additional claims. Returns: google.auth.jwt.OnDemandCredentials: A new credentials instance. """ new_additional_claims = copy.deepcopy(self._additional_claims) new_additional_claims.update(additional_claims or {}) return self.__class__( self._signer, issuer=issuer if issuer is not None else self._issuer, subject=subject if subject is not None else self._subject, additional_claims=new_additional_claims, max_cache_size=self._cache.maxsize, quota_project_id=self._quota_project_id, ) @_helpers.copy_docstring(google.auth.credentials.CredentialsWithQuotaProject) def with_quota_project(self, quota_project_id): return self.__class__( self._signer, issuer=self._issuer, subject=self._subject, additional_claims=self._additional_claims, max_cache_size=self._cache.maxsize, quota_project_id=quota_project_id, ) @property def valid(self): """Checks the validity of the credentials. These credentials are always valid because it generates tokens on demand. """ return True def _make_jwt_for_audience(self, audience): """Make a new JWT for the given audience. Args: audience (str): The intended audience. Returns: Tuple[bytes, datetime]: The encoded JWT and the expiration. """ now = _helpers.utcnow() lifetime = datetime.timedelta(seconds=self._token_lifetime) expiry = now + lifetime payload = { "iss": self._issuer, "sub": self._subject, "iat": _helpers.datetime_to_secs(now), "exp": _helpers.datetime_to_secs(expiry), "aud": audience, } payload.update(self._additional_claims) jwt = encode(self._signer, payload) return jwt, expiry def _get_jwt_for_audience(self, audience): """Get a JWT For a given audience. If there is already an existing, non-expired token in the cache for the audience, that token is used. Otherwise, a new token will be created. Args: audience (str): The intended audience. Returns: bytes: The encoded JWT. """ token, expiry = self._cache.get(audience, (None, None)) if token is None or expiry < _helpers.utcnow(): token, expiry = self._make_jwt_for_audience(audience) self._cache[audience] = token, expiry return token def refresh(self, request): """Raises an exception, these credentials can not be directly refreshed. Args: request (Any): Unused. Raises: google.auth.RefreshError """ # pylint: disable=unused-argument # (pylint doesn't correctly recognize overridden methods.) raise exceptions.RefreshError( "OnDemandCredentials can not be directly refreshed." ) def before_request(self, request, method, url, headers): """Performs credential-specific before request logic. Args: request (Any): Unused. JWT credentials do not need to make an HTTP request to refresh. method (str): The request's HTTP method. url (str): The request's URI. This is used as the audience claim when generating the JWT. headers (Mapping): The request's headers. """ # pylint: disable=unused-argument # (pylint doesn't correctly recognize overridden methods.) parts = urllib.parse.urlsplit(url) # Strip query string and fragment audience = urllib.parse.urlunsplit( (parts.scheme, parts.netloc, parts.path, "", "") ) token = self._get_jwt_for_audience(audience) self.apply(headers, token=token) @_helpers.copy_docstring(google.auth.credentials.Signing) def sign_bytes(self, message): return self._signer.sign(message) @property # type: ignore @_helpers.copy_docstring(google.auth.credentials.Signing) def signer_email(self): return self._issuer @property # type: ignore @_helpers.copy_docstring(google.auth.credentials.Signing) def signer(self): return self._signer	30.37 KB	2025-04-14 17:36
📄 metrics.py # Copyright 2023 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ We use x-goog-api-client header to report metrics. This module provides the constants and helper methods to construct x-goog-api-client header. """ import platform from google.auth import version API_CLIENT_HEADER = "x-goog-api-client" # BYOID Specific consts BYOID_HEADER_SECTION = "google-byoid-sdk" # Auth request type REQUEST_TYPE_ACCESS_TOKEN = "auth-request-type/at" REQUEST_TYPE_ID_TOKEN = "auth-request-type/it" REQUEST_TYPE_MDS_PING = "auth-request-type/mds" REQUEST_TYPE_REAUTH_START = "auth-request-type/re-start" REQUEST_TYPE_REAUTH_CONTINUE = "auth-request-type/re-cont" # Credential type CRED_TYPE_USER = "cred-type/u" CRED_TYPE_SA_ASSERTION = "cred-type/sa" CRED_TYPE_SA_JWT = "cred-type/jwt" CRED_TYPE_SA_MDS = "cred-type/mds" CRED_TYPE_SA_IMPERSONATE = "cred-type/imp" # Versions def python_and_auth_lib_version(): return "gl-python/{} auth/{}".format(platform.python_version(), version.__version__) # Token request metric header values # x-goog-api-client header value for access token request via metadata server. # Example: "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/mds" def token_request_access_token_mds(): return "{} {} {}".format( python_and_auth_lib_version(), REQUEST_TYPE_ACCESS_TOKEN, CRED_TYPE_SA_MDS ) # x-goog-api-client header value for ID token request via metadata server. # Example: "gl-python/3.7 auth/1.1 auth-request-type/it cred-type/mds" def token_request_id_token_mds(): return "{} {} {}".format( python_and_auth_lib_version(), REQUEST_TYPE_ID_TOKEN, CRED_TYPE_SA_MDS ) # x-goog-api-client header value for impersonated credentials access token request. # Example: "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/imp" def token_request_access_token_impersonate(): return "{} {} {}".format( python_and_auth_lib_version(), REQUEST_TYPE_ACCESS_TOKEN, CRED_TYPE_SA_IMPERSONATE, ) # x-goog-api-client header value for impersonated credentials ID token request. # Example: "gl-python/3.7 auth/1.1 auth-request-type/it cred-type/imp" def token_request_id_token_impersonate(): return "{} {} {}".format( python_and_auth_lib_version(), REQUEST_TYPE_ID_TOKEN, CRED_TYPE_SA_IMPERSONATE ) # x-goog-api-client header value for service account credentials access token # request (assertion flow). # Example: "gl-python/3.7 auth/1.1 auth-request-type/at cred-type/sa" def token_request_access_token_sa_assertion(): return "{} {} {}".format( python_and_auth_lib_version(), REQUEST_TYPE_ACCESS_TOKEN, CRED_TYPE_SA_ASSERTION ) # x-goog-api-client header value for service account credentials ID token # request (assertion flow). # Example: "gl-python/3.7 auth/1.1 auth-request-type/it cred-type/sa" def token_request_id_token_sa_assertion(): return "{} {} {}".format( python_and_auth_lib_version(), REQUEST_TYPE_ID_TOKEN, CRED_TYPE_SA_ASSERTION ) # x-goog-api-client header value for user credentials token request. # Example: "gl-python/3.7 auth/1.1 cred-type/u" def token_request_user(): return "{} {}".format(python_and_auth_lib_version(), CRED_TYPE_USER) # Miscellenous metrics # x-goog-api-client header value for metadata server ping. # Example: "gl-python/3.7 auth/1.1 auth-request-type/mds" def mds_ping(): return "{} {}".format(python_and_auth_lib_version(), REQUEST_TYPE_MDS_PING) # x-goog-api-client header value for reauth start endpoint calls. # Example: "gl-python/3.7 auth/1.1 auth-request-type/re-start" def reauth_start(): return "{} {}".format(python_and_auth_lib_version(), REQUEST_TYPE_REAUTH_START) # x-goog-api-client header value for reauth continue endpoint calls. # Example: "gl-python/3.7 auth/1.1 cred-type/re-cont" def reauth_continue(): return "{} {}".format(python_and_auth_lib_version(), REQUEST_TYPE_REAUTH_CONTINUE) # x-goog-api-client header value for BYOID calls to the Security Token Service exchange token endpoint. # Example: "gl-python/3.7 auth/1.1 google-byoid-sdk source/aws sa-impersonation/true sa-impersonation/true" def byoid_metrics_header(metrics_options): header = "{} {}".format(python_and_auth_lib_version(), BYOID_HEADER_SECTION) for key, value in metrics_options.items(): header = "{} {}/{}".format(header, key, value) return header def add_metric_header(headers, metric_header_value): """Add x-goog-api-client header with the given value. Args: headers (Mapping[str, str]): The headers to which we will add the metric header. metric_header_value (Optional[str]): If value is None, do nothing; if headers already has a x-goog-api-client header, append the value to the existing header; otherwise add a new x-goog-api-client header with the given value. """ if not metric_header_value: return if API_CLIENT_HEADER not in headers: headers[API_CLIENT_HEADER] = metric_header_value else: headers[API_CLIENT_HEADER] += " " + metric_header_value	5.48 KB	2025-04-14 17:36
📄 pluggable.py # Copyright 2022 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """Pluggable Credentials. Pluggable Credentials are initialized using external_account arguments which are typically loaded from third-party executables. Unlike other credentials that can be initialized with a list of explicit arguments, secrets or credentials, external account clients use the environment and hints/guidelines provided by the external_account JSON file to retrieve credentials and exchange them for Google access tokens. Example credential_source for pluggable credential: { "executable": { "command": "/path/to/get/credentials.sh --arg1=value1 --arg2=value2", "timeout_millis": 5000, "output_file": "/path/to/generated/cached/credentials" } } """ try: from collections.abc import Mapping # Python 2.7 compatibility except ImportError: # pragma: NO COVER from collections import Mapping # type: ignore import json import os import subprocess import sys import time from google.auth import _helpers from google.auth import exceptions from google.auth import external_account # The max supported executable spec version. EXECUTABLE_SUPPORTED_MAX_VERSION = 1 EXECUTABLE_TIMEOUT_MILLIS_DEFAULT = 30 * 1000 # 30 seconds EXECUTABLE_TIMEOUT_MILLIS_LOWER_BOUND = 5 * 1000 # 5 seconds EXECUTABLE_TIMEOUT_MILLIS_UPPER_BOUND = 120 * 1000 # 2 minutes EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_LOWER_BOUND = 30 * 1000 # 30 seconds EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_UPPER_BOUND = 30 * 60 * 1000 # 30 minutes class Credentials(external_account.Credentials): """External account credentials sourced from executables.""" def __init__( self, audience, subject_token_type, token_url, credential_source, args, kwargs ): """Instantiates an external account credentials object from a executables. Args: audience (str): The STS audience field. subject_token_type (str): The subject token type. token_url (str): The STS endpoint URL. credential_source (Mapping): The credential source dictionary used to provide instructions on how to retrieve external credential to be exchanged for Google access tokens. Example credential_source for pluggable credential: { "executable": { "command": "/path/to/get/credentials.sh --arg1=value1 --arg2=value2", "timeout_millis": 5000, "output_file": "/path/to/generated/cached/credentials" } } args (List): Optional positional arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method. kwargs (Mapping): Optional keyword arguments passed into the underlying :meth:`~external_account.Credentials.__init__` method. Raises: google.auth.exceptions.RefreshError: If an error is encountered during access token retrieval logic. google.auth.exceptions.InvalidValue: For invalid parameters. google.auth.exceptions.MalformedError: For invalid parameters. .. note:: Typically one of the helper constructors :meth:`from_file` or :meth:`from_info` are used instead of calling the constructor directly. """ self.interactive = kwargs.pop("interactive", False) super(Credentials, self).__init__( audience=audience, subject_token_type=subject_token_type, token_url=token_url, credential_source=credential_source, args, kwargs ) if not isinstance(credential_source, Mapping): self._credential_source_executable = None raise exceptions.MalformedError( "Missing credential_source. The credential_source is not a dict." ) self._credential_source_executable = credential_source.get("executable") if not self._credential_source_executable: raise exceptions.MalformedError( "Missing credential_source. An 'executable' must be provided." ) self._credential_source_executable_command = self._credential_source_executable.get( "command" ) self._credential_source_executable_timeout_millis = self._credential_source_executable.get( "timeout_millis" ) self._credential_source_executable_interactive_timeout_millis = self._credential_source_executable.get( "interactive_timeout_millis" ) self._credential_source_executable_output_file = self._credential_source_executable.get( "output_file" ) # Dummy value. This variable is only used via injection, not exposed to ctor self._tokeninfo_username = "" if not self._credential_source_executable_command: raise exceptions.MalformedError( "Missing command field. Executable command must be provided." ) if not self._credential_source_executable_timeout_millis: self._credential_source_executable_timeout_millis = ( EXECUTABLE_TIMEOUT_MILLIS_DEFAULT ) elif ( self._credential_source_executable_timeout_millis < EXECUTABLE_TIMEOUT_MILLIS_LOWER_BOUND or self._credential_source_executable_timeout_millis > EXECUTABLE_TIMEOUT_MILLIS_UPPER_BOUND ): raise exceptions.InvalidValue("Timeout must be between 5 and 120 seconds.") if self._credential_source_executable_interactive_timeout_millis: if ( self._credential_source_executable_interactive_timeout_millis < EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_LOWER_BOUND or self._credential_source_executable_interactive_timeout_millis > EXECUTABLE_INTERACTIVE_TIMEOUT_MILLIS_UPPER_BOUND ): raise exceptions.InvalidValue( "Interactive timeout must be between 30 seconds and 30 minutes." ) @_helpers.copy_docstring(external_account.Credentials) def retrieve_subject_token(self, request): self._validate_running_mode() # Check output file. if self._credential_source_executable_output_file is not None: try: with open( self._credential_source_executable_output_file, encoding="utf-8" ) as output_file: response = json.load(output_file) except Exception: pass else: try: # If the cached response is expired, _parse_subject_token will raise an error which will be ignored and we will call the executable again. subject_token = self._parse_subject_token(response) if ( "expiration_time" not in response ): # Always treat missing expiration_time as expired and proceed to executable run. raise exceptions.RefreshError except (exceptions.MalformedError, exceptions.InvalidValue): raise except exceptions.RefreshError: pass else: return subject_token if not _helpers.is_python_3(): raise exceptions.RefreshError( "Pluggable auth is only supported for python 3.7+" ) # Inject env vars. env = os.environ.copy() self._inject_env_variables(env) env["GOOGLE_EXTERNAL_ACCOUNT_REVOKE"] = "0" # Run executable. exe_timeout = ( self._credential_source_executable_interactive_timeout_millis / 1000 if self.interactive else self._credential_source_executable_timeout_millis / 1000 ) exe_stdin = sys.stdin if self.interactive else None exe_stdout = sys.stdout if self.interactive else subprocess.PIPE exe_stderr = sys.stdout if self.interactive else subprocess.STDOUT result = subprocess.run( self._credential_source_executable_command.split(), timeout=exe_timeout, stdin=exe_stdin, stdout=exe_stdout, stderr=exe_stderr, env=env, ) if result.returncode != 0: raise exceptions.RefreshError( "Executable exited with non-zero return code {}. Error: {}".format( result.returncode, result.stdout ) ) # Handle executable output. response = json.loads(result.stdout.decode("utf-8")) if result.stdout else None if not response and self._credential_source_executable_output_file is not None: response = json.load( open(self._credential_source_executable_output_file, encoding="utf-8") ) subject_token = self._parse_subject_token(response) return subject_token def revoke(self, request): """Revokes the subject token using the credential_source object. Args: request (google.auth.transport.Request): A callable used to make HTTP requests. Raises: google.auth.exceptions.RefreshError: If the executable revocation not properly executed. """ if not self.interactive: raise exceptions.InvalidValue( "Revoke is only enabled under interactive mode." ) self._validate_running_mode() if not _helpers.is_python_3(): raise exceptions.RefreshError( "Pluggable auth is only supported for python 3.7+" ) # Inject variables env = os.environ.copy() self._inject_env_variables(env) env["GOOGLE_EXTERNAL_ACCOUNT_REVOKE"] = "1" # Run executable result = subprocess.run( self._credential_source_executable_command.split(), timeout=self._credential_source_executable_interactive_timeout_millis / 1000, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, ) if result.returncode != 0: raise exceptions.RefreshError( "Auth revoke failed on executable. Exit with non-zero return code {}. Error: {}".format( result.returncode, result.stdout ) ) response = json.loads(result.stdout.decode("utf-8")) self._validate_revoke_response(response) @property def external_account_id(self): """Returns the external account identifier. When service account impersonation is used the identifier is the service account email. Without service account impersonation, this returns None, unless it is being used by the Google Cloud CLI which populates this field. """ return self.service_account_email or self._tokeninfo_username @classmethod def from_info(cls, info, kwargs): """Creates a Pluggable Credentials instance from parsed external account info. Args: info (Mapping[str, str]): The Pluggable external account info in Google format. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.pluggable.Credentials: The constructed credentials. Raises: google.auth.exceptions.InvalidValue: For invalid parameters. google.auth.exceptions.MalformedError: For invalid parameters. """ return super(Credentials, cls).from_info(info, kwargs) @classmethod def from_file(cls, filename, kwargs): """Creates an Pluggable Credentials instance from an external account json file. Args: filename (str): The path to the Pluggable external account json file. kwargs: Additional arguments to pass to the constructor. Returns: google.auth.pluggable.Credentials: The constructed credentials. """ return super(Credentials, cls).from_file(filename, **kwargs) def _inject_env_variables(self, env): env["GOOGLE_EXTERNAL_ACCOUNT_AUDIENCE"] = self._audience env["GOOGLE_EXTERNAL_ACCOUNT_TOKEN_TYPE"] = self._subject_token_type env["GOOGLE_EXTERNAL_ACCOUNT_ID"] = self.external_account_id env["GOOGLE_EXTERNAL_ACCOUNT_INTERACTIVE"] = "1" if self.interactive else "0" if self._service_account_impersonation_url is not None: env[ "GOOGLE_EXTERNAL_ACCOUNT_IMPERSONATED_EMAIL" ] = self.service_account_email if self._credential_source_executable_output_file is not None: env[ "GOOGLE_EXTERNAL_ACCOUNT_OUTPUT_FILE" ] = self._credential_source_executable_output_file def _parse_subject_token(self, response): self._validate_response_schema(response) if not response["success"]: if "code" not in response or "message" not in response: raise exceptions.MalformedError( "Error code and message fields are required in the response." ) raise exceptions.RefreshError( "Executable returned unsuccessful response: code: {}, message: {}.".format( response["code"], response["message"] ) ) if "expiration_time" in response and response["expiration_time"] < time.time(): raise exceptions.RefreshError( "The token returned by the executable is expired." ) if "token_type" not in response: raise exceptions.MalformedError( "The executable response is missing the token_type field." ) if ( response["token_type"] == "urn:ietf:params:oauth:token-type:jwt" or response["token_type"] == "urn:ietf:params:oauth:token-type:id_token" ): # OIDC return response["id_token"] elif response["token_type"] == "urn:ietf:params:oauth:token-type:saml2": # SAML return response["saml_response"] else: raise exceptions.RefreshError("Executable returned unsupported token type.") def _validate_revoke_response(self, response): self._validate_response_schema(response) if not response["success"]: raise exceptions.RefreshError("Revoke failed with unsuccessful response.") def _validate_response_schema(self, response): if "version" not in response: raise exceptions.MalformedError( "The executable response is missing the version field." ) if response["version"] > EXECUTABLE_SUPPORTED_MAX_VERSION: raise exceptions.RefreshError( "Executable returned unsupported version {}.".format( response["version"] ) ) if "success" not in response: raise exceptions.MalformedError( "The executable response is missing the success field." ) def _validate_running_mode(self): env_allow_executables = os.environ.get( "GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES" ) if env_allow_executables != "1": raise exceptions.MalformedError( "Executables need to be explicitly allowed (set GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES to '1') to run." ) if self.interactive and not self._credential_source_executable_output_file: raise exceptions.MalformedError( "An output_file must be specified in the credential configuration for interactive mode." ) if ( self.interactive and not self._credential_source_executable_interactive_timeout_millis ): raise exceptions.InvalidOperation( "Interactive mode cannot run without an interactive timeout." ) if self.interactive and not self.is_workforce_pool: raise exceptions.InvalidValue( "Interactive mode is only enabled for workforce pool." ) def _create_default_metrics_options(self): metrics_options = super(Credentials, self)._create_default_metrics_options() metrics_options["source"] = "executable" return metrics_options	16.92 KB	2025-04-14 17:36
📄 py.typed # Marker file for PEP 561. # The google-auth package uses inline types.	74 B	2025-04-14 17:36
📁 transport	-	2025-06-25 08:26
📄 version.py # Copyright 2021 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. __version__ = "2.39.0"	598 B	2025-04-14 17:36

Edit File:

Terminal