Bugfix/8762 Multiple changes to align S3 multipart checksum support with AWS functionality

moto/s3/models.py

@@ -127,6 +127,8 @@ def __init__(
         lock_legal_status: Optional[str] = None,
         lock_until: Optional[str] = None,
         checksum_value: Optional[str] = None,
+        checksum_type: Optional[str] = None,
+        checksum_parts: Optional[int] = None,
     ):
         ManagedState.__init__(
             self,
@@ -169,6 +171,8 @@ def __init__(
         self.lock_legal_status = lock_legal_status
         self.lock_until = lock_until
         self.checksum_value = checksum_value
+        self.checksum_type = checksum_type
+        self.checksum_parts = checksum_parts
 
         # Default metadata values
         self._metadata["Content-Type"] = "binary/octet-stream"
@@ -445,12 +449,14 @@ def __init__(
 
     def complete(
         self, body: Iterator[Tuple[int, str]]
-    ) -> Tuple[bytes, str, Optional[str]]:
-        checksum_algo = self.metadata.get("x-amz-checksum-algorithm")
+    ) -> Tuple[bytearray, str, Optional[Tuple[str, str, int]]]:
+        checksum_algo: Optional[str] = self.metadata.get("x-amz-checksum-algorithm")
+        checksum_type: Optional[str] = self.metadata.get("x-amz-checksum-type")
         decode_hex = codecs.getdecoder("hex_codec")
         total = bytearray()
         md5s = bytearray()
-        checksum = bytearray()
+
+        checksum_builder = MultipartChecksumBuilder(checksum_algo, checksum_type)
 
         last = None
         count = 0
@@ -466,10 +472,7 @@ def complete(
                 raise EntityTooSmall()
             md5s.extend(decode_hex(part_etag)[0])  # type: ignore
             total.extend(part.value)
-            if checksum_algo:
-                checksum.extend(
-                    compute_checksum(part.value, checksum_algo, encode_base64=False)
-                )
+            checksum_builder.addPart(part.value)
             last = part
             count += 1
 
@@ -478,11 +481,7 @@ def complete(
 
         full_etag = md5_hash()
         full_etag.update(bytes(md5s))
-        if checksum_algo:
-            encoded_checksum = compute_checksum(checksum, checksum_algo).decode("utf-8")
-        else:
-            encoded_checksum = None
-        return total, f"{full_etag.hexdigest()}-{count}", encoded_checksum
+        return total, f"{full_etag.hexdigest()}-{count}", checksum_builder.build()
 
     def set_part(self, part_id: int, value: bytes) -> FakeKey:
         if part_id < 1:
@@ -514,6 +513,52 @@ def dispose(self) -> None:
             part.dispose()
 
 
+class MultipartChecksumBuilder:
+    def __init__(self, checksum_algorithm: Optional[str], checksum_type: Optional[str]):
+        self.complete_object = bytearray()
+        self.checksum = bytearray()
+
+        self.part_count = 0
+
+        # Checksum algorithm defaults to CRC64NVME when none specified
+        self.checksum_algorithm: str = checksum_algorithm or "CRC64NVME"
+
+        # Checksum type defaults to COMPOSITE except for CRC64NVME which only supports FULL_OBJECT
+        self.checksum_type: str = checksum_type or ""
+        if not self.checksum_type:
+            self.checksum_type = (
+                "FULL_OBJECT" if self.checksum_algorithm == "CRC64NVME" else "COMPOSITE"
+            )
+
+    def addPart(self, part: bytes) -> None:
+        if self.checksum_type == "COMPOSITE":
+            self.checksum.extend(
+                compute_checksum(part, self.checksum_algorithm, encode_base64=False)
+            )
+        else:
+            self.complete_object.extend(part)
+
+        self.part_count += 1
+
+    def build(self) -> Optional[Tuple[str, str, int]]:
+        if self.checksum_type == "COMPOSITE":
+            return (
+                self.checksum_type,
+                compute_checksum(self.checksum, self.checksum_algorithm).decode(
+                    "utf-8"
+                ),
+                self.part_count,
+            )
+
+        return (
+            self.checksum_type,
+            compute_checksum(self.complete_object, self.checksum_algorithm).decode(
+                "utf-8"
+            ),
+            1,
+        )
+
+
 class FakeGrantee(BaseModel):
     def __init__(self, grantee_id: str = "", uri: str = "", display_name: str = ""):
         self.id = grantee_id
@@ -2332,6 +2377,8 @@ def get_object_attributes(
             response_keys["etag"] = key.etag.replace('"', "")
         if "Checksum" in attributes_to_get and key.checksum_value is not None:
             response_keys["checksum"] = {key.checksum_algorithm: key.checksum_value}
+            if key.checksum_type:
+                response_keys["checksum"]["type"] = key.checksum_type
         if "ObjectSize" in attributes_to_get:
             response_keys["size"] = key.size
         if "StorageClass" in attributes_to_get:
@@ -2613,7 +2660,7 @@ def complete_multipart_upload(
     ) -> Optional[FakeKey]:
         bucket = self.get_bucket(bucket_name)
         multipart = bucket.multiparts[multipart_id]
-        value, etag, checksum = multipart.complete(body)
+        value, etag, checksum_details = multipart.complete(body)
         if value is not None:
             del bucket.multiparts[multipart_id]
 
@@ -2632,9 +2679,11 @@ def complete_multipart_upload(
         )
         key.set_metadata(multipart.metadata)
 
-        if checksum:
+        if checksum_details:
             key.checksum_algorithm = multipart.metadata.get("x-amz-checksum-algorithm")
-            key.checksum_value = checksum
+            key.checksum_type = checksum_details[0]
+            key.checksum_value = checksum_details[1]
+            key.checksum_parts = checksum_details[2]
 
         self.put_object_tagging(key, multipart.tags)
         self.put_object_acl(

moto/s3/responses.py

@@ -1552,10 +1552,17 @@ def get_object(self) -> TYPE_RESPONSE:
             self.headers.get("x-amz-checksum-mode") == "ENABLED"
             and key.checksum_algorithm
         ):
+            qualified_checksum = key.checksum_value
+            if key.checksum_type == "COMPOSITE":
+                qualified_checksum = f"{key.checksum_value}-{key.checksum_parts}"
+
             response_headers[f"x-amz-checksum-{key.checksum_algorithm.lower()}"] = (
-                key.checksum_value
+                qualified_checksum
             )
 
+            if key.checksum_type:
+                response_headers["x-amz-checksum-type"] = key.checksum_type
+
         response_headers.update(key.metadata)
         response_headers.update({"Accept-Ranges": "bytes"})
 
@@ -3578,10 +3585,12 @@ def _invalid_headers(self, url: str, headers: Dict[str, str]) -> bool:
     {% if etag is not none %}<ETag>{{ etag }}</ETag>{% endif %}
     {% if checksum is not none %}
       <Checksum>
+      {% if "CRC64NVME" in checksum %}<ChecksumCRC64NVME>{{ checksum["CRC64NVME"] }}</ChecksumCRC64NVME>{% endif %}
       {% if "CRC32" in checksum %}<ChecksumCRC32>{{ checksum["CRC32"] }}</ChecksumCRC32>{% endif %}
       {% if "CRC32C" in checksum %}<ChecksumCRC32C>{{ checksum["CRC32C"] }}</ChecksumCRC32C>{% endif %}
       {% if "SHA1" in checksum %}<ChecksumSHA1>{{ checksum["SHA1"] }}</ChecksumSHA1>{% endif %}
       {% if "SHA256" in checksum %}<ChecksumSHA256>{{ checksum["SHA256"] }}</ChecksumSHA256>{% endif %}
+      {% if "type" in checksum %}<ChecksumType>{{ checksum["type"] }}</ChecksumType>{% endif %}
       </Checksum>
     {% endif %}
     {% if size is not none %}<ObjectSize>{{ size }}</ObjectSize>{% endif %}

moto/s3/utils.py

@@ -7,6 +7,7 @@
 from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 from urllib.parse import urlparse
 
+from awscrt import checksums as crt_checksums
 from requests.structures import CaseInsensitiveDict
 
 from moto.settings import S3_IGNORE_SUBDOMAIN_BUCKETNAME
@@ -25,6 +26,7 @@
     "content-disposition",
     "x-robots-tag",
     "x-amz-checksum-algorithm",
+    "x-amz-checksum-type",
     "x-amz-content-sha256",
     "x-amz-content-crc32",
     "x-amz-content-crc32c",
@@ -209,6 +211,8 @@ def compute_checksum(body: bytes, algorithm: str, encode_base64: bool = True) ->
             hashed_body = binascii.crc32(body).to_bytes(4, "big")
     elif algorithm == "CRC32":
         hashed_body = binascii.crc32(body).to_bytes(4, "big")
+    elif algorithm == "CRC64NVME":
+        hashed_body = crt_checksums.crc64nvme(body).to_bytes(8, "big")
     else:
         hashed_body = _hash(hashlib.sha256, (body,))
     if encode_base64: