Add support for checksum algorithms in aws

object_store/src/aws/checksum.rs

@@ -0,0 +1,79 @@
+// 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 ring::digest::digest as ring_digest;
+use ring::digest::SHA256 as DigestSha256;
+
+#[allow(non_camel_case_types)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+/// Enum representing checksum algorithm supported by S3.
+pub enum Checksum {
+    /// SHA-256 algorithm.
+    SHA256,
+
+    #[cfg(feature = "sha1")]
+    SHA1,
+
+    #[cfg(feature = "crc32")]
+    CRC32,
+
+    #[cfg(feature = "crc32c")]
+    CRC32C,
+}
+
+impl Checksum {
+    pub(super) fn digest(&self, bytes: &[u8]) -> Vec<u8> {
+        match self {
+            Self::SHA256 => ring_digest(&DigestSha256, bytes).as_ref().to_owned(),
+
+            #[cfg(feature = "sha1")]
+            Checksum::SHA1 => todo!(),
+
+            #[cfg(feature = "crc32")]
+            Checksum::CRC32 => todo!(),
+
+            #[cfg(feature = "crc32c")]
+            Checksum::CRC32C => todo!(),
+        }
+    }
+
+    pub(super) fn header_name(&self) -> &'static str {
+        match self {
+            Self::SHA256 => "x-amz-checksum-sha256",
+
+            #[cfg(feature = "sha1")]
+            Checksum::SHA1 => "x-amz-checksum-sha1",
+
+            #[cfg(feature = "crc32")]
+            Checksum::CRC32 => "x-amz-checksum-crc32",
+
+            #[cfg(feature = "crc32c")]
+            Checksum::CRC32C => "x-amz-checksum-crc32c",
+        }
+    }
+}
+
+impl TryFrom<&String> for Checksum {
+    type Error = ();
+
+    fn try_from(value: &String) -> Result<Self, Self::Error> {
+        match value.as_str() {
+            "sha256" => Ok(Self::SHA256),
+            _ => unreachable!(),
+        }
+    }
+}

object_store/src/aws/client.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::aws::checksum::Checksum;
 use crate::aws::credential::{AwsCredential, CredentialExt, CredentialProvider};
 use crate::aws::STRICT_PATH_ENCODE_SET;
 use crate::client::pagination::stream_paginated;
@@ -26,6 +27,8 @@ use crate::{
     BoxStream, ClientOptions, ListResult, MultipartId, ObjectMeta, Path, Result,
     RetryConfig, StreamExt,
 };
+use base64::prelude::BASE64_STANDARD;
+use base64::Engine;
 use bytes::{Buf, Bytes};
 use chrono::{DateTime, Utc};
 use percent_encoding::{utf8_percent_encode, PercentEncode};
@@ -205,6 +208,7 @@ pub struct S3Config {
     pub retry_config: RetryConfig,
     pub client_options: ClientOptions,
     pub sign_payload: bool,
+    pub checksum: Option<Checksum>,
 }
 
 impl S3Config {
@@ -262,6 +266,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                None,
             )
             .send_retry(&self.config.retry_config)
             .await
@@ -281,10 +286,22 @@ impl S3Client {
     ) -> Result<Response> {
         let credential = self.get_credential().await?;
         let url = self.config.path_url(path);
-
         let mut builder = self.client.request(Method::PUT, url);
+
+        let mut payload_checksum = None;
+
         if let Some(bytes) = bytes {
-            builder = builder.body(bytes)
+            payload_checksum = match self.config().checksum {
+                Some(checksum) => {
+                    let digest = checksum.digest(&bytes);
+                    builder = builder
+                        .header(checksum.header_name(), BASE64_STANDARD.encode(&digest));
+                    Some(digest)
+                }
+                None => None,
+            };
+
+            builder = builder.body(bytes);
         }
 
         if let Some(value) = self.config().client_options.get_content_type(path) {
@@ -298,6 +315,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                payload_checksum,
             )
             .send_retry(&self.config.retry_config)
             .await
@@ -325,6 +343,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                None,
             )
             .send_retry(&self.config.retry_config)
             .await
@@ -349,6 +368,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                None,
             )
             .send_retry(&self.config.retry_config)
             .await
@@ -395,6 +415,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                None,
             )
             .send_retry(&self.config.retry_config)
             .await
@@ -438,6 +459,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                None,
             )
             .send_retry(&self.config.retry_config)
             .await
@@ -482,6 +504,7 @@ impl S3Client {
                 &self.config.region,
                 "s3",
                 self.config.sign_payload,
+                None,
             )
             .send_retry(&self.config.retry_config)
             .await

object_store/src/aws/credential.rs

@@ -84,7 +84,7 @@ const AUTH_HEADER: &str = "authorization";
 const ALL_HEADERS: &[&str; 4] = &[DATE_HEADER, HASH_HEADER, TOKEN_HEADER, AUTH_HEADER];
 
 impl<'a> RequestSigner<'a> {
-    fn sign(&self, request: &mut Request) {
+    fn sign(&self, request: &mut Request, pre_calculated_digest: Option<Vec<u8>>) {
         if let Some(ref token) = self.credential.token {
             let token_val = HeaderValue::from_str(token).unwrap();
             request.headers_mut().insert(TOKEN_HEADER, token_val);
@@ -101,9 +101,13 @@ impl<'a> RequestSigner<'a> {
         request.headers_mut().insert(DATE_HEADER, date_val);
 
         let digest = if self.sign_payload {
-            match request.body() {
-                None => EMPTY_SHA256_HASH.to_string(),
-                Some(body) => hex_digest(body.as_bytes().unwrap()),
+            if let Some(digest) = pre_calculated_digest {
+                hex_encode(&digest)
+            } else {
+                match request.body() {
+                    None => EMPTY_SHA256_HASH.to_string(),
+                    Some(body) => hex_digest(body.as_bytes().unwrap()),
+                }
             }
         } else {
             UNSIGNED_PAYLOAD_LITERAL.to_string()
@@ -165,6 +169,7 @@ pub trait CredentialExt {
         region: &str,
         service: &str,
         sign_payload: bool,
+        payload_sha256: Option<Vec<u8>>,
     ) -> Self;
 }
 
@@ -175,6 +180,7 @@ impl CredentialExt for RequestBuilder {
         region: &str,
         service: &str,
         sign_payload: bool,
+        payload_sha256: Option<Vec<u8>>,
     ) -> Self {
         // Hack around lack of access to underlying request
         // https://github.com/seanmonstar/reqwest/issues/1212
@@ -193,7 +199,7 @@ impl CredentialExt for RequestBuilder {
             sign_payload,
         };
 
-        signer.sign(&mut request);
+        signer.sign(&mut request, payload_sha256);
 
         for header in ALL_HEADERS {
             if let Some(val) = request.headers_mut().remove(*header) {
@@ -627,7 +633,7 @@ mod tests {
             sign_payload: true,
         };
 
-        signer.sign(&mut request);
+        signer.sign(&mut request, None);
         assert_eq!(request.headers().get(AUTH_HEADER).unwrap(), "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20220806/us-east-1/ec2/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=a3c787a7ed37f7fdfbfd2d7056a3d7c9d85e6d52a2bfbec73793c0be6e7862d4")
     }
 
@@ -665,7 +671,7 @@ mod tests {
             sign_payload: false,
         };
 
-        signer.sign(&mut request);
+        signer.sign(&mut request, None);
         assert_eq!(request.headers().get(AUTH_HEADER).unwrap(), "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20220806/us-east-1/ec2/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=653c3d8ea261fd826207df58bc2bb69fbb5003e9eb3c0ef06e4a51f2a81d8699")
     }
 
@@ -702,7 +708,7 @@ mod tests {
             sign_payload: true,
         };
 
-        signer.sign(&mut request);
+        signer.sign(&mut request, None);
         assert_eq!(request.headers().get(AUTH_HEADER).unwrap(), "AWS4-HMAC-SHA256 Credential=H20ABqCkLZID4rLe/20220809/us-east-1/s3/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=9ebf2f92872066c99ac94e573b4e1b80f4dbb8a32b1e8e23178318746e7d1b4d")
     }