2 activity_lists::SharedInboxActivities,
3 check_is_apub_id_valid,
5 fetcher::user_or_community::UserOrCommunity,
6 http::{community::receive_group_inbox, person::receive_person_inbox},
12 web::{Bytes, BytesMut, Payload},
16 use anyhow::{anyhow, Context};
17 use futures::StreamExt;
19 use lemmy_api_common::blocking;
23 signatures::verify_signature,
24 traits::{ActivityHandler, ActorType},
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 pub async fn shared_inbox(
44 context: web::Data<LemmyContext>,
45 ) -> Result<HttpResponse, LemmyError> {
46 let unparsed = payload_to_string(payload).await?;
47 info!("Received shared inbox activity {}", unparsed);
48 let activity_data: ActivityCommonFields = serde_json::from_str(&unparsed)?;
49 let activity = serde_json::from_str::<WithContext<SharedInboxActivities>>(&unparsed)?;
50 match activity.inner() {
51 SharedInboxActivities::GroupInboxActivities(g) => {
52 receive_group_inbox(*g, activity_data, request, &context).await
54 SharedInboxActivities::PersonInboxActivities(p) => {
55 receive_person_inbox(*p, activity_data, request, &context).await
60 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
61 let mut bytes = BytesMut::new();
62 while let Some(item) = payload.next().await {
63 bytes.extend_from_slice(&item?);
65 let mut unparsed = String::new();
66 Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
70 #[derive(Clone, Debug, Deserialize, Serialize)]
71 #[serde(rename_all = "camelCase")]
72 pub(crate) struct ActivityCommonFields {
74 pub(crate) actor: Url,
77 // TODO: move most of this code to library
78 async fn receive_activity<'a, T>(
81 activity_data: ActivityCommonFields,
82 context: &LemmyContext,
83 ) -> Result<HttpResponse, LemmyError>
85 T: ActivityHandler<DataType = LemmyContext>
93 check_is_apub_id_valid(&activity_data.actor, false, &context.settings())?;
94 let request_counter = &mut 0;
95 let actor = ObjectId::<UserOrCommunity>::new(activity_data.actor)
96 .dereference(context, request_counter)
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_data.id).await? {
102 return Ok(HttpResponse::Ok().finish());
104 info!("Verifying activity {}", activity_data.id.to_string());
106 .verify(&Data::new(context.clone()), request_counter)
108 assert_activity_not_local(&activity_data.id, &context.settings().hostname)?;
110 // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
111 // if we receive the same activity twice in very quick succession.
112 let object_value = serde_json::to_value(&activity)?;
113 insert_activity(&activity_data.id, object_value, false, true, context.pool()).await?;
115 info!("Receiving activity {}", activity_data.id.to_string());
117 .receive(&Data::new(context.clone()), request_counter)
119 Ok(HttpResponse::Ok().finish())
122 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
124 fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
129 .content_type(APUB_JSON_CONTENT_TYPE)
130 .json(WithContext::new(data))
133 fn create_json_apub_response(data: serde_json::Value) -> HttpResponse<Body> {
135 .content_type(APUB_JSON_CONTENT_TYPE)
139 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
144 .content_type(APUB_JSON_CONTENT_TYPE)
145 .status(StatusCode::GONE)
146 .json(WithContext::new(data))
149 #[derive(Deserialize)]
150 pub struct ActivityQuery {
155 /// Return the ActivityPub json representation of a local activity over HTTP.
156 pub(crate) async fn get_activity(
157 info: web::Path<ActivityQuery>,
158 context: web::Data<LemmyContext>,
159 ) -> Result<HttpResponse<Body>, LemmyError> {
160 let settings = context.settings();
161 let activity_id = Url::parse(&format!(
162 "{}/activities/{}/{}",
163 settings.get_protocol_and_hostname(),
168 let activity = blocking(context.pool(), move |conn| {
169 Activity::read_from_apub_id(conn, &activity_id)
173 let sensitive = activity.sensitive.unwrap_or(true);
174 if !activity.local || sensitive {
175 Ok(HttpResponse::NotFound().finish())
177 Ok(create_json_apub_response(activity.data))
181 pub(crate) async fn is_activity_already_known(
184 ) -> Result<bool, LemmyError> {
185 let activity_id = activity_id.to_owned().into();
186 let existing = blocking(pool, move |conn| {
187 Activity::read_from_apub_id(conn, &activity_id)
196 fn assert_activity_not_local(id: &Url, hostname: &str) -> Result<(), LemmyError> {
197 let activity_domain = id.domain().context(location_info!())?;
199 if activity_domain == hostname {
202 "Error: received activity which was sent by local instance: {:?}",