Upgrade axum and http (#123)

* Upgrade axum and http

* Fix formatting

* use expect

---------

Co-authored-by: Felix Ableitner <me@nutomic.com>

Cargo.toml

@@ -10,8 +10,8 @@ documentation = "https://docs.rs/activitypub_federation/"
 
 [features]
 default = ["actix-web", "axum"]
-actix-web = ["dep:actix-web"]
+actix-web = ["dep:actix-web", "dep:http02"]
-axum = ["dep:axum", "dep:tower", "dep:hyper", "dep:http-body-util"]
+axum = ["dep:axum", "dep:tower"]
 diesel = ["dep:diesel"]
 
 [lints.rust]
@@ -37,18 +37,18 @@ serde = { version = "1.0.204", features = ["derive"] }
 async-trait = "0.1.81"
 url = { version = "2.5.2", features = ["serde"] }
 serde_json = { version = "1.0.120", features = ["preserve_order"] }
-reqwest = { version = "0.11.27", default-features = false, features = [
+reqwest = { version = "0.12.5", default-features = false, features = [
   "json",
   "stream",
   "rustls-tls",
 ] }
-reqwest-middleware = "0.2.5"
+reqwest-middleware = "0.3.2"
 tracing = "0.1.40"
 base64 = "0.22.1"
 rand = "0.8.5"
 rsa = "0.9.6"
 once_cell = "1.19.0"
-http = "0.2.12"
+http = "1.1.0"
 sha2 = { version = "0.10.8", features = ["oid"] }
 thiserror = "1.0.62"
 derive_builder = "0.20.0"
@@ -56,7 +56,7 @@ itertools = "0.13.0"
 dyn-clone = "1.0.17"
 enum_delegate = "0.2.0"
 httpdate = "1.0.3"
-http-signature-normalization-reqwest = { version = "0.10.0", default-features = false, features = [
+http-signature-normalization-reqwest = { version = "0.12.0", default-features = false, features = [
   "sha-2",
   "middleware",
   "default-spawner",
@@ -84,26 +84,17 @@ moka = { version = "0.12.8", features = ["future"] }
 
 # Actix-web
 actix-web = { version = "4.8.0", default-features = false, optional = true }
+http02 = { package = "http", version = "0.2.12", optional = true }
 
 # Axum
-axum = { version = "0.6.20", features = [
+axum = { version = "0.7.5", features = ["json"], default-features = false, optional = true }
-  "json",
-  "headers",
-], default-features = false, optional = true }
 tower = { version = "0.4.13", optional = true }
-hyper = { version = "0.14", optional = true }
-http-body-util = { version = "0.1.2", optional = true }
 
 [dev-dependencies]
 anyhow = "1.0.86"
+axum = { version = "0.7.5", features = ["macros"] }
+axum-extra = { version = "0.9.3", features = ["typed-header"] }
 env_logger = "0.11.3"
-tower-http = { version = "0.5.2", features = ["map-request-body", "util"] }
-axum = { version = "0.6.20", features = [
-  "http1",
-  "tokio",
-  "query",
-], default-features = false }
-axum-macros = "0.3.8"
 tokio = { version = "1.38.0", features = ["full"] }
 
 [profile.dev]

docs/06_http_endpoints_axum.md

@@ -15,9 +15,9 @@ The next step is to allow other servers to fetch our actors and objects. For thi
 # use activitypub_federation::config::FederationMiddleware;
 # use axum::routing::get;
 # use crate::activitypub_federation::traits::Object;
-# use axum::headers::ContentType;
+# use axum_extra::headers::ContentType;
 # use activitypub_federation::FEDERATION_CONTENT_TYPE;
-# use axum::TypedHeader;
+# use axum_extra::TypedHeader;
 # use axum::response::IntoResponse;
 # use http::HeaderMap;
 # async fn generate_user_html(_: String, _: Data<DbConnection>) -> axum::response::Response { todo!() }
@@ -34,10 +34,9 @@ async fn main() -> Result<(), Error> {
         .layer(FederationMiddleware::new(data));
 
     let addr = SocketAddr::from(([127, 0, 0, 1], 3000));
+    let listener = tokio::net::TcpListener::bind(addr).await?;
     tracing::debug!("listening on {}", addr);
-    axum::Server::bind(&addr)
+    axum::serve(listener, app.into_make_service()).await?;
-        .serve(app.into_make_service())
-        .await?;
     Ok(())
 }
 

examples/live_federation/http.rs

@@ -14,11 +14,11 @@ use activitypub_federation::{
     traits::Object,
 };
 use axum::{
+    debug_handler,
     extract::{Path, Query},
     response::{IntoResponse, Response},
     Json,
 };
-use axum_macros::debug_handler;
 use http::StatusCode;
 use serde::Deserialize;
 

examples/live_federation/main.rs

@@ -64,9 +64,8 @@ async fn main() -> Result<(), Error> {
         .to_socket_addrs()?
         .next()
         .expect("Failed to lookup domain name");
-    axum::Server::bind(&addr)
+    let listener = tokio::net::TcpListener::bind(addr).await?;
-        .serve(app.into_make_service())
+    axum::serve(listener, app.into_make_service()).await?;
-        .await?;
 
     Ok(())
 }

examples/local_federation/axum/http.rs

@@ -14,13 +14,13 @@ use activitypub_federation::{
     traits::Object,
 };
 use axum::{
+    debug_handler,
     extract::{Path, Query},
     response::IntoResponse,
     routing::{get, post},
     Json,
     Router,
 };
-use axum_macros::debug_handler;
 use serde::Deserialize;
 use std::net::ToSocketAddrs;
 use tracing::info;
@@ -39,9 +39,14 @@ pub fn listen(config: &FederationConfig<DatabaseHandle>) -> Result<(), Error> {
         .to_socket_addrs()?
         .next()
         .expect("Failed to lookup domain name");
-    let server = axum::Server::bind(&addr).serve(app.into_make_service());
+    let fut = async move {
+        let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
+        axum::serve(listener, app.into_make_service())
+            .await
+            .unwrap();
+    };
 
-    tokio::spawn(server);
+    tokio::spawn(fut);
     Ok(())
 }
 

src/activity_queue.rs

@@ -451,8 +451,8 @@ mod tests {
             .route("/", post(dodgy_handler))
             .with_state(state);
 
-        axum::Server::bind(&"0.0.0.0:8002".parse().unwrap())
+        let listener = tokio::net::TcpListener::bind("0.0.0.0:8002").await.unwrap();
-            .serve(app.into_make_service())
+        axum::serve(listener, app.into_make_service())
             .await
             .unwrap();
     }

src/activity_sending.rs

@@ -251,8 +251,8 @@ mod tests {
             .route("/", post(dodgy_handler))
             .with_state(state);
 
-        axum::Server::bind(&"0.0.0.0:8001".parse().unwrap())
+        let listener = tokio::net::TcpListener::bind("0.0.0.0:8001").await.unwrap();
-            .serve(app.into_make_service())
+        axum::serve(listener, app.into_make_service())
             .await
             .unwrap();
     }

src/actix_web/http_compat.rs

@@ -0,0 +1,30 @@
+//! Remove these conversion helpers after actix-web upgrades to http 1.0
+
+use std::str::FromStr;
+
+pub fn header_value(v: &http02::HeaderValue) -> http::HeaderValue {
+    http::HeaderValue::from_bytes(v.as_bytes()).expect("can convert http types")
+}
+
+pub fn header_map<'a, H>(m: H) -> http::HeaderMap
+where
+    H: IntoIterator<Item = (&'a http02::HeaderName, &'a http02::HeaderValue)>,
+{
+    let mut new_map = http::HeaderMap::new();
+    for (n, v) in m {
+        new_map.insert(
+            http::HeaderName::from_lowercase(n.as_str().as_bytes())
+                .expect("can convert http types"),
+            header_value(v),
+        );
+    }
+    new_map
+}
+
+pub fn method(m: &http02::Method) -> http::Method {
+    http::Method::from_bytes(m.as_str().as_bytes()).expect("can convert http types")
+}
+
+pub fn uri(m: &http02::Uri) -> http::Uri {
+    http::Uri::from_str(&m.to_string()).expect("can convert http types")
+}

src/actix_web/inbox.rs

@@ -1,5 +1,6 @@
 //! Handles incoming activities, verifying HTTP signatures and other checks
 
+use super::http_compat;
 use crate::{
     config::Data,
     error::Error,
@@ -27,16 +28,18 @@ where
     <ActorT as Object>::Error: From<Error>,
     Datatype: Clone,
 {
-    verify_body_hash(request.headers().get("Digest"), &body)?;
+    let digest_header = request
+        .headers()
+        .get("Digest")
+        .map(http_compat::header_value);
+    verify_body_hash(digest_header.as_ref(), &body)?;
 
     let (activity, actor) = parse_received_activity::<Activity, ActorT, _>(&body, data).await?;
 
-    verify_signature(
+    let headers = http_compat::header_map(request.headers());
-        request.headers(),
+    let method = http_compat::method(request.method());
-        request.method(),
+    let uri = http_compat::uri(request.uri());
-        request.uri(),
+    verify_signature(&headers, &method, &uri, actor.public_key_pem())?;
-        actor.public_key_pem(),
-    )?;
 
     debug!("Receiving activity {}", activity.id().to_string());
     activity.verify(data).await?;
@@ -61,6 +64,16 @@ mod test {
     use serde_json::json;
     use url::Url;
 
+    /// Remove this conversion helper after actix-web upgrades to http 1.0
+    fn header_pair(
+        p: (&http::HeaderName, &http::HeaderValue),
+    ) -> (http02::HeaderName, http02::HeaderValue) {
+        (
+            http02::HeaderName::from_lowercase(p.0.as_str().as_bytes()).unwrap(),
+            http02::HeaderValue::from_bytes(p.1.as_bytes()).unwrap(),
+        )
+    }
+
     #[tokio::test]
     async fn test_receive_activity() {
         let (body, incoming_request, config) = setup_receive_test().await;
@@ -155,7 +168,7 @@ mod test {
         .unwrap();
         let mut incoming_request = TestRequest::post().uri(outgoing_request.url().path());
         for h in outgoing_request.headers() {
-            incoming_request = incoming_request.append_header(h);
+            incoming_request = incoming_request.append_header(header_pair(h));
         }
         incoming_request
     }

src/actix_web/mod.rs

@@ -1,5 +1,6 @@
 //! Utilities for using this library with actix-web framework
 
+mod http_compat;
 pub mod inbox;
 #[doc(hidden)]
 pub mod middleware;
@@ -25,7 +26,14 @@ where
     <A as Object>::Error: From<Error>,
     for<'de2> <A as Object>::Kind: Deserialize<'de2>,
 {
-    verify_body_hash(request.headers().get("Digest"), &body.unwrap_or_default())?;
+    let digest_header = request
+        .headers()
+        .get("Digest")
+        .map(http_compat::header_value);
+    verify_body_hash(digest_header.as_ref(), &body.unwrap_or_default())?;
 
-    http_signatures::signing_actor(request.headers(), request.method(), request.uri(), data).await
+    let headers = http_compat::header_map(request.headers());
+    let method = http_compat::method(request.method());
+    let uri = http_compat::uri(request.uri());
+    http_signatures::signing_actor(&headers, &method, &uri, data).await
 }

src/axum/inbox.rs

@@ -11,7 +11,7 @@ use crate::{
 };
 use axum::{
     async_trait,
-    body::{Bytes, HttpBody},
+    body::Body,
     extract::FromRequest,
     http::{Request, StatusCode},
     response::{IntoResponse, Response},
@@ -59,21 +59,17 @@ pub struct ActivityData {
 }
 
 #[async_trait]
-impl<S, B> FromRequest<S, B> for ActivityData
+impl<S> FromRequest<S> for ActivityData
 where
-    Bytes: FromRequest<S, B>,
-    B: HttpBody + Send + 'static,
     S: Send + Sync,
-    <B as HttpBody>::Error: std::fmt::Display,
-    <B as HttpBody>::Data: Send,
 {
     type Rejection = Response;
 
-    async fn from_request(req: Request<B>, _state: &S) -> Result<Self, Self::Rejection> {
+    async fn from_request(req: Request<Body>, _state: &S) -> Result<Self, Self::Rejection> {
         let (parts, body) = req.into_parts();
 
         // this wont work if the body is an long running stream
-        let bytes = hyper::body::to_bytes(body)
+        let bytes = axum::body::to_bytes(body, usize::MAX)
             .await
             .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response())?;