Verify digest without consuming body

http-signature-normalization-warp/Cargo.toml

@@ -16,9 +16,10 @@ bytes = "0.5.3"
 futures = "0.3.1"
 http = "0.2.0"
 http-signature-normalization-http = { version = "0.3.0", path = "../http-signature-normalization-http" }
+hyper = "0.13.1"
 serde = { version = "1.0.104", features = ["derive"], optional = true }
 serde_json = { version = "1.0.44", optional = true }
 serde_urlencoded = { version = "0.6.1", optional = true }
 sha2 = { version = "0.8.1", optional = true }
 thiserror = { version = "1.0.9", optional = true }
-warp = "0.2.0"
+warp = { version = "0.2.0", git = "https://git.asonix.dog/asonix/warp", branch = "asonix/body-map-test" }

http-signature-normalization-warp/src/digest/mod.rs

@@ -1,6 +1,6 @@
-use bytes::Bytes;
+use bytes::Buf;
 use std::str::FromStr;
-use warp::{body, header, Filter, Rejection};
+use warp::{header, Filter, Rejection};
 
 #[cfg(feature = "sha-2")]
 mod sha_2;
@@ -9,6 +9,7 @@ pub trait DigestVerify {
     fn verify(&mut self, digests: &[DigestPart], payload: &[u8]) -> bool;
 }
 
+#[derive(Clone, Debug)]
 pub struct DigestPart {
     pub algorithm: String,
     pub digest: String,
@@ -35,49 +36,33 @@ pub enum ParseDigestError {
 #[error("Could not parse request body")]
 pub struct ParseBodyError;
 
-pub fn verify_bytes(
+pub fn verify<F>(
-    verifier: impl DigestVerify + Clone + Send,
+    verifier: impl DigestVerify + Clone + Send + Sync + 'static,
-) -> impl Filter<Extract = (Bytes,), Error = Rejection> + Clone {
+    filter: F,
-    parse_digest_header().and(body::bytes()).and_then(
+) -> impl Filter<Extract = F::Extract, Error = Rejection> + Clone
-        move |parts: Vec<DigestPart>, bytes: Bytes| {
+where
+    F: Filter + Clone + Send + Sync + 'static,
+    F::Extract: warp::Reply,
+    F::Error: Into<Rejection>,
+{
+    filter.with(
+        warp::body::map_request_body(parse_digest_header(), move |body: hyper::body::Body, parts: (Vec<DigestPart>,)| {
             let mut verifier = verifier.clone();
             async move {
-                if verifier.verify(&parts, &bytes) {
+                let parts = parts.clone();
-                    Ok(bytes)
+                let buf = hyper::body::aggregate(body).await.ok()?;
+                let bytes: Vec<u8> = buf.bytes().to_owned();
+
+                if verifier.verify(&parts.0, &bytes)  {
+                    Some(bytes.into())
                 } else {
-                    Err(warp::reject::custom(VerifyError))
+                    None
                 }
             }
-        },
+        })
     )
 }
 
-pub fn verify_json<T>(
-    verifier: impl DigestVerify + Clone + Send,
-) -> impl Filter<Extract = (T,), Error = Rejection> + Clone
-where
-    T: serde::de::DeserializeOwned,
-{
-    verify_bytes(verifier).and_then(|bytes: Bytes| {
-        async move {
-            serde_json::from_slice(&bytes).map_err(|_| warp::reject::custom(ParseBodyError))
-        }
-    })
-}
-
-pub fn verify_form<T>(
-    verifier: impl DigestVerify + Clone + Send,
-) -> impl Filter<Extract = (T,), Error = Rejection> + Clone
-where
-    T: serde::de::DeserializeOwned,
-{
-    verify_bytes(verifier).and_then(|bytes: Bytes| {
-        async move {
-            serde_urlencoded::from_bytes(&bytes).map_err(|_| warp::reject::custom(ParseBodyError))
-        }
-    })
-}
-
 fn parse_digest_header() -> impl Filter<Extract = (Vec<DigestPart>,), Error = Rejection> + Clone {
     header::header::<Digest>("Digest").map(|d: Digest| d.parts)
 }