ydb/contrib/python/google-auth/py2/tests/test_aws.py

# Copyright 2020 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

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"
TOKEN = "IQoJb3JpZ2luX2VjEIz//////////wEaCXVzLWVhc3QtMiJGMEQCIH7MHX/Oy/OB8OlLQa9GrqU1B914+iMikqWQW7vPCKlgAiA/Lsv8Jcafn14owfxXn95FURZNKaaphj0ykpmS+Ki+CSq0AwhlEAAaDDA3NzA3MTM5MTk5NiIMx9sAeP1ovlMTMKLjKpEDwuJQg41/QUKx0laTZYjPlQvjwSqS3OB9P1KAXPWSLkliVMMqaHqelvMF/WO/glv3KwuTfQsavRNs3v5pcSEm4SPO3l7mCs7KrQUHwGP0neZhIKxEXy+Ls//1C/Bqt53NL+LSbaGv6RPHaX82laz2qElphg95aVLdYgIFY6JWV5fzyjgnhz0DQmy62/Vi8pNcM2/VnxeCQ8CC8dRDSt52ry2v+nc77vstuI9xV5k8mPtnaPoJDRANh0bjwY5Sdwkbp+mGRUJBAQRlNgHUJusefXQgVKBCiyJY4w3Csd8Bgj9IyDV+Azuy1jQqfFZWgP68LSz5bURyIjlWDQunO82stZ0BgplKKAa/KJHBPCp8Qi6i99uy7qh76FQAqgVTsnDuU6fGpHDcsDSGoCls2HgZjZFPeOj8mmRhFk1Xqvkbjuz8V1cJk54d3gIJvQt8gD2D6yJQZecnuGWd5K2e2HohvCc8Fc9kBl1300nUJPV+k4tr/A5R/0QfEKOZL1/k5lf1g9CREnrM8LVkGxCgdYMxLQow1uTL+QU67AHRRSp5PhhGX4Rek+01vdYSnJCMaPhSEgcLqDlQkhk6MPsyT91QMXcWmyO+cAZwUPwnRamFepuP4K8k2KVXs/LIJHLELwAZ0ekyaS7CptgOqS7uaSTFG3U+vzFZLEnGvWQ7y9IPNQZ+Dffgh4p3vF4J68y9049sI6Sr5d5wbKkcbm8hdCDHZcv4lnqohquPirLiFQ3q7B17V9krMPu3mz1cg4Ekgcrn/E09NTsxAqD8NcZ7C7ECom9r+X3zkDOxaajW6hu3Az8hGlyylDaMiFfRbBJpTIlxp7jfa7CxikNgNtEKLH9iCzvuSg2vhA=="
# To avoid json.dumps() differing behavior from one version to other,
# the JSON payload is hardcoded.
REQUEST_PARAMS = '{"KeySchema":[{"KeyType":"HASH","AttributeName":"Id"}],"TableName":"TestTable","AttributeDefinitions":[{"AttributeName":"Id","AttributeType":"S"}],"ProvisionedThroughput":{"WriteCapacityUnits":5,"ReadCapacityUnits":5}}'
# Each tuple contains the following entries:
# region, time, credentials, original_request, signed_request
TEST_FIXTURES = [
    # GET request (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=b27ccfbfa7df52a200ff74193ca6e32d4b48b8856fab7ebf1c595d0670a7e470",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with relative path (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-relative-relative.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-relative-relative.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/foo/bar/../..",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/foo/bar/../..",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=b27ccfbfa7df52a200ff74193ca6e32d4b48b8856fab7ebf1c595d0670a7e470",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with /./ path (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-slash-dot-slash.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-slash-dot-slash.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/./",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/./",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=b27ccfbfa7df52a200ff74193ca6e32d4b48b8856fab7ebf1c595d0670a7e470",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with pointless dot path (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-slash-pointless-dot.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-slash-pointless-dot.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/./foo",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/./foo",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=910e4d6c9abafaf87898e1eb4c929135782ea25bb0279703146455745391e63a",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with utf8 path (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-utf8.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-utf8.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/%E1%88%B4",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/%E1%88%B4",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=8d6634c189aa8c75c2e51e106b6b5121bed103fdb351f7d7d4381c738823af74",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with duplicate query key (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla-query-order-key-case.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla-query-order-key-case.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/?foo=Zoo&foo=aha",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/?foo=Zoo&foo=aha",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=be7148d34ebccdc6423b19085378aa0bee970bdc61d144bd1a8c48c33079ab09",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with duplicate out of order query key (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla-query-order-value.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla-query-order-value.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/?foo=b&foo=a",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/?foo=b&foo=a",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=feb926e49e382bec75c9d7dcb2a1b6dc8aa50ca43c25d2bc51143768c0875acc",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with utf8 query (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla-ut8-query.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-vanilla-ut8-query.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "GET",
            "url": "https://host.foo.com/?{}=bar".format(
                urllib.parse.unquote("%E1%88%B4")
            ),
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/?{}=bar".format(
                urllib.parse.unquote("%E1%88%B4")
            ),
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=6fb359e9a05394cc7074e0feb42573a2601abc0c869a953e8c5c12e4e01f1a8c",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # POST request with sorted headers (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-header-key-sort.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-header-key-sort.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "POST",
            "url": "https://host.foo.com/",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT", "ZOO": "zoobar"},
        },
        {
            "url": "https://host.foo.com/",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host;zoo, Signature=b7a95a52518abbca0964a999a880429ab734f35ebbf1235bd79a5de87756dc4a",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
                "ZOO": "zoobar",
            },
        },
    ),
    # POST request with upper case header value from AWS Python test harness.
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-header-value-case.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-header-value-case.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "POST",
            "url": "https://host.foo.com/",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT", "zoo": "ZOOBAR"},
        },
        {
            "url": "https://host.foo.com/",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host;zoo, Signature=273313af9d0c265c531e11db70bbd653f3ba074c1009239e8559d3987039cad7",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
                "zoo": "ZOOBAR",
            },
        },
    ),
    # POST request with header and no body (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-header-value-trim.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/get-header-value-trim.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "POST",
            "url": "https://host.foo.com/",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT", "p": "phfft"},
        },
        {
            "url": "https://host.foo.com/",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host;p, Signature=debf546796015d6f6ded8626f5ce98597c33b47b9164cf6b17b4642036fcb592",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
                "p": "phfft",
            },
        },
    ),
    # POST request with body and no header (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-x-www-form-urlencoded.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-x-www-form-urlencoded.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "POST",
            "url": "https://host.foo.com/",
            "headers": {
                "Content-Type": "application/x-www-form-urlencoded",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
            "data": "foo=bar",
        },
        {
            "url": "https://host.foo.com/",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=content-type;date;host, Signature=5a15b22cf462f047318703b92e6f4f38884e4a7ab7b1d6426ca46a8bd1c26cbc",
                "host": "host.foo.com",
                "Content-Type": "application/x-www-form-urlencoded",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
            "data": "foo=bar",
        },
    ),
    # POST request with querystring (AWS botocore tests).
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-vanilla-query.req
    # https://github.com/boto/botocore/blob/879f8440a4e9ace5d3cf145ce8b3d5e5ffb892ef/tests/unit/auth/aws4_testsuite/post-vanilla-query.sreq
    (
        "us-east-1",
        "2011-09-09T23:36:00Z",
        {
            "access_key_id": "AKIDEXAMPLE",
            "secret_access_key": "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
        },
        {
            "method": "POST",
            "url": "https://host.foo.com/?foo=bar",
            "headers": {"date": "Mon, 09 Sep 2011 23:36:00 GMT"},
        },
        {
            "url": "https://host.foo.com/?foo=bar",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20110909/us-east-1/host/aws4_request, SignedHeaders=date;host, Signature=b6e3b79003ce0743a491606ba1035a804593b0efb1e20a11cba83f8c25a57a92",
                "host": "host.foo.com",
                "date": "Mon, 09 Sep 2011 23:36:00 GMT",
            },
        },
    ),
    # GET request with session token credentials.
    (
        "us-east-2",
        "2020-08-11T06:55:22Z",
        {
            "access_key_id": ACCESS_KEY_ID,
            "secret_access_key": SECRET_ACCESS_KEY,
            "security_token": TOKEN,
        },
        {
            "method": "GET",
            "url": "https://ec2.us-east-2.amazonaws.com?Action=DescribeRegions&Version=2013-10-15",
        },
        {
            "url": "https://ec2.us-east-2.amazonaws.com?Action=DescribeRegions&Version=2013-10-15",
            "method": "GET",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential="
                + ACCESS_KEY_ID
                + "/20200811/us-east-2/ec2/aws4_request, SignedHeaders=host;x-amz-date;x-amz-security-token, Signature=631ea80cddfaa545fdadb120dc92c9f18166e38a5c47b50fab9fce476e022855",
                "host": "ec2.us-east-2.amazonaws.com",
                "x-amz-date": "20200811T065522Z",
                "x-amz-security-token": TOKEN,
            },
        },
    ),
    # POST request with session token credentials.
    (
        "us-east-2",
        "2020-08-11T06:55:22Z",
        {
            "access_key_id": ACCESS_KEY_ID,
            "secret_access_key": SECRET_ACCESS_KEY,
            "security_token": TOKEN,
        },
        {
            "method": "POST",
            "url": "https://sts.us-east-2.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15",
        },
        {
            "url": "https://sts.us-east-2.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential="
                + ACCESS_KEY_ID
                + "/20200811/us-east-2/sts/aws4_request, SignedHeaders=host;x-amz-date;x-amz-security-token, Signature=73452984e4a880ffdc5c392355733ec3f5ba310d5e0609a89244440cadfe7a7a",
                "host": "sts.us-east-2.amazonaws.com",
                "x-amz-date": "20200811T065522Z",
                "x-amz-security-token": TOKEN,
            },
        },
    ),
    # POST request with computed x-amz-date and no data.
    (
        "us-east-2",
        "2020-08-11T06:55:22Z",
        {"access_key_id": ACCESS_KEY_ID, "secret_access_key": SECRET_ACCESS_KEY},
        {
            "method": "POST",
            "url": "https://sts.us-east-2.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15",
        },
        {
            "url": "https://sts.us-east-2.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential="
                + ACCESS_KEY_ID
                + "/20200811/us-east-2/sts/aws4_request, SignedHeaders=host;x-amz-date, Signature=d095ba304919cd0d5570ba8a3787884ee78b860f268ed040ba23831d55536d56",
                "host": "sts.us-east-2.amazonaws.com",
                "x-amz-date": "20200811T065522Z",
            },
        },
    ),
    # POST request with session token and additional headers/data.
    (
        "us-east-2",
        "2020-08-11T06:55:22Z",
        {
            "access_key_id": ACCESS_KEY_ID,
            "secret_access_key": SECRET_ACCESS_KEY,
            "security_token": TOKEN,
        },
        {
            "method": "POST",
            "url": "https://dynamodb.us-east-2.amazonaws.com/",
            "headers": {
                "Content-Type": "application/x-amz-json-1.0",
                "x-amz-target": "DynamoDB_20120810.CreateTable",
            },
            "data": REQUEST_PARAMS,
        },
        {
            "url": "https://dynamodb.us-east-2.amazonaws.com/",
            "method": "POST",
            "headers": {
                "Authorization": "AWS4-HMAC-SHA256 Credential="
                + ACCESS_KEY_ID
                + "/20200811/us-east-2/dynamodb/aws4_request, SignedHeaders=content-type;host;x-amz-date;x-amz-security-token;x-amz-target, Signature=fdaa5b9cc9c86b80fe61eaf504141c0b3523780349120f2bd8145448456e0385",
                "host": "dynamodb.us-east-2.amazonaws.com",
                "x-amz-date": "20200811T065522Z",
                "Content-Type": "application/x-amz-json-1.0",
                "x-amz-target": "DynamoDB_20120810.CreateTable",
                "x-amz-security-token": TOKEN,
            },
            "data": REQUEST_PARAMS,
        },
    ),
]


class TestRequestSigner(object):
    @pytest.mark.parametrize(
        "region, time, credentials, original_request, signed_request", TEST_FIXTURES
    )
    @mock.patch("google.auth._helpers.utcnow")
    def test_get_request_options(
        self, utcnow, region, time, credentials, original_request, signed_request
    ):
        utcnow.return_value = datetime.datetime.strptime(time, "%Y-%m-%dT%H:%M:%SZ")
        request_signer = aws.RequestSigner(region)
        actual_signed_request = request_signer.get_request_options(
            credentials,
            original_request.get("url"),
            original_request.get("method"),
            original_request.get("data"),
            original_request.get("headers"),
        )

        assert actual_signed_request == signed_request

    def test_get_request_options_with_missing_scheme_url(self):
        request_signer = aws.RequestSigner("us-east-2")

        with pytest.raises(ValueError) as excinfo:
            request_signer.get_request_options(
                {
                    "access_key_id": ACCESS_KEY_ID,
                    "secret_access_key": SECRET_ACCESS_KEY,
                },
                "invalid",
                "POST",
            )

        assert excinfo.match(r"Invalid AWS service URL")

    def test_get_request_options_with_invalid_scheme_url(self):
        request_signer = aws.RequestSigner("us-east-2")

        with pytest.raises(ValueError) as excinfo:
            request_signer.get_request_options(
                {
                    "access_key_id": ACCESS_KEY_ID,
                    "secret_access_key": SECRET_ACCESS_KEY,
                },
                "http://invalid",
                "POST",
            )

        assert excinfo.match(r"Invalid AWS service URL")

    def test_get_request_options_with_missing_hostname_url(self):
        request_signer = aws.RequestSigner("us-east-2")

        with pytest.raises(ValueError) as excinfo:
            request_signer.get_request_options(
                {
                    "access_key_id": ACCESS_KEY_ID,
                    "secret_access_key": SECRET_ACCESS_KEY,
                },
                "https://",
                "POST",
            )

        assert excinfo.match(r"Invalid AWS service URL")


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,
        default_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,
            default_scopes=default_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(ValueError) 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(ValueError) 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(ValueError) 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(ValueError) 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_info(self):
        credentials = self.make_credentials(
            credential_source=self.CREDENTIAL_SOURCE.copy()
        )

        assert credentials.info == {
            "type": "external_account",
            "audience": AUDIENCE,
            "subject_token_type": SUBJECT_TOKEN_TYPE,
            "token_url": TOKEN_URL,
            "credential_source": self.CREDENTIAL_SOURCE,
        }

    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][1], REGION_URL
        )
        # Assert role request.
        self.assert_aws_metadata_request_kwargs(
            request.call_args_list[1][1], SECURITY_CREDS_URL
        )
        # Assert security credentials request.
        self.assert_aws_metadata_request_kwargs(
            request.call_args_list[2][1],
            "{}/{}".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][1], SECURITY_CREDS_URL
        )
        # Assert security credentials request.
        self.assert_aws_metadata_request_kwargs(
            new_request.call_args_list[1][1],
            "{}/{}".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_with_default_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)
        monkeypatch.setenv(environment_vars.AWS_DEFAULT_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_with_both_regions_set(
        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_DEFAULT_REGION, "Malformed AWS Region")
        # This test makes sure that the AWS_REGION gets used over AWS_DEFAULT_REGION,
        # So, AWS_DEFAULT_REGION is set to something that would cause the test to fail,
        # And AWS_REGION is set to the a valid value, and it should succeed
        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_ignore_default_scopes(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,
            # Default scopes should be ignored.
            default_scopes=["ignored"],
        )

        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][1], 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
        assert credentials.default_scopes == ["ignored"]

    @mock.patch("google.auth._helpers.utcnow")
    def test_refresh_success_without_impersonation_use_default_scopes(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=None,
            # Default scopes should be used since user specified scopes are none.
            default_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][1], token_headers, token_request_data
        )
        assert credentials.token == self.SUCCESS_RESPONSE["access_token"]
        assert credentials.quota_project_id == QUOTA_PROJECT_ID
        assert credentials.scopes is None
        assert credentials.default_scopes == SCOPES

    @mock.patch("google.auth._helpers.utcnow")
    def test_refresh_success_with_impersonation_ignore_default_scopes(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,
            # Default scopes should be ignored.
            default_scopes=["ignored"],
        )

        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][1], 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][1],
            impersonation_headers,
            impersonation_request_data,
        )
        assert credentials.token == impersonation_response["accessToken"]
        assert credentials.quota_project_id == QUOTA_PROJECT_ID
        assert credentials.scopes == SCOPES
        assert credentials.default_scopes == ["ignored"]

    @mock.patch("google.auth._helpers.utcnow")
    def test_refresh_success_with_impersonation_use_default_scopes(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=None,
            # Default scopes should be used since user specified scopes are none.
            default_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][1], 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][1],
            impersonation_headers,
            impersonation_request_data,
        )
        assert credentials.token == impersonation_response["accessToken"]
        assert credentials.quota_project_id == QUOTA_PROJECT_ID
        assert credentials.scopes is None
        assert credentials.default_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")