Change default S3 presigned URL signature from SigV2 to SigV4

SigV2 silently ignores conditional headers like IfNoneMatch and IfMatch,
creating security issues. SigV4 supports header signing and is the modern
AWS standard.

Breaking change: Users requiring SigV2 must explicitly set
signature_version='s3' in client config.

Author: adev-code

botocore/client.py

@@ -457,7 +457,7 @@ def _set_s3_presign_signature_version(
         # the customer hasn't set a signature version so we default the
         # signature version to sigv2.
         client_meta.events.register(
-            'choose-signer.s3', self._default_s3_presign_to_sigv2
+            'choose-signer.s3', self._default_s3_presign_to_sigv4
         )
 
     def _inject_s3_input_parameters(self, params, context, **kwargs):
@@ -469,11 +469,11 @@ def _inject_s3_input_parameters(self, params, context, **kwargs):
                     inject_parameter
                 ]
 
-    def _default_s3_presign_to_sigv2(self, signature_version, **kwargs):
+    def _default_s3_presign_to_sigv4(self, signature_version, **kwargs):
         """
-        Returns the 's3' (sigv2) signer if presigning an s3 request. This is
-        intended to be used to set the default signature version for the signer
-        to sigv2. Situations where an asymmetric signature is required are the
+        Returns the 's3v4' (sigv4) signer if presigning an s3 request. This is
+        intended to be used to set the default signature version for the signer to sigv4.
+        Situations where an asymmetric signature is required are the
         exception, for example MRAP needs v4a.
 
         :type signature_version: str
@@ -482,7 +482,7 @@ def _default_s3_presign_to_sigv2(self, signature_version, **kwargs):
         :type signing_name: str
         :param signing_name: The signing name of the service.
 
-        :return: 's3' if the request is an s3 presign request, None otherwise
+        :return: 's3v4' if the request is an s3 presign request, None otherwise
         """
         if signature_version.startswith('v4a'):
             return
@@ -492,7 +492,7 @@ def _default_s3_presign_to_sigv2(self, signature_version, **kwargs):
 
         for suffix in ['-query', '-presign-post']:
             if signature_version.endswith(suffix):
-                return f's3{suffix}'
+                return f's3v4{suffix}'
 
     def _register_importexport_events(
         self,

tests/functional/test_s3.py

@@ -2098,6 +2098,16 @@ def assert_is_v2_presigned_url(self, url):
         self.assertNotIn("X-Amz-Algorithm", qs_components)
         self.assertIn("Signature", qs_components)
 
+    def assert_is_v4_presigned_url(self, url):
+        qs_components = parse_qs(urlsplit(url).query)
+        # Assert that it looks like a v4 presigned url by asserting it has
+        # the v4 qs components.
+        self.assertIn("X-Amz-Credential", qs_components)
+        self.assertIn("X-Amz-Algorithm", qs_components)
+        self.assertEqual(
+            qs_components["X-Amz-Algorithm"], ["AWS4-HMAC-SHA256"]
+        )
+
     def test_generate_unauthed_url(self):
         config = Config(signature_version=botocore.UNSIGNED)
         client = self.session.create_client("s3", self.region, config=config)
@@ -2116,9 +2126,9 @@ def test_generate_unauthed_post(self):
         }
         self.assertEqual(parts, expected)
 
-    def test_default_presign_uses_sigv2(self):
+    def test_default_presign_uses_sigv4(self):
         url = self.client.generate_presigned_url(ClientMethod="list_buckets")
-        self.assertNotIn("Algorithm=AWS4-HMAC-SHA256", url)
+        self.assertIn("Algorithm=AWS4-HMAC-SHA256", url)
 
     def test_sigv4_presign(self):
         config = Config(signature_version="s3v4")
@@ -2210,43 +2220,43 @@ def test_presign_post_s3_accelerate(self):
         }
         self.assertEqual(parts, expected)
 
-    def test_presign_uses_v2_for_aws_global(self):
+    def test_presign_uses_v4_for_aws_global(self):
         client = self.session.create_client("s3", "aws-global")
         url = client.generate_presigned_url(
             "get_object", {"Bucket": "mybucket", "Key": "mykey"}
         )
-        self.assert_is_v2_presigned_url(url)
+        self.assert_is_v4_presigned_url(url)
 
-    def test_presign_uses_v2_for_default_region_with_us_east_1_regional(self):
+    def test_presign_uses_v4_for_default_region_with_us_east_1_regional(self):
         config = Config(s3={"us_east_1_regional_endpoint": "regional"})
         client = self.session.create_client("s3", config=config)
         url = client.generate_presigned_url(
             "get_object", {"Bucket": "mybucket", "Key": "mykey"}
         )
-        self.assert_is_v2_presigned_url(url)
+        self.assert_is_v4_presigned_url(url)
 
-    def test_presign_uses_v2_for_aws_global_with_us_east_1_regional(self):
+    def test_presign_uses_v4_for_aws_global_with_us_east_1_regional(self):
         config = Config(s3={"us_east_1_regional_endpoint": "regional"})
         client = self.session.create_client("s3", "aws-global", config=config)
         url = client.generate_presigned_url(
             "get_object", {"Bucket": "mybucket", "Key": "mykey"}
         )
-        self.assert_is_v2_presigned_url(url)
+        self.assert_is_v4_presigned_url(url)
 
-    def test_presign_uses_v2_for_us_east_1(self):
+    def test_presign_uses_v4_for_us_east_1(self):
         client = self.session.create_client("s3", "us-east-1")
         url = client.generate_presigned_url(
             "get_object", {"Bucket": "mybucket", "Key": "mykey"}
         )
-        self.assert_is_v2_presigned_url(url)
+        self.assert_is_v4_presigned_url(url)
 
-    def test_presign_uses_v2_for_us_east_1_with_us_east_1_regional(self):
+    def test_presign_uses_v4_for_us_east_1_with_us_east_1_regional(self):
         config = Config(s3={"us_east_1_regional_endpoint": "regional"})
         client = self.session.create_client("s3", "us-east-1", config=config)
         url = client.generate_presigned_url(
             "get_object", {"Bucket": "mybucket", "Key": "mykey"}
         )
-        self.assert_is_v2_presigned_url(url)
+        self.assert_is_v4_presigned_url(url)
 
 
 CHECKSUM_TEST_CASES = [

tests/unit/test_client.py

@@ -1985,6 +1985,54 @@ def test_validaties_retry_mode(self):
             botocore.config.Config(retries={'mode': 'turbo-mode'})
 
 
+class TestS3PresignSignatureVersion(unittest.TestCase):
+    """Test that S3 presigned URLs default to sigv4 instead of sigv2"""
+
+    def setUp(self):
+        self.client_creator = client.ClientCreator(
+            loader=mock.Mock(),
+            endpoint_resolver=mock.Mock(),
+            user_agent='test-agent',
+            event_emitter=mock.Mock(),
+            retry_handler_factory=mock.Mock(),
+            retry_config_translator=mock.Mock(),
+            response_parser_factory=mock.Mock(),
+            exceptions_factory=mock.Mock(),
+            config_store=mock.Mock(),
+            user_agent_creator=mock.Mock(),
+        )
+
+    def test_default_s3_presign_returns_s3v4_query(self):
+        result = self.client_creator._default_s3_presign_to_sigv4(
+            signature_version='v4-query'
+        )
+        self.assertEqual(result, 's3v4-query')
+
+    def test_default_s3_presign_returns_s3v4_presign_post(self):
+        result = self.client_creator._default_s3_presign_to_sigv4(
+            signature_version='v4-presign-post'
+        )
+        self.assertEqual(result, 's3v4-presign-post')
+
+    def test_default_s3_presign_ignores_v4a(self):
+        result = self.client_creator._default_s3_presign_to_sigv4(
+            signature_version='v4a-query'
+        )
+        self.assertIsNone(result)
+
+    def test_default_s3_presign_preserves_s3express(self):
+        result = self.client_creator._default_s3_presign_to_sigv4(
+            signature_version='v4-s3express-query'
+        )
+        self.assertEqual(result, 'v4-s3express-query')
+
+    def test_default_s3_presign_ignores_non_presign_signatures(self):
+        result = self.client_creator._default_s3_presign_to_sigv4(
+            signature_version='v4'
+        )
+        self.assertIsNone(result)
+
+
 class TestClientEndpointBridge(unittest.TestCase):
     def setUp(self):
         self.resolver = mock.Mock()