Revert "feat: workload identity federation support (#686)" (#691)
This reverts commit 5dcd2b1bdd9d21522636d959cffc49ee29dda88f.
diff --git a/google/auth/_default.py b/google/auth/_default.py
index 836c339..3b8c281 100644
--- a/google/auth/_default.py
+++ b/google/auth/_default.py
@@ -34,8 +34,7 @@
# Valid types accepted for file-based credentials.
_AUTHORIZED_USER_TYPE = "authorized_user"
_SERVICE_ACCOUNT_TYPE = "service_account"
-_EXTERNAL_ACCOUNT_TYPE = "external_account"
-_VALID_TYPES = (_AUTHORIZED_USER_TYPE, _SERVICE_ACCOUNT_TYPE, _EXTERNAL_ACCOUNT_TYPE)
+_VALID_TYPES = (_AUTHORIZED_USER_TYPE, _SERVICE_ACCOUNT_TYPE)
# Help message when no credentials can be found.
_HELP_MESSAGE = """\
@@ -71,12 +70,12 @@
def load_credentials_from_file(
- filename, scopes=None, default_scopes=None, quota_project_id=None, request=None
+ filename, scopes=None, default_scopes=None, quota_project_id=None
):
"""Loads Google credentials from a file.
- The credentials file must be a service account key, stored authorized
- user credentials or external account credentials.
+ The credentials file must be a service account key or stored authorized
+ user credentials.
Args:
filename (str): The full path to the credentials file.
@@ -86,18 +85,12 @@
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.
+ 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. External account credentials project
- IDs may not always be determined.
+ have the project ID information.
Raises:
google.auth.exceptions.DefaultCredentialsError: if the file is in the
@@ -153,18 +146,6 @@
credentials = credentials.with_quota_project(quota_project_id)
return credentials, info.get("project_id")
- elif credential_type == _EXTERNAL_ACCOUNT_TYPE:
- credentials, project_id = _get_external_account_credentials(
- info,
- filename,
- scopes=scopes,
- default_scopes=default_scopes,
- request=request,
- )
- if quota_project_id:
- credentials = credentials.with_quota_project(quota_project_id)
- return credentials, project_id
-
else:
raise exceptions.DefaultCredentialsError(
"The file {file} does not have a valid type. "
@@ -195,28 +176,9 @@
return credentials, project_id
-def _get_explicit_environ_credentials(request=None, scopes=None, default_scopes=None):
+def _get_explicit_environ_credentials():
"""Gets credentials from the GOOGLE_APPLICATION_CREDENTIALS environment
- variable.
-
- Args:
- 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.
- 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.
-
- Returns:
- Tuple[Optional[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.
- """
+ variable."""
explicit_file = os.environ.get(environment_vars.CREDENTIALS)
_LOGGER.debug(
@@ -225,11 +187,7 @@
if explicit_file is not None:
credentials, project_id = load_credentials_from_file(
- os.environ[environment_vars.CREDENTIALS],
- scopes=scopes,
- default_scopes=default_scopes,
- quota_project_id=None,
- request=request,
+ os.environ[environment_vars.CREDENTIALS]
)
return credentials, project_id
@@ -294,65 +252,6 @@
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 2 types of external_account credentials.
- try:
- # 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
- )
- except ValueError:
- 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:
- request = google.auth.transport.requests.Request()
-
- return credentials, credentials.get_project_id(request=request)
-
-
def default(scopes=None, request=None, quota_project_id=None, default_scopes=None):
"""Gets the default credentials for the current environment.
@@ -366,15 +265,6 @@
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.
2. If the `Google Cloud SDK`_ is installed and has application default
credentials set they are loaded and returned.
@@ -420,15 +310,11 @@
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
+ 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.
default_scopes (Optional[Sequence[str]]): Default scopes passed by a
Google client library. Use 'scopes' for user-defined scopes.
@@ -450,9 +336,7 @@
)
checkers = (
- lambda: _get_explicit_environ_credentials(
- request=request, scopes=scopes, default_scopes=default_scopes
- ),
+ _get_explicit_environ_credentials,
_get_gcloud_sdk_credentials,
_get_gae_credentials,
lambda: _get_gce_credentials(request),
diff --git a/google/auth/aws.py b/google/auth/aws.py
deleted file mode 100644
index b362dd3..0000000
--- a/google/auth/aws.py
+++ /dev/null
@@ -1,714 +0,0 @@
-# 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.
-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.
-
-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 hashlib
-import hmac
-import io
-import json
-import os
-import re
-
-from six.moves import http_client
-from six.moves import urllib
-
-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"
-
-
-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 (Mapping[str, str]): A dictionary containing
- 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.
- """
- # Get AWS credentials.
- access_key = aws_security_credentials.get("access_key_id")
- secret_key = aws_security_credentials.get("secret_access_key")
- security_token = aws_security_credentials.get("security_token")
-
- additional_headers = additional_headers or {}
-
- uri = urllib.parse.urlparse(url)
- # Validate provided URL.
- if not uri.hostname or uri.scheme != "https":
- raise ValueError("Invalid AWS service URL")
-
- header_map = _generate_authentication_header_map(
- host=uri.hostname,
- canonical_uri=os.path.normpath(uri.path or "/"),
- canonical_querystring=_get_canonical_querystring(uri.query),
- method=method,
- region=self._region_name,
- access_key=access_key,
- secret_key=secret_key,
- security_token=security_token,
- 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 security_token is not None:
- headers[_AWS_SECURITY_TOKEN_HEADER] = security_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,
- access_key,
- secret_key,
- security_token,
- 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.
- access_key (str): The AWS access key ID.
- secret_key (str): The AWS secret access key.
- security_token (Optional[str]): The AWS security session token. This is
- available for temporary sessions.
- 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 security_token is not None:
- full_headers[_AWS_SECURITY_TOKEN_HEADER] = security_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(secret_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, access_key, 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
-
-
-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,
- credential_source=None,
- service_account_impersonation_url=None,
- client_id=None,
- client_secret=None,
- quota_project_id=None,
- scopes=None,
- default_scopes=None,
- ):
- """Instantiates an AWS workload external account credentials object.
-
- 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.
- service_account_impersonation_url (Optional[str]): The optional
- service account impersonation getAccessToken URL.
- client_id (Optional[str]): The optional client ID.
- client_secret (Optional[str]): The optional client secret.
- 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.
-
- 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,
- service_account_impersonation_url=service_account_impersonation_url,
- client_id=client_id,
- client_secret=client_secret,
- quota_project_id=quota_project_id,
- scopes=scopes,
- default_scopes=default_scopes,
- )
- credential_source = credential_source or {}
- self._environment_id = credential_source.get("environment_id") or ""
- self._region_url = credential_source.get("region_url")
- self._security_credentials_url = credential_source.get("url")
- self._cred_verification_url = credential_source.get(
- "regional_cred_verification_url"
- )
- self._region = None
- self._request_signer = None
- self._target_resource = audience
-
- # Get the environment ID. Currently, only one version supported (v1).
- matches = re.match(r"^(aws)([\d]+)$", self._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 ValueError("No valid AWS 'credential_source' provided")
- elif int(env_version or "") != 1:
- raise ValueError(
- "aws version '{}' is not supported in the current build.".format(
- env_version
- )
- )
-
- 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 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._get_region(request, self._region_url)
- self._request_signer = RequestSigner(self._region)
-
- # Retrieve the AWS security credentials needed to generate the signed
- # request.
- aws_security_credentials = self._get_security_credentials(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.
- # Keeping inner keys in sorted order makes testing easier for Python
- # versions <=3.5 as the stringified JSON string would have a predictable
- # key order.
- 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 sorted(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 _get_region(self, request, url):
- """Retrieves the current AWS region from either the AWS_REGION
- environment variable or from the AWS metadata server.
-
- Args:
- request (google.auth.transport.Request): A callable used to make
- HTTP requests.
- url (str): The AWS metadata server region URL.
-
- Returns:
- str: The current AWS region.
-
- Raises:
- google.auth.exceptions.RefreshError: If an error occurs while
- retrieving the AWS region.
- """
- # 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
-
- if not self._region_url:
- raise exceptions.RefreshError("Unable to determine AWS region")
- response = request(url=self._region_url, method="GET")
-
- # 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 AWS region", 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_security_credentials(self, request):
- """Retrieves the AWS security credentials required for signing AWS
- requests from either the AWS security credentials environment variables
- or from the AWS metadata server.
-
- Args:
- request (google.auth.transport.Request): A callable used to make
- HTTP requests.
-
- Returns:
- Mapping[str, str]: The AWS security credentials dictionary object.
-
- Raises:
- google.auth.exceptions.RefreshError: If an error occurs while
- retrieving the AWS security credentials.
- """
-
- # 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 {
- "access_key_id": env_aws_access_key_id,
- "secret_access_key": env_aws_secret_access_key,
- "security_token": env_aws_session_token,
- }
-
- # Get role name.
- role_name = self._get_metadata_role_name(request)
-
- # Get security credentials.
- credentials = self._get_metadata_security_credentials(request, role_name)
-
- return {
- "access_key_id": credentials.get("AccessKeyId"),
- "secret_access_key": credentials.get("SecretAccessKey"),
- "security_token": credentials.get("Token"),
- }
-
- def _get_metadata_security_credentials(self, request, role_name):
- """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.
-
- 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"}
- 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", response_body
- )
-
- credentials_response = json.loads(response_body)
-
- return credentials_response
-
- def _get_metadata_role_name(self, request):
- """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.
-
- 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"
- )
- response = request(url=self._security_credentials_url, method="GET")
-
- # 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", response_body
- )
-
- return response_body
-
- @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.
- """
- return cls(
- audience=info.get("audience"),
- subject_token_type=info.get("subject_token_type"),
- token_url=info.get("token_url"),
- service_account_impersonation_url=info.get(
- "service_account_impersonation_url"
- ),
- 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"),
- **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.
- """
- with io.open(filename, "r", encoding="utf-8") as json_file:
- data = json.load(json_file)
- return cls.from_info(data, **kwargs)
diff --git a/google/auth/environment_vars.py b/google/auth/environment_vars.py
index 416bab0..46a8926 100644
--- a/google/auth/environment_vars.py
+++ b/google/auth/environment_vars.py
@@ -59,13 +59,3 @@
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."""
-
-# 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"
diff --git a/google/auth/exceptions.py b/google/auth/exceptions.py
index b6f686b..da06d86 100644
--- a/google/auth/exceptions.py
+++ b/google/auth/exceptions.py
@@ -43,8 +43,3 @@
class ClientCertError(GoogleAuthError):
"""Used to indicate that client certificate is missing or invalid."""
-
-
-class OAuthError(GoogleAuthError):
- """Used to indicate an error occurred during an OAuth related HTTP
- request."""
diff --git a/google/auth/external_account.py b/google/auth/external_account.py
deleted file mode 100644
index 0429ee0..0000000
--- a/google/auth/external_account.py
+++ /dev/null
@@ -1,305 +0,0 @@
-# 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 datetime
-import json
-
-import six
-
-from google.auth import _helpers
-from google.auth import credentials
-from google.auth import exceptions
-from google.auth import impersonated_credentials
-from google.oauth2 import sts
-from google.oauth2 import utils
-
-# 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/"
-
-
-@six.add_metaclass(abc.ABCMeta)
-class Credentials(credentials.Scoped, credentials.CredentialsWithQuotaProject):
- """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,
- client_id=None,
- client_secret=None,
- quota_project_id=None,
- scopes=None,
- default_scopes=None,
- ):
- """Instantiates an external account credentials object.
-
- 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.
- 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.
- 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.
- 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._token_url = token_url
- self._credential_source = credential_source
- self._service_account_impersonation_url = service_account_impersonation_url
- 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
-
- 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)
-
- if self._service_account_impersonation_url:
- self._impersonated_credentials = self._initialize_impersonated_credentials()
- else:
- self._impersonated_credentials = None
- self._project_id = 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
-
- @_helpers.copy_docstring(credentials.Scoped)
- def with_scopes(self, scopes, default_scopes=None):
- return self.__class__(
- audience=self._audience,
- subject_token_type=self._subject_token_type,
- token_url=self._token_url,
- credential_source=self._credential_source,
- service_account_impersonation_url=self._service_account_impersonation_url,
- client_id=self._client_id,
- client_secret=self._client_secret,
- quota_project_id=self._quota_project_id,
- scopes=scopes,
- default_scopes=default_scopes,
- )
-
- @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 pool.
-
- 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
- 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.
- if self.project_number and scopes:
- headers = {}
- url = _CLOUD_RESOURCE_MANAGER + self.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
- 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()
- 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,
- )
- self.token = response_data.get("access_token")
- lifetime = datetime.timedelta(seconds=response_data.get("expires_in"))
- self.expiry = now + lifetime
-
- @_helpers.copy_docstring(credentials.CredentialsWithQuotaProject)
- def with_quota_project(self, quota_project_id):
- # Return copy of instance with the provided quota project ID.
- return self.__class__(
- audience=self._audience,
- subject_token_type=self._subject_token_type,
- token_url=self._token_url,
- credential_source=self._credential_source,
- service_account_impersonation_url=self._service_account_impersonation_url,
- client_id=self._client_id,
- client_secret=self._client_secret,
- quota_project_id=quota_project_id,
- scopes=self._scopes,
- default_scopes=self._default_scopes,
- )
-
- 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.
- source_credentials = self.__class__(
- audience=self._audience,
- subject_token_type=self._subject_token_type,
- token_url=self._token_url,
- credential_source=self._credential_source,
- service_account_impersonation_url=None,
- client_id=self._client_id,
- client_secret=self._client_secret,
- quota_project_id=self._quota_project_id,
- scopes=self._scopes,
- default_scopes=self._default_scopes,
- )
-
- # Determine target_principal.
- start_index = self._service_account_impersonation_url.rfind("/")
- end_index = self._service_account_impersonation_url.find(":generateAccessToken")
- if start_index != -1 and end_index != -1 and start_index < end_index:
- start_index = start_index + 1
- target_principal = self._service_account_impersonation_url[
- start_index:end_index
- ]
- else:
- 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,
- )
diff --git a/google/auth/identity_pool.py b/google/auth/identity_pool.py
deleted file mode 100644
index 5362199..0000000
--- a/google/auth/identity_pool.py
+++ /dev/null
@@ -1,279 +0,0 @@
-# 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. 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.
-"""
-
-try:
- from collections.abc import Mapping
-# Python 2.7 compatibility
-except ImportError: # pragma: NO COVER
- from collections import Mapping
-import io
-import json
-import os
-
-from google.auth import _helpers
-from google.auth import exceptions
-from google.auth import external_account
-
-
-class Credentials(external_account.Credentials):
- """External account credentials sourced from files and URLs."""
-
- def __init__(
- self,
- audience,
- subject_token_type,
- token_url,
- credential_source,
- service_account_impersonation_url=None,
- client_id=None,
- client_secret=None,
- quota_project_id=None,
- scopes=None,
- default_scopes=None,
- ):
- """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.
- 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 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"
- }
-
- service_account_impersonation_url (Optional[str]): The optional service account
- impersonation getAccessToken URL.
- client_id (Optional[str]): The optional client ID.
- client_secret (Optional[str]): The optional client secret.
- 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.
-
- 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,
- service_account_impersonation_url=service_account_impersonation_url,
- client_id=client_id,
- client_secret=client_secret,
- quota_project_id=quota_project_id,
- scopes=scopes,
- default_scopes=default_scopes,
- )
- if not isinstance(credential_source, Mapping):
- self._credential_source_file = None
- self._credential_source_url = None
- else:
- self._credential_source_file = credential_source.get("file")
- self._credential_source_url = credential_source.get("url")
- 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"
- )
- # environment_id is only supported in AWS or dedicated future external
- # account credentials.
- if "environment_id" in credential_source:
- raise ValueError(
- "Invalid Identity Pool credential_source field 'environment_id'"
- )
- if self._credential_source_format_type not in ["text", "json"]:
- raise ValueError(
- "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 ValueError(
- "Missing subject_token_field_name for JSON credential_source format"
- )
- else:
- self._credential_source_field_name = None
-
- if self._credential_source_file and self._credential_source_url:
- raise ValueError(
- "Ambiguous credential_source. 'file' is mutually exclusive with 'url'."
- )
- if not self._credential_source_file and not self._credential_source_url:
- raise ValueError(
- "Missing credential_source. A 'file' or 'url' must be provided."
- )
-
- @_helpers.copy_docstring(external_account.Credentials)
- def retrieve_subject_token(self, request):
- return self._parse_token_data(
- self._get_token_data(request),
- self._credential_source_format_type,
- self._credential_source_field_name,
- )
-
- def _get_token_data(self, request):
- if self._credential_source_file:
- return self._get_file_data(self._credential_source_file)
- else:
- return self._get_url_data(
- request, self._credential_source_url, self._credential_source_headers
- )
-
- def _get_file_data(self, filename):
- if not os.path.exists(filename):
- raise exceptions.RefreshError("File '{}' was not found.".format(filename))
-
- with io.open(filename, "r", encoding="utf-8") as file_obj:
- return file_obj.read(), filename
-
- def _get_url_data(self, request, url, headers):
- response = request(url=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 != 200:
- raise exceptions.RefreshError(
- "Unable to retrieve Identity Pool subject token", response_body
- )
-
- return response_body, url
-
- def _parse_token_data(
- self, token_content, format_type="text", subject_token_field_name=None
- ):
- content, filename = token_content
- if format_type == "text":
- token = content
- else:
- try:
- # Parse file content as JSON.
- response_data = json.loads(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(
- filename, subject_token_field_name
- )
- )
- if not token:
- raise exceptions.RefreshError(
- "Missing subject_token in the credential_source file"
- )
- return token
-
- @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.
- """
- return cls(
- audience=info.get("audience"),
- subject_token_type=info.get("subject_token_type"),
- token_url=info.get("token_url"),
- service_account_impersonation_url=info.get(
- "service_account_impersonation_url"
- ),
- 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"),
- **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.
- """
- with io.open(filename, "r", encoding="utf-8") as json_file:
- data = json.load(json_file)
- return cls.from_info(data, **kwargs)
diff --git a/google/auth/impersonated_credentials.py b/google/auth/impersonated_credentials.py
index b8a6c49..4d15837 100644
--- a/google/auth/impersonated_credentials.py
+++ b/google/auth/impersonated_credentials.py
@@ -65,9 +65,7 @@
_DEFAULT_TOKEN_URI = "https://oauth2.googleapis.com/token"
-def _make_iam_token_request(
- request, principal, headers, body, iam_endpoint_override=None
-):
+def _make_iam_token_request(request, principal, headers, body):
"""Makes a request to the Google Cloud IAM service for an access token.
Args:
request (Request): The Request object to use.
@@ -75,9 +73,6 @@
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
@@ -87,7 +82,7 @@
`iamcredentials.googleapis.com` is not enabled or the
`Service Account Token Creator` is not assigned
"""
- iam_endpoint = iam_endpoint_override or _IAM_ENDPOINT.format(principal)
+ iam_endpoint = _IAM_ENDPOINT.format(principal)
body = json.dumps(body).encode("utf-8")
@@ -190,7 +185,6 @@
delegates=None,
lifetime=_DEFAULT_TOKEN_LIFETIME_SECS,
quota_project_id=None,
- iam_endpoint_override=None,
):
"""
Args:
@@ -215,9 +209,6 @@
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 (Optiona[str]): The full IAM endpoint override
- with the target_principal embedded. This is useful when supporting
- impersonation with regional endpoints.
"""
super(Credentials, self).__init__()
@@ -235,7 +226,6 @@
self.token = None
self.expiry = _helpers.utcnow()
self._quota_project_id = quota_project_id
- self._iam_endpoint_override = iam_endpoint_override
@_helpers.copy_docstring(credentials.Credentials)
def refresh(self, request):
@@ -270,7 +260,6 @@
principal=self._target_principal,
headers=headers,
body=body,
- iam_endpoint_override=self._iam_endpoint_override,
)
def sign_bytes(self, message):
@@ -313,7 +302,6 @@
delegates=self._delegates,
lifetime=self._lifetime,
quota_project_id=quota_project_id,
- iam_endpoint_override=self._iam_endpoint_override,
)
diff --git a/google/oauth2/sts.py b/google/oauth2/sts.py
deleted file mode 100644
index ae3c014..0000000
--- a/google/oauth2/sts.py
+++ /dev/null
@@ -1,155 +0,0 @@
-# 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.
-
-"""OAuth 2.0 Token Exchange Spec.
-
-This module defines a token exchange utility based on the `OAuth 2.0 Token
-Exchange`_ spec. This will be mainly used to exchange external credentials
-for GCP access tokens in workload identity pools to access Google APIs.
-
-The implementation will support various types of client authentication as
-allowed in the spec.
-
-A deviation on the spec will be for additional Google specific options that
-cannot be easily mapped to parameters defined in the RFC.
-
-The returned dictionary response will be based on the `rfc8693 section 2.2.1`_
-spec JSON response.
-
-.. _OAuth 2.0 Token Exchange: https://tools.ietf.org/html/rfc8693
-.. _rfc8693 section 2.2.1: https://tools.ietf.org/html/rfc8693#section-2.2.1
-"""
-
-import json
-
-from six.moves import http_client
-from six.moves import urllib
-
-from google.oauth2 import utils
-
-
-_URLENCODED_HEADERS = {"Content-Type": "application/x-www-form-urlencoded"}
-
-
-class Client(utils.OAuthClientAuthHandler):
- """Implements the OAuth 2.0 token exchange spec based on
- https://tools.ietf.org/html/rfc8693.
- """
-
- def __init__(self, token_exchange_endpoint, client_authentication=None):
- """Initializes an STS client instance.
-
- Args:
- token_exchange_endpoint (str): The token exchange endpoint.
- client_authentication (Optional(google.oauth2.oauth2_utils.ClientAuthentication)):
- The optional OAuth client authentication credentials if available.
- """
- super(Client, self).__init__(client_authentication)
- self._token_exchange_endpoint = token_exchange_endpoint
-
- def exchange_token(
- self,
- request,
- grant_type,
- subject_token,
- subject_token_type,
- resource=None,
- audience=None,
- scopes=None,
- requested_token_type=None,
- actor_token=None,
- actor_token_type=None,
- additional_options=None,
- additional_headers=None,
- ):
- """Exchanges the provided token for another type of token based on the
- rfc8693 spec.
-
- Args:
- request (google.auth.transport.Request): A callable used to make
- HTTP requests.
- grant_type (str): The OAuth 2.0 token exchange grant type.
- subject_token (str): The OAuth 2.0 token exchange subject token.
- subject_token_type (str): The OAuth 2.0 token exchange subject token type.
- resource (Optional[str]): The optional OAuth 2.0 token exchange resource field.
- audience (Optional[str]): The optional OAuth 2.0 token exchange audience field.
- scopes (Optional[Sequence[str]]): The optional list of scopes to use.
- requested_token_type (Optional[str]): The optional OAuth 2.0 token exchange requested
- token type.
- actor_token (Optional[str]): The optional OAuth 2.0 token exchange actor token.
- actor_token_type (Optional[str]): The optional OAuth 2.0 token exchange actor token type.
- additional_options (Optional[Mapping[str, str]]): The optional additional
- non-standard Google specific options.
- additional_headers (Optional[Mapping[str, str]]): The optional additional
- headers to pass to the token exchange endpoint.
-
- Returns:
- Mapping[str, str]: The token exchange JSON-decoded response data containing
- the requested token and its expiration time.
-
- Raises:
- google.auth.exceptions.OAuthError: If the token endpoint returned
- an error.
- """
- # Initialize request headers.
- headers = _URLENCODED_HEADERS.copy()
- # Inject additional headers.
- if additional_headers:
- for k, v in dict(additional_headers).items():
- headers[k] = v
- # Initialize request body.
- request_body = {
- "grant_type": grant_type,
- "resource": resource,
- "audience": audience,
- "scope": " ".join(scopes or []),
- "requested_token_type": requested_token_type,
- "subject_token": subject_token,
- "subject_token_type": subject_token_type,
- "actor_token": actor_token,
- "actor_token_type": actor_token_type,
- "options": None,
- }
- # Add additional non-standard options.
- if additional_options:
- request_body["options"] = urllib.parse.quote(json.dumps(additional_options))
- # Remove empty fields in request body.
- for k, v in dict(request_body).items():
- if v is None or v == "":
- del request_body[k]
- # Apply OAuth client authentication.
- self.apply_client_authentication_options(headers, request_body)
-
- # Execute request.
- response = request(
- url=self._token_exchange_endpoint,
- method="POST",
- headers=headers,
- body=urllib.parse.urlencode(request_body).encode("utf-8"),
- )
-
- response_body = (
- response.data.decode("utf-8")
- if hasattr(response.data, "decode")
- else response.data
- )
-
- # If non-200 response received, translate to OAuthError exception.
- if response.status != http_client.OK:
- utils.handle_error_response(response_body)
-
- response_data = json.loads(response_body)
-
- # Return successful response.
- return response_data
diff --git a/google/oauth2/utils.py b/google/oauth2/utils.py
deleted file mode 100644
index efda796..0000000
--- a/google/oauth2/utils.py
+++ /dev/null
@@ -1,171 +0,0 @@
-# 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.
-
-"""OAuth 2.0 Utilities.
-
-This module provides implementations for various OAuth 2.0 utilities.
-This includes `OAuth error handling`_ and
-`Client authentication for OAuth flows`_.
-
-OAuth error handling
---------------------
-This will define interfaces for handling OAuth related error responses as
-stated in `RFC 6749 section 5.2`_.
-This will include a common function to convert these HTTP error responses to a
-:class:`google.auth.exceptions.OAuthError` exception.
-
-
-Client authentication for OAuth flows
--------------------------------------
-We introduce an interface for defining client authentication credentials based
-on `RFC 6749 section 2.3.1`_. This will expose the following
-capabilities:
-
- * Ability to support basic authentication via request header.
- * Ability to support bearer token authentication via request header.
- * Ability to support client ID / secret authentication via request body.
-
-.. _RFC 6749 section 2.3.1: https://tools.ietf.org/html/rfc6749#section-2.3.1
-.. _RFC 6749 section 5.2: https://tools.ietf.org/html/rfc6749#section-5.2
-"""
-
-import abc
-import base64
-import enum
-import json
-
-import six
-
-from google.auth import exceptions
-
-
-# OAuth client authentication based on
-# https://tools.ietf.org/html/rfc6749#section-2.3.
-class ClientAuthType(enum.Enum):
- basic = 1
- request_body = 2
-
-
-class ClientAuthentication(object):
- """Defines the client authentication credentials for basic and request-body
- types based on https://tools.ietf.org/html/rfc6749#section-2.3.1.
- """
-
- def __init__(self, client_auth_type, client_id, client_secret=None):
- """Instantiates a client authentication object containing the client ID
- and secret credentials for basic and response-body auth.
-
- Args:
- client_auth_type (google.oauth2.oauth_utils.ClientAuthType): The
- client authentication type.
- client_id (str): The client ID.
- client_secret (Optional[str]): The client secret.
- """
- self.client_auth_type = client_auth_type
- self.client_id = client_id
- self.client_secret = client_secret
-
-
-@six.add_metaclass(abc.ABCMeta)
-class OAuthClientAuthHandler(object):
- """Abstract class for handling client authentication in OAuth-based
- operations.
- """
-
- def __init__(self, client_authentication=None):
- """Instantiates an OAuth client authentication handler.
-
- Args:
- client_authentication (Optional[google.oauth2.utils.ClientAuthentication]):
- The OAuth client authentication credentials if available.
- """
- super(OAuthClientAuthHandler, self).__init__()
- self._client_authentication = client_authentication
-
- def apply_client_authentication_options(
- self, headers, request_body=None, bearer_token=None
- ):
- """Applies client authentication on the OAuth request's headers or POST
- body.
-
- Args:
- headers (Mapping[str, str]): The HTTP request header.
- request_body (Optional[Mapping[str, str]): The HTTP request body
- dictionary. For requests that do not support request body, this
- is None and will be ignored.
- bearer_token (Optional[str]): The optional bearer token.
- """
- # Inject authenticated header.
- self._inject_authenticated_headers(headers, bearer_token)
- # Inject authenticated request body.
- if bearer_token is None:
- self._inject_authenticated_request_body(request_body)
-
- def _inject_authenticated_headers(self, headers, bearer_token=None):
- if bearer_token is not None:
- headers["Authorization"] = "Bearer %s" % bearer_token
- elif (
- self._client_authentication is not None
- and self._client_authentication.client_auth_type is ClientAuthType.basic
- ):
- username = self._client_authentication.client_id
- password = self._client_authentication.client_secret or ""
-
- credentials = base64.b64encode(
- ("%s:%s" % (username, password)).encode()
- ).decode()
- headers["Authorization"] = "Basic %s" % credentials
-
- def _inject_authenticated_request_body(self, request_body):
- if (
- self._client_authentication is not None
- and self._client_authentication.client_auth_type
- is ClientAuthType.request_body
- ):
- if request_body is None:
- raise exceptions.OAuthError(
- "HTTP request does not support request-body"
- )
- else:
- request_body["client_id"] = self._client_authentication.client_id
- request_body["client_secret"] = (
- self._client_authentication.client_secret or ""
- )
-
-
-def handle_error_response(response_body):
- """Translates an error response from an OAuth operation into an
- OAuthError exception.
-
- Args:
- response_body (str): The decoded response data.
-
- Raises:
- google.auth.exceptions.OAuthError
- """
- try:
- error_components = []
- error_data = json.loads(response_body)
-
- error_components.append("Error code {}".format(error_data["error"]))
- if "error_description" in error_data:
- error_components.append(": {}".format(error_data["error_description"]))
- if "error_uri" in error_data:
- error_components.append(" - {}".format(error_data["error_uri"]))
- error_details = "".join(error_components)
- # If no details could be extracted, use the response data.
- except (KeyError, ValueError):
- error_details = response_body
-
- raise exceptions.OAuthError(error_details, response_body)