fix copy in place for bucket with encryption (#6303)

moto/s3/models.py

@@ -242,7 +242,7 @@ class FakeKey(BaseModel, ManagedState):
             res["x-amz-server-side-encryption"] = self.encryption
             if self.encryption == "aws:kms" and self.kms_key_id is not None:
                 res["x-amz-server-side-encryption-aws-kms-key-id"] = self.kms_key_id
-        if self.bucket_key_enabled is not None:
+        if self.encryption == "aws:kms" and self.bucket_key_enabled is not None:
             res[
                 "x-amz-server-side-encryption-bucket-key-enabled"
             ] = self.bucket_key_enabled
@@ -2373,6 +2373,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
         lock_legal_status: Optional[str] = None,
         lock_until: Optional[str] = None,
     ) -> None:
+        bucket = self.get_bucket(dest_bucket_name)
         if src_key.name == dest_key_name and src_key.bucket_name == dest_bucket_name:
             if src_key.encryption and src_key.encryption != "AES256" and not encryption:
                 # this a special case, as now S3 default to AES256 when not provided
@@ -2386,6 +2387,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
                     encryption,
                     mdirective == "REPLACE",
                     website_redirect_location,
+                    bucket.encryption,  # S3 will allow copy in place if the bucket has encryption configured
                 )
             ):
                 raise CopyObjectMustChangeSomething
@@ -2417,7 +2419,6 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
             new_key.set_expiry(None)
 
         # Send notifications that an object was copied
-        bucket = self.get_bucket(dest_bucket_name)
         notifications.send_event(
             self.account_id, notifications.S3_OBJECT_CREATE_COPY, bucket, new_key
         )

tests/test_s3/test_s3_copyobject.py

@@ -682,3 +682,39 @@ def test_copy_object_in_place_website_redirect_location():
 
     head_object = client.head_object(Bucket=bucket_name, Key=key)
     assert head_object["WebsiteRedirectLocation"] == "/test/direct"
+
+
+@mock_s3
+def test_copy_object_in_place_with_bucket_encryption():
+    # If a bucket has encryption configured, it will allow copy in place per default
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+    bucket_name = "test-bucket"
+    client.create_bucket(Bucket=bucket_name)
+    key = "source-key"
+
+    response = client.put_bucket_encryption(
+        Bucket=bucket_name,
+        ServerSideEncryptionConfiguration={
+            "Rules": [
+                {
+                    "ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"},
+                    "BucketKeyEnabled": False,
+                },
+            ]
+        },
+    )
+    assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+    response = client.put_object(
+        Body=b"",
+        Bucket=bucket_name,
+        Key=key,
+    )
+    assert response["ServerSideEncryption"] == "AES256"
+
+    response = client.copy_object(
+        Bucket=bucket_name,
+        CopySource={"Bucket": bucket_name, "Key": key},
+        Key=key,
+    )
+    assert response["ServerSideEncryption"] == "AES256"

tests/test_s3/test_s3_encryption.py

@@ -127,3 +127,24 @@ def test_encryption_status_on_copied_objects():
     # verify encryption status on object2 does have encryption
     res = s3.get_object(Bucket=bucket_name, Key="file2.txt")
     res.should.have.key("ServerSideEncryption").equals("AES256")
+
+
+@mock_s3
+def test_encryption_bucket_key_for_aes_not_returned():
+    bucket_name = str(uuid4())
+    s3 = boto3.client("s3", region_name="us-east-1")
+    s3.create_bucket(Bucket=bucket_name)
+    # enable encryption
+    sse_config = {
+        "Rules": [
+            {
+                "ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"},
+                "BucketKeyEnabled": False,
+            }
+        ]
+    }
+    s3.put_bucket_encryption(
+        Bucket=bucket_name, ServerSideEncryptionConfiguration=sse_config
+    )
+    res = s3.put_object(Bucket=bucket_name, Body=b"test", Key="file.txt")
+    res.shouldnt.have.key("BucketKeyEnabled")