crates/apub/src/http/mod.rs

   1 use crate::{
   2   check_is_apub_id_valid,
   3   extensions::signatures::verify_signature,
   4   fetcher::get_or_fetch_and_upsert_actor,
   5   http::{
   6     community::{receive_group_inbox, GroupInboxActivities},
   7     person::{receive_person_inbox, PersonInboxActivities},
   8   },
   9   insert_activity,
  10   APUB_JSON_CONTENT_TYPE,
  11 };
  12 use actix_web::{
  13   body::Body,
  14   web,
  15   web::{Bytes, BytesMut, Payload},
  16   HttpRequest,
  17   HttpResponse,
  18 };
  19 use anyhow::{anyhow, Context};
  20 use futures::StreamExt;
  21 use http::StatusCode;
  22 use lemmy_api_common::blocking;
  23 use lemmy_apub_lib::{ActivityFields, ActivityHandler};
  24 use lemmy_db_queries::{source::activity::Activity_, DbPool};
  25 use lemmy_db_schema::source::activity::Activity;
  26 use lemmy_utils::{location_info, settings::structs::Settings, LemmyError};
  27 use lemmy_websocket::LemmyContext;
  28 use log::{info, trace};
  29 use serde::{Deserialize, Serialize};
  30 use std::{fmt::Debug, io::Read};
  31 use url::Url;
  32
  33 mod comment;
  34 mod community;
  35 mod person;
  36 mod post;
  37 pub mod routes;
  38
  39 #[derive(Clone, Debug, Deserialize, Serialize, ActivityHandler, ActivityFields)]
  40 #[serde(untagged)]
  41 pub enum SharedInboxActivities {
  42   GroupInboxActivities(GroupInboxActivities),
  43   // Note, pm activities need to be at the end, otherwise comments will end up here. We can probably
  44   // avoid this problem by replacing createpm.object with our own struct, instead of NoteExt.
  45   PersonInboxActivities(PersonInboxActivities),
  46 }
  47
  48 pub async fn shared_inbox(
  49   request: HttpRequest,
  50   payload: Payload,
  51   context: web::Data<LemmyContext>,
  52 ) -> Result<HttpResponse, LemmyError> {
  53   let unparsed = payload_to_string(payload).await?;
  54   trace!("Received shared inbox activity {}", unparsed);
  55   let activity = serde_json::from_str::<SharedInboxActivities>(&unparsed)?;
  56   match activity {
  57     SharedInboxActivities::GroupInboxActivities(g) => {
  58       receive_group_inbox(g, request, &context).await
  59     }
  60     SharedInboxActivities::PersonInboxActivities(p) => {
  61       receive_person_inbox(p, request, &context).await
  62     }
  63   }
  64 }
  65
  66 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
  67   let mut bytes = BytesMut::new();
  68   while let Some(item) = payload.next().await {
  69     bytes.extend_from_slice(&item?);
  70   }
  71   let mut unparsed = String::new();
  72   Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
  73   Ok(unparsed)
  74 }
  75
  76 // TODO: move most of this code to library
  77 async fn receive_activity<'a, T>(
  78   request: HttpRequest,
  79   activity: T,
  80   context: &LemmyContext,
  81 ) -> Result<HttpResponse, LemmyError>
  82 where
  83   T: ActivityHandler
  84     + ActivityFields
  85     + Clone
  86     + Deserialize<'a>
  87     + Serialize
  88     + std::fmt::Debug
  89     + Send
  90     + 'static,
  91 {
  92   let request_counter = &mut 0;
  93   let actor = get_or_fetch_and_upsert_actor(activity.actor(), context, request_counter).await?;
  94   verify_signature(&request, &actor.public_key().context(location_info!())?)?;
  95
  96   // Do nothing if we received the same activity before
  97   if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
  98     return Ok(HttpResponse::Ok().finish());
  99   }
 100   check_is_apub_id_valid(activity.actor(), false)?;
 101   info!("Verifying activity {}", activity.id_unchecked().to_string());
 102   activity.verify(context, request_counter).await?;
 103   assert_activity_not_local(&activity)?;
 104
 105   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
 106   // if we receive the same activity twice in very quick succession.
 107   insert_activity(
 108     activity.id_unchecked(),
 109     activity.clone(),
 110     false,
 111     true,
 112     context.pool(),
 113   )
 114   .await?;
 115
 116   info!("Receiving activity {}", activity.id_unchecked().to_string());
 117   activity.receive(context, request_counter).await?;
 118   Ok(HttpResponse::Ok().finish())
 119 }
 120
 121 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
 122 /// headers.
 123 fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
 124 where
 125   T: Serialize,
 126 {
 127   HttpResponse::Ok()
 128     .content_type(APUB_JSON_CONTENT_TYPE)
 129     .json(data)
 130 }
 131
 132 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
 133 where
 134   T: Serialize,
 135 {
 136   HttpResponse::Gone()
 137     .content_type(APUB_JSON_CONTENT_TYPE)
 138     .status(StatusCode::GONE)
 139     .json(data)
 140 }
 141
 142 #[derive(Deserialize)]
 143 pub struct ActivityQuery {
 144   type_: String,
 145   id: String,
 146 }
 147
 148 /// Return the ActivityPub json representation of a local activity over HTTP.
 149 pub(crate) async fn get_activity(
 150   info: web::Path<ActivityQuery>,
 151   context: web::Data<LemmyContext>,
 152 ) -> Result<HttpResponse<Body>, LemmyError> {
 153   let settings = Settings::get();
 154   let activity_id = Url::parse(&format!(
 155     "{}/activities/{}/{}",
 156     settings.get_protocol_and_hostname(),
 157     info.type_,
 158     info.id
 159   ))?
 160   .into();
 161   let activity = blocking(context.pool(), move |conn| {
 162     Activity::read_from_apub_id(conn, &activity_id)
 163   })
 164   .await??;
 165
 166   let sensitive = activity.sensitive.unwrap_or(true);
 167   if !activity.local || sensitive {
 168     Ok(HttpResponse::NotFound().finish())
 169   } else {
 170     Ok(create_apub_response(&activity.data))
 171   }
 172 }
 173
 174 pub(crate) async fn is_activity_already_known(
 175   pool: &DbPool,
 176   activity_id: &Url,
 177 ) -> Result<bool, LemmyError> {
 178   let activity_id = activity_id.to_owned().into();
 179   let existing = blocking(pool, move |conn| {
 180     Activity::read_from_apub_id(conn, &activity_id)
 181   })
 182   .await?;
 183   match existing {
 184     Ok(_) => Ok(true),
 185     Err(_) => Ok(false),
 186   }
 187 }
 188
 189 fn assert_activity_not_local<T: Debug + ActivityFields>(activity: &T) -> Result<(), LemmyError> {
 190   let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
 191
 192   if activity_domain == Settings::get().hostname {
 193     return Err(
 194       anyhow!(
 195         "Error: received activity which was sent by local instance: {:?}",
 196         activity
 197       )
 198       .into(),
 199     );
 200   }
 201   Ok(())
 202 }