Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

S3: add a checksum to a copied object if requested so (#6245)

Browse Source
This commit is contained in:
rafcio19 2023-04-22 21:45:00 +01:00 committed by GitHub
parent f54f4a666f
commit 7fad08214d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 51 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
moto/s3/responses.py
View File

@ -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 = """<?xml version="1.0" encoding="UTF-8"?>
</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"?>

31
tests/test_s3/test_s3_copyobject.py
View File

@ -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)