diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs index 5e2d7679c316..f755c6a8dfea 100644 --- a/core/core/src/raw/ops.rs +++ b/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, if_not_exists: bool, user_metadata: Option>, + checksum: Option, } impl OpWrite { @@ -807,6 +809,17 @@ impl OpWrite { pub fn user_metadata(&self) -> Option<&HashMap> { 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 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 }, ) diff --git a/core/core/src/types/capability.rs b/core/core/src/types/capability.rs index f24064a58b8f..bbb01f52a49a 100644 --- a/core/core/src/types/capability.rs +++ b/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, diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs index 3c521ec61c1e..b62e591be6cc 100644 --- a/core/core/src/types/operator/operator_futures.rs +++ b/core/core/src/types/operator/operator_futures.rs @@ -925,6 +925,32 @@ impl>> FutureWrite { 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>> FutureWriter { 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`]. diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs index c35dd49b79c7..c7259b96cf1a 100644 --- a/core/core/src/types/options.rs +++ b/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, + + /// 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, } /// Options for copy operations. diff --git a/core/core/src/types/write/checksum.rs b/core/core/src/types/write/checksum.rs new file mode 100644 index 000000000000..ac44bcc6818e --- /dev/null +++ b/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) -> Self { + Self { + algorithm, + value: value.into(), + } + } + + /// Create a CRC32C checksum from its raw digest bytes. + pub fn crc32c(value: impl Into) -> Self { + Self::new(ChecksumAlgorithm::Crc32c, value) + } + + /// Create a SHA-1 checksum from its raw digest bytes. + pub fn sha1(value: impl Into) -> Self { + Self::new(ChecksumAlgorithm::Sha1, value) + } + + /// Create a SHA-256 checksum from its raw digest bytes. + pub fn sha256(value: impl Into) -> Self { + Self::new(ChecksumAlgorithm::Sha256, value) + } + + /// Create a CRC-64/NVME checksum from its raw digest bytes. + pub fn crc64nvme(value: impl Into) -> 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 + } +} diff --git a/core/core/src/types/write/mod.rs b/core/core/src/types/write/mod.rs index 1feaadf8749e..e5cdc54442ee 100644 --- a/core/core/src/types/write/mod.rs +++ b/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; diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs index 624351c6cd6f..ee81ec9b0b77 100644 --- a/core/services/s3/src/backend.rs +++ b/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 { + // 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()); diff --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs index 77877b341b59..0506d495a89d 100644 --- a/core/services/s3/src/core.rs +++ b/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 { + 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() { diff --git a/core/services/s3/src/writer.rs b/core/services/s3/src/writer.rs index 8d0b9130a94e..2f9f0fffc4a5 100644 --- a/core/services/s3/src/writer.rs +++ b/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 { + // 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;