From 4d9f87cd0a28263448b8c6e92aad84cfe07baad1 Mon Sep 17 00:00:00 2001
From: jonathan343 <jonathangee09@gmail.com>
Date: Thu, 3 Oct 2024 10:42:28 -0400
Subject: [PATCH] Add support for CRC64NVME when the CRT is available.

---
 botocore/httpchecksum.py        | 18 ++++++++++++++++--
 tests/unit/test_httpchecksum.py | 28 ++++++++++++++++++++++++++++
 2 files changed, 44 insertions(+), 2 deletions(-)

diff --git a/botocore/httpchecksum.py b/botocore/httpchecksum.py
index a97eb430d4..1991615e9c 100644
--- a/botocore/httpchecksum.py
+++ b/botocore/httpchecksum.py
@@ -109,6 +109,19 @@ def digest(self):
         return self._int_crc32c.to_bytes(4, byteorder="big")
 
 
+class CrtCrc64NvmeChecksum(BaseChecksum):
+    # Note: This class is only used if the CRT is available
+    def __init__(self):
+        self._int_crc64nvme = 0
+
+    def update(self, chunk):
+        new_checksum = crt_checksums.crc64nvme(chunk, self._int_crc64nvme)
+        self._int_crc64nvme = new_checksum & 0xFFFFFFFFFFFFFFFF
+
+    def digest(self):
+        return self._int_crc64nvme.to_bytes(8, byteorder="big")
+
+
 class Sha1Checksum(BaseChecksum):
     def __init__(self):
         self._checksum = sha1()
@@ -465,12 +478,13 @@ def _handle_bytes_response(http_response, response, algorithm):
     "sha1": Sha1Checksum,
     "sha256": Sha256Checksum,
 }
-_CRT_CHECKSUM_ALGORITHMS = ["crc32", "crc32c"]
+_CRT_CHECKSUM_ALGORITHMS = ["crc32", "crc32c", "crc64nvme"]
 if HAS_CRT:
     # Use CRT checksum implementations if available
     _CRT_CHECKSUM_CLS = {
         "crc32": CrtCrc32Checksum,
         "crc32c": CrtCrc32cChecksum,
+        "crc64nvme": CrtCrc64NvmeChecksum,
     }
     _CHECKSUM_CLS.update(_CRT_CHECKSUM_CLS)
     # Validate this list isn't out of sync with _CRT_CHECKSUM_CLS keys
@@ -478,4 +492,4 @@ def _handle_bytes_response(http_response, response, algorithm):
         name in _CRT_CHECKSUM_ALGORITHMS for name in _CRT_CHECKSUM_CLS.keys()
     )
 _SUPPORTED_CHECKSUM_ALGORITHMS = list(_CHECKSUM_CLS.keys())
-_ALGORITHMS_PRIORITY_LIST = ['crc32c', 'crc32', 'sha1', 'sha256']
+_ALGORITHMS_PRIORITY_LIST = ['crc32c', 'crc32', 'crc64nvme', 'sha1', 'sha256']
diff --git a/tests/unit/test_httpchecksum.py b/tests/unit/test_httpchecksum.py
index bc8265a9b3..1e46fc3268 100644
--- a/tests/unit/test_httpchecksum.py
+++ b/tests/unit/test_httpchecksum.py
@@ -26,6 +26,7 @@
     Crc32Checksum,
     CrtCrc32cChecksum,
     CrtCrc32Checksum,
+    CrtCrc64NvmeChecksum,
     Sha1Checksum,
     Sha256Checksum,
     StreamingChecksumBody,
@@ -193,6 +194,24 @@ def test_request_checksum_algorithm_model_no_crt_crc32c_unsupported(self):
                 str(context.exception),
             )
 
+    @unittest.skipIf(HAS_CRT, "Error only expected when CRT is not available")
+    def test_request_checksum_algorithm_model_no_crt_crc64nvme_unsupported(
+        self,
+    ):
+        request = self._build_request(b"")
+        operation_model = self._make_operation_model(
+            http_checksum={"requestAlgorithmMember": "Algorithm"},
+        )
+        params = {"Algorithm": "crc64nvme"}
+        with self.assertRaises(MissingDependencyException) as context:
+            resolve_request_checksum_algorithm(
+                request, operation_model, params
+            )
+            self.assertIn(
+                "Using CRC64NVME requires an additional dependency",
+                str(context.exception),
+            )
+
     def test_request_checksum_algorithm_model_legacy_md5(self):
         request = self._build_request(b"")
         operation_model = self._make_operation_model(required=True)
@@ -691,6 +710,10 @@ def test_crt_crc32(self):
     def test_crt_crc32c(self):
         self.assert_base64_checksum(CrtCrc32cChecksum, "yZRlqg==")
 
+    @requires_crt()
+    def test_crt_crc64nvme(self):
+        self.assert_base64_checksum(CrtCrc64NvmeChecksum, "jSnVw/bqjr4=")
+
 
 class TestCrtChecksumOverrides(unittest.TestCase):
     @requires_crt()
@@ -703,6 +726,11 @@ def test_crt_crc32c_available(self):
         actual_cls = _CHECKSUM_CLS.get("crc32c")
         self.assertEqual(actual_cls, CrtCrc32cChecksum)
 
+    @requires_crt()
+    def test_crt_crc64nvme_available(self):
+        actual_cls = _CHECKSUM_CLS.get("crc64nvme")
+        self.assertEqual(actual_cls, CrtCrc64NvmeChecksum)
+
 
 class TestStreamingChecksumBody(unittest.TestCase):
     def setUp(self):