Migrate to hyper v1

picoHz · Nov 16, 2024 · 719f695 · 719f695
1 parent 7dbfce8
commit 719f695
Show file tree

Hide file tree

Showing 15 changed files with 212 additions and 402 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/README.md b/README.md
@@ -24,7 +24,6 @@ Taxy is currently in early development. Please be aware that breaking changes ma
 - Allows live configuration updates via a REST API without restarting the service
 - Imports TLS certificates from the GUI or can generate a self-signed certificate
 - Provides Let's Encrypt support (ACME v2, HTTP challenge only) for seamless certificate provisioning
-- Supports automatic HTTP Brotli compression
 
 ## Screenshot
 

diff --git a/docs/content/_index.md b/docs/content/_index.md
@@ -16,7 +16,6 @@ sort_by = "weight"
 - Allows live configuration updates via a REST API without restarting the service
 - Imports TLS certificates from the GUI or can generate a self-signed certificate
 - Provides Let's Encrypt support (ACME v2, HTTP challenge only) for seamless certificate provisioning
-- Supports automatic HTTP Brotli compression
 
 # Installation
 

diff --git a/taxy/Cargo.toml b/taxy/Cargo.toml
@@ -28,7 +28,7 @@ axum-extra = { version = "0.9.4", features = ["cookie"] }
 axum-server = { version = "0.7.1", features = ["tls-rustls-no-provider"] }
 backoff = { version = "0.4.0", features = ["tokio"] }
 base64 = "0.22.1"
-brotli = "7.0.0"
+bytes = "1.8.0"
 clap = { version = "4.3.11", features = ["derive", "env"] }
 dashmap = "6.0.1"
 directories = "5.0.1"
@@ -41,8 +41,15 @@ hickory-resolver = { version = "0.24.1", features = [
     "tokio-runtime",
     "system-config",
 ] }
+http-body-util = "0.1.2"
 humantime-serde = "1.1.1"
-hyper = { version = "0.14.27", features = ["full"] }
+hyper = { version = "1.4.1", features = ["full"] }
+hyper-util = { version = "0.1.10", features = [
+    "full",
+    "http1",
+    "http2",
+    "server",
+] }
 include_dir = "0.7.3"
 indexmap = { version = "2.0.0", features = ["serde"] }
 instant-acme = "0.7.1"
@@ -83,14 +90,15 @@ tokio = { version = "1.29.1", features = [
     "signal",
     "io-util",
 ] }
-tokio-rustls = { version = "0.25.0", default-features = false, features = [
+tokio-rustls = { version = "0.26.0", default-features = false, features = [
     "tls12",
     "ring",
 ] }
 tokio-stream = { version = "0.1.14", features = ["sync", "net"] }
 toml = "0.8.8"
 toml_edit = { version = "0.22.9", features = ["serde"] }
 totp-rs = { version = "5.1.0", features = ["gen_secret", "zeroize"] }
+tower-service = "0.3.3"
 tower_governor = "0.4.3"
 tracing = { version = "0.1.37", features = ["release_max_level_info"] }
 tracing-appender = "0.2.2"
@@ -104,7 +112,7 @@ x509-parser = "0.16.0"
 built = "0.6.1"
 
 [dev-dependencies]
-mockito = "1.1.0"
+mockito = "1.6.1"
 net2 = "0.2.39"
 reqwest = { version = "0.12.1", default-features = false, features = [
     "rustls-tls",
@@ -115,6 +123,6 @@ reqwest = { version = "0.12.1", default-features = false, features = [
     "http2",
     "hickory-dns",
 ] }
-tokio-tungstenite = { version = "0.21.0", features = [
+tokio-tungstenite = { version = "0.24.0", features = [
     "rustls-tls-native-roots",
 ] }
diff --git a/taxy/src/proxy/http/compression.rs b/taxy/src/proxy/http/compression.rs
diff --git a/taxy/src/proxy/http/error.rs b/taxy/src/proxy/http/error.rs
@@ -1,4 +1,6 @@
-use hyper::{Body, Response, StatusCode};
+use bytes::Bytes;
+use http_body_util::{combinators::BoxBody, BodyExt, Full};
+use hyper::{body::Body, Response, StatusCode};
 use sailfish::TemplateOnce;
 use thiserror::Error;
 use tokio_rustls::rustls;
@@ -21,17 +23,22 @@ impl ProxyError {
     }
 }
 
-pub fn map_response(
-    res: Result<Response<Body>, anyhow::Error>,
-) -> Result<Response<Body>, anyhow::Error> {
+pub fn map_response<B>(
+    res: Result<Response<B>, anyhow::Error>,
+) -> Result<Response<BoxBody<Bytes, anyhow::Error>>, anyhow::Error>
+where
+    B: Body<Data = Bytes, Error = anyhow::Error> + Send + Sync + 'static,
+{
     match res {
-        Ok(res) => Ok(res),
+        Ok(res) => Ok(res.map(|body| BoxBody::new(body))),
         Err(err) => {
             let code = map_error(err);
             let ctx = ErrorTemplate {
                 code: code.as_u16(),
             };
-            let mut res = Response::new(Body::from(ctx.render_once().unwrap()));
+            let mut res = Response::new(BoxBody::new(
+                Full::new(Bytes::from(ctx.render_once().unwrap())).map_err(Into::into),
+            ));
             *res.status_mut() = code;
             Ok(res)
         }
@@ -42,30 +49,17 @@ fn map_error(err: anyhow::Error) -> StatusCode {
     if let Some(err) = err.downcast_ref::<ProxyError>() {
         return err.code();
     }
-    if let Some(err) = err.downcast_ref::<std::io::Error>() {
-        if err.kind() == std::io::ErrorKind::TimedOut {
-            return StatusCode::GATEWAY_TIMEOUT;
+    if let Some(err) = err.downcast_ref::<rustls::Error>() {
+        if matches!(err, rustls::Error::InvalidCertificate(_)) {
+            return StatusCode::from_u16(526).unwrap();
+        } else {
+            return StatusCode::from_u16(525).unwrap();
         }
     }
     if let Ok(err) = err.downcast::<hyper::Error>() {
-        let is_connect = err.is_connect();
-        if let Some(inner) = err.into_cause() {
-            if let Ok(err) = inner.downcast::<std::io::Error>() {
-                if err.kind() == std::io::ErrorKind::TimedOut {
-                    return StatusCode::GATEWAY_TIMEOUT;
-                }
-                if let Some(inner) = err.into_inner() {
-                    if let Ok(err) = inner.downcast::<rustls::Error>() {
-                        if matches!(*err, rustls::Error::InvalidCertificate(_)) {
-                            return StatusCode::from_u16(526).unwrap();
-                        } else {
-                            return StatusCode::from_u16(525).unwrap();
-                        }
-                    }
-                }
-            }
-        }
-        if is_connect {
+        if err.is_timeout() {
+            return StatusCode::GATEWAY_TIMEOUT;
+        } else {
             return StatusCode::from_u16(523).unwrap();
         }
     }

diff --git a/taxy/src/proxy/http/hyper_tls/client.rs b/taxy/src/proxy/http/hyper_tls/client.rs
@@ -4,11 +4,14 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
-use hyper::{client::connect::HttpConnector, service::Service, Uri};
-use tokio::io::{AsyncRead, AsyncWrite};
+use hyper::rt::{Read, Write};
+use hyper::Uri;
+use hyper_util::client::legacy::connect::HttpConnector;
+use hyper_util::rt::TokioIo;
 use tokio_rustls::rustls::pki_types::ServerName;
 use tokio_rustls::rustls::ClientConfig;
 use tokio_rustls::TlsConnector;
+use tower_service::Service;
 
 use super::stream::MaybeHttpsStream;
 
@@ -75,7 +78,7 @@ impl<T: fmt::Debug> fmt::Debug for HttpsConnector<T> {
 impl<T> Service<Uri> for HttpsConnector<T>
 where
     T: Service<Uri>,
-    T::Response: AsyncRead + AsyncWrite + Send + Unpin,
+    T::Response: Read + Write + Send + Unpin,
     T::Future: Send + 'static,
     T::Error: Into<BoxError>,
 {
@@ -104,13 +107,23 @@ where
             .trim_matches(|c| c == '[' || c == ']')
             .to_owned();
         let connecting = self.http.call(dst);
-        let tls = self.tls.clone();
+
+        let tls_connector = self.tls.clone();
+
         let fut = async move {
             let tcp = connecting.await.map_err(Into::into)?;
+
             let maybe = if is_https {
-                let tls = tls
-                    .connect(ServerName::try_from(host.as_str()).unwrap().to_owned(), tcp)
-                    .await?;
+                let stream = TokioIo::new(tcp);
+
+                let tls = TokioIo::new(
+                    tls_connector
+                        .connect(
+                            ServerName::try_from(host.as_str()).unwrap().to_owned(),
+                            stream,
+                        )
+                        .await?,
+                );
                 MaybeHttpsStream::Https(tls)
             } else {
                 MaybeHttpsStream::Http(tcp)
@@ -130,7 +143,7 @@ type BoxedFut<T> = Pin<Box<dyn Future<Output = Result<MaybeHttpsStream<T>, BoxEr
 /// A Future representing work to connect to a URL, and a TLS handshake.
 pub struct HttpsConnecting<T>(BoxedFut<T>);
 
-impl<T: AsyncRead + AsyncWrite + Unpin> Future for HttpsConnecting<T> {
+impl<T: Read + Write + Unpin> Future for HttpsConnecting<T> {
     type Output = Result<MaybeHttpsStream<T>, BoxError>;
 
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {