2 check_is_apub_id_valid,
3 extensions::signatures::verify_signature,
4 fetcher::get_or_fetch_and_upsert_actor,
6 community::{receive_group_inbox, GroupInboxActivities},
7 person::{receive_person_inbox, PersonInboxActivities},
10 APUB_JSON_CONTENT_TYPE,
15 web::{Bytes, BytesMut, Payload},
19 use anyhow::{anyhow, Context};
20 use futures::StreamExt;
22 use lemmy_api_common::blocking;
23 use lemmy_apub_lib::{ActivityFields, ActivityHandler};
24 use lemmy_db_queries::{source::activity::Activity_, DbPool};
25 use lemmy_db_schema::source::activity::Activity;
26 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
27 use lemmy_websocket::LemmyContext;
28 use log::{info, trace};
29 use serde::{Deserialize, Serialize};
30 use std::{fmt::Debug, io::Read};
39 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
41 pub enum SharedInboxActivities {
42 GroupInboxActivities(GroupInboxActivities),
43 // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
44 // avoid this problem by replacing createpm.object with our own struct, instead of NoteExt.
45 PersonInboxActivities(PersonInboxActivities),
48 pub async fn shared_inbox(
51 context: web::Data<LemmyContext>,
52 ) -> Result<HttpResponse, LemmyError> {
53 let unparsed = payload_to_string(payload).await?;
54 trace!("Received shared inbox activity {}", unparsed);
55 let activity = serde_json::from_str::<SharedInboxActivities>(&unparsed)?;
57 SharedInboxActivities::GroupInboxActivities(g) => {
58 receive_group_inbox(g, request, &context).await
60 SharedInboxActivities::PersonInboxActivities(p) => {
61 receive_person_inbox(p, request, &context).await
66 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
67 let mut bytes = BytesMut::new();
68 while let Some(item) = payload.next().await {
69 bytes.extend_from_slice(&item?);
71 let mut unparsed = String::new();
72 Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
76 // TODO: move most of this code to library
77 async fn receive_activity<'a, T>(
80 context: &LemmyContext,
81 ) -> Result<HttpResponse, LemmyError>
92 let request_counter = &mut 0;
93 let actor = get_or_fetch_and_upsert_actor(activity.actor(), context, request_counter).await?;
94 verify_signature(&request, &actor.public_key().context(location_info!())?)?;
96 // Do nothing if we received the same activity before
97 if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
98 return Ok(HttpResponse::Ok().finish());
100 check_is_apub_id_valid(activity.actor(), false)?;
101 info!("Verifying activity {}", activity.id_unchecked().to_string());
102 activity.verify(context, request_counter).await?;
103 assert_activity_not_local(&activity)?;
105 // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
106 // if we receive the same activity twice in very quick succession.
108 activity.id_unchecked(),
116 info!("Receiving activity {}", activity.id_unchecked().to_string());
117 activity.receive(context, request_counter).await?;
118 Ok(HttpResponse::Ok().finish())
121 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
123 fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
128 .content_type(APUB_JSON_CONTENT_TYPE)
132 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
137 .content_type(APUB_JSON_CONTENT_TYPE)
138 .status(StatusCode::GONE)
142 #[derive(Deserialize)]
143 pub struct ActivityQuery {
148 /// Return the ActivityPub json representation of a local activity over HTTP.
149 pub(crate) async fn get_activity(
150 info: web::Path<ActivityQuery>,
151 context: web::Data<LemmyContext>,
152 ) -> Result<HttpResponse<Body>, LemmyError> {
153 let settings = Settings::get();
154 let activity_id = Url::parse(&format!(
155 "{}/activities/{}/{}",
156 settings.get_protocol_and_hostname(),
161 let activity = blocking(context.pool(), move |conn| {
162 Activity::read_from_apub_id(conn, &activity_id)
166 let sensitive = activity.sensitive.unwrap_or(true);
167 if !activity.local || sensitive {
168 Ok(HttpResponse::NotFound().finish())
170 Ok(create_apub_response(&activity.data))
174 pub(crate) async fn is_activity_already_known(
177 ) -> Result<bool, LemmyError> {
178 let activity_id = activity_id.to_owned().into();
179 let existing = blocking(pool, move |conn| {
180 Activity::read_from_apub_id(conn, &activity_id)
189 fn assert_activity_not_local<T: Debug + ActivityFields>(activity: &T) -> Result<(), LemmyError> {
190 let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
192 if activity_domain == Settings::get().hostname {
195 "Error: received activity which was sent by local instance: {:?}",