3 comment::{receive_create_comment, receive_update_comment},
5 receive_delete_community,
6 receive_remove_community,
7 receive_undo_delete_community,
8 receive_undo_remove_community,
11 receive_create_private_message,
12 receive_delete_private_message,
13 receive_undo_delete_private_message,
14 receive_update_private_message,
16 receive_unhandled_activity,
17 verify_activity_domains_valid,
20 assert_activity_not_local,
22 inbox_verify_http_signature,
23 is_activity_already_known,
24 is_addressed_to_community_followers,
25 is_addressed_to_local_person,
26 receive_for_community::{
27 receive_add_for_community,
28 receive_block_user_for_community,
29 receive_create_for_community,
30 receive_delete_for_community,
31 receive_dislike_for_community,
32 receive_like_for_community,
33 receive_remove_for_community,
34 receive_undo_for_community,
35 receive_update_for_community,
37 verify_is_addressed_to_public,
40 use activitystreams::{
41 activity::{Accept, ActorAndObject, Announce, Create, Delete, Follow, Remove, Undo, Update},
45 use actix_web::{web, HttpRequest, HttpResponse};
46 use anyhow::{anyhow, Context};
48 use lemmy_api_common::blocking;
50 check_is_apub_id_valid,
51 fetcher::community::get_or_fetch_and_upsert_community,
52 get_activity_to_and_cc,
56 use lemmy_db_queries::{source::person::Person_, ApubObject, Followable};
57 use lemmy_db_schema::source::{
58 community::{Community, CommunityFollower},
60 private_message::PrivateMessage,
62 use lemmy_utils::{location_info, LemmyError};
63 use lemmy_websocket::LemmyContext;
65 use serde::{Deserialize, Serialize};
67 use strum_macros::EnumString;
70 /// Allowed activities for person inbox.
71 #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Deserialize, Serialize)]
72 #[serde(rename_all = "PascalCase")]
73 pub enum PersonValidTypes {
74 Accept, // community accepted our follow request
75 Create, // create private message
76 Update, // edit private message
77 Delete, // private message or community deleted by creator
78 Undo, // private message or community restored
79 Remove, // community removed by admin
80 Announce, // post, comment or vote in community
83 pub type PersonAcceptedActivities = ActorAndObject<PersonValidTypes>;
85 /// Handler for all incoming activities to person inboxes.
86 pub async fn person_inbox(
88 input: web::Json<PersonAcceptedActivities>,
89 path: web::Path<String>,
90 context: web::Data<LemmyContext>,
91 ) -> Result<HttpResponse, LemmyError> {
92 let activity = input.into_inner();
93 // First of all check the http signature
94 let request_counter = &mut 0;
95 let actor = inbox_verify_http_signature(&activity, &context, request, request_counter).await?;
97 // Do nothing if we received the same activity before
98 let activity_id = get_activity_id(&activity, &actor.actor_id())?;
99 if is_activity_already_known(context.pool(), &activity_id).await? {
100 return Ok(HttpResponse::Ok().finish());
103 // Check if the activity is actually meant for us
104 let username = path.into_inner();
105 let person = blocking(&context.pool(), move |conn| {
106 Person::find_by_name(&conn, &username)
109 let to_and_cc = get_activity_to_and_cc(&activity);
110 // TODO: we should also accept activities that are sent to community followers
111 if !to_and_cc.contains(&&person.actor_id()) {
112 return Err(anyhow!("Activity delivered to wrong person").into());
115 assert_activity_not_local(&activity)?;
116 insert_activity(&activity_id, activity.clone(), false, true, context.pool()).await?;
118 person_receive_message(
120 Some(person.clone()),
128 /// Receives Accept/Follow, Announce, private messages and community (undo) remove, (undo) delete
129 pub(crate) async fn person_receive_message(
130 activity: PersonAcceptedActivities,
131 to_person: Option<Person>,
132 actor: &dyn ActorType,
133 context: &LemmyContext,
134 request_counter: &mut i32,
135 ) -> Result<HttpResponse, LemmyError> {
136 is_for_person_inbox(context, &activity).await?;
139 "User received activity {:?} from {}",
142 .context(location_info!())?
144 &actor.actor_id().to_string()
147 let any_base = activity.clone().into_any_base()?;
148 let kind = activity.kind().context(location_info!())?;
149 let actor_url = actor.actor_id();
151 PersonValidTypes::Accept => {
156 to_person.expect("person provided"),
161 PersonValidTypes::Announce => {
162 Box::pin(receive_announce(&context, any_base, actor, request_counter)).await?
164 PersonValidTypes::Create => {
165 Box::pin(receive_create(
173 PersonValidTypes::Update => {
174 Box::pin(receive_update(
182 PersonValidTypes::Delete => {
183 Box::pin(receive_delete(
191 PersonValidTypes::Undo => {
192 Box::pin(receive_undo(context, any_base, &actor_url, request_counter)).await?
194 PersonValidTypes::Remove => Box::pin(receive_remove(context, any_base, &actor_url)).await?,
197 // TODO: would be logical to move websocket notification code here
199 Ok(HttpResponse::Ok().finish())
202 /// Returns true if the activity is addressed directly to one or more local persons, or if it is
203 /// addressed to the followers collection of a remote community, and at least one local person follows
205 async fn is_for_person_inbox(
206 context: &LemmyContext,
207 activity: &PersonAcceptedActivities,
208 ) -> Result<(), LemmyError> {
209 let to_and_cc = get_activity_to_and_cc(activity);
210 // Check if it is addressed directly to any local person
211 if is_addressed_to_local_person(&to_and_cc, context.pool()).await? {
215 // Check if it is addressed to any followers collection of a remote community, and that the
216 // community has local followers.
217 let community = is_addressed_to_community_followers(&to_and_cc, context.pool()).await?;
218 if let Some(c) = community {
219 let community_id = c.id;
220 let has_local_followers = blocking(&context.pool(), move |conn| {
221 CommunityFollower::has_local_followers(conn, community_id)
226 anyhow!("Remote activity cant be addressed to followers of local community").into(),
229 if has_local_followers {
234 Err(anyhow!("Not addressed for any local person").into())
237 /// Handle accepted follows.
238 async fn receive_accept(
239 context: &LemmyContext,
241 actor: &dyn ActorType,
243 request_counter: &mut i32,
244 ) -> Result<(), LemmyError> {
245 let accept = Accept::from_any_base(activity)?.context(location_info!())?;
246 verify_activity_domains_valid(&accept, &actor.actor_id(), false)?;
248 let object = accept.object().to_owned().one().context(location_info!())?;
249 let follow = Follow::from_any_base(object)?.context(location_info!())?;
250 verify_activity_domains_valid(&follow, &person.actor_id(), false)?;
252 let community_uri = accept
255 .single_xsd_any_uri()
256 .context(location_info!())?;
259 get_or_fetch_and_upsert_community(&community_uri, context, request_counter).await?;
261 let community_id = community.id;
262 let person_id = person.id;
263 // This will throw an error if no follow was requested
264 blocking(&context.pool(), move |conn| {
265 CommunityFollower::follow_accepted(conn, community_id, person_id)
272 #[derive(EnumString)]
273 enum AnnouncableActivities {
285 /// Takes an announce and passes the inner activity to the appropriate handler.
286 pub async fn receive_announce(
287 context: &LemmyContext,
289 actor: &dyn ActorType,
290 request_counter: &mut i32,
291 ) -> Result<(), LemmyError> {
292 let announce = Announce::from_any_base(activity)?.context(location_info!())?;
293 verify_activity_domains_valid(&announce, &actor.actor_id(), false)?;
294 verify_is_addressed_to_public(&announce)?;
298 .as_single_kind_str()
299 .and_then(|s| s.parse().ok());
300 let inner_activity = announce
304 .context(location_info!())?;
306 let inner_id = inner_activity.id().context(location_info!())?.to_owned();
307 check_is_apub_id_valid(&inner_id, false)?;
308 if is_activity_already_known(context.pool(), &inner_id).await? {
312 use AnnouncableActivities::*;
315 receive_create_for_community(context, inner_activity, &inner_id, request_counter).await
318 receive_update_for_community(
328 receive_like_for_community(context, inner_activity, &inner_id, request_counter).await
331 receive_dislike_for_community(context, inner_activity, &inner_id, request_counter).await
334 receive_delete_for_community(
344 receive_remove_for_community(context, inner_activity, Some(announce), request_counter).await
347 receive_undo_for_community(
357 receive_add_for_community(context, inner_activity, Some(announce), request_counter).await
360 receive_block_user_for_community(context, inner_activity, Some(announce), request_counter)
363 _ => receive_unhandled_activity(inner_activity),
367 /// Receive either a new private message, or a new comment mention. We distinguish them by checking
368 /// whether the activity is public.
369 async fn receive_create(
370 context: &LemmyContext,
372 expected_domain: Url,
373 request_counter: &mut i32,
374 ) -> Result<(), LemmyError> {
375 let create = Create::from_any_base(activity)?.context(location_info!())?;
376 verify_activity_domains_valid(&create, &expected_domain, true)?;
377 if verify_is_addressed_to_public(&create).is_ok() {
378 receive_create_comment(create, context, request_counter).await
380 receive_create_private_message(&context, create, expected_domain, request_counter).await
384 /// Receive either an updated private message, or an updated comment mention. We distinguish
385 /// them by checking whether the activity is public.
386 async fn receive_update(
387 context: &LemmyContext,
389 expected_domain: Url,
390 request_counter: &mut i32,
391 ) -> Result<(), LemmyError> {
392 let update = Update::from_any_base(activity)?.context(location_info!())?;
393 verify_activity_domains_valid(&update, &expected_domain, true)?;
394 if verify_is_addressed_to_public(&update).is_ok() {
395 receive_update_comment(update, context, request_counter).await
397 receive_update_private_message(&context, update, expected_domain, request_counter).await
401 async fn receive_delete(
402 context: &LemmyContext,
404 expected_domain: &Url,
405 request_counter: &mut i32,
406 ) -> Result<(), LemmyError> {
407 use CommunityOrPrivateMessage::*;
409 let delete = Delete::from_any_base(any_base.clone())?.context(location_info!())?;
410 verify_activity_domains_valid(&delete, expected_domain, true)?;
411 let object_uri = delete
414 .single_xsd_any_uri()
415 .context(location_info!())?;
417 match find_community_or_private_message_by_id(context, object_uri).await? {
418 Community(c) => receive_delete_community(context, c).await,
419 PrivateMessage(p) => receive_delete_private_message(context, delete, p, request_counter).await,
423 async fn receive_remove(
424 context: &LemmyContext,
426 expected_domain: &Url,
427 ) -> Result<(), LemmyError> {
428 let remove = Remove::from_any_base(any_base.clone())?.context(location_info!())?;
429 verify_activity_domains_valid(&remove, expected_domain, true)?;
430 let object_uri = remove
433 .single_xsd_any_uri()
434 .context(location_info!())?;
435 let community = blocking(context.pool(), move |conn| {
436 Community::read_from_apub_id(conn, &object_uri.into())
439 receive_remove_community(&context, community).await
442 async fn receive_undo(
443 context: &LemmyContext,
445 expected_domain: &Url,
446 request_counter: &mut i32,
447 ) -> Result<(), LemmyError> {
448 let undo = Undo::from_any_base(any_base)?.context(location_info!())?;
449 verify_activity_domains_valid(&undo, expected_domain, true)?;
451 let inner_activity = undo.object().to_owned().one().context(location_info!())?;
452 let kind = inner_activity.kind_str();
455 let delete = Delete::from_any_base(inner_activity)?.context(location_info!())?;
456 verify_activity_domains_valid(&delete, expected_domain, true)?;
457 let object_uri = delete
460 .single_xsd_any_uri()
461 .context(location_info!())?;
462 use CommunityOrPrivateMessage::*;
463 match find_community_or_private_message_by_id(context, object_uri).await? {
464 Community(c) => receive_undo_delete_community(context, c).await,
465 PrivateMessage(p) => {
466 receive_undo_delete_private_message(context, undo, expected_domain, p, request_counter)
472 let remove = Remove::from_any_base(inner_activity)?.context(location_info!())?;
473 let object_uri = remove
476 .single_xsd_any_uri()
477 .context(location_info!())?;
478 let community = blocking(context.pool(), move |conn| {
479 Community::read_from_apub_id(conn, &object_uri.into())
482 receive_undo_remove_community(context, community).await
484 _ => receive_unhandled_activity(undo),
487 enum CommunityOrPrivateMessage {
488 Community(Community),
489 PrivateMessage(PrivateMessage),
492 async fn find_community_or_private_message_by_id(
493 context: &LemmyContext,
495 ) -> Result<CommunityOrPrivateMessage, LemmyError> {
496 let ap_id = apub_id.to_owned();
497 let community = blocking(context.pool(), move |conn| {
498 Community::read_from_apub_id(conn, &ap_id.into())
501 if let Ok(c) = community {
502 return Ok(CommunityOrPrivateMessage::Community(c));
505 let ap_id = apub_id.to_owned();
506 let private_message = blocking(context.pool(), move |conn| {
507 PrivateMessage::read_from_apub_id(conn, &ap_id.into())
510 if let Ok(p) = private_message {
511 return Ok(CommunityOrPrivateMessage::PrivateMessage(p));