use crate::{
+ activity_lists::SharedInboxActivities,
check_is_apub_id_valid,
- extensions::signatures::verify_signature,
- fetcher::get_or_fetch_and_upsert_actor,
- http::inbox_enums::SharedInboxActivities,
+ context::WithContext,
+ fetcher::user_or_community::UserOrCommunity,
+ http::{community::receive_group_inbox, person::receive_person_inbox},
insert_activity,
- APUB_JSON_CONTENT_TYPE,
};
use actix_web::{
body::Body,
use futures::StreamExt;
use http::StatusCode;
use lemmy_api_common::blocking;
-use lemmy_apub_lib::ActivityHandler;
-use lemmy_db_queries::{source::activity::Activity_, DbPool};
+use lemmy_apub_lib::{
+ data::Data,
+ object_id::ObjectId,
+ signatures::verify_signature,
+ traits::{ActivityHandler, ActorType},
+ APUB_JSON_CONTENT_TYPE,
+};
use lemmy_db_schema::source::activity::Activity;
-use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
+use lemmy_utils::{location_info, LemmyError};
use lemmy_websocket::LemmyContext;
+use log::info;
use serde::{Deserialize, Serialize};
use std::{fmt::Debug, io::Read};
use url::Url;
mod comment;
mod community;
-mod inbox_enums;
mod person;
mod post;
pub mod routes;
context: web::Data<LemmyContext>,
) -> Result<HttpResponse, LemmyError> {
let unparsed = payload_to_string(payload).await?;
- receive_activity::<SharedInboxActivities>(request, &unparsed, context).await
+ info!("Received shared inbox activity {}", unparsed);
+ let activity_data: ActivityCommonFields = serde_json::from_str(&unparsed)?;
+ let activity = serde_json::from_str::<WithContext<SharedInboxActivities>>(&unparsed)?;
+ match activity.inner() {
+ SharedInboxActivities::GroupInboxActivities(g) => {
+ receive_group_inbox(*g, activity_data, request, &context).await
+ }
+ SharedInboxActivities::PersonInboxActivities(p) => {
+ receive_person_inbox(*p, activity_data, request, &context).await
+ }
+ }
}
async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
Ok(unparsed)
}
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct ActivityCommonFields {
+ pub(crate) id: Url,
+ pub(crate) actor: Url,
+}
+
// TODO: move most of this code to library
async fn receive_activity<'a, T>(
request: HttpRequest,
- activity: &'a str,
- context: web::Data<LemmyContext>,
+ activity: T,
+ activity_data: ActivityCommonFields,
+ context: &LemmyContext,
) -> Result<HttpResponse, LemmyError>
where
- T: ActivityHandler + Clone + Deserialize<'a> + Serialize + std::fmt::Debug + Send + 'static,
+ T: ActivityHandler<DataType = LemmyContext>
+ + Clone
+ + Deserialize<'a>
+ + Serialize
+ + std::fmt::Debug
+ + Send
+ + 'static,
{
- let activity = serde_json::from_str::<T>(activity)?;
- let activity_data = activity.common();
-
+ check_is_apub_id_valid(&activity_data.actor, false, &context.settings())?;
let request_counter = &mut 0;
- let actor =
- get_or_fetch_and_upsert_actor(&activity_data.actor, &context, request_counter).await?;
- verify_signature(&request, &actor.public_key().context(location_info!())?)?;
+ let actor = ObjectId::<UserOrCommunity>::new(activity_data.actor)
+ .dereference(context, request_counter)
+ .await?;
+ verify_signature(&request, &actor.public_key())?;
- // Do nothing if we received the same activity before
- if is_activity_already_known(context.pool(), activity_data.id_unchecked()).await? {
- return Ok(HttpResponse::Ok().finish());
- }
- check_is_apub_id_valid(&activity_data.actor, false)?;
- println!(
- "Verifying activity {}",
- activity_data.id_unchecked().to_string()
- );
- activity.verify(&context, request_counter).await?;
- assert_activity_not_local(&activity)?;
+ info!("Verifying activity {}", activity_data.id.to_string());
+ activity
+ .verify(&Data::new(context.clone()), request_counter)
+ .await?;
+ assert_activity_not_local(&activity_data.id, &context.settings().hostname)?;
// Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
// if we receive the same activity twice in very quick succession.
- insert_activity(
- activity_data.id_unchecked(),
- activity.clone(),
- false,
- true,
- context.pool(),
- )
- .await?;
-
- println!(
- "Receiving activity {}",
- activity_data.id_unchecked().to_string()
- );
- activity.receive(&context, request_counter).await?;
+ let object_value = serde_json::to_value(&activity)?;
+ insert_activity(&activity_data.id, object_value, false, true, context.pool()).await?;
+
+ info!("Receiving activity {}", activity_data.id.to_string());
+ activity
+ .receive(&Data::new(context.clone()), request_counter)
+ .await?;
Ok(HttpResponse::Ok().finish())
}
where
T: Serialize,
{
+ HttpResponse::Ok()
+ .content_type(APUB_JSON_CONTENT_TYPE)
+ .json(WithContext::new(data))
+}
+
+fn create_json_apub_response(data: serde_json::Value) -> HttpResponse<Body> {
HttpResponse::Ok()
.content_type(APUB_JSON_CONTENT_TYPE)
.json(data)
HttpResponse::Gone()
.content_type(APUB_JSON_CONTENT_TYPE)
.status(StatusCode::GONE)
- .json(data)
+ .json(WithContext::new(data))
}
#[derive(Deserialize)]
info: web::Path<ActivityQuery>,
context: web::Data<LemmyContext>,
) -> Result<HttpResponse<Body>, LemmyError> {
- let settings = Settings::get();
+ let settings = context.settings();
let activity_id = Url::parse(&format!(
"{}/activities/{}/{}",
settings.get_protocol_and_hostname(),
if !activity.local || sensitive {
Ok(HttpResponse::NotFound().finish())
} else {
- Ok(create_apub_response(&activity.data))
- }
-}
-
-pub(crate) async fn is_activity_already_known(
- pool: &DbPool,
- activity_id: &Url,
-) -> Result<bool, LemmyError> {
- let activity_id = activity_id.to_owned().into();
- let existing = blocking(pool, move |conn| {
- Activity::read_from_apub_id(conn, &activity_id)
- })
- .await?;
- match existing {
- Ok(_) => Ok(true),
- Err(_) => Ok(false),
+ Ok(create_json_apub_response(activity.data))
}
}
-fn assert_activity_not_local<T: Debug + ActivityHandler>(activity: &T) -> Result<(), LemmyError> {
- let activity_domain = activity
- .common()
- .id_unchecked()
- .domain()
- .context(location_info!())?;
+fn assert_activity_not_local(id: &Url, hostname: &str) -> Result<(), LemmyError> {
+ let activity_domain = id.domain().context(location_info!())?;
- if activity_domain == Settings::get().hostname() {
+ if activity_domain == hostname {
return Err(
anyhow!(
"Error: received activity which was sent by local instance: {:?}",
- activity
+ id
)
.into(),
);
projects / lemmy.git / blobdiff