Fix handling of SSE-C keys when copying unencrypted to encrypted objects or objects with different keys

.changes/next-release/bugfix-s3-83199.json

@@ -0,0 +1,5 @@
+{
+  "type": "bugfix",
+  "category": "``s3``",
+  "description": "Fix handling of SSE-C keys when copying unencrypted to encrypted objects or objects with different encryption keys"
+}

awscli/customizations/s3/subcommands.py

@@ -1608,16 +1608,15 @@ def _map_sse_c_params(self, request_parameters, paths_type):
         # SSE-C key and algorithm. Note the reverse FileGenerator does
         # not need any of these because it is used only for sync operations
         # which only use ListObjects which does not require HeadObject.
-        RequestParamsMapper.map_head_object_params(
-            request_parameters['HeadObject'], self.parameters
-        )
         if paths_type == 's3s3':
+            RequestParamsMapper.map_head_object_params_with_copy_source_sse(
+                request_parameters['HeadObject'],
+                self.parameters,
+            )
+        else:
             RequestParamsMapper.map_head_object_params(
                 request_parameters['HeadObject'],
-                {
-                    'sse_c': self.parameters.get('sse_c_copy_source'),
-                    'sse_c_key': self.parameters.get('sse_c_copy_source_key'),
-                },
+                self.parameters,
             )
 
     def _get_s3_handler_params(self):

awscli/customizations/s3/subscribers.py

@@ -330,8 +330,9 @@ def _get_head_object_response(self, future):
             'Bucket': copy_source['Bucket'],
             'Key': copy_source['Key'],
         }
-        utils.RequestParamsMapper.map_head_object_params(
-            head_object_params, self._cli_params
+        utils.RequestParamsMapper.map_head_object_params_with_copy_source_sse(
+            head_object_params,
+            self._cli_params,
         )
         return self._client.head_object(**head_object_params)
 

awscli/customizations/s3/utils.py

@@ -529,6 +529,19 @@ def map_head_object_params(cls, request_params, cli_params):
         cls._set_sse_c_request_params(request_params, cli_params)
         cls._set_request_payer_param(request_params, cli_params)
 
+    @classmethod
+    def map_head_object_params_with_copy_source_sse(
+        cls, request_params, cli_params
+    ):
+        """
+        Map CLI params to HeadObject request params,
+        using the SSE-C headers from the copy source
+        """
+        cls._set_sse_c_request_params_with_copy_source_sse(
+            request_params, cli_params
+        )
+        cls._set_request_payer_param(request_params, cli_params)
+
     @classmethod
     def map_create_multipart_upload_params(cls, request_params, cli_params):
         """Map CLI params to CreateMultipartUpload request params"""
@@ -662,6 +675,18 @@ def _set_sse_c_request_params(cls, request_params, cli_params):
             request_params['SSECustomerAlgorithm'] = cli_params['sse_c']
             request_params['SSECustomerKey'] = cli_params['sse_c_key']
 
+    @classmethod
+    def _set_sse_c_request_params_with_copy_source_sse(
+        cls, request_params, cli_params
+    ):
+        if cli_params.get('sse_c_copy_source'):
+            request_params['SSECustomerAlgorithm'] = cli_params[
+                'sse_c_copy_source'
+            ]
+            request_params['SSECustomerKey'] = cli_params[
+                'sse_c_copy_source_key'
+            ]
+
     @classmethod
     def _set_sse_c_copy_source_request_params(cls, request_params, cli_params):
         if cli_params.get('sse_c_copy_source'):

tests/functional/s3/test_cp_command.py

@@ -44,6 +44,10 @@ class BaseCPCommandTest(BaseS3TransferCommandTest):
 
 
 class TestCPCommand(BaseCPCommandTest):
+    def setUp(self):
+        super().setUp()
+        self.multipart_threshold = 8 * MB
+
     def test_operations_used_in_upload(self):
         full_path = self.files.create_file('foo.txt', 'mycontent')
         cmdline = '%s %s s3://bucket/key.txt' % (self.prefix, full_path)
@@ -411,7 +415,7 @@ def test_no_overwrite_flag_on_copy_when_small_object_exists_on_target(
     ):
         cmdline = f'{self.prefix} s3://bucket1/key.txt s3://bucket/key.txt --no-overwrite'
         self.parsed_responses = [
-            self.head_object_response(ContentLength=5), 
+            self.head_object_response(ContentLength=5),
             self.precondition_failed_error_response(),
         ]
         self.run_cmd(cmdline, expected_rc=0)
@@ -978,8 +982,15 @@ def test_cp_with_sse_c_copy_source_fileb(self):
         self.run_cmd(cmdline, expected_rc=0)
         self.assertEqual(len(self.operations_called), 2)
         self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
-        self.assertEqual(self.operations_called[1][0].name, 'CopyObject')
+        expected_head_args = {
+            'Bucket': 'bucket-one',
+            'Key': 'key.txt',
+            'SSECustomerAlgorithm': 'AES256',
+            'SSECustomerKey': key_contents,
+        }
+        self.assertDictEqual(self.operations_called[0][1], expected_head_args)
 
+        self.assertEqual(self.operations_called[1][0].name, 'CopyObject')
         expected_args = {
             'Key': 'key.txt',
             'Bucket': 'bucket',
@@ -989,6 +1000,184 @@ def test_cp_with_sse_c_copy_source_fileb(self):
         }
         self.assertDictEqual(self.operations_called[1][1], expected_args)
 
+    def test_s3s3_cp_with_destination_sse_c(self):
+        """S3->S3 copy with an unencrypted source and encrypted destination"""
+        self.parsed_responses = [
+            self.head_object_response(),
+            self.copy_object_response(),
+        ]
+        cmdline = (
+            '%s s3://bucket-one/key.txt s3://bucket/key.txt '
+            '--sse-c --sse-c-key destination-key' % self.prefix
+        )
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(len(self.operations_called), 2)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        expected_head_args = {
+            'Bucket': 'bucket-one',
+            'Key': 'key.txt',
+            # don't expect to see SSE-c params for the source
+        }
+        self.assertDictEqual(self.operations_called[0][1], expected_head_args)
+
+        self.assertEqual(self.operations_called[1][0].name, 'CopyObject')
+        expected_copy_args = {
+            'Key': 'key.txt',
+            'Bucket': 'bucket',
+            'CopySource': {'Bucket': 'bucket-one', 'Key': 'key.txt'},
+            'SSECustomerAlgorithm': 'AES256',
+            'SSECustomerKey': 'destination-key',
+        }
+        self.assertDictEqual(self.operations_called[1][1], expected_copy_args)
+
+    def test_s3s3_cp_with_different_sse_c_keys(self):
+        """S3->S3 copy with different SSE-C keys for source and destination"""
+        self.parsed_responses = [
+            self.head_object_response(),
+            self.copy_object_response(),
+        ]
+        cmdline = (
+            '%s s3://bucket-one/key.txt s3://bucket/key.txt '
+            '--sse-c-copy-source --sse-c-copy-source-key foo --sse-c --sse-c-key bar'
+            % self.prefix
+        )
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(len(self.operations_called), 2)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        expected_head_args = {
+            'Bucket': 'bucket-one',
+            'Key': 'key.txt',
+            'SSECustomerAlgorithm': 'AES256',
+            'SSECustomerKey': 'foo',
+        }
+        self.assertDictEqual(self.operations_called[0][1], expected_head_args)
+
+        self.assertEqual(self.operations_called[1][0].name, 'CopyObject')
+        expected_copy_args = {
+            'Key': 'key.txt',
+            'Bucket': 'bucket',
+            'CopySource': {'Bucket': 'bucket-one', 'Key': 'key.txt'},
+            'SSECustomerAlgorithm': 'AES256',
+            'SSECustomerKey': 'bar',
+            'CopySourceSSECustomerAlgorithm': 'AES256',
+            'CopySourceSSECustomerKey': 'foo',
+        }
+        self.assertDictEqual(self.operations_called[1][1], expected_copy_args)
+
+    def test_s3s3_cp_with_destination_sse_c_multipart(self):
+        """S3->S3 multipart copy with unencrypted source and encrypted destination"""
+        self.parsed_responses = [
+            self.head_object_response(ContentLength=self.multipart_threshold),
+            self.get_object_tagging_response({}),
+            self.create_mpu_response('upload_id'),
+            self.upload_part_copy_response(),
+            self.complete_mpu_response(),
+        ]
+        cmdline = (
+            '%s s3://bucket-one/key.txt s3://bucket/key.txt '
+            '--sse-c --sse-c-key destination-key' % self.prefix
+        )
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assert_operations_called(
+            [
+                self.head_object_request(
+                    'bucket-one',
+                    'key.txt',
+                    # no SSE-C params — source is unencrypted
+                ),
+                ('GetObjectTagging', mock.ANY),
+                self.create_mpu_request(
+                    'bucket',
+                    'key.txt',
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='destination-key',
+                ),
+                self.upload_part_copy_request(
+                    'bucket-one',
+                    'key.txt',
+                    'bucket',
+                    'key.txt',
+                    'upload_id',
+                    PartNumber=mock.ANY,
+                    CopySourceRange=mock.ANY,
+                    CopySourceIfMatch='"foo-1"',
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='destination-key',
+                ),
+                self.complete_mpu_request(
+                    'bucket',
+                    'key.txt',
+                    'upload_id',
+                    num_parts=1,
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='destination-key',
+                ),
+            ]
+        )
+
+    def test_s3s3_cp_with_different_sse_c_keys_multipart(self):
+        """S3->S3 multipart copy with different SSE-C keys"""
+        self.parsed_responses = [
+            self.head_object_response(ContentLength=self.multipart_threshold),
+            self.get_object_tagging_response({}),
+            self.create_mpu_response('upload_id'),
+            self.upload_part_copy_response(),
+            self.complete_mpu_response(),
+        ]
+        cmdline = (
+            '%s s3://bucket-one/key.txt s3://bucket/key.txt '
+            '--sse-c-copy-source --sse-c-copy-source-key source-key --sse-c --sse-c-key destination-key'
+            % self.prefix
+        )
+        self.run_cmd(cmdline, expected_rc=0)
+        self.assertEqual(self.operations_called[0][0].name, 'HeadObject')
+        expected_head_args = {
+            'Bucket': 'bucket-one',
+            'Key': 'key.txt',
+            'SSECustomerAlgorithm': 'AES256',
+            'SSECustomerKey': 'source-key',
+        }
+        self.assertDictEqual(self.operations_called[0][1], expected_head_args)
+        self.assert_operations_called(
+            [
+                self.head_object_request(
+                    'bucket-one',
+                    'key.txt',
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='source-key',
+                ),
+                ('GetObjectTagging', mock.ANY),
+                self.create_mpu_request(
+                    'bucket',
+                    'key.txt',
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='destination-key',
+                ),
+                self.upload_part_copy_request(
+                    'bucket-one',
+                    'key.txt',
+                    'bucket',
+                    'key.txt',
+                    'upload_id',
+                    PartNumber=mock.ANY,
+                    CopySourceRange=mock.ANY,
+                    CopySourceIfMatch='"foo-1"',
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='destination-key',
+                    CopySourceSSECustomerAlgorithm='AES256',
+                    CopySourceSSECustomerKey='source-key',
+                ),
+                self.complete_mpu_request(
+                    'bucket',
+                    'key.txt',
+                    'upload_id',
+                    num_parts=1,
+                    SSECustomerAlgorithm='AES256',
+                    SSECustomerKey='destination-key',
+                ),
+            ]
+        )
+
     # Note ideally the kms sse with a key id would be integration tests
     # However, you cannot delete kms keys so there would be no way to clean
     # up the tests