crates/apub/src/activities/mod.rs

   1 use crate::{
   2   objects::{community::ApubCommunity, person::ApubPerson},
   3   protocol::activities::{create_or_update::page::CreateOrUpdatePage, CreateOrUpdateType},
   4   CONTEXT,
   5 };
   6 use activitypub_federation::{
   7   activity_queue::send_activity,
   8   config::Data,
   9   fetch::object_id::ObjectId,
  10   kinds::public,
  11   protocol::context::WithContext,
  12   traits::{ActivityHandler, Actor},
  13 };
  14 use anyhow::anyhow;
  15 use lemmy_api_common::{
  16   context::LemmyContext,
  17   send_activity::{ActivityChannel, SendActivityData},
  18 };
  19 use lemmy_db_schema::{
  20   newtypes::CommunityId,
  21   source::{
  22     activity::{SentActivity, SentActivityForm},
  23     community::Community,
  24     instance::Instance,
  25   },
  26 };
  27 use lemmy_db_views_actor::structs::{CommunityPersonBanView, CommunityView};
  28 use lemmy_utils::{
  29   error::{LemmyError, LemmyErrorExt, LemmyErrorType, LemmyResult},
  30   spawn_try_task,
  31   SYNCHRONOUS_FEDERATION,
  32 };
  33 use moka::future::Cache;
  34 use once_cell::sync::Lazy;
  35 use serde::Serialize;
  36 use std::{ops::Deref, sync::Arc, time::Duration};
  37 use tracing::info;
  38 use url::{ParseError, Url};
  39 use uuid::Uuid;
  40
  41 pub mod block;
  42 pub mod community;
  43 pub mod create_or_update;
  44 pub mod deletion;
  45 pub mod following;
  46 pub mod unfederated;
  47 pub mod voting;
  48
  49 /// Amount of time that the list of dead instances is cached. This is only updated once a day,
  50 /// so there is no harm in caching it for a longer time.
  51 pub static DEAD_INSTANCE_LIST_CACHE_DURATION: Duration = Duration::from_secs(30 * 60);
  52
  53 /// Checks that the specified Url actually identifies a Person (by fetching it), and that the person
  54 /// doesn't have a site ban.
  55 #[tracing::instrument(skip_all)]
  56 async fn verify_person(
  57   person_id: &ObjectId<ApubPerson>,
  58   context: &Data<LemmyContext>,
  59 ) -> Result<(), LemmyError> {
  60   let person = person_id.dereference(context).await?;
  61   if person.banned {
  62     return Err(anyhow!("Person {} is banned", person_id))
  63       .with_lemmy_type(LemmyErrorType::CouldntUpdateComment);
  64   }
  65   Ok(())
  66 }
  67
  68 /// Fetches the person and community to verify their type, then checks if person is banned from site
  69 /// or community.
  70 #[tracing::instrument(skip_all)]
  71 pub(crate) async fn verify_person_in_community(
  72   person_id: &ObjectId<ApubPerson>,
  73   community: &ApubCommunity,
  74   context: &Data<LemmyContext>,
  75 ) -> Result<(), LemmyError> {
  76   let person = person_id.dereference(context).await?;
  77   if person.banned {
  78     return Err(LemmyErrorType::PersonIsBannedFromSite)?;
  79   }
  80   let person_id = person.id;
  81   let community_id = community.id;
  82   let is_banned = CommunityPersonBanView::get(&mut context.pool(), person_id, community_id)
  83     .await
  84     .is_ok();
  85   if is_banned {
  86     return Err(LemmyErrorType::PersonIsBannedFromCommunity)?;
  87   }
  88
  89   Ok(())
  90 }
  91
  92 /// Verify that mod action in community was performed by a moderator.
  93 ///
  94 /// * `mod_id` - Activitypub ID of the mod or admin who performed the action
  95 /// * `object_id` - Activitypub ID of the actor or object that is being moderated
  96 /// * `community` - The community inside which moderation is happening
  97 #[tracing::instrument(skip_all)]
  98 pub(crate) async fn verify_mod_action(
  99   mod_id: &ObjectId<ApubPerson>,
 100   object_id: &Url,
 101   community_id: CommunityId,
 102   context: &Data<LemmyContext>,
 103 ) -> Result<(), LemmyError> {
 104   let mod_ = mod_id.dereference(context).await?;
 105
 106   let is_mod_or_admin =
 107     CommunityView::is_mod_or_admin(&mut context.pool(), mod_.id, community_id).await?;
 108   if is_mod_or_admin {
 109     return Ok(());
 110   }
 111
 112   // mod action comes from the same instance as the moderated object, so it was presumably done
 113   // by an instance admin.
 114   // TODO: federate instance admin status and check it here
 115   if mod_id.inner().domain() == object_id.domain() {
 116     return Ok(());
 117   }
 118
 119   Err(LemmyErrorType::NotAModerator)?
 120 }
 121
 122 pub(crate) fn verify_is_public(to: &[Url], cc: &[Url]) -> Result<(), LemmyError> {
 123   if ![to, cc].iter().any(|set| set.contains(&public())) {
 124     return Err(LemmyErrorType::ObjectIsNotPublic)?;
 125   }
 126   Ok(())
 127 }
 128
 129 pub(crate) fn verify_community_matches<T>(
 130   a: &ObjectId<ApubCommunity>,
 131   b: T,
 132 ) -> Result<(), LemmyError>
 133 where
 134   T: Into<ObjectId<ApubCommunity>>,
 135 {
 136   let b: ObjectId<ApubCommunity> = b.into();
 137   if a != &b {
 138     return Err(LemmyErrorType::InvalidCommunity)?;
 139   }
 140   Ok(())
 141 }
 142
 143 pub(crate) fn check_community_deleted_or_removed(community: &Community) -> Result<(), LemmyError> {
 144   if community.deleted || community.removed {
 145     Err(LemmyErrorType::CannotCreatePostOrCommentInDeletedOrRemovedCommunity)?
 146   } else {
 147     Ok(())
 148   }
 149 }
 150
 151 /// Generate a unique ID for an activity, in the format:
 152 /// `http(s)://example.com/receive/create/202daf0a-1489-45df-8d2e-c8a3173fed36`
 153 fn generate_activity_id<T>(kind: T, protocol_and_hostname: &str) -> Result<Url, ParseError>
 154 where
 155   T: ToString,
 156 {
 157   let id = format!(
 158     "{}/activities/{}/{}",
 159     protocol_and_hostname,
 160     kind.to_string().to_lowercase(),
 161     Uuid::new_v4()
 162   );
 163   Url::parse(&id)
 164 }
 165
 166 #[tracing::instrument(skip_all)]
 167 async fn send_lemmy_activity<Activity, ActorT>(
 168   data: &Data<LemmyContext>,
 169   activity: Activity,
 170   actor: &ActorT,
 171   mut inbox: Vec<Url>,
 172   sensitive: bool,
 173 ) -> Result<(), LemmyError>
 174 where
 175   Activity: ActivityHandler + Serialize + Send + Sync + Clone,
 176   ActorT: Actor,
 177   Activity: ActivityHandler<Error = LemmyError>,
 178 {
 179   static CACHE: Lazy<Cache<(), Arc<Vec<String>>>> = Lazy::new(|| {
 180     Cache::builder()
 181       .max_capacity(1)
 182       .time_to_live(DEAD_INSTANCE_LIST_CACHE_DURATION)
 183       .build()
 184   });
 185   let dead_instances = CACHE
 186     .try_get_with((), async {
 187       Ok::<_, diesel::result::Error>(Arc::new(Instance::dead_instances(&mut data.pool()).await?))
 188     })
 189     .await?;
 190
 191   inbox.retain(|i| {
 192     let domain = i.domain().expect("has domain").to_string();
 193     !dead_instances.contains(&domain)
 194   });
 195   info!("Sending activity {}", activity.id().to_string());
 196   let activity = WithContext::new(activity, CONTEXT.deref().clone());
 197
 198   let form = SentActivityForm {
 199     ap_id: activity.id().clone().into(),
 200     data: serde_json::to_value(activity.clone())?,
 201     sensitive,
 202   };
 203   SentActivity::create(&mut data.pool(), form).await?;
 204   send_activity(activity, actor, inbox, data).await?;
 205
 206   Ok(())
 207 }
 208
 209 pub async fn handle_outgoing_activities(context: Data<LemmyContext>) -> LemmyResult<()> {
 210   while let Some(data) = ActivityChannel::retrieve_activity().await {
 211     match_outgoing_activities(data, &context.reset_request_count()).await?
 212   }
 213   Ok(())
 214 }
 215
 216 pub async fn match_outgoing_activities(
 217   data: SendActivityData,
 218   context: &Data<LemmyContext>,
 219 ) -> LemmyResult<()> {
 220   let fed_task = match data {
 221     SendActivityData::CreatePost(post) => {
 222       let creator_id = post.creator_id;
 223       CreateOrUpdatePage::send(
 224         post,
 225         creator_id,
 226         CreateOrUpdateType::Create,
 227         context.reset_request_count(),
 228       )
 229     }
 230   };
 231   if *SYNCHRONOUS_FEDERATION {
 232     fed_task.await?;
 233   } else {
 234     spawn_try_task(fed_task);
 235   }
 236   Ok(())
 237 }