From e7f98b4bb174ac4e636c4e28a6779cee441a111a Mon Sep 17 00:00:00 2001 From: kyleknap Date: Mon, 11 Dec 2023 10:51:52 -0800 Subject: [PATCH] Plumb IMDS v1 opt out configuration to region provider The AWS CLI v2 currently has a seperate, but similar, implementation from upstream botocore. In the future, we should be able to consolidate the implementations to make porting more streamlined. --- awscli/utils.py | 5 ++- tests/unit/test_utils.py | 77 +++++++++++++++++++++++----------------- 2 files changed, 49 insertions(+), 33 deletions(-) diff --git a/awscli/utils.py b/awscli/utils.py index a062d8acaca3..fcaa594d0910 100644 --- a/awscli/utils.py +++ b/awscli/utils.py @@ -135,7 +135,10 @@ def _create_fetcher(self): 'ec2_metadata_service_endpoint'), 'ec2_metadata_service_endpoint_mode': resolve_imds_endpoint_mode( self._session - ) + ), + 'ec2_metadata_v1_disabled': self._session.get_config_variable( + 'ec2_metadata_v1_disabled' + ), } fetcher = InstanceMetadataRegionFetcher( timeout=metadata_timeout, diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 1e422c41d073..2d777bc083f0 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -21,7 +21,7 @@ import botocore import botocore.model import botocore.session as session -from botocore.exceptions import ConnectionClosedError +from botocore.exceptions import ConnectionClosedError, MetadataRetrievalError from awscli.clidriver import create_clidriver from awscli.testutils import unittest, skip_if_windows, mock from awscli.compat import is_windows @@ -316,7 +316,7 @@ def test_not_create_process_if_stream_not_created(self): self.assertEqual(self.popen.call_count, 0) -class TestInstanceMetadataRegionFetcher(unittest.TestCase): +class BaseIMDSRegionTest(unittest.TestCase): def setUp(self): urllib3_session_send = 'botocore.httpsession.URLLib3Session.send' self._urllib3_patch = mock.patch(urllib3_session_send) @@ -324,9 +324,13 @@ def setUp(self): self._imds_responses = [] self._send.side_effect = self.get_imds_response self._region = 'us-mars-1a' + self.environ = {} + self.environ_patch = mock.patch('os.environ', self.environ) + self.environ_patch.start() def tearDown(self): self._urllib3_patch.stop() + self.environ_patch.stop() def add_imds_response(self, body, status_code=200): response = botocore.awsrequest.AWSResponse( @@ -354,6 +358,8 @@ def get_imds_response(self, request): raise response return response + +class TestInstanceMetadataRegionFetcher(BaseIMDSRegionTest): def test_disabled_by_environment(self): env = {'AWS_EC2_METADATA_DISABLED': 'true'} fetcher = InstanceMetadataRegionFetcher(env=env) @@ -454,15 +460,7 @@ def test_exhaust_retries_on_region_request(self): self.assertEqual(result, None) -class TestIMDSRegionProvider(unittest.TestCase): - def setUp(self): - self.environ = {} - self.environ_patch = mock.patch('os.environ', self.environ) - self.environ_patch.start() - - def tearDown(self): - self.environ_patch.stop() - +class TestIMDSRegionProvider(BaseIMDSRegionTest): def assert_does_provide_expected_value(self, fetcher_region=None, expected_result=None,): fake_session = mock.Mock(spec=session.Session) @@ -484,74 +482,89 @@ def test_does_provide_none(self): expected_result=None, ) - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_use_truncated_user_agent(self, send): + def test_use_truncated_user_agent(self): driver = create_clidriver() driver.session.user_agent_version = '3.0' + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertEqual(args[0].headers['User-Agent'], 'aws-cli/3.0') - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_can_use_ipv6(self, send): + def test_can_use_ipv6(self): driver = create_clidriver() driver.session.set_config_variable('imds_use_ipv6', True) + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertIn('[fd00:ec2::254]', args[0].url) - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_use_ipv4_by_default(self, send): + def test_use_ipv4_by_default(self): driver = create_clidriver() + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertIn('169.254.169.254', args[0].url) - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_can_set_imds_endpoint_mode_to_ipv4(self, send): + def test_can_set_imds_endpoint_mode_to_ipv4(self): driver = create_clidriver() driver.session.set_config_variable( 'ec2_metadata_service_endpoint_mode', 'ipv4') + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertIn('169.254.169.254', args[0].url) - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_can_set_imds_endpoint_mode_to_ipv6(self, send): + def test_can_set_imds_endpoint_mode_to_ipv6(self): driver = create_clidriver() driver.session.set_config_variable( 'ec2_metadata_service_endpoint_mode', 'ipv6') + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertIn('[fd00:ec2::254]', args[0].url) - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_can_set_imds_service_endpoint(self, send): + def test_can_set_imds_service_endpoint(self): driver = create_clidriver() driver.session.set_config_variable( 'ec2_metadata_service_endpoint', 'http://myendpoint/') + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertIn('http://myendpoint/', args[0].url) - @mock.patch('botocore.httpsession.URLLib3Session.send') - def test_imds_service_endpoint_overrides_ipv6_endpoint(self, send): + def test_imds_service_endpoint_overrides_ipv6_endpoint(self): driver = create_clidriver() driver.session.set_config_variable( 'ec2_metadata_service_endpoint_mode', 'ipv6') driver.session.set_config_variable( 'ec2_metadata_service_endpoint', 'http://myendpoint/') + self.add_imds_token_response() + self.add_get_region_imds_response() provider = IMDSRegionProvider(driver.session) provider.provide() - args, _ = send.call_args + args, _ = self._send.call_args self.assertIn('http://myendpoint/', args[0].url) + def test_disable_ec2_metadata_v1(self): + driver = create_clidriver() + driver.session.set_config_variable('ec2_metadata_v1_disabled', True) + self.add_imds_response(b'', 404) + provider = IMDSRegionProvider(driver.session) + with self.assertRaises(MetadataRetrievalError): + provider.provide() + class TestLazyPager(unittest.TestCase): def setUp(self):