actix: update version to 0.11.1

actix/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "http-signature-normalization-actix"
 description = "An HTTP Signatures library that leaves the signing to you"
-version = "0.10.1"
+version = "0.11.1"
 authors = ["asonix <asonix@asonix.dog>"]
 license = "AGPL-3.0"
 readme = "README.md"
@@ -32,12 +32,13 @@ actix-http = { version = "3.0.2", default-features = false }
 actix-rt = "2.6.0"
 actix-web = { version = "4.0.0", default-features = false, optional = true }
 awc = { version = "3.0.0", default-features = false, optional = true }
-base64 = { version = "0.13", optional = true }
-futures-util = { version = "0.3", default-features = false }
+base64 = { version = "0.22", optional = true }
+futures-core = "0.3.28"
 http-signature-normalization = { version = "0.7.0", path = ".." }
-ring = { version = "0.16.20", optional = true }
+ring = { version = "0.17.5", optional = true }
 sha2 = { version = "0.10", optional = true }
 sha3 = { version = "0.10", optional = true }
+streem = "0.2.0"
 thiserror = "1.0"
 tokio = { version = "1", default-features = false, features = ["sync"] }
 tracing = "0.1"
@@ -46,7 +47,7 @@ tracing-futures = "0.2"
 
 [dev-dependencies]
 actix-rt = "2.6.0"
-tracing-actix-web = { version = "0.6.0" }
+tracing-actix-web = { version = "0.7.0" }
 tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
 
 [package.metadata.docs.rs]

actix/examples/client.rs

@@ -1,5 +1,6 @@
 use actix_rt::task::JoinError;
 use awc::Client;
+use base64::{engine::general_purpose::STANDARD, Engine};
 use http_signature_normalization_actix::{digest::ring::Sha256, prelude::*, Canceled};
 use tracing::{error, info};
 use tracing_error::ErrorLayer;
@@ -14,7 +15,7 @@ async fn request(config: Config) -> Result<(), Box<dyn std::error::Error>> {
         .append_header(("Accept", "text/plain"))
         .signature_with_digest(config, "my-key-id", digest, "Hewwo-owo", |s| {
             info!("Signing String\n{}", s);
-            Ok(base64::encode(s)) as Result<_, MyError>
+            Ok(STANDARD.encode(s)) as Result<_, MyError>
         })
         .await?
         .send()

actix/examples/server.rs

@@ -1,4 +1,5 @@
 use actix_web::{http::StatusCode, web, App, HttpRequest, HttpResponse, HttpServer, ResponseError};
+use base64::{engine::general_purpose::STANDARD, Engine};
 use http_signature_normalization_actix::{digest::ring::Sha256, prelude::*};
 use std::future::{ready, Ready};
 use tracing::info;
@@ -29,7 +30,7 @@ impl SignatureVerify for MyVerify {
             return ready(Err(MyError::Key));
         }
 
-        let decoded = match base64::decode(&signature) {
+        let decoded = match STANDARD.decode(&signature) {
             Ok(decoded) => decoded,
             Err(_) => return ready(Err(MyError::Decode)),
         };

actix/src/digest/middleware.rs

@@ -1,6 +1,6 @@
 //! Types for setting up Digest middleware verification
 
-use crate::{DefaultSpawner, Spawn};
+use crate::{Canceled, DefaultSpawner, Spawn};
 
 use super::{DigestPart, DigestVerify};
 use actix_web::{
@@ -10,16 +10,14 @@ use actix_web::{
     http::{header::HeaderValue, StatusCode},
     web, FromRequest, HttpMessage, HttpRequest, HttpResponse, ResponseError,
 };
-use futures_util::{
-    future::LocalBoxFuture,
-    stream::{Stream, StreamExt},
-};
+use futures_core::{future::LocalBoxFuture, Stream};
 use std::{
     future::{ready, Ready},
     pin::Pin,
     task::{Context, Poll},
 };
-use tokio::sync::mpsc;
+use streem::{from_fn::Yielder, IntoStreamer};
+use tokio::sync::{mpsc, oneshot};
 use tracing::{debug, Span};
 use tracing_error::SpanTrace;
 
@@ -87,10 +85,10 @@ enum VerifyErrorKind {
 struct RxStream<T>(mpsc::Receiver<T>);
 
 impl<T> Stream for RxStream<T> {
-    type Item = T;
+    type Item = Result<T, PayloadError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        Pin::new(&mut self.0).poll_recv(cx)
+        Pin::new(&mut self.0).poll_recv(cx).map(|opt| opt.map(Ok))
     }
 }
 
@@ -127,22 +125,30 @@ where
     }
 }
 
+struct VerifiedReceiver {
+    rx: Option<oneshot::Receiver<()>>,
+}
+
 impl FromRequest for DigestVerified {
     type Error = VerifyError;
-    type Future = Ready<Result<Self, Self::Error>>;
+    type Future = LocalBoxFuture<'static, Result<Self, Self::Error>>;
 
     fn from_request(req: &HttpRequest, _: &mut Payload) -> Self::Future {
         let res = req
-            .extensions()
-            .get::<Self>()
-            .copied()
+            .extensions_mut()
+            .get_mut::<VerifiedReceiver>()
+            .and_then(|r| r.rx.take())
             .ok_or_else(|| VerifyError::new(&Span::current(), VerifyErrorKind::Extension));
 
         if res.is_err() {
             debug!("Failed to fetch DigestVerified from request");
         }
 
-        ready(res)
+        Box::pin(async move {
+            res?.await
+                .map_err(|_| VerifyError::new(&Span::current(), VerifyErrorKind::Dropped))
+                .map(|()| DigestVerified)
+        })
     }
 }
 
@@ -201,23 +207,23 @@ where
                     )));
                 }
             };
-            let payload = req.take_payload();
+
             let spawner = self.1.clone();
+            let digest = self.3.clone();
+            let (verify_tx, verify_rx) = oneshot::channel();
 
-            let (tx, rx) = mpsc::channel(1);
-            let f1 = span.in_scope(|| verify_payload(spawner, vec, self.3.clone(), payload, tx));
-
+            let payload = req.take_payload();
             let payload: Pin<Box<dyn Stream<Item = Result<web::Bytes, PayloadError>> + 'static>> =
-                Box::pin(RxStream(rx).map(Ok));
+                Box::pin(streem::try_from_fn(|yielder| async move {
+                    verify_payload(yielder, spawner, vec, digest, payload, verify_tx).await
+                }));
             req.set_payload(payload.into());
-            req.extensions_mut().insert(DigestVerified);
 
-            let f2 = self.0.call(req);
+            req.extensions_mut().insert(VerifiedReceiver {
+                rx: Some(verify_rx),
+            });
 
-            Box::pin(async move {
-                let (_, res) = futures_util::future::join(f1, f2).await;
-                res
-            })
+            Box::pin(self.0.call(req))
         } else if self.2 {
             Box::pin(ready(Err(VerifyError::new(
                 &span,
@@ -230,41 +236,79 @@ where
     }
 }
 
-#[tracing::instrument(name = "Verify Payload", skip(spawner, verify_digest, payload, tx))]
+fn canceled_error(error: Canceled) -> PayloadError {
+    PayloadError::Io(std::io::Error::new(std::io::ErrorKind::Other, error))
+}
+
+fn verified_error(error: VerifyError) -> PayloadError {
+    PayloadError::Io(std::io::Error::new(std::io::ErrorKind::Other, error))
+}
+
 async fn verify_payload<T, Spawner>(
+    yielder: Yielder<Result<web::Bytes, PayloadError>>,
     spawner: Spawner,
     vec: Vec<DigestPart>,
     mut verify_digest: T,
-    mut payload: Payload,
-    tx: mpsc::Sender<web::Bytes>,
-) -> Result<(), actix_web::Error>
+    payload: Payload,
+    verify_tx: oneshot::Sender<()>,
+) -> Result<(), PayloadError>
 where
     T: DigestVerify + Clone + Send + 'static,
     Spawner: Spawn,
 {
-    while let Some(res) = payload.next().await {
-        let bytes = res?;
-        let bytes2 = bytes.clone();
-        verify_digest = spawner
-            .spawn_blocking(move || {
-                verify_digest.update(bytes2.as_ref());
-                Ok(verify_digest) as Result<T, VerifyError>
-            })
-            .await??;
+    let mut payload = payload.into_streamer();
 
-        tx.send(bytes)
-            .await
-            .map_err(|_| VerifyError::new(&Span::current(), VerifyErrorKind::Dropped))?;
+    let mut error = None;
+
+    while let Some(bytes) = payload.try_next().await? {
+        if error.is_none() {
+            let bytes2 = bytes.clone();
+            let mut verify_digest2 = verify_digest.clone();
+
+            let task = spawner.spawn_blocking(move || {
+                verify_digest2.update(bytes2.as_ref());
+                Ok(verify_digest2) as Result<T, VerifyError>
+            });
+
+            yielder.yield_ok(bytes).await;
+
+            match task.await {
+                Ok(Ok(digest)) => {
+                    verify_digest = digest;
+                }
+                Ok(Err(e)) => {
+                    error = Some(verified_error(e));
+                }
+                Err(e) => {
+                    error = Some(canceled_error(e));
+                }
+            }
+        } else {
+            yielder.yield_ok(bytes).await;
+        }
+    }
+
+    if let Some(error) = error {
+        return Err(error);
     }
 
     let verified = spawner
         .spawn_blocking(move || Ok(verify_digest.verify(&vec)) as Result<_, VerifyError>)
-        .await??;
+        .await
+        .map_err(canceled_error)?
+        .map_err(verified_error)?;
 
     if verified {
+        if verify_tx.send(()).is_err() {
+            debug!("handler dropped");
+        }
+
         Ok(())
     } else {
-        Err(VerifyError::new(&Span::current(), VerifyErrorKind::Verify).into())
+        Err(verified_error(VerifyError::new(
+            &Span::current(),
+            VerifyErrorKind::Verify,
+        )))
     }
 }
 

actix/src/digest/ring.rs

@@ -96,11 +96,12 @@ impl DigestName for Sha512 {
 mod client {
     use super::*;
     use crate::digest::DigestCreate;
+    use base64::prelude::*;
 
     fn create(mut context: ring::digest::Context, input: &[u8]) -> String {
         context.update(input);
         let digest = context.finish();
-        base64::encode(digest.as_ref())
+        BASE64_STANDARD.encode(digest.as_ref())
     }
 
     impl DigestCreate for Sha256 {
@@ -126,6 +127,7 @@ mod client {
 mod server {
     use super::*;
     use crate::digest::{DigestPart, DigestVerify};
+    use base64::prelude::*;
     use tracing::{debug, warn};
 
     fn verify(context: ring::digest::Context, name: &str, parts: &[DigestPart]) -> bool {
@@ -135,7 +137,7 @@ mod server {
         {
             debug!("Verifying digest type, {}", name);
             let digest = context.finish();
-            let encoded = base64::encode(digest.as_ref());
+            let encoded = BASE64_STANDARD.encode(digest.as_ref());
 
             return part.digest == encoded;
         }

actix/src/digest/sha2.rs

@@ -21,13 +21,14 @@ impl DigestName for Sha512 {
 mod client {
     use super::*;
     use crate::digest::DigestCreate;
+    use base64::prelude::*;
 
     fn create<D: sha2::Digest + sha2::digest::FixedOutputReset>(
         digest: &mut D,
         input: &[u8],
     ) -> String {
         sha2::Digest::update(digest, input);
-        base64::encode(&digest.finalize_reset())
+        BASE64_STANDARD.encode(&digest.finalize_reset())
     }
 
     impl DigestCreate for Sha224 {
@@ -59,6 +60,7 @@ mod client {
 mod server {
     use super::*;
     use crate::digest::{DigestPart, DigestVerify};
+    use base64::prelude::*;
     use tracing::{debug, warn};
 
     fn verify<D: sha2::Digest + sha2::digest::FixedOutputReset>(
@@ -71,7 +73,7 @@ mod server {
             .find(|p| p.algorithm.to_lowercase() == name.to_lowercase())
         {
             debug!("Verifying digest type, {}", name);
-            let encoded = base64::encode(&digest.finalize_reset());
+            let encoded = BASE64_STANDARD.encode(&digest.finalize_reset());
 
             return part.digest == encoded;
         }

actix/src/digest/sha3.rs

@@ -40,17 +40,18 @@ impl DigestName for Sha3_512 {
     const NAME: &'static str = "SHA3-512";
 }
 
-#[cfg(features = "client")]
+#[cfg(feature = "client")]
 mod client {
     use super::*;
     use crate::digest::DigestCreate;
+    use base64::prelude::*;
 
     fn create<D: sha3::Digest + sha3::digest::FixedOutputReset>(
         digest: &mut D,
         input: &[u8],
     ) -> String {
-        digest.update(input);
-        base64::encode(&digest.finalize_reset())
+        sha3::Digest::update(digest, input);
+        BASE64_STANDARD.encode(&digest.finalize_reset())
     }
 
     impl DigestCreate for Sha3_224 {
@@ -112,6 +113,7 @@ mod client {
 mod server {
     use super::*;
     use crate::digest::{DigestPart, DigestVerify};
+    use base64::prelude::*;
     use tracing::{debug, warn};
 
     fn verify<D: sha3::Digest + sha3::digest::FixedOutputReset>(
@@ -124,7 +126,7 @@ mod server {
             .find(|p| p.algorithm.to_lowercase() == name.to_lowercase())
         {
             debug!("Verifying digest type, {}", name);
-            let encoded = base64::encode(&digest.finalize_reset());
+            let encoded = BASE64_STANDARD.encode(&digest.finalize_reset());
 
             return part.digest == encoded;
         }

actix/src/middleware.rs

@@ -7,7 +7,7 @@ use actix_web::{
     http::StatusCode,
     Error, FromRequest, HttpMessage, HttpRequest, HttpResponse, ResponseError,
 };
-use futures_util::future::LocalBoxFuture;
+use futures_core::future::LocalBoxFuture;
 use std::{
     collections::HashSet,
     future::{ready, Ready},

flake.lock

@@ -5,11 +5,11 @@
         "systems": "systems"
       },
       "locked": {
-        "lastModified": 1689068808,
-        "narHash": "sha256-6ixXo3wt24N/melDWjq70UuHQLxGV8jZvooRanIHXw0=",
+        "lastModified": 1694529238,
+        "narHash": "sha256-zsNZZGTGnMOf9YpHKJqMSsa0dXbfmxeoJ7xHlrt+xmY=",
         "owner": "numtide",
         "repo": "flake-utils",
-        "rev": "919d646de7be200f3bf08cb76ae1f09402b6f9b4",
+        "rev": "ff7b65b44d01cf9ba6a71320833626af21126384",
         "type": "github"
       },
       "original": {
@@ -20,11 +20,11 @@
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1690272529,
-        "narHash": "sha256-MakzcKXEdv/I4qJUtq/k/eG+rVmyOZLnYNC2w1mB59Y=",
+        "lastModified": 1700794826,
+        "narHash": "sha256-RyJTnTNKhO0yqRpDISk03I/4A67/dp96YRxc86YOPgU=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "ef99fa5c5ed624460217c31ac4271cfb5cb2502c",
+        "rev": "5a09cb4b393d58f9ed0d9ca1555016a8543c2ac8",
         "type": "github"
       },
       "original": {

reqwest/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "http-signature-normalization-reqwest"
 description = "An HTTP Signatures library that leaves the signing to you"
-version = "0.10.0"
+version = "0.12.0"
 authors = ["asonix <asonix@asonix.dog>"]
 license = "AGPL-3.0"
 readme = "README.md"
@@ -25,12 +25,12 @@ required-features = ["default-spawner", "ring"]
 
 [dependencies]
 async-trait = "0.1.71"
-base64 = { version = "0.13", optional = true }
+base64 = { version = "0.22", optional = true }
 http-signature-normalization = { version = "0.7.0", path = ".." }
 httpdate = "1.0.2"
-reqwest = { version = "0.11", default-features = false, features = ["json"] }
-reqwest-middleware = { version = "0.2.0", optional = true }
-ring = { version = "0.16.20", optional = true }
+reqwest = { version = "0.12", default-features = false, features = ["json"] }
+reqwest-middleware = { version = "0.3.0", optional = true }
+ring = { version = "0.17.5", optional = true }
 sha2 = { version = "0.10", optional = true }
 sha3 = { version = "0.10", optional = true }
 thiserror = "1.0"
@@ -39,7 +39,7 @@ tokio = { version = "1", default-features = false, features = [
 ], optional = true }
 
 [dev-dependencies]
-pretty_env_logger = "0.4"
+pretty_env_logger = "0.5"
 tokio = { version = "1", default-features = false, features = [
   "rt-multi-thread",
   "macros",

reqwest/examples/client.rs

@@ -1,3 +1,4 @@
+use base64::{engine::general_purpose::STANDARD, Engine};
 use http_signature_normalization_reqwest::{digest::ring::Sha256, prelude::*};
 use reqwest::{
     header::{ACCEPT, USER_AGENT},
@@ -15,7 +16,7 @@ async fn request(config: Config) -> Result<(), Box<dyn std::error::Error + Send
         .header(ACCEPT, "text/plain")
         .signature_with_digest(config, "my-key-id", digest, "Hewwo-owo", |s| {
             println!("Signing String\n{}", s);
-            Ok(base64::encode(s)) as Result<_, MyError>
+            Ok(STANDARD.encode(s)) as Result<_, MyError>
         })
         .await?;
 

reqwest/src/digest/ring.rs

@@ -81,9 +81,11 @@ impl Default for Sha512 {
 }
 
 fn create(mut context: ring::digest::Context, input: &[u8]) -> String {
+    use base64::prelude::*;
+
     context.update(input);
     let digest = context.finish();
-    base64::encode(digest.as_ref())
+    BASE64_STANDARD.encode(digest.as_ref())
 }
 
 impl DigestCreate for Sha256 {

reqwest/src/digest/sha2.rs

@@ -1,3 +1,4 @@
+use base64::prelude::*;
 use sha2::{Sha224, Sha256, Sha384, Sha512};
 
 use super::DigestCreate;
@@ -7,7 +8,7 @@ fn create<D: sha2::Digest + sha2::digest::FixedOutputReset>(
     input: &[u8],
 ) -> String {
     sha2::Digest::update(digest, input);
-    base64::encode(&digest.finalize_reset())
+    BASE64_STANDARD.encode(&digest.finalize_reset())
 }
 
 impl DigestCreate for Sha224 {

reqwest/src/digest/sha3.rs

@@ -1,3 +1,4 @@
+use base64::prelude::*;
 use sha3::{
     Keccak224, Keccak256, Keccak256Full, Keccak384, Keccak512, Sha3_224, Sha3_256, Sha3_384,
     Sha3_512,
@@ -10,7 +11,7 @@ fn create<D: sha3::Digest + sha3::digest::FixedOutputReset>(
     input: &[u8],
 ) -> String {
     sha3::Digest::update(digest, input);
-    base64::encode(&digest.finalize_reset())
+    BASE64_STANDARD.encode(&digest.finalize_reset())
 }
 
 impl DigestCreate for Sha3_224 {