Autopopulate CopySourceSSECustomerKeyMD5 for s3

botocore/handlers.py

@@ -170,21 +170,37 @@ def sse_md5(params, **kwargs):
     encryption key. This handler does both if the MD5 has not been set by
     the caller.
     """
-    if not _needs_s3_sse_customization(params):
+    _sse_md5(params, 'SSECustomer')
+
+
+def copy_source_sse_md5(params, **kwargs):
+    """
+    S3 server-side encryption requires the encryption key to be sent to the
+    server base64 encoded, as well as a base64-encoded MD5 hash of the
+    encryption key. This handler does both if the MD5 has not been set by
+    the caller specifically if the parameter is for the copy-source sse-c key.
+    """
+    _sse_md5(params, 'CopySourceSSECustomer')
+
+
+def _sse_md5(params, sse_member_prefix='SSECustomer'):
+    if not _needs_s3_sse_customization(params, sse_member_prefix):
         return
-    key_as_bytes = params['SSECustomerKey']
+    sse_key_member = sse_member_prefix + 'Key'
+    sse_md5_member = sse_member_prefix + 'KeyMD5'
+    key_as_bytes = params[sse_key_member]
     if isinstance(key_as_bytes, six.text_type):
         key_as_bytes = key_as_bytes.encode('utf-8')
     key_md5_str = base64.b64encode(
         hashlib.md5(key_as_bytes).digest()).decode('utf-8')
     key_b64_encoded = base64.b64encode(key_as_bytes).decode('utf-8')
-    params['SSECustomerKey'] = key_b64_encoded
-    params['SSECustomerKeyMD5'] = key_md5_str
+    params[sse_key_member] = key_b64_encoded
+    params[sse_md5_member] = key_md5_str
 
 
-def _needs_s3_sse_customization(params):
-    return (params.get('SSECustomerKey') is not None and
-            'SSECustomerKeyMD5' not in params)
+def _needs_s3_sse_customization(params, sse_member_prefix):
+    return (params.get(sse_member_prefix + 'Key') is not None and
+            sse_member_prefix + 'KeyMD5' not in params)
 
 
 def register_retries_for_service(service_data, session,
@@ -545,9 +561,11 @@ def change_get_to_post(request, **kwargs):
     ('before-parameter-build.s3.GetObject', sse_md5),
     ('before-parameter-build.s3.PutObject', sse_md5),
     ('before-parameter-build.s3.CopyObject', sse_md5),
+    ('before-parameter-build.s3.CopyObject', copy_source_sse_md5),
     ('before-parameter-build.s3.CreateMultipartUpload', sse_md5),
     ('before-parameter-build.s3.UploadPart', sse_md5),
     ('before-parameter-build.s3.UploadPartCopy', sse_md5),
+    ('before-parameter-build.s3.UploadPartCopy', copy_source_sse_md5),
     ('before-parameter-build.ec2.RunInstances', base64_encode_user_data),
     ('before-parameter-build.autoscaling.CreateLaunchConfiguration',
      base64_encode_user_data),
@@ -576,6 +594,10 @@ def change_get_to_post(request, **kwargs):
     # S3 SSE documentation modifications
     ('docs.*.s3.*.complete-section',
      AutoPopulatedParam('SSECustomerKeyMD5').document_auto_populated_param),
+    # S3 SSE Copy Source documentation modifications
+    ('docs.*.s3.*.complete-section',
+     AutoPopulatedParam(
+        'CopySourceSSECustomerKeyMD5').document_auto_populated_param),
     # The following S3 operations cannot actually accept a ContentMD5
     ('docs.*.s3.*.complete-section',
      HideParamFromOperations(

tests/functional/docs/test_s3.py

@@ -21,6 +21,12 @@ def test_auto_populates_sse_customer_key_md5(self):
             method_name='put_object',
             param_name='SSECustomerKeyMD5')
 
+    def test_auto_populates_copy_source_sse_customer_key_md5(self):
+        self.assert_is_documented_as_autopopulated_param(
+            service_name='s3',
+            method_name='copy_object',
+            param_name='CopySourceSSECustomerKeyMD5')
+
     def test_hides_content_md5_when_impossible_to_provide(self):
         modified_methods = ['delete_objects', 'put_bucket_acl',
                             'put_bucket_cors', 'put_bucket_lifecycle',

tests/integration/test_s3.py

@@ -822,6 +822,40 @@ def test_make_request_with_sse(self):
                                    SSECustomerKey=key_str)['Body'].read(),
             b'mycontents2')
 
+    def test_make_request_with_sse_copy_source(self):
+        encrypt_key = 'a' * 32
+        other_encrypt_key = 'b' * 32
+
+        # Upload the object using one encrypt key
+        self.client.put_object(
+            Bucket=self.bucket_name, Key='foo.txt',
+            Body=six.BytesIO(b'mycontents'), SSECustomerAlgorithm='AES256',
+            SSECustomerKey=encrypt_key)
+        self.addCleanup(self.client.delete_object,
+                        Bucket=self.bucket_name, Key='foo.txt')
+
+        # Copy the object using the original encryption key as the copy source
+        # and encrypt with a new encryption key.
+        self.client.copy_object(
+            Bucket=self.bucket_name,
+            CopySource=self.bucket_name+'/foo.txt',
+            Key='bar.txt', CopySourceSSECustomerAlgorithm='AES256',
+            CopySourceSSECustomerKey=encrypt_key,
+            SSECustomerAlgorithm='AES256',
+            SSECustomerKey=other_encrypt_key
+        )
+        self.addCleanup(self.client.delete_object,
+                        Bucket=self.bucket_name, Key='bar.txt')
+
+        # Download the object using the new encryption key.
+        # The content should not have changed.
+        self.assertEqual(
+            self.client.get_object(
+                Bucket=self.bucket_name, Key='bar.txt',
+                SSECustomerAlgorithm='AES256',
+                SSECustomerKey=other_encrypt_key)['Body'].read(),
+            b'mycontents')
+
 
 class TestS3UTF8Headers(BaseS3ClientTest):
     def test_can_set_utf_8_headers(self):

tests/unit/test_handlers.py

@@ -189,6 +189,25 @@ def test_sse_params_as_str(self):
         self.assertEqual(params['SSECustomerKeyMD5'],
                          'N7UdGUp1E+RbVvZSTy1R8g==')
 
+    def test_copy_source_sse_params(self):
+        for op in ['CopyObject', 'UploadPartCopy']:
+            event = 'before-parameter-build.s3.%s' % op
+            params = {'CopySourceSSECustomerKey': b'bar',
+                      'CopySourceSSECustomerAlgorithm': 'AES256'}
+            self.session.emit(event, params=params, model=mock.Mock())
+            self.assertEqual(params['CopySourceSSECustomerKey'], 'YmFy')
+            self.assertEqual(params['CopySourceSSECustomerKeyMD5'],
+                             'N7UdGUp1E+RbVvZSTy1R8g==')
+
+    def test_copy_source_sse_params_as_str(self):
+        event = 'before-parameter-build.s3.CopyObject'
+        params = {'CopySourceSSECustomerKey': 'bar',
+                  'CopySourceSSECustomerAlgorithm': 'AES256'}
+        self.session.emit(event, params=params, model=mock.Mock())
+        self.assertEqual(params['CopySourceSSECustomerKey'], 'YmFy')
+        self.assertEqual(params['CopySourceSSECustomerKeyMD5'],
+                         'N7UdGUp1E+RbVvZSTy1R8g==')
+
     def test_route53_resource_id(self):
         event = 'before-parameter-build.route53.GetHostedZone'
         params = {'Id': '/hostedzone/ABC123',