2 check_is_apub_id_valid,
4 fetcher::get_or_fetch_and_upsert_actor,
6 community::{receive_group_inbox, GroupInboxActivities},
7 person::{receive_person_inbox, PersonInboxActivities},
14 web::{Bytes, BytesMut, Payload},
18 use anyhow::{anyhow, Context};
19 use futures::StreamExt;
21 use lemmy_api_common::blocking;
24 signatures::verify_signature,
25 traits::{ActivityFields, ActivityHandler},
26 APUB_JSON_CONTENT_TYPE,
28 use lemmy_db_schema::{source::activity::Activity, DbPool};
29 use lemmy_utils::{location_info, LemmyError};
30 use lemmy_websocket::LemmyContext;
32 use serde::{Deserialize, Serialize};
33 use std::{fmt::Debug, io::Read};
42 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
44 #[activity_handler(LemmyContext)]
45 pub enum SharedInboxActivities {
46 GroupInboxActivities(GroupInboxActivities),
47 // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
48 // avoid this problem by replacing createpm.object with our own struct, instead of NoteExt.
49 PersonInboxActivities(PersonInboxActivities),
52 pub async fn shared_inbox(
55 context: web::Data<LemmyContext>,
56 ) -> Result<HttpResponse, LemmyError> {
57 let unparsed = payload_to_string(payload).await?;
58 info!("Received shared inbox activity {}", unparsed);
59 let activity = serde_json::from_str::<WithContext<SharedInboxActivities>>(&unparsed)?;
60 match activity.inner() {
61 SharedInboxActivities::GroupInboxActivities(g) => {
62 receive_group_inbox(g, request, &context).await
64 SharedInboxActivities::PersonInboxActivities(p) => {
65 receive_person_inbox(p, request, &context).await
70 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
71 let mut bytes = BytesMut::new();
72 while let Some(item) = payload.next().await {
73 bytes.extend_from_slice(&item?);
75 let mut unparsed = String::new();
76 Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
80 // TODO: move most of this code to library
81 async fn receive_activity<'a, T>(
84 context: &LemmyContext,
85 ) -> Result<HttpResponse, LemmyError>
87 T: ActivityHandler<DataType = LemmyContext>
96 let request_counter = &mut 0;
98 get_or_fetch_and_upsert_actor(activity.actor().clone(), context, request_counter).await?;
99 verify_signature(&request, &actor.public_key().context(location_info!())?)?;
101 // Do nothing if we received the same activity before
102 if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
103 return Ok(HttpResponse::Ok().finish());
105 check_is_apub_id_valid(activity.actor(), false, &context.settings())?;
106 info!("Verifying activity {}", activity.id_unchecked().to_string());
108 .verify(&Data::new(context.clone()), request_counter)
110 assert_activity_not_local(&activity, &context.settings().hostname)?;
112 // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
113 // if we receive the same activity twice in very quick succession.
115 activity.id_unchecked(),
123 info!("Receiving activity {}", activity.id_unchecked().to_string());
125 .receive(&Data::new(context.clone()), request_counter)
127 Ok(HttpResponse::Ok().finish())
130 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
132 fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
137 .content_type(APUB_JSON_CONTENT_TYPE)
138 .json(WithContext::new(data))
141 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
146 .content_type(APUB_JSON_CONTENT_TYPE)
147 .status(StatusCode::GONE)
148 .json(WithContext::new(data))
151 #[derive(Deserialize)]
152 pub struct ActivityQuery {
157 /// Return the ActivityPub json representation of a local activity over HTTP.
158 pub(crate) async fn get_activity(
159 info: web::Path<ActivityQuery>,
160 context: web::Data<LemmyContext>,
161 ) -> Result<HttpResponse<Body>, LemmyError> {
162 let settings = context.settings();
163 let activity_id = Url::parse(&format!(
164 "{}/activities/{}/{}",
165 settings.get_protocol_and_hostname(),
170 let activity = blocking(context.pool(), move |conn| {
171 Activity::read_from_apub_id(conn, &activity_id)
175 let sensitive = activity.sensitive.unwrap_or(true);
176 if !activity.local || sensitive {
177 Ok(HttpResponse::NotFound().finish())
179 Ok(create_apub_response(&activity.data))
183 pub(crate) async fn is_activity_already_known(
186 ) -> Result<bool, LemmyError> {
187 let activity_id = activity_id.to_owned().into();
188 let existing = blocking(pool, move |conn| {
189 Activity::read_from_apub_id(conn, &activity_id)
198 fn assert_activity_not_local<T: Debug + ActivityFields>(
201 ) -> Result<(), LemmyError> {
202 let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
204 if activity_domain == hostname {
207 "Error: received activity which was sent by local instance: {:?}",