2 check_is_apub_id_valid,
3 fetcher::get_or_fetch_and_upsert_actor,
5 community::{receive_group_inbox, GroupInboxActivities},
6 person::{receive_person_inbox, PersonInboxActivities},
13 web::{Bytes, BytesMut, Payload},
17 use anyhow::{anyhow, Context};
18 use futures::StreamExt;
20 use lemmy_api_common::blocking;
23 signatures::verify_signature,
24 traits::{ActivityFields, ActivityHandler},
25 APUB_JSON_CONTENT_TYPE,
27 use lemmy_db_schema::{source::activity::Activity, DbPool};
28 use lemmy_utils::{location_info, LemmyError};
29 use lemmy_websocket::LemmyContext;
31 use serde::{Deserialize, Serialize};
32 use std::{fmt::Debug, io::Read};
41 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
43 #[activity_handler(LemmyContext)]
44 pub enum SharedInboxActivities {
45 GroupInboxActivities(GroupInboxActivities),
46 // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
47 // avoid this problem by replacing createpm.object with our own struct, instead of NoteExt.
48 PersonInboxActivities(PersonInboxActivities),
51 pub async fn shared_inbox(
54 context: web::Data<LemmyContext>,
55 ) -> Result<HttpResponse, LemmyError> {
56 let unparsed = payload_to_string(payload).await?;
57 info!("Received shared inbox activity {}", unparsed);
58 let activity = serde_json::from_str::<SharedInboxActivities>(&unparsed)?;
60 SharedInboxActivities::GroupInboxActivities(g) => {
61 receive_group_inbox(g, request, &context).await
63 SharedInboxActivities::PersonInboxActivities(p) => {
64 receive_person_inbox(p, request, &context).await
69 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
70 let mut bytes = BytesMut::new();
71 while let Some(item) = payload.next().await {
72 bytes.extend_from_slice(&item?);
74 let mut unparsed = String::new();
75 Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
79 // TODO: move most of this code to library
80 async fn receive_activity<'a, T>(
83 context: &LemmyContext,
84 ) -> Result<HttpResponse, LemmyError>
86 T: ActivityHandler<DataType = LemmyContext>
95 let request_counter = &mut 0;
97 get_or_fetch_and_upsert_actor(activity.actor().clone(), context, request_counter).await?;
98 verify_signature(&request, &actor.public_key().context(location_info!())?)?;
100 // Do nothing if we received the same activity before
101 if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
102 return Ok(HttpResponse::Ok().finish());
104 check_is_apub_id_valid(activity.actor(), false, &context.settings())?;
105 info!("Verifying activity {}", activity.id_unchecked().to_string());
107 .verify(&Data::new(context.clone()), request_counter)
109 assert_activity_not_local(&activity, &context.settings().hostname)?;
111 // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
112 // if we receive the same activity twice in very quick succession.
114 activity.id_unchecked(),
122 info!("Receiving activity {}", activity.id_unchecked().to_string());
124 .receive(&Data::new(context.clone()), request_counter)
126 Ok(HttpResponse::Ok().finish())
129 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
131 fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
136 .content_type(APUB_JSON_CONTENT_TYPE)
140 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
145 .content_type(APUB_JSON_CONTENT_TYPE)
146 .status(StatusCode::GONE)
150 #[derive(Deserialize)]
151 pub struct ActivityQuery {
156 /// Return the ActivityPub json representation of a local activity over HTTP.
157 pub(crate) async fn get_activity(
158 info: web::Path<ActivityQuery>,
159 context: web::Data<LemmyContext>,
160 ) -> Result<HttpResponse<Body>, LemmyError> {
161 let settings = context.settings();
162 let activity_id = Url::parse(&format!(
163 "{}/activities/{}/{}",
164 settings.get_protocol_and_hostname(),
169 let activity = blocking(context.pool(), move |conn| {
170 Activity::read_from_apub_id(conn, &activity_id)
174 let sensitive = activity.sensitive.unwrap_or(true);
175 if !activity.local || sensitive {
176 Ok(HttpResponse::NotFound().finish())
178 Ok(create_apub_response(&activity.data))
182 pub(crate) async fn is_activity_already_known(
185 ) -> Result<bool, LemmyError> {
186 let activity_id = activity_id.to_owned().into();
187 let existing = blocking(pool, move |conn| {
188 Activity::read_from_apub_id(conn, &activity_id)
197 fn assert_activity_not_local<T: Debug + ActivityFields>(
200 ) -> Result<(), LemmyError> {
201 let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
203 if activity_domain == hostname {
206 "Error: received activity which was sent by local instance: {:?}",