Merge pull request 'Garage v1.0' (#683) from next-0.10 into main

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

12 files changed, 1997 insertions, 375 deletions

diff --git a/src/api/s3/api_server.rs b/src/api/s3/api_server.rs
index 1ed30996..1737af33 100644
--- a/src/api/s3/api_server.rs
+++ b/src/api/s3/api_server.rs
@@ -325,7 +325,7 @@ impl ApiHandler for S3ApiServer {
 					part_number_marker: part_number_marker.map(|p| p.min(10000)),
 					max_parts: max_parts.unwrap_or(1000).clamp(1, 1000),
 				};
-				handle_list_parts(ctx, &query).await
+				handle_list_parts(ctx, req, &query).await
 			}
 			Endpoint::DeleteObjects {} => handle_delete_objects(ctx, req, content_sha256).await,
 			Endpoint::GetBucketWebsite {} => handle_get_website(ctx).await,
diff --git a/src/api/s3/checksum.rs b/src/api/s3/checksum.rs
new file mode 100644
index 00000000..c9dc001c
--- /dev/null
+++ b/src/api/s3/checksum.rs
@@ -0,0 +1,406 @@
+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 garage_util::error::OkOrMessage;
+
+use garage_model::s3::object_table::*;
+
+use crate::s3::error::*;
+
+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)]
+pub(crate) 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(crate) 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(crate) 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(crate) fn init(expected: &ExpectedChecksums, require_md5: bool) -> Self {
+		let mut ret = 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());
+		}
+		ret
+	}
+
+	pub(crate) 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(crate) 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(crate) 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())),
+		}
+	}
+}
+
+// ----
+
+#[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)
+	}
+}
+
+// ----
+
+/// Extract the value of the x-amz-checksum-algorithm header
+pub(crate) 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)),
+		_ => Err(Error::bad_request("invalid checksum algorithm")),
+	}
+}
+
+/// Extract the value of any of the x-amz-checksum-* headers
+pub(crate) 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 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 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 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 ret.len() > 1 {
+		return Err(Error::bad_request(
+			"multiple x-amz-checksum-* headers given",
+		));
+	}
+	Ok(ret.pop())
+}
+
+/// Checks for the presense of x-amz-checksum-algorithm
+/// if so extract the corrseponding x-amz-checksum-* value
+pub(crate) fn request_checksum_algorithm_value(
+	headers: &HeaderMap<HeaderValue>,
+) -> Result<Option<ChecksumValue>, Error> {
+	match headers.get(X_AMZ_CHECKSUM_ALGORITHM) {
+		Some(x) if x == "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)))
+		}
+		Some(x) if x == "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)))
+		}
+		Some(x) if x == "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)))
+		}
+		Some(x) if x == "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)))
+		}
+		Some(_) => Err(Error::bad_request("invalid x-amz-checksum-algorithm")),
+		None => Ok(None),
+	}
+}
+
+pub(crate) 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/s3/copy.rs b/src/api/s3/copy.rs
index 3c2bd483..411a6917 100644
--- a/src/api/s3/copy.rs
+++ b/src/api/s3/copy.rs
@@ -1,17 +1,18 @@
 use std::pin::Pin;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
-use futures::{stream, stream::Stream, StreamExt};
-use md5::{Digest as Md5Digest, Md5};
+use futures::{stream, stream::Stream, StreamExt, TryStreamExt};
 
 use bytes::Bytes;
 use hyper::{Request, Response};
 use serde::Serialize;
 
 use garage_net::bytes_buf::BytesBuf;
+use garage_net::stream::read_stream_to_end;
 use garage_rpc::rpc_helper::OrderTag;
 use garage_table::*;
 use garage_util::data::*;
+use garage_util::error::Error as GarageError;
 use garage_util::time::*;
 
 use garage_model::s3::block_ref_table::*;
@@ -21,11 +22,16 @@ use garage_model::s3::version_table::*;
 
 use crate::helpers::*;
 use crate::s3::api_server::{ReqBody, ResBody};
+use crate::s3::checksum::*;
+use crate::s3::encryption::EncryptionParams;
 use crate::s3::error::*;
+use crate::s3::get::full_object_byte_stream;
 use crate::s3::multipart;
-use crate::s3::put::get_headers;
+use crate::s3::put::{get_headers, save_stream, ChecksumMode, SaveStreamResult};
 use crate::s3::xml::{self as s3_xml, xmlns_tag};
 
+// -------- CopyObject ---------
+
 pub async fn handle_copy(
 	ctx: ReqCtx,
 	req: &Request<ReqBody>,
@@ -33,13 +39,9 @@ pub async fn handle_copy(
 ) -> Result<Response<ResBody>, Error> {
 	let copy_precondition = CopyPreconditionHeaders::parse(req)?;
 
-	let source_object = get_copy_source(&ctx, req).await?;
+	let checksum_algorithm = request_checksum_algorithm(req.headers())?;
 
-	let ReqCtx {
-		garage,
-		bucket_id: dest_bucket_id,
-		..
-	} = ctx;
+	let source_object = get_copy_source(&ctx, req).await?;
 
 	let (source_version, source_version_data, source_version_meta) =
 		extract_source_info(&source_object)?;
@@ -47,26 +49,150 @@ pub async fn handle_copy(
 	// Check precondition, e.g. x-amz-copy-source-if-match
 	copy_precondition.check(source_version, &source_version_meta.etag)?;
 
+	// Determine encryption parameters
+	let (source_encryption, source_object_meta_inner) =
+		EncryptionParams::check_decrypt_for_copy_source(
+			&ctx.garage,
+			req.headers(),
+			&source_version_meta.encryption,
+		)?;
+	let dest_encryption = EncryptionParams::new_from_headers(&ctx.garage, req.headers())?;
+
+	// Extract source checksum info before source_object_meta_inner is consumed
+	let source_checksum = source_object_meta_inner.checksum;
+	let source_checksum_algorithm = source_checksum.map(|x| x.algorithm());
+
+	// If source object has a checksum, the destination object must as well.
+	// The x-amz-checksum-algorihtm header allows to change that algorithm,
+	// but if it is absent, we must use the same as before
+	let checksum_algorithm = checksum_algorithm.or(source_checksum_algorithm);
+
+	// Determine metadata of destination object
+	let was_multipart = source_version_meta.etag.contains('-');
+	let dest_object_meta = ObjectVersionMetaInner {
+		headers: match req.headers().get("x-amz-metadata-directive") {
+			Some(v) if v == hyper::header::HeaderValue::from_static("REPLACE") => {
+				get_headers(req.headers())?
+			}
+			_ => source_object_meta_inner.into_owned().headers,
+		},
+		checksum: source_checksum,
+	};
+
+	// Do actual object copying
+	//
+	// In any of the following scenarios, we need to read the whole object
+	// data and re-write it again:
+	//
+	// - the data needs to be decrypted or encrypted
+	// - the requested checksum algorithm requires us to recompute a checksum
+	// - the original object was a multipart upload and a checksum algorithm
+	//   is defined (AWS specifies that in this case, we must recompute the
+	//   checksum from scratch as if this was a single big object and not
+	//   a multipart object, as the checksums are not computed in the same way)
+	//
+	// In other cases, we can just copy the metadata and reference the same blocks.
+	//
+	// See: https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html
+
+	let must_recopy = !EncryptionParams::is_same(&source_encryption, &dest_encryption)
+		|| source_checksum_algorithm != checksum_algorithm
+		|| (was_multipart && checksum_algorithm.is_some());
+
+	let res = if !must_recopy {
+		// In most cases, we can just copy the metadata and link blocks of the
+		// old object from the new object.
+		handle_copy_metaonly(
+			ctx,
+			dest_key,
+			dest_object_meta,
+			dest_encryption,
+			source_version,
+			source_version_data,
+			source_version_meta,
+		)
+		.await?
+	} else {
+		let expected_checksum = ExpectedChecksums {
+			md5: None,
+			sha256: None,
+			extra: source_checksum,
+		};
+		let checksum_mode = if was_multipart || source_checksum_algorithm != checksum_algorithm {
+			ChecksumMode::Calculate(checksum_algorithm)
+		} else {
+			ChecksumMode::Verify(&expected_checksum)
+		};
+		// If source and dest encryption use different keys,
+		// we must decrypt content and re-encrypt, so rewrite all data blocks.
+		handle_copy_reencrypt(
+			ctx,
+			dest_key,
+			dest_object_meta,
+			dest_encryption,
+			source_version,
+			source_version_data,
+			source_encryption,
+			checksum_mode,
+		)
+		.await?
+	};
+
+	let last_modified = msec_to_rfc3339(res.version_timestamp);
+	let result = CopyObjectResult {
+		last_modified: s3_xml::Value(last_modified),
+		etag: s3_xml::Value(format!("\"{}\"", res.etag)),
+	};
+	let xml = s3_xml::to_xml_with_header(&result)?;
+
+	let mut resp = Response::builder()
+		.header("Content-Type", "application/xml")
+		.header("x-amz-version-id", hex::encode(res.version_uuid))
+		.header(
+			"x-amz-copy-source-version-id",
+			hex::encode(source_version.uuid),
+		);
+	dest_encryption.add_response_headers(&mut resp);
+	Ok(resp.body(string_body(xml))?)
+}
+
+async fn handle_copy_metaonly(
+	ctx: ReqCtx,
+	dest_key: &str,
+	dest_object_meta: ObjectVersionMetaInner,
+	dest_encryption: EncryptionParams,
+	source_version: &ObjectVersion,
+	source_version_data: &ObjectVersionData,
+	source_version_meta: &ObjectVersionMeta,
+) -> Result<SaveStreamResult, Error> {
+	let ReqCtx {
+		garage,
+		bucket_id: dest_bucket_id,
+		..
+	} = ctx;
+
 	// Generate parameters for copied object
 	let new_uuid = gen_uuid();
 	let new_timestamp = now_msec();
 
-	// Implement x-amz-metadata-directive: REPLACE
-	let new_meta = match req.headers().get("x-amz-metadata-directive") {
-		Some(v) if v == hyper::header::HeaderValue::from_static("REPLACE") => ObjectVersionMeta {
-			headers: get_headers(req.headers())?,
-			size: source_version_meta.size,
-			etag: source_version_meta.etag.clone(),
-		},
-		_ => source_version_meta.clone(),
+	let new_meta = ObjectVersionMeta {
+		encryption: dest_encryption.encrypt_meta(dest_object_meta)?,
+		size: source_version_meta.size,
+		etag: source_version_meta.etag.clone(),
 	};
 
-	let etag = new_meta.etag.to_string();
+	let res = SaveStreamResult {
+		version_uuid: new_uuid,
+		version_timestamp: new_timestamp,
+		etag: new_meta.etag.clone(),
+	};
 
 	// Save object copy
 	match source_version_data {
 		ObjectVersionData::DeleteMarker => unreachable!(),
 		ObjectVersionData::Inline(_meta, bytes) => {
+			// bytes is either plaintext before&after or encrypted with the
+			// same keys, so it's ok to just copy it as is
 			let dest_object_version = ObjectVersion {
 				uuid: new_uuid,
 				timestamp: new_timestamp,
@@ -97,7 +223,8 @@ pub async fn handle_copy(
 				uuid: new_uuid,
 				timestamp: new_timestamp,
 				state: ObjectVersionState::Uploading {
-					headers: new_meta.headers.clone(),
+					encryption: new_meta.encryption.clone(),
+					checksum_algorithm: None,
 					multipart: false,
 				},
 			};
@@ -164,23 +291,42 @@ pub async fn handle_copy(
 		}
 	}
 
-	let last_modified = msec_to_rfc3339(new_timestamp);
-	let result = CopyObjectResult {
-		last_modified: s3_xml::Value(last_modified),
-		etag: s3_xml::Value(format!("\"{}\"", etag)),
-	};
-	let xml = s3_xml::to_xml_with_header(&result)?;
+	Ok(res)
+}
 
-	Ok(Response::builder()
-		.header("Content-Type", "application/xml")
-		.header("x-amz-version-id", hex::encode(new_uuid))
-		.header(
-			"x-amz-copy-source-version-id",
-			hex::encode(source_version.uuid),
-		)
-		.body(string_body(xml))?)
+async fn handle_copy_reencrypt(
+	ctx: ReqCtx,
+	dest_key: &str,
+	dest_object_meta: ObjectVersionMetaInner,
+	dest_encryption: EncryptionParams,
+	source_version: &ObjectVersion,
+	source_version_data: &ObjectVersionData,
+	source_encryption: EncryptionParams,
+	checksum_mode: ChecksumMode<'_>,
+) -> Result<SaveStreamResult, Error> {
+	// basically we will read the source data (decrypt if necessary)
+	// and save that in a new object (encrypt if necessary),
+	// by combining the code used in getobject and putobject
+	let source_stream = full_object_byte_stream(
+		ctx.garage.clone(),
+		source_version,
+		source_version_data,
+		source_encryption,
+	);
+
+	save_stream(
+		&ctx,
+		dest_object_meta,
+		dest_encryption,
+		source_stream.map_err(|e| Error::from(GarageError::from(e))),
+		&dest_key.to_string(),
+		checksum_mode,
+	)
+	.await
 }
 
+// -------- UploadPartCopy ---------
+
 pub async fn handle_upload_part_copy(
 	ctx: ReqCtx,
 	req: &Request<ReqBody>,
@@ -193,7 +339,7 @@ pub async fn handle_upload_part_copy(
 	let dest_upload_id = multipart::decode_upload_id(upload_id)?;
 
 	let dest_key = dest_key.to_string();
-	let (source_object, (_, _, mut dest_mpu)) = futures::try_join!(
+	let (source_object, (_, dest_version, mut dest_mpu)) = futures::try_join!(
 		get_copy_source(&ctx, req),
 		multipart::get_upload(&ctx, &dest_key, &dest_upload_id)
 	)?;
@@ -206,6 +352,24 @@ pub async fn handle_upload_part_copy(
 	// Check precondition on source, e.g. x-amz-copy-source-if-match
 	copy_precondition.check(source_object_version, &source_version_meta.etag)?;
 
+	// Determine encryption parameters
+	let (source_encryption, _) = EncryptionParams::check_decrypt_for_copy_source(
+		&garage,
+		req.headers(),
+		&source_version_meta.encryption,
+	)?;
+	let (dest_object_encryption, dest_object_checksum_algorithm) = match dest_version.state {
+		ObjectVersionState::Uploading {
+			encryption,
+			checksum_algorithm,
+			..
+		} => (encryption, checksum_algorithm),
+		_ => unreachable!(),
+	};
+	let (dest_encryption, _) =
+		EncryptionParams::check_decrypt(&garage, req.headers(), &dest_object_encryption)?;
+	let same_encryption = EncryptionParams::is_same(&source_encryption, &dest_encryption);
+
 	// Check source range is valid
 	let source_range = match req.headers().get("x-amz-copy-source-range") {
 		Some(range) => {
@@ -227,21 +391,16 @@ pub async fn handle_upload_part_copy(
 	};
 
 	// Check source version is not inlined
-	match source_version_data {
-		ObjectVersionData::DeleteMarker => unreachable!(),
-		ObjectVersionData::Inline(_meta, _bytes) => {
-			// This is only for small files, we don't bother handling this.
-			// (in AWS UploadPartCopy works for parts at least 5MB which
-			// is never the case of an inline object)
-			return Err(Error::bad_request(
-				"Source object is too small (minimum part size is 5Mb)",
-			));
-		}
-		ObjectVersionData::FirstBlock(_meta, _first_block_hash) => (),
-	};
+	if matches!(source_version_data, ObjectVersionData::Inline(_, _)) {
+		// This is only for small files, we don't bother handling this.
+		// (in AWS UploadPartCopy works for parts at least 5MB which
+		// is never the case of an inline object)
+		return Err(Error::bad_request(
+			"Source object is too small (minimum part size is 5Mb)",
+		));
+	}
 
-	// Fetch source versin with its block list,
-	// and destination version to check part hasn't yet been uploaded
+	// Fetch source version with its block list
 	let source_version = garage
 		.version_table
 		.get(&source_object_version.uuid, &EmptyKey)
@@ -251,7 +410,9 @@ pub async fn handle_upload_part_copy(
 	// We want to reuse blocks from the source version as much as possible.
 	// However, we still need to get the data from these blocks
 	// because we need to know it to calculate the MD5sum of the part
-	// which is used as its ETag.
+	// which is used as its ETag. For encrypted sources or destinations,
+	// we must always read(+decrypt) and then write(+encrypt), so we
+	// can never reuse data blocks as is.
 
 	// First, calculate what blocks we want to keep,
 	// and the subrange of the block to take, if the bounds of the
@@ -300,7 +461,9 @@ pub async fn handle_upload_part_copy(
 		dest_mpu_part_key,
 		MpuPart {
 			version: dest_version_id,
+			// These are all filled in later (bottom of this function)
 			etag: None,
+			checksum: None,
 			size: None,
 		},
 	);
@@ -313,32 +476,55 @@ pub async fn handle_upload_part_copy(
 		},
 		false,
 	);
+	// write an empty version now to be the parent of the block_ref entries
+	garage.version_table.insert(&dest_version).await?;
 
 	// Now, actually copy the blocks
-	let mut md5hasher = Md5::new();
+	let mut checksummer = Checksummer::init(&Default::default(), !dest_encryption.is_encrypted())
+		.add(dest_object_checksum_algorithm);
 
 	// First, create a stream that is able to read the source blocks
 	// and extract the subrange if necessary.
 	// The second returned value is an Option<Hash>, that is Some
 	// if and only if the block returned is a block that already existed
-	// in the Garage data store (thus we don't need to save it again).
+	// in the Garage data store and can be reused as-is instead of having
+	// to save it again. This excludes encrypted source blocks that we had
+	// to decrypt.
 	let garage2 = garage.clone();
 	let order_stream = OrderTag::stream();
 	let source_blocks = stream::iter(blocks_to_copy)
 		.enumerate()
-		.flat_map(|(i, (block_hash, range_to_copy))| {
+		.map(|(i, (block_hash, range_to_copy))| {
 			let garage3 = garage2.clone();
-			stream::once(async move {
-				let data = garage3
-					.block_manager
-					.rpc_get_block(&block_hash, Some(order_stream.order(i as u64)))
+			async move {
+				let stream = source_encryption
+					.get_block(&garage3, &block_hash, Some(order_stream.order(i as u64)))
 					.await?;
+				let data = read_stream_to_end(stream).await?.into_bytes();
+				// For each item, we return a tuple of:
+				// 1. the full data block (decrypted)
+				// 2. an Option<Hash> that indicates the hash of the block in the block store,
+				//    only if it can be re-used as-is in the copied object
 				match range_to_copy {
-					Some(r) => Ok((data.slice(r), None)),
-					None => Ok((data, Some(block_hash))),
+					Some(r) => {
+						// If we are taking a subslice of the data, we cannot reuse the block as-is
+						Ok((data.slice(r), None))
+					}
+					None if same_encryption => {
+						// If the data is unencrypted before & after, or if we are using
+						// the same encryption key, we can reuse the stored block, no need
+						// to re-send it to storage nodes.
+						Ok((data, Some(block_hash)))
+					}
+					None => {
+						// If we are decrypting / (re)encrypting with different keys,
+						// we cannot reuse the block as-is
+						Ok((data, None))
+					}
 				}
-			})
+			}
 		})
+		.buffered(2)
 		.peekable();
 
 	// The defragmenter is a custom stream (defined below) that concatenates
@@ -346,22 +532,39 @@ pub async fn handle_upload_part_copy(
 	// It returns a series of (Vec<u8>, Option<Hash>).
 	// When it is done, it returns an empty vec.
 	// Same as the previous iterator, the Option is Some(_) if and only if
-	// it's an existing block of the Garage data store.
+	// it's an existing block of the Garage data store that can be reused.
 	let mut defragmenter = Defragmenter::new(garage.config.block_size, Box::pin(source_blocks));
 
 	let mut current_offset = 0;
 	let mut next_block = defragmenter.next().await?;
 
+	// TODO this could be optimized similarly to read_and_put_blocks
+	// low priority because uploadpartcopy is rarely used
 	loop {
 		let (data, existing_block_hash) = next_block;
 		if data.is_empty() {
 			break;
 		}
 
-		md5hasher.update(&data[..]);
-
-		let must_upload = existing_block_hash.is_none();
-		let final_hash = existing_block_hash.unwrap_or_else(|| blake2sum(&data[..]));
+		let data_len = data.len() as u64;
+
+		let (checksummer_updated, (data_to_upload, final_hash)) =
+			tokio::task::spawn_blocking(move || {
+				checksummer.update(&data[..]);
+
+				let tup = match existing_block_hash {
+					Some(hash) if same_encryption => (None, hash),
+					_ => {
+						let data_enc = dest_encryption.encrypt_block(data)?;
+						let hash = blake2sum(&data_enc);
+						(Some(data_enc), hash)
+					}
+				};
+				Ok::<_, Error>((checksummer, tup))
+			})
+			.await
+			.unwrap()?;
+		checksummer = checksummer_updated;
 
 		dest_version.blocks.clear();
 		dest_version.blocks.put(
@@ -371,10 +574,10 @@ pub async fn handle_upload_part_copy(
 			},
 			VersionBlock {
 				hash: final_hash,
-				size: data.len() as u64,
+				size: data_len,
 			},
 		);
-		current_offset += data.len() as u64;
+		current_offset += data_len;
 
 		let block_ref = BlockRef {
 			block: final_hash,
@@ -382,36 +585,34 @@ pub async fn handle_upload_part_copy(
 			deleted: false.into(),
 		};
 
-		let garage2 = garage.clone();
-		let res = futures::try_join!(
+		let (_, _, _, next) = futures::try_join!(
 			// Thing 1: if the block is not exactly a block that existed before,
 			// we need to insert that data as a new block.
-			async move {
-				if must_upload {
-					garage2
+			async {
+				if let Some(final_data) = data_to_upload {
+					garage
 						.block_manager
-						.rpc_put_block(final_hash, data, None)
+						.rpc_put_block(final_hash, final_data, dest_encryption.is_encrypted(), None)
 						.await
 				} else {
 					Ok(())
 				}
 			},
-			async {
-				// Thing 2: we need to insert the block in the version
-				garage.version_table.insert(&dest_version).await?;
-				// Thing 3: we need to add a block reference
-				garage.block_ref_table.insert(&block_ref).await
-			},
-			// Thing 4: we need to prefetch the next block
+			// Thing 2: we need to insert the block in the version
+			garage.version_table.insert(&dest_version),
+			// Thing 3: we need to add a block reference
+			garage.block_ref_table.insert(&block_ref),
+			// Thing 4: we need to read the next block
 			defragmenter.next(),
 		)?;
-		next_block = res.2;
+		next_block = next;
 	}
 
 	assert_eq!(current_offset, source_range.length);
 
-	let data_md5sum = md5hasher.finalize();
-	let etag = hex::encode(data_md5sum);
+	let checksums = checksummer.finalize();
+	let etag = dest_encryption.etag_from_md5(&checksums.md5);
+	let checksum = checksums.extract(dest_object_checksum_algorithm);
 
 	// Put the part's ETag in the Versiontable
 	dest_mpu.parts.put(
@@ -419,6 +620,7 @@ pub async fn handle_upload_part_copy(
 		MpuPart {
 			version: dest_version_id,
 			etag: Some(etag.clone()),
+			checksum,
 			size: Some(current_offset),
 		},
 	);
@@ -431,13 +633,14 @@ pub async fn handle_upload_part_copy(
 		last_modified: s3_xml::Value(msec_to_rfc3339(source_object_version.timestamp)),
 	})?;
 
-	Ok(Response::builder()
+	let mut resp = Response::builder()
 		.header("Content-Type", "application/xml")
 		.header(
 			"x-amz-copy-source-version-id",
 			hex::encode(source_object_version.uuid),
-		)
-		.body(string_body(resp_xml))?)
+		);
+	dest_encryption.add_response_headers(&mut resp);
+	Ok(resp.body(string_body(resp_xml))?)
 }
 
 async fn get_copy_source(ctx: &ReqCtx, req: &Request<ReqBody>) -> Result<Object, Error> {
diff --git a/src/api/s3/encryption.rs b/src/api/s3/encryption.rs
new file mode 100644
index 00000000..2e6ed65c
--- /dev/null
+++ b/src/api/s3/encryption.rs
@@ -0,0 +1,595 @@
+use std::borrow::Cow;
+use std::convert::TryInto;
+use std::pin::Pin;
+
+use aes_gcm::{
+	aead::stream::{DecryptorLE31, EncryptorLE31, StreamLE31},
+	aead::{Aead, AeadCore, KeyInit, OsRng},
+	aes::cipher::crypto_common::rand_core::RngCore,
+	aes::cipher::typenum::Unsigned,
+	Aes256Gcm, Key, Nonce,
+};
+use base64::prelude::*;
+use bytes::Bytes;
+
+use futures::stream::Stream;
+use futures::task;
+use tokio::io::BufReader;
+
+use http::header::{HeaderMap, HeaderName, HeaderValue};
+
+use garage_net::bytes_buf::BytesBuf;
+use garage_net::stream::{stream_asyncread, ByteStream};
+use garage_rpc::rpc_helper::OrderTag;
+use garage_util::data::Hash;
+use garage_util::error::Error as GarageError;
+use garage_util::migrate::Migrate;
+
+use garage_model::garage::Garage;
+use garage_model::s3::object_table::{ObjectVersionEncryption, ObjectVersionMetaInner};
+
+use crate::common_error::*;
+use crate::s3::checksum::Md5Checksum;
+use crate::s3::error::Error;
+
+const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM: HeaderName =
+	HeaderName::from_static("x-amz-server-side-encryption-customer-algorithm");
+const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY: HeaderName =
+	HeaderName::from_static("x-amz-server-side-encryption-customer-key");
+const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5: HeaderName =
+	HeaderName::from_static("x-amz-server-side-encryption-customer-key-md5");
+
+const X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM: HeaderName =
+	HeaderName::from_static("x-amz-copy-source-server-side-encryption-customer-algorithm");
+const X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY: HeaderName =
+	HeaderName::from_static("x-amz-copy-source-server-side-encryption-customer-key");
+const X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5: HeaderName =
+	HeaderName::from_static("x-amz-copy-source-server-side-encryption-customer-key-md5");
+
+const CUSTOMER_ALGORITHM_AES256: &[u8] = b"AES256";
+
+type Md5Output = md5::digest::Output<md5::Md5Core>;
+
+type StreamNonceSize = aes_gcm::aead::stream::NonceSize<Aes256Gcm, StreamLE31<Aes256Gcm>>;
+
+// Data blocks are encrypted by smaller chunks of size 4096 bytes,
+// so that data can be streamed when reading.
+// This size has to be known and has to be constant, or data won't be
+// readable anymore. DO NOT CHANGE THIS VALUE.
+const STREAM_ENC_PLAIN_CHUNK_SIZE: usize = 0x1000; // 4096 bytes
+const STREAM_ENC_CYPER_CHUNK_SIZE: usize = STREAM_ENC_PLAIN_CHUNK_SIZE + 16;
+
+#[derive(Clone, Copy)]
+pub enum EncryptionParams {
+	Plaintext,
+	SseC {
+		client_key: Key<Aes256Gcm>,
+		client_key_md5: Md5Output,
+		compression_level: Option<i32>,
+	},
+}
+
+impl EncryptionParams {
+	pub fn is_encrypted(&self) -> bool {
+		!matches!(self, Self::Plaintext)
+	}
+
+	pub fn is_same(a: &Self, b: &Self) -> bool {
+		let relevant_info = |x: &Self| match x {
+			Self::Plaintext => None,
+			Self::SseC {
+				client_key,
+				compression_level,
+				..
+			} => Some((*client_key, compression_level.is_some())),
+		};
+		relevant_info(a) == relevant_info(b)
+	}
+
+	pub fn new_from_headers(
+		garage: &Garage,
+		headers: &HeaderMap,
+	) -> Result<EncryptionParams, Error> {
+		let key = parse_request_headers(
+			headers,
+			&X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM,
+			&X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY,
+			&X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
+		)?;
+		match key {
+			Some((client_key, client_key_md5)) => Ok(EncryptionParams::SseC {
+				client_key,
+				client_key_md5,
+				compression_level: garage.config.compression_level,
+			}),
+			None => Ok(EncryptionParams::Plaintext),
+		}
+	}
+
+	pub fn add_response_headers(&self, resp: &mut http::response::Builder) {
+		if let Self::SseC { client_key_md5, .. } = self {
+			let md5 = BASE64_STANDARD.encode(&client_key_md5);
+
+			resp.headers_mut().unwrap().insert(
+				X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM,
+				HeaderValue::from_bytes(CUSTOMER_ALGORITHM_AES256).unwrap(),
+			);
+			resp.headers_mut().unwrap().insert(
+				X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
+				HeaderValue::from_bytes(md5.as_bytes()).unwrap(),
+			);
+		}
+	}
+
+	pub fn check_decrypt<'a>(
+		garage: &Garage,
+		headers: &HeaderMap,
+		obj_enc: &'a ObjectVersionEncryption,
+	) -> Result<(Self, Cow<'a, ObjectVersionMetaInner>), Error> {
+		let key = parse_request_headers(
+			headers,
+			&X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM,
+			&X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY,
+			&X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
+		)?;
+		Self::check_decrypt_common(garage, key, obj_enc)
+	}
+
+	pub fn check_decrypt_for_copy_source<'a>(
+		garage: &Garage,
+		headers: &HeaderMap,
+		obj_enc: &'a ObjectVersionEncryption,
+	) -> Result<(Self, Cow<'a, ObjectVersionMetaInner>), Error> {
+		let key = parse_request_headers(
+			headers,
+			&X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM,
+			&X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY,
+			&X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
+		)?;
+		Self::check_decrypt_common(garage, key, obj_enc)
+	}
+
+	fn check_decrypt_common<'a>(
+		garage: &Garage,
+		key: Option<(Key<Aes256Gcm>, Md5Output)>,
+		obj_enc: &'a ObjectVersionEncryption,
+	) -> Result<(Self, Cow<'a, ObjectVersionMetaInner>), Error> {
+		match (key, &obj_enc) {
+			(
+				Some((client_key, client_key_md5)),
+				ObjectVersionEncryption::SseC { inner, compressed },
+			) => {
+				let enc = Self::SseC {
+					client_key,
+					client_key_md5,
+					compression_level: if *compressed {
+						Some(garage.config.compression_level.unwrap_or(1))
+					} else {
+						None
+					},
+				};
+				let plaintext = enc.decrypt_blob(&inner)?;
+				let inner = ObjectVersionMetaInner::decode(&plaintext)
+					.ok_or_internal_error("Could not decode encrypted metadata")?;
+				Ok((enc, Cow::Owned(inner)))
+			}
+			(None, ObjectVersionEncryption::Plaintext { inner }) => {
+				Ok((Self::Plaintext, Cow::Borrowed(inner)))
+			}
+			(_, ObjectVersionEncryption::SseC { .. }) => {
+				Err(Error::bad_request("Object is encrypted"))
+			}
+			(Some(_), _) => {
+				// TODO: should this be an OK scenario?
+				Err(Error::bad_request("Trying to decrypt a plaintext object"))
+			}
+		}
+	}
+
+	pub fn encrypt_meta(
+		&self,
+		meta: ObjectVersionMetaInner,
+	) -> Result<ObjectVersionEncryption, Error> {
+		match self {
+			Self::SseC {
+				compression_level, ..
+			} => {
+				let plaintext = meta.encode().map_err(GarageError::from)?;
+				let ciphertext = self.encrypt_blob(&plaintext)?;
+				Ok(ObjectVersionEncryption::SseC {
+					inner: ciphertext.into_owned(),
+					compressed: compression_level.is_some(),
+				})
+			}
+			Self::Plaintext => Ok(ObjectVersionEncryption::Plaintext { inner: meta }),
+		}
+	}
+
+	// ---- generating object Etag values ----
+	pub fn etag_from_md5(&self, md5sum: &Option<Md5Checksum>) -> String {
+		match self {
+			Self::Plaintext => md5sum
+				.map(|x| hex::encode(&x[..]))
+				.expect("md5 digest should have been computed"),
+			Self::SseC { .. } => {
+				// AWS specifies that for encrypted objects, the Etag is not
+				// the md5sum of the data, but doesn't say what it is.
+				// So we just put some random bytes.
+				let mut random = [0u8; 16];
+				OsRng.fill_bytes(&mut random);
+				hex::encode(&random)
+			}
+		}
+	}
+
+	// ---- generic function for encrypting / decrypting blobs ----
+	// Prepends a randomly-generated nonce to the encrypted value.
+	// This is used for encrypting object metadata and inlined data for small objects.
+	// This does not compress anything.
+
+	pub fn encrypt_blob<'a>(&self, blob: &'a [u8]) -> Result<Cow<'a, [u8]>, Error> {
+		match self {
+			Self::SseC { client_key, .. } => {
+				let cipher = Aes256Gcm::new(&client_key);
+				let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
+				let ciphertext = cipher
+					.encrypt(&nonce, blob)
+					.ok_or_internal_error("Encryption failed")?;
+				Ok(Cow::Owned([nonce.to_vec(), ciphertext].concat()))
+			}
+			Self::Plaintext => Ok(Cow::Borrowed(blob)),
+		}
+	}
+
+	pub fn decrypt_blob<'a>(&self, blob: &'a [u8]) -> Result<Cow<'a, [u8]>, Error> {
+		match self {
+			Self::SseC { client_key, .. } => {
+				let cipher = Aes256Gcm::new(&client_key);
+				let nonce_size = <Aes256Gcm as AeadCore>::NonceSize::to_usize();
+				let nonce = Nonce::from_slice(
+					blob.get(..nonce_size)
+						.ok_or_internal_error("invalid encrypted data")?,
+				);
+				let plaintext = cipher
+					.decrypt(nonce, &blob[nonce_size..])
+					.ok_or_bad_request(
+						"Invalid encryption key, could not decrypt object metadata.",
+					)?;
+				Ok(Cow::Owned(plaintext))
+			}
+			Self::Plaintext => Ok(Cow::Borrowed(blob)),
+		}
+	}
+
+	// ----  function for encrypting / decrypting byte streams ----
+
+	/// Get a data block from the storage node, and decrypt+decompress it
+	/// if necessary. If object is plaintext, just get it without any processing.
+	pub async fn get_block(
+		&self,
+		garage: &Garage,
+		hash: &Hash,
+		order: Option<OrderTag>,
+	) -> Result<ByteStream, GarageError> {
+		let raw_block = garage
+			.block_manager
+			.rpc_get_block_streaming(hash, order)
+			.await?;
+		Ok(self.decrypt_block_stream(raw_block))
+	}
+
+	pub fn decrypt_block_stream(&self, stream: ByteStream) -> ByteStream {
+		match self {
+			Self::Plaintext => stream,
+			Self::SseC {
+				client_key,
+				compression_level,
+				..
+			} => {
+				let plaintext = DecryptStream::new(stream, *client_key);
+				if compression_level.is_some() {
+					let reader = stream_asyncread(Box::pin(plaintext));
+					let reader = BufReader::new(reader);
+					let reader = async_compression::tokio::bufread::ZstdDecoder::new(reader);
+					Box::pin(tokio_util::io::ReaderStream::new(reader))
+				} else {
+					Box::pin(plaintext)
+				}
+			}
+		}
+	}
+
+	/// Encrypt a data block if encryption is set, for use before
+	/// putting the data blocks into storage
+	pub fn encrypt_block(&self, block: Bytes) -> Result<Bytes, Error> {
+		match self {
+			Self::Plaintext => Ok(block),
+			Self::SseC {
+				client_key,
+				compression_level,
+				..
+			} => {
+				let block = if let Some(level) = compression_level {
+					Cow::Owned(
+						garage_block::zstd_encode(block.as_ref(), *level)
+							.ok_or_internal_error("failed to compress data block")?,
+					)
+				} else {
+					Cow::Borrowed(block.as_ref())
+				};
+
+				let mut ret = Vec::with_capacity(block.len() + 32 + block.len() / 64);
+
+				let mut nonce: Nonce<StreamNonceSize> = Default::default();
+				OsRng.fill_bytes(&mut nonce);
+				ret.extend_from_slice(nonce.as_slice());
+
+				let mut cipher = EncryptorLE31::<Aes256Gcm>::new(&client_key, &nonce);
+				let mut iter = block.chunks(STREAM_ENC_PLAIN_CHUNK_SIZE).peekable();
+
+				if iter.peek().is_none() {
+					// Empty stream: we encrypt an empty last chunk
+					let chunk_enc = cipher
+						.encrypt_last(&[][..])
+						.ok_or_internal_error("failed to encrypt chunk")?;
+					ret.extend_from_slice(&chunk_enc);
+				} else {
+					loop {
+						let chunk = iter.next().unwrap();
+						if iter.peek().is_some() {
+							let chunk_enc = cipher
+								.encrypt_next(chunk)
+								.ok_or_internal_error("failed to encrypt chunk")?;
+							assert_eq!(chunk.len(), STREAM_ENC_PLAIN_CHUNK_SIZE);
+							assert_eq!(chunk_enc.len(), STREAM_ENC_CYPER_CHUNK_SIZE);
+							ret.extend_from_slice(&chunk_enc);
+						} else {
+							// use encrypt_last for the last chunk
+							let chunk_enc = cipher
+								.encrypt_last(chunk)
+								.ok_or_internal_error("failed to encrypt chunk")?;
+							ret.extend_from_slice(&chunk_enc);
+							break;
+						}
+					}
+				}
+
+				Ok(ret.into())
+			}
+		}
+	}
+}
+
+fn parse_request_headers(
+	headers: &HeaderMap,
+	alg_header: &HeaderName,
+	key_header: &HeaderName,
+	md5_header: &HeaderName,
+) -> Result<Option<(Key<Aes256Gcm>, Md5Output)>, Error> {
+	let alg = headers.get(alg_header).map(HeaderValue::as_bytes);
+	let key = headers.get(key_header).map(HeaderValue::as_bytes);
+	let md5 = headers.get(md5_header).map(HeaderValue::as_bytes);
+
+	match alg {
+		Some(CUSTOMER_ALGORITHM_AES256) => {
+			use md5::{Digest, Md5};
+
+			let key_b64 =
+				key.ok_or_bad_request("Missing server-side-encryption-customer-key header")?;
+			let key_bytes: [u8; 32] = BASE64_STANDARD
+				.decode(&key_b64)
+				.ok_or_bad_request(
+					"Invalid server-side-encryption-customer-key header: invalid base64",
+				)?
+				.try_into()
+				.ok()
+				.ok_or_bad_request(
+					"Invalid server-side-encryption-customer-key header: invalid length",
+				)?;
+
+			let md5_b64 =
+				md5.ok_or_bad_request("Missing server-side-encryption-customer-key-md5 header")?;
+			let md5_bytes = BASE64_STANDARD.decode(&md5_b64).ok_or_bad_request(
+				"Invalid server-side-encryption-customer-key-md5 header: invalid bass64",
+			)?;
+
+			let mut hasher = Md5::new();
+			hasher.update(&key_bytes[..]);
+			let our_md5 = hasher.finalize();
+			if our_md5.as_slice() != md5_bytes.as_slice() {
+				return Err(Error::bad_request(
+					"Server-side encryption client key MD5 checksum does not match",
+				));
+			}
+
+			Ok(Some((key_bytes.into(), our_md5)))
+		}
+		Some(alg) => Err(Error::InvalidEncryptionAlgorithm(
+			String::from_utf8_lossy(alg).into_owned(),
+		)),
+		None => {
+			if key.is_some() || md5.is_some() {
+				Err(Error::bad_request(
+					"Unexpected server-side-encryption-customer-key{,-md5} header(s)",
+				))
+			} else {
+				Ok(None)
+			}
+		}
+	}
+}
+
+// ---- encrypt & decrypt streams ----
+
+#[pin_project::pin_project]
+struct DecryptStream {
+	#[pin]
+	stream: ByteStream,
+	done_reading: bool,
+	buf: BytesBuf,
+	key: Key<Aes256Gcm>,
+	state: DecryptStreamState,
+}
+
+enum DecryptStreamState {
+	Starting,
+	Running(DecryptorLE31<Aes256Gcm>),
+	Done,
+}
+
+impl DecryptStream {
+	fn new(stream: ByteStream, key: Key<Aes256Gcm>) -> Self {
+		Self {
+			stream,
+			done_reading: false,
+			buf: BytesBuf::new(),
+			key,
+			state: DecryptStreamState::Starting,
+		}
+	}
+}
+
+impl Stream for DecryptStream {
+	type Item = Result<Bytes, std::io::Error>;
+
+	fn poll_next(
+		self: Pin<&mut Self>,
+		cx: &mut task::Context<'_>,
+	) -> task::Poll<Option<Self::Item>> {
+		use std::task::Poll;
+
+		let mut this = self.project();
+
+		// The first bytes of the stream should contain the starting nonce.
+		// If we don't have a Running state, it means that we haven't
+		// yet read the nonce.
+		while matches!(this.state, DecryptStreamState::Starting) {
+			let nonce_size = StreamNonceSize::to_usize();
+			if let Some(nonce) = this.buf.take_exact(nonce_size) {
+				let nonce = Nonce::from_slice(nonce.as_ref());
+				*this.state = DecryptStreamState::Running(DecryptorLE31::new(&this.key, nonce));
+				break;
+			}
+
+			match futures::ready!(this.stream.as_mut().poll_next(cx)) {
+				Some(Ok(bytes)) => {
+					this.buf.extend(bytes);
+				}
+				Some(Err(e)) => {
+					return Poll::Ready(Some(Err(e)));
+				}
+				None => {
+					return Poll::Ready(Some(Err(std::io::Error::new(
+						std::io::ErrorKind::UnexpectedEof,
+						"Decrypt: unexpected EOF, could not read nonce",
+					))));
+				}
+			}
+		}
+
+		// Read at least one byte more than the encrypted chunk size
+		// (if possible), so that we know if we are decrypting the
+		// last chunk or not.
+		while !*this.done_reading && this.buf.len() <= STREAM_ENC_CYPER_CHUNK_SIZE {
+			match futures::ready!(this.stream.as_mut().poll_next(cx)) {
+				Some(Ok(bytes)) => {
+					this.buf.extend(bytes);
+				}
+				Some(Err(e)) => {
+					return Poll::Ready(Some(Err(e)));
+				}
+				None => {
+					*this.done_reading = true;
+					break;
+				}
+			}
+		}
+
+		if matches!(this.state, DecryptStreamState::Done) {
+			if !this.buf.is_empty() {
+				return Poll::Ready(Some(Err(std::io::Error::new(
+					std::io::ErrorKind::Other,
+					"Decrypt: unexpected bytes after last encrypted chunk",
+				))));
+			}
+			return Poll::Ready(None);
+		}
+
+		let res = if this.buf.len() > STREAM_ENC_CYPER_CHUNK_SIZE {
+			// we have strictly more bytes than the encrypted chunk size,
+			// so we know this is not the last
+			let DecryptStreamState::Running(ref mut cipher) = this.state else {
+				unreachable!()
+			};
+			let chunk = this.buf.take_exact(STREAM_ENC_CYPER_CHUNK_SIZE).unwrap();
+			let chunk_dec = cipher.decrypt_next(chunk.as_ref());
+			if let Ok(c) = &chunk_dec {
+				assert_eq!(c.len(), STREAM_ENC_PLAIN_CHUNK_SIZE);
+			}
+			chunk_dec
+		} else {
+			// We have one encrypted chunk size or less, even though we tried
+			// to read more, so this is the last chunk. Decrypt using the
+			// appropriate decrypt_last() function that then destroys the cipher.
+			let state = std::mem::replace(this.state, DecryptStreamState::Done);
+			let DecryptStreamState::Running(cipher) = state else {
+				unreachable!()
+			};
+			let chunk = this.buf.take_all();
+			cipher.decrypt_last(chunk.as_ref())
+		};
+
+		match res {
+			Ok(bytes) if bytes.is_empty() => Poll::Ready(None),
+			Ok(bytes) => Poll::Ready(Some(Ok(bytes.into()))),
+			Err(_) => Poll::Ready(Some(Err(std::io::Error::new(
+				std::io::ErrorKind::Other,
+				"Decryption failed",
+			)))),
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	use futures::stream::StreamExt;
+	use garage_net::stream::read_stream_to_end;
+
+	fn stream() -> ByteStream {
+		Box::pin(
+			futures::stream::iter(16usize..1024)
+				.map(|i| Ok(Bytes::from(vec![(i % 256) as u8; (i * 37) % 1024]))),
+		)
+	}
+
+	async fn test_block_enc(compression_level: Option<i32>) {
+		let enc = EncryptionParams::SseC {
+			client_key: Aes256Gcm::generate_key(&mut OsRng),
+			client_key_md5: Default::default(), // not needed
+			compression_level,
+		};
+
+		let block_plain = read_stream_to_end(stream()).await.unwrap().into_bytes();
+
+		let block_enc = enc.encrypt_block(block_plain.clone()).unwrap();
+
+		let block_dec =
+			enc.decrypt_block_stream(Box::pin(futures::stream::once(async { Ok(block_enc) })));
+		let block_dec = read_stream_to_end(block_dec).await.unwrap().into_bytes();
+
+		assert_eq!(block_plain, block_dec);
+		assert!(block_dec.len() > 128000);
+	}
+
+	#[tokio::test]
+	async fn test_encrypt_block() {
+		test_block_enc(None).await
+	}
+
+	#[tokio::test]
+	async fn test_encrypt_block_compressed() {
+		test_block_enc(Some(1)).await
+	}
+}
diff --git a/src/api/s3/error.rs b/src/api/s3/error.rs
index f86c19a6..2855e0b3 100644
--- a/src/api/s3/error.rs
+++ b/src/api/s3/error.rs
@@ -65,6 +65,14 @@ pub enum Error {
 	#[error(display = "Invalid HTTP range: {:?}", _0)]
 	InvalidRange(#[error(from)] (http_range::HttpRangeParseError, u64)),
 
+	/// The client sent a range header with invalid value
+	#[error(display = "Invalid encryption algorithm: {:?}, should be AES256", _0)]
+	InvalidEncryptionAlgorithm(String),
+
+	/// The client sent invalid XML data
+	#[error(display = "Invalid digest: {}", _0)]
+	InvalidDigest(String),
+
 	/// The client sent a request for an action not supported by garage
 	#[error(display = "Unimplemented action: {}", _0)]
 	NotImplemented(String),
@@ -125,7 +133,9 @@ impl Error {
 			Error::NotImplemented(_) => "NotImplemented",
 			Error::InvalidXml(_) => "MalformedXML",
 			Error::InvalidRange(_) => "InvalidRange",
+			Error::InvalidDigest(_) => "InvalidDigest",
 			Error::InvalidUtf8Str(_) | Error::InvalidUtf8String(_) => "InvalidRequest",
+			Error::InvalidEncryptionAlgorithm(_) => "InvalidEncryptionAlgorithmError",
 		}
 	}
 }
@@ -143,6 +153,8 @@ impl ApiError for Error {
 			| Error::InvalidPart
 			| Error::InvalidPartOrder
 			| Error::EntityTooSmall
+			| Error::InvalidDigest(_)
+			| Error::InvalidEncryptionAlgorithm(_)
 			| Error::InvalidXml(_)
 			| Error::InvalidUtf8Str(_)
 			| Error::InvalidUtf8String(_) => StatusCode::BAD_REQUEST,
diff --git a/src/api/s3/get.rs b/src/api/s3/get.rs
index ed996fb1..f5d3cf11 100644
--- a/src/api/s3/get.rs
+++ b/src/api/s3/get.rs
@@ -1,10 +1,12 @@
 //! Function related to GET and HEAD requests
+use std::collections::BTreeMap;
 use std::convert::TryInto;
 use std::sync::Arc;
 use std::time::{Duration, UNIX_EPOCH};
 
+use bytes::Bytes;
 use futures::future;
-use futures::stream::{self, StreamExt};
+use futures::stream::{self, Stream, StreamExt};
 use http::header::{
 	ACCEPT_RANGES, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE,
 	CONTENT_LENGTH, CONTENT_RANGE, CONTENT_TYPE, ETAG, EXPIRES, IF_MODIFIED_SINCE, IF_NONE_MATCH,
@@ -25,6 +27,8 @@ use garage_model::s3::version_table::*;
 
 use crate::helpers::*;
 use crate::s3::api_server::ResBody;
+use crate::s3::checksum::{add_checksum_response_headers, X_AMZ_CHECKSUM_MODE};
+use crate::s3::encryption::EncryptionParams;
 use crate::s3::error::*;
 
 const X_AMZ_MP_PARTS_COUNT: &str = "x-amz-mp-parts-count";
@@ -42,6 +46,9 @@ pub struct GetObjectOverrides {
 fn object_headers(
 	version: &ObjectVersion,
 	version_meta: &ObjectVersionMeta,
+	meta_inner: &ObjectVersionMetaInner,
+	encryption: EncryptionParams,
+	checksum_mode: ChecksumMode,
 ) -> http::response::Builder {
 	debug!("Version meta: {:?}", version_meta);
 
@@ -49,7 +56,6 @@ fn object_headers(
 	let date_str = httpdate::fmt_http_date(date);
 
 	let mut resp = Response::builder()
-		.header(CONTENT_TYPE, version_meta.headers.content_type.to_string())
 		.header(LAST_MODIFIED, date_str)
 		.header(ACCEPT_RANGES, "bytes".to_string());
 
@@ -57,10 +63,31 @@ fn object_headers(
 		resp = resp.header(ETAG, format!("\"{}\"", version_meta.etag));
 	}
 
-	for (k, v) in version_meta.headers.other.iter() {
-		resp = resp.header(k, v.to_string());
+	// When metadata is retrieved through the REST API, Amazon S3 combines headers that
+	// have the same name (ignoring case) into a comma-delimited list.
+	// See: https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingMetadata.html
+	let mut headers_by_name = BTreeMap::new();
+	for (name, value) in meta_inner.headers.iter() {
+		match headers_by_name.get_mut(name) {
+			None => {
+				headers_by_name.insert(name, vec![value.as_str()]);
+			}
+			Some(headers) => {
+				headers.push(value.as_str());
+			}
+		}
+	}
+
+	for (name, values) in headers_by_name {
+		resp = resp.header(name, values.join(","));
+	}
+
+	if checksum_mode.enabled {
+		resp = add_checksum_response_headers(&meta_inner.checksum, resp);
 	}
 
+	encryption.add_response_headers(&mut resp);
+
 	resp
 }
 
@@ -175,21 +202,33 @@ pub async fn handle_head_without_ctx(
 		return Ok(cached);
 	}
 
+	let (encryption, headers) =
+		EncryptionParams::check_decrypt(&garage, req.headers(), &version_meta.encryption)?;
+
+	let checksum_mode = checksum_mode(&req);
+
 	if let Some(pn) = part_number {
 		match version_data {
-			ObjectVersionData::Inline(_, bytes) => {
+			ObjectVersionData::Inline(_, _) => {
 				if pn != 1 {
 					return Err(Error::InvalidPart);
 				}
-				Ok(object_headers(object_version, version_meta)
-					.header(CONTENT_LENGTH, format!("{}", bytes.len()))
-					.header(
-						CONTENT_RANGE,
-						format!("bytes 0-{}/{}", bytes.len() - 1, bytes.len()),
-					)
-					.header(X_AMZ_MP_PARTS_COUNT, "1")
-					.status(StatusCode::PARTIAL_CONTENT)
-					.body(empty_body())?)
+				let bytes_len = version_meta.size;
+				Ok(object_headers(
+					object_version,
+					version_meta,
+					&headers,
+					encryption,
+					checksum_mode,
+				)
+				.header(CONTENT_LENGTH, format!("{}", bytes_len))
+				.header(
+					CONTENT_RANGE,
+					format!("bytes 0-{}/{}", bytes_len - 1, bytes_len),
+				)
+				.header(X_AMZ_MP_PARTS_COUNT, "1")
+				.status(StatusCode::PARTIAL_CONTENT)
+				.body(empty_body())?)
 			}
 			ObjectVersionData::FirstBlock(_, _) => {
 				let version = garage
@@ -201,28 +240,40 @@ pub async fn handle_head_without_ctx(
 				let (part_offset, part_end) =
 					calculate_part_bounds(&version, pn).ok_or(Error::InvalidPart)?;
 
-				Ok(object_headers(object_version, version_meta)
-					.header(CONTENT_LENGTH, format!("{}", part_end - part_offset))
-					.header(
-						CONTENT_RANGE,
-						format!(
-							"bytes {}-{}/{}",
-							part_offset,
-							part_end - 1,
-							version_meta.size
-						),
-					)
-					.header(X_AMZ_MP_PARTS_COUNT, format!("{}", version.n_parts()?))
-					.status(StatusCode::PARTIAL_CONTENT)
-					.body(empty_body())?)
+				Ok(object_headers(
+					object_version,
+					version_meta,
+					&headers,
+					encryption,
+					checksum_mode,
+				)
+				.header(CONTENT_LENGTH, format!("{}", part_end - part_offset))
+				.header(
+					CONTENT_RANGE,
+					format!(
+						"bytes {}-{}/{}",
+						part_offset,
+						part_end - 1,
+						version_meta.size
+					),
+				)
+				.header(X_AMZ_MP_PARTS_COUNT, format!("{}", version.n_parts()?))
+				.status(StatusCode::PARTIAL_CONTENT)
+				.body(empty_body())?)
 			}
 			_ => unreachable!(),
 		}
 	} else {
-		Ok(object_headers(object_version, version_meta)
-			.header(CONTENT_LENGTH, format!("{}", version_meta.size))
-			.status(StatusCode::OK)
-			.body(empty_body())?)
+		Ok(object_headers(
+			object_version,
+			version_meta,
+			&headers,
+			encryption,
+			checksum_mode,
+		)
+		.header(CONTENT_LENGTH, format!("{}", version_meta.size))
+		.status(StatusCode::OK)
+		.body(empty_body())?)
 	}
 }
 
@@ -273,23 +324,55 @@ pub async fn handle_get_without_ctx(
 		return Ok(cached);
 	}
 
+	let (enc, headers) =
+		EncryptionParams::check_decrypt(&garage, req.headers(), &last_v_meta.encryption)?;
+
+	let checksum_mode = checksum_mode(&req);
+
 	match (part_number, parse_range_header(req, last_v_meta.size)?) {
 		(Some(_), Some(_)) => Err(Error::bad_request(
 			"Cannot specify both partNumber and Range header",
 		)),
-		(Some(pn), None) => handle_get_part(garage, last_v, last_v_data, last_v_meta, pn).await,
+		(Some(pn), None) => {
+			handle_get_part(
+				garage,
+				last_v,
+				last_v_data,
+				last_v_meta,
+				enc,
+				&headers,
+				pn,
+				checksum_mode,
+			)
+			.await
+		}
 		(None, Some(range)) => {
 			handle_get_range(
 				garage,
 				last_v,
 				last_v_data,
 				last_v_meta,
+				enc,
+				&headers,
 				range.start,
 				range.start + range.length,
+				checksum_mode,
+			)
+			.await
+		}
+		(None, None) => {
+			handle_get_full(
+				garage,
+				last_v,
+				last_v_data,
+				last_v_meta,
+				enc,
+				&headers,
+				overrides,
+				checksum_mode,
 			)
 			.await
 		}
-		(None, None) => handle_get_full(garage, last_v, last_v_data, last_v_meta, overrides).await,
 	}
 }
 
@@ -298,17 +381,43 @@ async fn handle_get_full(
 	version: &ObjectVersion,
 	version_data: &ObjectVersionData,
 	version_meta: &ObjectVersionMeta,
+	encryption: EncryptionParams,
+	meta_inner: &ObjectVersionMetaInner,
 	overrides: GetObjectOverrides,
+	checksum_mode: ChecksumMode,
 ) -> Result<Response<ResBody>, Error> {
-	let mut resp_builder = object_headers(version, version_meta)
-		.header(CONTENT_LENGTH, format!("{}", version_meta.size))
-		.status(StatusCode::OK);
+	let mut resp_builder = object_headers(
+		version,
+		version_meta,
+		&meta_inner,
+		encryption,
+		checksum_mode,
+	)
+	.header(CONTENT_LENGTH, format!("{}", version_meta.size))
+	.status(StatusCode::OK);
 	getobject_override_headers(overrides, &mut resp_builder)?;
 
+	let stream = full_object_byte_stream(garage, version, version_data, encryption);
+
+	Ok(resp_builder.body(response_body_from_stream(stream))?)
+}
+
+pub fn full_object_byte_stream(
+	garage: Arc<Garage>,
+	version: &ObjectVersion,
+	version_data: &ObjectVersionData,
+	encryption: EncryptionParams,
+) -> ByteStream {
 	match &version_data {
 		ObjectVersionData::DeleteMarker => unreachable!(),
 		ObjectVersionData::Inline(_, bytes) => {
-			Ok(resp_builder.body(bytes_body(bytes.to_vec().into()))?)
+			let bytes = bytes.to_vec();
+			Box::pin(futures::stream::once(async move {
+				encryption
+					.decrypt_blob(&bytes)
+					.map(|x| Bytes::from(x.to_vec()))
+					.map_err(std_error_from_read_error)
+			}))
 		}
 		ObjectVersionData::FirstBlock(_, first_block_hash) => {
 			let (tx, rx) = mpsc::channel::<ByteStream>(2);
@@ -324,19 +433,18 @@ async fn handle_get_full(
 						garage2.version_table.get(&version_uuid, &EmptyKey).await
 					});
 
-					let stream_block_0 = garage
-						.block_manager
-						.rpc_get_block_streaming(&first_block_hash, Some(order_stream.order(0)))
+					let stream_block_0 = encryption
+						.get_block(&garage, &first_block_hash, Some(order_stream.order(0)))
 						.await?;
+
 					tx.send(stream_block_0)
 						.await
 						.ok_or_message("channel closed")?;
 
 					let version = version_fut.await.unwrap()?.ok_or(Error::NoSuchKey)?;
 					for (i, (_, vb)) in version.blocks.items().iter().enumerate().skip(1) {
-						let stream_block_i = garage
-							.block_manager
-							.rpc_get_block_streaming(&vb.hash, Some(order_stream.order(i as u64)))
+						let stream_block_i = encryption
+							.get_block(&garage, &vb.hash, Some(order_stream.order(i as u64)))
 							.await?;
 						tx.send(stream_block_i)
 							.await
@@ -354,8 +462,7 @@ async fn handle_get_full(
 				}
 			});
 
-			let body = response_body_from_block_stream(rx);
-			Ok(resp_builder.body(body)?)
+			Box::pin(tokio_stream::wrappers::ReceiverStream::new(rx).flatten())
 		}
 	}
 }
@@ -365,13 +472,16 @@ async fn handle_get_range(
 	version: &ObjectVersion,
 	version_data: &ObjectVersionData,
 	version_meta: &ObjectVersionMeta,
+	encryption: EncryptionParams,
+	meta_inner: &ObjectVersionMetaInner,
 	begin: u64,
 	end: u64,
+	checksum_mode: ChecksumMode,
 ) -> Result<Response<ResBody>, Error> {
 	// Here we do not use getobject_override_headers because we don't
 	// want to add any overridden headers (those should not be added
 	// when returning PARTIAL_CONTENT)
-	let resp_builder = object_headers(version, version_meta)
+	let resp_builder = object_headers(version, version_meta, meta_inner, encryption, checksum_mode)
 		.header(CONTENT_LENGTH, format!("{}", end - begin))
 		.header(
 			CONTENT_RANGE,
@@ -382,6 +492,7 @@ async fn handle_get_range(
 	match &version_data {
 		ObjectVersionData::DeleteMarker => unreachable!(),
 		ObjectVersionData::Inline(_meta, bytes) => {
+			let bytes = encryption.decrypt_blob(&bytes)?;
 			if end as usize <= bytes.len() {
 				let body = bytes_body(bytes[begin as usize..end as usize].to_vec().into());
 				Ok(resp_builder.body(body)?)
@@ -398,7 +509,8 @@ async fn handle_get_range(
 				.await?
 				.ok_or(Error::NoSuchKey)?;
 
-			let body = body_from_blocks_range(garage, version.blocks.items(), begin, end);
+			let body =
+				body_from_blocks_range(garage, encryption, version.blocks.items(), begin, end);
 			Ok(resp_builder.body(body)?)
 		}
 	}
@@ -409,17 +521,28 @@ async fn handle_get_part(
 	object_version: &ObjectVersion,
 	version_data: &ObjectVersionData,
 	version_meta: &ObjectVersionMeta,
+	encryption: EncryptionParams,
+	meta_inner: &ObjectVersionMetaInner,
 	part_number: u64,
+	checksum_mode: ChecksumMode,
 ) -> Result<Response<ResBody>, Error> {
 	// Same as for get_range, no getobject_override_headers
-	let resp_builder =
-		object_headers(object_version, version_meta).status(StatusCode::PARTIAL_CONTENT);
+	let resp_builder = object_headers(
+		object_version,
+		version_meta,
+		meta_inner,
+		encryption,
+		checksum_mode,
+	)
+	.status(StatusCode::PARTIAL_CONTENT);
 
 	match version_data {
 		ObjectVersionData::Inline(_, bytes) => {
 			if part_number != 1 {
 				return Err(Error::InvalidPart);
 			}
+			let bytes = encryption.decrypt_blob(&bytes)?;
+			assert_eq!(bytes.len() as u64, version_meta.size);
 			Ok(resp_builder
 				.header(CONTENT_LENGTH, format!("{}", bytes.len()))
 				.header(
@@ -427,7 +550,7 @@ async fn handle_get_part(
 					format!("bytes {}-{}/{}", 0, bytes.len() - 1, bytes.len()),
 				)
 				.header(X_AMZ_MP_PARTS_COUNT, "1")
-				.body(bytes_body(bytes.to_vec().into()))?)
+				.body(bytes_body(bytes.into_owned().into()))?)
 		}
 		ObjectVersionData::FirstBlock(_, _) => {
 			let version = garage
@@ -439,7 +562,8 @@ async fn handle_get_part(
 			let (begin, end) =
 				calculate_part_bounds(&version, part_number).ok_or(Error::InvalidPart)?;
 
-			let body = body_from_blocks_range(garage, version.blocks.items(), begin, end);
+			let body =
+				body_from_blocks_range(garage, encryption, version.blocks.items(), begin, end);
 
 			Ok(resp_builder
 				.header(CONTENT_LENGTH, format!("{}", end - begin))
@@ -492,8 +616,23 @@ fn calculate_part_bounds(v: &Version, part_number: u64) -> Option<(u64, u64)> {
 	None
 }
 
+struct ChecksumMode {
+	enabled: bool,
+}
+
+fn checksum_mode(req: &Request<impl Body>) -> ChecksumMode {
+	ChecksumMode {
+		enabled: req
+			.headers()
+			.get(X_AMZ_CHECKSUM_MODE)
+			.map(|x| x == "ENABLED")
+			.unwrap_or(false),
+	}
+}
+
 fn body_from_blocks_range(
 	garage: Arc<Garage>,
+	encryption: EncryptionParams,
 	all_blocks: &[(VersionBlockKey, VersionBlock)],
 	begin: u64,
 	end: u64,
@@ -523,12 +662,11 @@ fn body_from_blocks_range(
 
 	tokio::spawn(async move {
 		match async {
-			let garage = garage.clone();
 			for (i, (block, block_offset)) in blocks.iter().enumerate() {
-				let block_stream = garage
-					.block_manager
-					.rpc_get_block_streaming(&block.hash, Some(order_stream.order(i as u64)))
-					.await?
+				let block_stream = encryption
+					.get_block(&garage, &block.hash, Some(order_stream.order(i as u64)))
+					.await?;
+				let block_stream = block_stream
 					.scan(*block_offset, move |chunk_offset, chunk| {
 						let r = match chunk {
 							Ok(chunk_bytes) => {
@@ -588,19 +726,30 @@ fn body_from_blocks_range(
 }
 
 fn response_body_from_block_stream(rx: mpsc::Receiver<ByteStream>) -> ResBody {
-	let body_stream = tokio_stream::wrappers::ReceiverStream::new(rx)
-		.flatten()
-		.map(|x| {
-			x.map(hyper::body::Frame::data)
-				.map_err(|e| Error::from(garage_util::error::Error::from(e)))
-		});
+	let body_stream = tokio_stream::wrappers::ReceiverStream::new(rx).flatten();
+	response_body_from_stream(body_stream)
+}
+
+fn response_body_from_stream<S>(stream: S) -> ResBody
+where
+	S: Stream<Item = Result<Bytes, std::io::Error>> + Send + Sync + 'static,
+{
+	let body_stream = stream.map(|x| {
+		x.map(hyper::body::Frame::data)
+			.map_err(|e| Error::from(garage_util::error::Error::from(e)))
+	});
 	ResBody::new(http_body_util::StreamBody::new(body_stream))
 }
 
 fn error_stream_item<E: std::fmt::Display>(e: E) -> ByteStream {
-	let err = std::io::Error::new(
+	Box::pin(stream::once(future::ready(Err(std_error_from_read_error(
+		e,
+	)))))
+}
+
+fn std_error_from_read_error<E: std::fmt::Display>(e: E) -> std::io::Error {
+	std::io::Error::new(
 		std::io::ErrorKind::Other,
-		format!("Error while getting object data: {}", e),
-	);
-	Box::pin(stream::once(future::ready(Err(err))))
+		format!("Error while reading object data: {}", e),
+	)
 }
diff --git a/src/api/s3/list.rs b/src/api/s3/list.rs
index 302c03f4..648bace2 100644
--- a/src/api/s3/list.rs
+++ b/src/api/s3/list.rs
@@ -2,7 +2,7 @@ use std::collections::{BTreeMap, BTreeSet};
 use std::iter::{Iterator, Peekable};
 
 use base64::prelude::*;
-use hyper::Response;
+use hyper::{Request, Response};
 
 use garage_util::data::*;
 use garage_util::error::Error as GarageError;
@@ -15,7 +15,8 @@ use garage_table::EnumerationOrder;
 
 use crate::encoding::*;
 use crate::helpers::*;
-use crate::s3::api_server::ResBody;
+use crate::s3::api_server::{ReqBody, ResBody};
+use crate::s3::encryption::EncryptionParams;
 use crate::s3::error::*;
 use crate::s3::multipart as s3_multipart;
 use crate::s3::xml as s3_xml;
@@ -271,13 +272,21 @@ pub async fn handle_list_multipart_upload(
 
 pub async fn handle_list_parts(
 	ctx: ReqCtx,
+	req: Request<ReqBody>,
 	query: &ListPartsQuery,
 ) -> Result<Response<ResBody>, Error> {
 	debug!("ListParts {:?}", query);
 
 	let upload_id = s3_multipart::decode_upload_id(&query.upload_id)?;
 
-	let (_, _, mpu) = s3_multipart::get_upload(&ctx, &query.key, &upload_id).await?;
+	let (_, object_version, mpu) = s3_multipart::get_upload(&ctx, &query.key, &upload_id).await?;
+
+	let object_encryption = match object_version.state {
+		ObjectVersionState::Uploading { encryption, .. } => encryption,
+		_ => unreachable!(),
+	};
+	let encryption_res =
+		EncryptionParams::check_decrypt(&ctx.garage, req.headers(), &object_encryption);
 
 	let (info, next) = fetch_part_info(query, &mpu)?;
 
@@ -296,11 +305,40 @@ pub async fn handle_list_parts(
 		is_truncated: s3_xml::Value(format!("{}", next.is_some())),
 		parts: info
 			.iter()
-			.map(|part| s3_xml::PartItem {
-				etag: s3_xml::Value(format!("\"{}\"", part.etag)),
-				last_modified: s3_xml::Value(msec_to_rfc3339(part.timestamp)),
-				part_number: s3_xml::IntValue(part.part_number as i64),
-				size: s3_xml::IntValue(part.size as i64),
+			.map(|part| {
+				// hide checksum if object is encrypted and the decryption
+				// keys are not provided
+				let checksum = part.checksum.filter(|_| encryption_res.is_ok());
+				s3_xml::PartItem {
+					etag: s3_xml::Value(format!("\"{}\"", part.etag)),
+					last_modified: s3_xml::Value(msec_to_rfc3339(part.timestamp)),
+					part_number: s3_xml::IntValue(part.part_number as i64),
+					size: s3_xml::IntValue(part.size as i64),
+					checksum_crc32: match &checksum {
+						Some(ChecksumValue::Crc32(x)) => {
+							Some(s3_xml::Value(BASE64_STANDARD.encode(&x)))
+						}
+						_ => None,
+					},
+					checksum_crc32c: match &checksum {
+						Some(ChecksumValue::Crc32c(x)) => {
+							Some(s3_xml::Value(BASE64_STANDARD.encode(&x)))
+						}
+						_ => None,
+					},
+					checksum_sha1: match &checksum {
+						Some(ChecksumValue::Sha1(x)) => {
+							Some(s3_xml::Value(BASE64_STANDARD.encode(&x)))
+						}
+						_ => None,
+					},
+					checksum_sha256: match &checksum {
+						Some(ChecksumValue::Sha256(x)) => {
+							Some(s3_xml::Value(BASE64_STANDARD.encode(&x)))
+						}
+						_ => None,
+					},
+				}
 			})
 			.collect(),
 
@@ -346,6 +384,7 @@ struct PartInfo<'a> {
 	timestamp: u64,
 	part_number: u64,
 	size: u64,
+	checksum: Option<ChecksumValue>,
 }
 
 enum ExtractionResult {
@@ -486,6 +525,7 @@ fn fetch_part_info<'a>(
 				timestamp: pk.timestamp,
 				etag,
 				size,
+				checksum: p.checksum,
 			};
 			match parts.last_mut() {
 				Some(lastpart) if lastpart.part_number == pk.part_number => {
@@ -944,10 +984,13 @@ mod tests {
 			timestamp: TS,
 			state: ObjectVersionState::Uploading {
 				multipart: true,
-				headers: ObjectVersionHeaders {
-					content_type: "text/plain".to_string(),
-					other: BTreeMap::<String, String>::new(),
+				encryption: ObjectVersionEncryption::Plaintext {
+					inner: ObjectVersionMetaInner {
+						headers: vec![],
+						checksum: None,
+					},
 				},
+				checksum_algorithm: None,
 			},
 		}
 	}
@@ -1136,6 +1179,7 @@ mod tests {
 					version: uuid,
 					size: Some(3),
 					etag: Some("etag1".into()),
+					checksum: None,
 				},
 			),
 			(
@@ -1147,6 +1191,7 @@ mod tests {
 					version: uuid,
 					size: None,
 					etag: None,
+					checksum: None,
 				},
 			),
 			(
@@ -1158,6 +1203,7 @@ mod tests {
 					version: uuid,
 					size: Some(10),
 					etag: Some("etag2".into()),
+					checksum: None,
 				},
 			),
 			(
@@ -1169,6 +1215,7 @@ mod tests {
 					version: uuid,
 					size: Some(7),
 					etag: Some("etag3".into()),
+					checksum: None,
 				},
 			),
 			(
@@ -1180,6 +1227,7 @@ mod tests {
 					version: uuid,
 					size: Some(5),
 					etag: Some("etag4".into()),
+					checksum: None,
 				},
 			),
 		];
@@ -1218,12 +1266,14 @@ mod tests {
 					etag: "etag1",
 					timestamp: TS,
 					part_number: 1,
-					size: 3
+					size: 3,
+					checksum: None,
 				},
 				PartInfo {
 					etag: "etag2",
 					timestamp: TS,
 					part_number: 3,
+					checksum: None,
 					size: 10
 				},
 			]
@@ -1239,12 +1289,14 @@ mod tests {
 				PartInfo {
 					etag: "etag3",
 					timestamp: TS,
+					checksum: None,
 					part_number: 5,
 					size: 7
 				},
 				PartInfo {
 					etag: "etag4",
 					timestamp: TS,
+					checksum: None,
 					part_number: 8,
 					size: 5
 				},
@@ -1268,24 +1320,28 @@ mod tests {
 				PartInfo {
 					etag: "etag1",
 					timestamp: TS,
+					checksum: None,
 					part_number: 1,
 					size: 3
 				},
 				PartInfo {
 					etag: "etag2",
 					timestamp: TS,
+					checksum: None,
 					part_number: 3,
 					size: 10
 				},
 				PartInfo {
 					etag: "etag3",
 					timestamp: TS,
+					checksum: None,
 					part_number: 5,
 					size: 7
 				},
 				PartInfo {
 					etag: "etag4",
 					timestamp: TS,
+					checksum: None,
 					part_number: 8,
 					size: 5
 				},
diff --git a/src/api/s3/mod.rs b/src/api/s3/mod.rs
index cbdb94ab..b9bb1a6f 100644
--- a/src/api/s3/mod.rs
+++ b/src/api/s3/mod.rs
@@ -13,5 +13,7 @@ mod post_object;
 mod put;
 mod website;
 
+mod checksum;
+mod encryption;
 mod router;
 pub mod xml;
diff --git a/src/api/s3/multipart.rs b/src/api/s3/multipart.rs
index 1d5aeb26..3db3e8aa 100644
--- a/src/api/s3/multipart.rs
+++ b/src/api/s3/multipart.rs
@@ -1,9 +1,10 @@
 use std::collections::HashMap;
+use std::convert::TryInto;
 use std::sync::Arc;
 
+use base64::prelude::*;
 use futures::prelude::*;
 use hyper::{Request, Response};
-use md5::{Digest as Md5Digest, Md5};
 
 use garage_table::*;
 use garage_util::data::*;
@@ -16,6 +17,8 @@ use garage_model::s3::version_table::*;
 
 use crate::helpers::*;
 use crate::s3::api_server::{ReqBody, ResBody};
+use crate::s3::checksum::*;
+use crate::s3::encryption::EncryptionParams;
 use crate::s3::error::*;
 use crate::s3::put::*;
 use crate::s3::xml as s3_xml;
@@ -40,6 +43,16 @@ pub async fn handle_create_multipart_upload(
 	let timestamp = next_timestamp(existing_object.as_ref());
 
 	let headers = get_headers(req.headers())?;
+	let meta = ObjectVersionMetaInner {
+		headers,
+		checksum: None,
+	};
+
+	// Determine whether object should be encrypted, and if so the key
+	let encryption = EncryptionParams::new_from_headers(&garage, req.headers())?;
+	let object_encryption = encryption.encrypt_meta(meta)?;
+
+	let checksum_algorithm = request_checksum_algorithm(req.headers())?;
 
 	// Create object in object table
 	let object_version = ObjectVersion {
@@ -47,7 +60,8 @@ pub async fn handle_create_multipart_upload(
 		timestamp,
 		state: ObjectVersionState::Uploading {
 			multipart: true,
-			headers,
+			encryption: object_encryption,
+			checksum_algorithm,
 		},
 	};
 	let object = Object::new(*bucket_id, key.to_string(), vec![object_version]);
@@ -68,7 +82,9 @@ pub async fn handle_create_multipart_upload(
 	};
 	let xml = s3_xml::to_xml_with_header(&result)?;
 
-	Ok(Response::new(string_body(xml)))
+	let mut resp = Response::builder();
+	encryption.add_response_headers(&mut resp);
+	Ok(resp.body(string_body(xml))?)
 }
 
 pub async fn handle_put_part(
@@ -83,20 +99,37 @@ pub async fn handle_put_part(
 
 	let upload_id = decode_upload_id(upload_id)?;
 
-	let content_md5 = match req.headers().get("content-md5") {
-		Some(x) => Some(x.to_str()?.to_string()),
-		None => None,
+	let expected_checksums = ExpectedChecksums {
+		md5: match req.headers().get("content-md5") {
+			Some(x) => Some(x.to_str()?.to_string()),
+			None => None,
+		},
+		sha256: content_sha256,
+		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 stream = body_stream(req.into_body());
+	let (req_head, req_body) = req.into_parts();
+	let stream = body_stream(req_body);
 	let mut chunker = StreamChunker::new(stream, garage.config.block_size);
 
-	let ((_, _, mut mpu), first_block) =
+	let ((_, object_version, mut mpu), first_block) =
 		futures::try_join!(get_upload(&ctx, &key, &upload_id), chunker.next(),)?;
 
+	// Check encryption params
+	let (object_encryption, checksum_algorithm) = match object_version.state {
+		ObjectVersionState::Uploading {
+			encryption,
+			checksum_algorithm,
+			..
+		} => (encryption, checksum_algorithm),
+		_ => unreachable!(),
+	};
+	let (encryption, _) =
+		EncryptionParams::check_decrypt(&garage, &req_head.headers, &object_encryption)?;
+
 	// Check object is valid and part can be accepted
 	let first_block = first_block.ok_or_bad_request("Empty body")?;
 
@@ -122,7 +155,9 @@ pub async fn handle_put_part(
 		mpu_part_key,
 		MpuPart {
 			version: version_uuid,
+			// all these are filled in later, at the end of this function
 			etag: None,
+			checksum: None,
 			size: None,
 		},
 	);
@@ -136,24 +171,31 @@ pub async fn handle_put_part(
 	garage.version_table.insert(&version).await?;
 
 	// Copy data to version
-	let (total_size, data_md5sum, data_sha256sum, _) =
-		read_and_put_blocks(&ctx, &version, part_number, first_block, &mut chunker).await?;
+	let checksummer =
+		Checksummer::init(&expected_checksums, !encryption.is_encrypted()).add(checksum_algorithm);
+	let (total_size, checksums, _) = read_and_put_blocks(
+		&ctx,
+		&version,
+		encryption,
+		part_number,
+		first_block,
+		&mut chunker,
+		checksummer,
+	)
+	.await?;
 
 	// Verify that checksums map
-	ensure_checksum_matches(
-		data_md5sum.as_slice(),
-		data_sha256sum,
-		content_md5.as_deref(),
-		content_sha256,
-	)?;
+	checksums.verify(&expected_checksums)?;
 
 	// Store part etag in version
-	let data_md5sum_hex = hex::encode(data_md5sum);
+	let etag = encryption.etag_from_md5(&checksums.md5);
+
 	mpu.parts.put(
 		mpu_part_key,
 		MpuPart {
 			version: version_uuid,
-			etag: Some(data_md5sum_hex.clone()),
+			etag: Some(etag.clone()),
+			checksum: checksums.extract(checksum_algorithm),
 			size: Some(total_size),
 		},
 	);
@@ -163,11 +205,10 @@ pub async fn handle_put_part(
 	// We won't have to clean up on drop.
 	interrupted_cleanup.cancel();
 
-	let response = Response::builder()
-		.header("ETag", format!("\"{}\"", data_md5sum_hex))
-		.body(empty_body())
-		.unwrap();
-	Ok(response)
+	let mut resp = Response::builder().header("ETag", format!("\"{}\"", etag));
+	encryption.add_response_headers(&mut resp);
+	let resp = add_checksum_response_headers(&expected_checksums.extra, resp);
+	Ok(resp.body(empty_body())?)
 }
 
 struct InterruptedCleanup(Option<InterruptedCleanupInner>);
@@ -214,10 +255,11 @@ pub async fn handle_complete_multipart_upload(
 		bucket_name,
 		..
 	} = &ctx;
+	let (req_head, req_body) = req.into_parts();
+
+	let expected_checksum = request_checksum_value(&req_head.headers)?;
 
-	let body = http_body_util::BodyExt::collect(req.into_body())
-		.await?
-		.to_bytes();
+	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[..])?;
@@ -241,8 +283,12 @@ pub async fn handle_complete_multipart_upload(
 		return Err(Error::bad_request("No data was uploaded"));
 	}
 
-	let headers = match object_version.state {
-		ObjectVersionState::Uploading { headers, .. } => headers,
+	let (object_encryption, checksum_algorithm) = match object_version.state {
+		ObjectVersionState::Uploading {
+			encryption,
+			checksum_algorithm,
+			..
+		} => (encryption, checksum_algorithm),
 		_ => unreachable!(),
 	};
 
@@ -270,6 +316,13 @@ pub async fn handle_complete_multipart_upload(
 	for req_part in body_list_of_parts.iter() {
 		match have_parts.get(&req_part.part_number) {
 			Some(part) if part.etag.as_ref() == Some(&req_part.etag) && part.size.is_some() => {
+				// alternative version: if req_part.checksum.is_some() && part.checksum != req_part.checksum {
+				if part.checksum != req_part.checksum {
+					return Err(Error::InvalidDigest(format!(
+						"Invalid checksum for part {}: in request = {:?}, uploaded part = {:?}",
+						req_part.part_number, req_part.checksum, part.checksum
+					)));
+				}
 				parts.push(*part)
 			}
 			_ => return Err(Error::InvalidPart),
@@ -317,18 +370,23 @@ pub async fn handle_complete_multipart_upload(
 	});
 	garage.block_ref_table.insert_many(block_refs).await?;
 
-	// Calculate etag of final object
+	// Calculate checksum and etag of final object
 	// To understand how etags are calculated, read more here:
+	// https://docs.aws.amazon.com/AmazonS3/latest/userguide/checking-object-integrity.html
 	// https://teppen.io/2018/06/23/aws_s3_etags/
-	let mut etag_md5_hasher = Md5::new();
+	let mut checksummer = MultipartChecksummer::init(checksum_algorithm);
 	for part in parts.iter() {
-		etag_md5_hasher.update(part.etag.as_ref().unwrap().as_bytes());
+		checksummer.update(part.etag.as_ref().unwrap(), part.checksum)?;
 	}
-	let etag = format!(
-		"{}-{}",
-		hex::encode(etag_md5_hasher.finalize()),
-		parts.len()
-	);
+	let (checksum_md5, checksum_extra) = checksummer.finalize();
+
+	if expected_checksum.is_some() && checksum_extra != expected_checksum {
+		return Err(Error::InvalidDigest(
+			"Failed to validate x-amz-checksum-*".into(),
+		));
+	}
+
+	let etag = format!("{}-{}", hex::encode(&checksum_md5[..]), parts.len());
 
 	// Calculate total size of final object
 	let total_size = parts.iter().map(|x| x.size.unwrap()).sum();
@@ -341,10 +399,24 @@ pub async fn handle_complete_multipart_upload(
 		return Err(e);
 	}
 
+	// If there is a checksum algorithm, update metadata with checksum
+	let object_encryption = match checksum_algorithm {
+		None => object_encryption,
+		Some(_) => {
+			let (encryption, meta) =
+				EncryptionParams::check_decrypt(&garage, &req_head.headers, &object_encryption)?;
+			let new_meta = ObjectVersionMetaInner {
+				headers: meta.into_owned().headers,
+				checksum: checksum_extra,
+			};
+			encryption.encrypt_meta(new_meta)?
+		}
+	};
+
 	// Write final object version
 	object_version.state = ObjectVersionState::Complete(ObjectVersionData::FirstBlock(
 		ObjectVersionMeta {
-			headers,
+			encryption: object_encryption,
 			size: total_size,
 			etag: etag.clone(),
 		},
@@ -361,10 +433,28 @@ pub async fn handle_complete_multipart_upload(
 		bucket: s3_xml::Value(bucket_name.to_string()),
 		key: s3_xml::Value(key),
 		etag: s3_xml::Value(format!("\"{}\"", etag)),
+		checksum_crc32: match &checksum_extra {
+			Some(ChecksumValue::Crc32(x)) => Some(s3_xml::Value(BASE64_STANDARD.encode(&x))),
+			_ => None,
+		},
+		checksum_crc32c: match &checksum_extra {
+			Some(ChecksumValue::Crc32c(x)) => Some(s3_xml::Value(BASE64_STANDARD.encode(&x))),
+			_ => None,
+		},
+		checksum_sha1: match &checksum_extra {
+			Some(ChecksumValue::Sha1(x)) => Some(s3_xml::Value(BASE64_STANDARD.encode(&x))),
+			_ => None,
+		},
+		checksum_sha256: match &checksum_extra {
+			Some(ChecksumValue::Sha256(x)) => Some(s3_xml::Value(BASE64_STANDARD.encode(&x))),
+			_ => None,
+		},
 	};
 	let xml = s3_xml::to_xml_with_header(&result)?;
 
-	Ok(Response::new(string_body(xml)))
+	let resp = Response::builder();
+	let resp = add_checksum_response_headers(&expected_checksum, resp);
+	Ok(resp.body(string_body(xml))?)
 }
 
 pub async fn handle_abort_multipart_upload(
@@ -433,6 +523,7 @@ pub fn decode_upload_id(id: &str) -> Result<Uuid, Error> {
 struct CompleteMultipartUploadPart {
 	etag: String,
 	part_number: u64,
+	checksum: Option<ChecksumValue>,
 }
 
 fn parse_complete_multipart_upload_body(
@@ -458,9 +549,41 @@ fn parse_complete_multipart_upload_body(
 				.children()
 				.find(|e| e.has_tag_name("PartNumber"))?
 				.text()?;
+			let checksum = if let Some(crc32) =
+				item.children().find(|e| e.has_tag_name("ChecksumCRC32"))
+			{
+				Some(ChecksumValue::Crc32(
+					BASE64_STANDARD.decode(crc32.text()?).ok()?[..]
+						.try_into()
+						.ok()?,
+				))
+			} else if let Some(crc32c) = item.children().find(|e| e.has_tag_name("ChecksumCRC32C"))
+			{
+				Some(ChecksumValue::Crc32c(
+					BASE64_STANDARD.decode(crc32c.text()?).ok()?[..]
+						.try_into()
+						.ok()?,
+				))
+			} else if let Some(sha1) = item.children().find(|e| e.has_tag_name("ChecksumSHA1")) {
+				Some(ChecksumValue::Sha1(
+					BASE64_STANDARD.decode(sha1.text()?).ok()?[..]
+						.try_into()
+						.ok()?,
+				))
+			} else if let Some(sha256) = item.children().find(|e| e.has_tag_name("ChecksumSHA256"))
+			{
+				Some(ChecksumValue::Sha256(
+					BASE64_STANDARD.decode(sha256.text()?).ok()?[..]
+						.try_into()
+						.ok()?,
+				))
+			} else {
+				None
+			};
 			parts.push(CompleteMultipartUploadPart {
 				etag: etag.trim_matches('"').to_string(),
 				part_number: part_number.parse().ok()?,
+				checksum,
 			});
 		} else {
 			return None;
diff --git a/src/api/s3/post_object.rs b/src/api/s3/post_object.rs
index 66f8174c..2c106b3b 100644
--- a/src/api/s3/post_object.rs
+++ b/src/api/s3/post_object.rs
@@ -14,12 +14,15 @@ use multer::{Constraints, Multipart, SizeLimit};
 use serde::Deserialize;
 
 use garage_model::garage::Garage;
+use garage_model::s3::object_table::*;
 
 use crate::helpers::*;
 use crate::s3::api_server::ResBody;
+use crate::s3::checksum::*;
 use crate::s3::cors::*;
+use crate::s3::encryption::EncryptionParams;
 use crate::s3::error::*;
-use crate::s3::put::{get_headers, save_stream};
+use crate::s3::put::{get_headers, save_stream, ChecksumMode};
 use crate::s3::xml as s3_xml;
 use crate::signature::payload::{verify_v4, Authorization};
 
@@ -48,13 +51,17 @@ pub async fn handle_post_object(
 	let mut multipart = Multipart::with_constraints(stream, boundary, constraints);
 
 	let mut params = HeaderMap::new();
-	let field = loop {
+	let file_field = loop {
 		let field = if let Some(field) = multipart.next_field().await? {
 			field
 		} else {
 			return Err(Error::bad_request("Request did not contain a file"));
 		};
-		let name: HeaderName = if let Some(Ok(name)) = field.name().map(TryInto::try_into) {
+		let name: HeaderName = if let Some(Ok(name)) = field
+			.name()
+			.map(str::to_ascii_lowercase)
+			.map(TryInto::try_into)
+		{
 			name
 		} else {
 			continue;
@@ -96,7 +103,7 @@ pub async fn handle_post_object(
 
 	let key = if key.contains("${filename}") {
 		// if no filename is provided, don't replace. This matches the behavior of AWS.
-		if let Some(filename) = field.file_name() {
+		if let Some(filename) = file_field.file_name() {
 			key.replace("${filename}", filename)
 		} else {
 			key.to_owned()
@@ -143,9 +150,8 @@ pub async fn handle_post_object(
 	let mut conditions = decoded_policy.into_conditions()?;
 
 	for (param_key, value) in params.iter() {
-		let mut param_key = param_key.to_string();
-		param_key.make_ascii_lowercase();
-		match param_key.as_str() {
+		let param_key = param_key.as_str();
+		match param_key {
 			"policy" | "x-amz-signature" => (), // this is always accepted, as it's required to validate other fields
 			"content-type" => {
 				let conds = conditions.params.remove("content-type").ok_or_else(|| {
@@ -190,7 +196,7 @@ pub async fn handle_post_object(
 					// how aws seems to behave.
 					continue;
 				}
-				let conds = conditions.params.remove(&param_key).ok_or_else(|| {
+				let conds = conditions.params.remove(param_key).ok_or_else(|| {
 					Error::bad_request(format!("Key '{}' is not allowed in policy", param_key))
 				})?;
 				for cond in conds {
@@ -218,8 +224,24 @@ pub async fn handle_post_object(
 
 	let headers = get_headers(&params)?;
 
-	let stream = field.map(|r| r.map_err(Into::into));
+	let expected_checksums = ExpectedChecksums {
+		md5: params
+			.get("content-md5")
+			.map(HeaderValue::to_str)
+			.transpose()?
+			.map(str::to_string),
+		sha256: None,
+		extra: request_checksum_algorithm_value(&params)?,
+	};
+
+	let meta = ObjectVersionMetaInner {
+		headers,
+		checksum: expected_checksums.extra,
+	};
+
+	let encryption = EncryptionParams::new_from_headers(&garage, &params)?;
 
+	let stream = file_field.map(|r| r.map_err(Into::into));
 	let ctx = ReqCtx {
 		garage,
 		bucket_id,
@@ -228,17 +250,17 @@ pub async fn handle_post_object(
 		api_key,
 	};
 
-	let (_, md5) = save_stream(
+	let res = save_stream(
 		&ctx,
-		headers,
+		meta,
+		encryption,
 		StreamLimiter::new(stream, conditions.content_length),
 		&key,
-		None,
-		None,
+		ChecksumMode::Verify(&expected_checksums),
 	)
 	.await?;
 
-	let etag = format!("\"{}\"", md5);
+	let etag = format!("\"{}\"", res.etag);
 
 	let mut resp = if let Some(mut target) = params
 		.get("success_action_redirect")
@@ -252,11 +274,12 @@ pub async fn handle_post_object(
 			.append_pair("key", &key)
 			.append_pair("etag", &etag);
 		let target = target.to_string();
-		Response::builder()
+		let mut resp = Response::builder()
 			.status(StatusCode::SEE_OTHER)
 			.header(header::LOCATION, target.clone())
-			.header(header::ETAG, etag)
-			.body(string_body(target))?
+			.header(header::ETAG, etag);
+		encryption.add_response_headers(&mut resp);
+		resp.body(string_body(target))?
 	} else {
 		let path = head
 			.uri
@@ -283,9 +306,10 @@ pub async fn handle_post_object(
 			.get("success_action_status")
 			.and_then(|h| h.to_str().ok())
 			.unwrap_or("204");
-		let builder = Response::builder()
+		let mut builder = Response::builder()
 			.header(header::LOCATION, location.clone())
 			.header(header::ETAG, etag.clone());
+		encryption.add_response_headers(&mut builder);
 		match action {
 			"200" => builder.status(StatusCode::OK).body(empty_body())?,
 			"201" => {
diff --git a/src/api/s3/put.rs b/src/api/s3/put.rs
index 685cca80..1e3b1b44 100644
--- a/src/api/s3/put.rs
+++ b/src/api/s3/put.rs
@@ -1,12 +1,9 @@
-use std::collections::{BTreeMap, HashMap};
+use std::collections::HashMap;
 use std::sync::Arc;
 
-use base64::prelude::*;
 use futures::prelude::*;
 use futures::stream::FuturesOrdered;
 use futures::try_join;
-use md5::{digest::generic_array::*, Digest as Md5Digest, Md5};
-use sha2::Sha256;
 
 use tokio::sync::mpsc;
 
@@ -22,7 +19,6 @@ use opentelemetry::{
 use garage_net::bytes_buf::BytesBuf;
 use garage_rpc::rpc_helper::OrderTag;
 use garage_table::*;
-use garage_util::async_hash::*;
 use garage_util::data::*;
 use garage_util::error::Error as GarageError;
 use garage_util::time::*;
@@ -36,10 +32,24 @@ use garage_model::s3::version_table::*;
 
 use crate::helpers::*;
 use crate::s3::api_server::{ReqBody, ResBody};
+use crate::s3::checksum::*;
+use crate::s3::encryption::EncryptionParams;
 use crate::s3::error::*;
 
 const PUT_BLOCKS_MAX_PARALLEL: usize = 3;
 
+pub(crate) struct SaveStreamResult {
+	pub(crate) version_uuid: Uuid,
+	pub(crate) version_timestamp: u64,
+	/// Etag WITHOUT THE QUOTES (just the hex value)
+	pub(crate) etag: String,
+}
+
+pub(crate) enum ChecksumMode<'a> {
+	Verify(&'a ExpectedChecksums),
+	Calculate(Option<ChecksumAlgorithm>),
+}
+
 pub async fn handle_put(
 	ctx: ReqCtx,
 	req: Request<ReqBody>,
@@ -50,26 +60,51 @@ pub async fn handle_put(
 	let headers = get_headers(req.headers())?;
 	debug!("Object headers: {:?}", headers);
 
-	let content_md5 = match req.headers().get("content-md5") {
-		Some(x) => Some(x.to_str()?.to_string()),
-		None => None,
+	let expected_checksums = ExpectedChecksums {
+		md5: match req.headers().get("content-md5") {
+			Some(x) => Some(x.to_str()?.to_string()),
+			None => None,
+		},
+		sha256: content_sha256,
+		extra: request_checksum_value(req.headers())?,
+	};
+
+	let meta = ObjectVersionMetaInner {
+		headers,
+		checksum: expected_checksums.extra,
 	};
 
+	// 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());
 
-	save_stream(&ctx, headers, stream, key, content_md5, content_sha256)
-		.await
-		.map(|(uuid, md5)| put_response(uuid, md5))
+	let res = save_stream(
+		&ctx,
+		meta,
+		encryption,
+		stream,
+		key,
+		ChecksumMode::Verify(&expected_checksums),
+	)
+	.await?;
+
+	let mut resp = Response::builder()
+		.header("x-amz-version-id", hex::encode(res.version_uuid))
+		.header("ETag", format!("\"{}\"", res.etag));
+	encryption.add_response_headers(&mut resp);
+	let resp = add_checksum_response_headers(&expected_checksums.extra, resp);
+	Ok(resp.body(empty_body())?)
 }
 
 pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	ctx: &ReqCtx,
-	headers: ObjectVersionHeaders,
+	mut meta: ObjectVersionMetaInner,
+	encryption: EncryptionParams,
 	body: S,
 	key: &String,
-	content_md5: Option<String>,
-	content_sha256: Option<FixedBytes32>,
-) -> Result<(Uuid, String), Error> {
+	checksum_mode: ChecksumMode<'_>,
+) -> Result<SaveStreamResult, Error> {
 	let ReqCtx {
 		garage, bucket_id, ..
 	} = ctx;
@@ -86,43 +121,55 @@ 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 {
+		ChecksumMode::Verify(expected) => Checksummer::init(expected, !encryption.is_encrypted()),
+		ChecksumMode::Calculate(algo) => {
+			Checksummer::init(&Default::default(), !encryption.is_encrypted()).add(algo)
+		}
+	};
+
 	// If body is small enough, store it directly in the object table
 	// as "inline data". We can then return immediately.
 	if first_block.len() < INLINE_THRESHOLD {
-		let mut md5sum = Md5::new();
-		md5sum.update(&first_block[..]);
-		let data_md5sum = md5sum.finalize();
-		let data_md5sum_hex = hex::encode(data_md5sum);
+		checksummer.update(&first_block);
+		let checksums = checksummer.finalize();
 
-		let data_sha256sum = sha256sum(&first_block[..]);
-		let size = first_block.len() as u64;
-
-		ensure_checksum_matches(
-			data_md5sum.as_slice(),
-			data_sha256sum,
-			content_md5.as_deref(),
-			content_sha256,
-		)?;
+		match checksum_mode {
+			ChecksumMode::Verify(expected) => {
+				checksums.verify(&expected)?;
+			}
+			ChecksumMode::Calculate(algo) => {
+				meta.checksum = checksums.extract(algo);
+			}
+		};
 
+		let size = first_block.len() as u64;
 		check_quotas(ctx, size, existing_object.as_ref()).await?;
 
+		let etag = encryption.etag_from_md5(&checksums.md5);
+		let inline_data = encryption.encrypt_blob(&first_block)?.to_vec();
+
 		let object_version = ObjectVersion {
 			uuid: version_uuid,
 			timestamp: version_timestamp,
 			state: ObjectVersionState::Complete(ObjectVersionData::Inline(
 				ObjectVersionMeta {
-					headers,
+					encryption: encryption.encrypt_meta(meta)?,
 					size,
-					etag: data_md5sum_hex.clone(),
+					etag: etag.clone(),
 				},
-				first_block.to_vec(),
+				inline_data,
 			)),
 		};
 
 		let object = Object::new(*bucket_id, key.into(), vec![object_version]);
 		garage.object_table.insert(&object).await?;
 
-		return Ok((version_uuid, data_md5sum_hex));
+		return Ok(SaveStreamResult {
+			version_uuid,
+			version_timestamp,
+			etag,
+		});
 	}
 
 	// The following consists in many steps that can each fail.
@@ -142,7 +189,8 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 		uuid: version_uuid,
 		timestamp: version_timestamp,
 		state: ObjectVersionState::Uploading {
-			headers: headers.clone(),
+			encryption: encryption.encrypt_meta(meta.clone())?,
+			checksum_algorithm: None, // don't care; overwritten later
 			multipart: false,
 		},
 	};
@@ -163,26 +211,39 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	);
 	garage.version_table.insert(&version).await?;
 
-	// Transfer data and verify checksum
-	let (total_size, data_md5sum, data_sha256sum, first_block_hash) =
-		read_and_put_blocks(ctx, &version, 1, first_block, &mut chunker).await?;
-
-	ensure_checksum_matches(
-		data_md5sum.as_slice(),
-		data_sha256sum,
-		content_md5.as_deref(),
-		content_sha256,
-	)?;
+	// Transfer data
+	let (total_size, checksums, first_block_hash) = read_and_put_blocks(
+		ctx,
+		&version,
+		encryption,
+		1,
+		first_block,
+		&mut chunker,
+		checksummer,
+	)
+	.await?;
+
+	// Verify checksums are ok / add calculated checksum to metadata
+	match checksum_mode {
+		ChecksumMode::Verify(expected) => {
+			checksums.verify(&expected)?;
+		}
+		ChecksumMode::Calculate(algo) => {
+			meta.checksum = checksums.extract(algo);
+		}
+	};
 
+	// Verify quotas are respsected
 	check_quotas(ctx, total_size, existing_object.as_ref()).await?;
 
 	// Save final object state, marked as Complete
-	let md5sum_hex = hex::encode(data_md5sum);
+	let etag = encryption.etag_from_md5(&checksums.md5);
+
 	object_version.state = ObjectVersionState::Complete(ObjectVersionData::FirstBlock(
 		ObjectVersionMeta {
-			headers,
+			encryption: encryption.encrypt_meta(meta)?,
 			size: total_size,
-			etag: md5sum_hex.clone(),
+			etag: etag.clone(),
 		},
 		first_block_hash,
 	));
@@ -193,34 +254,11 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	// We won't have to clean up on drop.
 	interrupted_cleanup.cancel();
 
-	Ok((version_uuid, md5sum_hex))
-}
-
-/// Validate MD5 sum against content-md5 header
-/// and sha256sum against signed content-sha256
-pub(crate) fn ensure_checksum_matches(
-	data_md5sum: &[u8],
-	data_sha256sum: garage_util::data::FixedBytes32,
-	content_md5: Option<&str>,
-	content_sha256: Option<garage_util::data::FixedBytes32>,
-) -> Result<(), Error> {
-	if let Some(expected_sha256) = content_sha256 {
-		if expected_sha256 != data_sha256sum {
-			return Err(Error::bad_request(
-				"Unable to validate x-amz-content-sha256",
-			));
-		} else {
-			trace!("Successfully validated x-amz-content-sha256");
-		}
-	}
-	if let Some(expected_md5) = content_md5 {
-		if expected_md5.trim_matches('"') != BASE64_STANDARD.encode(data_md5sum) {
-			return Err(Error::bad_request("Unable to validate content-md5"));
-		} else {
-			trace!("Successfully validated content-md5");
-		}
-	}
-	Ok(())
+	Ok(SaveStreamResult {
+		version_uuid,
+		version_timestamp,
+		etag,
+	})
 }
 
 /// Check that inserting this object with this size doesn't exceed bucket quotas
@@ -248,7 +286,7 @@ pub(crate) async fn check_quotas(
 		.await?;
 
 	let counters = counters
-		.map(|x| x.filtered_values(&garage.system.ring.borrow()))
+		.map(|x| x.filtered_values(&garage.system.cluster_layout()))
 		.unwrap_or_default();
 
 	let (prev_cnt_obj, prev_cnt_size) = match prev_object {
@@ -290,10 +328,12 @@ pub(crate) async fn check_quotas(
 pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
 	ctx: &ReqCtx,
 	version: &Version,
+	encryption: EncryptionParams,
 	part_number: u64,
 	first_block: Bytes,
 	chunker: &mut StreamChunker<S>,
-) -> Result<(u64, GenericArray<u8, typenum::U16>, Hash, Hash), Error> {
+	checksummer: Checksummer,
+) -> Result<(u64, Checksums, Hash), Error> {
 	let tracer = opentelemetry::global::tracer("garage");
 
 	let (block_tx, mut block_rx) = mpsc::channel::<Result<Bytes, Error>>(2);
@@ -321,20 +361,20 @@ pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> +
 
 	let (block_tx2, mut block_rx2) = mpsc::channel::<Result<Bytes, Error>>(1);
 	let hash_stream = async {
-		let md5hasher = AsyncHasher::<Md5>::new();
-		let sha256hasher = AsyncHasher::<Sha256>::new();
+		let mut checksummer = checksummer;
 		while let Some(next) = block_rx.recv().await {
 			match next {
 				Ok(block) => {
 					block_tx2.send(Ok(block.clone())).await?;
-					futures::future::join(
-						md5hasher.update(block.clone()),
-						sha256hasher.update(block.clone()),
-					)
+					checksummer = tokio::task::spawn_blocking(move || {
+						checksummer.update(&block);
+						checksummer
+					})
 					.with_context(Context::current_with_span(
 						tracer.start("Hash block (md5, sha256)"),
 					))
-					.await;
+					.await
+					.unwrap()
 				}
 				Err(e) => {
 					block_tx2.send(Err(e)).await?;
@@ -343,27 +383,38 @@ pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> +
 			}
 		}
 		drop(block_tx2);
-		Ok::<_, mpsc::error::SendError<_>>(futures::join!(
-			md5hasher.finalize(),
-			sha256hasher.finalize()
-		))
+		Ok::<_, mpsc::error::SendError<_>>(checksummer)
 	};
 
-	let (block_tx3, mut block_rx3) = mpsc::channel::<Result<(Bytes, Hash), Error>>(1);
-	let hash_blocks = async {
+	let (block_tx3, mut block_rx3) = mpsc::channel::<Result<(Bytes, u64, Hash), Error>>(1);
+	let encrypt_hash_blocks = async {
 		let mut first_block_hash = None;
 		while let Some(next) = block_rx2.recv().await {
 			match next {
 				Ok(block) => {
-					let hash = async_blake2sum(block.clone())
-						.with_context(Context::current_with_span(
-							tracer.start("Hash block (blake2)"),
-						))
-						.await;
-					if first_block_hash.is_none() {
-						first_block_hash = Some(hash);
+					let unencrypted_len = block.len() as u64;
+					let res = tokio::task::spawn_blocking(move || {
+						let block = encryption.encrypt_block(block)?;
+						let hash = blake2sum(&block);
+						Ok((block, hash))
+					})
+					.with_context(Context::current_with_span(
+						tracer.start("Encrypt and hash (blake2) block"),
+					))
+					.await
+					.unwrap();
+					match res {
+						Ok((block, hash)) => {
+							if first_block_hash.is_none() {
+								first_block_hash = Some(hash);
+							}
+							block_tx3.send(Ok((block, unencrypted_len, hash))).await?;
+						}
+						Err(e) => {
+							block_tx3.send(Err(e)).await?;
+							break;
+						}
 					}
-					block_tx3.send(Ok((block, hash))).await?;
 				}
 				Err(e) => {
 					block_tx3.send(Err(e)).await?;
@@ -398,7 +449,7 @@ pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> +
 					block_rx3.recv().await
 				}
 			};
-			let (block, hash) = tokio::select! {
+			let (block, unencrypted_len, hash) = tokio::select! {
 				result = write_futs_next => {
 					result?;
 					continue;
@@ -410,17 +461,18 @@ pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> +
 			};
 
 			// For next block to be written: count its size and spawn future to write it
-			let offset = written_bytes;
-			written_bytes += block.len() as u64;
 			write_futs.push_back(put_block_and_meta(
 				ctx,
 				version,
 				part_number,
-				offset,
+				written_bytes,
 				hash,
 				block,
+				unencrypted_len,
+				encryption.is_encrypted(),
 				order_stream.order(written_bytes),
 			));
+			written_bytes += unencrypted_len;
 		}
 		while let Some(res) = write_futs.next().await {
 			res?;
@@ -429,17 +481,15 @@ pub(crate) async fn read_and_put_blocks<S: Stream<Item = Result<Bytes, Error>> +
 	};
 
 	let (_, stream_hash_result, block_hash_result, final_result) =
-		futures::join!(read_blocks, hash_stream, hash_blocks, put_blocks);
+		futures::join!(read_blocks, hash_stream, encrypt_hash_blocks, put_blocks);
 
 	let total_size = final_result?;
 	// unwrap here is ok, because if hasher failed, it is because something failed
 	// later in the pipeline which already caused a return at the ? on previous line
-	let (data_md5sum, data_sha256sum) = stream_hash_result.unwrap();
 	let first_block_hash = block_hash_result.unwrap();
+	let checksums = stream_hash_result.unwrap().finalize();
 
-	let data_sha256sum = Hash::try_from(&data_sha256sum[..]).unwrap();
-
-	Ok((total_size, data_md5sum, data_sha256sum, first_block_hash))
+	Ok((total_size, checksums, first_block_hash))
 }
 
 async fn put_block_and_meta(
@@ -449,6 +499,8 @@ async fn put_block_and_meta(
 	offset: u64,
 	hash: Hash,
 	block: Bytes,
+	size: u64,
+	is_encrypted: bool,
 	order_tag: OrderTag,
 ) -> Result<(), GarageError> {
 	let ReqCtx { garage, .. } = ctx;
@@ -459,10 +511,7 @@ async fn put_block_and_meta(
 			part_number,
 			offset,
 		},
-		VersionBlock {
-			hash,
-			size: block.len() as u64,
-		},
+		VersionBlock { hash, size },
 	);
 
 	let block_ref = BlockRef {
@@ -474,7 +523,7 @@ async fn put_block_and_meta(
 	futures::try_join!(
 		garage
 			.block_manager
-			.rpc_put_block(hash, block, Some(order_tag)),
+			.rpc_put_block(hash, block, is_encrypted, Some(order_tag)),
 		garage.version_table.insert(&version),
 		garage.block_ref_table.insert(&block_ref),
 	)?;
@@ -517,14 +566,6 @@ impl<S: Stream<Item = Result<Bytes, Error>> + Unpin> StreamChunker<S> {
 	}
 }
 
-pub fn put_response(version_uuid: Uuid, md5sum_hex: String) -> Response<ResBody> {
-	Response::builder()
-		.header("x-amz-version-id", hex::encode(version_uuid))
-		.header("ETag", format!("\"{}\"", md5sum_hex))
-		.body(empty_body())
-		.unwrap()
-}
-
 struct InterruptedCleanup(Option<InterruptedCleanupInner>);
 struct InterruptedCleanupInner {
 	garage: Arc<Garage>,
@@ -559,57 +600,35 @@ impl Drop for InterruptedCleanup {
 
 // ============ helpers ============
 
-pub(crate) fn get_mime_type(headers: &HeaderMap<HeaderValue>) -> Result<String, Error> {
-	Ok(headers
-		.get(hyper::header::CONTENT_TYPE)
-		.map(|x| x.to_str())
-		.unwrap_or(Ok("blob"))?
-		.to_string())
-}
-
-pub(crate) fn get_headers(headers: &HeaderMap<HeaderValue>) -> Result<ObjectVersionHeaders, Error> {
-	let content_type = get_mime_type(headers)?;
-	let mut other = BTreeMap::new();
+pub(crate) fn get_headers(headers: &HeaderMap<HeaderValue>) -> Result<HeaderList, Error> {
+	let mut ret = Vec::new();
 
 	// Preserve standard headers
 	let standard_header = vec![
+		hyper::header::CONTENT_TYPE,
 		hyper::header::CACHE_CONTROL,
 		hyper::header::CONTENT_DISPOSITION,
 		hyper::header::CONTENT_ENCODING,
 		hyper::header::CONTENT_LANGUAGE,
 		hyper::header::EXPIRES,
 	];
-	for h in standard_header.iter() {
-		if let Some(v) = headers.get(h) {
-			match v.to_str() {
-				Ok(v_str) => {
-					other.insert(h.to_string(), v_str.to_string());
-				}
-				Err(e) => {
-					warn!("Discarding header {}, error in .to_str(): {}", h, e);
-				}
-			}
+	for name in standard_header.iter() {
+		if let Some(value) = headers.get(name) {
+			ret.push((name.to_string(), value.to_str()?.to_string()));
 		}
 	}
 
 	// Preserve x-amz-meta- headers
-	for (k, v) in headers.iter() {
-		if k.as_str().starts_with("x-amz-meta-") {
-			match std::str::from_utf8(v.as_bytes()) {
-				Ok(v_str) => {
-					other.insert(k.to_string(), v_str.to_string());
-				}
-				Err(e) => {
-					warn!("Discarding header {}, error in .to_str(): {}", k, e);
-				}
-			}
+	for (name, value) in headers.iter() {
+		if name.as_str().starts_with("x-amz-meta-") {
+			ret.push((
+				name.to_string(),
+				std::str::from_utf8(value.as_bytes())?.to_string(),
+			));
 		}
 	}
 
-	Ok(ObjectVersionHeaders {
-		content_type,
-		other,
-	})
+	Ok(ret)
 }
 
 pub(crate) fn next_timestamp(existing_object: Option<&Object>) -> u64 {
diff --git a/src/api/s3/xml.rs b/src/api/s3/xml.rs
index 06f11288..1e569ade 100644
--- a/src/api/s3/xml.rs
+++ b/src/api/s3/xml.rs
@@ -131,6 +131,14 @@ pub struct CompleteMultipartUploadResult {
 	pub key: Value,
 	#[serde(rename = "ETag")]
 	pub etag: Value,
+	#[serde(rename = "ChecksumCRC32")]
+	pub checksum_crc32: Option<Value>,
+	#[serde(rename = "ChecksumCRC32C")]
+	pub checksum_crc32c: Option<Value>,
+	#[serde(rename = "ChecksumSHA1")]
+	pub checksum_sha1: Option<Value>,
+	#[serde(rename = "ChecksumSHA256")]
+	pub checksum_sha256: Option<Value>,
 }
 
 #[derive(Debug, Serialize, PartialEq, Eq)]
@@ -197,6 +205,14 @@ pub struct PartItem {
 	pub part_number: IntValue,
 	#[serde(rename = "Size")]
 	pub size: IntValue,
+	#[serde(rename = "ChecksumCRC32")]
+	pub checksum_crc32: Option<Value>,
+	#[serde(rename = "ChecksumCRC32C")]
+	pub checksum_crc32c: Option<Value>,
+	#[serde(rename = "ChecksumSHA1")]
+	pub checksum_sha1: Option<Value>,
+	#[serde(rename = "ChecksumSHA256")]
+	pub checksum_sha256: Option<Value>,
 }
 
 #[derive(Debug, Serialize, PartialEq, Eq)]
@@ -500,6 +516,10 @@ mod tests {
 			bucket: Value("mybucket".to_string()),
 			key: Value("a/plop".to_string()),
 			etag: Value("\"3858f62230ac3c915f300c664312c11f-9\"".to_string()),
+			checksum_crc32: None,
+			checksum_crc32c: None,
+			checksum_sha1: Some(Value("ZJAnHyG8PeKz9tI8UTcHrJos39A=".into())),
+			checksum_sha256: None,
 		};
 		assert_eq!(
 			to_xml_with_header(&result)?,
@@ -509,6 +529,7 @@ mod tests {
 	<Bucket>mybucket</Bucket>\
 	<Key>a/plop</Key>\
 	<ETag>&quot;3858f62230ac3c915f300c664312c11f-9&quot;</ETag>\
+    <ChecksumSHA1>ZJAnHyG8PeKz9tI8UTcHrJos39A=</ChecksumSHA1>\
 </CompleteMultipartUploadResult>"
 		);
 		Ok(())
@@ -780,12 +801,22 @@ mod tests {
 					last_modified: Value("2010-11-10T20:48:34.000Z".to_string()),
 					part_number: IntValue(2),
 					size: IntValue(10485760),
+					checksum_crc32: None,
+					checksum_crc32c: None,
+					checksum_sha256: Some(Value(
+						"5RQ3A5uk0w7ojNjvegohch4JRBBGN/cLhsNrPzfv/hA=".into(),
+					)),
+					checksum_sha1: None,
 				},
 				PartItem {
 					etag: Value("\"aaaa18db4cc2f85cedef654fccc4a4x8\"".to_string()),
 					last_modified: Value("2010-11-10T20:48:33.000Z".to_string()),
 					part_number: IntValue(3),
 					size: IntValue(10485760),
+					checksum_sha256: None,
+					checksum_crc32c: None,
+					checksum_crc32: Some(Value("ZJAnHyG8=".into())),
+					checksum_sha1: None,
 				},
 			],
 			initiator: Initiator {
@@ -820,12 +851,14 @@ mod tests {
     <LastModified>2010-11-10T20:48:34.000Z</LastModified>\
     <PartNumber>2</PartNumber>\
     <Size>10485760</Size>\
+    <ChecksumSHA256>5RQ3A5uk0w7ojNjvegohch4JRBBGN/cLhsNrPzfv/hA=</ChecksumSHA256>\
   </Part>\
   <Part>\
     <ETag>&quot;aaaa18db4cc2f85cedef654fccc4a4x8&quot;</ETag>\
     <LastModified>2010-11-10T20:48:33.000Z</LastModified>\
     <PartNumber>3</PartNumber>\
     <Size>10485760</Size>\
+    <ChecksumCRC32>ZJAnHyG8=</ChecksumCRC32>\
   </Part>\
   <Initiator>\
       <DisplayName>umat-user-11116a31-17b5-4fb7-9df5-b288870f11xx</DisplayName>\