feat: Add crc32c checksums to S3 Service (#4533)

* Add crc32c checksums to S3

* Properly encode checksum based on https://docs.aws.amazon.com/AmazonS3/latest/API/API_Checksum.html

* Fix formatting

* Remove S3 prefix, don't copy body & declare checksum_algorithm in initiate MultipartUpload

* Improve documentation

* Fix formatting

* Refactor ChecksumAlgorithm

* Fix formatting

---------

Co-authored-by: Jan Wackerbauer <[email protected]>

Author: JWackerbauer

core/Cargo.lock

@@ -350,6 +350,7 @@ prometheus-client = { version = "0.22.2", optional = true }
 tracing = { version = "0.1", optional = true }
 # for layers-dtrace
 probe = { version = "0.5.1", optional = true }
+crc32c = "0.6.5"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 getrandom = { version = "0.2", features = ["js"] }

core/Cargo.toml

@@ -204,6 +204,12 @@ pub struct S3Config {
     ///
     /// For example, R2 doesn't support stat with `response_content_type` query.
     pub disable_stat_with_override: bool,
+    /// Checksum Algorithm to use when sending checksums in HTTP headers.
+    /// This is necessary when writing to AWS S3 Buckets with Object Lock enabled for example.
+    ///
+    /// Available options:
+    /// - "crc32c"
+    pub checksum_algorithm: Option<String>,
 }
 
 impl Debug for S3Config {
@@ -663,6 +669,17 @@ impl S3Builder {
         self
     }
 
+    /// Set checksum algorithm of this backend.
+    /// This is necessary when writing to AWS S3 Buckets with Object Lock enabled for example.
+    ///
+    /// Available options:
+    /// - "crc32c"
+    pub fn checksum_algorithm(&mut self, checksum_algorithm: &str) -> &mut Self {
+        self.config.checksum_algorithm = Some(checksum_algorithm.to_string());
+
+        self
+    }
+
     /// Detect region of S3 bucket.
     ///
     /// # Args
@@ -858,6 +875,17 @@ impl Builder for S3Builder {
                 })?),
             };
 
+        let checksum_algorithm = match self.config.checksum_algorithm.as_deref() {
+            Some("crc32c") => Some(ChecksumAlgorithm::Crc32c),
+            None => None,
+            _ => {
+                return Err(Error::new(
+                    ErrorKind::ConfigInvalid,
+                    "{v} is not a supported checksum_algorithm.",
+                ))
+            }
+        };
+
         let client = if let Some(client) = self.http_client.take() {
             client
         } else {
@@ -979,6 +1007,7 @@ impl Builder for S3Builder {
                 credential_loaded: AtomicBool::new(false),
                 client,
                 batch_max_operations,
+                checksum_algorithm,
             }),
         })
     }

core/src/services/s3/backend.rs

@@ -17,12 +17,15 @@
 
 use std::fmt;
 use std::fmt::Debug;
+use std::fmt::Display;
 use std::fmt::Formatter;
 use std::fmt::Write;
 use std::sync::atomic;
 use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use bytes::Bytes;
 use http::header::HeaderName;
 use http::header::CACHE_CONTROL;
@@ -88,6 +91,7 @@ pub struct S3Core {
     pub credential_loaded: AtomicBool,
     pub client: HttpClient,
     pub batch_max_operations: usize,
+    pub checksum_algorithm: Option<ChecksumAlgorithm>,
 }
 
 impl Debug for S3Core {
@@ -246,6 +250,35 @@ impl S3Core {
 
         req
     }
+
+    pub fn insert_checksum_header(
+        &self,
+        mut req: http::request::Builder,
+        body: &Buffer,
+    ) -> http::request::Builder {
+        if let Some(checksum_algorithm) = self.checksum_algorithm.as_ref() {
+            let checksum = match checksum_algorithm {
+                ChecksumAlgorithm::Crc32c => {
+                    let mut crc = 0u32;
+                    body.clone()
+                        .for_each(|b| crc = crc32c::crc32c_append(crc, &b));
+                    BASE64_STANDARD.encode(crc.to_be_bytes())
+                }
+            };
+            req = req.header(checksum_algorithm.to_header_name(), checksum);
+        }
+        req
+    }
+
+    pub fn insert_checksum_type_header(
+        &self,
+        mut req: http::request::Builder,
+    ) -> http::request::Builder {
+        if let Some(checksum_algorithm) = self.checksum_algorithm.as_ref() {
+            req = req.header("x-amz-checksum-algorithm", checksum_algorithm.to_string());
+        }
+        req
+    }
 }
 
 impl S3Core {
@@ -408,6 +441,9 @@ impl S3Core {
         // Set SSE headers.
         req = self.insert_sse_headers(req, true);
 
+        // Set Checksum header.
+        req = self.insert_checksum_header(req, &body);
+
         // Set body
         let req = req.body(body).map_err(new_request_build_error)?;
 
@@ -573,6 +609,9 @@ impl S3Core {
         // Set SSE headers.
         let req = self.insert_sse_headers(req, true);
 
+        // Set SSE headers.
+        let req = self.insert_checksum_type_header(req);
+
         let mut req = req.body(Buffer::new()).map_err(new_request_build_error)?;
 
         self.sign(&mut req).await?;
@@ -605,6 +644,9 @@ impl S3Core {
         // Set SSE headers.
         req = self.insert_sse_headers(req, true);
 
+        // Set Checksum header.
+        req = self.insert_checksum_header(req, &body);
+
         // Set body
         let req = req.body(body).map_err(new_request_build_error)?;
 
@@ -821,6 +863,28 @@ pub struct OutputCommonPrefix {
     pub prefix: String,
 }
 
+pub enum ChecksumAlgorithm {
+    Crc32c,
+}
+impl ChecksumAlgorithm {
+    pub fn to_header_name(&self) -> HeaderName {
+        match self {
+            Self::Crc32c => HeaderName::from_static("x-amz-checksum-crc32c"),
+        }
+    }
+}
+impl Display for ChecksumAlgorithm {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                Self::Crc32c => "CRC32C",
+            }
+        )
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use bytes::Buf;