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