api: streaming: organize code paths to accept unsigned and trailer

src/api/common/signature/streaming.rs

@@ -25,17 +25,13 @@ pub fn parse_streaming_body(
 ) -> Result<Request<ReqBody>, Error> {
 	match checked_signature.content_sha256_header {
 		ContentSha256Header::StreamingPayload { signed, trailer } => {
-			if trailer {
+			if !signed && !trailer {
 				return Err(Error::bad_request(
-					"STREAMING-*-TRAILER is not supported by Garage",
+					"STREAMING-UNSIGNED-PAYLOAD is not a valid combination",
-				));
-			}
-			if !signed {
-				return Err(Error::bad_request(
-					"STREAMING-UNSIGNED-PAYLOAD-* is not supported by Garage",
 				));
 			}
 
+			let sign_params = if signed {
 				let signature = checked_signature
 					.signature_header
 					.clone()
@@ -65,13 +61,24 @@ pub fn parse_streaming_body(
 				let date: DateTime<Utc> = Utc.from_utc_datetime(&date);
 
 				let scope = compute_scope(&date, region, service);
-			let signing_hmac = crate::signature::signing_hmac(&date, &secret_key, region, service)
+				let signing_hmac =
+					crate::signature::signing_hmac(&date, &secret_key, region, service)
 						.ok_or_internal_error("Unable to build signing HMAC")?;
 
+				Some(SignParams {
+					datetime: date,
+					scope,
+					signing_hmac,
+					previous_signature: signature,
+				})
+			} else {
+				None
+			};
+
 			Ok(req.map(move |body| {
 				let stream = body_stream::<_, Error>(body);
 				let signed_payload_stream =
-					SignedPayloadStream::new(stream, signing_hmac, date, &scope, signature)
+					StreamingPayloadStream::new(stream, sign_params, trailer)
 						.map(|x| x.map(hyper::body::Frame::data))
 						.map_err(Error::from);
 				ReqBody::new(StreamBody::new(signed_payload_stream))
@@ -122,13 +129,13 @@ mod payload {
 	}
 
 	#[derive(Debug, Clone)]
-	pub struct Header {
+	pub struct ChunkHeader {
 		pub size: usize,
-		pub signature: Hash,
+		pub signature: Option<Hash>,
 	}
 
-	impl Header {
+	impl ChunkHeader {
-		pub fn parse(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+		pub fn parse_signed(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
 			use nom::bytes::streaming::tag;
 			use nom::character::streaming::hex_digit1;
 			use nom::combinator::map_res;
@@ -149,96 +156,136 @@ mod payload {
 
 			let (input, _) = try_parse!(tag("\r\n")(input));
 
-			let header = Header {
+			let header = ChunkHeader {
 				size: size as usize,
-				signature,
+				signature: Some(signature),
+			};
+
+			Ok((input, header))
+		}
+
+		pub fn parse_unsigned(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			use nom::bytes::streaming::tag;
+			use nom::number::streaming::hex_u32;
+
+			macro_rules! try_parse {
+				($expr:expr) => {
+					$expr.map_err(|e| e.map(Error::Parser))?
+				};
+			}
+
+			let (input, size) = try_parse!(hex_u32(input));
+			let (input, _) = try_parse!(tag("\r\n")(input));
+
+			let header = ChunkHeader {
+				size: size as usize,
+				signature: None,
 			};
 
 			Ok((input, header))
 		}
 	}
+
+	#[derive(Debug, Clone)]
+	pub struct TrailerChunk {
+		pub content: String,
+		pub signature: Option<Hash>,
+	}
+
+	impl TrailerChunk {
+		pub fn parse_signed(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			todo!()
+		}
+		pub fn parse_unsigned(input: &[u8]) -> nom::IResult<&[u8], Self, Error<&[u8]>> {
+			todo!()
+		}
+	}
 }
 
 #[derive(Debug)]
-pub enum SignedPayloadStreamError {
+pub enum StreamingPayloadError {
 	Stream(Error),
 	InvalidSignature,
 	Message(String),
 }
 
-impl SignedPayloadStreamError {
+impl StreamingPayloadError {
 	fn message(msg: &str) -> Self {
-		SignedPayloadStreamError::Message(msg.into())
+		StreamingPayloadError::Message(msg.into())
 	}
 }
 
-impl From<SignedPayloadStreamError> for Error {
+impl From<StreamingPayloadError> for Error {
-	fn from(err: SignedPayloadStreamError) -> Self {
+	fn from(err: StreamingPayloadError) -> Self {
 		match err {
-			SignedPayloadStreamError::Stream(e) => e,
+			StreamingPayloadError::Stream(e) => e,
-			SignedPayloadStreamError::InvalidSignature => {
+			StreamingPayloadError::InvalidSignature => {
 				Error::bad_request("Invalid payload signature")
 			}
-			SignedPayloadStreamError::Message(e) => {
+			StreamingPayloadError::Message(e) => {
 				Error::bad_request(format!("Chunk format error: {}", e))
 			}
 		}
 	}
 }
 
-impl<I> From<payload::Error<I>> for SignedPayloadStreamError {
+impl<I> From<payload::Error<I>> for StreamingPayloadError {
 	fn from(err: payload::Error<I>) -> Self {
 		Self::message(err.description())
 	}
 }
 
-impl<I> From<nom::error::Error<I>> for SignedPayloadStreamError {
+impl<I> From<nom::error::Error<I>> for StreamingPayloadError {
 	fn from(err: nom::error::Error<I>) -> Self {
 		Self::message(err.code.description())
 	}
 }
 
-struct SignedPayload {
+enum StreamingPayloadChunk {
-	header: payload::Header,
+	Chunk {
+		header: payload::ChunkHeader,
 		data: Bytes,
+	},
+	Trailer(payload::TrailerChunk),
 }
 
-#[pin_project::pin_project]
+struct SignParams {
-pub struct SignedPayloadStream<S>
-where
-	S: Stream<Item = Result<Bytes, Error>>,
-{
-	#[pin]
-	stream: S,
-	buf: bytes::BytesMut,
 	datetime: DateTime<Utc>,
 	scope: String,
 	signing_hmac: HmacSha256,
 	previous_signature: Hash,
 }
 
-impl<S> SignedPayloadStream<S>
+#[pin_project::pin_project]
+pub struct StreamingPayloadStream<S>
 where
 	S: Stream<Item = Result<Bytes, Error>>,
 {
-	pub fn new(
+	#[pin]
 	stream: S,
-		signing_hmac: HmacSha256,
+	buf: bytes::BytesMut,
-		datetime: DateTime<Utc>,
+	signing: Option<SignParams>,
-		scope: &str,
+	has_trailer: bool,
-		seed_signature: Hash,
+}
-	) -> Self {
+
+impl<S> StreamingPayloadStream<S>
+where
+	S: Stream<Item = Result<Bytes, Error>>,
+{
+	pub fn new(stream: S, signing: Option<SignParams>, has_trailer: bool) -> Self {
 		Self {
 			stream,
 			buf: bytes::BytesMut::new(),
-			datetime,
+			signing,
-			scope: scope.into(),
+			has_trailer,
-			signing_hmac,
-			previous_signature: seed_signature,
 		}
 	}
 
-	fn parse_next(input: &[u8]) -> nom::IResult<&[u8], SignedPayload, SignedPayloadStreamError> {
+	fn parse_next(
+		input: &[u8],
+		is_signed: bool,
+		has_trailer: bool,
+	) -> nom::IResult<&[u8], StreamingPayloadChunk, StreamingPayloadError> {
 		use nom::bytes::streaming::{tag, take};
 
 		macro_rules! try_parse {
@@ -247,33 +294,46 @@ where
 			};
 		}
 
-		let (input, header) = try_parse!(payload::Header::parse(input));
+		let (input, header) = if is_signed {
+			try_parse!(payload::ChunkHeader::parse_signed(input))
+		} else {
+			try_parse!(payload::ChunkHeader::parse_unsigned(input))
+		};
 
 		// 0-sized chunk is the last
 		if header.size == 0 {
+			if has_trailer {
+				let (input, trailer) = if is_signed {
+					try_parse!(payload::TrailerChunk::parse_signed(input))
+				} else {
+					try_parse!(payload::TrailerChunk::parse_unsigned(input))
+				};
+				return Ok((input, StreamingPayloadChunk::Trailer(trailer)));
+			} else {
 				return Ok((
 					input,
-				SignedPayload {
+					StreamingPayloadChunk::Chunk {
 						header,
 						data: Bytes::new(),
 					},
 				));
 			}
+		}
 
 		let (input, data) = try_parse!(take::<_, _, nom::error::Error<_>>(header.size)(input));
 		let (input, _) = try_parse!(tag::<_, _, nom::error::Error<_>>("\r\n")(input));
 
 		let data = Bytes::from(data.to_vec());
 
-		Ok((input, SignedPayload { header, data }))
+		Ok((input, StreamingPayloadChunk::Chunk { header, data }))
 	}
 }
 
-impl<S> Stream for SignedPayloadStream<S>
+impl<S> Stream for StreamingPayloadStream<S>
 where
 	S: Stream<Item = Result<Bytes, Error>> + Unpin,
 {
-	type Item = Result<Bytes, SignedPayloadStreamError>;
+	type Item = Result<Bytes, StreamingPayloadError>;
 
 	fn poll_next(
 		self: Pin<&mut Self>,
@@ -284,7 +344,8 @@ where
 		let mut this = self.project();
 
 		loop {
-			let (input, payload) = match Self::parse_next(this.buf) {
+			let (input, payload) =
+				match Self::parse_next(this.buf, this.signing.is_some(), *this.has_trailer) {
 					Ok(res) => res,
 					Err(nom::Err::Incomplete(_)) => {
 						match futures::ready!(this.stream.as_mut().poll_next(cx)) {
@@ -293,10 +354,10 @@ where
 								continue;
 							}
 							Some(Err(e)) => {
-							return Poll::Ready(Some(Err(SignedPayloadStreamError::Stream(e))))
+								return Poll::Ready(Some(Err(StreamingPayloadError::Stream(e))))
 							}
 							None => {
-							return Poll::Ready(Some(Err(SignedPayloadStreamError::message(
+								return Poll::Ready(Some(Err(StreamingPayloadError::message(
 									"Unexpected EOF",
 								))));
 							}
@@ -307,32 +368,47 @@ where
 					}
 				};
 
-			// 0-sized chunk is the last
+			match payload {
-			if payload.data.is_empty() {
+				StreamingPayloadChunk::Chunk { data, header } => {
-				return Poll::Ready(None);
+					if let Some(signing) = this.signing.as_mut() {
-			}
+						let data_sha256sum = sha256sum(&data);
-
-			let data_sha256sum = sha256sum(&payload.data);
 
 						let expected_signature = compute_streaming_payload_signature(
-				this.signing_hmac,
+							&signing.signing_hmac,
-				*this.datetime,
+							signing.datetime,
-				this.scope,
+							&signing.scope,
-				*this.previous_signature,
+							signing.previous_signature,
 							data_sha256sum,
 						)
 						.map_err(|e| {
-				SignedPayloadStreamError::Message(format!("Could not build signature: {}", e))
+							StreamingPayloadError::Message(format!(
+								"Could not build signature: {}",
+								e
+							))
 						})?;
 
-			if payload.header.signature != expected_signature {
+						if header.signature.unwrap() != expected_signature {
-				return Poll::Ready(Some(Err(SignedPayloadStreamError::InvalidSignature)));
+							return Poll::Ready(Some(Err(StreamingPayloadError::InvalidSignature)));
+						}
+
+						signing.previous_signature = header.signature.unwrap();
 					}
 
 					*this.buf = input.into();
-			*this.previous_signature = payload.header.signature;
 
-			return Poll::Ready(Some(Ok(payload.data)));
+					// 0-sized chunk is the last
+					if data.is_empty() {
+						// if there was a trailer, it would have been returned by the parser
+						assert!(!self.has_trailer);
+						return Poll::Ready(None);
+					}
+
+					return Poll::Ready(Some(Ok(data)));
+				}
+				StreamingPayloadChunk::Trailer(trailer) => {
+					todo!()
+				}
+			}
 		}
 	}
 
@@ -345,7 +421,7 @@ where
 mod tests {
 	use futures::prelude::*;
 
-	use super::{SignedPayloadStream, SignedPayloadStreamError};
+	use super::{StreamingPayloadError, StreamingPayloadStream};
 
 	#[tokio::test]
 	async fn test_interrupted_signed_payload_stream() {
@@ -368,11 +444,11 @@ mod tests {
 		let seed_signature = Hash::default();
 
 		let mut stream =
-			SignedPayloadStream::new(body, signing_hmac, datetime, &scope, seed_signature);
+			StreamingPayloadStream::new(body, signing_hmac, datetime, &scope, seed_signature);
 
 		assert!(stream.try_next().await.is_err());
 		match stream.try_next().await {
-			Err(SignedPayloadStreamError::Message(msg)) if msg == "Unexpected EOF" => {}
+			Err(StreamingPayloadError::Message(msg)) if msg == "Unexpected EOF" => {}
 			item => panic!(
 				"Unexpected result, expected early EOF error, got {:?}",
 				item