Add support for EKS container credentials

.changes/next-release/feature-ContainerProvider-78717.json

@@ -0,0 +1,5 @@
+{
+  "type": "feature",
+  "category": "ContainerProvider",
+  "description": "Added Support for EKS container credentials"
+}

awscli/botocore/credentials.py

@@ -1809,6 +1809,7 @@ class ContainerProvider(CredentialProvider):
     ENV_VAR = 'AWS_CONTAINER_CREDENTIALS_RELATIVE_URI'
     ENV_VAR_FULL = 'AWS_CONTAINER_CREDENTIALS_FULL_URI'
     ENV_VAR_AUTH_TOKEN = 'AWS_CONTAINER_AUTHORIZATION_TOKEN'
+    ENV_VAR_AUTH_TOKEN_FILE = 'AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE'
 
     def __init__(self, environ=None, fetcher=None):
         if environ is None:
@@ -1842,12 +1843,20 @@ def _retrieve_or_fail(self):
         )
 
     def _build_headers(self):
-        headers = {}
-        auth_token = self._environ.get(self.ENV_VAR_AUTH_TOKEN)
+        auth_token = None
+        if self.ENV_VAR_AUTH_TOKEN_FILE in self._environ:
+            auth_token_file_path = self._environ[self.ENV_VAR_AUTH_TOKEN_FILE]
+            with open(auth_token_file_path) as token_file:
+                auth_token = token_file.read()
+        elif self.ENV_VAR_AUTH_TOKEN in self._environ:
+            auth_token = self._environ[self.ENV_VAR_AUTH_TOKEN]
         if auth_token is not None:
-            return {
-                'Authorization': auth_token
-            }
+            self._validate_auth_token(auth_token)
+            return {'Authorization': auth_token}
+
+    def _validate_auth_token(self, auth_token):
+        if "\r" in auth_token or "\n" in auth_token:
+            raise ValueError("Auth token value is not a legal header value")
 
     def _create_fetcher(self, full_uri, headers):
         def fetch_creds():

awscli/botocore/utils.py

@@ -27,6 +27,7 @@
 import warnings
 import weakref
 from datetime import datetime as _DatetimeClass
+from ipaddress import ip_address
 from pathlib import Path
 
 import botocore
@@ -2715,7 +2716,12 @@ class ContainerMetadataFetcher(object):
     RETRY_ATTEMPTS = 3
     SLEEP_TIME = 1
     IP_ADDRESS = '169.254.170.2'
-    _ALLOWED_HOSTS = [IP_ADDRESS, 'localhost', '127.0.0.1']
+    _ALLOWED_HOSTS = [
+        IP_ADDRESS,
+        '169.254.170.23',
+        'fd00:ec2::23',
+        'localhost',
+    ]
 
     def __init__(self, session=None, sleep=time.sleep):
         if session is None:
@@ -2739,21 +2745,30 @@ def retrieve_full_uri(self, full_url, headers=None):
 
     def _validate_allowed_url(self, full_url):
         parsed = botocore.compat.urlparse(full_url)
+        if self._is_loopback_address(parsed.hostname):
+            return
         is_whitelisted_host = self._check_if_whitelisted_host(
             parsed.hostname)
         if not is_whitelisted_host:
             raise ValueError(
-                "Unsupported host '%s'.  Can only "
-                "retrieve metadata from these hosts: %s" %
-                (parsed.hostname, ', '.join(self._ALLOWED_HOSTS)))
+                f"Unsupported host '{parsed.hostname}'.  Can only retrieve metadata "
+                f"from a loopback address or one of these hosts: {', '.join(self._ALLOWED_HOSTS)}"
+            )
+
+    def _is_loopback_address(self, hostname):
+        try:
+            ip = ip_address(hostname)
+            return ip.is_loopback
+        except ValueError:
+            return False
 
     def _check_if_whitelisted_host(self, host):
         if host in self._ALLOWED_HOSTS:
             return True
         return False
 
     def retrieve_uri(self, relative_uri):
-        """Retrieve JSON metadata from ECS metadata.
+        """Retrieve JSON metadata from container metadata.
 
         :type relative_uri: str
         :param relative_uri: A relative URI, e.g "/foo/bar?id=123"
@@ -2790,19 +2805,21 @@ def _get_response(self, full_url, headers, timeout):
             if response.status_code != 200:
                 raise MetadataRetrievalError(
                     error_msg=(
-                        "Received non 200 response (%s) from ECS metadata: %s"
-                    ) % (response.status_code, response_text))
+                        f"Received non 200 response {response.status_code} "
+                        f"from container metadata: {response_text}"
+                    )
+                )
             try:
                 return json.loads(response_text)
             except ValueError:
-                error_msg = (
-                    "Unable to parse JSON returned from ECS metadata services"
-                )
+                error_msg = "Unable to parse JSON returned from container metadata services"
                 logger.debug('%s:%s', error_msg, response_text)
                 raise MetadataRetrievalError(error_msg=error_msg)
         except RETRYABLE_HTTP_ERRORS as e:
-            error_msg = ("Received error when attempting to retrieve "
-                         "ECS metadata: %s" % e)
+            error_msg = (
+                "Received error when attempting to retrieve "
+                f"container metadata: {e}"
+            )
             raise MetadataRetrievalError(error_msg=error_msg)
 
     def full_url(self, relative_uri):

tests/unit/botocore/test_credentials.py

@@ -2805,6 +2805,14 @@ def test_refresh_giving_expired_credentials_raises_exception(self):
 
 
 class TestContainerProvider(BaseEnvVar):
+    def setUp(self):
+        super().setUp()
+        self.tempdir = tempfile.mkdtemp()
+
+    def tearDown(self):
+        super().tearDown()
+        shutil.rmtree(self.tempdir)
+
     def test_noop_if_env_var_is_not_set(self):
         # The 'AWS_CONTAINER_CREDENTIALS_RELATIVE_URI' env var
         # is not present as an env var.
@@ -2962,6 +2970,58 @@ def test_can_pass_basic_auth_token(self):
         self.assertEqual(creds.token, 'token')
         self.assertEqual(creds.method, 'container-role')
 
+    def test_can_pass_auth_token_from_file(self):
+        token_file_path = os.path.join(self.tempdir, 'token.jwt')
+        with open(token_file_path, 'w') as token_file:
+            token_file.write('Basic auth-token')
+        environ = {
+            'AWS_CONTAINER_CREDENTIALS_FULL_URI': 'http://localhost/foo',
+            'AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE': token_file_path,
+        }
+        fetcher = self.create_fetcher()
+        timeobj = datetime.now(tzlocal())
+        timestamp = (timeobj + timedelta(hours=24)).isoformat()
+        fetcher.retrieve_full_uri.return_value = {
+            "AccessKeyId": "access_key",
+            "SecretAccessKey": "secret_key",
+            "Token": "token",
+            "Expiration": timestamp,
+        }
+        provider = credentials.ContainerProvider(environ, fetcher)
+        creds = provider.load()
+
+        fetcher.retrieve_full_uri.assert_called_with(
+            'http://localhost/foo',
+            headers={'Authorization': 'Basic auth-token'},
+        )
+        self.assertEqual(creds.access_key, 'access_key')
+        self.assertEqual(creds.secret_key, 'secret_key')
+        self.assertEqual(creds.token, 'token')
+        self.assertEqual(creds.method, 'container-role')
+
+    def test_throws_error_on_invalid_token_file(self):
+        token_file_path = '/some/path/token.jwt'
+        environ = {
+            'AWS_CONTAINER_CREDENTIALS_FULL_URI': 'http://localhost/foo',
+            'AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE': token_file_path,
+        }
+        fetcher = self.create_fetcher()
+        provider = credentials.ContainerProvider(environ, fetcher)
+
+        with self.assertRaises(FileNotFoundError):
+            provider.load()
+
+    def test_throws_error_on_illegal_header(self):
+        environ = {
+            'AWS_CONTAINER_CREDENTIALS_FULL_URI': 'http://localhost/foo',
+            'AWS_CONTAINER_AUTHORIZATION_TOKEN': 'invalid\r\ntoken',
+        }
+        fetcher = self.create_fetcher()
+        provider = credentials.ContainerProvider(environ, fetcher)
+
+        with self.assertRaises(ValueError):
+            provider.load()
+
 
 class TestProcessProvider(BaseEnvVar):
     def setUp(self):

tests/unit/botocore/test_utils.py

@@ -2053,9 +2053,12 @@ def get_s3_outpost_request(self, **s3_request_kwargs):
         request.context['s3_accesspoint'] = accesspoint_context
         return request
 
-    def get_s3_accesspoint_request(self, accesspoint_name=None,
-                                   accesspoint_context=None,
-                                   **s3_request_kwargs):
+    def get_s3_accesspoint_request(
+        self,
+        accesspoint_name=None,
+        accesspoint_context=None,
+        **s3_request_kwargs,
+    ):
         if not accesspoint_name:
             accesspoint_name = self.accesspoint_name
         request = self.get_s3_request(accesspoint_name, **s3_request_kwargs)
@@ -2385,8 +2388,8 @@ def test_error_raised_on_no_json_response(self):
         self.assertEqual(self.http.send.call_count, fetcher.RETRY_ATTEMPTS)
 
     def test_can_retrieve_full_uri_with_fixed_ip(self):
-        self.assert_can_retrieve_metadata_from(
-            'http://%s/foo?id=1' % ContainerMetadataFetcher.IP_ADDRESS)
+        uri = f'http://{ContainerMetadataFetcher.IP_ADDRESS}/foo?id=1'
+        self.assert_can_retrieve_metadata_from(uri)
 
     def test_localhost_http_is_allowed(self):
         self.assert_can_retrieve_metadata_from('http://localhost/foo')
@@ -2403,6 +2406,21 @@ def test_can_use_127_ip_addr(self):
     def test_can_use_127_ip_addr_with_port(self):
         self.assert_can_retrieve_metadata_from('https://127.0.0.1:8080/foo')
 
+    def test_can_use_eks_ipv4_addr(self):
+        uri = 'http://169.254.170.23/credentials'
+        self.assert_can_retrieve_metadata_from(uri)
+
+    def test_can_use_eks_ipv6_addr(self):
+        uri = 'http://[fd00:ec2::23]/credentials'
+        self.assert_can_retrieve_metadata_from(uri)
+
+    def test_can_use_eks_ipv6_addr_with_port(self):
+        uri = 'https://[fd00:ec2::23]:8000'
+        self.assert_can_retrieve_metadata_from(uri)
+
+    def test_can_use_loopback_v6_uri(self):
+        self.assert_can_retrieve_metadata_from('http://[::1]/credentials')
+
     def test_link_local_http_is_not_allowed(self):
         self.assert_host_is_not_allowed('http://169.254.0.1/foo')