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_user,
  11     user_inbox::{user_receive_message, UserAcceptedActivities},
  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_db_queries::{ApubObject, DbPool};
  19 use lemmy_db_schema::source::community::Community;
  20 use lemmy_structs::blocking;
  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 }
  40
  41 // TODO: this isnt entirely correct, cause some of these receive are not ActorAndObject,
  42 //       but it still works due to the anybase conversion
  43 pub type AcceptedActivities = ActorAndObject<ValidTypes>;
  44
  45 /// Handler for all incoming requests to shared inbox.
  46 pub async fn shared_inbox(
  47   request: HttpRequest,
  48   input: web::Json<AcceptedActivities>,
  49   context: web::Data<LemmyContext>,
  50 ) -> Result<HttpResponse, LemmyError> {
  51   let activity = input.into_inner();
  52   // First of all check the http signature
  53   let request_counter = &mut 0;
  54   let actor = inbox_verify_http_signature(&activity, &context, request, request_counter).await?;
  55
  56   // Do nothing if we received the same activity before
  57   let actor_id = actor.actor_id();
  58   let activity_id = get_activity_id(&activity, &actor_id)?;
  59   if is_activity_already_known(context.pool(), &activity_id).await? {
  60     return Ok(HttpResponse::Ok().finish());
  61   }
  62
  63   assert_activity_not_local(&activity)?;
  64   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
  65   // if we receive the same activity twice in very quick succession.
  66   insert_activity(&activity_id, activity.clone(), false, true, context.pool()).await?;
  67
  68   let activity_any_base = activity.clone().into_any_base()?;
  69   let mut res: Option<HttpResponse> = None;
  70   let to_and_cc = get_activity_to_and_cc(&activity);
  71   // Handle community first, so in case the sender is banned by the community, it will error out.
  72   // If we handled the user receive first, the activity would be inserted to the database before the
  73   // community could check for bans.
  74   // Note that an activity can be addressed to a community and to a user (or multiple users) at the
  75   // same time. In this case we still only handle it once, to avoid duplicate websocket
  76   // notifications.
  77   let community = extract_local_community_from_destinations(&to_and_cc, context.pool()).await?;
  78   if let Some(community) = community {
  79     let community_activity = CommunityAcceptedActivities::from_any_base(activity_any_base.clone())?
  80       .context(location_info!())?;
  81     res = Some(
  82       community_receive_message(
  83         community_activity,
  84         community,
  85         actor.as_ref(),
  86         &context,
  87         request_counter,
  88       )
  89       .await?,
  90     );
  91   } else if is_addressed_to_local_user(&to_and_cc, context.pool()).await? {
  92     let user_activity = UserAcceptedActivities::from_any_base(activity_any_base.clone())?
  93       .context(location_info!())?;
  94     // `to_user` is only used for follow activities (which we dont receive here), so no need to pass
  95     // it in
  96     user_receive_message(
  97       user_activity,
  98       None,
  99       actor.as_ref(),
 100       &context,
 101       request_counter,
 102     )
 103     .await?;
 104   } else if is_addressed_to_community_followers(&to_and_cc, context.pool())
 105     .await?
 106     .is_some()
 107   {
 108     let user_activity = UserAcceptedActivities::from_any_base(activity_any_base.clone())?
 109       .context(location_info!())?;
 110     res = Some(
 111       user_receive_message(
 112         user_activity,
 113         None,
 114         actor.as_ref(),
 115         &context,
 116         request_counter,
 117       )
 118       .await?,
 119     );
 120   }
 121
 122   // If none of those, throw an error
 123   if let Some(r) = res {
 124     Ok(r)
 125   } else {
 126     Ok(HttpResponse::NotImplemented().finish())
 127   }
 128 }
 129
 130 /// If `to_and_cc` contains the ID of a local community, return that community, otherwise return
 131 /// None.
 132 ///
 133 /// This doesnt handle the case where an activity is addressed to multiple communities (because
 134 /// Lemmy doesnt generate such activities).
 135 async fn extract_local_community_from_destinations(
 136   to_and_cc: &[Url],
 137   pool: &DbPool,
 138 ) -> Result<Option<Community>, LemmyError> {
 139   for url in to_and_cc {
 140     let url = url.to_owned();
 141     let community = blocking(&pool, move |conn| {
 142       Community::read_from_apub_id(&conn, &url.into())
 143     })
 144     .await?;
 145     if let Ok(c) = community {
 146       if c.local {
 147         return Ok(Some(c));
 148       }
 149     }
 150   }
 151   Ok(None)
 152 }