Compare commits

..

2 commits

Author SHA1 Message Date
Felix Ableitner
4aa9658710 doctest 2025-01-15 12:29:03 +01:00
Felix Ableitner
095d0aee2d Add verify_is_remote_object 2025-01-15 12:06:43 +01:00
46 changed files with 601 additions and 4410 deletions

1
.gitignore vendored
View file

@ -1,5 +1,6 @@
/target /target
/.idea /.idea
/Cargo.lock
perf.data* perf.data*
flamegraph.svg flamegraph.svg

View file

@ -1,4 +1,4 @@
edition = "2021" edition="2021"
imports_layout = "HorizontalVertical" imports_layout="HorizontalVertical"
imports_granularity = "Crate" imports_granularity="Crate"
reorder_imports = true reorder_imports=true

View file

@ -1,5 +1,5 @@
variables: variables:
- &rust_image "rust:1.91-bullseye" - &rust_image "rust:1.78-bullseye"
steps: steps:
cargo_fmt: cargo_fmt:

3309
Cargo.lock generated

File diff suppressed because it is too large Load diff

View file

@ -1,6 +1,6 @@
[package] [package]
name = "activitypub_federation" name = "activitypub_federation"
version = "0.7.0-beta.11" version = "0.6.1"
edition = "2021" edition = "2021"
description = "High-level Activitypub framework" description = "High-level Activitypub framework"
keywords = ["activitypub", "activitystreams", "federation", "fediverse"] keywords = ["activitypub", "activitystreams", "federation", "fediverse"]
@ -12,7 +12,11 @@ documentation = "https://docs.rs/activitypub_federation/"
default = ["actix-web", "axum"] default = ["actix-web", "axum"]
actix-web = ["dep:actix-web", "dep:http02"] actix-web = ["dep:actix-web", "dep:http02"]
axum = ["dep:axum", "dep:tower"] axum = ["dep:axum", "dep:tower"]
axum-original-uri = ["dep:axum", "axum/original-uri"] diesel = ["dep:diesel"]
[lints.rust]
warnings = "deny"
deprecated = "deny"
[lints.clippy] [lints.clippy]
perf = { level = "deny", priority = -1 } perf = { level = "deny", priority = -1 }
@ -28,71 +32,70 @@ redundant_closure_for_method_calls = "deny"
unwrap_used = "deny" unwrap_used = "deny"
[dependencies] [dependencies]
chrono = { version = "0.4.42", features = ["clock"], default-features = false } chrono = { version = "0.4.38", features = ["clock"], default-features = false }
serde = { version = "1.0.228", features = ["derive"] } serde = { version = "1.0.204", features = ["derive"] }
async-trait = "0.1.89" async-trait = "0.1.81"
url = { version = "2.5.8", features = ["serde"] } url = { version = "2.5.2", features = ["serde"] }
serde_json = { version = "1.0.149", features = ["preserve_order"] } serde_json = { version = "1.0.120", features = ["preserve_order"] }
reqwest = { version = "0.13.1", default-features = false, features = [ reqwest = { version = "0.12.5", default-features = false, features = [
"json", "json",
"stream", "stream",
"rustls-tls",
] } ] }
reqwest-middleware = "0.5.0" reqwest-middleware = "0.3.2"
tracing = "0.1.44" tracing = "0.1.40"
base64 = "0.22.1" base64 = "0.22.1"
rand = "0.8.5" rand = "0.8.5"
rsa = "0.9.10" rsa = "0.9.6"
http = "1.4.0" once_cell = "1.19.0"
sha2 = { version = "0.10.9", features = ["oid"] } http = "1.1.0"
thiserror = "2.0.17" sha2 = { version = "0.10.8", features = ["oid"] }
derive_builder = "0.20.2" thiserror = "1.0.62"
itertools = "0.14.0" derive_builder = "0.20.0"
dyn-clone = "1.0.20" itertools = "0.13.0"
dyn-clone = "1.0.17"
enum_delegate = "0.2.0" enum_delegate = "0.2.0"
httpdate = "1.0.3" httpdate = "1.0.3"
http-signature-normalization-reqwest = { version = "0.14.0", default-features = false, features = [ http-signature-normalization-reqwest = { version = "0.12.0", default-features = false, features = [
"sha-2", "sha-2",
"middleware", "middleware",
"default-spawner", "default-spawner",
] } ] }
http-signature-normalization = "0.7.0" http-signature-normalization = "0.7.0"
bytes = "1.11.0" bytes = "1.6.1"
futures-core = { version = "0.3.31", default-features = false } futures-core = { version = "0.3.30", default-features = false }
pin-project-lite = "0.2.16" pin-project-lite = "0.2.14"
activitystreams-kinds = "0.3.0" activitystreams-kinds = "0.3.0"
regex = { version = "1.12.2", default-features = false, features = [ regex = { version = "1.10.5", default-features = false, features = [
"std", "std",
"unicode", "unicode",
] } ] }
tokio = { version = "1.49.0", features = [ tokio = { version = "1.38.0", features = [
"sync", "sync",
"rt", "rt",
"rt-multi-thread", "rt-multi-thread",
"time", "time",
] } ] }
futures = "0.3.31" diesel = { version = "2.2.1", features = [
moka = { version = "0.12.12", features = ["future"] } "postgres",
either = "1.15.0" ], default-features = false, optional = true }
futures = "0.3.30"
moka = { version = "0.12.8", features = ["future"] }
# Actix-web # Actix-web
actix-web = { version = "4.12.1", default-features = false, optional = true } actix-web = { version = "4.8.0", default-features = false, optional = true }
http02 = { package = "http", version = "0.2.12", optional = true } http02 = { package = "http", version = "0.2.12", optional = true }
# Axum # Axum
axum = { version = "0.8.8", features = [ axum = { version = "0.7.5", features = ["json"], default-features = false, optional = true }
"json", tower = { version = "0.4.13", optional = true }
], default-features = false, optional = true }
tower = { version = "0.5.2", optional = true }
[dev-dependencies] [dev-dependencies]
anyhow = "1.0.100" anyhow = "1.0.86"
axum = { version = "0.8.8", features = ["macros"] } axum = { version = "0.7.5", features = ["macros"] }
axum-extra = { version = "0.12.5", features = ["typed-header"] } axum-extra = { version = "0.9.3", features = ["typed-header"] }
env_logger = "0.11.8" env_logger = "0.11.3"
tokio = { version = "1.49.0", features = ["full"] } tokio = { version = "1.38.0", features = ["full"] }
reqwest = { version = "0.13.1",features = [
"rustls"
] }
[profile.dev] [profile.dev]
strip = "symbols" strip = "symbols"
@ -105,8 +108,3 @@ path = "examples/local_federation/main.rs"
[[example]] [[example]]
name = "live_federation" name = "live_federation"
path = "examples/live_federation/main.rs" path = "examples/live_federation/main.rs"
# Speedup RSA key generation
# https://github.com/RustCrypto/RSA/blob/master/README.md#example
[profile.dev.package.num-bigint-dig]
opt-level = 3

View file

@ -10,7 +10,7 @@ A high-level framework for [ActivityPub](https://www.w3.org/TR/activitypub/) fed
The ActivityPub protocol is a decentralized social networking protocol. It allows web servers to exchange data using JSON over HTTP. Data can be fetched on demand, and also delivered directly to inboxes for live updates. The ActivityPub protocol is a decentralized social networking protocol. It allows web servers to exchange data using JSON over HTTP. Data can be fetched on demand, and also delivered directly to inboxes for live updates.
Activitypub has the potential to form the basis of the next generation of social media. This is because it has a number of major advantages compared to existing platforms and alternative technologies: While Activitypub is not in widespread use yet, is has the potential to form the basis of the next generation of social media. This is because it has a number of major advantages compared to existing platforms and alternative technologies:
- **Interoperability**: Imagine being able to comment under a Youtube video directly from twitter.com, and having the comment shown under the video on youtube.com. Or following a Subreddit from Facebook. Such functionality is already available on the equivalent Fediverse platforms, thanks to common usage of Activitypub. - **Interoperability**: Imagine being able to comment under a Youtube video directly from twitter.com, and having the comment shown under the video on youtube.com. Or following a Subreddit from Facebook. Such functionality is already available on the equivalent Fediverse platforms, thanks to common usage of Activitypub.
- **Ease of use**: From a user perspective, decentralized social media works almost identically to existing websites: a website with email and password based login. Unlike pure peer-to-peer networks, it is not necessary to handle private keys or install any local software. - **Ease of use**: From a user perspective, decentralized social media works almost identically to existing websites: a website with email and password based login. Unlike pure peer-to-peer networks, it is not necessary to handle private keys or install any local software.

View file

@ -39,4 +39,4 @@ let user: DbUser = webfinger_resolve_actor("ruud@lemmy.world", &data).await?;
# }).unwrap(); # }).unwrap();
``` ```
Note that webfinger queries don't contain a leading `@`. It is possible that there are multiple Activitypub IDs returned for a single webfinger query in case of multiple actors with the same name (for example Lemmy permits group and person with the same name). In this case `webfinger_resolve_actor` automatically loops and returns the first item which can be dereferenced successfully to the given type. Note that webfinger queries don't contain a leading `@`. It is possible tha there are multiple Activitypub IDs returned for a single webfinger query in case of multiple actors with the same name (for example Lemmy permits group and person with the same name). In this case `webfinger_resolve_actor` automatically loops and returns the first item which can be dereferenced successfully to the given type.

View file

@ -1,6 +1,6 @@
## Sending and receiving activities ## Sending and receiving activities
Activitypub propagates actions across servers using `Activities`. For this each actor has an inbox and a public/private key pair. We already defined a `Person` actor with keypair. Whats left is to define an activity. This is similar to the way we defined `Person` and `Note` structs before. In this case we need to implement the [Activity](trait@crate::traits::Activity) trait. Activitypub propagates actions across servers using `Activities`. For this each actor has an inbox and a public/private key pair. We already defined a `Person` actor with keypair. Whats left is to define an activity. This is similar to the way we defined `Person` and `Note` structs before. In this case we need to implement the [ActivityHandler](trait@crate::traits::ActivityHandler) trait.
``` ```
# use serde::{Deserialize, Serialize}; # use serde::{Deserialize, Serialize};
@ -10,7 +10,7 @@ Activitypub propagates actions across servers using `Activities`. For this each
# use activitypub_federation::fetch::object_id::ObjectId; # use activitypub_federation::fetch::object_id::ObjectId;
# use activitypub_federation::traits::tests::{DbConnection, DbUser}; # use activitypub_federation::traits::tests::{DbConnection, DbUser};
# use activitystreams_kinds::activity::FollowType; # use activitystreams_kinds::activity::FollowType;
# use activitypub_federation::traits::Activity; # use activitypub_federation::traits::ActivityHandler;
# use activitypub_federation::config::Data; # use activitypub_federation::config::Data;
# async fn send_accept() -> Result<(), Error> { Ok(()) } # async fn send_accept() -> Result<(), Error> { Ok(()) }
@ -25,7 +25,7 @@ pub struct Follow {
} }
#[async_trait] #[async_trait]
impl Activity for Follow { impl ActivityHandler for Follow {
type DataType = DbConnection; type DataType = DbConnection;
type Error = Error; type Error = Error;
@ -59,14 +59,14 @@ Next its time to setup the actual HTTP handler for the inbox. For this we first
# use activitypub_federation::axum::inbox::{ActivityData, receive_activity}; # use activitypub_federation::axum::inbox::{ActivityData, receive_activity};
# use activitypub_federation::config::Data; # use activitypub_federation::config::Data;
# use activitypub_federation::protocol::context::WithContext; # use activitypub_federation::protocol::context::WithContext;
# use activitypub_federation::traits::Activity; # use activitypub_federation::traits::ActivityHandler;
# use activitypub_federation::traits::tests::{DbConnection, DbUser, Follow}; # use activitypub_federation::traits::tests::{DbConnection, DbUser, Follow};
# use serde::{Deserialize, Serialize}; # use serde::{Deserialize, Serialize};
# use url::Url; # use url::Url;
#[derive(Deserialize, Serialize, Debug)] #[derive(Deserialize, Serialize, Debug)]
#[serde(untagged)] #[serde(untagged)]
#[enum_delegate::implement(Activity)] #[enum_delegate::implement(ActivityHandler)]
pub enum PersonAcceptedActivities { pub enum PersonAcceptedActivities {
Follow(Follow), Follow(Follow),
} }

View file

@ -32,13 +32,6 @@ impl Object for SearchableDbObjects {
type Kind = SearchableObjects; type Kind = SearchableObjects;
type Error = anyhow::Error; type Error = anyhow::Error;
fn id(&self) -> Url {
match self {
SearchableDbObjects::User(p) => p.federation_id.clone(),
SearchableDbObjects::Post(n) => n.federation_id.clone(),
}
}
async fn read_from_id( async fn read_from_id(
object_id: Url, object_id: Url,
data: &Data<Self::DataType>, data: &Data<Self::DataType>,

View file

@ -11,7 +11,7 @@ use activitypub_federation::{
fetch::object_id::ObjectId, fetch::object_id::ObjectId,
kinds::activity::CreateType, kinds::activity::CreateType,
protocol::{context::WithContext, helpers::deserialize_one_or_many}, protocol::{context::WithContext, helpers::deserialize_one_or_many},
traits::{Activity, Object}, traits::{ActivityHandler, Object},
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use url::Url; use url::Url;
@ -50,7 +50,7 @@ impl CreatePost {
} }
#[async_trait::async_trait] #[async_trait::async_trait]
impl Activity for CreatePost { impl ActivityHandler for CreatePost {
type DataType = DatabaseHandle; type DataType = DatabaseHandle;
type Error = crate::error::Error; type Error = crate::error::Error;

View file

@ -55,8 +55,8 @@ async fn main() -> Result<(), Error> {
info!("Listen with HTTP server on {BIND_ADDRESS}"); info!("Listen with HTTP server on {BIND_ADDRESS}");
let config = config.clone(); let config = config.clone();
let app = Router::new() let app = Router::new()
.route("/{user}", get(http_get_user)) .route("/:user", get(http_get_user))
.route("/{user}/inbox", post(http_post_user_inbox)) .route("/:user/inbox", post(http_post_user_inbox))
.route("/.well-known/webfinger", get(webfinger)) .route("/.well-known/webfinger", get(webfinger))
.layer(FederationMiddleware::new(config)); .layer(FederationMiddleware::new(config));

View file

@ -5,7 +5,7 @@ use activitypub_federation::{
http_signatures::generate_actor_keypair, http_signatures::generate_actor_keypair,
kinds::actor::PersonType, kinds::actor::PersonType,
protocol::{public_key::PublicKey, verification::verify_domains_match}, protocol::{public_key::PublicKey, verification::verify_domains_match},
traits::{Activity, Actor, Object}, traits::{ActivityHandler, Actor, Object},
}; };
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -29,7 +29,7 @@ pub struct DbUser {
/// List of all activities which this actor can receive. /// List of all activities which this actor can receive.
#[derive(Deserialize, Serialize, Debug)] #[derive(Deserialize, Serialize, Debug)]
#[serde(untagged)] #[serde(untagged)]
#[enum_delegate::implement(Activity)] #[enum_delegate::implement(ActivityHandler)]
pub enum PersonAcceptedActivities { pub enum PersonAcceptedActivities {
CreateNote(CreatePost), CreateNote(CreatePost),
} }
@ -69,10 +69,6 @@ impl Object for DbUser {
type Kind = Person; type Kind = Person;
type Error = Error; type Error = Error;
fn id(&self) -> Url {
self.ap_id.inner().clone()
}
fn last_refreshed_at(&self) -> Option<DateTime<Utc>> { fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
Some(self.last_refreshed_at) Some(self.last_refreshed_at)
} }
@ -126,6 +122,10 @@ impl Object for DbUser {
} }
impl Actor for DbUser { impl Actor for DbUser {
fn id(&self) -> Url {
self.ap_id.inner().clone()
}
fn public_key_pem(&self) -> &str { fn public_key_pem(&self) -> &str {
&self.public_key &self.public_key
} }

View file

@ -50,10 +50,6 @@ impl Object for DbPost {
type Kind = Note; type Kind = Note;
type Error = Error; type Error = Error;
fn id(&self) -> Url {
self.ap_id.inner().clone()
}
async fn read_from_id( async fn read_from_id(
_object_id: Url, _object_id: Url,
_data: &Data<Self::DataType>, _data: &Data<Self::DataType>,

View file

@ -3,7 +3,7 @@ use activitypub_federation::{
config::Data, config::Data,
fetch::object_id::ObjectId, fetch::object_id::ObjectId,
kinds::activity::AcceptType, kinds::activity::AcceptType,
traits::Activity, traits::ActivityHandler,
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use url::Url; use url::Url;
@ -30,7 +30,7 @@ impl Accept {
} }
#[async_trait::async_trait] #[async_trait::async_trait]
impl Activity for Accept { impl ActivityHandler for Accept {
type DataType = DatabaseHandle; type DataType = DatabaseHandle;
type Error = crate::error::Error; type Error = crate::error::Error;

View file

@ -8,7 +8,7 @@ use activitypub_federation::{
fetch::object_id::ObjectId, fetch::object_id::ObjectId,
kinds::activity::CreateType, kinds::activity::CreateType,
protocol::helpers::deserialize_one_or_many, protocol::helpers::deserialize_one_or_many,
traits::{Activity, Object}, traits::{ActivityHandler, Object},
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use url::Url; use url::Url;
@ -38,7 +38,7 @@ impl CreatePost {
} }
#[async_trait::async_trait] #[async_trait::async_trait]
impl Activity for CreatePost { impl ActivityHandler for CreatePost {
type DataType = DatabaseHandle; type DataType = DatabaseHandle;
type Error = crate::error::Error; type Error = crate::error::Error;

View file

@ -8,7 +8,7 @@ use activitypub_federation::{
config::Data, config::Data,
fetch::object_id::ObjectId, fetch::object_id::ObjectId,
kinds::activity::FollowType, kinds::activity::FollowType,
traits::{Activity, Actor}, traits::{ActivityHandler, Actor},
}; };
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use url::Url; use url::Url;
@ -35,7 +35,7 @@ impl Follow {
} }
#[async_trait::async_trait] #[async_trait::async_trait]
impl Activity for Follow { impl ActivityHandler for Follow {
type DataType = DatabaseHandle; type DataType = DatabaseHandle;
type Error = crate::error::Error; type Error = crate::error::Error;

View file

@ -8,7 +8,7 @@ use activitypub_federation::{
config::{Data, FederationConfig, FederationMiddleware}, config::{Data, FederationConfig, FederationMiddleware},
fetch::webfinger::{build_webfinger_response, extract_webfinger_name}, fetch::webfinger::{build_webfinger_response, extract_webfinger_name},
protocol::context::WithContext, protocol::context::WithContext,
traits::Object, traits::{Actor, Object},
FEDERATION_CONTENT_TYPE, FEDERATION_CONTENT_TYPE,
}; };
use actix_web::{web, web::Bytes, App, HttpRequest, HttpResponse, HttpServer}; use actix_web::{web, web::Bytes, App, HttpRequest, HttpResponse, HttpServer};

View file

@ -29,10 +29,9 @@ pub fn listen(config: &FederationConfig<DatabaseHandle>) -> Result<(), Error> {
let hostname = config.domain(); let hostname = config.domain();
info!("Listening with axum on {hostname}"); info!("Listening with axum on {hostname}");
let config = config.clone(); let config = config.clone();
let app = Router::new() let app = Router::new()
.route("/{user}/inbox", post(http_post_user_inbox)) .route("/:user/inbox", post(http_post_user_inbox))
.route("/{user}", get(http_get_user)) .route("/:user", get(http_get_user))
.route("/.well-known/webfinger", get(webfinger)) .route("/.well-known/webfinger", get(webfinger))
.layer(FederationMiddleware::new(config)); .layer(FederationMiddleware::new(config));

View file

@ -7,7 +7,6 @@ use crate::{
}; };
use error::Error; use error::Error;
use std::{env::args, str::FromStr}; use std::{env::args, str::FromStr};
use tokio::try_join;
use tracing::log::{info, LevelFilter}; use tracing::log::{info, LevelFilter};
mod activities; mod activities;
@ -35,10 +34,8 @@ async fn main() -> Result<(), Error> {
.map(|arg| Webserver::from_str(&arg).unwrap()) .map(|arg| Webserver::from_str(&arg).unwrap())
.unwrap_or(Webserver::Axum); .unwrap_or(Webserver::Axum);
let (alpha, beta) = try_join!( let alpha = new_instance("localhost:8001", "alpha".to_string()).await?;
new_instance("localhost:8001", "alpha".to_string()), let beta = new_instance("localhost:8002", "beta".to_string()).await?;
new_instance("localhost:8002", "beta".to_string())
)?;
listen(&alpha, &webserver)?; listen(&alpha, &webserver)?;
listen(&beta, &webserver)?; listen(&beta, &webserver)?;
info!("Local instances started"); info!("Local instances started");

View file

@ -13,7 +13,7 @@ use activitypub_federation::{
http_signatures::generate_actor_keypair, http_signatures::generate_actor_keypair,
kinds::actor::PersonType, kinds::actor::PersonType,
protocol::{context::WithContext, public_key::PublicKey, verification::verify_domains_match}, protocol::{context::WithContext, public_key::PublicKey, verification::verify_domains_match},
traits::{Activity, Actor, Object}, traits::{ActivityHandler, Actor, Object},
}; };
use chrono::{DateTime, Utc}; use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -37,7 +37,7 @@ pub struct DbUser {
/// List of all activities which this actor can receive. /// List of all activities which this actor can receive.
#[derive(Deserialize, Serialize, Debug)] #[derive(Deserialize, Serialize, Debug)]
#[serde(untagged)] #[serde(untagged)]
#[enum_delegate::implement(Activity)] #[enum_delegate::implement(ActivityHandler)]
pub enum PersonAcceptedActivities { pub enum PersonAcceptedActivities {
Follow(Follow), Follow(Follow),
Accept(Accept), Accept(Accept),
@ -103,16 +103,16 @@ impl DbUser {
Ok(()) Ok(())
} }
pub(crate) async fn send<A>( pub(crate) async fn send<Activity>(
&self, &self,
activity: A, activity: Activity,
recipients: Vec<Url>, recipients: Vec<Url>,
use_queue: bool, use_queue: bool,
data: &Data<DatabaseHandle>, data: &Data<DatabaseHandle>,
) -> Result<(), Error> ) -> Result<(), Error>
where where
A: Activity + Serialize + Debug + Send + Sync, Activity: ActivityHandler + Serialize + Debug + Send + Sync,
<A as Activity>::Error: From<anyhow::Error> + From<serde_json::Error>, <Activity as ActivityHandler>::Error: From<anyhow::Error> + From<serde_json::Error>,
{ {
let activity = WithContext::new_default(activity); let activity = WithContext::new_default(activity);
// Send through queue in some cases and bypass it in others to test both code paths // Send through queue in some cases and bypass it in others to test both code paths
@ -134,10 +134,6 @@ impl Object for DbUser {
type Kind = Person; type Kind = Person;
type Error = Error; type Error = Error;
fn id(&self) -> Url {
self.ap_id.inner().clone()
}
fn last_refreshed_at(&self) -> Option<DateTime<Utc>> { fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
Some(self.last_refreshed_at) Some(self.last_refreshed_at)
} }
@ -191,6 +187,10 @@ impl Object for DbUser {
} }
impl Actor for DbUser { impl Actor for DbUser {
fn id(&self) -> Url {
self.ap_id.inner().clone()
}
fn public_key_pem(&self) -> &str { fn public_key_pem(&self) -> &str {
&self.public_key &self.public_key
} }

View file

@ -47,10 +47,6 @@ impl Object for DbPost {
type Kind = Note; type Kind = Note;
type Error = Error; type Error = Error;
fn id(&self) -> Url {
self.ap_id.inner().clone()
}
async fn read_from_id( async fn read_from_id(
object_id: Url, object_id: Url,
data: &Data<Self::DataType>, data: &Data<Self::DataType>,

View file

@ -6,7 +6,7 @@ use crate::{
activity_sending::{build_tasks, SendActivityTask}, activity_sending::{build_tasks, SendActivityTask},
config::Data, config::Data,
error::Error, error::Error,
traits::{Activity, Actor}, traits::{ActivityHandler, Actor},
}; };
use futures_core::Future; use futures_core::Future;
@ -33,18 +33,18 @@ use url::Url;
/// ///
/// - `activity`: The activity to be sent, gets converted to json /// - `activity`: The activity to be sent, gets converted to json
/// - `private_key`: Private key belonging to the actor who sends the activity, for signing HTTP /// - `private_key`: Private key belonging to the actor who sends the activity, for signing HTTP
/// signature. Generated with [crate::http_signatures::generate_actor_keypair]. /// signature. Generated with [crate::http_signatures::generate_actor_keypair].
/// - `inboxes`: List of remote actor inboxes that should receive the activity. Ignores local actor /// - `inboxes`: List of remote actor inboxes that should receive the activity. Ignores local actor
/// inboxes. Should be built by calling [crate::traits::Actor::shared_inbox_or_inbox] /// inboxes. Should be built by calling [crate::traits::Actor::shared_inbox_or_inbox]
/// for each target actor. /// for each target actor.
pub async fn queue_activity<A, Datatype, ActorType>( pub async fn queue_activity<Activity, Datatype, ActorType>(
activity: &A, activity: &Activity,
actor: &ActorType, actor: &ActorType,
inboxes: Vec<Url>, inboxes: Vec<Url>,
data: &Data<Datatype>, data: &Data<Datatype>,
) -> Result<(), Error> ) -> Result<(), Error>
where where
A: Activity + Serialize + Debug, Activity: ActivityHandler + Serialize + Debug,
Datatype: Clone, Datatype: Clone,
ActorType: Actor, ActorType: Actor,
{ {

View file

@ -7,7 +7,7 @@ use crate::{
error::Error, error::Error,
http_signatures::sign_request, http_signatures::sign_request,
reqwest_shim::ResponseExt, reqwest_shim::ResponseExt,
traits::{Activity, Actor}, traits::{ActivityHandler, Actor},
FEDERATION_CONTENT_TYPE, FEDERATION_CONTENT_TYPE,
}; };
use bytes::Bytes; use bytes::Bytes;
@ -52,16 +52,16 @@ impl SendActivityTask {
/// ///
/// - `activity`: The activity to be sent, gets converted to json /// - `activity`: The activity to be sent, gets converted to json
/// - `inboxes`: List of remote actor inboxes that should receive the activity. Ignores local actor /// - `inboxes`: List of remote actor inboxes that should receive the activity. Ignores local actor
/// inboxes. Should be built by calling [crate::traits::Actor::shared_inbox_or_inbox] /// inboxes. Should be built by calling [crate::traits::Actor::shared_inbox_or_inbox]
/// for each target actor. /// for each target actor.
pub async fn prepare<A, Datatype, ActorType>( pub async fn prepare<Activity, Datatype, ActorType>(
activity: &A, activity: &Activity,
actor: &ActorType, actor: &ActorType,
inboxes: Vec<Url>, inboxes: Vec<Url>,
data: &Data<Datatype>, data: &Data<Datatype>,
) -> Result<Vec<SendActivityTask>, Error> ) -> Result<Vec<SendActivityTask>, Error>
where where
A: Activity + Serialize + Debug, Activity: ActivityHandler + Serialize + Debug,
Datatype: Clone, Datatype: Clone,
ActorType: Actor, ActorType: Actor,
{ {
@ -136,14 +136,14 @@ impl SendActivityTask {
} }
} }
pub(crate) async fn build_tasks<A, Datatype, ActorType>( pub(crate) async fn build_tasks<'a, Activity, Datatype, ActorType>(
activity: &A, activity: &'a Activity,
actor: &ActorType, actor: &ActorType,
inboxes: Vec<Url>, inboxes: Vec<Url>,
data: &Data<Datatype>, data: &Data<Datatype>,
) -> Result<Vec<SendActivityTask>, Error> ) -> Result<Vec<SendActivityTask>, Error>
where where
A: Activity + Serialize + Debug, Activity: ActivityHandler + Serialize + Debug,
Datatype: Clone, Datatype: Clone,
ActorType: Actor, ActorType: Actor,
{ {

View file

@ -6,7 +6,7 @@ use crate::{
error::Error, error::Error,
http_signatures::{verify_body_hash, verify_signature}, http_signatures::{verify_body_hash, verify_signature},
parse_received_activity, parse_received_activity,
traits::{Activity, Actor, Object}, traits::{ActivityHandler, Actor, Object},
}; };
use actix_web::{web::Bytes, HttpRequest, HttpResponse}; use actix_web::{web::Bytes, HttpRequest, HttpResponse};
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
@ -14,77 +14,17 @@ use tracing::debug;
/// Handles incoming activities, verifying HTTP signatures and other checks /// Handles incoming activities, verifying HTTP signatures and other checks
/// ///
/// After successful validation, activities are passed to respective [trait@Activity]. /// After successful validation, activities are passed to respective [trait@ActivityHandler].
pub async fn receive_activity<A, ActorT, Datatype>( pub async fn receive_activity<Activity, ActorT, Datatype>(
request: HttpRequest, request: HttpRequest,
body: Bytes, body: Bytes,
data: &Data<Datatype>, data: &Data<Datatype>,
) -> Result<HttpResponse, <A as Activity>::Error> ) -> Result<HttpResponse, <Activity as ActivityHandler>::Error>
where where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + 'static, Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Sync + 'static, ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>, for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>, <Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>,
Datatype: Clone,
{
let (activity, _) = do_stuff::<A, ActorT, Datatype>(request, body, data).await?;
do_more_stuff(activity, data).await
}
/// Workaround required so we can use references for the hook, instead of cloning data.
pub trait ReceiveActivityHook<A, ActorT, Datatype>
where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + Clone + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Clone + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>,
Datatype: Clone,
{
/// Called when a new activity is recived
fn hook(
self,
activity: &A,
actor: &ActorT,
data: &Data<Datatype>,
) -> impl std::future::Future<Output = Result<(), <A as Activity>::Error>>;
}
/// Same as [receive_activity], only that it calls the provided hook function before
/// calling activity verify and receive functions.
pub async fn receive_activity_with_hook<A, ActorT, Datatype>(
request: HttpRequest,
body: Bytes,
hook: impl ReceiveActivityHook<A, ActorT, Datatype>,
data: &Data<Datatype>,
) -> Result<HttpResponse, <A as Activity>::Error>
where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + Clone + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Sync + Clone + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>,
Datatype: Clone,
{
let (activity, actor) = do_stuff::<A, ActorT, Datatype>(request, body, data).await?;
hook.hook(&activity, &actor, data).await?;
do_more_stuff(activity, data).await
}
async fn do_stuff<A, ActorT, Datatype>(
request: HttpRequest,
body: Bytes,
data: &Data<Datatype>,
) -> Result<(A, ActorT), <A as Activity>::Error>
where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Sync + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>, <ActorT as Object>::Error: From<Error>,
Datatype: Clone, Datatype: Clone,
{ {
@ -94,24 +34,13 @@ where
.map(http_compat::header_value); .map(http_compat::header_value);
verify_body_hash(digest_header.as_ref(), &body)?; verify_body_hash(digest_header.as_ref(), &body)?;
let (activity, actor) = parse_received_activity::<A, ActorT, _>(&body, data).await?; let (activity, actor) = parse_received_activity::<Activity, ActorT, _>(&body, data).await?;
let headers = http_compat::header_map(request.headers()); let headers = http_compat::header_map(request.headers());
let method = http_compat::method(request.method()); let method = http_compat::method(request.method());
let uri = http_compat::uri(request.uri()); let uri = http_compat::uri(request.uri());
verify_signature(&headers, &method, &uri, actor.public_key_pem())?; verify_signature(&headers, &method, &uri, actor.public_key_pem())?;
Ok((activity, actor))
}
async fn do_more_stuff<A, Datatype>(
activity: A,
data: &Data<Datatype>,
) -> Result<HttpResponse, <A as Activity>::Error>
where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + 'static,
Datatype: Clone,
{
debug!("Receiving activity {}", activity.id().to_string()); debug!("Receiving activity {}", activity.id().to_string());
activity.verify(data).await?; activity.verify(data).await?;
activity.receive(data).await?; activity.receive(data).await?;
@ -146,38 +75,15 @@ mod test {
} }
#[tokio::test] #[tokio::test]
async fn test_receive_activity_hook() { async fn test_receive_activity() {
let (body, incoming_request, config) = setup_receive_test().await; let (body, incoming_request, config) = setup_receive_test().await;
let res = receive_activity_with_hook::<Follow, DbUser, DbConnection>( receive_activity::<Follow, DbUser, DbConnection>(
incoming_request.to_http_request(), incoming_request.to_http_request(),
body, body,
Dummy,
&config.to_request_data(), &config.to_request_data(),
) )
.await; .await
assert_eq!(res.err(), Some(Error::Other("test-error".to_string()))); .unwrap();
}
struct Dummy;
impl<A, ActorT, Datatype> ReceiveActivityHook<A, ActorT, Datatype> for Dummy
where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + Clone + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Clone + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>,
Datatype: Clone,
{
async fn hook(
self,
_activity: &A,
_actor: &ActorT,
_data: &Data<Datatype>,
) -> Result<(), <A as Activity>::Error> {
// ensure that hook gets called by returning this value
Err(Error::Other("test-error".to_string()).into())
}
} }
#[tokio::test] #[tokio::test]
@ -216,14 +122,14 @@ mod test {
let (_, _, config) = setup_receive_test().await; let (_, _, config) = setup_receive_test().await;
let actor = Url::parse("http://ds9.lemmy.ml/u/lemmy_alpha").unwrap(); let actor = Url::parse("http://ds9.lemmy.ml/u/lemmy_alpha").unwrap();
let activity_id = "http://localhost:123/1"; let id = "http://localhost:123/1";
let activity = json!({ let activity = json!({
"actor": actor.as_str(), "actor": actor.as_str(),
"to": ["https://www.w3.org/ns/activitystreams#Public"], "to": ["https://www.w3.org/ns/activitystreams#Public"],
"object": "http://ds9.lemmy.ml/post/1", "object": "http://ds9.lemmy.ml/post/1",
"cc": ["http://enterprise.lemmy.ml/c/main"], "cc": ["http://enterprise.lemmy.ml/c/main"],
"type": "Delete", "type": "Delete",
"id": activity_id "id": id
} }
); );
let body: Bytes = serde_json::to_vec(&activity).unwrap().into(); let body: Bytes = serde_json::to_vec(&activity).unwrap().into();
@ -238,8 +144,8 @@ mod test {
.await; .await;
match res { match res {
Err(Error::ParseReceivedActivity { err: _, id }) => { Err(Error::ParseReceivedActivity(_, url)) => {
assert_eq!(activity_id, id.expect("has url").as_str()); assert_eq!(id, url.expect("has url").as_str());
} }
_ => unreachable!(), _ => unreachable!(),
} }

View file

@ -4,7 +4,6 @@ mod http_compat;
pub mod inbox; pub mod inbox;
#[doc(hidden)] #[doc(hidden)]
pub mod middleware; pub mod middleware;
pub mod response;
use crate::{ use crate::{
config::Data, config::Data,
@ -23,7 +22,7 @@ pub async fn signing_actor<A>(
data: &Data<<A as Object>::DataType>, data: &Data<<A as Object>::DataType>,
) -> Result<A, <A as Object>::Error> ) -> Result<A, <A as Object>::Error>
where where
A: Object + Actor + Send + Sync, A: Object + Actor,
<A as Object>::Error: From<Error>, <A as Object>::Error: From<Error>,
for<'de2> <A as Object>::Kind: Deserialize<'de2>, for<'de2> <A as Object>::Kind: Deserialize<'de2>,
{ {

View file

@ -1,50 +0,0 @@
//! Generate HTTP responses for Activitypub ojects
use crate::{
protocol::{context::WithContext, tombstone::Tombstone},
FEDERATION_CONTENT_TYPE,
};
use actix_web::HttpResponse;
use http02::header::VARY;
use serde::Serialize;
use serde_json::Value;
use url::Url;
/// Generates HTTP response to serve the object for fetching from other instances.
///
/// If possible use [Object.http_response]
/// which also handles redirects for remote objects and deletions.
///
/// `federation_context` is the value of `@context`.
pub fn create_http_response<T: Serialize>(
data: T,
federation_context: &Value,
) -> Result<HttpResponse, serde_json::Error> {
let json = serde_json::to_string_pretty(&WithContext::new(data, federation_context.clone()))?;
Ok(HttpResponse::Ok()
.content_type(FEDERATION_CONTENT_TYPE)
.insert_header((VARY, "Accept"))
.body(json))
}
pub(crate) fn create_tombstone_response(
id: Url,
federation_context: &Value,
) -> Result<HttpResponse, serde_json::Error> {
let tombstone = Tombstone::new(id);
let json =
serde_json::to_string_pretty(&WithContext::new(tombstone, federation_context.clone()))?;
Ok(HttpResponse::Gone()
.content_type(FEDERATION_CONTENT_TYPE)
.status(actix_web::http::StatusCode::GONE)
.insert_header((VARY, "Accept"))
.body(json))
}
pub(crate) fn redirect_remote_object(url: &Url) -> HttpResponse {
let mut res = HttpResponse::PermanentRedirect();
res.insert_header((actix_web::http::header::LOCATION, url.as_str()));
res.finish()
}

View file

@ -7,9 +7,10 @@ use crate::{
error::Error, error::Error,
http_signatures::verify_signature, http_signatures::verify_signature,
parse_received_activity, parse_received_activity,
traits::{Activity, Actor, Object}, traits::{ActivityHandler, Actor, Object},
}; };
use axum::{ use axum::{
async_trait,
body::Body, body::Body,
extract::FromRequest, extract::FromRequest,
http::{Request, StatusCode}, http::{Request, StatusCode},
@ -20,20 +21,20 @@ use serde::de::DeserializeOwned;
use tracing::debug; use tracing::debug;
/// Handles incoming activities, verifying HTTP signatures and other checks /// Handles incoming activities, verifying HTTP signatures and other checks
pub async fn receive_activity<A, ActorT, Datatype>( pub async fn receive_activity<Activity, ActorT, Datatype>(
activity_data: ActivityData, activity_data: ActivityData,
data: &Data<Datatype>, data: &Data<Datatype>,
) -> Result<(), <A as Activity>::Error> ) -> Result<(), <Activity as ActivityHandler>::Error>
where where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + 'static, Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Sync + 'static, ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>, for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>, <Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>, <ActorT as Object>::Error: From<Error>,
Datatype: Clone, Datatype: Clone,
{ {
let (activity, actor) = let (activity, actor) =
parse_received_activity::<A, ActorT, _>(&activity_data.body, data).await?; parse_received_activity::<Activity, ActorT, _>(&activity_data.body, data).await?;
verify_signature( verify_signature(
&activity_data.headers, &activity_data.headers,
@ -57,6 +58,7 @@ pub struct ActivityData {
body: Vec<u8>, body: Vec<u8>,
} }
#[async_trait]
impl<S> FromRequest<S> for ActivityData impl<S> FromRequest<S> for ActivityData
where where
S: Send + Sync, S: Send + Sync,
@ -64,21 +66,7 @@ where
type Rejection = Response; type Rejection = Response;
async fn from_request(req: Request<Body>, _state: &S) -> Result<Self, Self::Rejection> { async fn from_request(req: Request<Body>, _state: &S) -> Result<Self, Self::Rejection> {
#[allow(unused_mut)] let (parts, body) = req.into_parts();
let (mut parts, body) = req.into_parts();
// take the full URI to handle nested routers
// OriginalUri::from_request_parts has an Infallible error type
#[cfg(feature = "axum-original-uri")]
let uri = {
use axum::extract::{FromRequestParts, OriginalUri};
OriginalUri::from_request_parts(&mut parts, _state)
.await
.expect("infallible")
.0
};
#[cfg(not(feature = "axum-original-uri"))]
let uri = parts.uri;
// this wont work if the body is an long running stream // this wont work if the body is an long running stream
let bytes = axum::body::to_bytes(body, usize::MAX) let bytes = axum::body::to_bytes(body, usize::MAX)
@ -88,7 +76,7 @@ where
Ok(Self { Ok(Self {
headers: parts.headers, headers: parts.headers,
method: parts.method, method: parts.method,
uri, uri: parts.uri,
body: bytes.to_vec(), body: bytes.to_vec(),
}) })
} }

View file

@ -1,5 +1,5 @@
use crate::config::{Data, FederationConfig, FederationMiddleware}; use crate::config::{Data, FederationConfig, FederationMiddleware};
use axum::{body::Body, extract::FromRequestParts, http::Request, response::Response}; use axum::{async_trait, body::Body, extract::FromRequestParts, http::Request, response::Response};
use http::{request::Parts, StatusCode}; use http::{request::Parts, StatusCode};
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use tower::{Layer, Service}; use tower::{Layer, Service};
@ -43,6 +43,7 @@ where
} }
} }
#[async_trait]
impl<S, T: Clone + 'static> FromRequestParts<S> for Data<T> impl<S, T: Clone + 'static> FromRequestParts<S> for Data<T>
where where
S: Send + Sync, S: Send + Sync,

View file

@ -19,16 +19,14 @@ use crate::{
error::Error, error::Error,
http_signatures::sign_request, http_signatures::sign_request,
protocol::verification::verify_domains_match, protocol::verification::verify_domains_match,
traits::{Activity, Actor}, traits::{ActivityHandler, Actor},
utils::validate_ip,
}; };
use async_trait::async_trait; use async_trait::async_trait;
use bytes::Bytes; use bytes::Bytes;
use derive_builder::Builder; use derive_builder::Builder;
use dyn_clone::{clone_trait_object, DynClone}; use dyn_clone::{clone_trait_object, DynClone};
use moka::future::Cache; use moka::future::Cache;
use regex::Regex; use reqwest::Request;
use reqwest::{redirect::Policy, Client, Request};
use reqwest_middleware::{ClientWithMiddleware, RequestBuilder}; use reqwest_middleware::{ClientWithMiddleware, RequestBuilder};
use rsa::{pkcs8::DecodePrivateKey, RsaPrivateKey}; use rsa::{pkcs8::DecodePrivateKey, RsaPrivateKey};
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
@ -37,7 +35,6 @@ use std::{
sync::{ sync::{
atomic::{AtomicU32, Ordering}, atomic::{AtomicU32, Ordering},
Arc, Arc,
OnceLock,
}, },
time::Duration, time::Duration,
}; };
@ -57,14 +54,9 @@ pub struct FederationConfig<T: Clone> {
/// [crate::fetch::object_id::ObjectId] for more details. /// [crate::fetch::object_id::ObjectId] for more details.
#[builder(default = "20")] #[builder(default = "20")]
pub(crate) http_fetch_limit: u32, pub(crate) http_fetch_limit: u32,
#[builder(default = "default_client()")] #[builder(default = "reqwest::Client::default().into()")]
/// HTTP client used for all outgoing requests. When passing a custom client here you should /// HTTP client used for all outgoing requests. Middleware can be used to add functionality
/// also disable redirects and set timeouts. /// like log tracing or retry of failed requests.
///
/// Middleware can be used to add functionality like log tracing or retry of failed requests.
/// Redirects are disabled by default, because automatic redirect URLs can't be validated.
/// Instead a single redirect is handled manually. The default client sets a timeout of 10s
/// to avoid excessive resource usage when connecting to dead servers.
pub(crate) client: ClientWithMiddleware, pub(crate) client: ClientWithMiddleware,
/// Run library in debug mode. This allows usage of http and localhost urls. It also sends /// Run library in debug mode. This allows usage of http and localhost urls. It also sends
/// outgoing activities synchronously, not in background thread. This helps to make tests /// outgoing activities synchronously, not in background thread. This helps to make tests
@ -113,20 +105,18 @@ pub struct FederationConfig<T: Clone> {
pub(crate) queue_retry_count: usize, pub(crate) queue_retry_count: usize,
} }
pub(crate) fn domain_regex() -> &'static Regex {
static DOMAIN_REGEX: OnceLock<Regex> = OnceLock::new();
DOMAIN_REGEX.get_or_init(|| Regex::new(r"^[a-zA-Z0-9.-]*$").expect("compile regex"))
}
impl<T: Clone> FederationConfig<T> { impl<T: Clone> FederationConfig<T> {
/// Returns a new config builder with default values. /// Returns a new config builder with default values.
pub fn builder() -> FederationConfigBuilder<T> { pub fn builder() -> FederationConfigBuilder<T> {
FederationConfigBuilder::default() FederationConfigBuilder::default()
} }
pub(crate) async fn verify_url_and_domain<A, Datatype>(&self, activity: &A) -> Result<(), Error> pub(crate) async fn verify_url_and_domain<Activity, Datatype>(
&self,
activity: &Activity,
) -> Result<(), Error>
where where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + 'static, Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
{ {
verify_domains_match(activity.id(), activity.actor())?; verify_domains_match(activity.id(), activity.actor())?;
self.verify_url_valid(activity.id()).await?; self.verify_url_valid(activity.id()).await?;
@ -169,35 +159,17 @@ impl<T: Clone> FederationConfig<T> {
return Ok(()); return Ok(());
} }
let Some(domain) = url.domain() else { if url.domain().is_none() {
return Err(Error::UrlVerificationError("Url must have a domain")); return Err(Error::UrlVerificationError("Url must have a domain"));
};
if !domain_regex().is_match(domain) {
return Err(Error::UrlVerificationError("Invalid characters in domain"));
} }
// Extra checks only for production mode if url.domain() == Some("localhost") && !self.debug {
if !self.debug { return Err(Error::UrlVerificationError(
if url.port().is_some() { "Localhost is only allowed in debug mode",
return Err(Error::UrlVerificationError("Explicit port is not allowed")); ));
}
let allow_local = std::env::var("DANGER_FEDERATION_ALLOW_LOCAL_IP").is_ok();
if !allow_local && validate_ip(&url).await.is_err() {
return Err(Error::DomainResolveError(domain.to_string()));
}
} }
// It is valid but uncommon for domains to end with `.` char. Drop this so it cant be used self.url_verifier.verify(url).await?;
// to bypass domain blocklist. Avoid cloning url in common case.
if domain.ends_with('.') {
let mut url = url.clone();
let domain = &domain[0..domain.len() - 1];
url.set_host(Some(domain))?;
self.url_verifier.verify(&url).await?;
} else {
self.url_verifier.verify(url).await?;
}
Ok(()) Ok(())
} }
@ -233,7 +205,7 @@ impl<T: Clone> FederationConfigBuilder<T> {
let private_key = let private_key =
RsaPrivateKey::from_pkcs8_pem(&private_key_pem).expect("Could not decode PEM data"); RsaPrivateKey::from_pkcs8_pem(&private_key_pem).expect("Could not decode PEM data");
self.signed_fetch_actor = Some(Some(Arc::new((actor.id().clone(), private_key)))); self.signed_fetch_actor = Some(Some(Arc::new((actor.id(), private_key))));
self self
} }
@ -331,10 +303,9 @@ clone_trait_object!(UrlVerifier);
/// prevent denial of service attacks, where an attacker triggers fetching of recursive objects. /// prevent denial of service attacks, where an attacker triggers fetching of recursive objects.
/// ///
/// <https://www.w3.org/TR/activitypub/#security-recursive-objects> /// <https://www.w3.org/TR/activitypub/#security-recursive-objects>
#[derive(Clone)]
pub struct Data<T: Clone> { pub struct Data<T: Clone> {
pub(crate) config: FederationConfig<T>, pub(crate) config: FederationConfig<T>,
pub(crate) request_counter: RequestCounter, pub(crate) request_counter: AtomicU32,
} }
impl<T: Clone> Data<T> { impl<T: Clone> Data<T> {
@ -357,7 +328,7 @@ impl<T: Clone> Data<T> {
} }
/// Total number of outgoing HTTP requests made with this data. /// Total number of outgoing HTTP requests made with this data.
pub fn request_count(&self) -> u32 { pub fn request_count(&self) -> u32 {
self.request_counter.0.load(Ordering::Relaxed) self.request_counter.load(Ordering::Relaxed)
} }
/// Add HTTP signature to arbitrary request /// Add HTTP signature to arbitrary request
@ -378,15 +349,6 @@ impl<T: Clone> Data<T> {
) )
.await .await
} }
/// Resolve domain of the url and throw error if it points to local/private IP.
pub async fn is_valid_ip(&self, url: &Url) -> Result<(), Error> {
if self.config.debug {
return Ok(());
}
validate_ip(url).await
}
} }
impl<T: Clone> Deref for Data<T> { impl<T: Clone> Deref for Data<T> {
@ -397,16 +359,6 @@ impl<T: Clone> Deref for Data<T> {
} }
} }
/// Wrapper to implement `Clone`
#[derive(Default)]
pub(crate) struct RequestCounter(pub(crate) AtomicU32);
impl Clone for RequestCounter {
fn clone(&self) -> Self {
RequestCounter(self.0.load(Ordering::Relaxed).into())
}
}
/// Middleware for HTTP handlers which provides access to [Data] /// Middleware for HTTP handlers which provides access to [Data]
#[derive(Clone)] #[derive(Clone)]
pub struct FederationMiddleware<T: Clone>(pub(crate) FederationConfig<T>); pub struct FederationMiddleware<T: Clone>(pub(crate) FederationConfig<T>);
@ -418,17 +370,6 @@ impl<T: Clone> FederationMiddleware<T> {
} }
} }
fn default_client() -> ClientWithMiddleware {
let timeout = Duration::from_secs(10);
Client::builder()
.redirect(Policy::none())
.timeout(timeout)
.connect_timeout(timeout)
.build()
.unwrap_or_else(|_| Client::default())
.into()
}
#[cfg(test)] #[cfg(test)]
#[allow(clippy::unwrap_used)] #[allow(clippy::unwrap_used)]
mod test { mod test {

View file

@ -28,9 +28,6 @@ pub enum Error {
/// url verification error /// url verification error
#[error("URL failed verification: {0}")] #[error("URL failed verification: {0}")]
UrlVerificationError(&'static str), UrlVerificationError(&'static str),
/// Resolving domain points to local IP.
#[error("Resolving domain {0} points to local IP address. This may indicate an attacker attempting to access internal services. If intentional, you can ignore this error by setting DANGER_FEDERATION_ALLOW_LOCAL_IP=1")]
DomainResolveError(String),
/// Incoming activity has invalid digest for body /// Incoming activity has invalid digest for body
#[error("Incoming activity has invalid digest for body")] #[error("Incoming activity has invalid digest for body")]
ActivityBodyDigestInvalid, ActivityBodyDigestInvalid,
@ -47,16 +44,11 @@ pub enum Error {
#[error("Failed to parse object {1} with content {2}: {0}")] #[error("Failed to parse object {1} with content {2}: {0}")]
ParseFetchedObject(serde_json::Error, Url, String), ParseFetchedObject(serde_json::Error, Url, String),
/// Failed to parse an activity received from another instance /// Failed to parse an activity received from another instance
#[error("Failed to parse incoming activity {}: {0}", match .id { #[error("Failed to parse incoming activity {}: {0}", match .1 {
Some(t) => format!("with id {t}"), Some(t) => format!("with id {t}"),
None => String::new(), None => String::new(),
})] })]
ParseReceivedActivity { ParseReceivedActivity(serde_json::Error, Option<Url>),
/// The parse error
err: serde_json::Error,
/// ID of the Activitypub object which caused this error
id: Option<Url>,
},
/// Reqwest Middleware Error /// Reqwest Middleware Error
#[error(transparent)] #[error(transparent)]
ReqwestMiddleware(#[from] reqwest_middleware::Error), ReqwestMiddleware(#[from] reqwest_middleware::Error),
@ -86,9 +78,6 @@ pub enum Error {
/// Attempted to fetch object but the response's id field doesn't match /// Attempted to fetch object but the response's id field doesn't match
#[error("Attempted to fetch object from {0} but the response's id field doesn't match")] #[error("Attempted to fetch object from {0} but the response's id field doesn't match")]
FetchWrongId(Url), FetchWrongId(Url),
/// I/O error from OS
#[error(transparent)]
IoError(#[from] std::io::Error),
/// Other generic errors /// Other generic errors
#[error("{0}")] #[error("{0}")]
Other(String), Other(String),

View file

@ -102,3 +102,92 @@ where
self.0.eq(&other.0) && self.1 == other.1 self.0.eq(&other.0) && self.1 == other.1
} }
} }
#[cfg(feature = "diesel")]
const _IMPL_DIESEL_NEW_TYPE_FOR_COLLECTION_ID: () = {
use diesel::{
backend::Backend,
deserialize::{FromSql, FromStaticSqlRow},
expression::AsExpression,
internal::derives::as_expression::Bound,
pg::Pg,
query_builder::QueryId,
serialize,
serialize::{Output, ToSql},
sql_types::{HasSqlType, SingleValue, Text},
Expression,
Queryable,
};
// TODO: this impl only works for Postgres db because of to_string() call which requires reborrow
impl<Kind, ST> ToSql<ST, Pg> for CollectionId<Kind>
where
Kind: Collection,
for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
String: ToSql<ST, Pg>,
{
fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
let v = self.0.to_string();
<String as ToSql<Text, Pg>>::to_sql(&v, &mut out.reborrow())
}
}
impl<'expr, Kind, ST> AsExpression<ST> for &'expr CollectionId<Kind>
where
Kind: Collection,
for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
Bound<ST, String>: Expression<SqlType = ST>,
ST: SingleValue,
{
type Expression = Bound<ST, &'expr str>;
fn as_expression(self) -> Self::Expression {
Bound::new(self.0.as_str())
}
}
impl<Kind, ST> AsExpression<ST> for CollectionId<Kind>
where
Kind: Collection,
for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
Bound<ST, String>: Expression<SqlType = ST>,
ST: SingleValue,
{
type Expression = Bound<ST, String>;
fn as_expression(self) -> Self::Expression {
Bound::new(self.0.to_string())
}
}
impl<Kind, ST, DB> FromSql<ST, DB> for CollectionId<Kind>
where
Kind: Collection + Send + 'static,
for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
String: FromSql<ST, DB>,
DB: Backend,
DB: HasSqlType<ST>,
{
fn from_sql(
raw: DB::RawValue<'_>,
) -> Result<Self, Box<dyn ::std::error::Error + Send + Sync>> {
let string: String = FromSql::<ST, DB>::from_sql(raw)?;
Ok(CollectionId::parse(&string)?)
}
}
impl<Kind, ST, DB> Queryable<ST, DB> for CollectionId<Kind>
where
Kind: Collection + Send + 'static,
for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
String: FromStaticSqlRow<ST, DB>,
DB: Backend,
DB: HasSqlType<ST>,
{
type Row = String;
fn build(row: Self::Row) -> diesel::deserialize::Result<Self> {
Ok(CollectionId::parse(&row)?)
}
}
impl<Kind> QueryId for CollectionId<Kind>
where
Kind: Collection + 'static,
for<'de2> <Kind as Collection>::Kind: Deserialize<'de2>,
{
type QueryId = Self;
}
};

View file

@ -11,7 +11,7 @@ use crate::{
FEDERATION_CONTENT_TYPE, FEDERATION_CONTENT_TYPE,
}; };
use bytes::Bytes; use bytes::Bytes;
use http::{header::LOCATION, HeaderValue, StatusCode}; use http::{HeaderValue, StatusCode};
use serde::de::DeserializeOwned; use serde::de::DeserializeOwned;
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
use tracing::info; use tracing::info;
@ -59,7 +59,7 @@ pub async fn fetch_object_http<T: Clone, Kind: DeserializeOwned>(
r#"application/ld+json; profile="https://www.w3.org/ns/activitystreams""#, // activitypub standard r#"application/ld+json; profile="https://www.w3.org/ns/activitystreams""#, // activitypub standard
r#"application/activity+json; charset=utf-8"#, // mastodon r#"application/activity+json; charset=utf-8"#, // mastodon
]; ];
let res = fetch_object_http_with_accept(url, data, &FETCH_CONTENT_TYPE, false).await?; let res = fetch_object_http_with_accept(url, data, &FETCH_CONTENT_TYPE).await?;
// Ensure correct content-type to prevent vulnerabilities, with case insensitive comparison. // Ensure correct content-type to prevent vulnerabilities, with case insensitive comparison.
let content_type = res let content_type = res
@ -74,7 +74,6 @@ pub async fn fetch_object_http<T: Clone, Kind: DeserializeOwned>(
// Ensure id field matches final url after redirect // Ensure id field matches final url after redirect
if res.object_id.as_ref() != Some(&res.url) { if res.object_id.as_ref() != Some(&res.url) {
if let Some(res_object_id) = res.object_id { if let Some(res_object_id) = res.object_id {
data.config.verify_url_valid(&res_object_id).await?;
// If id is different but still on the same domain, attempt to request object // If id is different but still on the same domain, attempt to request object
// again from url in id field. // again from url in id field.
if res_object_id.domain() == res.url.domain() { if res_object_id.domain() == res.url.domain() {
@ -100,13 +99,12 @@ async fn fetch_object_http_with_accept<T: Clone, Kind: DeserializeOwned>(
url: &Url, url: &Url,
data: &Data<T>, data: &Data<T>,
content_type: &HeaderValue, content_type: &HeaderValue,
recursive: bool,
) -> Result<FetchObjectResponse<Kind>, Error> { ) -> Result<FetchObjectResponse<Kind>, Error> {
let config = &data.config; let config = &data.config;
config.verify_url_valid(url).await?; config.verify_url_valid(url).await?;
info!("Fetching remote object {}", url.to_string()); info!("Fetching remote object {}", url.to_string());
let mut counter = data.request_counter.0.fetch_add(1, Ordering::SeqCst); let mut counter = data.request_counter.fetch_add(1, Ordering::SeqCst);
// fetch_add returns old value so we need to increment manually here // fetch_add returns old value so we need to increment manually here
counter += 1; counter += 1;
if counter > config.http_fetch_limit { if counter > config.http_fetch_limit {
@ -133,19 +131,6 @@ async fn fetch_object_http_with_accept<T: Clone, Kind: DeserializeOwned>(
req.send().await? req.send().await?
}; };
// Allow a single redirect using recursion. Further redirects are ignored.
let location = res.headers().get(LOCATION).and_then(|l| l.to_str().ok());
if let (Some(location), false) = (location, recursive) {
let location = location.parse()?;
return Box::pin(fetch_object_http_with_accept(
&location,
data,
content_type,
true,
))
.await;
}
if res.status() == StatusCode::GONE { if res.status() == StatusCode::GONE {
return Err(Error::ObjectDeleted(url.clone())); return Err(Error::ObjectDeleted(url.clone()));
} }

View file

@ -10,7 +10,7 @@ use url::Url;
impl<T> FromStr for ObjectId<T> impl<T> FromStr for ObjectId<T>
where where
T: Object + Send + Sync + Debug + 'static, T: Object + Send + Debug + 'static,
for<'de2> <T as Object>::Kind: Deserialize<'de2>, for<'de2> <T as Object>::Kind: Deserialize<'de2>,
{ {
type Err = url::ParseError; type Err = url::ParseError;
@ -61,7 +61,7 @@ where
impl<Kind> ObjectId<Kind> impl<Kind> ObjectId<Kind>
where where
Kind: Object + Send + Sync + Debug + 'static, Kind: Object + Send + Debug + 'static,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>, for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
{ {
/// Construct a new objectid instance /// Construct a new objectid instance
@ -164,7 +164,6 @@ where
if let Err(Error::ObjectDeleted(url)) = res { if let Err(Error::ObjectDeleted(url)) = res {
if let Some(db_object) = db_object { if let Some(db_object) = db_object {
db_object.delete(data).await?; db_object.delete(data).await?;
return Ok(db_object);
} }
return Err(Error::ObjectDeleted(url).into()); return Err(Error::ObjectDeleted(url).into());
} }
@ -272,7 +271,95 @@ where
} }
} }
/// Internal only #[cfg(feature = "diesel")]
const _IMPL_DIESEL_NEW_TYPE_FOR_OBJECT_ID: () = {
use diesel::{
backend::Backend,
deserialize::{FromSql, FromStaticSqlRow},
expression::AsExpression,
internal::derives::as_expression::Bound,
pg::Pg,
query_builder::QueryId,
serialize,
serialize::{Output, ToSql},
sql_types::{HasSqlType, SingleValue, Text},
Expression,
Queryable,
};
// TODO: this impl only works for Postgres db because of to_string() call which requires reborrow
impl<Kind, ST> ToSql<ST, Pg> for ObjectId<Kind>
where
Kind: Object,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
String: ToSql<ST, Pg>,
{
fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Pg>) -> serialize::Result {
let v = self.0.to_string();
<String as ToSql<Text, Pg>>::to_sql(&v, &mut out.reborrow())
}
}
impl<'expr, Kind, ST> AsExpression<ST> for &'expr ObjectId<Kind>
where
Kind: Object,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
Bound<ST, String>: Expression<SqlType = ST>,
ST: SingleValue,
{
type Expression = Bound<ST, &'expr str>;
fn as_expression(self) -> Self::Expression {
Bound::new(self.0.as_str())
}
}
impl<Kind, ST> AsExpression<ST> for ObjectId<Kind>
where
Kind: Object,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
Bound<ST, String>: Expression<SqlType = ST>,
ST: SingleValue,
{
type Expression = Bound<ST, String>;
fn as_expression(self) -> Self::Expression {
Bound::new(self.0.to_string())
}
}
impl<Kind, ST, DB> FromSql<ST, DB> for ObjectId<Kind>
where
Kind: Object + Send + 'static,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
String: FromSql<ST, DB>,
DB: Backend,
DB: HasSqlType<ST>,
{
fn from_sql(
raw: DB::RawValue<'_>,
) -> Result<Self, Box<dyn ::std::error::Error + Send + Sync>> {
let string: String = FromSql::<ST, DB>::from_sql(raw)?;
Ok(ObjectId::parse(&string)?)
}
}
impl<Kind, ST, DB> Queryable<ST, DB> for ObjectId<Kind>
where
Kind: Object + Send + 'static,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
String: FromStaticSqlRow<ST, DB>,
DB: Backend,
DB: HasSqlType<ST>,
{
type Row = String;
fn build(row: Self::Row) -> diesel::deserialize::Result<Self> {
Ok(ObjectId::parse(&row)?)
}
}
impl<Kind> QueryId for ObjectId<Kind>
where
Kind: Object + 'static,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
{
type QueryId = Self;
}
};
#[cfg(test)] #[cfg(test)]
#[allow(clippy::unwrap_used)] #[allow(clippy::unwrap_used)]
pub mod tests { pub mod tests {

View file

@ -1,5 +1,5 @@
use crate::{ use crate::{
config::{domain_regex, Data}, config::Data,
error::Error, error::Error,
fetch::{fetch_object_http_with_accept, object_id::ObjectId}, fetch::{fetch_object_http_with_accept, object_id::ObjectId},
traits::{Actor, Object}, traits::{Actor, Object},
@ -7,9 +7,10 @@ use crate::{
}; };
use http::HeaderValue; use http::HeaderValue;
use itertools::Itertools; use itertools::Itertools;
use once_cell::sync::Lazy;
use regex::Regex; use regex::Regex;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::{collections::HashMap, fmt::Display, sync::LazyLock}; use std::{collections::HashMap, fmt::Display};
use tracing::debug; use tracing::debug;
use url::Url; use url::Url;
@ -45,7 +46,7 @@ pub async fn webfinger_resolve_actor<T: Clone, Kind>(
data: &Data<T>, data: &Data<T>,
) -> Result<Kind, <Kind as Object>::Error> ) -> Result<Kind, <Kind as Object>::Error>
where where
Kind: Object + Actor + Send + Sync + 'static + Object<DataType = T>, Kind: Object + Actor + Send + 'static + Object<DataType = T>,
for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>, for<'de2> <Kind as Object>::Kind: serde::Deserialize<'de2>,
<Kind as Object>::Error: From<crate::error::Error> + Send + Sync + Display, <Kind as Object>::Error: From<crate::error::Error> + Send + Sync + Display,
{ {
@ -53,31 +54,21 @@ where
.splitn(2, '@') .splitn(2, '@')
.collect_tuple() .collect_tuple()
.ok_or(WebFingerError::WrongFormat.into_crate_error())?; .ok_or(WebFingerError::WrongFormat.into_crate_error())?;
// For production mode make sure that domain doesnt contain any port or path.
if !data.config.debug && !domain_regex().is_match(domain) {
return Err(Error::UrlVerificationError("Invalid characters in domain").into());
}
let protocol = if data.config.debug { "http" } else { "https" }; let protocol = if data.config.debug { "http" } else { "https" };
let fetch_url = let fetch_url =
format!("{protocol}://{domain}/.well-known/webfinger?resource=acct:{identifier}"); format!("{protocol}://{domain}/.well-known/webfinger?resource=acct:{identifier}");
debug!("Fetching webfinger url: {}", &fetch_url); debug!("Fetching webfinger url: {}", &fetch_url);
let res = fetch_object_http_with_accept::<_, Webfinger>( let res: Webfinger = fetch_object_http_with_accept(
&Url::parse(&fetch_url).map_err(Error::UrlParse)?, &Url::parse(&fetch_url).map_err(Error::UrlParse)?,
data, data,
&WEBFINGER_CONTENT_TYPE, &WEBFINGER_CONTENT_TYPE,
false,
) )
.await?; .await?
if res.url.as_str() != fetch_url { .object;
data.config.verify_url_valid(&res.url).await?;
}
debug_assert_eq!(res.object.subject, format!("acct:{identifier}")); debug_assert_eq!(res.subject, format!("acct:{identifier}"));
let links: Vec<Url> = res let links: Vec<Url> = res
.object
.links .links
.iter() .iter()
.filter(|link| { .filter(|link| {
@ -88,7 +79,6 @@ where
} }
}) })
.filter_map(|l| l.href.clone()) .filter_map(|l| l.href.clone())
.rev()
.collect(); .collect();
for l in links { for l in links {
@ -130,8 +120,8 @@ pub fn extract_webfinger_name<'i, T>(query: &'i str, data: &Data<T>) -> Result<&
where where
T: Clone, T: Clone,
{ {
static WEBFINGER_REGEX: LazyLock<Regex> = static WEBFINGER_REGEX: Lazy<Regex> =
LazyLock::new(|| Regex::new(r"^acct:([\p{L}0-9_\.\-]+)@(.*)$").expect("compile regex")); Lazy::new(|| Regex::new(r"^acct:([\p{L}0-9_\.\-]+)@(.*)$").expect("compile regex"));
// Regex to extract usernames from webfinger query. Supports different alphabets using `\p{L}`. // Regex to extract usernames from webfinger query. Supports different alphabets using `\p{L}`.
// TODO: This should use a URL parser // TODO: This should use a URL parser
let captures = WEBFINGER_REGEX let captures = WEBFINGER_REGEX
@ -223,7 +213,7 @@ pub fn build_webfinger_response_with_type(
} }
/// A webfinger response with information about a `Person` or other type of actor. /// A webfinger response with information about a `Person` or other type of actor.
#[derive(Serialize, Deserialize, Debug, Default, PartialEq)] #[derive(Serialize, Deserialize, Debug, Default)]
pub struct Webfinger { pub struct Webfinger {
/// The actor which is described here, for example `acct:LemmyDev@mastodon.social` /// The actor which is described here, for example `acct:LemmyDev@mastodon.social`
pub subject: String, pub subject: String,
@ -238,7 +228,7 @@ pub struct Webfinger {
} }
/// A single link included as part of a [Webfinger] response. /// A single link included as part of a [Webfinger] response.
#[derive(Serialize, Deserialize, Debug, Default, PartialEq)] #[derive(Serialize, Deserialize, Debug, Default)]
pub struct WebfingerLink { pub struct WebfingerLink {
/// Relationship of the link, such as `self` or `http://webfinger.net/rel/profile-page` /// Relationship of the link, such as `self` or `http://webfinger.net/rel/profile-page`
pub rel: Option<String>, pub rel: Option<String>,

View file

@ -19,6 +19,7 @@ use http_signature_normalization_reqwest::{
prelude::{Config, SignExt}, prelude::{Config, SignExt},
DefaultSpawner, DefaultSpawner,
}; };
use once_cell::sync::Lazy;
use reqwest::Request; use reqwest::Request;
use reqwest_middleware::RequestBuilder; use reqwest_middleware::RequestBuilder;
use rsa::{ use rsa::{
@ -29,7 +30,7 @@ use rsa::{
}; };
use serde::Deserialize; use serde::Deserialize;
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use std::{collections::BTreeMap, fmt::Debug, sync::LazyLock, time::Duration}; use std::{collections::BTreeMap, fmt::Debug, time::Duration};
use tracing::debug; use tracing::debug;
use url::Url; use url::Url;
@ -53,10 +54,6 @@ impl Keypair {
} }
/// Generate a random asymmetric keypair for ActivityPub HTTP signatures. /// Generate a random asymmetric keypair for ActivityPub HTTP signatures.
///
/// Note that this method is very slow in debug mode. To make it faster, follow
/// instructions in the RSA crate's readme.
/// <https://github.com/RustCrypto/RSA/blob/master/README.md>
pub fn generate_actor_keypair() -> Result<Keypair, Error> { pub fn generate_actor_keypair() -> Result<Keypair, Error> {
let mut rng = rand::thread_rng(); let mut rng = rand::thread_rng();
let rsa = RsaPrivateKey::new(&mut rng, 2048)?; let rsa = RsaPrivateKey::new(&mut rng, 2048)?;
@ -85,9 +82,9 @@ pub(crate) async fn sign_request(
private_key: RsaPrivateKey, private_key: RsaPrivateKey,
http_signature_compat: bool, http_signature_compat: bool,
) -> Result<Request, Error> { ) -> Result<Request, Error> {
static CONFIG: LazyLock<Config<DefaultSpawner>> = static CONFIG: Lazy<Config<DefaultSpawner>> =
LazyLock::new(|| Config::new().set_expiration(EXPIRES_AFTER)); Lazy::new(|| Config::new().set_expiration(EXPIRES_AFTER));
static CONFIG_COMPAT: LazyLock<Config> = LazyLock::new(|| { static CONFIG_COMPAT: Lazy<Config> = Lazy::new(|| {
Config::new() Config::new()
.mastodon_compat() .mastodon_compat()
.set_expiration(EXPIRES_AFTER) .set_expiration(EXPIRES_AFTER)
@ -149,7 +146,7 @@ pub(crate) async fn signing_actor<'a, A, H>(
data: &Data<<A as Object>::DataType>, data: &Data<<A as Object>::DataType>,
) -> Result<A, <A as Object>::Error> ) -> Result<A, <A as Object>::Error>
where where
A: Object + Actor + Send + Sync, A: Object + Actor,
<A as Object>::Error: From<Error>, <A as Object>::Error: From<Error>,
for<'de2> <A as Object>::Kind: Deserialize<'de2>, for<'de2> <A as Object>::Kind: Deserialize<'de2>,
H: IntoIterator<Item = (&'a HeaderName, &'a HeaderValue)>, H: IntoIterator<Item = (&'a HeaderName, &'a HeaderValue)>,
@ -188,7 +185,7 @@ fn verify_signature_inner(
uri: &Uri, uri: &Uri,
public_key: &str, public_key: &str,
) -> Result<(), Error> { ) -> Result<(), Error> {
static CONFIG: LazyLock<http_signature_normalization::Config> = LazyLock::new(|| { static CONFIG: Lazy<http_signature_normalization::Config> = Lazy::new(|| {
http_signature_normalization::Config::new() http_signature_normalization::Config::new()
.set_expiration(EXPIRES_AFTER) .set_expiration(EXPIRES_AFTER)
.require_digest() .require_digest()
@ -280,7 +277,6 @@ pub(crate) fn verify_body_hash(
Ok(()) Ok(())
} }
/// Internal only
#[cfg(test)] #[cfg(test)]
#[allow(clippy::unwrap_used)] #[allow(clippy::unwrap_used)]
pub mod test { pub mod test {
@ -291,10 +287,9 @@ pub mod test {
use rsa::{pkcs1::DecodeRsaPrivateKey, pkcs8::DecodePrivateKey}; use rsa::{pkcs1::DecodeRsaPrivateKey, pkcs8::DecodePrivateKey};
use std::str::FromStr; use std::str::FromStr;
static ACTOR_ID: LazyLock<Url> = static ACTOR_ID: Lazy<Url> = Lazy::new(|| Url::parse("https://example.com/u/alice").unwrap());
LazyLock::new(|| Url::parse("https://example.com/u/alice").unwrap()); static INBOX_URL: Lazy<Url> =
static INBOX_URL: LazyLock<Url> = Lazy::new(|| Url::parse("https://example.com/u/alice/inbox").unwrap());
LazyLock::new(|| Url::parse("https://example.com/u/alice/inbox").unwrap());
#[tokio::test] #[tokio::test]
async fn test_sign() { async fn test_sign() {
@ -383,7 +378,6 @@ pub mod test {
assert_eq!(invalid, Err(Error::ActivityBodyDigestInvalid)); assert_eq!(invalid, Err(Error::ActivityBodyDigestInvalid));
} }
/// Internal only, return hardcoded keypair for testing
pub fn test_keypair() -> Keypair { pub fn test_keypair() -> Keypair {
let rsa = RsaPrivateKey::from_pkcs1_pem(PRIVATE_KEY).unwrap(); let rsa = RsaPrivateKey::from_pkcs1_pem(PRIVATE_KEY).unwrap();
let pkey = RsaPublicKey::from(&rsa); let pkey = RsaPublicKey::from(&rsa);

View file

@ -23,13 +23,12 @@ pub mod http_signatures;
pub mod protocol; pub mod protocol;
pub(crate) mod reqwest_shim; pub(crate) mod reqwest_shim;
pub mod traits; pub mod traits;
mod utils;
use crate::{ use crate::{
config::Data, config::Data,
error::Error, error::Error,
fetch::object_id::ObjectId, fetch::object_id::ObjectId,
traits::{Activity, Actor, Object}, traits::{ActivityHandler, Actor, Object},
}; };
pub use activitystreams_kinds as kinds; pub use activitystreams_kinds as kinds;
@ -41,22 +40,22 @@ pub const FEDERATION_CONTENT_TYPE: &str = "application/activity+json";
/// Deserialize incoming inbox activity to the given type, perform basic /// Deserialize incoming inbox activity to the given type, perform basic
/// validation and extract the actor. /// validation and extract the actor.
async fn parse_received_activity<A, ActorT, Datatype>( async fn parse_received_activity<Activity, ActorT, Datatype>(
body: &[u8], body: &[u8],
data: &Data<Datatype>, data: &Data<Datatype>,
) -> Result<(A, ActorT), <A as Activity>::Error> ) -> Result<(Activity, ActorT), <Activity as ActivityHandler>::Error>
where where
A: Activity<DataType = Datatype> + DeserializeOwned + Send + 'static, Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
ActorT: Object<DataType = Datatype> + Actor + Send + Sync + 'static, ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>, for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<A as Activity>::Error: From<Error> + From<<ActorT as Object>::Error>, <Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>, <ActorT as Object>::Error: From<Error>,
Datatype: Clone, Datatype: Clone,
{ {
let activity: A = serde_json::from_slice(body).map_err(|err| { let activity: Activity = serde_json::from_slice(body).map_err(|e| {
// Attempt to include activity id in error message // Attempt to include activity id in error message
let id = extract_id(body).ok(); let id = extract_id(body).ok();
Error::ParseReceivedActivity { err, id } Error::ParseReceivedActivity(e, id)
})?; })?;
data.config.verify_url_and_domain(&activity).await?; data.config.verify_url_and_domain(&activity).await?;
let actor = ObjectId::<ActorT>::from(activity.actor().clone()) let actor = ObjectId::<ActorT>::from(activity.actor().clone())

View file

@ -19,7 +19,7 @@
//! Ok::<(), serde_json::error::Error>(()) //! Ok::<(), serde_json::error::Error>(())
//! ``` //! ```
use crate::{config::Data, traits::Activity}; use crate::{config::Data, traits::ActivityHandler};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::Value; use serde_json::Value;
use url::Url; use url::Url;
@ -55,12 +55,12 @@ impl<T> WithContext<T> {
} }
#[async_trait::async_trait] #[async_trait::async_trait]
impl<T> Activity for WithContext<T> impl<T> ActivityHandler for WithContext<T>
where where
T: Activity + Send + Sync, T: ActivityHandler + Send + Sync,
{ {
type DataType = <T as Activity>::DataType; type DataType = <T as ActivityHandler>::DataType;
type Error = <T as Activity>::Error; type Error = <T as ActivityHandler>::Error;
fn id(&self) -> &Url { fn id(&self) -> &Url {
self.inner.id() self.inner.id()

View file

@ -1,22 +1,13 @@
//! Serde deserialization functions which help to receive differently shaped data //! Serde deserialization functions which help to receive differently shaped data
use activitystreams_kinds::public; use serde::{Deserialize, Deserializer};
use itertools::Itertools;
use serde::{de::Error, Deserialize, Deserializer};
use serde_json::Value;
use url::Url;
/// Deserialize JSON single value or array into `Vec<Url>`. /// Deserialize JSON single value or array into Vec.
/// ///
/// Useful if your application can handle multiple values for a field, but another federated /// Useful if your application can handle multiple values for a field, but another federated
/// platform only sends a single one. /// platform only sends a single one.
/// ///
/// Also accepts common `Public` aliases for recipient fields. Some implementations send `Public`
/// or `as:Public` instead of the canonical `https://www.w3.org/ns/activitystreams#Public` URL
/// in fields such as `to` and `cc`.
///
/// ``` /// ```
/// # use activitypub_federation::kinds::public;
/// # use activitypub_federation::protocol::helpers::deserialize_one_or_many; /// # use activitypub_federation::protocol::helpers::deserialize_one_or_many;
/// # use url::Url; /// # use url::Url;
/// #[derive(serde::Deserialize)] /// #[derive(serde::Deserialize)]
@ -34,39 +25,24 @@ use url::Url;
/// "https://lemmy.ml/u/bob" /// "https://lemmy.ml/u/bob"
/// ]}"#)?; /// ]}"#)?;
/// assert_eq!(multiple.to.len(), 2); /// assert_eq!(multiple.to.len(), 2);
/// /// Ok::<(), anyhow::Error>(())
/// let note: Note = serde_json::from_str(r#"{"to": ["Public", "as:Public"]}"#)?; pub fn deserialize_one_or_many<'de, T, D>(deserializer: D) -> Result<Vec<T>, D::Error>
/// assert_eq!(note.to, vec![public()]);
/// # Ok::<(), anyhow::Error>(())
/// ```
pub fn deserialize_one_or_many<'de, D>(deserializer: D) -> Result<Vec<Url>, D::Error>
where where
T: Deserialize<'de>,
D: Deserializer<'de>, D: Deserializer<'de>,
{ {
#[derive(Deserialize)] #[derive(Deserialize)]
#[serde(untagged)] #[serde(untagged)]
enum OneOrMany { enum OneOrMany<T> {
Many(Vec<Value>), One(T),
One(Value), Many(Vec<T>),
} }
let result: OneOrMany = Deserialize::deserialize(deserializer)?; let result: OneOrMany<T> = Deserialize::deserialize(deserializer)?;
let values = match result { Ok(match result {
OneOrMany::Many(list) => list,
OneOrMany::One(value) => vec![value], OneOrMany::One(value) => vec![value],
OneOrMany::Many(values) => values, })
};
values
.into_iter()
.map(|value| match value {
Value::String(value) if matches!(value.as_str(), "Public" | "as:Public") => {
Ok(public())
}
Value::String(value) => Url::parse(&value).map_err(D::Error::custom),
value => Url::deserialize(value).map_err(D::Error::custom),
})
.collect::<Result<Vec<_>, _>>()
.map(|values| values.into_iter().unique().collect())
} }
/// Deserialize JSON single value or single element array into single value. /// Deserialize JSON single value or single element array into single value.
@ -140,35 +116,8 @@ where
Ok(inner) Ok(inner)
} }
/// Deserialize either single value or last item from an array into an optional field.
pub fn deserialize_last<'de, T, D>(deserializer: D) -> Result<Option<T>, D::Error>
where
T: Deserialize<'de>,
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum MaybeArray<T> {
Simple(T),
Array(Vec<T>),
None,
}
let result = Deserialize::deserialize(deserializer)?;
Ok(match result {
MaybeArray::Simple(value) => Some(value),
MaybeArray::Array(value) => value.into_iter().last(),
MaybeArray::None => None,
})
}
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::deserialize_one_or_many;
use activitystreams_kinds::public;
use anyhow::Result;
use serde::Deserialize;
#[test] #[test]
fn deserialize_one_multiple_values() { fn deserialize_one_multiple_values() {
use crate::protocol::helpers::deserialize_one; use crate::protocol::helpers::deserialize_one;
@ -184,70 +133,4 @@ mod tests {
); );
assert!(note.is_err()); assert!(note.is_err());
} }
#[test]
fn deserialize_one_or_many_single_public_aliases() -> Result<()> {
use url::Url;
#[derive(Deserialize)]
struct Note {
#[serde(deserialize_with = "deserialize_one_or_many")]
to: Vec<Url>,
}
for alias in ["Public", "as:Public"] {
let note = serde_json::from_str::<Note>(&format!(r#"{{"to": "{alias}"}}"#))?;
assert_eq!(note.to, vec![public()]);
}
Ok(())
}
#[test]
fn deserialize_one_or_many_array() -> Result<()> {
use url::Url;
#[derive(Deserialize)]
struct Note {
#[serde(deserialize_with = "deserialize_one_or_many")]
to: Vec<Url>,
}
let note = serde_json::from_str::<Note>(
r#"{
"to": [
"https://example.com/c/main",
"Public",
"as:Public",
"https://www.w3.org/ns/activitystreams#Public"
]
}"#,
)?;
assert_eq!(
note.to,
vec![Url::parse("https://example.com/c/main")?, public(),]
);
Ok(())
}
#[test]
fn deserialize_one_or_many_leaves_other_strings_unchanged() -> Result<()> {
use url::Url;
#[derive(Deserialize)]
struct Note {
#[serde(deserialize_with = "deserialize_one_or_many")]
to: Vec<Url>,
content: String,
}
let note = serde_json::from_str::<Note>(r#"{"to": "Public", "content": "Public"}"#)?;
assert_eq!(note.to, vec![public()]);
assert_eq!(note.content, "Public");
Ok(())
}
} }

View file

@ -3,6 +3,5 @@
pub mod context; pub mod context;
pub mod helpers; pub mod helpers;
pub mod public_key; pub mod public_key;
pub mod tombstone;
pub mod values; pub mod values;
pub mod verification; pub mod verification;

View file

@ -1,27 +0,0 @@
//! Tombstone is used to serve deleted objects
use crate::kinds::object::TombstoneType;
use serde::{Deserialize, Serialize};
use url::Url;
/// Represents a local object that was deleted
///
/// <https://www.w3.org/TR/activitystreams-vocabulary/#dfn-tombstone>
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Tombstone {
/// Id of the deleted object
pub id: Url,
#[serde(rename = "type")]
pub(crate) kind: TombstoneType,
}
impl Tombstone {
/// Create a new tombstone for the given object id
pub fn new(id: Url) -> Tombstone {
Tombstone {
id,
kind: TombstoneType::Tombstone,
}
}
}

View file

@ -63,7 +63,7 @@ pub fn verify_is_remote_object<Kind, R: Clone>(
data: &Data<<Kind as Object>::DataType>, data: &Data<<Kind as Object>::DataType>,
) -> Result<(), Error> ) -> Result<(), Error>
where where
Kind: Object<DataType = R> + Send + Sync + 'static, Kind: Object<DataType = R> + Send + 'static,
for<'de2> <Kind as Object>::Kind: Deserialize<'de2>, for<'de2> <Kind as Object>::Kind: Deserialize<'de2>,
{ {
if id.is_local(data) { if id.is_local(data) {

View file

@ -10,8 +10,8 @@ use std::{
task::{Context, Poll}, task::{Context, Poll},
}; };
/// 1 MB /// 200KB
const MAX_BODY_SIZE: usize = 1024 * 1024; const MAX_BODY_SIZE: usize = 204800;
pin_project! { pin_project! {
pub struct BytesFuture { pub struct BytesFuture {
@ -66,7 +66,7 @@ impl Future for TextFuture {
/// Reqwest doesn't limit the response body size by default nor does it offer an option to configure one. /// Reqwest doesn't limit the response body size by default nor does it offer an option to configure one.
/// Since we have to fetch data from untrusted sources, not restricting the maximum size is a DoS hazard for us. /// Since we have to fetch data from untrusted sources, not restricting the maximum size is a DoS hazard for us.
/// ///
/// This shim reimplements the `bytes`, `json`, and `text` functions and restricts the bodies length. /// This shim reimplements the `bytes`, `json`, and `text` functions and restricts the bodies to 100KB.
/// ///
/// TODO: Remove this shim as soon as reqwest gets support for size-limited bodies. /// TODO: Remove this shim as soon as reqwest gets support for size-limited bodies.
pub trait ResponseExt { pub trait ResponseExt {

View file

@ -7,10 +7,6 @@ use serde::Deserialize;
use std::{fmt::Debug, ops::Deref}; use std::{fmt::Debug, ops::Deref};
use url::Url; use url::Url;
/// `Either` implementations for traits
pub mod either;
pub mod tests;
/// Helper for converting between database structs and federated protocol structs. /// Helper for converting between database structs and federated protocol structs.
/// ///
/// ``` /// ```
@ -53,8 +49,6 @@ pub mod tests;
/// type Kind = Note; /// type Kind = Note;
/// type Error = anyhow::Error; /// type Error = anyhow::Error;
/// ///
/// fn id(&self) -> Url { self.ap_id.inner().clone() }
///
/// async fn read_from_id(object_id: Url, data: &Data<Self::DataType>) -> Result<Option<Self>, Self::Error> { /// async fn read_from_id(object_id: Url, data: &Data<Self::DataType>) -> Result<Option<Self>, Self::Error> {
/// // Attempt to read object from local database. Return Ok(None) if not found. /// // Attempt to read object from local database. Return Ok(None) if not found.
/// let post: Option<DbPost> = data.read_post_from_json_id(object_id).await?; /// let post: Option<DbPost> = data.read_post_from_json_id(object_id).await?;
@ -109,9 +103,6 @@ pub trait Object: Sized + Debug {
/// Error type returned by handler methods /// Error type returned by handler methods
type Error; type Error;
/// `id` field of the object
fn id(&self) -> Url;
/// Returns the last time this object was updated. /// Returns the last time this object was updated.
/// ///
/// If this returns `Some` and the value is too long ago, the object is refetched from the /// If this returns `Some` and the value is too long ago, the object is refetched from the
@ -136,15 +127,10 @@ pub trait Object: Sized + Debug {
/// Mark remote object as deleted in local database. /// Mark remote object as deleted in local database.
/// ///
/// Called when a `Delete` activity is received, or if fetch returns a `Tombstone` object. /// Called when a `Delete` activity is received, or if fetch returns a `Tombstone` object.
async fn delete(&self, _data: &Data<Self::DataType>) -> Result<(), Self::Error> { async fn delete(self, _data: &Data<Self::DataType>) -> Result<(), Self::Error> {
Ok(()) Ok(())
} }
/// Returns true if the object was deleted
fn is_deleted(&self) -> bool {
false
}
/// Convert database type to Activitypub type. /// Convert database type to Activitypub type.
/// ///
/// Called when a local object gets fetched by another instance over HTTP, or when an object /// Called when a local object gets fetched by another instance over HTTP, or when an object
@ -170,40 +156,6 @@ pub trait Object: Sized + Debug {
/// should write the received object to database. Note that there is no distinction between /// should write the received object to database. Note that there is no distinction between
/// create and update, so an `upsert` operation should be used. /// create and update, so an `upsert` operation should be used.
async fn from_json(json: Self::Kind, data: &Data<Self::DataType>) -> Result<Self, Self::Error>; async fn from_json(json: Self::Kind, data: &Data<Self::DataType>) -> Result<Self, Self::Error>;
/// Generates HTTP response to serve the object for fetching from other instances.
///
/// - If the object has a remote domain, sends a redirect to the original instance.
/// - If [Object.is_deleted] returns true, returns a [crate::protocol::tombstone::Tombstone] instead.
/// - Otherwise serves the object JSON using [Object.into_json] and pretty-print
///
/// `federation_context` is the value of `@context`.
#[cfg(feature = "actix-web")]
async fn http_response(
self,
federation_context: &serde_json::Value,
data: &Data<Self::DataType>,
) -> Result<actix_web::HttpResponse, Self::Error>
where
Self::Error: From<serde_json::Error>,
Self::Kind: serde::Serialize + Send,
{
use crate::actix_web::response::{
create_http_response,
create_tombstone_response,
redirect_remote_object,
};
let id = self.id();
let res = if !data.config.is_local_url(&id) {
redirect_remote_object(&id)
} else if !self.is_deleted() {
let json = self.into_json(data).await?;
create_http_response(json, federation_context)?
} else {
create_tombstone_response(id.clone(), federation_context)?
};
Ok(res)
}
} }
/// Handler for receiving incoming activities. /// Handler for receiving incoming activities.
@ -213,7 +165,7 @@ pub trait Object: Sized + Debug {
/// # use url::Url; /// # use url::Url;
/// # use activitypub_federation::fetch::object_id::ObjectId; /// # use activitypub_federation::fetch::object_id::ObjectId;
/// # use activitypub_federation::config::Data; /// # use activitypub_federation::config::Data;
/// # use activitypub_federation::traits::Activity; /// # use activitypub_federation::traits::ActivityHandler;
/// # use activitypub_federation::traits::tests::{DbConnection, DbUser}; /// # use activitypub_federation::traits::tests::{DbConnection, DbUser};
/// #[derive(serde::Deserialize)] /// #[derive(serde::Deserialize)]
/// struct Follow { /// struct Follow {
@ -225,7 +177,7 @@ pub trait Object: Sized + Debug {
/// } /// }
/// ///
/// #[async_trait::async_trait] /// #[async_trait::async_trait]
/// impl Activity for Follow { /// impl ActivityHandler for Follow {
/// type DataType = DbConnection; /// type DataType = DbConnection;
/// type Error = anyhow::Error; /// type Error = anyhow::Error;
/// ///
@ -251,7 +203,7 @@ pub trait Object: Sized + Debug {
/// ``` /// ```
#[async_trait] #[async_trait]
#[enum_delegate::register] #[enum_delegate::register]
pub trait Activity { pub trait ActivityHandler {
/// App data type passed to handlers. Must be identical to /// App data type passed to handlers. Must be identical to
/// [crate::config::FederationConfigBuilder::app_data] type. /// [crate::config::FederationConfigBuilder::app_data] type.
type DataType: Clone + Send + Sync; type DataType: Clone + Send + Sync;
@ -279,6 +231,9 @@ pub trait Activity {
/// Trait to allow retrieving common Actor data. /// Trait to allow retrieving common Actor data.
pub trait Actor: Object + Send + 'static { pub trait Actor: Object + Send + 'static {
/// `id` field of the actor
fn id(&self) -> Url;
/// The actor's public key for verifying signatures of incoming activities. /// The actor's public key for verifying signatures of incoming activities.
/// ///
/// Use [generate_actor_keypair](crate::http_signatures::generate_actor_keypair) to create the /// Use [generate_actor_keypair](crate::http_signatures::generate_actor_keypair) to create the
@ -296,7 +251,7 @@ pub trait Actor: Object + Send + 'static {
/// Generates a public key struct for use in the actor json representation /// Generates a public key struct for use in the actor json representation
fn public_key(&self) -> PublicKey { fn public_key(&self) -> PublicKey {
PublicKey::new(self.id().clone(), self.public_key_pem().to_string()) PublicKey::new(self.id(), self.public_key_pem().to_string())
} }
/// The actor's shared inbox, if any /// The actor's shared inbox, if any
@ -312,9 +267,9 @@ pub trait Actor: Object + Send + 'static {
/// Allow for boxing of enum variants /// Allow for boxing of enum variants
#[async_trait] #[async_trait]
impl<T> Activity for Box<T> impl<T> ActivityHandler for Box<T>
where where
T: Activity + Send + Sync, T: ActivityHandler + Send + Sync,
{ {
type DataType = T::DataType; type DataType = T::DataType;
type Error = T::Error; type Error = T::Error;
@ -376,3 +331,207 @@ pub trait Collection: Sized {
data: &Data<Self::DataType>, data: &Data<Self::DataType>,
) -> Result<Self, Self::Error>; ) -> Result<Self, Self::Error>;
} }
/// Some impls of these traits for use in tests. Dont use this from external crates.
///
/// TODO: Should be using `cfg[doctest]` but blocked by <https://github.com/rust-lang/rust/issues/67295>
#[doc(hidden)]
#[allow(clippy::unwrap_used)]
pub mod tests {
use super::{async_trait, ActivityHandler, Actor, Data, Debug, Object, PublicKey, Url};
use crate::{
error::Error,
fetch::object_id::ObjectId,
http_signatures::{generate_actor_keypair, Keypair},
protocol::verification::verify_domains_match,
};
use activitystreams_kinds::{activity::FollowType, actor::PersonType};
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
#[derive(Clone)]
pub struct DbConnection;
impl DbConnection {
pub async fn read_post_from_json_id<T>(&self, _: Url) -> Result<Option<T>, Error> {
Ok(None)
}
pub async fn read_local_user(&self, _: &str) -> Result<DbUser, Error> {
todo!()
}
pub async fn upsert<T>(&self, _: &T) -> Result<(), Error> {
Ok(())
}
pub async fn add_follower(&self, _: DbUser, _: DbUser) -> Result<(), Error> {
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Person {
#[serde(rename = "type")]
pub kind: PersonType,
pub preferred_username: String,
pub id: ObjectId<DbUser>,
pub inbox: Url,
pub public_key: PublicKey,
}
#[derive(Debug, Clone)]
pub struct DbUser {
pub name: String,
pub federation_id: Url,
pub inbox: Url,
pub public_key: String,
#[allow(dead_code)]
private_key: Option<String>,
pub followers: Vec<Url>,
pub local: bool,
}
pub static DB_USER_KEYPAIR: Lazy<Keypair> = Lazy::new(|| generate_actor_keypair().unwrap());
pub static DB_USER: Lazy<DbUser> = Lazy::new(|| DbUser {
name: String::new(),
federation_id: "https://localhost/123".parse().unwrap(),
inbox: "https://localhost/123/inbox".parse().unwrap(),
public_key: DB_USER_KEYPAIR.public_key.clone(),
private_key: Some(DB_USER_KEYPAIR.private_key.clone()),
followers: vec![],
local: false,
});
#[async_trait]
impl Object for DbUser {
type DataType = DbConnection;
type Kind = Person;
type Error = Error;
async fn read_from_id(
_object_id: Url,
_data: &Data<Self::DataType>,
) -> Result<Option<Self>, Self::Error> {
Ok(Some(DB_USER.clone()))
}
async fn into_json(self, _data: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
Ok(Person {
preferred_username: self.name.clone(),
kind: Default::default(),
id: self.federation_id.clone().into(),
inbox: self.inbox.clone(),
public_key: self.public_key(),
})
}
async fn verify(
json: &Self::Kind,
expected_domain: &Url,
_data: &Data<Self::DataType>,
) -> Result<(), Self::Error> {
verify_domains_match(json.id.inner(), expected_domain)?;
Ok(())
}
async fn from_json(
json: Self::Kind,
_data: &Data<Self::DataType>,
) -> Result<Self, Self::Error> {
Ok(DbUser {
name: json.preferred_username,
federation_id: json.id.into(),
inbox: json.inbox,
public_key: json.public_key.public_key_pem,
private_key: None,
followers: vec![],
local: false,
})
}
}
impl Actor for DbUser {
fn id(&self) -> Url {
self.federation_id.clone()
}
fn public_key_pem(&self) -> &str {
&self.public_key
}
fn private_key_pem(&self) -> Option<String> {
self.private_key.clone()
}
fn inbox(&self) -> Url {
self.inbox.clone()
}
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct Follow {
pub actor: ObjectId<DbUser>,
pub object: ObjectId<DbUser>,
#[serde(rename = "type")]
pub kind: FollowType,
pub id: Url,
}
#[async_trait]
impl ActivityHandler for Follow {
type DataType = DbConnection;
type Error = Error;
fn id(&self) -> &Url {
&self.id
}
fn actor(&self) -> &Url {
self.actor.inner()
}
async fn verify(&self, _: &Data<Self::DataType>) -> Result<(), Self::Error> {
Ok(())
}
async fn receive(self, _data: &Data<Self::DataType>) -> Result<(), Self::Error> {
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Note {}
#[derive(Debug, Clone)]
pub struct DbPost {}
#[async_trait]
impl Object for DbPost {
type DataType = DbConnection;
type Kind = Note;
type Error = Error;
async fn read_from_id(
_: Url,
_: &Data<Self::DataType>,
) -> Result<Option<Self>, Self::Error> {
todo!()
}
async fn into_json(self, _: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
todo!()
}
async fn verify(
_: &Self::Kind,
_: &Url,
_: &Data<Self::DataType>,
) -> Result<(), Self::Error> {
todo!()
}
async fn from_json(_: Self::Kind, _: &Data<Self::DataType>) -> Result<Self, Self::Error> {
todo!()
}
}
}

View file

@ -1,133 +0,0 @@
use super::{Actor, Object};
use crate::{config::Data, error::Error};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use either::Either;
use serde::{Deserialize, Serialize};
use std::fmt::Debug;
use url::Url;
#[doc(hidden)]
#[derive(Serialize, Deserialize)]
#[serde(untagged)]
pub enum UntaggedEither<L, R> {
Left(L),
Right(R),
}
#[async_trait]
impl<T, R, E, D> Object for Either<T, R>
where
T: Object + Object<Error = E, DataType = D> + Send + Sync,
R: Object + Object<Error = E, DataType = D> + Send + Sync,
<T as Object>::Kind: Send + Sync,
<R as Object>::Kind: Send + Sync,
D: Sync + Send + Clone,
E: From<Error> + Debug,
{
type DataType = D;
type Kind = UntaggedEither<T::Kind, R::Kind>;
type Error = E;
/// `id` field of the object
fn id(&self) -> Url {
match self {
Either::Left(l) => l.id(),
Either::Right(r) => r.id(),
}
}
fn last_refreshed_at(&self) -> Option<DateTime<Utc>> {
match self {
Either::Left(l) => l.last_refreshed_at(),
Either::Right(r) => r.last_refreshed_at(),
}
}
async fn read_from_id(
object_id: Url,
data: &Data<Self::DataType>,
) -> Result<Option<Self>, Self::Error> {
let l = T::read_from_id(object_id.clone(), data).await?;
if let Some(l) = l {
return Ok(Some(Either::Left(l)));
}
let r = R::read_from_id(object_id.clone(), data).await?;
if let Some(r) = r {
return Ok(Some(Either::Right(r)));
}
Ok(None)
}
async fn delete(&self, data: &Data<Self::DataType>) -> Result<(), Self::Error> {
match self {
Either::Left(l) => l.delete(data).await,
Either::Right(r) => r.delete(data).await,
}
}
fn is_deleted(&self) -> bool {
match self {
Either::Left(l) => l.is_deleted(),
Either::Right(r) => r.is_deleted(),
}
}
async fn into_json(self, data: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
Ok(match self {
Either::Left(l) => UntaggedEither::Left(l.into_json(data).await?),
Either::Right(r) => UntaggedEither::Right(r.into_json(data).await?),
})
}
async fn verify(
json: &Self::Kind,
expected_domain: &Url,
data: &Data<Self::DataType>,
) -> Result<(), Self::Error> {
match json {
UntaggedEither::Left(l) => T::verify(l, expected_domain, data).await?,
UntaggedEither::Right(r) => R::verify(r, expected_domain, data).await?,
};
Ok(())
}
async fn from_json(json: Self::Kind, data: &Data<Self::DataType>) -> Result<Self, Self::Error> {
Ok(match json {
UntaggedEither::Left(l) => Either::Left(T::from_json(l, data).await?),
UntaggedEither::Right(r) => Either::Right(R::from_json(r, data).await?),
})
}
}
#[async_trait]
impl<T, R, E, D> Actor for Either<T, R>
where
T: Actor + Object + Object<Error = E, DataType = D> + Send + Sync + 'static,
R: Actor + Object + Object<Error = E, DataType = D> + Send + Sync + 'static,
<T as Object>::Kind: Send + Sync,
<R as Object>::Kind: Send + Sync,
D: Sync + Send + Clone,
E: From<Error> + Debug,
{
fn public_key_pem(&self) -> &str {
match self {
Either::Left(l) => l.public_key_pem(),
Either::Right(r) => r.public_key_pem(),
}
}
fn private_key_pem(&self) -> Option<String> {
match self {
Either::Left(l) => l.private_key_pem(),
Either::Right(r) => r.private_key_pem(),
}
}
fn inbox(&self) -> Url {
match self {
Either::Left(l) => l.inbox(),
Either::Right(r) => r.inbox(),
}
}
}

View file

@ -1,201 +0,0 @@
#![doc(hidden)]
#![allow(clippy::unwrap_used)]
//! Some impls of these traits for use in tests. Dont use this from external crates.
//!
//! TODO: Should be using `cfg[doctest]` but blocked by <https://github.com/rust-lang/rust/issues/67295>
use super::{async_trait, Activity, Actor, Data, Debug, Object, PublicKey, Url};
use crate::{
error::Error,
fetch::object_id::ObjectId,
http_signatures::{generate_actor_keypair, Keypair},
protocol::verification::verify_domains_match,
};
use activitystreams_kinds::{activity::FollowType, actor::PersonType};
use serde::{Deserialize, Serialize};
use std::sync::LazyLock;
#[derive(Clone)]
pub struct DbConnection;
impl DbConnection {
pub async fn read_post_from_json_id<T>(&self, _: Url) -> Result<Option<T>, Error> {
Ok(None)
}
pub async fn read_local_user(&self, _: &str) -> Result<DbUser, Error> {
todo!()
}
pub async fn upsert<T>(&self, _: &T) -> Result<(), Error> {
Ok(())
}
pub async fn add_follower(&self, _: DbUser, _: DbUser) -> Result<(), Error> {
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Person {
#[serde(rename = "type")]
pub kind: PersonType,
pub preferred_username: String,
pub id: ObjectId<DbUser>,
pub inbox: Url,
pub public_key: PublicKey,
}
#[derive(Debug, Clone)]
pub struct DbUser {
pub name: String,
pub federation_id: Url,
pub inbox: Url,
pub public_key: String,
#[allow(dead_code)]
private_key: Option<String>,
pub followers: Vec<Url>,
pub local: bool,
}
pub static DB_USER_KEYPAIR: LazyLock<Keypair> = LazyLock::new(|| generate_actor_keypair().unwrap());
pub static DB_USER: LazyLock<DbUser> = LazyLock::new(|| DbUser {
name: String::new(),
federation_id: "https://localhost/123".parse().unwrap(),
inbox: "https://localhost/123/inbox".parse().unwrap(),
public_key: DB_USER_KEYPAIR.public_key.clone(),
private_key: Some(DB_USER_KEYPAIR.private_key.clone()),
followers: vec![],
local: false,
});
#[async_trait]
impl Object for DbUser {
type DataType = DbConnection;
type Kind = Person;
type Error = Error;
fn id(&self) -> Url {
self.federation_id.clone()
}
async fn read_from_id(
_object_id: Url,
_data: &Data<Self::DataType>,
) -> Result<Option<Self>, Self::Error> {
Ok(Some(DB_USER.clone()))
}
async fn into_json(self, _data: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
Ok(Person {
preferred_username: self.name.clone(),
kind: Default::default(),
id: self.federation_id.clone().into(),
inbox: self.inbox.clone(),
public_key: self.public_key(),
})
}
async fn verify(
json: &Self::Kind,
expected_domain: &Url,
_data: &Data<Self::DataType>,
) -> Result<(), Self::Error> {
verify_domains_match(json.id.inner(), expected_domain)?;
Ok(())
}
async fn from_json(
json: Self::Kind,
_data: &Data<Self::DataType>,
) -> Result<Self, Self::Error> {
Ok(DbUser {
name: json.preferred_username,
federation_id: json.id.into(),
inbox: json.inbox,
public_key: json.public_key.public_key_pem,
private_key: None,
followers: vec![],
local: false,
})
}
}
impl Actor for DbUser {
fn public_key_pem(&self) -> &str {
&self.public_key
}
fn private_key_pem(&self) -> Option<String> {
self.private_key.clone()
}
fn inbox(&self) -> Url {
self.inbox.clone()
}
}
#[derive(Deserialize, Serialize, Clone, Debug)]
#[serde(rename_all = "camelCase")]
pub struct Follow {
pub actor: ObjectId<DbUser>,
pub object: ObjectId<DbUser>,
#[serde(rename = "type")]
pub kind: FollowType,
pub id: Url,
}
#[async_trait]
impl Activity for Follow {
type DataType = DbConnection;
type Error = Error;
fn id(&self) -> &Url {
&self.id
}
fn actor(&self) -> &Url {
self.actor.inner()
}
async fn verify(&self, _: &Data<Self::DataType>) -> Result<(), Self::Error> {
Ok(())
}
async fn receive(self, _data: &Data<Self::DataType>) -> Result<(), Self::Error> {
Ok(())
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Note {}
#[derive(Debug, Clone)]
pub struct DbPost {
pub federation_id: Url,
}
#[async_trait]
impl Object for DbPost {
type DataType = DbConnection;
type Kind = Note;
type Error = Error;
fn id(&self) -> Url {
todo!()
}
async fn read_from_id(_: Url, _: &Data<Self::DataType>) -> Result<Option<Self>, Self::Error> {
todo!()
}
async fn into_json(self, _: &Data<Self::DataType>) -> Result<Self::Kind, Self::Error> {
todo!()
}
async fn verify(_: &Self::Kind, _: &Url, _: &Data<Self::DataType>) -> Result<(), Self::Error> {
todo!()
}
async fn from_json(_: Self::Kind, _: &Data<Self::DataType>) -> Result<Self, Self::Error> {
todo!()
}
}

View file

@ -1,78 +0,0 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use crate::error::Error;
use tokio::net::lookup_host;
use url::{Host, Url};
// TODO: Use is_global() once stabilized
// https://doc.rust-lang.org/std/net/enum.IpAddr.html#method.is_global
pub(crate) async fn validate_ip(url: &Url) -> Result<(), Error> {
let mut ip = vec![];
let host = url
.host()
.ok_or(Error::UrlVerificationError("Url must have a domain"))?;
match host {
Host::Domain(domain) => ip.extend(
lookup_host((domain.to_owned(), 80))
.await?
.map(|s| s.ip().to_canonical()),
),
Host::Ipv4(ipv4) => ip.push(ipv4.into()),
Host::Ipv6(ipv6) => ip.push(ipv6.into()),
};
let invalid_ip = ip.into_iter().any(|addr| match addr {
IpAddr::V4(addr) => v4_is_invalid(addr),
IpAddr::V6(addr) => v6_is_invalid(addr),
});
if invalid_ip {
return Err(Error::DomainResolveError(host.to_string()));
}
Ok(())
}
fn v4_is_invalid(v4: Ipv4Addr) -> bool {
v4.is_private()
|| v4.is_loopback()
|| v4.is_link_local()
|| v4.is_multicast()
|| v4.is_documentation()
|| v4.is_unspecified()
|| v4.is_broadcast()
}
fn v6_is_invalid(v6: Ipv6Addr) -> bool {
v6.is_loopback()
|| v6.is_multicast()
|| v6.is_unique_local()
|| v6.is_unicast_link_local()
|| v6.is_unspecified()
|| v6_is_documentation(v6)
|| v6.to_ipv4_mapped().is_some_and(v4_is_invalid)
}
fn v6_is_documentation(v6: std::net::Ipv6Addr) -> bool {
matches!(
v6.segments(),
[0x2001, 0xdb8, ..] | [0x3fff, 0..=0x0fff, ..]
)
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod test {
use super::*;
#[tokio::test]
async fn test_is_valid_ip() -> Result<(), Error> {
assert!(validate_ip(&Url::parse("http://example.com")?)
.await
.is_ok());
assert!(validate_ip(&Url::parse("http://172.66.147.243")?)
.await
.is_ok());
assert!(validate_ip(&Url::parse("http://localhost")?).await.is_err());
assert!(validate_ip(&Url::parse("http://127.0.0.1")?).await.is_err());
Ok(())
}
}