Improve api for send_activity()

This commit is contained in:
Felix Ableitner 2022-06-03 14:00:16 +02:00
parent d7e401eeed
commit f9b822860e
7 changed files with 127 additions and 86 deletions

16
Cargo.lock generated
View file

@ -20,6 +20,7 @@ dependencies = [
"http",
"http-signature-normalization-actix",
"http-signature-normalization-reqwest",
"itertools",
"once_cell",
"openssl",
"rand",
@ -520,6 +521,12 @@ version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8c97b9233581d84b8e1e689cdd3a47b6f69770084fc246e86a7f78b0d9c1d4a5"
[[package]]
name = "either"
version = "1.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457"
[[package]]
name = "encoding_rs"
version = "0.8.31"
@ -899,6 +906,15 @@ version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "879d54834c8c76457ef4293a689b2a8c59b076067ad77b15efafbb05f92a592b"
[[package]]
name = "itertools"
version = "0.10.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a9a9d19fa1e79b6215ff29b9d6880b706147f16e9b1dbb1e4e5947b5b02bc5e3"
dependencies = [
"either",
]
[[package]]
name = "itoa"
version = "1.0.2"

View file

@ -31,6 +31,7 @@ http-signature-normalization-reqwest = { version = "0.5.0", default-features = f
background-jobs = "0.12.0"
thiserror = "1.0.31"
derive_builder = "0.11.2"
itertools = "0.10.3"
[dev-dependencies]
activitystreams-kinds = "0.2.1"

View file

@ -4,7 +4,11 @@ use crate::{
instance::InstanceHandle,
objects::person::MyUser,
};
use activitypub_federation::{core::object_id::ObjectId, data::Data, traits::ActivityHandler};
use activitypub_federation::{
core::object_id::ObjectId,
data::Data,
traits::{ActivityHandler, Actor},
};
use activitystreams_kinds::activity::FollowType;
use serde::{Deserialize, Serialize};
use url::Url;
@ -74,9 +78,8 @@ impl ActivityHandler for Follow {
let accept = Accept::new(local_user.ap_id.clone(), self, id.clone());
local_user
.send(
id,
accept,
vec![follower.inbox.clone()],
vec![follower.shared_inbox_or_inbox()],
data.local_instance(),
)
.await?;

View file

@ -7,19 +7,17 @@ use crate::{
};
use activitypub_federation::{
core::{
activity_queue::SendActivity,
inbox::ActorPublicKey,
activity_queue::send_activity,
object_id::ObjectId,
signatures::{Keypair, PublicKey},
},
deser::context::WithContext,
traits::ApubObject,
traits::{ActivityHandler, Actor, ApubObject},
LocalInstance,
};
use activitypub_federation_derive::activity_handler;
use activitystreams_kinds::actor::PersonType;
use serde::{Deserialize, Serialize};
use tracing::log::debug;
use url::Url;
#[derive(Debug, Clone)]
@ -87,9 +85,8 @@ impl MyUser {
let id = generate_object_id(instance.local_instance().hostname())?;
let follow = Follow::new(self.ap_id.clone(), other.ap_id.clone(), id.clone());
self.send(
id,
follow,
vec![other.inbox.clone()],
vec![other.shared_inbox_or_inbox()],
instance.local_instance(),
)
.await?;
@ -104,30 +101,31 @@ impl MyUser {
let user: MyUser = ObjectId::new(f)
.dereference(instance, instance.local_instance(), &mut 0)
.await?;
inboxes.push(user.inbox);
inboxes.push(user.shared_inbox_or_inbox());
}
self.send(id, &create, inboxes, instance.local_instance())
self.send(create, inboxes, instance.local_instance())
.await?;
Ok(())
}
pub(crate) async fn send<Activity: Serialize>(
pub(crate) async fn send<Activity>(
&self,
activity_id: Url,
activity: Activity,
inboxes: Vec<Url>,
recipients: Vec<Url>,
local_instance: &LocalInstance,
) -> Result<(), Error> {
let serialized = serde_json::to_string_pretty(&WithContext::new_default(activity))?;
debug!("Sending activity: {}", &serialized);
SendActivity {
activity_id,
actor_public_key: self.public_key(),
actor_private_key: self.private_key.clone().expect("has private key"),
inboxes,
activity: serialized,
}
.send(local_instance)
) -> Result<(), <Activity as ActivityHandler>::Error>
where
Activity: ActivityHandler + Serialize,
<Activity as ActivityHandler>::Error: From<anyhow::Error> + From<serde_json::Error>,
{
let activity = WithContext::new_default(activity);
send_activity(
activity,
self.public_key(),
self.private_key.clone().expect("has private key"),
recipients,
local_instance,
)
.await?;
Ok(())
}
@ -186,8 +184,12 @@ impl ApubObject for MyUser {
}
}
impl ActorPublicKey for MyUser {
impl Actor for MyUser {
fn public_key(&self) -> &str {
&self.public_key
}
fn inbox(&self) -> Url {
self.inbox.clone()
}
}

View file

@ -1,5 +1,6 @@
use crate::{
core::signatures::{sign_request, PublicKey},
traits::ActivityHandler,
utils::verify_url_valid,
Error,
InstanceSettings,
@ -16,70 +17,76 @@ use background_jobs::{
WorkerConfig,
};
use http::{header::HeaderName, HeaderMap, HeaderValue};
use itertools::Itertools;
use reqwest_middleware::ClientWithMiddleware;
use serde::{Deserialize, Serialize};
use std::{fmt::Debug, future::Future, pin::Pin, time::Duration};
use tracing::{info, warn};
use url::Url;
/// Necessary data for sending out an activity
#[derive(Debug)]
pub struct SendActivity {
/// Id of the sent activity, used for logging
pub activity_id: Url,
/// Public key and actor id of the sender
pub actor_public_key: PublicKey,
/// Signing key of sender for HTTP signatures
pub actor_private_key: String,
/// List of Activitypub inboxes that the activity gets delivered to
pub inboxes: Vec<Url>,
/// Activity json
pub activity: String,
}
/// Send out the given activity to all inboxes, automatically generating the HTTP signatures. By
/// default, sending is done on a background thread, and automatically retried on failure with
/// exponential backoff.
///
/// - `activity`: The activity to be sent, gets converted to json
/// - `public_key`: The sending actor's public key. In fact, we only need the key id for signing
/// - `private_key`: The sending actor's private key for signing HTTP signature
/// - `recipients`: List of actors who should receive the activity. This gets deduplicated, and
/// local/invalid inbox urls removed
pub async fn send_activity<Activity>(
activity: Activity,
public_key: PublicKey,
private_key: String,
recipients: Vec<Url>,
instance: &LocalInstance,
) -> Result<(), <Activity as ActivityHandler>::Error>
where
Activity: ActivityHandler + Serialize,
<Activity as ActivityHandler>::Error: From<anyhow::Error> + From<serde_json::Error>,
{
let activity_id = activity.id();
let activity_serialized = serde_json::to_string_pretty(&activity)?;
let inboxes: Vec<Url> = recipients
.into_iter()
.unique()
.filter(|i| !instance.is_local_url(i))
.filter(|i| verify_url_valid(i, &instance.settings).is_ok())
.collect();
impl SendActivity {
/// Send out the given activity to all inboxes, automatically generating the HTTP signatures. By
/// default, sending is done on a background thread, and automatically retried on failure with
/// exponential backoff.
///
/// For debugging or testing, you might want to set [[InstanceSettings.testing_send_sync]].
pub async fn send(self, instance: &LocalInstance) -> Result<(), Error> {
let activity_queue = &instance.activity_queue;
for inbox in self.inboxes {
if verify_url_valid(&inbox, &instance.settings).is_err() {
continue;
let activity_queue = &instance.activity_queue;
for inbox in inboxes {
if verify_url_valid(&inbox, &instance.settings).is_err() {
continue;
}
let message = SendActivityTask {
activity_id: activity_id.clone(),
inbox,
activity: activity_serialized.clone(),
public_key: public_key.clone(),
private_key: private_key.clone(),
};
if instance.settings.debug {
let res = do_send(message, &instance.client, instance.settings.request_timeout).await;
// Don't fail on error, as we intentionally do some invalid actions in tests, to verify that
// they are rejected on the receiving side. These errors shouldn't bubble up to make the API
// call fail. This matches the behaviour in production.
if let Err(e) = res {
warn!("{}", e);
}
let message = SendActivityTask {
activity_id: self.activity_id.clone(),
inbox,
activity: self.activity.clone(),
public_key: self.actor_public_key.clone(),
private_key: self.actor_private_key.clone(),
};
if instance.settings.debug {
let res =
do_send(message, &instance.client, instance.settings.request_timeout).await;
// Don't fail on error, as we intentionally do some invalid actions in tests, to verify that
// they are rejected on the receiving side. These errors shouldn't bubble up to make the API
// call fail. This matches the behaviour in production.
if let Err(e) = res {
warn!("{}", e);
}
} else {
activity_queue.queue::<SendActivityTask>(message).await?;
let stats = activity_queue.get_stats().await?;
info!(
} else {
activity_queue.queue::<SendActivityTask>(message).await?;
let stats = activity_queue.get_stats().await?;
info!(
"Activity queue stats: pending: {}, running: {}, dead (this hour): {}, complete (this hour): {}",
stats.pending,
stats.running,
stats.dead.this_hour(),
stats.complete.this_hour()
);
}
}
Ok(())
}
Ok(())
}
#[derive(Clone, Debug, Deserialize, Serialize)]

View file

@ -1,7 +1,7 @@
use crate::{
core::{object_id::ObjectId, signatures::verify_signature},
data::Data,
traits::{ActivityHandler, ApubObject},
traits::{ActivityHandler, Actor, ApubObject},
utils::{verify_domains_match, verify_url_valid},
Error,
LocalInstance,
@ -10,13 +10,8 @@ use actix_web::{HttpRequest, HttpResponse};
use serde::de::DeserializeOwned;
use tracing::log::debug;
pub trait ActorPublicKey {
/// Returns the actor's public key for verification of HTTP signatures
fn public_key(&self) -> &str;
}
/// Receive an activity and perform some basic checks, including HTTP signature verification.
pub async fn receive_activity<Activity, Actor, Datatype>(
pub async fn receive_activity<Activity, ActorT, Datatype>(
request: HttpRequest,
activity: Activity,
local_instance: &LocalInstance,
@ -24,11 +19,11 @@ pub async fn receive_activity<Activity, Actor, Datatype>(
) -> Result<HttpResponse, <Activity as ActivityHandler>::Error>
where
Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
Actor: ApubObject<DataType = Datatype> + ActorPublicKey + Send + 'static,
for<'de2> <Actor as ApubObject>::ApubType: serde::Deserialize<'de2>,
ActorT: ApubObject<DataType = Datatype> + Actor + Send + 'static,
for<'de2> <ActorT as ApubObject>::ApubType: serde::Deserialize<'de2>,
<Activity as ActivityHandler>::Error:
From<anyhow::Error> + From<Error> + From<<Actor as ApubObject>::Error>,
<Actor as ApubObject>::Error: From<Error> + From<anyhow::Error>,
From<anyhow::Error> + From<Error> + From<<ActorT as ApubObject>::Error>,
<ActorT as ApubObject>::Error: From<Error> + From<anyhow::Error>,
{
verify_domains_match(activity.id(), activity.actor())?;
verify_url_valid(activity.id(), &local_instance.settings)?;
@ -37,7 +32,7 @@ where
}
let request_counter = &mut 0;
let actor = ObjectId::<Actor>::new(activity.actor().clone())
let actor = ObjectId::<ActorT>::new(activity.actor().clone())
.dereference(data, local_instance, request_counter)
.await?;
verify_signature(&request, actor.public_key())?;

View file

@ -89,3 +89,20 @@ pub trait ApubObject {
where
Self: Sized;
}
pub trait Actor: ApubObject {
/// Returns the actor's public key for verification of HTTP signatures
fn public_key(&self) -> &str;
/// The inbox where activities for this user should be sent to
fn inbox(&self) -> Url;
/// The actor's shared inbox, if any
fn shared_inbox(&self) -> Option<Url> {
None
}
fn shared_inbox_or_inbox(&self) -> Url {
self.shared_inbox().unwrap_or_else(|| self.inbox())
}
}