[v2] Register feature ids for IMDS and HTTP credentials (aws#9745)

Author: aemous

awscli/botocore/credentials.py

@@ -41,6 +41,7 @@
     UnknownCredentialError,
 )
 from botocore.tokens import SSOTokenProvider
+from botocore.useragent import register_feature_id
 from botocore.utils import (
     ArnParser,
     ContainerMetadataFetcher,
@@ -1130,6 +1131,7 @@ def load(self):
         metadata = fetcher.retrieve_iam_role_credentials()
         if not metadata:
             return None
+        register_feature_id('CREDENTIALS_IMDS')
         logger.debug(
             'Found credentials from IAM Role: %s', metadata['role_name']
         )
@@ -2056,6 +2058,7 @@ def fetch_creds():
                 response = self._fetcher.retrieve_full_uri(
                     full_uri, headers=headers
                 )
+                register_feature_id('CREDENTIALS_HTTP')
             except MetadataRetrievalError as e:
                 logger.debug(
                     "Error retrieving container metadata: %s", e, exc_info=True

awscli/botocore/useragent.py

@@ -76,6 +76,8 @@
     'FLEXIBLE_CHECKSUMS_REQ_WHEN_REQUIRED': 'a',
     'FLEXIBLE_CHECKSUMS_RES_WHEN_SUPPORTED': 'b',
     'FLEXIBLE_CHECKSUMS_RES_WHEN_REQUIRED': 'c',
+    'CREDENTIALS_HTTP': 'z',
+    'CREDENTIALS_IMDS': '0',
     'BEARER_SERVICE_ENV_VARS': '3',
 }
 

tests/functional/botocore/test_credentials.py

@@ -19,7 +19,8 @@
 import threading
 import time
 import uuid
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
 
 import pytest
 from botocore import UNSIGNED
@@ -51,6 +52,7 @@
 
 from tests import (
     BaseEnvVar,
+    ClientHTTPStubber,
     BaseSessionTest,
     IntegerRefresher,
     SessionHTTPStubber,
@@ -60,9 +62,13 @@
     temporary_file,
     unittest,
 )
+from tests.functional.test_useragent import (
+    get_captured_ua_strings,
+    parse_registered_feature_ids,
+)
 
-TIME_IN_ONE_HOUR = datetime.utcnow() + timedelta(hours=1)
-TIME_IN_SIX_MONTHS = datetime.utcnow() + timedelta(hours=4320)
+TIME_IN_ONE_HOUR = datetime.now(tz=timezone.utc) + timedelta(hours=1)
+TIME_IN_SIX_MONTHS = datetime.now(tz=timezone.utc) + timedelta(hours=4320)
 
 
 class TestCredentialRefreshRaces(unittest.TestCase):
@@ -1085,3 +1091,91 @@ def add_credential_response(self, stubber):
             }
         }
         stubber.add_response(body=json.dumps(response).encode('utf-8'))
+
+
+class TestFeatureIdRegistered:
+    @patch(
+        "botocore.utils.InstanceMetadataFetcher.retrieve_iam_role_credentials"
+    )
+    @patch("botocore.credentials.ContainerProvider.load", return_value=None)
+    @patch("botocore.credentials.ConfigProvider.load", return_value=None)
+    @patch(
+        "botocore.credentials.SharedCredentialProvider.load", return_value=None
+    )
+    @patch("botocore.credentials.EnvProvider.load", return_value=None)
+    def test_user_agent_has_imds_credentials_feature_id(
+        self,
+        _unused_mock_env_load,
+        _unused_mock_shared_load,
+        _unused_mock_config_load,
+        _unused_mock_container_load,
+        mock_retrieve_iam_role_credentials,
+        patched_session,
+    ):
+        fake_creds = {
+            "role_name": "FAKEROLE",
+            "access_key": "FAKEACCESSKEY",
+            "secret_key": "FAKESECRET",
+            "token": "FAKETOKEN",
+            "expiry_time": "2099-01-01T00:00:00Z",
+        }
+        mock_retrieve_iam_role_credentials.return_value = fake_creds
+
+        client = patched_session.create_client("s3", region_name="us-east-1")
+        with ClientHTTPStubber(client, strict=True) as http_stubber:
+            # We want to call this twice to assert that the feature id exists
+            # for multiple calls with the same credentials
+            http_stubber.add_response()
+            http_stubber.add_response()
+            client.list_buckets()
+            client.list_buckets()
+
+        ua_string = get_captured_ua_strings(http_stubber)
+        feature_list_one = parse_registered_feature_ids(ua_string[0])
+        feature_list_two = parse_registered_feature_ids(ua_string[1])
+        assert '0' in feature_list_one and '0' in feature_list_two
+
+    @patch("botocore.credentials.ContainerMetadataFetcher.retrieve_full_uri")
+    @patch("botocore.credentials.ConfigProvider.load", return_value=None)
+    @patch(
+        "botocore.credentials.SharedCredentialProvider.load", return_value=None
+    )
+    @patch("botocore.credentials.EnvProvider.load", return_value=None)
+    def test_user_agent_has_http_credentials_feature_id(
+        self,
+        _unused_mock_env_load,
+        _unused_mock_shared_load,
+        _unused_mock_config_load,
+        mock_load_http_credentials,
+        monkeypatch,
+        patched_session,
+    ):
+        environ = {
+            'AWS_CONTAINER_CREDENTIALS_FULL_URI': 'http://localhost/foo',
+            'AWS_CONTAINER_AUTHORIZATION_TOKEN': 'Basic auth-token',
+        }
+        for var in environ:
+            monkeypatch.setenv(var, environ[var])
+
+        fake_creds = {
+            "AccessKeyId": "FAKEACCESSKEY",
+            "SecretAccessKey": "FAKESECRET",
+            "Token": "FAKETOKEN",
+            "Expiration": "2099-01-01T00:00:00Z",
+            "AccountId": "01234567890",
+        }
+        mock_load_http_credentials.return_value = fake_creds
+
+        client = patched_session.create_client("s3", region_name="us-east-1")
+        with ClientHTTPStubber(client, strict=True) as http_stubber:
+            # We want to call this twice to assert that the feature id exists
+            # for multiple calls with the same credentials
+            http_stubber.add_response()
+            http_stubber.add_response()
+            client.list_buckets()
+            client.list_buckets()
+
+        ua_string = get_captured_ua_strings(http_stubber)
+        feature_list_one = parse_registered_feature_ids(ua_string[0])
+        feature_list_two = parse_registered_feature_ids(ua_string[1])
+        assert 'z' in feature_list_one and 'z' in feature_list_two