Merge pull request 'implement STREAMING-*-PAYLOAD-TRAILER' (#960) from fix-824 into main

Reviewed-on: https://git.deuxfleurs.fr/Deuxfleurs/garage/pulls/960

2 files changed, 173 insertions, 12 deletions

diff --git a/src/garage/tests/s3/objects.rs b/src/garage/tests/s3/objects.rs
index 77eca2b1..dfc5253d 100644
--- a/src/garage/tests/s3/objects.rs
+++ b/src/garage/tests/s3/objects.rs
@@ -189,12 +189,14 @@ async fn test_getobject() {
 
 #[tokio::test]
 async fn test_metadata() {
+	use aws_sdk_s3::primitives::{DateTime, DateTimeFormat};
+
 	let ctx = common::context();
 	let bucket = ctx.create_bucket("testmetadata");
 
 	let etag = "\"46cf18a9b447991b450cad3facf5937e\"";
-	let exp = aws_sdk_s3::primitives::DateTime::from_secs(10000000000);
-	let exp2 = aws_sdk_s3::primitives::DateTime::from_secs(10000500000);
+	let exp = DateTime::from_secs(10000000000);
+	let exp2 = DateTime::from_secs(10000500000);
 
 	{
 		// Note. The AWS client SDK adds a Content-Type header
@@ -227,7 +229,7 @@ async fn test_metadata() {
 		assert_eq!(o.content_disposition, None);
 		assert_eq!(o.content_encoding, None);
 		assert_eq!(o.content_language, None);
-		assert_eq!(o.expires, None);
+		assert_eq!(o.expires_string, None);
 		assert_eq!(o.metadata.unwrap_or_default().len(), 0);
 
 		let o = ctx
@@ -250,7 +252,10 @@ async fn test_metadata() {
 		assert_eq!(o.content_disposition.unwrap().as_str(), "cddummy");
 		assert_eq!(o.content_encoding.unwrap().as_str(), "cedummy");
 		assert_eq!(o.content_language.unwrap().as_str(), "cldummy");
-		assert_eq!(o.expires.unwrap(), exp);
+		assert_eq!(
+			o.expires_string.unwrap(),
+			exp.fmt(DateTimeFormat::HttpDate).unwrap()
+		);
 	}
 
 	{
@@ -288,7 +293,10 @@ async fn test_metadata() {
 		assert_eq!(o.content_disposition.unwrap().as_str(), "cdtest");
 		assert_eq!(o.content_encoding.unwrap().as_str(), "cetest");
 		assert_eq!(o.content_language.unwrap().as_str(), "cltest");
-		assert_eq!(o.expires.unwrap(), exp2);
+		assert_eq!(
+			o.expires_string.unwrap(),
+			exp2.fmt(DateTimeFormat::HttpDate).unwrap()
+		);
 		let mut meta = o.metadata.unwrap();
 		assert_eq!(meta.remove("testmeta").unwrap(), "hello people");
 		assert_eq!(meta.remove("nice-unicode-meta").unwrap(), "宅配便");
@@ -314,7 +322,10 @@ async fn test_metadata() {
 		assert_eq!(o.content_disposition.unwrap().as_str(), "cddummy");
 		assert_eq!(o.content_encoding.unwrap().as_str(), "cedummy");
 		assert_eq!(o.content_language.unwrap().as_str(), "cldummy");
-		assert_eq!(o.expires.unwrap(), exp);
+		assert_eq!(
+			o.expires_string.unwrap(),
+			exp.fmt(DateTimeFormat::HttpDate).unwrap()
+		);
 	}
 }
 
diff --git a/src/garage/tests/s3/streaming_signature.rs b/src/garage/tests/s3/streaming_signature.rs
index 351aa422..a86feefc 100644
--- a/src/garage/tests/s3/streaming_signature.rs
+++ b/src/garage/tests/s3/streaming_signature.rs
@@ -1,5 +1,8 @@
 use std::collections::HashMap;
 
+use base64::prelude::*;
+use crc32fast::Hasher as Crc32;
+
 use crate::common;
 use crate::common::ext::CommandExt;
 use common::custom_requester::BodySignature;
@@ -21,7 +24,7 @@ async fn test_putobject_streaming() {
 		let content_type = "text/csv";
 		let mut headers = HashMap::new();
 		headers.insert("content-type".to_owned(), content_type.to_owned());
-		let _ = ctx
+		let res = ctx
 			.custom_request
 			.builder(bucket.clone())
 			.method(Method::PUT)
@@ -29,10 +32,11 @@ async fn test_putobject_streaming() {
 			.signed_headers(headers)
 			.vhost_style(true)
 			.body(vec![])
-			.body_signature(BodySignature::Streaming(10))
+			.body_signature(BodySignature::Streaming { chunk_size: 10 })
 			.send()
 			.await
 			.unwrap();
+		assert!(res.status().is_success(), "got response: {:?}", res);
 
 		// assert_eq!(r.e_tag.unwrap().as_str(), etag);
 		// We return a version ID here
@@ -65,7 +69,146 @@ async fn test_putobject_streaming() {
 	{
 		let etag = "\"46cf18a9b447991b450cad3facf5937e\"";
 
-		let _ = ctx
+		let mut crc32 = Crc32::new();
+		crc32.update(&BODY[..]);
+		let crc32 = BASE64_STANDARD.encode(&u32::to_be_bytes(crc32.finalize())[..]);
+
+		let mut headers = HashMap::new();
+		headers.insert("x-amz-checksum-crc32".to_owned(), crc32.clone());
+
+		let res = ctx
+			.custom_request
+			.builder(bucket.clone())
+			.method(Method::PUT)
+			//.path(CTRL_KEY.to_owned()) at the moment custom_request does not encode url so this
+			//fail
+			.path("abc".to_owned())
+			.vhost_style(true)
+			.signed_headers(headers)
+			.body(BODY.to_vec())
+			.body_signature(BodySignature::Streaming { chunk_size: 16 })
+			.send()
+			.await
+			.unwrap();
+		assert!(res.status().is_success(), "got response: {:?}", res);
+
+		// assert_eq!(r.e_tag.unwrap().as_str(), etag);
+		// assert!(r.version_id.is_some());
+
+		let o = ctx
+			.client
+			.get_object()
+			.bucket(&bucket)
+			//.key(CTRL_KEY)
+			.key("abc")
+			.checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
+			.send()
+			.await
+			.unwrap();
+
+		assert_bytes_eq!(o.body, BODY);
+		assert_eq!(o.e_tag.unwrap(), etag);
+		assert!(o.last_modified.is_some());
+		assert_eq!(o.content_length.unwrap(), 62);
+		assert_eq!(o.parts_count, None);
+		assert_eq!(o.tag_count, None);
+		assert_eq!(o.checksum_crc32.unwrap(), crc32);
+	}
+}
+
+#[tokio::test]
+async fn test_putobject_streaming_unsigned_trailer() {
+	let ctx = common::context();
+	let bucket = ctx.create_bucket("putobject-streaming-unsigned-trailer");
+
+	{
+		// Send an empty object (can serve as a directory marker)
+		// with a content type
+		let etag = "\"d41d8cd98f00b204e9800998ecf8427e\"";
+		let content_type = "text/csv";
+		let mut headers = HashMap::new();
+		headers.insert("content-type".to_owned(), content_type.to_owned());
+
+		let empty_crc32 = BASE64_STANDARD.encode(&u32::to_be_bytes(Crc32::new().finalize())[..]);
+
+		let res = ctx
+			.custom_request
+			.builder(bucket.clone())
+			.method(Method::PUT)
+			.path(STD_KEY.to_owned())
+			.signed_headers(headers)
+			.vhost_style(true)
+			.body(vec![])
+			.body_signature(BodySignature::StreamingUnsignedTrailer {
+				chunk_size: 10,
+				trailer_algorithm: "x-amz-checksum-crc32".into(),
+				trailer_value: empty_crc32,
+			})
+			.send()
+			.await
+			.unwrap();
+		assert!(res.status().is_success(), "got response: {:?}", res);
+
+		// assert_eq!(r.e_tag.unwrap().as_str(), etag);
+		// We return a version ID here
+		// We should check if Amazon is returning one when versioning is not enabled
+		// assert!(r.version_id.is_some());
+
+		//let _version = r.version_id.unwrap();
+
+		let o = ctx
+			.client
+			.get_object()
+			.bucket(&bucket)
+			.key(STD_KEY)
+			.send()
+			.await
+			.unwrap();
+
+		assert_bytes_eq!(o.body, b"");
+		assert_eq!(o.e_tag.unwrap(), etag);
+		// We do not return version ID
+		// We should check if Amazon is returning one when versioning is not enabled
+		// assert_eq!(o.version_id.unwrap(), _version);
+		assert_eq!(o.content_type.unwrap(), content_type);
+		assert!(o.last_modified.is_some());
+		assert_eq!(o.content_length.unwrap(), 0);
+		assert_eq!(o.parts_count, None);
+		assert_eq!(o.tag_count, None);
+	}
+
+	{
+		let etag = "\"46cf18a9b447991b450cad3facf5937e\"";
+
+		let mut crc32 = Crc32::new();
+		crc32.update(&BODY[..]);
+		let crc32 = BASE64_STANDARD.encode(&u32::to_be_bytes(crc32.finalize())[..]);
+
+		// try sending with wrong crc32, check that it fails
+		let err_res = ctx
+			.custom_request
+			.builder(bucket.clone())
+			.method(Method::PUT)
+			//.path(CTRL_KEY.to_owned()) at the moment custom_request does not encode url so this
+			//fail
+			.path("abc".to_owned())
+			.vhost_style(true)
+			.body(BODY.to_vec())
+			.body_signature(BodySignature::StreamingUnsignedTrailer {
+				chunk_size: 16,
+				trailer_algorithm: "x-amz-checksum-crc32".into(),
+				trailer_value: "2Yp9Yw==".into(),
+			})
+			.send()
+			.await
+			.unwrap();
+		assert!(
+			err_res.status().is_client_error(),
+			"got response: {:?}",
+			err_res
+		);
+
+		let res = ctx
 			.custom_request
 			.builder(bucket.clone())
 			.method(Method::PUT)
@@ -74,10 +217,15 @@ async fn test_putobject_streaming() {
 			.path("abc".to_owned())
 			.vhost_style(true)
 			.body(BODY.to_vec())
-			.body_signature(BodySignature::Streaming(16))
+			.body_signature(BodySignature::StreamingUnsignedTrailer {
+				chunk_size: 16,
+				trailer_algorithm: "x-amz-checksum-crc32".into(),
+				trailer_value: crc32.clone(),
+			})
 			.send()
 			.await
 			.unwrap();
+		assert!(res.status().is_success(), "got response: {:?}", res);
 
 		// assert_eq!(r.e_tag.unwrap().as_str(), etag);
 		// assert!(r.version_id.is_some());
@@ -88,6 +236,7 @@ async fn test_putobject_streaming() {
 			.bucket(&bucket)
 			//.key(CTRL_KEY)
 			.key("abc")
+			.checksum_mode(aws_sdk_s3::types::ChecksumMode::Enabled)
 			.send()
 			.await
 			.unwrap();
@@ -98,6 +247,7 @@ async fn test_putobject_streaming() {
 		assert_eq!(o.content_length.unwrap(), 62);
 		assert_eq!(o.parts_count, None);
 		assert_eq!(o.tag_count, None);
+		assert_eq!(o.checksum_crc32.unwrap(), crc32);
 	}
 }
 
@@ -119,7 +269,7 @@ async fn test_create_bucket_streaming() {
 			.custom_request
 			.builder(bucket.to_owned())
 			.method(Method::PUT)
-			.body_signature(BodySignature::Streaming(10))
+			.body_signature(BodySignature::Streaming { chunk_size: 10 })
 			.send()
 			.await
 			.unwrap();
@@ -174,7 +324,7 @@ async fn test_put_website_streaming() {
 			.method(Method::PUT)
 			.query_params(query)
 			.body(website_config.as_bytes().to_vec())
-			.body_signature(BodySignature::Streaming(10))
+			.body_signature(BodySignature::Streaming { chunk_size: 10 })
 			.send()
 			.await
 			.unwrap();