Convert content-type to lowercase for comparison (#114)

* Convert content-type to lowercase for comparison

* rust 1.78

* clippy priority

* upgrade dep

.woodpecker.yml

@@ -1,54 +1,56 @@
-pipeline:
+variables:
+  - &rust_image "rust:1.78-bullseye"
+
+steps:
   cargo_fmt:
     image: rustdocker/rust:nightly
     commands:
       - /root/.cargo/bin/cargo fmt -- --check
-
+    when:
-  cargo_check:
+      - event: pull_request
-    image: rust:1.70-bullseye
-    environment:
-      CARGO_HOME: .cargo
-    commands:
-      - cargo check --all-features --all-targets
 
   cargo_clippy:
-    image: rust:1.70-bullseye
+    image: *rust_image
     environment:
       CARGO_HOME: .cargo
     commands:
       - rustup component add clippy
-      - cargo clippy --all-targets --all-features --
+      - cargo clippy --all-targets --all-features
-        -D warnings -D deprecated -D clippy::perf -D clippy::complexity
+    when:
-        -D clippy::dbg_macro -D clippy::inefficient_to_string
+      - event: pull_request
-        -D clippy::items-after-statements -D clippy::implicit_clone
-        -D clippy::wildcard_imports -D clippy::cast_lossless
-        -D clippy::manual_string_new -D clippy::redundant_closure_for_method_calls
-      - cargo clippy --all-features -- -D clippy::unwrap_used
 
   cargo_test:
-    image: rust:1.70-bullseye
+    image: *rust_image
     environment:
       CARGO_HOME: .cargo
     commands:
       - cargo test --all-features --no-fail-fast
+    when:
+      - event: pull_request
 
   cargo_doc:
-    image: rust:1.70-bullseye
+    image: *rust_image
     environment:
       CARGO_HOME: .cargo
     commands:
       - cargo doc --all-features
+    when:
+      - event: pull_request
 
   cargo_run_actix_example:
-    image: rust:1.70-bullseye
+    image: *rust_image
     environment:
       CARGO_HOME: .cargo
     commands:
       - cargo run --example local_federation actix-web
+    when:
+      - event: pull_request
 
   cargo_run_axum_example:
-    image: rust:1.70-bullseye
+    image: *rust_image
     environment:
       CARGO_HOME: .cargo
     commands:
       - cargo run --example local_federation axum
+    when:
+      - event: pull_request

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "activitypub_federation"
-version = "0.5.2"
+version = "0.5.6"
 edition = "2021"
 description = "High-level Activitypub framework"
 keywords = ["activitypub", "activitystreams", "federation", "fediverse"]
@@ -14,22 +14,39 @@ actix-web = ["dep:actix-web"]
 axum = ["dep:axum", "dep:tower", "dep:hyper", "dep:http-body-util"]
 diesel = ["dep:diesel"]
 
+[lints.rust]
+warnings = "deny"
+deprecated = "deny"
+
+[lints.clippy]
+perf = { level = "deny", priority = -1 }
+complexity = { level = "deny", priority = -1 }
+dbg_macro  = "deny"
+inefficient_to_string = "deny"
+items-after-statements  = "deny" 
+implicit_clone = "deny"
+wildcard_imports = "deny"
+cast_lossless = "deny"
+manual_string_new = "deny"
+redundant_closure_for_method_calls = "deny"
+unwrap_used = "deny"
+
 [dependencies]
-chrono = { version = "0.4.34", features = ["clock"], default-features = false }
+chrono = { version = "0.4.38", features = ["clock"], default-features = false }
-serde = { version = "1.0.197", features = ["derive"] }
+serde = { version = "1.0.200", features = ["derive"] }
-async-trait = "0.1.77"
+async-trait = "0.1.80"
 url = { version = "2.5.0", features = ["serde"] }
-serde_json = { version = "1.0.114", features = ["preserve_order"] }
+serde_json = { version = "1.0.116", features = ["preserve_order"] }
-reqwest = { version = "0.11.24", features = ["json", "stream"] }
+reqwest = { version = "0.11.27", features = ["json", "stream"] }
-reqwest-middleware = "0.2.4"
+reqwest-middleware = "0.2.5"
 tracing = "0.1.40"
-base64 = "0.21.7"
+base64 = "0.22.1"
 openssl = "0.10.64"
 once_cell = "1.19.0"
-http = "0.2.11"
+http = "0.2.12"
 sha2 = "0.10.8"
-thiserror = "1.0.57"
+thiserror = "1.0.59"
-derive_builder = "0.12.0"
+derive_builder = "0.20.0"
 itertools = "0.12.1"
 dyn-clone = "1.0.17"
 enum_delegate = "0.2.0"
@@ -40,20 +57,20 @@ http-signature-normalization-reqwest = { version = "0.10.0", default-features =
   "default-spawner",
 ] }
 http-signature-normalization = "0.7.0"
-bytes = "1.5.0"
+bytes = "1.6.0"
 futures-core = { version = "0.3.30", default-features = false }
-pin-project-lite = "0.2.13"
+pin-project-lite = "0.2.14"
 activitystreams-kinds = "0.3.0"
-regex = { version = "1.10.3", default-features = false, features = ["std", "unicode-case"] }
+regex = { version = "1.10.5", default-features = false, features = ["std", "unicode"] }
-tokio = { version = "1.36.0", features = [
+tokio = { version = "1.37.0", features = [
   "sync",
   "rt",
   "rt-multi-thread",
   "time",
 ] }
-diesel = { version = "2.1.4", features = ["postgres"], default-features = false, optional = true }
+diesel = { version = "2.1.6", features = ["postgres"], default-features = false, optional = true }
 futures = "0.3.30"
-moka = { version = "0.12.5", features = ["future"] }
+moka = { version = "0.12.7", features = ["future"] }
 
 # Actix-web
 actix-web = { version = "4.5.1", default-features = false, optional = true }
@@ -65,12 +82,12 @@ axum = { version = "0.6.20", features = [
 ], default-features = false, optional = true }
 tower = { version = "0.4.13", optional = true }
 hyper = { version = "0.14", optional = true }
-http-body-util = {version = "0.1.0", optional = true }
+http-body-util = {version = "0.1.1", optional = true }
 
 [dev-dependencies]
-anyhow = "1.0.80"
+anyhow = "1.0.82"
 rand = "0.8.5"
-env_logger = "0.10.2"
+env_logger = "0.11.3"
 tower-http = { version = "0.5.2", features = ["map-request-body", "util"] }
 axum = { version = "0.6.20", features = [
   "http1",
@@ -78,7 +95,7 @@ axum = { version = "0.6.20", features = [
   "query",
 ], default-features = false }
 axum-macros = "0.3.8"
-tokio = { version = "1.36.0", features = ["full"] }
+tokio = { version = "1.37.0", features = ["full"] }
 
 [profile.dev]
 strip = "symbols"

examples/live_federation/main.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::unwrap_used)]
+
 use crate::{
     database::Database,
     http::{http_get_user, http_post_user_inbox, webfinger},

examples/live_federation/objects/post.rs

@@ -21,7 +21,6 @@ pub struct DbPost {
     pub text: String,
     pub ap_id: ObjectId<DbPost>,
     pub creator: ObjectId<DbUser>,
-    pub local: bool,
 }
 
 #[derive(Deserialize, Serialize, Debug)]
@@ -59,7 +58,15 @@ impl Object for DbPost {
     }
 
     async fn into_json(self, _data: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
-        unimplemented!()
+        Ok(Note {
+            kind: NoteType::Note,
+            id: self.ap_id,
+            content: self.text,
+            attributed_to: self.creator,
+            to: vec![public()],
+            tag: vec![],
+            in_reply_to: None,
+        })
     }
 
     async fn verify(
@@ -81,7 +88,6 @@ impl Object for DbPost {
             text: json.content,
             ap_id: json.id.clone(),
             creator: json.attributed_to.clone(),
-            local: false,
         };
 
         let mention = Mention {

examples/local_federation/instance.rs

@@ -28,6 +28,7 @@ pub async fn new_instance(
         .domain(hostname)
         .signed_fetch_actor(&system_user)
         .app_data(database)
+        .url_verifier(Box::new(MyUrlVerifier()))
         .debug(true)
         .build()
         .await?;

examples/local_federation/main.rs

@@ -1,3 +1,5 @@
+#![allow(clippy::unwrap_used)]
+
 use crate::{
     instance::{listen, new_instance, Webserver},
     objects::post::DbPost,

src/activity_queue.rs

@@ -416,6 +416,7 @@ async fn retry<T, E: Display + Debug, F: Future<Output = Result<T, E>>, A: FnMut
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod tests {
     use super::*;
     use crate::http_signatures::generate_actor_keypair;

src/activity_sending.rs

@@ -12,14 +12,17 @@ use crate::{
 };
 use bytes::Bytes;
 use futures::StreamExt;
+use http::StatusCode;
 use httpdate::fmt_http_date;
 use itertools::Itertools;
 use openssl::pkey::{PKey, Private};
-use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
+use reqwest::{
+    header::{HeaderMap, HeaderName, HeaderValue},
+    Response,
+};
 use reqwest_middleware::ClientWithMiddleware;
 use serde::Serialize;
 use std::{
-    self,
     fmt::{Debug, Display},
     time::{Duration, SystemTime},
 };
@@ -90,20 +93,30 @@ impl SendActivityTask {
         )
         .await?;
         let response = client.execute(request).await?;
+        self.handle_response(response).await
+    }
 
-        match response {
+    /// Based on the HTTP status code determines if an activity was delivered successfully. In that case
-            o if o.status().is_success() => {
+    /// Ok is returned. Otherwise it returns Err and the activity send should be retried later.
+    ///
+    /// Equivalent code in mastodon: https://github.com/mastodon/mastodon/blob/v4.2.8/app/helpers/jsonld_helper.rb#L215-L217
+    async fn handle_response(&self, response: Response) -> Result<(), Error> {
+        match response.status() {
+            status if status.is_success() => {
                 debug!("Activity {self} delivered successfully");
                 Ok(())
             }
-            o if o.status().is_client_error() => {
+            status
-                let text = o.text_limited().await?;
+                if status.is_client_error()
+                    && status != StatusCode::REQUEST_TIMEOUT
+                    && status != StatusCode::TOO_MANY_REQUESTS =>
+            {
+                let text = response.text_limited().await?;
                 debug!("Activity {self} was rejected, aborting: {text}");
                 Ok(())
             }
-            o => {
+            status => {
-                let status = o.status();
+                let text = response.text_limited().await?;
-                let text = o.text_limited().await?;
 
                 Err(Error::Other(format!(
                     "Activity {self} failure with status {status}: {text}",
@@ -211,11 +224,10 @@ pub(crate) fn generate_request_headers(inbox_url: &Url) -> HeaderMap {
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod tests {
     use super::*;
     use crate::{config::FederationConfig, http_signatures::generate_actor_keypair};
-    use bytes::Bytes;
-    use http::StatusCode;
     use std::{
         sync::{atomic::AtomicUsize, Arc},
         time::Instant,
@@ -289,4 +301,48 @@ mod tests {
         info!("Queue Sent: {:?}", start.elapsed());
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_handle_response() {
+        let keypair = generate_actor_keypair().unwrap();
+        let message = SendActivityTask {
+            actor_id: "http://localhost:8001".parse().unwrap(),
+            activity_id: "http://localhost:8001/activity".parse().unwrap(),
+            activity: "{}".into(),
+            inbox: "http://localhost:8001".parse().unwrap(),
+            private_key: keypair.private_key().unwrap(),
+            http_signature_compat: true,
+        };
+
+        let res = |status| {
+            http::Response::builder()
+                .status(status)
+                .body(vec![])
+                .unwrap()
+                .into()
+        };
+
+        assert!(message.handle_response(res(StatusCode::OK)).await.is_ok());
+        assert!(message
+            .handle_response(res(StatusCode::BAD_REQUEST))
+            .await
+            .is_ok());
+
+        assert!(message
+            .handle_response(res(StatusCode::MOVED_PERMANENTLY))
+            .await
+            .is_err());
+        assert!(message
+            .handle_response(res(StatusCode::REQUEST_TIMEOUT))
+            .await
+            .is_err());
+        assert!(message
+            .handle_response(res(StatusCode::TOO_MANY_REQUESTS))
+            .await
+            .is_err());
+        assert!(message
+            .handle_response(res(StatusCode::INTERNAL_SERVER_ERROR))
+            .await
+            .is_err());
+    }
 }

src/actix_web/inbox.rs

@@ -45,6 +45,7 @@ where
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod test {
     use super::*;
     use crate::{

src/config.rs

@@ -174,11 +174,17 @@ impl<T: Clone> FederationConfig<T> {
     /// Returns true if the url refers to this instance. Handles hostnames like `localhost:8540` for
     /// local debugging.
     pub(crate) fn is_local_url(&self, url: &Url) -> bool {
-        let mut domain = url.host_str().expect("id has domain").to_string();
+        match url.host_str() {
-        if let Some(port) = url.port() {
+            Some(domain) => {
-            domain = format!("{}:{}", domain, port);
+                let domain = if let Some(port) = url.port() {
+                    format!("{}:{}", domain, port)
+                } else {
+                    domain.to_string()
+                };
+                domain == self.domain
+            }
+            None => false,
         }
-        domain == self.domain
     }
 
     /// Returns the local domain
@@ -341,3 +347,34 @@ impl<T: Clone> FederationMiddleware<T> {
         FederationMiddleware(config)
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used)]
+mod test {
+    use super::*;
+
+    async fn config() -> FederationConfig<i32> {
+        FederationConfig::builder()
+            .domain("example.com")
+            .app_data(1)
+            .build()
+            .await
+            .unwrap()
+    }
+
+    #[tokio::test]
+    async fn test_url_is_local() -> Result<(), Error> {
+        let config = config().await;
+        assert!(config.is_local_url(&Url::parse("http://example.com")?));
+        assert!(!config.is_local_url(&Url::parse("http://other.com")?));
+        // ensure that missing domain doesnt cause crash
+        assert!(!config.is_local_url(&Url::parse("http://127.0.0.1")?));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_get_domain() {
+        let config = config().await;
+        assert_eq!("example.com", config.domain());
+    }
+}

src/fetch/mod.rs

@@ -53,26 +53,35 @@ pub async fn fetch_object_http<T: Clone, Kind: DeserializeOwned>(
     url: &Url,
     data: &Data<T>,
 ) -> Result<FetchObjectResponse<Kind>, Error> {
-    static CONTENT_TYPE: HeaderValue = HeaderValue::from_static(FEDERATION_CONTENT_TYPE);
+    static FETCH_CONTENT_TYPE: HeaderValue = HeaderValue::from_static(FEDERATION_CONTENT_TYPE);
-    static ALT_CONTENT_TYPE: HeaderValue = HeaderValue::from_static(
+    const VALID_RESPONSE_CONTENT_TYPES: [&str; 3] = [
-        r#"application/ld+json; profile="https://www.w3.org/ns/activitystreams""#,
+        FEDERATION_CONTENT_TYPE, // lemmy
-    );
+        r#"application/ld+json; profile="https://www.w3.org/ns/activitystreams""#, // activitypub standard
-    static ALT_CONTENT_TYPE_MASTODON: HeaderValue =
+        r#"application/activity+json; charset=utf-8"#,                             // mastodon
-        HeaderValue::from_static(r#"application/activity+json; charset=utf-8"#);
+    ];
-    let res = fetch_object_http_with_accept(url, data, &CONTENT_TYPE).await?;
+    let res = fetch_object_http_with_accept(url, data, &FETCH_CONTENT_TYPE).await?;
 
-    // Ensure correct content-type to prevent vulnerabilities.
+    // Ensure correct content-type to prevent vulnerabilities, with case insensitive comparison.
-    if res.content_type.as_ref() != Some(&CONTENT_TYPE)
+    let content_type = res
-        && res.content_type.as_ref() != Some(&ALT_CONTENT_TYPE)
+        .content_type
-        && res.content_type.as_ref() != Some(&ALT_CONTENT_TYPE_MASTODON)
+        .as_ref()
-    {
+        .and_then(|c| Some(c.to_str().ok()?.to_lowercase()))
+        .ok_or(Error::FetchInvalidContentType(res.url.clone()))?;
+    if !VALID_RESPONSE_CONTENT_TYPES.contains(&content_type.as_str()) {
         return Err(Error::FetchInvalidContentType(res.url));
     }
 
-    // Ensure id field matches final url
+    // Ensure id field matches final url after redirect
     if res.object_id.as_ref() != Some(&res.url) {
         return Err(Error::FetchWrongId(res.url));
     }
+
+    // Dont allow fetching local object. Only check this after the request as a local url
+    // may redirect to a remote object.
+    if data.config.is_local_url(&res.url) {
+        return Err(Error::NotFound);
+    }
+
     Ok(res)
 }
 
@@ -84,8 +93,6 @@ async fn fetch_object_http_with_accept<T: Clone, Kind: DeserializeOwned>(
     content_type: &HeaderValue,
 ) -> Result<FetchObjectResponse<Kind>, Error> {
     let config = &data.config;
-    // dont fetch local objects this way
-    debug_assert!(url.domain() != Some(&config.domain));
     config.verify_url_valid(url).await?;
     info!("Fetching remote object {}", url.to_string());
 

src/fetch/object_id.rs

@@ -88,19 +88,13 @@ where
         <Kind as Object>::Error: From<Error>,
     {
         let db_object = self.dereference_from_db(data).await?;
-        // if its a local object, only fetch it from the database and not over http
-        if data.config.is_local_url(&self.0) {
-            return match db_object {
-                None => Err(Error::NotFound.into()),
-                Some(o) => Ok(o),
-            };
-        }
 
         // object found in database
         if let Some(object) = db_object {
-            // object is old and should be refetched
             if let Some(last_refreshed_at) = object.last_refreshed_at() {
-                if should_refetch_object(last_refreshed_at) {
+                let is_local = data.config.is_local_url(&self.0);
+                if !is_local && should_refetch_object(last_refreshed_at) {
+                    // object is outdated and should be refetched
                     return self.dereference_from_http(data, Some(object)).await;
                 }
             }
@@ -175,6 +169,11 @@ where
         Kind::verify(&res.object, redirect_url, data).await?;
         Kind::from_json(res.object, data).await
     }
+
+    /// Returns true if the object's domain matches the one defined in [[FederationConfig.domain]].
+    pub fn is_local(&self, data: &Data<<Kind as Object>::DataType>) -> bool {
+        data.config.is_local_url(&self.0)
+    }
 }
 
 /// Need to implement clone manually, to avoid requiring Kind to be Clone
@@ -345,9 +344,10 @@ const _IMPL_DIESEL_NEW_TYPE_FOR_OBJECT_ID: () = {
 };
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 pub mod tests {
     use super::*;
-    use crate::{fetch::object_id::should_refetch_object, traits::tests::DbUser};
+    use crate::traits::tests::DbUser;
 
     #[test]
     fn test_deserialize() {

src/fetch/webfinger.rs

@@ -245,6 +245,7 @@ pub struct WebfingerLink {
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod tests {
     use super::*;
     use crate::{

src/http_signatures.rs

@@ -189,8 +189,11 @@ fn verify_signature_inner(
     uri: &Uri,
     public_key: &str,
 ) -> Result<(), Error> {
-    static CONFIG: Lazy<http_signature_normalization::Config> =
+    static CONFIG: Lazy<http_signature_normalization::Config> = Lazy::new(|| {
-        Lazy::new(|| http_signature_normalization::Config::new().set_expiration(EXPIRES_AFTER));
+        http_signature_normalization::Config::new()
+            .set_expiration(EXPIRES_AFTER)
+            .require_digest()
+    });
 
     let path_and_query = uri.path_and_query().map(PathAndQuery::as_str).unwrap_or("");
 
@@ -275,6 +278,7 @@ pub(crate) fn verify_body_hash(
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 pub mod test {
     use super::*;
     use crate::activity_sending::generate_request_headers;

src/traits.rs

@@ -343,7 +343,7 @@ pub mod tests {
         error::Error,
         fetch::object_id::ObjectId,
         http_signatures::{generate_actor_keypair, Keypair},
-        protocol::{public_key::PublicKey, verification::verify_domains_match},
+        protocol::verification::verify_domains_match,
     };
     use activitystreams_kinds::{activity::FollowType, actor::PersonType};
     use once_cell::sync::Lazy;