crates/apub/src/inbox/shared_inbox.rs

   1 use crate::{
   2   inbox::{
   3     assert_activity_not_local,
   4     community_inbox::{community_receive_message, CommunityAcceptedActivities},
   5     get_activity_id,
   6     get_activity_to_and_cc,
   7     inbox_verify_http_signature,
   8     is_activity_already_known,
   9     is_addressed_to_community_followers,
  10     is_addressed_to_local_person,
  11     person_inbox::{person_receive_message, PersonAcceptedActivities},
  12   },
  13   insert_activity,
  14 };
  15 use activitystreams::{activity::ActorAndObject, prelude::*};
  16 use actix_web::{web, HttpRequest, HttpResponse};
  17 use anyhow::Context;
  18 use lemmy_api_structs::blocking;
  19 use lemmy_db_queries::{ApubObject, DbPool};
  20 use lemmy_db_schema::source::community::Community;
  21 use lemmy_utils::{location_info, LemmyError};
  22 use lemmy_websocket::LemmyContext;
  23 use serde::{Deserialize, Serialize};
  24 use std::fmt::Debug;
  25 use url::Url;
  26
  27 /// Allowed activity types for shared inbox.
  28 #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Deserialize, Serialize)]
  29 #[serde(rename_all = "PascalCase")]
  30 pub enum ValidTypes {
  31   Create,
  32   Update,
  33   Like,
  34   Dislike,
  35   Delete,
  36   Undo,
  37   Remove,
  38   Announce,
  39   Add,
  40 }
  41
  42 // TODO: this isnt entirely correct, cause some of these receive are not ActorAndObject,
  43 //       but it still works due to the anybase conversion
  44 pub type AcceptedActivities = ActorAndObject<ValidTypes>;
  45
  46 /// Handler for all incoming requests to shared inbox.
  47 pub async fn shared_inbox(
  48   request: HttpRequest,
  49   input: web::Json<AcceptedActivities>,
  50   context: web::Data<LemmyContext>,
  51 ) -> Result<HttpResponse, LemmyError> {
  52   let activity = input.into_inner();
  53   // First of all check the http signature
  54   let request_counter = &mut 0;
  55   let actor = inbox_verify_http_signature(&activity, &context, request, request_counter).await?;
  56
  57   // Do nothing if we received the same activity before
  58   let actor_id = actor.actor_id();
  59   let activity_id = get_activity_id(&activity, &actor_id)?;
  60   if is_activity_already_known(context.pool(), &activity_id).await? {
  61     return Ok(HttpResponse::Ok().finish());
  62   }
  63
  64   assert_activity_not_local(&activity)?;
  65   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
  66   // if we receive the same activity twice in very quick succession.
  67   insert_activity(&activity_id, activity.clone(), false, true, context.pool()).await?;
  68
  69   let activity_any_base = activity.clone().into_any_base()?;
  70   let mut res: Option<HttpResponse> = None;
  71   let to_and_cc = get_activity_to_and_cc(&activity);
  72   // Handle community first, so in case the sender is banned by the community, it will error out.
  73   // If we handled the person receive first, the activity would be inserted to the database before the
  74   // community could check for bans.
  75   // Note that an activity can be addressed to a community and to a person (or multiple persons) at the
  76   // same time. In this case we still only handle it once, to avoid duplicate websocket
  77   // notifications.
  78   let community = extract_local_community_from_destinations(&to_and_cc, context.pool()).await?;
  79   if let Some(community) = community {
  80     let community_activity = CommunityAcceptedActivities::from_any_base(activity_any_base.clone())?
  81       .context(location_info!())?;
  82     res = Some(
  83       Box::pin(community_receive_message(
  84         community_activity,
  85         community,
  86         actor.as_ref(),
  87         &context,
  88         request_counter,
  89       ))
  90       .await?,
  91     );
  92   } else if is_addressed_to_local_person(&to_and_cc, context.pool()).await? {
  93     let person_activity = PersonAcceptedActivities::from_any_base(activity_any_base.clone())?
  94       .context(location_info!())?;
  95     // `to_person` is only used for follow activities (which we dont receive here), so no need to pass
  96     // it in
  97     Box::pin(person_receive_message(
  98       person_activity,
  99       None,
 100       actor.as_ref(),
 101       &context,
 102       request_counter,
 103     ))
 104     .await?;
 105   } else if is_addressed_to_community_followers(&to_and_cc, context.pool())
 106     .await?
 107     .is_some()
 108   {
 109     let person_activity = PersonAcceptedActivities::from_any_base(activity_any_base.clone())?
 110       .context(location_info!())?;
 111     res = Some(
 112       Box::pin(person_receive_message(
 113         person_activity,
 114         None,
 115         actor.as_ref(),
 116         &context,
 117         request_counter,
 118       ))
 119       .await?,
 120     );
 121   }
 122
 123   // If none of those, throw an error
 124   if let Some(r) = res {
 125     Ok(r)
 126   } else {
 127     Ok(HttpResponse::NotImplemented().finish())
 128   }
 129 }
 130
 131 /// If `to_and_cc` contains the ID of a local community, return that community, otherwise return
 132 /// None.
 133 ///
 134 /// This doesnt handle the case where an activity is addressed to multiple communities (because
 135 /// Lemmy doesnt generate such activities).
 136 async fn extract_local_community_from_destinations(
 137   to_and_cc: &[Url],
 138   pool: &DbPool,
 139 ) -> Result<Option<Community>, LemmyError> {
 140   for url in to_and_cc {
 141     let url = url.to_owned();
 142     let community = blocking(&pool, move |conn| {
 143       Community::read_from_apub_id(&conn, &url.into())
 144     })
 145     .await?;
 146     if let Ok(c) = community {
 147       if c.local {
 148         return Ok(Some(c));
 149       }
 150     }
 151   }
 152   Ok(None)
 153 }