crates/apub_receive/src/inbox/shared_inbox.rs

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