| # 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") |