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

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

8 files changed, 991 insertions, 218 deletions

diff --git a/src/api/common/Cargo.toml b/src/api/common/Cargo.toml
index 5b9cf479..c33d585d 100644
--- 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
diff --git a/src/api/common/cors.rs b/src/api/common/cors.rs
index 14369b56..09b55c13 100644
--- 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
diff --git a/src/api/common/signature/body.rs b/src/api/common/signature/body.rs
new file mode 100644
index 00000000..96be0d5b
--- /dev/null
+++ 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)
+	}
+}
diff --git a/src/api/common/signature/checksum.rs b/src/api/common/signature/checksum.rs
new file mode 100644
index 00000000..3c5e7c53
--- /dev/null
+++ b/src/api/common/signature/checksum.rs
@@ -0,0 +1,323 @@
+use std::convert::{TryFrom, TryInto};
+use std::hash::Hasher;
+
+use base64::prelude::*;
+use crc32c::Crc32cHasher as Crc32c;
+use crc32fast::Hasher as Crc32;
+use md5::{Digest, Md5};
+use sha1::Sha1;
+use sha2::Sha256;
+
+use http::{HeaderMap, HeaderName, HeaderValue};
+
+use garage_util::data::*;
+
+use super::*;
+
+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");
+pub const X_AMZ_CHECKSUM_MODE: HeaderName = HeaderName::from_static("x-amz-checksum-mode");
+pub const X_AMZ_CHECKSUM_CRC32: HeaderName = HeaderName::from_static("x-amz-checksum-crc32");
+pub const X_AMZ_CHECKSUM_CRC32C: HeaderName = HeaderName::from_static("x-amz-checksum-crc32c");
+pub const X_AMZ_CHECKSUM_SHA1: HeaderName = HeaderName::from_static("x-amz-checksum-sha1");
+pub const X_AMZ_CHECKSUM_SHA256: HeaderName = HeaderName::from_static("x-amz-checksum-sha256");
+
+pub type Crc32Checksum = [u8; 4];
+pub type Crc32cChecksum = [u8; 4];
+pub type Md5Checksum = [u8; 16];
+pub type Sha1Checksum = [u8; 20];
+pub type Sha256Checksum = [u8; 32];
+
+#[derive(Debug, Default, Clone)]
+pub struct ExpectedChecksums {
+	// base64-encoded md5 (content-md5 header)
+	pub md5: Option<String>,
+	// content_sha256 (as a Hash / FixedBytes32)
+	pub sha256: Option<Hash>,
+	// extra x-amz-checksum-* header
+	pub extra: Option<ChecksumValue>,
+}
+
+pub struct Checksummer {
+	pub crc32: Option<Crc32>,
+	pub crc32c: Option<Crc32c>,
+	pub md5: Option<Md5>,
+	pub sha1: Option<Sha1>,
+	pub sha256: Option<Sha256>,
+}
+
+#[derive(Default)]
+pub struct Checksums {
+	pub crc32: Option<Crc32Checksum>,
+	pub crc32c: Option<Crc32cChecksum>,
+	pub md5: Option<Md5Checksum>,
+	pub sha1: Option<Sha1Checksum>,
+	pub sha256: Option<Sha256Checksum>,
+}
+
+impl Checksummer {
+	pub fn new() -> Self {
+		Self {
+			crc32: None,
+			crc32c: None,
+			md5: None,
+			sha1: None,
+			sha256: None,
+		}
+	}
+
+	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 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());
+			}
+			Some(ChecksumAlgorithm::Crc32c) => {
+				self.crc32c = Some(Crc32c::default());
+			}
+			Some(ChecksumAlgorithm::Sha1) => {
+				self.sha1 = Some(Sha1::new());
+			}
+			Some(ChecksumAlgorithm::Sha256) => {
+				self.sha256 = Some(Sha256::new());
+			}
+			None => (),
+		}
+		self
+	}
+
+	pub fn update(&mut self, bytes: &[u8]) {
+		if let Some(crc32) = &mut self.crc32 {
+			crc32.update(bytes);
+		}
+		if let Some(crc32c) = &mut self.crc32c {
+			crc32c.write(bytes);
+		}
+		if let Some(md5) = &mut self.md5 {
+			md5.update(bytes);
+		}
+		if let Some(sha1) = &mut self.sha1 {
+			sha1.update(bytes);
+		}
+		if let Some(sha256) = &mut self.sha256 {
+			sha256.update(bytes);
+		}
+	}
+
+	pub fn finalize(self) -> Checksums {
+		Checksums {
+			crc32: self.crc32.map(|x| u32::to_be_bytes(x.finalize())),
+			crc32c: self
+				.crc32c
+				.map(|x| u32::to_be_bytes(u32::try_from(x.finish()).unwrap())),
+			md5: self.md5.map(|x| x.finalize()[..].try_into().unwrap()),
+			sha1: self.sha1.map(|x| x.finalize()[..].try_into().unwrap()),
+			sha256: self.sha256.map(|x| x.finalize()[..].try_into().unwrap()),
+		}
+	}
+}
+
+impl Checksums {
+	pub fn verify(&self, expected: &ExpectedChecksums) -> Result<(), Error> {
+		if let Some(expected_md5) = &expected.md5 {
+			match self.md5 {
+				Some(md5) if BASE64_STANDARD.encode(&md5) == expected_md5.trim_matches('"') => (),
+				_ => {
+					return Err(Error::InvalidDigest(
+						"MD5 checksum verification failed (from content-md5)".into(),
+					))
+				}
+			}
+		}
+		if let Some(expected_sha256) = &expected.sha256 {
+			match self.sha256 {
+				Some(sha256) if &sha256[..] == expected_sha256.as_slice() => (),
+				_ => {
+					return Err(Error::InvalidDigest(
+						"SHA256 checksum verification failed (from x-amz-content-sha256)".into(),
+					))
+				}
+			}
+		}
+		if let Some(extra) = expected.extra {
+			let algo = extra.algorithm();
+			if self.extract(Some(algo)) != Some(extra) {
+				return Err(Error::InvalidDigest(format!(
+					"Failed to validate checksum for algorithm {:?}",
+					algo
+				)));
+			}
+		}
+		Ok(())
+	}
+
+	pub fn extract(&self, algo: Option<ChecksumAlgorithm>) -> Option<ChecksumValue> {
+		match algo {
+			None => None,
+			Some(ChecksumAlgorithm::Crc32) => Some(ChecksumValue::Crc32(self.crc32.unwrap())),
+			Some(ChecksumAlgorithm::Crc32c) => Some(ChecksumValue::Crc32c(self.crc32c.unwrap())),
+			Some(ChecksumAlgorithm::Sha1) => Some(ChecksumValue::Sha1(self.sha1.unwrap())),
+			Some(ChecksumAlgorithm::Sha256) => Some(ChecksumValue::Sha256(self.sha256.unwrap())),
+		}
+	}
+}
+
+// ----
+
+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 fn request_checksum_algorithm(
+	headers: &HeaderMap<HeaderValue>,
+) -> Result<Option<ChecksumAlgorithm>, Error> {
+	match headers.get(X_AMZ_CHECKSUM_ALGORITHM) {
+		None => Ok(None),
+		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 fn request_checksum_value(
+	headers: &HeaderMap<HeaderValue>,
+) -> Result<Option<ChecksumValue>, Error> {
+	let mut ret = vec![];
+
+	if headers.contains_key(X_AMZ_CHECKSUM_CRC32) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Crc32)?);
+	}
+	if headers.contains_key(X_AMZ_CHECKSUM_CRC32C) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Crc32c)?);
+	}
+	if headers.contains_key(X_AMZ_CHECKSUM_SHA1) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Sha1)?);
+	}
+	if headers.contains_key(X_AMZ_CHECKSUM_SHA256) {
+		ret.push(extract_checksum_value(headers, ChecksumAlgorithm::Sha256)?);
+	}
+
+	if ret.len() > 1 {
+		return Err(Error::bad_request(
+			"multiple x-amz-checksum-* headers given",
+		));
+	}
+	Ok(ret.pop())
+}
+
+/// Checks for the presence of x-amz-checksum-algorithm
+/// if so extract the corresponding x-amz-checksum-* value
+pub fn extract_checksum_value(
+	headers: &HeaderMap<HeaderValue>,
+	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(ChecksumValue::Crc32(crc32))
+		}
+		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(ChecksumValue::Crc32c(crc32c))
+		}
+		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(ChecksumValue::Sha1(sha1))
+		}
+		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(ChecksumValue::Sha256(sha256))
+		}
+	}
+}
+
+pub fn add_checksum_response_headers(
+	checksum: &Option<ChecksumValue>,
+	mut resp: http::response::Builder,
+) -> http::response::Builder {
+	match checksum {
+		Some(ChecksumValue::Crc32(crc32)) => {
+			resp = resp.header(X_AMZ_CHECKSUM_CRC32, BASE64_STANDARD.encode(&crc32));
+		}
+		Some(ChecksumValue::Crc32c(crc32c)) => {
+			resp = resp.header(X_AMZ_CHECKSUM_CRC32C, BASE64_STANDARD.encode(&crc32c));
+		}
+		Some(ChecksumValue::Sha1(sha1)) => {
+			resp = resp.header(X_AMZ_CHECKSUM_SHA1, BASE64_STANDARD.encode(&sha1));
+		}
+		Some(ChecksumValue::Sha256(sha256)) => {
+			resp = resp.header(X_AMZ_CHECKSUM_SHA256, BASE64_STANDARD.encode(&sha256));
+		}
+		None => (),
+	}
+	resp
+}
diff --git a/src/api/common/signature/error.rs b/src/api/common/signature/error.rs
index 2d92a072..b2f396b5 100644
--- 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
diff --git a/src/api/common/signature/mod.rs b/src/api/common/signature/mod.rs
index 6514da43..50fbd304 100644
--- 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"))?;
-
-	let req = streaming::parse_streaming_body(
-		&api_key,
+) -> Result<VerifiedRequest, Error> {
+	let checked_signature = payload::check_payload_signature(&garage, &mut req, service).await?;
+
+	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(
diff --git a/src/api/common/signature/payload.rs b/src/api/common/signature/payload.rs
index 81541e4a..2d5f8603 100644
--- 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
diff --git a/src/api/common/signature/streaming.rs b/src/api/common/signature/streaming.rs
index e223d1b1..64362727 100644
--- 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")?;
-
-			let secret_key = &api_key
-				.state
-				.as_option()
-				.ok_or_internal_error("Deleted key state")?
-				.secret_key;
-
-			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")?;
+	debug!(
+		"Content signature mode: {:?}",
+		checked_signature.content_sha256_header
+	);
+
+	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",
+				));
+			}
+
+			// 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",
+					));
+				}
+			}
+
+			// 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: 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,
+				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),
+}
+
+struct SignParams {
+	datetime: DateTime<Utc>,
+	scope: String,
+	signing_hmac: HmacSha256,
+	previous_signature: Hash,
 }
 
 #[pin_project::pin_project]
-pub struct SignedPayloadStream<S>
+pub struct StreamingPayloadStream<S>
 where
 	S: Stream<Item = Result<Bytes, Error>>,
 {
 	#[pin]
 	stream: S,
 	buf: bytes::BytesMut,
-	datetime: DateTime<Utc>,
-	scope: String,
-	signing_hmac: HmacSha256,
-	previous_signature: Hash,
+	signing: Option<SignParams>,
+	has_trailer: bool,
+	done: bool,
 }
 
-impl<S> SignedPayloadStream<S>
+impl<S> 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 {
+	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))))
+			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",
+								))));
+							}
 						}
-						None => {
-							return Poll::Ready(Some(Err(SignedPayloadStreamError::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();
 					}
-				}
-				Err(nom::Err::Error(e)) | Err(nom::Err::Failure(e)) => {
-					return Poll::Ready(Some(Err(e)))
-				}
-			};
 
-			// 0-sized chunk is the last
-			if payload.data.is_empty() {
-				return Poll::Ready(None);
-			}
+					*this.buf = input.into();
 
-			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)));
-			}
+					// 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))));
+				}
+				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.previous_signature = payload.header.signature;
+					*this.buf = input.into();
+					*this.done = true;
 
-			return Poll::Ready(Some(Ok(payload.data)));
+					let mut trailers_map = HeaderMap::new();
+					trailers_map.insert(trailer.header_name, trailer.header_value);
+
+					return Poll::Ready(Some(Ok(Frame::trailers(trailers_map))));
+				}
+			}
 		}
 	}
 
@@ -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