feat(core,services/s3): support supplying a precomputed checksum on write

core/core/src/raw/ops.rs

@@ -20,6 +20,7 @@
 //! By using ops, users can add more context for operation.
 
 use crate::BytesRange;
+use crate::Checksum;
 use crate::options;
 use crate::raw::*;
 
@@ -680,6 +681,7 @@ pub struct OpWrite {
     if_none_match: Option<String>,
     if_not_exists: bool,
     user_metadata: Option<HashMap<String, String>>,
+    checksum: Option<Checksum>,
 }
 
 impl OpWrite {
@@ -807,6 +809,17 @@ impl OpWrite {
     pub fn user_metadata(&self) -> Option<&HashMap<String, String>> {
         self.user_metadata.as_ref()
     }
+
+    /// Set a precomputed checksum for this write operation.
+    pub fn with_checksum(mut self, checksum: Checksum) -> Self {
+        self.checksum = Some(checksum);
+        self
+    }
+
+    /// Get the precomputed checksum from the op.
+    pub fn checksum(&self) -> Option<&Checksum> {
+        self.checksum.as_ref()
+    }
 }
 
 /// Args for `writer` operation.
@@ -858,6 +871,7 @@ impl From<options::WriteOptions> for (OpWrite, OpWriter) {
                 if_none_match: value.if_none_match,
                 if_not_exists: value.if_not_exists,
                 user_metadata: value.user_metadata,
+                checksum: value.checksum,
             },
             OpWriter { chunk: value.chunk },
         )

core/core/src/types/capability.rs

@@ -125,6 +125,8 @@ pub struct Capability {
     pub write_with_if_not_exists: bool,
     /// Indicates if custom user metadata can be attached during write operations.
     pub write_with_user_metadata: bool,
+    /// Indicates if a precomputed checksum can be supplied during write operations.
+    pub write_with_checksum: bool,
     /// Maximum size supported for multipart uploads.
     /// For example, AWS S3 supports up to 5GiB per part in multipart uploads.
     pub write_multi_max_size: Option<usize>,

core/core/src/types/operator/operator_futures.rs

@@ -925,6 +925,32 @@ impl<F: Future<Output = Result<Metadata>>> FutureWrite<F> {
         self.args.0.user_metadata = Some(HashMap::from_iter(data));
         self
     }
+
+    /// Sets a precomputed checksum to verify the integrity of this write.
+    ///
+    /// Refer to [`options::WriteOptions::checksum`] for more details.
+    ///
+    /// ### Example
+    ///
+    /// ```
+    /// # use opendal_core::Result;
+    /// # use opendal_core::Operator;
+    /// use opendal_core::Checksum;
+    ///
+    /// # async fn test(op: Operator) -> Result<()> {
+    /// // 32 raw bytes of a SHA-256 digest computed ahead of time.
+    /// let digest = vec![0u8; 32];
+    /// let _ = op
+    ///     .write_with("path/to/file", vec![0; 4096])
+    ///     .checksum(Checksum::sha256(digest))
+    ///     .await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn checksum(mut self, checksum: Checksum) -> Self {
+        self.args.0.checksum = Some(checksum);
+        self
+    }
 }
 
 /// Future that generated by [`Operator::writer_with`].
@@ -1264,6 +1290,14 @@ impl<F: Future<Output = Result<Writer>>> FutureWriter<F> {
         self.args.user_metadata = Some(HashMap::from_iter(data));
         self
     }
+
+    /// Sets a precomputed checksum to verify the integrity of this write.
+    ///
+    /// Refer to [`options::WriteOptions::checksum`] for more details.
+    pub fn checksum(mut self, checksum: Checksum) -> Self {
+        self.args.checksum = Some(checksum);
+        self
+    }
 }
 
 /// Future that generated by [`Operator::delete_with`].

core/core/src/types/options.rs

@@ -17,6 +17,7 @@
 
 //! Options module provides options definitions for operations.
 
+use crate::Checksum;
 use crate::raw::Timestamp;
 use crate::types::BytesRange;
 use std::collections::HashMap;
@@ -528,6 +529,23 @@ pub struct WriteOptions {
     /// - Lower operation costs
     /// - Better utilize network bandwidth
     pub chunk: Option<usize>,
+
+    /// Sets a precomputed checksum to verify the integrity of this write.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::write_with_checksum`] before using this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If supported, the checksum is sent alongside the data and the service
+    ///   rejects the write when the received data does not match.
+    /// - The checksum applies to the whole object. Services that fall back to
+    ///   multipart uploads cannot apply a whole-object checksum and will return
+    ///   an error; keep such writes within a single request (for example by
+    ///   raising the chunk size and disabling concurrency).
+    /// - If not supported, the value is ignored.
+    pub checksum: Option<Checksum>,
 }
 
 /// Options for copy operations.

core/core/src/types/write/checksum.rs

@@ -0,0 +1,100 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+
+/// The algorithm used to compute a [`Checksum`].
+///
+/// This mirrors the additional checksum algorithms that object storage
+/// services such as AWS S3 accept for verifying the integrity of uploaded
+/// data.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+#[non_exhaustive]
+pub enum ChecksumAlgorithm {
+    /// CRC32C (Castagnoli) checksum.
+    Crc32c,
+    /// SHA-1 checksum.
+    Sha1,
+    /// SHA-256 checksum.
+    Sha256,
+    /// CRC-64/NVME checksum.
+    Crc64Nvme,
+}
+
+/// A precomputed checksum of the data being written.
+///
+/// Supplying a checksum lets the service verify the integrity of an upload
+/// against a value you already know, instead of computing one from the body.
+///
+/// The `value` holds the raw digest bytes (for example the 32 bytes of a
+/// SHA-256 digest), not a hex or base64 encoded string. Services encode the
+/// value as required by their wire format.
+///
+/// # Example
+///
+/// ```
+/// use opendal_core::Checksum;
+///
+/// // 32 raw bytes of a SHA-256 digest computed ahead of time.
+/// let digest = vec![0u8; 32];
+/// let checksum = Checksum::sha256(digest);
+/// ```
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Checksum {
+    algorithm: ChecksumAlgorithm,
+    value: Bytes,
+}
+
+impl Checksum {
+    /// Create a checksum from an algorithm and its raw digest bytes.
+    pub fn new(algorithm: ChecksumAlgorithm, value: impl Into<Bytes>) -> Self {
+        Self {
+            algorithm,
+            value: value.into(),
+        }
+    }
+
+    /// Create a CRC32C checksum from its raw digest bytes.
+    pub fn crc32c(value: impl Into<Bytes>) -> Self {
+        Self::new(ChecksumAlgorithm::Crc32c, value)
+    }
+
+    /// Create a SHA-1 checksum from its raw digest bytes.
+    pub fn sha1(value: impl Into<Bytes>) -> Self {
+        Self::new(ChecksumAlgorithm::Sha1, value)
+    }
+
+    /// Create a SHA-256 checksum from its raw digest bytes.
+    pub fn sha256(value: impl Into<Bytes>) -> Self {
+        Self::new(ChecksumAlgorithm::Sha256, value)
+    }
+
+    /// Create a CRC-64/NVME checksum from its raw digest bytes.
+    pub fn crc64nvme(value: impl Into<Bytes>) -> Self {
+        Self::new(ChecksumAlgorithm::Crc64Nvme, value)
+    }
+
+    /// Get the algorithm of this checksum.
+    pub fn algorithm(&self) -> ChecksumAlgorithm {
+        self.algorithm
+    }
+
+    /// Get the raw digest bytes of this checksum.
+    pub fn value(&self) -> &Bytes {
+        &self.value
+    }
+}

core/core/src/types/write/mod.rs

@@ -18,6 +18,10 @@
 mod writer;
 pub use writer::Writer;
 
+mod checksum;
+pub use checksum::Checksum;
+pub use checksum::ChecksumAlgorithm;
+
 mod buffer_sink;
 pub use buffer_sink::BufferSink;
 mod futures_async_writer;

core/services/s3/src/backend.rs

@@ -46,6 +46,7 @@ use crate::S3_SCHEME;
 use crate::config::S3Config;
 use crate::copier::S3Copiers;
 use crate::copier::new_s3_copier;
+use crate::core::ChecksumAlgorithm;
 use crate::core::parse_error;
 use crate::core::*;
 use crate::deleter::S3Deleter;
@@ -951,6 +952,7 @@ impl Builder for S3Builder {
                     write_with_if_match: true,
                     write_with_if_not_exists: true,
                     write_with_user_metadata: true,
+                    write_with_checksum: true,
 
                     // The min multipart size of S3 is 5 MiB.
                     //
@@ -1102,6 +1104,16 @@ impl Service for S3Backend {
     }
 
     fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
+        // A supplied checksum covers the whole object, which only maps cleanly
+        // onto a single PutObject request. Append uploads write incrementally,
+        // so reject the combination instead of silently dropping the checksum.
+        if args.checksum().is_some() && args.append() {
+            return Err(Error::new(
+                ErrorKind::Unsupported,
+                "a precomputed checksum cannot be used together with append writes",
+            ));
+        }
+
         let output: S3Writers = {
             let writer = S3Writer::new(self.core.clone(), ctx.clone(), path, args.clone());
 

core/services/s3/src/core.rs

@@ -125,6 +125,28 @@ fn format_crc32c_iter(body: Buffer) -> String {
     BASE64_STANDARD.encode(crc.to_be_bytes())
 }
 
+/// Attach a user-supplied checksum to a request as the matching
+/// `x-amz-checksum-*` header, base64-encoding the raw digest bytes.
+fn insert_user_checksum_header(
+    req: http::request::Builder,
+    checksum: &Checksum,
+) -> Result<http::request::Builder> {
+    let header = match checksum.algorithm() {
+        opendal_core::ChecksumAlgorithm::Crc32c => "x-amz-checksum-crc32c",
+        opendal_core::ChecksumAlgorithm::Sha1 => "x-amz-checksum-sha1",
+        opendal_core::ChecksumAlgorithm::Sha256 => "x-amz-checksum-sha256",
+        opendal_core::ChecksumAlgorithm::Crc64Nvme => "x-amz-checksum-crc64nvme",
+        algorithm => {
+            return Err(Error::new(
+                ErrorKind::Unsupported,
+                format!("checksum algorithm {algorithm:?} is not supported by S3"),
+            ));
+        }
+    };
+
+    Ok(req.header(header, BASE64_STANDARD.encode(checksum.value())))
+}
+
 impl Debug for S3Core {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("S3Core")
@@ -574,9 +596,10 @@ impl S3Core {
         // Set SSE headers.
         req = self.insert_sse_headers(req, true);
 
-        // Calculate Checksum.
-        if let Some(checksum) = self.calculate_checksum(&body) {
-            // Set Checksum header.
+        // Prefer a user-supplied checksum, otherwise compute one if configured.
+        if let Some(checksum) = args.checksum() {
+            req = insert_user_checksum_header(req, checksum)?;
+        } else if let Some(checksum) = self.calculate_checksum(&body) {
             req = self.insert_checksum_header(req, &checksum);
         }
 
@@ -1588,6 +1611,18 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_insert_user_checksum_header() {
+        // The raw digest bytes are base64-encoded into the matching header.
+        let checksum = Checksum::sha256(Bytes::from_static(&[0x01, 0x02, 0x03]));
+        let req = insert_user_checksum_header(Request::put("https://example.com"), &checksum)
+            .expect("sha256 is supported")
+            .body(())
+            .unwrap();
+
+        assert_eq!(req.headers().get("x-amz-checksum-sha256").unwrap(), "AQID");
+    }
+
     /// This example is from https://docs.aws.amazon.com/AmazonS3/latest/API/API_CreateMultipartUpload.html#API_CreateMultipartUpload_Examples
     #[test]
     fn test_deserialize_initiate_multipart_upload_result() {

core/services/s3/src/writer.rs

@@ -22,6 +22,7 @@ use constants::X_AMZ_OBJECT_SIZE;
 use constants::X_AMZ_VERSION_ID;
 use http::StatusCode;
 
+use crate::core::ChecksumAlgorithm;
 use crate::core::S3Error;
 use crate::core::from_s3_error;
 use crate::core::parse_error;
@@ -112,6 +113,17 @@ impl oio::MultipartWrite for S3Writer {
         size: u64,
         body: Buffer,
     ) -> Result<oio::MultipartPart> {
+        // A supplied checksum covers the whole object and cannot be applied to
+        // individual parts, so reject it instead of silently uploading without
+        // the integrity check the caller asked for.
+        if self.op.checksum().is_some() {
+            return Err(Error::new(
+                ErrorKind::Unsupported,
+                "a precomputed checksum cannot be used with multipart uploads; \
+                 keep the write within a single request via a larger chunk size",
+            ));
+        }
+
         // AWS S3 requires part number must between [1..=10000]
         let part_number = part_number + 1;