Finish upgrading axum/hyper

Author: jarhodes314

Cargo.lock

@@ -78,12 +78,13 @@ assert_matches = "1.5"
 async-compat = "0.2.1"
 async-log = "2.0"
 async-trait = "0.1"
-async-tungstenite = { version = "0.23", features = [
+async-tungstenite = { version = "0.28", features = [
     "tokio-runtime",
     "tokio-rustls-native-certs",
 ] }
 axum = "0.8.1"
-axum-server = { version = "0.5.1", features = ["tls-rustls"] }
+axum-extra = "0.10.0"
+axum-server = { version = "0.7.1", features = ["tls-rustls"] }
 backoff = { version = "0.4", features = ["tokio"] }
 base64 = "0.13"
 bytes = "1.4"
@@ -111,19 +112,20 @@ futures-timer = "3.0"
 futures-util = "0.3.25"
 glob = "0.3"
 heck = "0.4.1"
-http = "0.2"
-http-body = "0.4"
+http = "1.2"
+http-body = "1.0.0"
+http-body-util = "0.1.2"
 httparse = "1.9.3"
 humantime = "2.1.0"
-hyper = { version = "0.14", default-features = false }
-hyper-util = { version = "0.1" }
+hyper = { version = "1.5", default-features = false }
 hyper-rustls = { version = "0.27.5", default-features = false, features = [
-    "tokio-runtime",
+    "native-tokio",
     "tls12",
     "rustls-native-certs",
     "http1",
     "http2",
 ] }
+hyper-util = { version = "0.1" }
 itertools = "0.13"
 lazy_static = "1.4"
 log = "0.4"
@@ -155,9 +157,9 @@ regex = "1.4"
 reqwest = { version = "0.12", default-features = false }
 rpassword = "5.0"
 rstest = "0.16.0"
-rumqttc = { git = "https://github.com/bytebeamio/rumqtt", ref = "49c3b5f81360299c316b052778905a7420bea579" }
+rumqttc = { git = "https://github.com/bytebeamio/rumqtt", rev = "49c3b5f81360299c316b052778905a7420bea579" }
 rumqttd = "0.19"
-rustls = "0.23"
+rustls = { version = "0.23", default-features = false }
 rustls-native-certs = "0.6.3"
 rustls-pemfile = "2.2"
 serde = "1.0"
@@ -178,20 +180,20 @@ thiserror = "1.0"
 time = "0.3"
 tokio = { version = "1.37", default-features = false }
 tokio-rustls = "0.26.1"
-tokio-tungstenite = { version = "0.20.0" }
+tokio-tungstenite = { version = "0.26.1" }
 tokio-util = { version = "0.7", features = ["codec"] }
 toml = "0.8"
 tower = "0.4"
 tracing = { version = "0.1", features = ["attributes", "log"] }
 tracing-subscriber = { version = "0.3", features = ["time", "env-filter"] }
 try-traits = "0.1"
-tungstenite = "0.20"
+tungstenite = "0.26.1"
 url = "2.3"
 uzers = "0.11"
 walkdir = "2"
 which = "4.2"
 whoami = "1.5.0"
-ws_stream_tungstenite = "0.11"
+ws_stream_tungstenite = "0.14"
 x509-parser = "0.16"
 yansi = "1.0.1"
 zeroize = "1.5"

Cargo.toml

@@ -21,8 +21,11 @@ axum-server = { workspace = true }
 camino = { workspace = true }
 futures = { workspace = true }
 hyper = { workspace = true }
+hyper-util = { workspace = true, features = [
+    "client",
+    "client-legacy",
+], optional = true }
 pin-project = { workspace = true }
-hyper-util = { workspace = true, features = ["client", "client-legacy"], optional = true }
 reqwest = { workspace = true, features = [
     "rustls-tls-native-roots",
 ], optional = true }

crates/common/axum_tls/Cargo.toml

@@ -87,9 +87,9 @@ where
             } else {
                 let (stream, service) = acceptor.accept(stream, service).await?;
                 let server_conn = stream.get_ref().1;
-                let cert = (|| {
-                    X509Certificate::from_der(&server_conn.peer_certificates()?.first()?).ok()
-                })();
+                let cert =
+                    (|| X509Certificate::from_der(server_conn.peer_certificates()?.first()?).ok())(
+                    );
                 let certificate_info = TlsData {
                     common_name: common_name(cert.as_ref()).map(Arc::from),
                     is_secure: true,
@@ -117,7 +117,6 @@ mod tests {
     use reqwest::Certificate;
     use reqwest::Client;
     use reqwest::Identity;
-    use rustls::crypto::CryptoProvider;
     use rustls::pki_types::pem::PemObject as _;
     use rustls::pki_types::CertificateDer;
     use rustls::pki_types::PrivateKeyDer;
@@ -173,9 +172,12 @@ mod tests {
 
     #[tokio::test]
     async fn acceptor_rejects_untrusted_client_certificates() {
-        let permitted_certificate = rcgen::generate_simple_self_signed(vec!["not-my-client".into()]).unwrap();
+        let permitted_certificate =
+            rcgen::generate_simple_self_signed(vec!["not-my-client".into()]).unwrap();
         let mut roots = RootCertStore::empty();
-        roots.add(permitted_certificate.serialize_der().unwrap().into()).unwrap();
+        roots
+            .add(permitted_certificate.serialize_der().unwrap().into())
+            .unwrap();
         let server = Server::with_trusted_roots(roots);
         let client = Client::builder()
             .add_root_certificate(server.certificate.clone())
@@ -257,7 +259,6 @@ mod tests {
         }
 
         fn start(trusted_roots: Option<RootCertStore>) -> Self {
-            let _ = CryptoProvider::install_default(rustls::crypto::ring::default_provider());
             let mut port = 3000;
             let listener = loop {
                 if let Ok(listener) = TcpListener::bind::<SocketAddr>(([127, 0, 0, 1], port).into())

crates/common/axum_tls/src/acceptor.rs

@@ -220,17 +220,10 @@ mod tests {
     }
 
     mod server_accepts {
-        use rustls::crypto::CryptoProvider;
-
         use super::*;
 
-        fn init_crypto() {
-            let _ = CryptoProvider::install_default(rustls::crypto::ring::default_provider());
-        }
-
         #[tokio::test]
         async fn alg_ed25519_pkcs8() {
-            init_crypto();
             let key = test_data("ed25519.key");
             let cert = test_data("ed25519.crt");
 
@@ -242,7 +235,6 @@ mod tests {
 
         #[tokio::test]
         async fn alg_ec() {
-            init_crypto();
             let key = test_data("ec.key");
             let cert = test_data("ec.crt");
 
@@ -254,7 +246,6 @@ mod tests {
 
         #[tokio::test]
         async fn alg_ec_pkcs8() {
-            init_crypto();
             let key = test_data("ec.pkcs8.key");
             let cert = test_data("ec.crt");
 
@@ -266,7 +257,6 @@ mod tests {
 
         #[tokio::test]
         async fn alg_rsa_pkcs8() {
-            init_crypto();
             let key = test_data("rsa.pkcs8.key");
             let cert = test_data("rsa.crt");
 
@@ -278,7 +268,6 @@ mod tests {
 
         #[tokio::test]
         async fn alg_rsa_pkcs1() {
-            init_crypto();
             let key = test_data("rsa.pkcs1.key");
             let cert = test_data("rsa.crt");
 
@@ -310,7 +299,7 @@ mod tests {
                 .context("reading certs")?;
             let key_der = parse_key_to_der(key)?;
             let cert = reqwest::tls::Certificate::from_der(
-                &*chain.first().expect("chain should contain certificate"),
+                chain.first().expect("chain should contain certificate"),
             )
             .context("converting certificate to reqwest::tls::Certificate")?;
             let config = ssl_config(chain, key_der, None)?;

crates/common/axum_tls/src/files.rs

@@ -182,7 +182,6 @@ pub fn read_cert_chain(
     })?;
     let mut cert_reader = BufReader::new(f);
     certs(&mut cert_reader)
-        .map(|der_chain| Ok(CertificateDer::from(der_chain?)))
         .collect::<Result<Vec<_>, _>>()
         .map_err(|e| CertificateError::CertificateParseFailed {
             path: cert_file.as_ref().to_path_buf(),

crates/common/certificate/src/parse_root_certificate.rs

@@ -283,13 +283,13 @@ impl Uploader {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use axum::extract::BodyStream;
     use axum::http::StatusCode;
     use axum::routing::put;
     use axum::Router;
     use backoff::ExponentialBackoffBuilder;
     use futures::future::pending;
     use futures::stream::StreamExt;
+    use std::future::IntoFuture as _;
     use std::sync::atomic::AtomicBool;
     use std::sync::atomic::Ordering;
     use std::sync::Arc;
@@ -450,7 +450,7 @@ mod tests {
 
         let app = Router::new().route(
             "/target.txt",
-            put(|mut body: BodyStream| async move {
+            put(|body: axum::body::Body| async move {
                 let res = async {
                     if is_first_attempt.fetch_and(false, Ordering::SeqCst) {
                         Ok(StatusCode::INTERNAL_SERVER_ERROR)
@@ -460,7 +460,8 @@ mod tests {
                                 .await
                                 .context("creating file")?,
                         );
-                        while let Some(chunk) = body.next().await {
+                        let mut body_stream = body.into_data_stream();
+                        while let Some(chunk) = body_stream.next().await {
                             file.write_all(&chunk.context("receiving chunk")?)
                                 .await
                                 .context("writing chunk")?;
@@ -484,11 +485,7 @@ mod tests {
             }),
         );
 
-        let server_task = tokio::spawn(
-            axum::Server::from_tcp(listener.into_std().unwrap())
-                .unwrap()
-                .serve(app.into_make_service()),
-        );
+        let server_task = tokio::spawn(axum::serve(listener, app).into_future());
 
         tokio::time::sleep(Duration::from_millis(50)).await;
 

crates/common/upload/src/upload.rs

@@ -585,7 +585,7 @@ fn check_device_status_c8y(
 
     mqtt_options.set_keep_alive(RESPONSE_TIMEOUT);
 
-    let (mut client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
+    let (client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
     connection
         .eventloop
         .network_options
@@ -705,7 +705,7 @@ fn check_device_status_azure(
 
     mqtt_options.set_keep_alive(RESPONSE_TIMEOUT);
 
-    let (mut client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
+    let (client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
     let mut acknowledged = false;
 
     if tedge_config.mqtt.bridge.built_in {
@@ -810,7 +810,7 @@ fn check_device_status_aws(
         .rumqttc_options()?;
     mqtt_options.set_keep_alive(RESPONSE_TIMEOUT);
 
-    let (mut client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
+    let (client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
     let mut acknowledged = false;
 
     if tedge_config.mqtt.bridge.built_in {

crates/core/tedge/src/cli/connect/command.rs

@@ -25,7 +25,7 @@ pub(crate) fn get_connected_c8y_url(
         .rumqttc_options()?;
     mqtt_options.set_keep_alive(RESPONSE_TIMEOUT);
 
-    let (mut client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
+    let (client, mut connection) = rumqttc::Client::new(mqtt_options, 10);
     connection
         .eventloop
         .network_options

crates/core/tedge/src/cli/connect/jwt_token.rs

@@ -22,7 +22,7 @@ fn disconnect_if_interrupted(client: Client) -> Arc<AtomicBool> {
         unsafe {
             let _ = signal_hook::low_level::register(*signal, move || {
                 interrupted.store(true, Ordering::Relaxed);
-                let mut client = interrupter.lock().unwrap();
+                let client = interrupter.lock().unwrap();
                 let _ = client.disconnect();
             });
         }