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