diff --git a/CHANGELOG.md b/CHANGELOG.md index 94dd92594..d05e4e7d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,13 @@ [1]: https://pypi.org/project/google-auth/#history +### [1.22.1](https://www.github.com/googleapis/google-auth-library-python/compare/v1.22.0...v1.22.1) (2020-10-05) + + +### Bug Fixes + +* move aiohttp to extra as it is currently internal surface ([#619](https://www.github.com/googleapis/google-auth-library-python/issues/619)) ([a924011](https://www.github.com/googleapis/google-auth-library-python/commit/a9240111e7af29338624d98ee10aed31462f4d19)), closes [#618](https://www.github.com/googleapis/google-auth-library-python/issues/618) + ## [1.22.0](https://www.github.com/googleapis/google-auth-library-python/compare/v1.21.3...v1.22.0) (2020-09-28) diff --git a/google/auth/aws.py b/google/auth/aws.py index d4c75c0c8..3a6d53750 100644 --- a/google/auth/aws.py +++ b/google/auth/aws.py @@ -12,21 +12,41 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Helper functions for AWS related operations +"""AWS Credentials and AWS Signature V4 Request Signer. This module provides a basic implementation of the `AWS Signature Version 4`_ request signing algorithm. +This module also provides credentials for AWS workloads that 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. + +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" @@ -302,3 +322,377 @@ def _generate_authentication_header_map( 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, + ): + """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. + + .. 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, + ) + 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 exceptions.GoogleAuthError( + "No valid AWS 'credential_source' provided" + ) + elif int(env_version or "") != 1: + raise exceptions.GoogleAuthError( + "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: + google.auth.exceptions.GoogleAuthError: 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/crypt/_cryptography_rsa.py b/google/auth/crypt/_cryptography_rsa.py index e94bc681e..916c9d80a 100644 --- a/google/auth/crypt/_cryptography_rsa.py +++ b/google/auth/crypt/_cryptography_rsa.py @@ -25,23 +25,10 @@ from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import padding import cryptography.x509 -import pkg_resources from google.auth import _helpers from google.auth.crypt import base -_IMPORT_ERROR_MSG = ( - "cryptography>=1.4.0 is required to use cryptography-based RSA " "implementation." -) - -try: # pragma: NO COVER - release = pkg_resources.get_distribution("cryptography").parsed_version - if release < pkg_resources.parse_version("1.4.0"): - raise ImportError(_IMPORT_ERROR_MSG) -except pkg_resources.DistributionNotFound: # pragma: NO COVER - raise ImportError(_IMPORT_ERROR_MSG) - - _CERTIFICATE_MARKER = b"-----BEGIN CERTIFICATE-----" _BACKEND = backends.default_backend() _PADDING = padding.PKCS1v15() diff --git a/google/auth/crypt/es256.py b/google/auth/crypt/es256.py index 6955efcc5..c6d617606 100644 --- a/google/auth/crypt/es256.py +++ b/google/auth/crypt/es256.py @@ -25,22 +25,10 @@ from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature import cryptography.x509 -import pkg_resources from google.auth import _helpers from google.auth.crypt import base -_IMPORT_ERROR_MSG = ( - "cryptography>=1.4.0 is required to use cryptography-based ECDSA " "algorithms" -) - -try: # pragma: NO COVER - release = pkg_resources.get_distribution("cryptography").parsed_version - if release < pkg_resources.parse_version("1.4.0"): - raise ImportError(_IMPORT_ERROR_MSG) -except pkg_resources.DistributionNotFound: # pragma: NO COVER - raise ImportError(_IMPORT_ERROR_MSG) - _CERTIFICATE_MARKER = b"-----BEGIN CERTIFICATE-----" _BACKEND = backends.default_backend() diff --git a/google/auth/environment_vars.py b/google/auth/environment_vars.py index 46a892664..416bab0c0 100644 --- a/google/auth/environment_vars.py +++ b/google/auth/environment_vars.py @@ -59,3 +59,13 @@ 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/setup.py b/setup.py index a62a080b4..d3b740a62 100644 --- a/setup.py +++ b/setup.py @@ -31,11 +31,12 @@ 'aiohttp >= 3.6.2, < 4.0.0dev; python_version>="3.6"', ) +extras = {"aiohttp": "aiohttp >= 3.6.2, < 4.0.0dev; python_version>='3.6'"} with io.open("README.rst", "r") as fh: long_description = fh.read() -version = "1.22.0" +version = "1.22.1" setup( name="google-auth", @@ -48,6 +49,7 @@ packages=find_packages(exclude=("tests*", "system_tests*")), namespace_packages=("google",), install_requires=DEPENDENCIES, + extras_require=extras, python_requires=">=2.7,!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*", license="Apache 2.0", keywords="google auth oauth client", diff --git a/tests/test_aws.py b/tests/test_aws.py index 26728aa83..5322c32c6 100644 --- a/tests/test_aws.py +++ b/tests/test_aws.py @@ -13,14 +13,39 @@ # limitations under the License. import datetime +import json import mock import pytest +from six.moves import http_client from six.moves import urllib +from google.auth import _helpers from google.auth import aws +from google.auth import environment_vars +from google.auth import exceptions +from google.auth import transport +CLIENT_ID = "username" +CLIENT_SECRET = "password" +# Base64 encoding of "username:password". +BASIC_AUTH_ENCODING = "dXNlcm5hbWU6cGFzc3dvcmQ=" +SERVICE_ACCOUNT_EMAIL = "service-1234@service-name.iam.gserviceaccount.com" +SERVICE_ACCOUNT_IMPERSONATION_URL = ( + "https://us-east1-iamcredentials.googleapis.com/v1/projects/-" + + "/serviceAccounts/{}:generateAccessToken".format(SERVICE_ACCOUNT_EMAIL) +) +QUOTA_PROJECT_ID = "QUOTA_PROJECT_ID" +SCOPES = ["scope1", "scope2"] +TOKEN_URL = "https://sts.googleapis.com/v1/token" +SUBJECT_TOKEN_TYPE = "urn:ietf:params:aws:token-type:aws4_request" +AUDIENCE = "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID" +REGION_URL = "http://169.254.169.254/latest/meta-data/placement/availability-zone" +SECURITY_CREDS_URL = "http://169.254.169.254/latest/meta-data/iam/security-credentials" +CRED_VERIFICATION_URL = ( + "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15" +) # Sample AWS security credentials to be used with tests that require a session token. ACCESS_KEY_ID = "ASIARD4OQDT6A77FR3CL" SECRET_ACCESS_KEY = "Y8AfSaucF37G4PpvfguKZ3/l7Id4uocLXxX0+VTx" @@ -498,3 +523,719 @@ def test_get_request_options( ) assert actual_signed_request == signed_request + + +class TestCredentials(object): + AWS_REGION = "us-east-2" + AWS_ROLE = "gcp-aws-role" + AWS_SECURITY_CREDENTIALS_RESPONSE = { + "AccessKeyId": ACCESS_KEY_ID, + "SecretAccessKey": SECRET_ACCESS_KEY, + "Token": TOKEN, + } + AWS_SIGNATURE_TIME = "2020-08-11T06:55:22Z" + CREDENTIAL_SOURCE = { + "environment_id": "aws1", + "region_url": REGION_URL, + "url": SECURITY_CREDS_URL, + "regional_cred_verification_url": CRED_VERIFICATION_URL, + } + SUCCESS_RESPONSE = { + "access_token": "ACCESS_TOKEN", + "issued_token_type": "urn:ietf:params:oauth:token-type:access_token", + "token_type": "Bearer", + "expires_in": 3600, + "scope": " ".join(SCOPES), + } + + @classmethod + def make_serialized_aws_signed_request( + cls, + aws_security_credentials, + region_name="us-east-2", + url="https://sts.us-east-2.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15", + ): + """Utility to generate serialize AWS signed requests. + This makes it easy to assert generated subject tokens based on the + provided AWS security credentials, regions and AWS STS endpoint. + """ + request_signer = aws.RequestSigner(region_name) + signed_request = request_signer.get_request_options( + aws_security_credentials, url, "POST" + ) + reformatted_signed_request = { + "url": signed_request.get("url"), + "method": signed_request.get("method"), + "headers": [ + { + "key": "Authorization", + "value": signed_request.get("headers").get("Authorization"), + }, + {"key": "host", "value": signed_request.get("headers").get("host")}, + { + "key": "x-amz-date", + "value": signed_request.get("headers").get("x-amz-date"), + }, + ], + } + # Include security token if available. + if "security_token" in aws_security_credentials: + reformatted_signed_request.get("headers").append( + { + "key": "x-amz-security-token", + "value": signed_request.get("headers").get("x-amz-security-token"), + } + ) + # Append x-goog-cloud-target-resource header. + reformatted_signed_request.get("headers").append( + {"key": "x-goog-cloud-target-resource", "value": AUDIENCE} + ), + return urllib.parse.quote( + json.dumps( + reformatted_signed_request, separators=(",", ":"), sort_keys=True + ) + ) + + @classmethod + def make_mock_request( + cls, + region_status=None, + region_name=None, + role_status=None, + role_name=None, + security_credentials_status=None, + security_credentials_data=None, + token_status=None, + token_data=None, + impersonation_status=None, + impersonation_data=None, + ): + """Utility function to generate a mock HTTP request object. + This will facilitate testing various edge cases by specify how the + various endpoints will respond while generating a Google Access token + in an AWS environment. + """ + responses = [] + if region_status: + # AWS region request. + region_response = mock.create_autospec(transport.Response, instance=True) + region_response.status = region_status + if region_name: + region_response.data = "{}b".format(region_name).encode("utf-8") + responses.append(region_response) + + if role_status: + # AWS role name request. + role_response = mock.create_autospec(transport.Response, instance=True) + role_response.status = role_status + if role_name: + role_response.data = role_name.encode("utf-8") + responses.append(role_response) + + if security_credentials_status: + # AWS security credentials request. + security_credentials_response = mock.create_autospec( + transport.Response, instance=True + ) + security_credentials_response.status = security_credentials_status + if security_credentials_data: + security_credentials_response.data = json.dumps( + security_credentials_data + ).encode("utf-8") + responses.append(security_credentials_response) + + if token_status: + # GCP token exchange request. + token_response = mock.create_autospec(transport.Response, instance=True) + token_response.status = token_status + token_response.data = json.dumps(token_data).encode("utf-8") + responses.append(token_response) + + if impersonation_status: + # Service account impersonation request. + impersonation_response = mock.create_autospec( + transport.Response, instance=True + ) + impersonation_response.status = impersonation_status + impersonation_response.data = json.dumps(impersonation_data).encode("utf-8") + responses.append(impersonation_response) + + request = mock.create_autospec(transport.Request) + request.side_effect = responses + + return request + + @classmethod + def make_credentials( + cls, + credential_source, + client_id=None, + client_secret=None, + quota_project_id=None, + scopes=None, + service_account_impersonation_url=None, + ): + return aws.Credentials( + audience=AUDIENCE, + subject_token_type=SUBJECT_TOKEN_TYPE, + token_url=TOKEN_URL, + service_account_impersonation_url=service_account_impersonation_url, + credential_source=credential_source, + client_id=client_id, + client_secret=client_secret, + quota_project_id=quota_project_id, + scopes=scopes, + ) + + @classmethod + def assert_aws_metadata_request_kwargs(cls, request_kwargs, url, headers=None): + assert request_kwargs["url"] == url + # All used AWS metadata server endpoints use GET HTTP method. + assert request_kwargs["method"] == "GET" + if headers: + assert request_kwargs["headers"] == headers + else: + assert "headers" not in request_kwargs + # None of the endpoints used require any data in request. + assert "body" not in request_kwargs + + @classmethod + def assert_token_request_kwargs( + cls, request_kwargs, headers, request_data, token_url=TOKEN_URL + ): + assert request_kwargs["url"] == token_url + assert request_kwargs["method"] == "POST" + assert request_kwargs["headers"] == headers + assert request_kwargs["body"] is not None + body_tuples = urllib.parse.parse_qsl(request_kwargs["body"]) + assert len(body_tuples) == len(request_data.keys()) + for (k, v) in body_tuples: + assert v.decode("utf-8") == request_data[k.decode("utf-8")] + + @classmethod + def assert_impersonation_request_kwargs( + cls, + request_kwargs, + headers, + request_data, + service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL, + ): + assert request_kwargs["url"] == service_account_impersonation_url + assert request_kwargs["method"] == "POST" + assert request_kwargs["headers"] == headers + assert request_kwargs["body"] is not None + body_json = json.loads(request_kwargs["body"].decode("utf-8")) + assert body_json == request_data + + @mock.patch.object(aws.Credentials, "__init__", return_value=None) + def test_from_info_full_options(self, mock_init): + credentials = aws.Credentials.from_info( + { + "audience": AUDIENCE, + "subject_token_type": SUBJECT_TOKEN_TYPE, + "token_url": TOKEN_URL, + "service_account_impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL, + "client_id": CLIENT_ID, + "client_secret": CLIENT_SECRET, + "quota_project_id": QUOTA_PROJECT_ID, + "credential_source": self.CREDENTIAL_SOURCE, + } + ) + + # Confirm aws.Credentials instance initialized with the expected parameters. + assert isinstance(credentials, aws.Credentials) + mock_init.assert_called_once_with( + audience=AUDIENCE, + subject_token_type=SUBJECT_TOKEN_TYPE, + token_url=TOKEN_URL, + service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL, + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + credential_source=self.CREDENTIAL_SOURCE, + quota_project_id=QUOTA_PROJECT_ID, + ) + + @mock.patch.object(aws.Credentials, "__init__", return_value=None) + def test_from_info_required_options_only(self, mock_init): + credentials = aws.Credentials.from_info( + { + "audience": AUDIENCE, + "subject_token_type": SUBJECT_TOKEN_TYPE, + "token_url": TOKEN_URL, + "credential_source": self.CREDENTIAL_SOURCE, + } + ) + + # Confirm aws.Credentials instance initialized with the expected parameters. + assert isinstance(credentials, aws.Credentials) + mock_init.assert_called_once_with( + audience=AUDIENCE, + subject_token_type=SUBJECT_TOKEN_TYPE, + token_url=TOKEN_URL, + service_account_impersonation_url=None, + client_id=None, + client_secret=None, + credential_source=self.CREDENTIAL_SOURCE, + quota_project_id=None, + ) + + @mock.patch.object(aws.Credentials, "__init__", return_value=None) + def test_from_file_full_options(self, mock_init, tmpdir): + info = { + "audience": AUDIENCE, + "subject_token_type": SUBJECT_TOKEN_TYPE, + "token_url": TOKEN_URL, + "service_account_impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL, + "client_id": CLIENT_ID, + "client_secret": CLIENT_SECRET, + "quota_project_id": QUOTA_PROJECT_ID, + "credential_source": self.CREDENTIAL_SOURCE, + } + config_file = tmpdir.join("config.json") + config_file.write(json.dumps(info)) + credentials = aws.Credentials.from_file(str(config_file)) + + # Confirm aws.Credentials instance initialized with the expected parameters. + assert isinstance(credentials, aws.Credentials) + mock_init.assert_called_once_with( + audience=AUDIENCE, + subject_token_type=SUBJECT_TOKEN_TYPE, + token_url=TOKEN_URL, + service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL, + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + credential_source=self.CREDENTIAL_SOURCE, + quota_project_id=QUOTA_PROJECT_ID, + ) + + @mock.patch.object(aws.Credentials, "__init__", return_value=None) + def test_from_file_required_options_only(self, mock_init, tmpdir): + info = { + "audience": AUDIENCE, + "subject_token_type": SUBJECT_TOKEN_TYPE, + "token_url": TOKEN_URL, + "credential_source": self.CREDENTIAL_SOURCE, + } + config_file = tmpdir.join("config.json") + config_file.write(json.dumps(info)) + credentials = aws.Credentials.from_file(str(config_file)) + + # Confirm aws.Credentials instance initialized with the expected parameters. + assert isinstance(credentials, aws.Credentials) + mock_init.assert_called_once_with( + audience=AUDIENCE, + subject_token_type=SUBJECT_TOKEN_TYPE, + token_url=TOKEN_URL, + service_account_impersonation_url=None, + client_id=None, + client_secret=None, + credential_source=self.CREDENTIAL_SOURCE, + quota_project_id=None, + ) + + def test_constructor_invalid_credential_source(self): + # Provide invalid credential source. + credential_source = {"unsupported": "value"} + + with pytest.raises(exceptions.GoogleAuthError) as excinfo: + self.make_credentials(credential_source=credential_source) + + assert excinfo.match(r"No valid AWS 'credential_source' provided") + + def test_constructor_invalid_environment_id(self): + # Provide invalid environment_id. + credential_source = self.CREDENTIAL_SOURCE.copy() + credential_source["environment_id"] = "azure1" + + with pytest.raises(exceptions.GoogleAuthError) as excinfo: + self.make_credentials(credential_source=credential_source) + + assert excinfo.match(r"No valid AWS 'credential_source' provided") + + def test_constructor_missing_cred_verification_url(self): + # regional_cred_verification_url is a required field. + credential_source = self.CREDENTIAL_SOURCE.copy() + credential_source.pop("regional_cred_verification_url") + + with pytest.raises(exceptions.GoogleAuthError) as excinfo: + self.make_credentials(credential_source=credential_source) + + assert excinfo.match(r"No valid AWS 'credential_source' provided") + + def test_constructor_invalid_environment_id_version(self): + # Provide an unsupported version. + credential_source = self.CREDENTIAL_SOURCE.copy() + credential_source["environment_id"] = "aws3" + + with pytest.raises(exceptions.GoogleAuthError) as excinfo: + self.make_credentials(credential_source=credential_source) + + assert excinfo.match(r"aws version '3' is not supported in the current build.") + + def test_retrieve_subject_token_missing_region_url(self): + # When AWS_REGION envvar is not available, region_url is required for + # determining the current AWS region. + credential_source = self.CREDENTIAL_SOURCE.copy() + credential_source.pop("region_url") + credentials = self.make_credentials(credential_source=credential_source) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.retrieve_subject_token(None) + + assert excinfo.match(r"Unable to determine AWS region") + + @mock.patch("google.auth._helpers.utcnow") + def test_retrieve_subject_token_success_temp_creds_no_environment_vars( + self, utcnow + ): + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + request = self.make_mock_request( + region_status=http_client.OK, + region_name=self.AWS_REGION, + role_status=http_client.OK, + role_name=self.AWS_ROLE, + security_credentials_status=http_client.OK, + security_credentials_data=self.AWS_SECURITY_CREDENTIALS_RESPONSE, + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + subject_token = credentials.retrieve_subject_token(request) + + assert subject_token == self.make_serialized_aws_signed_request( + { + "access_key_id": ACCESS_KEY_ID, + "secret_access_key": SECRET_ACCESS_KEY, + "security_token": TOKEN, + } + ) + # Assert region request. + self.assert_aws_metadata_request_kwargs( + request.call_args_list[0].kwargs, REGION_URL + ) + # Assert role request. + self.assert_aws_metadata_request_kwargs( + request.call_args_list[1].kwargs, SECURITY_CREDS_URL + ) + # Assert security credentials request. + self.assert_aws_metadata_request_kwargs( + request.call_args_list[2].kwargs, + "{}/{}".format(SECURITY_CREDS_URL, self.AWS_ROLE), + {"Content-Type": "application/json"}, + ) + + # Retrieve subject_token again. Region should not be queried again. + new_request = self.make_mock_request( + role_status=http_client.OK, + role_name=self.AWS_ROLE, + security_credentials_status=http_client.OK, + security_credentials_data=self.AWS_SECURITY_CREDENTIALS_RESPONSE, + ) + + credentials.retrieve_subject_token(new_request) + + # Only 2 requests should be sent as the region is cached. + assert len(new_request.call_args_list) == 2 + # Assert role request. + self.assert_aws_metadata_request_kwargs( + new_request.call_args_list[0].kwargs, SECURITY_CREDS_URL + ) + # Assert security credentials request. + self.assert_aws_metadata_request_kwargs( + new_request.call_args_list[1].kwargs, + "{}/{}".format(SECURITY_CREDS_URL, self.AWS_ROLE), + {"Content-Type": "application/json"}, + ) + + @mock.patch("google.auth._helpers.utcnow") + def test_retrieve_subject_token_success_permanent_creds_no_environment_vars( + self, utcnow + ): + # Simualte a permanent credential without a session token is + # returned by the security-credentials endpoint. + security_creds_response = self.AWS_SECURITY_CREDENTIALS_RESPONSE.copy() + security_creds_response.pop("Token") + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + request = self.make_mock_request( + region_status=http_client.OK, + region_name=self.AWS_REGION, + role_status=http_client.OK, + role_name=self.AWS_ROLE, + security_credentials_status=http_client.OK, + security_credentials_data=security_creds_response, + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + subject_token = credentials.retrieve_subject_token(request) + + assert subject_token == self.make_serialized_aws_signed_request( + {"access_key_id": ACCESS_KEY_ID, "secret_access_key": SECRET_ACCESS_KEY} + ) + + @mock.patch("google.auth._helpers.utcnow") + def test_retrieve_subject_token_success_environment_vars(self, utcnow, monkeypatch): + monkeypatch.setenv(environment_vars.AWS_ACCESS_KEY_ID, ACCESS_KEY_ID) + monkeypatch.setenv(environment_vars.AWS_SECRET_ACCESS_KEY, SECRET_ACCESS_KEY) + monkeypatch.setenv(environment_vars.AWS_SESSION_TOKEN, TOKEN) + monkeypatch.setenv(environment_vars.AWS_REGION, self.AWS_REGION) + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + subject_token = credentials.retrieve_subject_token(None) + + assert subject_token == self.make_serialized_aws_signed_request( + { + "access_key_id": ACCESS_KEY_ID, + "secret_access_key": SECRET_ACCESS_KEY, + "security_token": TOKEN, + } + ) + + @mock.patch("google.auth._helpers.utcnow") + def test_retrieve_subject_token_success_environment_vars_no_session_token( + self, utcnow, monkeypatch + ): + monkeypatch.setenv(environment_vars.AWS_ACCESS_KEY_ID, ACCESS_KEY_ID) + monkeypatch.setenv(environment_vars.AWS_SECRET_ACCESS_KEY, SECRET_ACCESS_KEY) + monkeypatch.setenv(environment_vars.AWS_REGION, self.AWS_REGION) + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + subject_token = credentials.retrieve_subject_token(None) + + assert subject_token == self.make_serialized_aws_signed_request( + {"access_key_id": ACCESS_KEY_ID, "secret_access_key": SECRET_ACCESS_KEY} + ) + + @mock.patch("google.auth._helpers.utcnow") + def test_retrieve_subject_token_success_environment_vars_except_region( + self, utcnow, monkeypatch + ): + monkeypatch.setenv(environment_vars.AWS_ACCESS_KEY_ID, ACCESS_KEY_ID) + monkeypatch.setenv(environment_vars.AWS_SECRET_ACCESS_KEY, SECRET_ACCESS_KEY) + monkeypatch.setenv(environment_vars.AWS_SESSION_TOKEN, TOKEN) + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + # Region will be queried since it is not found in envvars. + request = self.make_mock_request( + region_status=http_client.OK, region_name=self.AWS_REGION + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + subject_token = credentials.retrieve_subject_token(request) + + assert subject_token == self.make_serialized_aws_signed_request( + { + "access_key_id": ACCESS_KEY_ID, + "secret_access_key": SECRET_ACCESS_KEY, + "security_token": TOKEN, + } + ) + + def test_retrieve_subject_token_error_determining_aws_region(self): + # Simulate error in retrieving the AWS region. + request = self.make_mock_request(region_status=http_client.BAD_REQUEST) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.retrieve_subject_token(request) + + assert excinfo.match(r"Unable to retrieve AWS region") + + def test_retrieve_subject_token_error_determining_aws_role(self): + # Simulate error in retrieving the AWS role name. + request = self.make_mock_request( + region_status=http_client.OK, + region_name=self.AWS_REGION, + role_status=http_client.BAD_REQUEST, + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.retrieve_subject_token(request) + + assert excinfo.match(r"Unable to retrieve AWS role name") + + def test_retrieve_subject_token_error_determining_security_creds_url(self): + # Simulate the security-credentials url is missing. This is needed for + # determining the AWS security credentials when not found in envvars. + credential_source = self.CREDENTIAL_SOURCE.copy() + credential_source.pop("url") + request = self.make_mock_request( + region_status=http_client.OK, region_name=self.AWS_REGION + ) + credentials = self.make_credentials(credential_source=credential_source) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.retrieve_subject_token(request) + + assert excinfo.match( + r"Unable to determine the AWS metadata server security credentials endpoint" + ) + + def test_retrieve_subject_token_error_determining_aws_security_creds(self): + # Simulate error in retrieving the AWS security credentials. + request = self.make_mock_request( + region_status=http_client.OK, + region_name=self.AWS_REGION, + role_status=http_client.OK, + role_name=self.AWS_ROLE, + security_credentials_status=http_client.BAD_REQUEST, + ) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.retrieve_subject_token(request) + + assert excinfo.match(r"Unable to retrieve AWS security credentials") + + @mock.patch("google.auth._helpers.utcnow") + def test_refresh_success_without_impersonation(self, utcnow): + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + expected_subject_token = self.make_serialized_aws_signed_request( + { + "access_key_id": ACCESS_KEY_ID, + "secret_access_key": SECRET_ACCESS_KEY, + "security_token": TOKEN, + } + ) + token_headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Authorization": "Basic " + BASIC_AUTH_ENCODING, + } + token_request_data = { + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "audience": AUDIENCE, + "requested_token_type": "urn:ietf:params:oauth:token-type:access_token", + "scope": " ".join(SCOPES), + "subject_token": expected_subject_token, + "subject_token_type": SUBJECT_TOKEN_TYPE, + } + request = self.make_mock_request( + region_status=http_client.OK, + region_name=self.AWS_REGION, + role_status=http_client.OK, + role_name=self.AWS_ROLE, + security_credentials_status=http_client.OK, + security_credentials_data=self.AWS_SECURITY_CREDENTIALS_RESPONSE, + token_status=http_client.OK, + token_data=self.SUCCESS_RESPONSE, + ) + credentials = self.make_credentials( + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + credential_source=self.CREDENTIAL_SOURCE, + quota_project_id=QUOTA_PROJECT_ID, + scopes=SCOPES, + ) + + credentials.refresh(request) + + assert len(request.call_args_list) == 4 + # Fourth request should be sent to GCP STS endpoint. + self.assert_token_request_kwargs( + request.call_args_list[3].kwargs, token_headers, token_request_data + ) + assert credentials.token == self.SUCCESS_RESPONSE["access_token"] + assert credentials.quota_project_id == QUOTA_PROJECT_ID + assert credentials.scopes == SCOPES + + @mock.patch("google.auth._helpers.utcnow") + def test_refresh_success_with_impersonation(self, utcnow): + utcnow.return_value = datetime.datetime.strptime( + self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ" + ) + expire_time = ( + _helpers.utcnow().replace(microsecond=0) + datetime.timedelta(seconds=3600) + ).isoformat("T") + "Z" + expected_subject_token = self.make_serialized_aws_signed_request( + { + "access_key_id": ACCESS_KEY_ID, + "secret_access_key": SECRET_ACCESS_KEY, + "security_token": TOKEN, + } + ) + token_headers = { + "Content-Type": "application/x-www-form-urlencoded", + "Authorization": "Basic " + BASIC_AUTH_ENCODING, + } + token_request_data = { + "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange", + "audience": AUDIENCE, + "requested_token_type": "urn:ietf:params:oauth:token-type:access_token", + "scope": "https://www.googleapis.com/auth/iam", + "subject_token": expected_subject_token, + "subject_token_type": SUBJECT_TOKEN_TYPE, + } + # Service account impersonation request/response. + impersonation_response = { + "accessToken": "SA_ACCESS_TOKEN", + "expireTime": expire_time, + } + impersonation_headers = { + "Content-Type": "application/json", + "authorization": "Bearer {}".format(self.SUCCESS_RESPONSE["access_token"]), + "x-goog-user-project": QUOTA_PROJECT_ID, + } + impersonation_request_data = { + "delegates": None, + "scope": SCOPES, + "lifetime": "3600s", + } + request = self.make_mock_request( + region_status=http_client.OK, + region_name=self.AWS_REGION, + role_status=http_client.OK, + role_name=self.AWS_ROLE, + security_credentials_status=http_client.OK, + security_credentials_data=self.AWS_SECURITY_CREDENTIALS_RESPONSE, + token_status=http_client.OK, + token_data=self.SUCCESS_RESPONSE, + impersonation_status=http_client.OK, + impersonation_data=impersonation_response, + ) + credentials = self.make_credentials( + client_id=CLIENT_ID, + client_secret=CLIENT_SECRET, + credential_source=self.CREDENTIAL_SOURCE, + service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL, + quota_project_id=QUOTA_PROJECT_ID, + scopes=SCOPES, + ) + + credentials.refresh(request) + + assert len(request.call_args_list) == 5 + # Fourth request should be sent to GCP STS endpoint. + self.assert_token_request_kwargs( + request.call_args_list[3].kwargs, token_headers, token_request_data + ) + # Fifth request should be sent to iamcredentials endpoint for service + # account impersonation. + self.assert_impersonation_request_kwargs( + request.call_args_list[4].kwargs, + impersonation_headers, + impersonation_request_data, + ) + assert credentials.token == impersonation_response["accessToken"] + assert credentials.quota_project_id == QUOTA_PROJECT_ID + assert credentials.scopes == SCOPES + + def test_refresh_with_retrieve_subject_token_error(self): + request = self.make_mock_request(region_status=http_client.BAD_REQUEST) + credentials = self.make_credentials(credential_source=self.CREDENTIAL_SOURCE) + + with pytest.raises(exceptions.RefreshError) as excinfo: + credentials.refresh(request) + + assert excinfo.match(r"Unable to retrieve AWS region")