diff --git a/moto/s3/models.py b/moto/s3/models.py index 930d76256..75ad72fac 100644 --- a/moto/s3/models.py +++ b/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 ) diff --git a/tests/test_s3/test_s3_copyobject.py b/tests/test_s3/test_s3_copyobject.py index e28819b9f..156918a84 100644 --- a/tests/test_s3/test_s3_copyobject.py +++ b/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" diff --git a/tests/test_s3/test_s3_encryption.py b/tests/test_s3/test_s3_encryption.py index 7a9829d70..3c566435f 100644 --- a/tests/test_s3/test_s3_encryption.py +++ b/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")