crates/apub/src/activities/mod.rs

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