diff --git a/moto/s3/responses.py b/moto/s3/responses.py index a39ada5d6..fc5ce8306 100644 --- a/moto/s3/responses.py +++ b/moto/s3/responses.py @@ -1617,6 +1617,19 @@ class S3Response(BaseResponse): if tdirective == "REPLACE": tagging = self._tagging_from_headers(request.headers) self.backend.set_key_tags(new_key, tagging) + + # checksum stuff, do we need to compute hash of the copied object + checksum_algorithm = request.headers.get("x-amz-checksum-algorithm") + if checksum_algorithm: + checksum_value = compute_checksum( + key_to_copy.value, algorithm=checksum_algorithm + ).decode("utf-8") + response_headers.update( + {"Checksum": {f"Checksum{checksum_algorithm}": checksum_value}} + ) + new_key.checksum_algorithm = checksum_algorithm + new_key.checksum_value = checksum_value + template = self.response_template(S3_OBJECT_COPY_RESPONSE) response_headers.update(new_key.response_dict) return 200, response_headers, template.render(key=new_key) @@ -2598,10 +2611,17 @@ S3_BUCKET_CORS_RESPONSE = """ """ +# https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html S3_OBJECT_COPY_RESPONSE = """\ {{ key.etag }} {{ key.last_modified_ISO8601 }} + {% if key.checksum_value %} + {% if "CRC32" in key.checksum_algorithm %}{{ key.checksum_value }}{% endif %} + {% if "CRC32C" in key.checksum_algorithm %}{{ key.checksum_value }}{% endif %} + {% if "SHA1" in key.checksum_algorithm %}{{ key.checksum_value }}{% endif %} + {% if "SHA256" in key.checksum_algorithm %}{{ key.checksum_value }}{% endif %} + {% endif %} """ S3_MULTIPART_INITIATE_RESPONSE = """ diff --git a/tests/test_s3/test_s3_copyobject.py b/tests/test_s3/test_s3_copyobject.py index c360624f8..6bb64dee4 100644 --- a/tests/test_s3/test_s3_copyobject.py +++ b/tests/test_s3/test_s3_copyobject.py @@ -36,6 +36,37 @@ def test_copy_key_boto3(key_name): resp["Body"].read().should.equal(b"some value") +@mock_s3 +def test_copy_key_boto3_with_sha256_checksum(): + # Setup + s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME) + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + key_name = "key" + new_key = "new_key" + bucket = "foobar" + expected_hash = "YWIzZDA3ZjMxNjljY2JkMGVkNmM0YjQ1ZGUyMTUxOWY5ZjkzOGM3MmQyNDEyNDk5OGFhYjk0OWNlODNiYjUxYg==" + + s3.create_bucket(Bucket=bucket) + key = s3.Object("foobar", key_name) + key.put(Body=b"some value") + + # Execute + key2 = s3.Object(bucket, new_key) + key2.copy( + CopySource={"Bucket": bucket, "Key": key_name}, + ExtraArgs={"ChecksumAlgorithm": "SHA256"}, + ) + + # Verify + resp = client.get_object_attributes( + Bucket=bucket, Key=new_key, ObjectAttributes=["Checksum"] + ) + + assert "Checksum" in resp + assert "ChecksumSHA256" in resp["Checksum"] + assert resp["Checksum"]["ChecksumSHA256"] == expected_hash + + @mock_s3 def test_copy_key_with_version_boto3(): s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)