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

Reviewed-on: #960
2025-02-19 09:59:32 +00:00 · 2025-02-19 09:59:32 +00:00 · f64ec6e542
commit f64ec6e542
parent 859b38b0d2 6d38907dac
36 changed files with 1366 additions and 578 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -238,9 +238,9 @@ checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26"

 [[package]]
 name = "aws-credential-types"
-version = "1.1.4"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "33cc49dcdd31c8b6e79850a179af4c367669150c7ac0135f176c61bec81a70f7"
+checksum = "60e8f6b615cb5fc60a98132268508ad104310f0cfb25a1c22eee76efdf9154da"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-runtime-api",
@ -250,15 +250,16 @@ dependencies = [

 [[package]]
 name = "aws-runtime"
-version = "1.1.4"
+version = "1.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eb031bff99877c26c28895766f7bb8484a05e24547e370768d6cc9db514662aa"
+checksum = "76dd04d39cc12844c0994f2c9c5a6f5184c22e9188ec1ff723de41910a21dcad"
 dependencies = [
 "aws-credential-types",
 "aws-sigv4",
 "aws-smithy-async",
 "aws-smithy-eventstream",
 "aws-smithy-http",
+ "aws-smithy-runtime",
 "aws-smithy-runtime-api",
 "aws-smithy-types",
 "aws-types",
@ -266,6 +267,7 @@ dependencies = [
 "fastrand",
 "http 0.2.12",
 "http-body 0.4.6",
+ "once_cell",
 "percent-encoding",
 "pin-project-lite",
 "tracing",
@ -274,9 +276,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-config"
-version = "1.13.0"
+version = "1.62.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4af4f5b0f64563ada272e009cc95027effb546110ed85d014611420ac0d97858"
+checksum = "0f94d79b8eef608af51b5415d13f5c670dec177880c6f78cd27bea968e5c9b76"
 dependencies = [
 "aws-credential-types",
 "aws-runtime",
@ -296,9 +298,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.14.0"
+version = "1.68.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "951f7730f51a2155c711c85c79f337fbc02a577fa99d2a0a8059acfce5392113"
+checksum = "bc5ddf1dc70287dc9a2f953766a1fe15e3e74aef02fd1335f2afa475c9b4f4fc"
 dependencies = [
 "aws-credential-types",
 "aws-runtime",
@ -314,20 +316,25 @@ dependencies = [
 "aws-smithy-xml",
 "aws-types",
 "bytes",
+ "fastrand",
+ "hex",
+ "hmac",
 "http 0.2.12",
 "http-body 0.4.6",
+ "lru",
 "once_cell",
 "percent-encoding",
 "regex-lite",
+ "sha2",
 "tracing",
 "url",
 ]

 [[package]]
 name = "aws-sigv4"
-version = "1.1.4"
+version = "1.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c371c6b0ac54d4605eb6f016624fb5c7c2925d315fdf600ac1bf21b19d5f1742"
+checksum = "9bfe75fad52793ce6dec0dc3d4b1f388f038b5eb866c8d4d7f3a8e21b5ea5051"
 dependencies = [
 "aws-credential-types",
 "aws-smithy-eventstream",
@ -354,9 +361,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-async"
-version = "1.1.4"
+version = "1.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72ee2d09cce0ef3ae526679b522835d63e75fb427aca5413cd371e490d52dcc6"
+checksum = "fa59d1327d8b5053c54bf2eaae63bf629ba9e904434d0835a28ed3c0ed0a614e"
 dependencies = [
 "futures-util",
 "pin-project-lite",
@ -365,9 +372,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-checksums"
-version = "0.60.4"
+version = "0.60.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be2acd1b9c6ae5859999250ed5a62423aedc5cf69045b844432de15fa2f31f2b"
+checksum = "ba1a71073fca26775c8b5189175ea8863afb1c9ea2cceb02a5de5ad9dfbaa795"
 dependencies = [
 "aws-smithy-http",
 "aws-smithy-types",
@ -386,9 +393,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-eventstream"
-version = "0.60.4"
+version = "0.60.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e6363078f927f612b970edf9d1903ef5cef9a64d1e8423525ebb1f0a1633c858"
+checksum = "8b18559a41e0c909b77625adf2b8c50de480a8041e5e4a3f5f7d177db70abc5a"
 dependencies = [
 "aws-smithy-types",
 "bytes",
@ -397,9 +404,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-http"
-version = "0.60.4"
+version = "0.60.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dab56aea3cd9e1101a0a999447fb346afb680ab1406cebc44b32346e25b4117d"
+checksum = "7809c27ad8da6a6a68c454e651d4962479e81472aa19ae99e59f9aba1f9713cc"
 dependencies = [
 "aws-smithy-eventstream",
 "aws-smithy-runtime-api",
@ -418,18 +425,18 @@ dependencies = [

 [[package]]
 name = "aws-smithy-json"
-version = "0.60.4"
+version = "0.61.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fd3898ca6518f9215f62678870064398f00031912390efd03f1f6ef56d83aa8e"
+checksum = "623a51127f24c30776c8b374295f2df78d92517386f77ba30773f15a30ce1422"
 dependencies = [
 "aws-smithy-types",
 ]

 [[package]]
 name = "aws-smithy-runtime"
-version = "1.1.4"
+version = "1.7.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fafdab38f40ad7816e7da5dec279400dd505160780083759f01441af1bbb10ea"
+checksum = "d526a12d9ed61fadefda24abe2e682892ba288c2018bcb38b1b4c111d13f6d92"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-http",
@ -440,6 +447,8 @@ dependencies = [
 "h2 0.3.24",
 "http 0.2.12",
 "http-body 0.4.6",
+ "http-body 1.0.1",
+ "httparse",
 "hyper 0.14.32",
 "hyper-rustls 0.24.2",
 "once_cell",
@ -452,31 +461,36 @@ dependencies = [

 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.1.4"
+version = "1.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c18276dd28852f34b3bf501f4f3719781f4999a51c7bff1a5c6dc8c4529adc29"
+checksum = "92165296a47a812b267b4f41032ff8069ab7ff783696d217f0994a0d7ab585cd"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-types",
 "bytes",
 "http 0.2.12",
+ "http 1.2.0",
 "pin-project-lite",
 "tokio",
 "tracing",
+ "zeroize",
 ]

 [[package]]
 name = "aws-smithy-types"
-version = "1.1.4"
+version = "1.2.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb3e134004170d3303718baa2a4eb4ca64ee0a1c0a7041dca31b38be0fb414f3"
+checksum = "c7b8a53819e42f10d0821f56da995e1470b199686a1809168db6ca485665f042"
 dependencies = [
 "base64-simd",
 "bytes",
 "bytes-utils",
 "futures-core",
 "http 0.2.12",
+ "http 1.2.0",
 "http-body 0.4.6",
+ "http-body 1.0.1",
+ "http-body-util",
 "itoa",
 "num-integer",
 "pin-project-lite",
@ -490,24 +504,23 @@ dependencies = [

 [[package]]
 name = "aws-smithy-xml"
-version = "0.60.4"
+version = "0.60.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8604a11b25e9ecaf32f9aa56b9fe253c5e2f606a3477f0071e96d3155a5ed218"
+checksum = "ab0b0166827aa700d3dc519f72f8b3a91c35d0b8d042dc5d643a91e6f80648fc"
 dependencies = [
 "xmlparser",
 ]

 [[package]]
 name = "aws-types"
-version = "1.1.4"
+version = "1.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "789bbe008e65636fe1b6dbbb374c40c8960d1232b96af5ff4aec349f9c4accf4"
+checksum = "dfbd0a668309ec1f66c0f6bda4840dd6d4796ae26d699ebc266d7cc95c6d040f"
 dependencies = [
 "aws-credential-types",
 "aws-smithy-async",
 "aws-smithy-runtime-api",
 "aws-smithy-types",
- "http 0.2.12",
 "rustc_version",
 "tracing",
 ]
@ -1116,6 +1129,12 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"

+[[package]]
+name = "foldhash"
+version = "0.1.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a0d2fde1f7b3d48b8395d5f2de76c18a528bd6a9cdde438df747bfcba3e05d6f"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@ -1230,6 +1249,7 @@ dependencies = [
 "bytes",
 "bytesize",
 "chrono",
+ "crc32fast",
 "format_table",
 "futures",
 "garage_api_admin",
@ -1301,8 +1321,11 @@ dependencies = [
 name = "garage_api_common"
 version = "1.0.1"
 dependencies = [
+ "base64 0.21.7",
 "bytes",
 "chrono",
+ "crc32c",
+ "crc32fast",
 "crypto-common",
 "err-derive",
 "futures",
@ -1316,11 +1339,13 @@ dependencies = [
 "hyper 1.6.0",
 "hyper-util",
 "idna 0.5.0",
+ "md-5",
 "nom",
 "opentelemetry",
 "pin-project",
 "serde",
 "serde_json",
+ "sha1",
 "sha2",
 "tokio",
 "tracing",
@ -1738,6 +1763,11 @@ name = "hashbrown"
 version = "0.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash",
+]

 [[package]]
 name = "hashlink"
@ -2594,6 +2624,15 @@ version = "0.4.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04cbf5b083de1c7e0222a7a51dbfdba1cbe1c6ab0b15e29fff3f6c077fd9cd9f"

+[[package]]
+name = "lru"
+version = "0.12.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38"
+dependencies = [
+ "hashbrown 0.15.2",
+]
+
 [[package]]
 name = "matchers"
 version = "0.1.0"
@ -4909,7 +4948,7 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.52.0",
 ]

 [[package]]
--- a/Cargo.toml
+++ b/Cargo.toml
@ -141,8 +141,8 @@ thiserror = "1.0"
 assert-json-diff = "2.0"
 rustc_version = "0.4.0"
 static_init = "1.0"
-aws-sdk-config = "1.13"
-aws-sdk-s3 = "1.14"
+aws-sdk-config = "1.62"
+aws-sdk-s3 = "=1.68"

 [profile.dev]
 #lto = "thin"	# disabled for now, adds 2-4 min to each CI build
--- a/script/dev-cluster.sh
+++ b/script/dev-cluster.sh
@ -11,7 +11,7 @@ PATH="${GARAGE_DEBUG}:${GARAGE_RELEASE}:${NIX_RELEASE}:$PATH"
 FANCYCOLORS=("41m" "42m" "44m" "45m" "100m" "104m")

 export RUST_BACKTRACE=1 
-export RUST_LOG=garage=info,garage_api=debug
+export RUST_LOG=garage=info,garage_api_common=debug,garage_api_s3=debug
 MAIN_LABEL="\e[${FANCYCOLORS[0]}[main]\e[49m"

 if [ -z "$GARAGE_BIN" ]; then
--- a/src/api/common/Cargo.toml
+++ b/src/api/common/Cargo.toml
@ -18,16 +18,21 @@ garage_model.workspace = true
 garage_table.workspace = true
 garage_util.workspace = true

+base64.workspace = true
 bytes.workspace = true
 chrono.workspace = true
+crc32fast.workspace = true
+crc32c.workspace = true
 crypto-common.workspace = true
 err-derive.workspace = true
 hex.workspace = true
 hmac.workspace = true
+md-5.workspace = true
 idna.workspace = true
 tracing.workspace = true
 nom.workspace = true
 pin-project.workspace = true
+sha1.workspace = true
 sha2.workspace = true

 futures.workspace = true
--- a/src/api/common/cors.rs
+++ b/src/api/common/cors.rs
@ -14,9 +14,9 @@ use crate::common_error::{
 };
 use crate::helpers::*;

-pub fn find_matching_cors_rule<'a>(
+pub fn find_matching_cors_rule<'a, B>(
 	bucket_params: &'a BucketParams,
-	req: &Request<impl Body>,
+	req: &Request<B>,
 ) -> Result<Option<&'a GarageCorsRule>, CommonError> {
 	if let Some(cors_config) = bucket_params.cors_config.get() {
 		if let Some(origin) = req.headers().get("Origin") {
@ -132,8 +132,8 @@ pub async fn handle_options_api(
 	}
 }

-pub fn handle_options_for_bucket(
-	req: &Request<IncomingBody>,
+pub fn handle_options_for_bucket<B>(
+	req: &Request<B>,
 	bucket_params: &BucketParams,
 ) -> Result<Response<EmptyBody>, CommonError> {
 	let origin = req
--- a/src/api/common/signature/body.rs
+++ b/src/api/common/signature/body.rs
@ -0,0 +1,135 @@
+use std::sync::Mutex;
+
+use futures::prelude::*;
+use futures::stream::BoxStream;
+use http_body_util::{BodyExt, StreamBody};
+use hyper::body::{Bytes, Frame};
+use serde::Deserialize;
+use tokio::sync::mpsc;
+use tokio::task;
+
+use super::*;
+
+use crate::signature::checksum::*;
+
+pub struct ReqBody {
+	// why need mutex to be sync??
+	pub(crate) stream: Mutex<BoxStream<'static, Result<Frame<Bytes>, Error>>>,
+	pub(crate) checksummer: Checksummer,
+	pub(crate) expected_checksums: ExpectedChecksums,
+	pub(crate) trailer_algorithm: Option<ChecksumAlgorithm>,
+}
+
+pub type StreamingChecksumReceiver = task::JoinHandle<Result<Checksums, Error>>;
+
+impl ReqBody {
+	pub fn add_expected_checksums(&mut self, more: ExpectedChecksums) {
+		if more.md5.is_some() {
+			self.expected_checksums.md5 = more.md5;
+		}
+		if more.sha256.is_some() {
+			self.expected_checksums.sha256 = more.sha256;
+		}
+		if more.extra.is_some() {
+			self.expected_checksums.extra = more.extra;
+		}
+		self.checksummer.add_expected(&self.expected_checksums);
+	}
+
+	pub fn add_md5(&mut self) {
+		self.checksummer.add_md5();
+	}
+
+	// ============ non-streaming =============
+
+	pub async fn json<T: for<'a> Deserialize<'a>>(self) -> Result<T, Error> {
+		let body = self.collect().await?;
+		let resp: T = serde_json::from_slice(&body).ok_or_bad_request("Invalid JSON")?;
+		Ok(resp)
+	}
+
+	pub async fn collect(self) -> Result<Bytes, Error> {
+		self.collect_with_checksums().await.map(|(b, _)| b)
+	}
+
+	pub async fn collect_with_checksums(mut self) -> Result<(Bytes, Checksums), Error> {
+		let stream: BoxStream<_> = self.stream.into_inner().unwrap();
+		let bytes = BodyExt::collect(StreamBody::new(stream)).await?.to_bytes();
+
+		self.checksummer.update(&bytes);
+		let checksums = self.checksummer.finalize();
+		checksums.verify(&self.expected_checksums)?;
+
+		Ok((bytes, checksums))
+	}
+
+	// ============ streaming =============
+
+	pub fn streaming_with_checksums(
+		self,
+	) -> (
+		BoxStream<'static, Result<Bytes, Error>>,
+		StreamingChecksumReceiver,
+	) {
+		let Self {
+			stream,
+			mut checksummer,
+			mut expected_checksums,
+			trailer_algorithm,
+		} = self;
+
+		let (frame_tx, mut frame_rx) = mpsc::channel::<Frame<Bytes>>(5);
+
+		let join_checksums = tokio::spawn(async move {
+			while let Some(frame) = frame_rx.recv().await {
+				match frame.into_data() {
+					Ok(data) => {
+						checksummer = tokio::task::spawn_blocking(move || {
+							checksummer.update(&data);
+							checksummer
+						})
+						.await
+						.unwrap()
+					}
+					Err(frame) => {
+						let trailers = frame.into_trailers().unwrap();
+						let algo = trailer_algorithm.unwrap();
+						expected_checksums.extra = Some(extract_checksum_value(&trailers, algo)?);
+						break;
+					}
+				}
+			}
+
+			if trailer_algorithm.is_some() && expected_checksums.extra.is_none() {
+				return Err(Error::bad_request("trailing checksum was not sent"));
+			}
+
+			let checksums = checksummer.finalize();
+			checksums.verify(&expected_checksums)?;
+
+			Ok(checksums)
+		});
+
+		let stream: BoxStream<_> = stream.into_inner().unwrap();
+		let stream = stream.filter_map(move |x| {
+			let frame_tx = frame_tx.clone();
+			async move {
+				match x {
+					Err(e) => Some(Err(e)),
+					Ok(frame) => {
+						if frame.is_data() {
+							let data = frame.data_ref().unwrap().clone();
+							let _ = frame_tx.send(frame).await;
+							Some(Ok(data))
+						} else {
+							let _ = frame_tx.send(frame).await;
+							None
+						}
+					}
+				}
+			}
+		});
+
+		(stream.boxed(), join_checksums)
+	}
+}
--- a/src/api/common/signature/checksum.rs
+++ b/src/api/common/signature/checksum.rs
@ -11,11 +11,12 @@ use sha2::Sha256;
 use http::{HeaderMap, HeaderName, HeaderValue};

 use garage_util::data::*;
-use garage_util::error::OkOrMessage;

-use garage_model::s3::object_table::*;
+use super::*;

-use crate::error::*;
+pub use garage_model::s3::object_table::{ChecksumAlgorithm, ChecksumValue};
+
+pub const CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");

 pub const X_AMZ_CHECKSUM_ALGORITHM: HeaderName =
 	HeaderName::from_static("x-amz-checksum-algorithm");
@ -31,8 +32,8 @@ pub type Md5Checksum = [u8; 16];
 pub type Sha1Checksum = [u8; 20];
 pub type Sha256Checksum = [u8; 32];

-#[derive(Debug, Default)]
-pub(crate) struct ExpectedChecksums {
+#[derive(Debug, Default, Clone)]
+pub struct ExpectedChecksums {
 	// base64-encoded md5 (content-md5 header)
 	pub md5: Option<String>,
 	// content_sha256 (as a Hash / FixedBytes32)
@ -41,7 +42,7 @@ pub(crate) struct ExpectedChecksums {
 	pub extra: Option<ChecksumValue>,
 }

-pub(crate) struct Checksummer {
+pub struct Checksummer {
 	pub crc32: Option<Crc32>,
 	pub crc32c: Option<Crc32c>,
 	pub md5: Option<Md5>,
@ -50,7 +51,7 @@ pub(crate) struct Checksummer {
 }

 #[derive(Default)]
-pub(crate) struct Checksums {
+pub struct Checksums {
 	pub crc32: Option<Crc32Checksum>,
 	pub crc32c: Option<Crc32cChecksum>,
 	pub md5: Option<Md5Checksum>,
@ -59,34 +60,48 @@ pub(crate) struct Checksums {
 }

 impl Checksummer {
-	pub(crate) fn init(expected: &ExpectedChecksums, require_md5: bool) -> Self {
-		let mut ret = Self {
+	pub fn new() -> Self {
+		Self {
 			crc32: None,
 			crc32c: None,
 			md5: None,
 			sha1: None,
 			sha256: None,
-		};
+		}
+	}

-		if expected.md5.is_some() || require_md5 {
-			ret.md5 = Some(Md5::new());
-		}
-		if expected.sha256.is_some() || matches!(&expected.extra, Some(ChecksumValue::Sha256(_))) {
-			ret.sha256 = Some(Sha256::new());
-		}
-		if matches!(&expected.extra, Some(ChecksumValue::Crc32(_))) {
-			ret.crc32 = Some(Crc32::new());
-		}
-		if matches!(&expected.extra, Some(ChecksumValue::Crc32c(_))) {
-			ret.crc32c = Some(Crc32c::default());
-		}
-		if matches!(&expected.extra, Some(ChecksumValue::Sha1(_))) {
-			ret.sha1 = Some(Sha1::new());
+	pub fn init(expected: &ExpectedChecksums, add_md5: bool) -> Self {
+		let mut ret = Self::new();
+		ret.add_expected(expected);
+		if add_md5 {
+			ret.add_md5();
 		}
 		ret
 	}

-	pub(crate) fn add(mut self, algo: Option<ChecksumAlgorithm>) -> Self {
+	pub fn add_md5(&mut self) {
+		self.md5 = Some(Md5::new());
+	}
+
+	pub fn add_expected(&mut self, expected: &ExpectedChecksums) {
+		if expected.md5.is_some() {
+			self.md5 = Some(Md5::new());
+		}
+		if expected.sha256.is_some() || matches!(&expected.extra, Some(ChecksumValue::Sha256(_))) {
+			self.sha256 = Some(Sha256::new());
+		}
+		if matches!(&expected.extra, Some(ChecksumValue::Crc32(_))) {
+			self.crc32 = Some(Crc32::new());
+		}
+		if matches!(&expected.extra, Some(ChecksumValue::Crc32c(_))) {
+			self.crc32c = Some(Crc32c::default());
+		}
+		if matches!(&expected.extra, Some(ChecksumValue::Sha1(_))) {
+			self.sha1 = Some(Sha1::new());
+		}
+	}
+
+	pub fn add(mut self, algo: Option<ChecksumAlgorithm>) -> Self {
 		match algo {
 			Some(ChecksumAlgorithm::Crc32) => {
 				self.crc32 = Some(Crc32::new());
@ -105,7 +120,7 @@ impl Checksummer {
 		self
 	}

-	pub(crate) fn update(&mut self, bytes: &[u8]) {
+	pub fn update(&mut self, bytes: &[u8]) {
 		if let Some(crc32) = &mut self.crc32 {
 			crc32.update(bytes);
 		}
@ -123,7 +138,7 @@ impl Checksummer {
 		}
 	}

-	pub(crate) fn finalize(self) -> Checksums {
+	pub fn finalize(self) -> Checksums {
 		Checksums {
 			crc32: self.crc32.map(|x| u32::to_be_bytes(x.finalize())),
 			crc32c: self
@ -183,153 +198,56 @@ impl Checksums {

 // ----

-#[derive(Default)]
-pub(crate) struct MultipartChecksummer {
-	pub md5: Md5,
-	pub extra: Option<MultipartExtraChecksummer>,
-}
-
-pub(crate) enum MultipartExtraChecksummer {
-	Crc32(Crc32),
-	Crc32c(Crc32c),
-	Sha1(Sha1),
-	Sha256(Sha256),
-}
-
-impl MultipartChecksummer {
-	pub(crate) fn init(algo: Option<ChecksumAlgorithm>) -> Self {
-		Self {
-			md5: Md5::new(),
-			extra: match algo {
-				None => None,
-				Some(ChecksumAlgorithm::Crc32) => {
-					Some(MultipartExtraChecksummer::Crc32(Crc32::new()))
-				}
-				Some(ChecksumAlgorithm::Crc32c) => {
-					Some(MultipartExtraChecksummer::Crc32c(Crc32c::default()))
-				}
-				Some(ChecksumAlgorithm::Sha1) => Some(MultipartExtraChecksummer::Sha1(Sha1::new())),
-				Some(ChecksumAlgorithm::Sha256) => {
-					Some(MultipartExtraChecksummer::Sha256(Sha256::new()))
-				}
-			},
-		}
-	}
-
-	pub(crate) fn update(
-		&mut self,
-		etag: &str,
-		checksum: Option<ChecksumValue>,
-	) -> Result<(), Error> {
-		self.md5
-			.update(&hex::decode(&etag).ok_or_message("invalid etag hex")?);
-		match (&mut self.extra, checksum) {
-			(None, _) => (),
-			(
-				Some(MultipartExtraChecksummer::Crc32(ref mut crc32)),
-				Some(ChecksumValue::Crc32(x)),
-			) => {
-				crc32.update(&x);
-			}
-			(
-				Some(MultipartExtraChecksummer::Crc32c(ref mut crc32c)),
-				Some(ChecksumValue::Crc32c(x)),
-			) => {
-				crc32c.write(&x);
-			}
-			(Some(MultipartExtraChecksummer::Sha1(ref mut sha1)), Some(ChecksumValue::Sha1(x))) => {
-				sha1.update(&x);
-			}
-			(
-				Some(MultipartExtraChecksummer::Sha256(ref mut sha256)),
-				Some(ChecksumValue::Sha256(x)),
-			) => {
-				sha256.update(&x);
-			}
-			(Some(_), b) => {
-				return Err(Error::internal_error(format!(
-					"part checksum was not computed correctly, got: {:?}",
-					b
-				)))
-			}
-		}
-		Ok(())
-	}
-
-	pub(crate) fn finalize(self) -> (Md5Checksum, Option<ChecksumValue>) {
-		let md5 = self.md5.finalize()[..].try_into().unwrap();
-		let extra = match self.extra {
-			None => None,
-			Some(MultipartExtraChecksummer::Crc32(crc32)) => {
-				Some(ChecksumValue::Crc32(u32::to_be_bytes(crc32.finalize())))
-			}
-			Some(MultipartExtraChecksummer::Crc32c(crc32c)) => Some(ChecksumValue::Crc32c(
-				u32::to_be_bytes(u32::try_from(crc32c.finish()).unwrap()),
-			)),
-			Some(MultipartExtraChecksummer::Sha1(sha1)) => {
-				Some(ChecksumValue::Sha1(sha1.finalize()[..].try_into().unwrap()))
-			}
-			Some(MultipartExtraChecksummer::Sha256(sha256)) => Some(ChecksumValue::Sha256(
-				sha256.finalize()[..].try_into().unwrap(),
-			)),
-		};
-		(md5, extra)
+pub fn parse_checksum_algorithm(algo: &str) -> Result<ChecksumAlgorithm, Error> {
+	match algo {
+		"CRC32" => Ok(ChecksumAlgorithm::Crc32),
+		"CRC32C" => Ok(ChecksumAlgorithm::Crc32c),
+		"SHA1" => Ok(ChecksumAlgorithm::Sha1),
+		"SHA256" => Ok(ChecksumAlgorithm::Sha256),
+		_ => Err(Error::bad_request("invalid checksum algorithm")),
 	}
 }

-// ----
-
 /// Extract the value of the x-amz-checksum-algorithm header
-pub(crate) fn request_checksum_algorithm(
+pub fn request_checksum_algorithm(
 	headers: &HeaderMap<HeaderValue>,
 ) -> Result<Option<ChecksumAlgorithm>, Error> {
 	match headers.get(X_AMZ_CHECKSUM_ALGORITHM) {
 		None => Ok(None),
-		Some(x) if x == "CRC32" => Ok(Some(ChecksumAlgorithm::Crc32)),
-		Some(x) if x == "CRC32C" => Ok(Some(ChecksumAlgorithm::Crc32c)),
-		Some(x) if x == "SHA1" => Ok(Some(ChecksumAlgorithm::Sha1)),
-		Some(x) if x == "SHA256" => Ok(Some(ChecksumAlgorithm::Sha256)),
+		Some(x) => parse_checksum_algorithm(x.to_str()?).map(Some),
+	}
+}
+
+pub fn request_trailer_checksum_algorithm(
+	headers: &HeaderMap<HeaderValue>,
+) -> Result<Option<ChecksumAlgorithm>, Error> {
+	match headers.get(X_AMZ_TRAILER).map(|x| x.to_str()).transpose()? {
+		None => Ok(None),
+		Some(x) if x == X_AMZ_CHECKSUM_CRC32 => Ok(Some(ChecksumAlgorithm::Crc32)),
+		Some(x) if x == X_AMZ_CHECKSUM_CRC32C => Ok(Some(ChecksumAlgorithm::Crc32c)),
+		Some(x) if x == X_AMZ_CHECKSUM_SHA1 => Ok(Some(ChecksumAlgorithm::Sha1)),
+		Some(x) if x == X_AMZ_CHECKSUM_SHA256 => Ok(Some(ChecksumAlgorithm::Sha256)),
 		_ => Err(Error::bad_request("invalid checksum algorithm")),
 	}
 }

 /// Extract the value of any of the x-amz-checksum-* headers
-pub(crate) fn request_checksum_value(
+pub fn request_checksum_value(
 	headers: &HeaderMap<HeaderValue>,
 ) -> Result<Option<ChecksumValue>, Error> {
 	let mut ret = vec![];

-	if let Some(crc32_str) = headers.get(X_AMZ_CHECKSUM_CRC32) {
-		let crc32 = BASE64_STANDARD
-			.decode(&crc32_str)
-			.ok()
-			.and_then(|x| x.try_into().ok())
-			.ok_or_bad_request("invalid x-amz-checksum-crc32 header")?;
-		ret.push(ChecksumValue::Crc32(crc32))
+	if headers.contains_key(X_AMZ_CHECKSUM_CRC32) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Crc32)?);
 	}
-	if let Some(crc32c_str) = headers.get(X_AMZ_CHECKSUM_CRC32C) {
-		let crc32c = BASE64_STANDARD
-			.decode(&crc32c_str)
-			.ok()
-			.and_then(|x| x.try_into().ok())
-			.ok_or_bad_request("invalid x-amz-checksum-crc32c header")?;
-		ret.push(ChecksumValue::Crc32c(crc32c))
+	if headers.contains_key(X_AMZ_CHECKSUM_CRC32C) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Crc32c)?);
 	}
-	if let Some(sha1_str) = headers.get(X_AMZ_CHECKSUM_SHA1) {
-		let sha1 = BASE64_STANDARD
-			.decode(&sha1_str)
-			.ok()
-			.and_then(|x| x.try_into().ok())
-			.ok_or_bad_request("invalid x-amz-checksum-sha1 header")?;
-		ret.push(ChecksumValue::Sha1(sha1))
+	if headers.contains_key(X_AMZ_CHECKSUM_SHA1) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Sha1)?);
 	}
-	if let Some(sha256_str) = headers.get(X_AMZ_CHECKSUM_SHA256) {
-		let sha256 = BASE64_STANDARD
-			.decode(&sha256_str)
-			.ok()
-			.and_then(|x| x.try_into().ok())
-			.ok_or_bad_request("invalid x-amz-checksum-sha256 header")?;
-		ret.push(ChecksumValue::Sha256(sha256))
+	if headers.contains_key(X_AMZ_CHECKSUM_SHA256) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Sha256)?);
 	}

 	if ret.len() > 1 {
@ -342,48 +260,47 @@ pub(crate) fn request_checksum_value(

 /// Checks for the presence of x-amz-checksum-algorithm
 /// if so extract the corresponding x-amz-checksum-* value
-pub(crate) fn request_checksum_algorithm_value(
+pub fn extract_checksum_value(
 	headers: &HeaderMap<HeaderValue>,
-) -> Result<Option<ChecksumValue>, Error> {
-	match headers.get(X_AMZ_CHECKSUM_ALGORITHM) {
-		Some(x) if x == "CRC32" => {
+	algo: ChecksumAlgorithm,
+) -> Result<ChecksumValue, Error> {
+	match algo {
+		ChecksumAlgorithm::Crc32 => {
 			let crc32 = headers
 				.get(X_AMZ_CHECKSUM_CRC32)
 				.and_then(|x| BASE64_STANDARD.decode(&x).ok())
 				.and_then(|x| x.try_into().ok())
 				.ok_or_bad_request("invalid x-amz-checksum-crc32 header")?;
-			Ok(Some(ChecksumValue::Crc32(crc32)))
+			Ok(ChecksumValue::Crc32(crc32))
 		}
-		Some(x) if x == "CRC32C" => {
+		ChecksumAlgorithm::Crc32c => {
 			let crc32c = headers
 				.get(X_AMZ_CHECKSUM_CRC32C)
 				.and_then(|x| BASE64_STANDARD.decode(&x).ok())
 				.and_then(|x| x.try_into().ok())
 				.ok_or_bad_request("invalid x-amz-checksum-crc32c header")?;
-			Ok(Some(ChecksumValue::Crc32c(crc32c)))
+			Ok(ChecksumValue::Crc32c(crc32c))
 		}
-		Some(x) if x == "SHA1" => {
+		ChecksumAlgorithm::Sha1 => {
 			let sha1 = headers
 				.get(X_AMZ_CHECKSUM_SHA1)
 				.and_then(|x| BASE64_STANDARD.decode(&x).ok())
 				.and_then(|x| x.try_into().ok())
 				.ok_or_bad_request("invalid x-amz-checksum-sha1 header")?;
-			Ok(Some(ChecksumValue::Sha1(sha1)))
+			Ok(ChecksumValue::Sha1(sha1))
 		}
-		Some(x) if x == "SHA256" => {
+		ChecksumAlgorithm::Sha256 => {
 			let sha256 = headers
 				.get(X_AMZ_CHECKSUM_SHA256)
 				.and_then(|x| BASE64_STANDARD.decode(&x).ok())
 				.and_then(|x| x.try_into().ok())
 				.ok_or_bad_request("invalid x-amz-checksum-sha256 header")?;
-			Ok(Some(ChecksumValue::Sha256(sha256)))
+			Ok(ChecksumValue::Sha256(sha256))
 		}
-		Some(_) => Err(Error::bad_request("invalid x-amz-checksum-algorithm")),
-		None => Ok(None),
 	}
 }

-pub(crate) fn add_checksum_response_headers(
+pub fn add_checksum_response_headers(
 	checksum: &Option<ChecksumValue>,
 	mut resp: http::response::Builder,
 ) -> http::response::Builder {
--- a/src/api/common/signature/error.rs
+++ b/src/api/common/signature/error.rs
@ -18,6 +18,10 @@ pub enum Error {
 	/// The request contained an invalid UTF-8 sequence in its path or in other parameters
 	#[error(display = "Invalid UTF-8: {}", _0)]
 	InvalidUtf8Str(#[error(source)] std::str::Utf8Error),
+
+	/// The provided digest (checksum) value was invalid
+	#[error(display = "Invalid digest: {}", _0)]
+	InvalidDigest(String),
 }

 impl<T> From<T> for Error
--- a/src/api/common/signature/mod.rs
+++ b/src/api/common/signature/mod.rs
@ -2,6 +2,7 @@ use chrono::{DateTime, Utc};
 use hmac::{Hmac, Mac};
 use sha2::Sha256;

+use hyper::header::HeaderName;
 use hyper::{body::Incoming as IncomingBody, Request};

 use garage_model::garage::Garage;
@ -10,6 +11,8 @@ use garage_util::data::{sha256sum, Hash};

 use error::*;

+pub mod body;
+pub mod checksum;
 pub mod error;
 pub mod payload;
 pub mod streaming;
@ -17,36 +20,73 @@ pub mod streaming;
 pub const SHORT_DATE: &str = "%Y%m%d";
 pub const LONG_DATETIME: &str = "%Y%m%dT%H%M%SZ";

+// ---- Constants used in AWSv4 signatures ----
+
+pub const X_AMZ_ALGORITHM: HeaderName = HeaderName::from_static("x-amz-algorithm");
+pub const X_AMZ_CREDENTIAL: HeaderName = HeaderName::from_static("x-amz-credential");
+pub const X_AMZ_DATE: HeaderName = HeaderName::from_static("x-amz-date");
+pub const X_AMZ_EXPIRES: HeaderName = HeaderName::from_static("x-amz-expires");
+pub const X_AMZ_SIGNEDHEADERS: HeaderName = HeaderName::from_static("x-amz-signedheaders");
+pub const X_AMZ_SIGNATURE: HeaderName = HeaderName::from_static("x-amz-signature");
+pub const X_AMZ_CONTENT_SHA256: HeaderName = HeaderName::from_static("x-amz-content-sha256");
+pub const X_AMZ_TRAILER: HeaderName = HeaderName::from_static("x-amz-trailer");
+
+/// Result of `sha256("")`
+pub(crate) const EMPTY_STRING_HEX_DIGEST: &str =
+	"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
+
+// Signature calculation algorithm
+pub const AWS4_HMAC_SHA256: &str = "AWS4-HMAC-SHA256";
 type HmacSha256 = Hmac<Sha256>;

+// Possible values for x-amz-content-sha256, in addition to the actual sha256
+pub const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
+pub const STREAMING_UNSIGNED_PAYLOAD_TRAILER: &str = "STREAMING-UNSIGNED-PAYLOAD-TRAILER";
+pub const STREAMING_AWS4_HMAC_SHA256_PAYLOAD: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD";
+
+// Used in the computation of StringToSign
+pub const AWS4_HMAC_SHA256_PAYLOAD: &str = "AWS4-HMAC-SHA256-PAYLOAD";
+
+// ---- enums to describe stuff going on in signature calculation ----
+
+#[derive(Debug)]
+pub enum ContentSha256Header {
+	UnsignedPayload,
+	Sha256Checksum(Hash),
+	StreamingPayload { trailer: bool, signed: bool },
+}
+
+// ---- top-level functions ----
+
+pub struct VerifiedRequest {
+	pub request: Request<streaming::ReqBody>,
+	pub access_key: Key,
+	pub content_sha256_header: ContentSha256Header,
+}
+
 pub async fn verify_request(
 	garage: &Garage,
 	mut req: Request<IncomingBody>,
 	service: &'static str,
-) -> Result<(Request<streaming::ReqBody>, Key, Option<Hash>), Error> {
-	let (api_key, mut content_sha256) =
-		payload::check_payload_signature(&garage, &mut req, service).await?;
-	let api_key =
-		api_key.ok_or_else(|| Error::forbidden("Garage does not support anonymous access yet"))?;
+) -> Result<VerifiedRequest, Error> {
+	let checked_signature = payload::check_payload_signature(&garage, &mut req, service).await?;

-	let req = streaming::parse_streaming_body(
-		&api_key,
+	let request = streaming::parse_streaming_body(
 		req,
-		&mut content_sha256,
+		&checked_signature,
 		&garage.config.s3_api.s3_region,
 		service,
 	)?;

-	Ok((req, api_key, content_sha256))
-}
+	let access_key = checked_signature
+		.key
+		.ok_or_else(|| Error::forbidden("Garage does not support anonymous access yet"))?;

-pub fn verify_signed_content(expected_sha256: Hash, body: &[u8]) -> Result<(), Error> {
-	if expected_sha256 != sha256sum(body) {
-		return Err(Error::bad_request(
-			"Request content hash does not match signed hash".to_string(),
-		));
-	}
-	Ok(())
+	Ok(VerifiedRequest {
+		request,
+		access_key,
+		content_sha256_header: checked_signature.content_sha256_header,
+	})
 }

 pub fn signing_hmac(
--- a/src/api/common/signature/payload.rs
+++ b/src/api/common/signature/payload.rs
@ -13,23 +13,9 @@ use garage_util::data::Hash;
 use garage_model::garage::Garage;
 use garage_model::key_table::*;

-use super::LONG_DATETIME;
-use super::{compute_scope, signing_hmac};
+use super::*;

 use crate::encoding::uri_encode;
-use crate::signature::error::*;
-
-pub const X_AMZ_ALGORITHM: HeaderName = HeaderName::from_static("x-amz-algorithm");
-pub const X_AMZ_CREDENTIAL: HeaderName = HeaderName::from_static("x-amz-credential");
-pub const X_AMZ_DATE: HeaderName = HeaderName::from_static("x-amz-date");
-pub const X_AMZ_EXPIRES: HeaderName = HeaderName::from_static("x-amz-expires");
-pub const X_AMZ_SIGNEDHEADERS: HeaderName = HeaderName::from_static("x-amz-signedheaders");
-pub const X_AMZ_SIGNATURE: HeaderName = HeaderName::from_static("x-amz-signature");
-pub const X_AMZ_CONTENT_SH256: HeaderName = HeaderName::from_static("x-amz-content-sha256");
-
-pub const AWS4_HMAC_SHA256: &str = "AWS4-HMAC-SHA256";
-pub const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
-pub const STREAMING_AWS4_HMAC_SHA256_PAYLOAD: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD";

 pub type QueryMap = HeaderMap<QueryValue>;
 pub struct QueryValue {
@ -39,11 +25,18 @@ pub struct QueryValue {
 	value: String,
 }

+#[derive(Debug)]
+pub struct CheckedSignature {
+	pub key: Option<Key>,
+	pub content_sha256_header: ContentSha256Header,
+	pub signature_header: Option<String>,
+}
+
 pub async fn check_payload_signature(
 	garage: &Garage,
 	request: &mut Request<IncomingBody>,
 	service: &'static str,
-) -> Result<(Option<Key>, Option<Hash>), Error> {
+) -> Result<CheckedSignature, Error> {
 	let query = parse_query_map(request.uri())?;

 	if query.contains_key(&X_AMZ_ALGORITHM) {
@ -57,17 +50,46 @@ pub async fn check_payload_signature(
 		// Unsigned (anonymous) request
 		let content_sha256 = request
 			.headers()
-			.get("x-amz-content-sha256")
-			.filter(|c| c.as_bytes() != UNSIGNED_PAYLOAD.as_bytes());
-		if let Some(content_sha256) = content_sha256 {
-			let sha256 = hex::decode(content_sha256)
-				.ok()
-				.and_then(|bytes| Hash::try_from(&bytes))
-				.ok_or_bad_request("Invalid content sha256 hash")?;
-			Ok((None, Some(sha256)))
+			.get(X_AMZ_CONTENT_SHA256)
+			.map(|x| x.to_str())
+			.transpose()?;
+		Ok(CheckedSignature {
+			key: None,
+			content_sha256_header: parse_x_amz_content_sha256(content_sha256)?,
+			signature_header: None,
+		})
+	}
+}
+
+fn parse_x_amz_content_sha256(header: Option<&str>) -> Result<ContentSha256Header, Error> {
+	let header = match header {
+		Some(x) => x,
+		None => return Ok(ContentSha256Header::UnsignedPayload),
+	};
+	if header == UNSIGNED_PAYLOAD {
+		Ok(ContentSha256Header::UnsignedPayload)
+	} else if let Some(rest) = header.strip_prefix("STREAMING-") {
+		let (trailer, algo) = if let Some(rest2) = rest.strip_suffix("-TRAILER") {
+			(true, rest2)
 		} else {
-			Ok((None, None))
-		}
+			(false, rest)
+		};
+		let signed = match algo {
+			AWS4_HMAC_SHA256_PAYLOAD => true,
+			UNSIGNED_PAYLOAD => false,
+			_ => {
+				return Err(Error::bad_request(
+					"invalid or unsupported x-amz-content-sha256",
+				))
+			}
+		};
+		Ok(ContentSha256Header::StreamingPayload { trailer, signed })
+	} else {
+		let sha256 = hex::decode(header)
+			.ok()
+			.and_then(|bytes| Hash::try_from(&bytes))
+			.ok_or_bad_request("Invalid content sha256 hash")?;
+		Ok(ContentSha256Header::Sha256Checksum(sha256))
 	}
 }

@ -76,7 +98,7 @@ async fn check_standard_signature(
 	service: &'static str,
 	request: &Request<IncomingBody>,
 	query: QueryMap,
-) -> Result<(Option<Key>, Option<Hash>), Error> {
+) -> Result<CheckedSignature, Error> {
 	let authorization = Authorization::parse_header(request.headers())?;

 	// Verify that all necessary request headers are included in signed_headers
@ -108,18 +130,13 @@ async fn check_standard_signature(

 	let key = verify_v4(garage, service, &authorization, string_to_sign.as_bytes()).await?;

-	let content_sha256 = if authorization.content_sha256 == UNSIGNED_PAYLOAD {
-		None
-	} else if authorization.content_sha256 == STREAMING_AWS4_HMAC_SHA256_PAYLOAD {
-		let bytes = hex::decode(authorization.signature).ok_or_bad_request("Invalid signature")?;
-		Some(Hash::try_from(&bytes).ok_or_bad_request("Invalid signature")?)
-	} else {
-		let bytes = hex::decode(authorization.content_sha256)
-			.ok_or_bad_request("Invalid content sha256 hash")?;
-		Some(Hash::try_from(&bytes).ok_or_bad_request("Invalid content sha256 hash")?)
-	};
+	let content_sha256_header = parse_x_amz_content_sha256(Some(&authorization.content_sha256))?;

-	Ok((Some(key), content_sha256))
+	Ok(CheckedSignature {
+		key: Some(key),
+		content_sha256_header,
+		signature_header: Some(authorization.signature),
+	})
 }

 async fn check_presigned_signature(
@ -127,7 +144,7 @@ async fn check_presigned_signature(
 	service: &'static str,
 	request: &mut Request<IncomingBody>,
 	mut query: QueryMap,
-) -> Result<(Option<Key>, Option<Hash>), Error> {
+) -> Result<CheckedSignature, Error> {
 	let algorithm = query.get(&X_AMZ_ALGORITHM).unwrap();
 	let authorization = Authorization::parse_presigned(&algorithm.value, &query)?;

@ -193,7 +210,11 @@ async fn check_presigned_signature(

 	// Presigned URLs always use UNSIGNED-PAYLOAD,
 	// so there is no sha256 hash to return.
-	Ok((Some(key), None))
+	Ok(CheckedSignature {
+		key: Some(key),
+		content_sha256_header: ContentSha256Header::UnsignedPayload,
+		signature_header: Some(authorization.signature),
+	})
 }

 pub fn parse_query_map(uri: &http::uri::Uri) -> Result<QueryMap, Error> {
@ -442,7 +463,7 @@ impl Authorization {
 			.to_string();

 		let content_sha256 = headers
-			.get(X_AMZ_CONTENT_SH256)
+			.get(X_AMZ_CONTENT_SHA256)
 			.ok_or_bad_request("Missing X-Amz-Content-Sha256 field")?;

 		let date = headers
--- a/src/api/common/signature/streaming.rs
+++ b/src/api/common/signature/streaming.rs
@ -1,84 +1,157 @@
 use std::pin::Pin;
+use std::sync::Mutex;

 use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
 use futures::prelude::*;
 use futures::task;
-use garage_model::key_table::Key;
 use hmac::Mac;
-use http_body_util::StreamBody;
-use hyper::body::{Bytes, Incoming as IncomingBody};
+use http::header::{HeaderMap, HeaderValue, CONTENT_ENCODING};
+use hyper::body::{Bytes, Frame, Incoming as IncomingBody};
 use hyper::Request;

 use garage_util::data::Hash;

-use super::{compute_scope, sha256sum, HmacSha256, LONG_DATETIME};
+use super::*;

-use crate::helpers::*;
-use crate::signature::error::*;
-use crate::signature::payload::{
-	STREAMING_AWS4_HMAC_SHA256_PAYLOAD, X_AMZ_CONTENT_SH256, X_AMZ_DATE,
-};
+use crate::helpers::body_stream;
+use crate::signature::checksum::*;
+use crate::signature::payload::CheckedSignature;

-pub const AWS4_HMAC_SHA256_PAYLOAD: &str = "AWS4-HMAC-SHA256-PAYLOAD";
-
-pub type ReqBody = BoxBody<Error>;
+pub use crate::signature::body::ReqBody;

 pub fn parse_streaming_body(
-	api_key: &Key,
-	req: Request<IncomingBody>,
-	content_sha256: &mut Option<Hash>,
+	mut req: Request<IncomingBody>,
+	checked_signature: &CheckedSignature,
 	region: &str,
 	service: &str,
 ) -> Result<Request<ReqBody>, Error> {
-	match req.headers().get(X_AMZ_CONTENT_SH256) {
-		Some(header) if header == STREAMING_AWS4_HMAC_SHA256_PAYLOAD => {
-			let signature = content_sha256
-				.take()
-				.ok_or_bad_request("No signature provided")?;
+	debug!(
+		"Content signature mode: {:?}",
+		checked_signature.content_sha256_header
+	);

-			let secret_key = &api_key
-				.state
-				.as_option()
-				.ok_or_internal_error("Deleted key state")?
-				.secret_key;
+	match checked_signature.content_sha256_header {
+		ContentSha256Header::StreamingPayload { signed, trailer } => {
+			// Sanity checks
+			if !signed && !trailer {
+				return Err(Error::bad_request(
+					"STREAMING-UNSIGNED-PAYLOAD without trailer is not a valid combination",
+				));
+			}

-			let date = req
-				.headers()
-				.get(X_AMZ_DATE)
-				.ok_or_bad_request("Missing X-Amz-Date field")?
-				.to_str()?;
-			let date: NaiveDateTime = NaiveDateTime::parse_from_str(date, LONG_DATETIME)
-				.ok_or_bad_request("Invalid date")?;
-			let date: DateTime<Utc> = Utc.from_utc_datetime(&date);
+			// Remove the aws-chunked component in the content-encoding: header
+			// Note: this header is not properly sent by minio client, so don't fail
+			// if it is absent from the request.
+			if let Some(content_encoding) = req.headers_mut().remove(CONTENT_ENCODING) {
+				if let Some(rest) = content_encoding.as_bytes().strip_prefix(b"aws-chunked,") {
+					req.headers_mut()
+						.insert(CONTENT_ENCODING, HeaderValue::from_bytes(rest).unwrap());
+				} else if content_encoding != "aws-chunked" {
+					return Err(Error::bad_request(
+						"content-encoding does not contain aws-chunked for STREAMING-*-PAYLOAD",
+					));
+				}
+			}

-			let scope = compute_scope(&date, region, service);
-			let signing_hmac = crate::signature::signing_hmac(&date, secret_key, region, service)
-				.ok_or_internal_error("Unable to build signing HMAC")?;
+			// If trailer header is announced, add the calculation of the requested checksum
+			let mut checksummer = Checksummer::init(&Default::default(), false);
+			let trailer_algorithm = if trailer {
+				let algo = Some(
+					request_trailer_checksum_algorithm(req.headers())?
+						.ok_or_bad_request("Missing x-amz-trailer header")?,
+				);
+				checksummer = checksummer.add(algo);
+				algo
+			} else {
+				None
+			};
+
+			// For signed variants, determine signing parameters
+			let sign_params = if signed {
+				let signature = checked_signature
+					.signature_header
+					.clone()
+					.ok_or_bad_request("No signature provided")?;
+				let signature = hex::decode(signature)
+					.ok()
+					.and_then(|bytes| Hash::try_from(&bytes))
+					.ok_or_bad_request("Invalid signature")?;
+
+				let secret_key = checked_signature
+					.key
+					.as_ref()
+					.ok_or_bad_request("Cannot sign streaming payload without signing key")?
+					.state
+					.as_option()
+					.ok_or_internal_error("Deleted key state")?
+					.secret_key
+					.to_string();
+
+				let date = req
+					.headers()
+					.get(X_AMZ_DATE)
+					.ok_or_bad_request("Missing X-Amz-Date field")?
+					.to_str()?;
+				let date: NaiveDateTime = NaiveDateTime::parse_from_str(date, LONG_DATETIME)
+					.ok_or_bad_request("Invalid date")?;
+				let date: DateTime<Utc> = Utc.from_utc_datetime(&date);
+
+				let scope = compute_scope(&date, region, service);
+				let signing_hmac =
+					crate::signature::signing_hmac(&date, &secret_key, region, service)
+						.ok_or_internal_error("Unable to build signing HMAC")?;
+
+				Some(SignParams {
+					datetime: date,
+					scope,
+					signing_hmac,
+					previous_signature: signature,
+				})
+			} else {
+				None
+			};

 			Ok(req.map(move |body| {
 				let stream = body_stream::<_, Error>(body);
+
 				let signed_payload_stream =
-					SignedPayloadStream::new(stream, signing_hmac, date, &scope, signature)
-						.map(|x| x.map(hyper::body::Frame::data))
-						.map_err(Error::from);
-				ReqBody::new(StreamBody::new(signed_payload_stream))
+					StreamingPayloadStream::new(stream, sign_params, trailer).map_err(Error::from);
+				ReqBody {
+					stream: Mutex::new(signed_payload_stream.boxed()),
+					checksummer,
+					expected_checksums: Default::default(),
+					trailer_algorithm,
+				}
 			}))
 		}
-		_ => Ok(req.map(|body| ReqBody::new(http_body_util::BodyExt::map_err(body, Error::from)))),
+		_ => Ok(req.map(|body| {
+			let expected_checksums = ExpectedChecksums {
+				sha256: match &checked_signature.content_sha256_header {
+					ContentSha256Header::Sha256Checksum(sha256) => Some(*sha256),
+					_ => None,
+				},
+				..Default::default()
+			};
+			let checksummer = Checksummer::init(&expected_checksums, false);
+
+			let stream = http_body_util::BodyStream::new(body).map_err(Error::from);
+			ReqBody {
+				stream: Mutex::new(stream.boxed()),
+				checksummer,
+				expected_checksums,
+				trailer_algorithm: None,
+			}
+		})),
 	}
 }

-/// Result of `sha256("")`
-const EMPTY_STRING_HEX_DIGEST: &str =
-	"e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
-
 fn compute_streaming_payload_signature(
 	signing_hmac: &HmacSha256,
 	date: DateTime<Utc>,
 	scope: &str,
 	previous_signature: Hash,
 	content_sha256: Hash,
-) -> Result<Hash, Error> {
+) -> Result<Hash, StreamingPayloadError> {
 	let string_to_sign = [
 		AWS4_HMAC_SHA256_PAYLOAD,
 		&date.format(LONG_DATETIME).to_string(),
@ -92,12 +165,49 @@ fn compute_streaming_payload_signature(
 	let mut hmac = signing_hmac.clone();
 	hmac.update(string_to_sign.as_bytes());

-	Ok(Hash::try_from(&hmac.finalize().into_bytes()).ok_or_internal_error("Invalid signature")?)
+	Hash::try_from(&hmac.finalize().into_bytes())
+		.ok_or_else(|| StreamingPayloadError::Message("Could not build signature".into()))
+}
+
+fn compute_streaming_trailer_signature(
+	signing_hmac: &HmacSha256,
+	date: DateTime<Utc>,
+	scope: &str,
+	previous_signature: Hash,
+	trailer_sha256: Hash,
+) -> Result<Hash, StreamingPayloadError> {
+	let string_to_sign = [
+		AWS4_HMAC_SHA256_PAYLOAD,
+		&date.format(LONG_DATETIME).to_string(),
+		scope,
+		&hex::encode(previous_signature),
+		&hex::encode(trailer_sha256),
+	]
+	.join("\n");
+
+	let mut hmac = signing_hmac.clone();
+	hmac.update(string_to_sign.as_bytes());
+
+	Hash::try_from(&hmac.finalize().into_bytes())
+		.ok_or_else(|| StreamingPayloadError::Message("Could not build signature".into()))
 }

 mod payload {
+	use http::{HeaderName, HeaderValue};
+
 	use garage_util::data::Hash;

+	use nom::bytes::streaming::{tag, take_while};
+	use nom::character::streaming::hex_digit1;
+	use nom::combinator::{map_res, opt};
+	use nom::number::streaming::hex_u32;
+
+	macro_rules! try_parse {
+		($expr:expr) => {
+			$expr.map_err(|e| e.map(Error::Parser))?
+		};
+	}
+
 	pub enum Error<I> {
 		Parser(nom::error::Error<I>),
 		BadSignature,
@ -113,24 +223,13 @@ mod payload {
 	}

 	#[derive(Debug, Clone)]
-	pub struct Header {
+	pub struct ChunkHeader {
 		pub size: usize,
-		pub signature: Hash,
+		pub signature: Option<Hash>,
 	}

-	impl Header {
-		pub fn parse(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
-			use nom::bytes::streaming::tag;
-			use nom::character::streaming::hex_digit1;
-			use nom::combinator::map_res;
-			use nom::number::streaming::hex_u32;
-
-			macro_rules! try_parse {
-				($expr:expr) => {
-					$expr.map_err(|e| e.map(Error::Parser))?
-				};
-			}
-
+	impl ChunkHeader {
+		pub fn parse_signed(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
 			let (input, size) = try_parse!(hex_u32(input));
 			let (input, _) = try_parse!(tag(";")(input));

@ -140,96 +239,172 @@ mod payload {

 			let (input, _) = try_parse!(tag("\r\n")(input));

-			let header = Header {
+			let header = ChunkHeader {
 				size: size as usize,
-				signature,
+				signature: Some(signature),
+			};
+
+			Ok((input, header))
+		}
+
+		pub fn parse_unsigned(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			let (input, size) = try_parse!(hex_u32(input));
+			let (input, _) = try_parse!(tag("\r\n")(input));
+
+			let header = ChunkHeader {
+				size: size as usize,
+				signature: None,
 			};

 			Ok((input, header))
 		}
 	}
+
+	#[derive(Debug, Clone)]
+	pub struct TrailerChunk {
+		pub header_name: HeaderName,
+		pub header_value: HeaderValue,
+		pub signature: Option<Hash>,
+	}
+
+	impl TrailerChunk {
+		fn parse_content(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			let (input, header_name) = try_parse!(map_res(
+				take_while(|c: u8| c.is_ascii_alphanumeric() || c == b'-'),
+				HeaderName::from_bytes
+			)(input));
+			let (input, _) = try_parse!(tag(b":")(input));
+			let (input, header_value) = try_parse!(map_res(
+				take_while(|c: u8| c.is_ascii_alphanumeric() || b"+/=".contains(&c)),
+				HeaderValue::from_bytes
+			)(input));
+
+			// Possible '\n' after the header value, depends on clients
+			// https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html
+			let (input, _) = try_parse!(opt(tag(b"\n"))(input));
+
+			let (input, _) = try_parse!(tag(b"\r\n")(input));
+
+			Ok((
+				input,
+				TrailerChunk {
+					header_name,
+					header_value,
+					signature: None,
+				},
+			))
+		}
+		pub fn parse_signed(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			let (input, trailer) = Self::parse_content(input)?;
+
+			let (input, _) = try_parse!(tag(b"x-amz-trailer-signature:")(input));
+			let (input, data) = try_parse!(map_res(hex_digit1, hex::decode)(input));
+			let signature = Hash::try_from(&data).ok_or(nom::Err::Failure(Error::BadSignature))?;
+			let (input, _) = try_parse!(tag(b"\r\n")(input));
+
+			Ok((
+				input,
+				TrailerChunk {
+					signature: Some(signature),
+					..trailer
+				},
+			))
+		}
+		pub fn parse_unsigned(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			let (input, trailer) = Self::parse_content(input)?;
+			let (input, _) = try_parse!(tag(b"\r\n")(input));
+
+			Ok((input, trailer))
+		}
+	}
 }

 #[derive(Debug)]
-pub enum SignedPayloadStreamError {
+pub enum StreamingPayloadError {
 	Stream(Error),
 	InvalidSignature,
 	Message(String),
 }

-impl SignedPayloadStreamError {
+impl StreamingPayloadError {
 	fn message(msg: &str) -> Self {
-		SignedPayloadStreamError::Message(msg.into())
+		StreamingPayloadError::Message(msg.into())
 	}
 }

-impl From<SignedPayloadStreamError> for Error {
-	fn from(err: SignedPayloadStreamError) -> Self {
+impl From<StreamingPayloadError> for Error {
+	fn from(err: StreamingPayloadError) -> Self {
 		match err {
-			SignedPayloadStreamError::Stream(e) => e,
-			SignedPayloadStreamError::InvalidSignature => {
+			StreamingPayloadError::Stream(e) => e,
+			StreamingPayloadError::InvalidSignature => {
 				Error::bad_request("Invalid payload signature")
 			}
-			SignedPayloadStreamError::Message(e) => {
+			StreamingPayloadError::Message(e) => {
 				Error::bad_request(format!("Chunk format error: {}", e))
 			}
 		}
 	}
 }

-impl<I> From<payload::Error<I>> for SignedPayloadStreamError {
+impl<I> From<payload::Error<I>> for StreamingPayloadError {
 	fn from(err: payload::Error<I>) -> Self {
 		Self::message(err.description())
 	}
 }

-impl<I> From<nom::error::Error<I>> for SignedPayloadStreamError {
+impl<I> From<nom::error::Error<I>> for StreamingPayloadError {
 	fn from(err: nom::error::Error<I>) -> Self {
 		Self::message(err.code.description())
 	}
 }

-struct SignedPayload {
-	header: payload::Header,
-	data: Bytes,
+enum StreamingPayloadChunk {
+	Chunk {
+		header: payload::ChunkHeader,
+		data: Bytes,
+	},
+	Trailer(payload::TrailerChunk),
 }

-#[pin_project::pin_project]
-pub struct SignedPayloadStream<S>
-where
-	S: Stream<Item = Result<Bytes, Error>>,
-{
-	#[pin]
-	stream: S,
-	buf: bytes::BytesMut,
+struct SignParams {
 	datetime: DateTime<Utc>,
 	scope: String,
 	signing_hmac: HmacSha256,
 	previous_signature: Hash,
 }

-impl<S> SignedPayloadStream<S>
+#[pin_project::pin_project]
+pub struct StreamingPayloadStream<S>
 where
 	S: Stream<Item = Result<Bytes, Error>>,
 {
-	pub fn new(
-		stream: S,
-		signing_hmac: HmacSha256,
-		datetime: DateTime<Utc>,
-		scope: &str,
-		seed_signature: Hash,
-	) -> Self {
+	#[pin]
+	stream: S,
+	buf: bytes::BytesMut,
+	signing: Option<SignParams>,
+	has_trailer: bool,
+	done: bool,
+}
+
+impl<S> StreamingPayloadStream<S>
+where
+	S: Stream<Item = Result<Bytes, Error>>,
+{
+	fn new(stream: S, signing: Option<SignParams>, has_trailer: bool) -> Self {
 		Self {
 			stream,
 			buf: bytes::BytesMut::new(),
-			datetime,
-			scope: scope.into(),
-			signing_hmac,
-			previous_signature: seed_signature,
+			signing,
+			has_trailer,
+			done: false,
 		}
 	}

-	fn parse_next(input: &[u8]) -> nom::IResult<&[u8], SignedPayload, SignedPayloadStreamError> {
+	fn parse_next(
+		input: &[u8],
+		is_signed: bool,
+		has_trailer: bool,
+	) -> nom::IResult<&[u8], StreamingPayloadChunk, StreamingPayloadError> {
 		use nom::bytes::streaming::{tag, take};

 		macro_rules! try_parse {
@ -238,17 +413,30 @@ where
 			};
 		}

-		let (input, header) = try_parse!(payload::Header::parse(input));
+		let (input, header) = if is_signed {
+			try_parse!(payload::ChunkHeader::parse_signed(input))
+		} else {
+			try_parse!(payload::ChunkHeader::parse_unsigned(input))
+		};

 		// 0-sized chunk is the last
 		if header.size == 0 {
-			return Ok((
-				input,
-				SignedPayload {
-					header,
-					data: Bytes::new(),
-				},
-			));
+			if has_trailer {
+				let (input, trailer) = if is_signed {
+					try_parse!(payload::TrailerChunk::parse_signed(input))
+				} else {
+					try_parse!(payload::TrailerChunk::parse_unsigned(input))
+				};
+				return Ok((input, StreamingPayloadChunk::Trailer(trailer)));
+			} else {
+				return Ok((
+					input,
+					StreamingPayloadChunk::Chunk {
+						header,
+						data: Bytes::new(),
+					},
+				));
+			}
 		}

 		let (input, data) = try_parse!(take::<_, _, nom::error::Error<_>>(header.size)(input));
@ -256,15 +444,15 @@ where

 		let data = Bytes::from(data.to_vec());

-		Ok((input, SignedPayload { header, data }))
+		Ok((input, StreamingPayloadChunk::Chunk { header, data }))
 	}
 }

-impl<S> Stream for SignedPayloadStream<S>
+impl<S> Stream for StreamingPayloadStream<S>
 where
 	S: Stream<Item = Result<Bytes, Error>> + Unpin,
 {
-	type Item = Result<Bytes, SignedPayloadStreamError>;
+	type Item = Result<Frame<Bytes>, StreamingPayloadError>;

 	fn poll_next(
 		self: Pin<&mut Self>,
@ -274,56 +462,105 @@ where

 		let mut this = self.project();

+		if *this.done {
+			return Poll::Ready(None);
+		}
+
 		loop {
-			let (input, payload) = match Self::parse_next(this.buf) {
-				Ok(res) => res,
-				Err(nom::Err::Incomplete(_)) => {
-					match futures::ready!(this.stream.as_mut().poll_next(cx)) {
-						Some(Ok(bytes)) => {
-							this.buf.extend(bytes);
-							continue;
-						}
-						Some(Err(e)) => {
-							return Poll::Ready(Some(Err(SignedPayloadStreamError::Stream(e))))
-						}
-						None => {
-							return Poll::Ready(Some(Err(SignedPayloadStreamError::message(
-								"Unexpected EOF",
-							))));
+			let (input, payload) =
+				match Self::parse_next(this.buf, this.signing.is_some(), *this.has_trailer) {
+					Ok(res) => res,
+					Err(nom::Err::Incomplete(_)) => {
+						match futures::ready!(this.stream.as_mut().poll_next(cx)) {
+							Some(Ok(bytes)) => {
+								this.buf.extend(bytes);
+								continue;
+							}
+							Some(Err(e)) => {
+								return Poll::Ready(Some(Err(StreamingPayloadError::Stream(e))))
+							}
+							None => {
+								return Poll::Ready(Some(Err(StreamingPayloadError::message(
+									"Unexpected EOF",
+								))));
+							}
 						}
 					}
+					Err(nom::Err::Error(e)) | Err(nom::Err::Failure(e)) => {
+						return Poll::Ready(Some(Err(e)))
+					}
+				};
+
+			match payload {
+				StreamingPayloadChunk::Chunk { data, header } => {
+					if let Some(signing) = this.signing.as_mut() {
+						let data_sha256sum = sha256sum(&data);
+
+						let expected_signature = compute_streaming_payload_signature(
+							&signing.signing_hmac,
+							signing.datetime,
+							&signing.scope,
+							signing.previous_signature,
+							data_sha256sum,
+						)?;
+
+						if header.signature.unwrap() != expected_signature {
+							return Poll::Ready(Some(Err(StreamingPayloadError::InvalidSignature)));
+						}
+
+						signing.previous_signature = header.signature.unwrap();
+					}
+
+					*this.buf = input.into();
+
+					// 0-sized chunk is the last
+					if data.is_empty() {
+						// if there was a trailer, it would have been returned by the parser
+						assert!(!*this.has_trailer);
+						*this.done = true;
+						return Poll::Ready(None);
+					}
+
+					return Poll::Ready(Some(Ok(Frame::data(data))));
 				}
-				Err(nom::Err::Error(e)) | Err(nom::Err::Failure(e)) => {
-					return Poll::Ready(Some(Err(e)))
+				StreamingPayloadChunk::Trailer(trailer) => {
+					trace!(
+						"In StreamingPayloadStream::poll_next: got trailer {:?}",
+						trailer
+					);
+
+					if let Some(signing) = this.signing.as_mut() {
+						let data = [
+							trailer.header_name.as_ref(),
+							&b":"[..],
+							trailer.header_value.as_ref(),
+							&b"\n"[..],
+						]
+						.concat();
+						let trailer_sha256sum = sha256sum(&data);
+
+						let expected_signature = compute_streaming_trailer_signature(
+							&signing.signing_hmac,
+							signing.datetime,
+							&signing.scope,
+							signing.previous_signature,
+							trailer_sha256sum,
+						)?;
+
+						if trailer.signature.unwrap() != expected_signature {
+							return Poll::Ready(Some(Err(StreamingPayloadError::InvalidSignature)));
+						}
+					}
+
+					*this.buf = input.into();
+					*this.done = true;
+
+					let mut trailers_map = HeaderMap::new();
+					trailers_map.insert(trailer.header_name, trailer.header_value);
+
+					return Poll::Ready(Some(Ok(Frame::trailers(trailers_map))));
 				}
-			};
-
-			// 0-sized chunk is the last
-			if payload.data.is_empty() {
-				return Poll::Ready(None);
 			}
-
-			let data_sha256sum = sha256sum(&payload.data);
-
-			let expected_signature = compute_streaming_payload_signature(
-				this.signing_hmac,
-				*this.datetime,
-				this.scope,
-				*this.previous_signature,
-				data_sha256sum,
-			)
-			.map_err(|e| {
-				SignedPayloadStreamError::Message(format!("Could not build signature: {}", e))
-			})?;
-
-			if payload.header.signature != expected_signature {
-				return Poll::Ready(Some(Err(SignedPayloadStreamError::InvalidSignature)));
-			}
-
-			*this.buf = input.into();
-			*this.previous_signature = payload.header.signature;
-
-			return Poll::Ready(Some(Ok(payload.data)));
 		}
 	}

@ -336,7 +573,7 @@ where
 mod tests {
 	use futures::prelude::*;

-	use super::{SignedPayloadStream, SignedPayloadStreamError};
+	use super::{SignParams, StreamingPayloadError, StreamingPayloadStream};

 	#[tokio::test]
 	async fn test_interrupted_signed_payload_stream() {
@ -358,12 +595,20 @@ mod tests {

 		let seed_signature = Hash::default();

-		let mut stream =
-			SignedPayloadStream::new(body, signing_hmac, datetime, &scope, seed_signature);
+		let mut stream = StreamingPayloadStream::new(
+			body,
+			Some(SignParams {
+				signing_hmac,
+				datetime,
+				scope,
+				previous_signature: seed_signature,
+			}),
+			false,
+		);

 		assert!(stream.try_next().await.is_err());
 		match stream.try_next().await {
-			Err(SignedPayloadStreamError::Message(msg)) if msg == "Unexpected EOF" => {}
+			Err(StreamingPayloadError::Message(msg)) if msg == "Unexpected EOF" => {}
 			item => panic!(
 				"Unexpected result, expected early EOF error, got {:?}",
 				item
--- a/src/api/k2v/api_server.rs
+++ b/src/api/k2v/api_server.rs
@ -81,7 +81,9 @@ impl ApiHandler for K2VApiServer {
 			return Ok(options_res.map(|_empty_body: EmptyBody| empty_body()));
 		}

-		let (req, api_key, _content_sha256) = verify_request(&garage, req, "k2v").await?;
+		let verified_request = verify_request(&garage, req, "k2v").await?;
+		let req = verified_request.request;
+		let api_key = verified_request.access_key;

 		let bucket_id = garage
 			.bucket_helper()
--- a/src/api/k2v/batch.rs
+++ b/src/api/k2v/batch.rs
@ -20,7 +20,7 @@ pub async fn handle_insert_batch(
 	let ReqCtx {
 		garage, bucket_id, ..
 	} = &ctx;
-	let items = parse_json_body::<Vec<InsertBatchItem>, _, Error>(req).await?;
+	let items = req.into_body().json::<Vec<InsertBatchItem>>().await?;

 	let mut items2 = vec![];
 	for it in items {
@ -47,7 +47,7 @@ pub async fn handle_read_batch(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
 ) -> Result<Response<ResBody>, Error> {
-	let queries = parse_json_body::<Vec<ReadBatchQuery>, _, Error>(req).await?;
+	let queries = req.into_body().json::<Vec<ReadBatchQuery>>().await?;

 	let resp_results = futures::future::join_all(
 		queries
@ -141,7 +141,7 @@ pub async fn handle_delete_batch(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
 ) -> Result<Response<ResBody>, Error> {
-	let queries = parse_json_body::<Vec<DeleteBatchQuery>, _, Error>(req).await?;
+	let queries = req.into_body().json::<Vec<DeleteBatchQuery>>().await?;

 	let resp_results = futures::future::join_all(
 		queries
@ -262,7 +262,7 @@ pub(crate) async fn handle_poll_range(
 	} = ctx;
 	use garage_model::k2v::sub::PollRange;

-	let query = parse_json_body::<PollRangeQuery, _, Error>(req).await?;
+	let query = req.into_body().json::<PollRangeQuery>().await?;

 	let timeout_msec = query.timeout.unwrap_or(300).clamp(1, 600) * 1000;

--- a/src/api/k2v/error.rs
+++ b/src/api/k2v/error.rs
@ -23,6 +23,10 @@ pub enum Error {
 	#[error(display = "Authorization header malformed, unexpected scope: {}", _0)]
 	AuthorizationHeaderMalformed(String),

+	/// The provided digest (checksum) value was invalid
+	#[error(display = "Invalid digest: {}", _0)]
+	InvalidDigest(String),
+
 	/// The object requested don't exists
 	#[error(display = "Key not found")]
 	NoSuchKey,
@ -54,6 +58,7 @@ impl From<SignatureError> for Error {
 				Self::AuthorizationHeaderMalformed(c)
 			}
 			SignatureError::InvalidUtf8Str(i) => Self::InvalidUtf8Str(i),
+			SignatureError::InvalidDigest(d) => Self::InvalidDigest(d),
 		}
 	}
 }
@ -71,6 +76,7 @@ impl Error {
 			Error::InvalidBase64(_) => "InvalidBase64",
 			Error::InvalidUtf8Str(_) => "InvalidUtf8String",
 			Error::InvalidCausalityToken => "CausalityToken",
+			Error::InvalidDigest(_) => "InvalidDigest",
 		}
 	}
 }
@ -85,6 +91,7 @@ impl ApiError for Error {
 			Error::AuthorizationHeaderMalformed(_)
 			| Error::InvalidBase64(_)
 			| Error::InvalidUtf8Str(_)
+			| Error::InvalidDigest(_)
 			| Error::InvalidCausalityToken => StatusCode::BAD_REQUEST,
 		}
 	}
--- a/src/api/k2v/item.rs
+++ b/src/api/k2v/item.rs
@ -144,9 +144,7 @@ pub async fn handle_insert_item(
 		.map(parse_causality_token)
 		.transpose()?;

-	let body = http_body_util::BodyExt::collect(req.into_body())
-		.await?
-		.to_bytes();
+	let body = req.into_body().collect().await?;

 	let value = DvvsValue::Value(body.to_vec());

--- a/src/api/s3/api_server.rs
+++ b/src/api/s3/api_server.rs
@ -121,7 +121,9 @@ impl ApiHandler for S3ApiServer {
 			return Ok(options_res.map(|_empty_body: EmptyBody| empty_body()));
 		}

-		let (req, api_key, content_sha256) = verify_request(&garage, req, "s3").await?;
+		let verified_request = verify_request(&garage, req, "s3").await?;
+		let req = verified_request.request;
+		let api_key = verified_request.access_key;

 		let bucket_name = match bucket_name {
 			None => {
@ -134,14 +136,7 @@ impl ApiHandler for S3ApiServer {

 		// Special code path for CreateBucket API endpoint
 		if let Endpoint::CreateBucket {} = endpoint {
-			return handle_create_bucket(
-				&garage,
-				req,
-				content_sha256,
-				&api_key.key_id,
-				bucket_name,
-			)
-			.await;
+			return handle_create_bucket(&garage, req, &api_key.key_id, bucket_name).await;
 		}

 		let bucket_id = garage
@ -179,7 +174,7 @@ impl ApiHandler for S3ApiServer {
 		let resp = match endpoint {
 			Endpoint::HeadObject {
 				key, part_number, ..
-			} => handle_head(ctx, &req, &key, part_number).await,
+			} => handle_head(ctx, &req.map(|_| ()), &key, part_number).await,
 			Endpoint::GetObject {
 				key,
 				part_number,
@ -199,20 +194,20 @@ impl ApiHandler for S3ApiServer {
 					response_content_type,
 					response_expires,
 				};
-				handle_get(ctx, &req, &key, part_number, overrides).await
+				handle_get(ctx, &req.map(|_| ()), &key, part_number, overrides).await
 			}
 			Endpoint::UploadPart {
 				key,
 				part_number,
 				upload_id,
-			} => handle_put_part(ctx, req, &key, part_number, &upload_id, content_sha256).await,
+			} => handle_put_part(ctx, req, &key, part_number, &upload_id).await,
 			Endpoint::CopyObject { key } => handle_copy(ctx, &req, &key).await,
 			Endpoint::UploadPartCopy {
 				key,
 				part_number,
 				upload_id,
 			} => handle_upload_part_copy(ctx, &req, &key, part_number, &upload_id).await,
-			Endpoint::PutObject { key } => handle_put(ctx, req, &key, content_sha256).await,
+			Endpoint::PutObject { key } => handle_put(ctx, req, &key).await,
 			Endpoint::AbortMultipartUpload { key, upload_id } => {
 				handle_abort_multipart_upload(ctx, &key, &upload_id).await
 			}
@ -221,7 +216,7 @@ impl ApiHandler for S3ApiServer {
 				handle_create_multipart_upload(ctx, &req, &key).await
 			}
 			Endpoint::CompleteMultipartUpload { key, upload_id } => {
-				handle_complete_multipart_upload(ctx, req, &key, &upload_id, content_sha256).await
+				handle_complete_multipart_upload(ctx, req, &key, &upload_id).await
 			}
 			Endpoint::CreateBucket {} => unreachable!(),
 			Endpoint::HeadBucket {} => {
@ -324,17 +319,15 @@ impl ApiHandler for S3ApiServer {
 				};
 				handle_list_parts(ctx, req, &query).await
 			}
-			Endpoint::DeleteObjects {} => handle_delete_objects(ctx, req, content_sha256).await,
+			Endpoint::DeleteObjects {} => handle_delete_objects(ctx, req).await,
 			Endpoint::GetBucketWebsite {} => handle_get_website(ctx).await,
-			Endpoint::PutBucketWebsite {} => handle_put_website(ctx, req, content_sha256).await,
+			Endpoint::PutBucketWebsite {} => handle_put_website(ctx, req).await,
 			Endpoint::DeleteBucketWebsite {} => handle_delete_website(ctx).await,
 			Endpoint::GetBucketCors {} => handle_get_cors(ctx).await,
-			Endpoint::PutBucketCors {} => handle_put_cors(ctx, req, content_sha256).await,
+			Endpoint::PutBucketCors {} => handle_put_cors(ctx, req).await,
 			Endpoint::DeleteBucketCors {} => handle_delete_cors(ctx).await,
 			Endpoint::GetBucketLifecycleConfiguration {} => handle_get_lifecycle(ctx).await,
-			Endpoint::PutBucketLifecycleConfiguration {} => {
-				handle_put_lifecycle(ctx, req, content_sha256).await
-			}
+			Endpoint::PutBucketLifecycleConfiguration {} => handle_put_lifecycle(ctx, req).await,
 			Endpoint::DeleteBucketLifecycle {} => handle_delete_lifecycle(ctx).await,
 			endpoint => Err(Error::NotImplemented(endpoint.name().to_owned())),
 		};
--- a/src/api/s3/bucket.rs
+++ b/src/api/s3/bucket.rs
@ -1,6 +1,5 @@
 use std::collections::HashMap;

-use http_body_util::BodyExt;
 use hyper::{Request, Response, StatusCode};

 use garage_model::bucket_alias_table::*;
@ -10,12 +9,10 @@ use garage_model::key_table::Key;
 use garage_model::permission::BucketKeyPerm;
 use garage_table::util::*;
 use garage_util::crdt::*;
-use garage_util::data::*;
 use garage_util::time::*;

 use garage_api_common::common_error::CommonError;
 use garage_api_common::helpers::*;
-use garage_api_common::signature::verify_signed_content;

 use crate::api_server::{ReqBody, ResBody};
 use crate::error::*;
@ -122,15 +119,10 @@ pub async fn handle_list_buckets(
 pub async fn handle_create_bucket(
 	garage: &Garage,
 	req: Request<ReqBody>,
-	content_sha256: Option<Hash>,
 	api_key_id: &String,
 	bucket_name: String,
 ) -> Result<Response<ResBody>, Error> {
-	let body = BodyExt::collect(req.into_body()).await?.to_bytes();
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
+	let body = req.into_body().collect().await?;

 	let cmd =
 		parse_create_bucket_xml(&body[..]).ok_or_bad_request("Invalid create bucket XML query")?;
--- a/src/api/s3/copy.rs
+++ b/src/api/s3/copy.rs
@ -21,9 +21,9 @@ use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;

 use garage_api_common::helpers::*;
+use garage_api_common::signature::checksum::*;

 use crate::api_server::{ReqBody, ResBody};
-use crate::checksum::*;
 use crate::encryption::EncryptionParams;
 use crate::error::*;
 use crate::get::full_object_byte_stream;
--- a/src/api/s3/cors.rs
+++ b/src/api/s3/cors.rs
@ -2,15 +2,11 @@ use quick_xml::de::from_reader;

 use hyper::{header::HeaderName, Method, Request, Response, StatusCode};

-use http_body_util::BodyExt;
-
 use serde::{Deserialize, Serialize};

 use garage_model::bucket_table::{Bucket, CorsRule as GarageCorsRule};
-use garage_util::data::*;

 use garage_api_common::helpers::*;
-use garage_api_common::signature::verify_signed_content;

 use crate::api_server::{ReqBody, ResBody};
 use crate::error::*;
@ -59,7 +55,6 @@ pub async fn handle_delete_cors(ctx: ReqCtx) -> Result<Response<ResBody>, Error>
 pub async fn handle_put_cors(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
 	let ReqCtx {
 		garage,
@ -68,11 +63,7 @@ pub async fn handle_put_cors(
 		..
 	} = ctx;

-	let body = BodyExt::collect(req.into_body()).await?.to_bytes();
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
+	let body = req.into_body().collect().await?;

 	let conf: CorsConfiguration = from_reader(&body as &[u8])?;
 	conf.validate()?;
--- a/src/api/s3/delete.rs
+++ b/src/api/s3/delete.rs
@ -1,4 +1,3 @@
-use http_body_util::BodyExt;
 use hyper::{Request, Response, StatusCode};

 use garage_util::data::*;
@ -6,7 +5,6 @@ use garage_util::data::*;
 use garage_model::s3::object_table::*;

 use garage_api_common::helpers::*;
-use garage_api_common::signature::verify_signed_content;

 use crate::api_server::{ReqBody, ResBody};
 use crate::error::*;
@ -68,13 +66,8 @@ pub async fn handle_delete(ctx: ReqCtx, key: &str) -> Result<Response<ResBody>,
 pub async fn handle_delete_objects(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
-	let body = BodyExt::collect(req.into_body()).await?.to_bytes();
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
+	let body = req.into_body().collect().await?;

 	let cmd_xml = roxmltree::Document::parse(std::str::from_utf8(&body)?)?;
 	let cmd = parse_delete_objects_xml(&cmd_xml).ok_or_bad_request("Invalid delete XML query")?;
--- a/src/api/s3/encryption.rs
+++ b/src/api/s3/encryption.rs
@ -29,8 +29,8 @@ use garage_model::garage::Garage;
 use garage_model::s3::object_table::{ObjectVersionEncryption, ObjectVersionMetaInner};

 use garage_api_common::common_error::*;
+use garage_api_common::signature::checksum::Md5Checksum;

-use crate::checksum::Md5Checksum;
 use crate::error::Error;

 const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM: HeaderName =
--- a/src/api/s3/error.rs
+++ b/src/api/s3/error.rs
@ -80,7 +80,7 @@ pub enum Error {
 	#[error(display = "Invalid encryption algorithm: {:?}, should be AES256", _0)]
 	InvalidEncryptionAlgorithm(String),

-	/// The client sent invalid XML data
+	/// The provided digest (checksum) value was invalid
 	#[error(display = "Invalid digest: {}", _0)]
 	InvalidDigest(String),

@ -119,6 +119,7 @@ impl From<SignatureError> for Error {
 				Self::AuthorizationHeaderMalformed(c)
 			}
 			SignatureError::InvalidUtf8Str(i) => Self::InvalidUtf8Str(i),
+			SignatureError::InvalidDigest(d) => Self::InvalidDigest(d),
 		}
 	}
 }
--- a/src/api/s3/get.rs
+++ b/src/api/s3/get.rs
@ -12,7 +12,7 @@ use http::header::{
 	CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, ETAG, EXPIRES, IF_MODIFIED_SINCE, IF_NONE_MATCH,
 	LAST_MODIFIED, RANGE,
 };
-use hyper::{body::Body, Request, Response, StatusCode};
+use hyper::{Request, Response, StatusCode};
 use tokio::sync::mpsc;

 use garage_net::stream::ByteStream;
@ -26,9 +26,9 @@ use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;

 use garage_api_common::helpers::*;
+use garage_api_common::signature::checksum::{add_checksum_response_headers, X_AMZ_CHECKSUM_MODE};

 use crate::api_server::ResBody;
-use crate::checksum::{add_checksum_response_headers, X_AMZ_CHECKSUM_MODE};
 use crate::encryption::EncryptionParams;
 use crate::error::*;

@ -118,7 +118,7 @@ fn getobject_override_headers(
 fn try_answer_cached(
 	version: &ObjectVersion,
 	version_meta: &ObjectVersionMeta,
-	req: &Request<impl Body>,
+	req: &Request<()>,
 ) -> Option<Response<ResBody>> {
 	// <trinity> It is possible, and is even usually the case, [that both If-None-Match and
 	// If-Modified-Since] are present in a request. In this situation If-None-Match takes
@ -157,7 +157,7 @@ fn try_answer_cached(
 /// Handle HEAD request
 pub async fn handle_head(
 	ctx: ReqCtx,
-	req: &Request<impl Body>,
+	req: &Request<()>,
 	key: &str,
 	part_number: Option<u64>,
 ) -> Result<Response<ResBody>, Error> {
@ -167,7 +167,7 @@ pub async fn handle_head(
 /// Handle HEAD request for website
 pub async fn handle_head_without_ctx(
 	garage: Arc<Garage>,
-	req: &Request<impl Body>,
+	req: &Request<()>,
 	bucket_id: Uuid,
 	key: &str,
 	part_number: Option<u64>,
@ -278,7 +278,7 @@ pub async fn handle_head_without_ctx(
 /// Handle GET request
 pub async fn handle_get(
 	ctx: ReqCtx,
-	req: &Request<impl Body>,
+	req: &Request<()>,
 	key: &str,
 	part_number: Option<u64>,
 	overrides: GetObjectOverrides,
@ -289,7 +289,7 @@ pub async fn handle_get(
 /// Handle GET request
 pub async fn handle_get_without_ctx(
 	garage: Arc<Garage>,
-	req: &Request<impl Body>,
+	req: &Request<()>,
 	bucket_id: Uuid,
 	key: &str,
 	part_number: Option<u64>,
@ -340,7 +340,12 @@ pub async fn handle_get_without_ctx(
 				enc,
 				&headers,
 				pn,
-				checksum_mode,
+				ChecksumMode {
+					// TODO: for multipart uploads, checksums of each part should be stored
+					// so that we can return the corresponding checksum here
+					// https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html
+					enabled: false,
+				},
 			)
 			.await
 		}
@ -354,7 +359,12 @@ pub async fn handle_get_without_ctx(
 				&headers,
 				range.start,
 				range.start + range.length,
-				checksum_mode,
+				ChecksumMode {
+					// TODO: for range queries that align with part boundaries,
+					// we should return the saved checksum of the part
+					// https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html
+					enabled: false,
+				},
 			)
 			.await
 		}
@ -577,7 +587,7 @@ async fn handle_get_part(
 }

 fn parse_range_header(
-	req: &Request<impl Body>,
+	req: &Request<()>,
 	total_size: u64,
 ) -> Result<Option<http_range::HttpRange>, Error> {
 	let range = match req.headers().get(RANGE) {
@ -618,7 +628,7 @@ struct ChecksumMode {
 	enabled: bool,
 }

-fn checksum_mode(req: &Request<impl Body>) -> ChecksumMode {
+fn checksum_mode(req: &Request<()>) -> ChecksumMode {
 	ChecksumMode {
 		enabled: req
 			.headers()
--- a/src/api/s3/lib.rs
+++ b/src/api/s3/lib.rs
@ -16,7 +16,6 @@ mod post_object;
 mod put;
 mod website;

-mod checksum;
 mod encryption;
 mod router;
 pub mod xml;
--- a/src/api/s3/lifecycle.rs
+++ b/src/api/s3/lifecycle.rs
@ -1,12 +1,10 @@
 use quick_xml::de::from_reader;

-use http_body_util::BodyExt;
 use hyper::{Request, Response, StatusCode};

 use serde::{Deserialize, Serialize};

 use garage_api_common::helpers::*;
-use garage_api_common::signature::verify_signed_content;

 use crate::api_server::{ReqBody, ResBody};
 use crate::error::*;
@ -16,7 +14,6 @@ use garage_model::bucket_table::{
 	parse_lifecycle_date, Bucket, LifecycleExpiration as GarageLifecycleExpiration,
 	LifecycleFilter as GarageLifecycleFilter, LifecycleRule as GarageLifecycleRule,
 };
-use garage_util::data::*;

 pub async fn handle_get_lifecycle(ctx: ReqCtx) -> Result<Response<ResBody>, Error> {
 	let ReqCtx { bucket_params, .. } = ctx;
@ -56,7 +53,6 @@ pub async fn handle_delete_lifecycle(ctx: ReqCtx) -> Result<Response<ResBody>, E
 pub async fn handle_put_lifecycle(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
 	let ReqCtx {
 		garage,
@ -65,11 +61,7 @@ pub async fn handle_put_lifecycle(
 		..
 	} = ctx;

-	let body = BodyExt::collect(req.into_body()).await?.to_bytes();
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
+	let body = req.into_body().collect().await?;

 	let conf: LifecycleConfiguration = from_reader(&body as &[u8])?;
 	let config = conf
--- a/src/api/s3/multipart.rs
+++ b/src/api/s3/multipart.rs
@ -1,13 +1,20 @@
 use std::collections::HashMap;
-use std::convert::TryInto;
+use std::convert::{TryFrom, TryInto};
+use std::hash::Hasher;
 use std::sync::Arc;

 use base64::prelude::*;
+use crc32c::Crc32cHasher as Crc32c;
+use crc32fast::Hasher as Crc32;
 use futures::prelude::*;
 use hyper::{Request, Response};
+use md5::{Digest, Md5};
+use sha1::Sha1;
+use sha2::Sha256;

 use garage_table::*;
 use garage_util::data::*;
+use garage_util::error::OkOrMessage;

 use garage_model::garage::Garage;
 use garage_model::s3::block_ref_table::*;
@ -16,10 +23,9 @@ use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;

 use garage_api_common::helpers::*;
-use garage_api_common::signature::verify_signed_content;
+use garage_api_common::signature::checksum::*;

 use crate::api_server::{ReqBody, ResBody};
-use crate::checksum::*;
 use crate::encryption::EncryptionParams;
 use crate::error::*;
 use crate::put::*;
@ -94,7 +100,6 @@ pub async fn handle_put_part(
 	key: &str,
 	part_number: u64,
 	upload_id: &str,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
 	let ReqCtx { garage, .. } = &ctx;

@ -105,17 +110,30 @@ pub async fn handle_put_part(
 			Some(x) => Some(x.to_str()?.to_string()),
 			None => None,
 		},
-		sha256: content_sha256,
+		sha256: None,
 		extra: request_checksum_value(req.headers())?,
 	};

-	// Read first chuck, and at the same time try to get object to see if it exists
 	let key = key.to_string();

-	let (req_head, req_body) = req.into_parts();
-	let stream = body_stream(req_body);
+	let (req_head, mut req_body) = req.into_parts();
+
+	// Before we stream the body, configure the needed checksums.
+	req_body.add_expected_checksums(expected_checksums.clone());
+	// TODO: avoid parsing encryption headers twice...
+	if !EncryptionParams::new_from_headers(&garage, &req_head.headers)?.is_encrypted() {
+		// For non-encrypted objects, we need to compute the md5sum in all cases
+		// (even if content-md5 is not set), because it is used as an etag of the
+		// part, which is in turn used in the etag computation of the whole object
+		req_body.add_md5();
+	}
+
+	let (stream, stream_checksums) = req_body.streaming_with_checksums();
+	let stream = stream.map_err(Error::from);
+
 	let mut chunker = StreamChunker::new(stream, garage.config.block_size);

+	// Read first chuck, and at the same time try to get object to see if it exists
 	let ((_, object_version, mut mpu), first_block) =
 		futures::try_join!(get_upload(&ctx, &key, &upload_id), chunker.next(),)?;

@ -172,21 +190,21 @@ pub async fn handle_put_part(
 	garage.version_table.insert(&version).await?;

 	// Copy data to version
-	let checksummer =
-		Checksummer::init(&expected_checksums, !encryption.is_encrypted()).add(checksum_algorithm);
-	let (total_size, checksums, _) = read_and_put_blocks(
+	let (total_size, _, _) = read_and_put_blocks(
 		&ctx,
 		&version,
 		encryption,
 		part_number,
 		first_block,
-		&mut chunker,
-		checksummer,
+		chunker,
+		Checksummer::new(),
 	)
 	.await?;

-	// Verify that checksums map
-	checksums.verify(&expected_checksums)?;
+	// Verify that checksums match
+	let checksums = stream_checksums
+		.await
+		.ok_or_internal_error("checksum calculation")??;

 	// Store part etag in version
 	let etag = encryption.etag_from_md5(&checksums.md5);
@ -248,7 +266,6 @@ pub async fn handle_complete_multipart_upload(
 	req: Request<ReqBody>,
 	key: &str,
 	upload_id: &str,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
 	let ReqCtx {
 		garage,
@ -260,11 +277,7 @@ pub async fn handle_complete_multipart_upload(

 	let expected_checksum = request_checksum_value(&req_head.headers)?;

-	let body = http_body_util::BodyExt::collect(req_body).await?.to_bytes();
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
+	let body = req_body.collect().await?;

 	let body_xml = roxmltree::Document::parse(std::str::from_utf8(&body)?)?;
 	let body_list_of_parts = parse_complete_multipart_upload_body(&body_xml)
@ -602,3 +615,99 @@ fn parse_complete_multipart_upload_body(

 	Some(parts)
 }
+
+// ====== checksummer ====
+
+#[derive(Default)]
+pub(crate) struct MultipartChecksummer {
+	pub md5: Md5,
+	pub extra: Option<MultipartExtraChecksummer>,
+}
+
+pub(crate) enum MultipartExtraChecksummer {
+	Crc32(Crc32),
+	Crc32c(Crc32c),
+	Sha1(Sha1),
+	Sha256(Sha256),
+}
+
+impl MultipartChecksummer {
+	pub(crate) fn init(algo: Option<ChecksumAlgorithm>) -> Self {
+		Self {
+			md5: Md5::new(),
+			extra: match algo {
+				None => None,
+				Some(ChecksumAlgorithm::Crc32) => {
+					Some(MultipartExtraChecksummer::Crc32(Crc32::new()))
+				}
+				Some(ChecksumAlgorithm::Crc32c) => {
+					Some(MultipartExtraChecksummer::Crc32c(Crc32c::default()))
+				}
+				Some(ChecksumAlgorithm::Sha1) => Some(MultipartExtraChecksummer::Sha1(Sha1::new())),
+				Some(ChecksumAlgorithm::Sha256) => {
+					Some(MultipartExtraChecksummer::Sha256(Sha256::new()))
+				}
+			},
+		}
+	}
+
+	pub(crate) fn update(
+		&mut self,
+		etag: &str,
+		checksum: Option<ChecksumValue>,
+	) -> Result<(), Error> {
+		self.md5
+			.update(&hex::decode(&etag).ok_or_message("invalid etag hex")?);
+		match (&mut self.extra, checksum) {
+			(None, _) => (),
+			(
+				Some(MultipartExtraChecksummer::Crc32(ref mut crc32)),
+				Some(ChecksumValue::Crc32(x)),
+			) => {
+				crc32.update(&x);
+			}
+			(
+				Some(MultipartExtraChecksummer::Crc32c(ref mut crc32c)),
+				Some(ChecksumValue::Crc32c(x)),
+			) => {
+				crc32c.write(&x);
+			}
+			(Some(MultipartExtraChecksummer::Sha1(ref mut sha1)), Some(ChecksumValue::Sha1(x))) => {
+				sha1.update(&x);
+			}
+			(
+				Some(MultipartExtraChecksummer::Sha256(ref mut sha256)),
+				Some(ChecksumValue::Sha256(x)),
+			) => {
+				sha256.update(&x);
+			}
+			(Some(_), b) => {
+				return Err(Error::internal_error(format!(
+					"part checksum was not computed correctly, got: {:?}",
+					b
+				)))
+			}
+		}
+		Ok(())
+	}
+
+	pub(crate) fn finalize(self) -> (Md5Checksum, Option<ChecksumValue>) {
+		let md5 = self.md5.finalize()[..].try_into().unwrap();
+		let extra = match self.extra {
+			None => None,
+			Some(MultipartExtraChecksummer::Crc32(crc32)) => {
+				Some(ChecksumValue::Crc32(u32::to_be_bytes(crc32.finalize())))
+			}
+			Some(MultipartExtraChecksummer::Crc32c(crc32c)) => Some(ChecksumValue::Crc32c(
+				u32::to_be_bytes(u32::try_from(crc32c.finish()).unwrap()),
+			)),
+			Some(MultipartExtraChecksummer::Sha1(sha1)) => {
+				Some(ChecksumValue::Sha1(sha1.finalize()[..].try_into().unwrap()))
+			}
+			Some(MultipartExtraChecksummer::Sha256(sha256)) => Some(ChecksumValue::Sha256(
+				sha256.finalize()[..].try_into().unwrap(),
+			)),
+		};
+		(md5, extra)
+	}
+}
--- a/src/api/s3/post_object.rs
+++ b/src/api/s3/post_object.rs
@ -18,10 +18,10 @@ use garage_model::s3::object_table::*;

 use garage_api_common::cors::*;
 use garage_api_common::helpers::*;
+use garage_api_common::signature::checksum::*;
 use garage_api_common::signature::payload::{verify_v4, Authorization};

 use crate::api_server::ResBody;
-use crate::checksum::*;
 use crate::encryption::EncryptionParams;
 use crate::error::*;
 use crate::put::{get_headers, save_stream, ChecksumMode};
@ -218,6 +218,7 @@ pub async fn handle_post_object(
 	// around here to make sure the rest of the machinery takes our acl into account.
 	let headers = get_headers(&params)?;

+	let checksum_algorithm = request_checksum_algorithm(&params)?;
 	let expected_checksums = ExpectedChecksums {
 		md5: params
 			.get("content-md5")
@ -225,7 +226,9 @@ pub async fn handle_post_object(
 			.transpose()?
 			.map(str::to_string),
 		sha256: None,
-		extra: request_checksum_algorithm_value(&params)?,
+		extra: checksum_algorithm
+			.map(|algo| extract_checksum_value(&params, algo))
+			.transpose()?,
 	};

 	let meta = ObjectVersionMetaInner {
--- a/src/api/s3/put.rs
+++ b/src/api/s3/put.rs
@ -31,9 +31,10 @@ use garage_model::s3::object_table::*;
 use garage_model::s3::version_table::*;

 use garage_api_common::helpers::*;
+use garage_api_common::signature::body::StreamingChecksumReceiver;
+use garage_api_common::signature::checksum::*;

 use crate::api_server::{ReqBody, ResBody};
-use crate::checksum::*;
 use crate::encryption::EncryptionParams;
 use crate::error::*;

@ -48,6 +49,10 @@ pub(crate) struct SaveStreamResult {

 pub(crate) enum ChecksumMode<'a> {
 	Verify(&'a ExpectedChecksums),
+	VerifyFrom {
+		checksummer: StreamingChecksumReceiver,
+		trailer_algo: Option<ChecksumAlgorithm>,
+	},
 	Calculate(Option<ChecksumAlgorithm>),
 }

@ -55,7 +60,6 @@ pub async fn handle_put(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
 	key: &String,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
 	// Retrieve interesting headers from request
 	let headers = get_headers(req.headers())?;
@ -66,9 +70,10 @@ pub async fn handle_put(
 			Some(x) => Some(x.to_str()?.to_string()),
 			None => None,
 		},
-		sha256: content_sha256,
+		sha256: None,
 		extra: request_checksum_value(req.headers())?,
 	};
+	let trailer_checksum_algorithm = request_trailer_checksum_algorithm(req.headers())?;

 	let meta = ObjectVersionMetaInner {
 		headers,
@ -78,7 +83,19 @@ pub async fn handle_put(
 	// Determine whether object should be encrypted, and if so the key
 	let encryption = EncryptionParams::new_from_headers(&ctx.garage, req.headers())?;

-	let stream = body_stream(req.into_body());
+	// The request body is a special ReqBody object (see garage_api_common::signature::body)
+	// which supports calculating checksums while streaming the data.
+	// Before we start streaming, we configure it to calculate all the checksums we need.
+	let mut req_body = req.into_body();
+	req_body.add_expected_checksums(expected_checksums.clone());
+	if !encryption.is_encrypted() {
+		// For non-encrypted objects, we need to compute the md5sum in all cases
+		// (even if content-md5 is not set), because it is used as the object etag
+		req_body.add_md5();
+	}
+
+	let (stream, checksummer) = req_body.streaming_with_checksums();
+	let stream = stream.map_err(Error::from);

 	let res = save_stream(
 		&ctx,
@ -86,7 +103,10 @@ pub async fn handle_put(
 		encryption,
 		stream,
 		key,
-		ChecksumMode::Verify(&expected_checksums),
+		ChecksumMode::VerifyFrom {
+			checksummer,
+			trailer_algo: trailer_checksum_algorithm,
+		},
 	)
 	.await?;

@ -122,10 +142,15 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	let version_uuid = gen_uuid();
 	let version_timestamp = next_timestamp(existing_object.as_ref());

-	let mut checksummer = match checksum_mode {
+	let mut checksummer = match &checksum_mode {
 		ChecksumMode::Verify(expected) => Checksummer::init(expected, !encryption.is_encrypted()),
 		ChecksumMode::Calculate(algo) => {
-			Checksummer::init(&Default::default(), !encryption.is_encrypted()).add(algo)
+			Checksummer::init(&Default::default(), !encryption.is_encrypted()).add(*algo)
+		}
+		ChecksumMode::VerifyFrom { .. } => {
+			// Checksums are calculated by the garage_api_common::signature module
+			// so here we can just have an empty checksummer that does nothing
+			Checksummer::new()
 		}
 	};

@ -133,7 +158,7 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	// as "inline data". We can then return immediately.
 	if first_block.len() < INLINE_THRESHOLD {
 		checksummer.update(&first_block);
-		let checksums = checksummer.finalize();
+		let mut checksums = checksummer.finalize();

 		match checksum_mode {
 			ChecksumMode::Verify(expected) => {
@ -142,6 +167,18 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 			ChecksumMode::Calculate(algo) => {
 				meta.checksum = checksums.extract(algo);
 			}
+			ChecksumMode::VerifyFrom {
+				checksummer,
+				trailer_algo,
+			} => {
+				drop(chunker);
+				checksums = checksummer
+					.await
+					.ok_or_internal_error("checksum calculation")??;
+				if let Some(algo) = trailer_algo {
+					meta.checksum = checksums.extract(Some(algo));
+				}
+			}
 		};

 		let size = first_block.len() as u64;
@ -213,13 +250,13 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	garage.version_table.insert(&version).await?;

 	// Transfer data
-	let (total_size, checksums, first_block_hash) = read_and_put_blocks(
+	let (total_size, mut checksums, first_block_hash) = read_and_put_blocks(
 		ctx,
 		&version,
 		encryption,
 		1,
 		first_block,
-		&mut chunker,
+		chunker,
 		checksummer,
 	)
 	.await?;
@ -232,6 +269,17 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 		ChecksumMode::Calculate(algo) => {
 			meta.checksum = checksums.extract(algo);
 		}
+		ChecksumMode::VerifyFrom {
+			checksummer,
+			trailer_algo,
+		} => {
+			checksums = checksummer
+				.await
+				.ok_or_internal_error("checksum calculation")??;
+			if let Some(algo) = trailer_algo {
+				meta.checksum = checksums.extract(Some(algo));
+			}
+		}
 	};

 	// Verify quotas are respsected
@ -332,7 +380,7 @@ pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> +
 	encryption: EncryptionParams,
 	part_number: u64,
 	first_block: Bytes,
-	chunker: &mut StreamChunker<S>,
+	mut chunker: StreamChunker<S>,
 	checksummer: Checksummer,
 ) -> Result<(u64, Checksums, Hash), Error> {
 	let tracer = opentelemetry::global::tracer("garage");
--- a/src/api/s3/website.rs
+++ b/src/api/s3/website.rs
@ -1,14 +1,11 @@
 use quick_xml::de::from_reader;

-use http_body_util::BodyExt;
 use hyper::{Request, Response, StatusCode};
 use serde::{Deserialize, Serialize};

 use garage_model::bucket_table::*;
-use garage_util::data::*;

 use garage_api_common::helpers::*;
-use garage_api_common::signature::verify_signed_content;

 use crate::api_server::{ReqBody, ResBody};
 use crate::error::*;
@ -61,7 +58,6 @@ pub async fn handle_delete_website(ctx: ReqCtx) -> Result<Response<ResBody>, Err
 pub async fn handle_put_website(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
-	content_sha256: Option<Hash>,
 ) -> Result<Response<ResBody>, Error> {
 	let ReqCtx {
 		garage,
@ -70,11 +66,7 @@ pub async fn handle_put_website(
 		..
 	} = ctx;

-	let body = BodyExt::collect(req.into_body()).await?.to_bytes();
-
-	if let Some(content_sha256) = content_sha256 {
-		verify_signed_content(content_sha256, &body[..])?;
-	}
+	let body = req.into_body().collect().await?;

 	let conf: WebsiteConfiguration = from_reader(&body as &[u8])?;
 	conf.validate()?;
--- a/src/garage/Cargo.toml
+++ b/src/garage/Cargo.toml
@ -71,10 +71,12 @@ hyper-util.workspace = true
 mktemp.workspace = true
 sha2.workspace = true

+
 static_init.workspace = true
 assert-json-diff.workspace = true
 serde_json.workspace = true
 base64.workspace = true
+crc32fast.workspace = true

 k2v-client.workspace = true

--- a/src/garage/tests/common/client.rs
+++ b/src/garage/tests/common/client.rs
@ -12,7 +12,7 @@ pub fn build_client(key: &Key) -> Client {
 		.endpoint_url(format!("http://127.0.0.1:{}", DEFAULT_PORT))
 		.region(super::REGION)
 		.credentials_provider(credentials)
-		.behavior_version(BehaviorVersion::v2023_11_09())
+		.behavior_version(BehaviorVersion::v2024_03_28())
 		.build();

 	Client::from_conf(config)
--- a/src/garage/tests/common/custom_requester.rs
+++ b/src/garage/tests/common/custom_requester.rs
@ -192,16 +192,13 @@ impl<'a> RequestBuilder<'a> {
 			.collect::<HeaderMap>();

 		let date = now.format(signature::LONG_DATETIME).to_string();
-		all_headers.insert(
-			signature::payload::X_AMZ_DATE,
-			HeaderValue::from_str(&date).unwrap(),
-		);
+		all_headers.insert(signature::X_AMZ_DATE, HeaderValue::from_str(&date).unwrap());
 		all_headers.insert(HOST, HeaderValue::from_str(&host).unwrap());

-		let body_sha = match self.body_signature {
+		let body_sha = match &self.body_signature {
 			BodySignature::Unsigned => "UNSIGNED-PAYLOAD".to_owned(),
 			BodySignature::Classic => hex::encode(garage_util::data::sha256sum(&self.body)),
-			BodySignature::Streaming(size) => {
+			BodySignature::Streaming { chunk_size } => {
 				all_headers.insert(
 					CONTENT_ENCODING,
 					HeaderValue::from_str("aws-chunked").unwrap(),
@ -216,18 +213,59 @@ impl<'a> RequestBuilder<'a> {
 				// code.
 				all_headers.insert(
 					CONTENT_LENGTH,
-					to_streaming_body(&self.body, size, String::new(), signer.clone(), now, "")
-						.len()
-						.to_string()
-						.try_into()
-						.unwrap(),
+					to_streaming_body(
+						&self.body,
+						*chunk_size,
+						String::new(),
+						signer.clone(),
+						now,
+						"",
+					)
+					.len()
+					.to_string()
+					.try_into()
+					.unwrap(),
 				);

 				"STREAMING-AWS4-HMAC-SHA256-PAYLOAD".to_owned()
 			}
+			BodySignature::StreamingUnsignedTrailer {
+				chunk_size,
+				trailer_algorithm,
+				trailer_value,
+			} => {
+				all_headers.insert(
+					CONTENT_ENCODING,
+					HeaderValue::from_str("aws-chunked").unwrap(),
+				);
+				all_headers.insert(
+					HeaderName::from_static("x-amz-decoded-content-length"),
+					HeaderValue::from_str(&self.body.len().to_string()).unwrap(),
+				);
+				all_headers.insert(
+					HeaderName::from_static("x-amz-trailer"),
+					HeaderValue::from_str(&trailer_algorithm).unwrap(),
+				);
+
+				all_headers.insert(
+					CONTENT_LENGTH,
+					to_streaming_unsigned_trailer_body(
+						&self.body,
+						*chunk_size,
+						&trailer_algorithm,
+						&trailer_value,
+					)
+					.len()
+					.to_string()
+					.try_into()
+					.unwrap(),
+				);
+
+				"STREAMING-UNSIGNED-PAYLOAD-TRAILER".to_owned()
+			}
 		};
 		all_headers.insert(
-			signature::payload::X_AMZ_CONTENT_SH256,
+			signature::X_AMZ_CONTENT_SHA256,
 			HeaderValue::from_str(&body_sha).unwrap(),
 		);

@ -276,10 +314,26 @@ impl<'a> RequestBuilder<'a> {
 		let mut request = Request::builder();
 		*request.headers_mut().unwrap() = all_headers;

-		let body = if let BodySignature::Streaming(size) = self.body_signature {
-			to_streaming_body(&self.body, size, signature, streaming_signer, now, &scope)
-		} else {
-			self.body.clone()
+		let body = match &self.body_signature {
+			BodySignature::Streaming { chunk_size } => to_streaming_body(
+				&self.body,
+				*chunk_size,
+				signature,
+				streaming_signer,
+				now,
+				&scope,
+			),
+			BodySignature::StreamingUnsignedTrailer {
+				chunk_size,
+				trailer_algorithm,
+				trailer_value,
+			} => to_streaming_unsigned_trailer_body(
+				&self.body,
+				*chunk_size,
+				&trailer_algorithm,
+				&trailer_value,
+			),
+			_ => self.body.clone(),
 		};
 		let request = request
 			.uri(uri)
@ -308,7 +362,14 @@ impl<'a> RequestBuilder<'a> {
 pub enum BodySignature {
 	Unsigned,
 	Classic,
-	Streaming(usize),
+	Streaming {
+		chunk_size: usize,
+	},
+	StreamingUnsignedTrailer {
+		chunk_size: usize,
+		trailer_algorithm: String,
+		trailer_value: String,
+	},
 }

 fn query_param_to_string(params: &HashMap<String, Option<String>>) -> String {
@ -363,3 +424,26 @@ fn to_streaming_body(

 	res
 }
+
+fn to_streaming_unsigned_trailer_body(
+	body: &[u8],
+	chunk_size: usize,
+	trailer_algorithm: &str,
+	trailer_value: &str,
+) -> Vec<u8> {
+	let mut res = Vec::with_capacity(body.len());
+	for chunk in body.chunks(chunk_size) {
+		let header = format!("{:x}\r\n", chunk.len());
+		res.extend_from_slice(header.as_bytes());
+		res.extend_from_slice(chunk);
+		res.extend_from_slice(b"\r\n");
+	}
+
+	res.extend_from_slice(b"0\r\n");
+	res.extend_from_slice(trailer_algorithm.as_bytes());
+	res.extend_from_slice(b":");
+	res.extend_from_slice(trailer_value.as_bytes());
+	res.extend_from_slice(b"\n\r\n\r\n");
+
+	res
+}
--- a/src/garage/tests/common/garage.rs
+++ b/src/garage/tests/common/garage.rs
@ -99,7 +99,10 @@ api_bind_addr = "127.0.0.1:{admin_port}"
 			.arg("server")
 			.stdout(stdout)
 			.stderr(stderr)
-			.env("RUST_LOG", "garage=debug,garage_api=trace")
+			.env(
+				"RUST_LOG",
+				"garage=debug,garage_api_common=trace,garage_api_s3=trace",
+			)
 			.spawn()
 			.expect("Could not start garage");

--- 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()
+		);
 	}
 }

--- 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,14 @@ 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)
@ -73,11 +84,13 @@ async fn test_putobject_streaming() {
 			//fail
 			.path("abc".to_owned())
 			.vhost_style(true)
+			.signed_headers(headers)
 			.body(BODY.to_vec())
-			.body_signature(BodySignature::Streaming(16))
+			.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());
@ -88,6 +101,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 +112,142 @@ 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);
+	}
+}
+
+#[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)
+			//.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: 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());
+
+		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);
 	}
 }

@ -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();
--- a/src/web/web_server.rs
+++ b/src/web/web_server.rs
@ -1,6 +1,6 @@
 use std::fs::{self, Permissions};
 use std::os::unix::prelude::PermissionsExt;
-use std::{convert::Infallible, sync::Arc};
+use std::sync::Arc;

 use tokio::net::{TcpListener, UnixListener};
 use tokio::sync::watch;
@ -163,6 +163,8 @@ impl WebServer {
 			metrics_tags.push(KeyValue::new("host", host_header.clone()));
 		}

+		let req = req.map(|_| ());
+
 		// The actual handler
 		let res = self
 			.serve_file(&req)
@ -218,7 +220,7 @@ impl WebServer {

 	async fn serve_file(
 		self: &Arc<Self>,
-		req: &Request<IncomingBody>,
+		req: &Request<()>,
 	) -> Result<Response<BoxBody<ApiError>>, Error> {
 		// Get http authority string (eg. [::1]:3902 or garage.tld:80)
 		let authority = req
@ -322,7 +324,7 @@ impl WebServer {
 				// Create a fake HTTP request with path = the error document
 				let req2 = Request::builder()
 					.uri(format!("http://{}/{}", host, &error_document))
-					.body(empty_body::<Infallible>())
+					.body(())
 					.unwrap();

 				match handle_get_without_ctx(