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, 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 =
  94     get_or_fetch_and_upsert_actor(activity.actor().clone(), context, request_counter).await?;
  95   verify_signature(&request, &actor.public_key().context(location_info!())?)?;
  96
  97   // Do nothing if we received the same activity before
  98   if is_activity_already_known(context.pool(), activity.id_unchecked()).await? {
  99     return Ok(HttpResponse::Ok().finish());
 100   }
 101   check_is_apub_id_valid(activity.actor(), false, &context.settings())?;
 102   info!("Verifying activity {}", activity.id_unchecked().to_string());
 103   activity.verify(context, request_counter).await?;
 104   assert_activity_not_local(&activity, &context.settings().hostname)?;
 105
 106   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
 107   // if we receive the same activity twice in very quick succession.
 108   insert_activity(
 109     activity.id_unchecked(),
 110     activity.clone(),
 111     false,
 112     true,
 113     context.pool(),
 114   )
 115   .await?;
 116
 117   info!("Receiving activity {}", activity.id_unchecked().to_string());
 118   activity.receive(context, request_counter).await?;
 119   Ok(HttpResponse::Ok().finish())
 120 }
 121
 122 /// Convert the data to json and turn it into an HTTP Response with the correct ActivityPub
 123 /// headers.
 124 fn create_apub_response<T>(data: &T) -> HttpResponse<Body>
 125 where
 126   T: Serialize,
 127 {
 128   HttpResponse::Ok()
 129     .content_type(APUB_JSON_CONTENT_TYPE)
 130     .json(data)
 131 }
 132
 133 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
 134 where
 135   T: Serialize,
 136 {
 137   HttpResponse::Gone()
 138     .content_type(APUB_JSON_CONTENT_TYPE)
 139     .status(StatusCode::GONE)
 140     .json(data)
 141 }
 142
 143 #[derive(Deserialize)]
 144 pub struct ActivityQuery {
 145   type_: String,
 146   id: String,
 147 }
 148
 149 /// Return the ActivityPub json representation of a local activity over HTTP.
 150 pub(crate) async fn get_activity(
 151   info: web::Path<ActivityQuery>,
 152   context: web::Data<LemmyContext>,
 153 ) -> Result<HttpResponse<Body>, LemmyError> {
 154   let settings = context.settings();
 155   let activity_id = Url::parse(&format!(
 156     "{}/activities/{}/{}",
 157     settings.get_protocol_and_hostname(),
 158     info.type_,
 159     info.id
 160   ))?
 161   .into();
 162   let activity = blocking(context.pool(), move |conn| {
 163     Activity::read_from_apub_id(conn, &activity_id)
 164   })
 165   .await??;
 166
 167   let sensitive = activity.sensitive.unwrap_or(true);
 168   if !activity.local || sensitive {
 169     Ok(HttpResponse::NotFound().finish())
 170   } else {
 171     Ok(create_apub_response(&activity.data))
 172   }
 173 }
 174
 175 pub(crate) async fn is_activity_already_known(
 176   pool: &DbPool,
 177   activity_id: &Url,
 178 ) -> Result<bool, LemmyError> {
 179   let activity_id = activity_id.to_owned().into();
 180   let existing = blocking(pool, move |conn| {
 181     Activity::read_from_apub_id(conn, &activity_id)
 182   })
 183   .await?;
 184   match existing {
 185     Ok(_) => Ok(true),
 186     Err(_) => Ok(false),
 187   }
 188 }
 189
 190 fn assert_activity_not_local<T: Debug + ActivityFields>(
 191   activity: &T,
 192   hostname: &str,
 193 ) -> Result<(), LemmyError> {
 194   let activity_domain = activity.id_unchecked().domain().context(location_info!())?;
 195
 196   if activity_domain == hostname {
 197     return Err(
 198       anyhow!(
 199         "Error: received activity which was sent by local instance: {:?}",
 200         activity
 201       )
 202       .into(),
 203     );
 204   }
 205   Ok(())
 206 }