crates/apub/src/http/mod.rs

   1 use crate::{
   2   activity_lists::SharedInboxActivities,
   3   check_is_apub_id_valid,
   4   context::WithContext,
   5   fetcher::user_or_community::UserOrCommunity,
   6   http::{community::receive_group_inbox, person::receive_person_inbox},
   7   insert_activity,
   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::{
  21   data::Data,
  22   object_id::ObjectId,
  23   signatures::verify_signature,
  24   traits::{ActivityHandler, ActorType},
  25   APUB_JSON_CONTENT_TYPE,
  26 };
  27 use lemmy_db_schema::{source::activity::Activity, DbPool};
  28 use lemmy_utils::{location_info, LemmyError};
  29 use lemmy_websocket::LemmyContext;
  30 use log::info;
  31 use serde::{Deserialize, Serialize};
  32 use std::{fmt::Debug, io::Read};
  33 use url::Url;
  34
  35 mod comment;
  36 mod community;
  37 mod person;
  38 mod post;
  39 pub mod routes;
  40
  41 pub async fn shared_inbox(
  42   request: HttpRequest,
  43   payload: Payload,
  44   context: web::Data<LemmyContext>,
  45 ) -> Result<HttpResponse, LemmyError> {
  46   let unparsed = payload_to_string(payload).await?;
  47   info!("Received shared inbox activity {}", unparsed);
  48   let activity_data: ActivityCommonFields = serde_json::from_str(&unparsed)?;
  49   let activity = serde_json::from_str::<WithContext<SharedInboxActivities>>(&unparsed)?;
  50   match activity.inner() {
  51     SharedInboxActivities::GroupInboxActivities(g) => {
  52       receive_group_inbox(*g, activity_data, request, &context).await
  53     }
  54     SharedInboxActivities::PersonInboxActivities(p) => {
  55       receive_person_inbox(*p, activity_data, request, &context).await
  56     }
  57   }
  58 }
  59
  60 async fn payload_to_string(mut payload: Payload) -> Result<String, LemmyError> {
  61   let mut bytes = BytesMut::new();
  62   while let Some(item) = payload.next().await {
  63     bytes.extend_from_slice(&item?);
  64   }
  65   let mut unparsed = String::new();
  66   Bytes::from(bytes).as_ref().read_to_string(&mut unparsed)?;
  67   Ok(unparsed)
  68 }
  69
  70 #[derive(Clone, Debug, Deserialize, Serialize)]
  71 #[serde(rename_all = "camelCase")]
  72 pub(crate) struct ActivityCommonFields {
  73   pub(crate) id: Url,
  74   pub(crate) actor: Url,
  75 }
  76
  77 // TODO: move most of this code to library
  78 async fn receive_activity<'a, T>(
  79   request: HttpRequest,
  80   activity: T,
  81   activity_data: ActivityCommonFields,
  82   context: &LemmyContext,
  83 ) -> Result<HttpResponse, LemmyError>
  84 where
  85   T: ActivityHandler<DataType = LemmyContext>
  86     + Clone
  87     + Deserialize<'a>
  88     + Serialize
  89     + std::fmt::Debug
  90     + Send
  91     + 'static,
  92 {
  93   check_is_apub_id_valid(&activity_data.actor, false, &context.settings())?;
  94   let request_counter = &mut 0;
  95   let actor = ObjectId::<UserOrCommunity>::new(activity_data.actor)
  96     .dereference(context, request_counter)
  97     .await?;
  98   verify_signature(&request, &actor.public_key())?;
  99
 100   // Do nothing if we received the same activity before
 101   if is_activity_already_known(context.pool(), &activity_data.id).await? {
 102     return Ok(HttpResponse::Ok().finish());
 103   }
 104   info!("Verifying activity {}", activity_data.id.to_string());
 105   activity
 106     .verify(&Data::new(context.clone()), request_counter)
 107     .await?;
 108   assert_activity_not_local(&activity_data.id, &context.settings().hostname)?;
 109
 110   // Log the activity, so we avoid receiving and parsing it twice. Note that this could still happen
 111   // if we receive the same activity twice in very quick succession.
 112   let object_value = serde_json::to_value(&activity)?;
 113   insert_activity(&activity_data.id, object_value, false, true, context.pool()).await?;
 114
 115   info!("Receiving activity {}", activity_data.id.to_string());
 116   activity
 117     .receive(&Data::new(context.clone()), request_counter)
 118     .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(WithContext::new(data))
 131 }
 132
 133 fn create_json_apub_response(data: serde_json::Value) -> HttpResponse<Body> {
 134   HttpResponse::Ok()
 135     .content_type(APUB_JSON_CONTENT_TYPE)
 136     .json(data)
 137 }
 138
 139 fn create_apub_tombstone_response<T>(data: &T) -> HttpResponse<Body>
 140 where
 141   T: Serialize,
 142 {
 143   HttpResponse::Gone()
 144     .content_type(APUB_JSON_CONTENT_TYPE)
 145     .status(StatusCode::GONE)
 146     .json(WithContext::new(data))
 147 }
 148
 149 #[derive(Deserialize)]
 150 pub struct ActivityQuery {
 151   type_: String,
 152   id: String,
 153 }
 154
 155 /// Return the ActivityPub json representation of a local activity over HTTP.
 156 pub(crate) async fn get_activity(
 157   info: web::Path<ActivityQuery>,
 158   context: web::Data<LemmyContext>,
 159 ) -> Result<HttpResponse<Body>, LemmyError> {
 160   let settings = context.settings();
 161   let activity_id = Url::parse(&format!(
 162     "{}/activities/{}/{}",
 163     settings.get_protocol_and_hostname(),
 164     info.type_,
 165     info.id
 166   ))?
 167   .into();
 168   let activity = blocking(context.pool(), move |conn| {
 169     Activity::read_from_apub_id(conn, &activity_id)
 170   })
 171   .await??;
 172
 173   let sensitive = activity.sensitive.unwrap_or(true);
 174   if !activity.local || sensitive {
 175     Ok(HttpResponse::NotFound().finish())
 176   } else {
 177     Ok(create_json_apub_response(activity.data))
 178   }
 179 }
 180
 181 pub(crate) async fn is_activity_already_known(
 182   pool: &DbPool,
 183   activity_id: &Url,
 184 ) -> Result<bool, LemmyError> {
 185   let activity_id = activity_id.to_owned().into();
 186   let existing = blocking(pool, move |conn| {
 187     Activity::read_from_apub_id(conn, &activity_id)
 188   })
 189   .await?;
 190   match existing {
 191     Ok(_) => Ok(true),
 192     Err(_) => Ok(false),
 193   }
 194 }
 195
 196 fn assert_activity_not_local(id: &Url, hostname: &str) -> Result<(), LemmyError> {
 197   let activity_domain = id.domain().context(location_info!())?;
 198
 199   if activity_domain == hostname {
 200     return Err(
 201       anyhow!(
 202         "Error: received activity which was sent by local instance: {:?}",
 203         id
 204       )
 205       .into(),
 206     );
 207   }
 208   Ok(())
 209 }