[s3] Allow KMS SSE parameters in copy_object (#4593)

moto/s3/models.py

@@ -2048,6 +2048,8 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         storage=None,
         acl=None,
         src_version_id=None,
+        encryption=None,
+        kms_key_id=None,
     ):
         key = self.get_object(src_bucket_name, src_key_name, version_id=src_version_id)
 
@@ -2057,8 +2059,8 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             value=key.value,
             storage=storage or key.storage_class,
             multipart=key.multipart,
-            encryption=key.encryption,
-            kms_key_id=key.kms_key_id,
+            encryption=encryption or key.encryption,
+            kms_key_id=kms_key_id or key.kms_key_id,
             bucket_key_enabled=key.bucket_key_enabled,
             lock_mode=key.lock_mode,
             lock_legal_status=key.lock_legal_status,

moto/s3/responses.py

@@ -1559,6 +1559,8 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
                     storage=storage_class,
                     acl=acl,
                     src_version_id=src_version_id,
+                    kms_key_id=kms_key_id,
+                    encryption=encryption,
                 )
             else:
                 raise MissingKey(key=src_key)

tests/test_s3/test_s3.py

@@ -28,7 +28,7 @@ import pytest
 
 import sure  # noqa # pylint: disable=unused-import
 
-from moto import settings, mock_s3, mock_s3_deprecated, mock_config
+from moto import settings, mock_s3, mock_s3_deprecated, mock_config, mock_kms
 import moto.s3.models as s3model
 from moto.core.exceptions import InvalidNextTokenException
 from moto.settings import get_s3_default_key_buffer_size, S3_UPLOAD_PART_MIN_SIZE
@@ -3378,7 +3378,7 @@ def test_boto3_head_object_with_versioning():
 
 @mock_s3
 def test_boto3_copy_non_existing_file():
-    s3 = boto3.resource("s3")
+    s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
     src = "srcbucket"
     target = "target"
     s3.create_bucket(Bucket=src)
@@ -6577,3 +6577,28 @@ def test_request_partial_content_should_contain_all_metadata():
     assert response["LastModified"] == obj.last_modified
     assert response["ContentLength"] == 4
     assert response["ContentRange"] == "bytes {}/{}".format(query_range, len(body))
+
+
+@mock_s3
+@mock_kms
+def test_boto3_copy_object_with_kms_encryption():
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    kms_client = boto3.client("kms", region_name=DEFAULT_REGION_NAME)
+    kms_key = kms_client.create_key()["KeyMetadata"]["KeyId"]
+
+    client.create_bucket(
+        Bucket="blah", CreateBucketConfiguration={"LocationConstraint": "eu-west-1"}
+    )
+
+    client.put_object(Bucket="blah", Key="test1", Body=b"test1")
+
+    client.copy_object(
+        CopySource={"Bucket": "blah", "Key": "test1"},
+        Bucket="blah",
+        Key="test2",
+        SSEKMSKeyId=kms_key,
+        ServerSideEncryption="aws:kms",
+    )
+    result = client.head_object(Bucket="blah", Key="test2")
+    assert result["SSEKMSKeyId"] == kms_key
+    assert result["ServerSideEncryption"] == "aws:kms"