S3: add a checksum to a copied object if requested so

moto/s3/responses.py

@@ -1619,6 +1619,19 @@ def _key_response_put(
             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)
@@ -2600,10 +2613,17 @@ def _invalid_headers(self, url: str, headers: Dict[str, str]) -> bool:
   </CORSConfiguration>
 """
 
+# https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html
 S3_OBJECT_COPY_RESPONSE = """\
 <CopyObjectResult xmlns="http://doc.s3.amazonaws.com/2006-03-01">
     <ETag>{{ key.etag }}</ETag>
     <LastModified>{{ key.last_modified_ISO8601 }}</LastModified>
+    {% if key.checksum_value  %}
+      {% if "CRC32" in key.checksum_algorithm %}<ChecksumCRC32>{{ key.checksum_value }}</ChecksumCRC32>{% endif %}
+      {% if "CRC32C" in key.checksum_algorithm %}<ChecksumCRC32C>{{ key.checksum_value }}</ChecksumCRC32C>{% endif %}
+      {% if "SHA1" in key.checksum_algorithm %}<ChecksumSHA1>{{ key.checksum_value }}</ChecksumSHA1>{% endif %}
+      {% if "SHA256" in key.checksum_algorithm %}<ChecksumSHA256>{{ key.checksum_value }}</ChecksumSHA256>{% endif %}
+    {% endif %}
 </CopyObjectResult>"""
 
 S3_MULTIPART_INITIATE_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>

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)