use crate::{
- check_is_apub_id_valid,
- context::WithContext,
- fetcher::get_or_fetch_and_upsert_actor,
- http::{
- community::{receive_group_inbox, GroupInboxActivities},
- person::{receive_person_inbox, PersonInboxActivities},
- },
- insert_activity,
+ activity_lists::SharedInboxActivities,
+ fetcher::user_or_community::UserOrCommunity,
+ protocol::objects::tombstone::Tombstone,
+ CONTEXT,
};
-use actix_web::{
- body::Body,
- web,
- web::{Bytes, BytesMut, Payload},
- HttpRequest,
- HttpResponse,
+use activitypub_federation::{
+ actix_web::inbox::receive_activity,
+ config::Data,
+ protocol::context::WithContext,
+ FEDERATION_CONTENT_TYPE,
};
-use anyhow::{anyhow, Context};
-use futures::StreamExt;
+use actix_web::{web, web::Bytes, HttpRequest, HttpResponse};
use http::StatusCode;
-use lemmy_api_common::blocking;
-use lemmy_apub_lib::{
- data::Data,
- signatures::verify_signature,
- traits::{ActivityFields, ActivityHandler},
- APUB_JSON_CONTENT_TYPE,
-};
-use lemmy_db_schema::{source::activity::Activity, DbPool};
-use lemmy_utils::{location_info, LemmyError};
-use lemmy_websocket::LemmyContext;
-use log::info;
+use lemmy_api_common::context::LemmyContext;
+use lemmy_db_schema::source::activity::SentActivity;
+use lemmy_utils::error::{LemmyError, LemmyErrorType, LemmyResult};
use serde::{Deserialize, Serialize};
-use std::{fmt::Debug, io::Read};
+use std::ops::Deref;
use url::Url;
mod comment;
mod person;
mod post;
pub mod routes;
-
-#[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
-#[serde(untagged)]
-#[activity_handler(LemmyContext)]
-pub enum SharedInboxActivities {
- GroupInboxActivities(GroupInboxActivities),
- // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
- // avoid this problem by replacing createpm.object with our own struct, instead of NoteExt.
- PersonInboxActivities(PersonInboxActivities),
-}
+pub mod site;
pub async fn shared_inbox(
request: HttpRequest,
- payload: Payload,
- context: web::Data<LemmyContext>,
-) -> Result<HttpResponse, LemmyError> {
- let unparsed = payload_to_string(payload).await?;
- info!("Received shared inbox activity {}", unparsed);
- let activity = serde_json::from_str::<WithContext<SharedInboxActivities>>(&unparsed)?;
- match activity.inner() {
- SharedInboxActivities::GroupInboxActivities(g) => {
- receive_group_inbox(g, request, &context).await
- }
- SharedInboxActivities::PersonInboxActivities(p) => {
- receive_person_inbox(p, request, &context).await
- }
- }
-}
-
-async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
- let mut bytes = BytesMut::new();
- while let Some(item) = payload.next().await {
- bytes.extend_from_slice(&item?);
- }
- let mut unparsed = String::new();
- Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
- Ok(unparsed)
+ body: Bytes,
+ data: Data<LemmyContext>,
+) -> LemmyResult<HttpResponse> {
+ receive_activity::<SharedInboxActivities, UserOrCommunity, LemmyContext>(request, body, &data)
+ .await
}
-// TODO: move most of this code to library
-async fn receive_activity<'a, T>(
- request: HttpRequest,
- activity: T,
- context: &LemmyContext,
-) -> Result<HttpResponse, LemmyError>
+/// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
+/// headers.
+///
+/// actix-web doesn't allow pretty-print for json so we need to do this manually.
+fn create_apub_response<T>(data: &T) -> LemmyResult<HttpResponse>
where
- T: ActivityHandler<DataType = LemmyContext>
- + ActivityFields
- + Clone
- + Deserialize<'a>
- + Serialize
- + std::fmt::Debug
- + Send
- + 'static,
+ T: Serialize,
{
- let request_counter = &mut 0;
- let actor =
- get_or_fetch_and_upsert_actor(activity.actor().clone(), context, request_counter).await?;
- verify_signature(&request, &actor.public_key().context(location_info!())?)?;
+ let json = serde_json::to_string_pretty(&WithContext::new(data, CONTEXT.clone()))?;
- // Do nothing if we received the same activity before
- if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
- return Ok(HttpResponse::Ok().finish());
- }
- check_is_apub_id_valid(activity.actor(), false, &context.settings())?;
- info!("Verifying activity {}", activity.id_unchecked().to_string());
- activity
- .verify(&Data::new(context.clone()), request_counter)
- .await?;
- assert_activity_not_local(&activity, &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.id_unchecked(),
- activity.clone(),
- false,
- true,
- context.pool(),
+ Ok(
+ HttpResponse::Ok()
+ .content_type(FEDERATION_CONTENT_TYPE)
+ .body(json),
)
- .await?;
-
- info!("Receiving activity {}", activity.id_unchecked().to_string());
- activity
- .receive(&Data::new(context.clone()), request_counter)
- .await?;
- Ok(HttpResponse::Ok().finish())
}
-/// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
-/// headers.
-fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
-where
- T: Serialize,
-{
- HttpResponse::Ok()
- .content_type(APUB_JSON_CONTENT_TYPE)
- .json(WithContext::new(data))
+fn create_apub_tombstone_response<T: Into<Url>>(id: T) -> LemmyResult<HttpResponse> {
+ let tombstone = Tombstone::new(id.into());
+ let json = serde_json::to_string_pretty(&WithContext::new(tombstone, CONTEXT.deref().clone()))?;
+
+ Ok(
+ HttpResponse::Gone()
+ .content_type(FEDERATION_CONTENT_TYPE)
+ .status(StatusCode::GONE)
+ .body(json),
+ )
}
-fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
-where
- T: Serialize,
-{
- HttpResponse::Gone()
- .content_type(APUB_JSON_CONTENT_TYPE)
- .status(StatusCode::GONE)
- .json(WithContext::new(data))
+fn err_object_not_local() -> LemmyError {
+ LemmyErrorType::ObjectNotLocal.into()
}
#[derive(Deserialize)]
}
/// Return the ActivityPub json representation of a local activity over HTTP.
+#[tracing::instrument(skip_all)]
pub(crate) async fn get_activity(
info: web::Path<ActivityQuery>,
context: web::Data<LemmyContext>,
-) -> Result<HttpResponse<Body>, LemmyError> {
+) -> Result<HttpResponse, LemmyError> {
let settings = context.settings();
let activity_id = Url::parse(&format!(
"{}/activities/{}/{}",
info.id
))?
.into();
- let activity = blocking(context.pool(), move |conn| {
- Activity::read_from_apub_id(conn, &activity_id)
- })
- .await??;
+ let activity = SentActivity::read_from_apub_id(&mut context.pool(), &activity_id).await?;
- let sensitive = activity.sensitive.unwrap_or(true);
- if !activity.local || sensitive {
- Ok(HttpResponse::NotFound().finish())
+ let sensitive = activity.sensitive;
+ if sensitive {
+ Ok(HttpResponse::Forbidden().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),
- }
-}
-
-fn assert_activity_not_local<T: Debug + ActivityFields>(
- activity: &T,
- hostname: &str,
-) -> Result<(), LemmyError> {
- let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
-
- if activity_domain == hostname {
- return Err(
- anyhow!(
- "Error: received activity which was sent by local instance: {:?}",
- activity
- )
- .into(),
- );
+ create_apub_response(&activity.data)
}
- Ok(())
}
projects / lemmy.git / blobdiff