X-Git-Url: http://these/git/?a=blobdiff_plain;f=crates%2Fapub%2Fsrc%2Factivities%2Fmod.rs;h=bc8379f111b9b070420d6f951f74592cf883265c;hb=70fae9d68d65b1e4d153e30d3c065cc315b75eaf;hp=f4c535ee69f9bcb359712ad8710e405856955650;hpb=1664cb326a2ed7aff749a6b2f66da7958e690e31;p=lemmy.git

diff --git a/crates/apub/src/activities/mod.rs b/crates/apub/src/activities/mod.rs
index f4c535ee..bc8379f1 100644
--- a/crates/apub/src/activities/mod.rs
+++ b/crates/apub/src/activities/mod.rs
@@ -1,157 +1,348 @@
+use self::following::send_follow_community;
 use crate::{
-  check_community_or_site_ban,
-  check_is_apub_id_valid,
-  fetcher::{community::get_or_fetch_and_upsert_community, person::get_or_fetch_and_upsert_person},
-  generate_moderators_url,
+  activities::{
+    block::{send_ban_from_community, send_ban_from_site},
+    community::{
+      collection_add::{send_add_mod_to_community, send_feature_post},
+      lock_page::send_lock_post,
+      update::send_update_community,
+    },
+    create_or_update::private_message::send_create_or_update_pm,
+    deletion::{
+      delete_user::delete_user,
+      send_apub_delete_in_community,
+      send_apub_delete_in_community_new,
+      send_apub_delete_private_message,
+      DeletableObjects,
+    },
+    voting::send_like_activity,
+  },
+  objects::{community::ApubCommunity, person::ApubPerson},
+  protocol::activities::{
+    community::report::Report,
+    create_or_update::{note::CreateOrUpdateNote, page::CreateOrUpdatePage},
+    CreateOrUpdateType,
+  },
+  CONTEXT,
+};
+use activitypub_federation::{
+  activity_queue::send_activity,
+  config::Data,
+  fetch::object_id::ObjectId,
+  kinds::public,
+  protocol::context::WithContext,
+  traits::{ActivityHandler, Actor},
 };
 use anyhow::anyhow;
-use lemmy_api_common::blocking;
-use lemmy_apub_lib::{verify_domains_match, ActivityCommonFields};
-use lemmy_db_queries::ApubObject;
+use lemmy_api_common::{
+  context::LemmyContext,
+  send_activity::{ActivityChannel, SendActivityData},
+};
 use lemmy_db_schema::{
-  source::{community::Community, person::Person},
-  DbUrl,
+  newtypes::CommunityId,
+  source::{
+    activity::{SentActivity, SentActivityForm},
+    community::Community,
+    instance::Instance,
+  },
+};
+use lemmy_db_views_actor::structs::{CommunityPersonBanView, CommunityView};
+use lemmy_utils::{
+  error::{LemmyError, LemmyErrorExt, LemmyErrorType, LemmyResult},
+  spawn_try_task,
+  SYNCHRONOUS_FEDERATION,
 };
-use lemmy_db_views_actor::community_view::CommunityView;
-use lemmy_utils::{settings::structs::Settings, LemmyError};
-use lemmy_websocket::LemmyContext;
-use serde::{Deserialize, Serialize};
-use strum_macros::ToString;
+use moka::future::Cache;
+use once_cell::sync::Lazy;
+use serde::Serialize;
+use std::{ops::Deref, sync::Arc, time::Duration};
+use tracing::info;
 use url::{ParseError, Url};
 use uuid::Uuid;
 
-pub mod comment;
+pub mod block;
 pub mod community;
+pub mod create_or_update;
 pub mod deletion;
 pub mod following;
-pub mod post;
-pub mod private_message;
-pub mod removal;
-pub mod send;
+pub mod unfederated;
 pub mod voting;
 
-#[derive(Clone, Debug, ToString, Deserialize, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub enum CreateOrUpdateType {
-  Create,
-  Update,
-}
+/// Amount of time that the list of dead instances is cached. This is only updated once a day,
+/// so there is no harm in caching it for a longer time.
+pub static DEAD_INSTANCE_LIST_CACHE_DURATION: Duration = Duration::from_secs(30 * 60);
 
 /// Checks that the specified Url actually identifies a Person (by fetching it), and that the person
 /// doesn't have a site ban.
+#[tracing::instrument(skip_all)]
 async fn verify_person(
-  person_id: &Url,
-  context: &LemmyContext,
-  request_counter: &mut i32,
+  person_id: &ObjectId<ApubPerson>,
+  context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
-  let person = get_or_fetch_and_upsert_person(person_id, context, request_counter).await?;
+  let person = person_id.dereference(context).await?;
   if person.banned {
-    return Err(anyhow!("Person {} is banned", person_id).into());
+    return Err(anyhow!("Person {} is banned", person_id))
+      .with_lemmy_type(LemmyErrorType::CouldntUpdateComment);
   }
   Ok(())
 }
 
-pub(crate) async fn extract_community(
-  cc: &[Url],
-  context: &LemmyContext,
-  request_counter: &mut i32,
-) -> Result<Community, LemmyError> {
-  let mut cc_iter = cc.iter();
-  loop {
-    if let Some(cid) = cc_iter.next() {
-      if let Ok(c) = get_or_fetch_and_upsert_community(cid, context, request_counter).await {
-        break Ok(c);
-      }
-    } else {
-      return Err(anyhow!("No community found in cc").into());
-    }
-  }
-}
-
 /// Fetches the person and community to verify their type, then checks if person is banned from site
 /// or community.
+#[tracing::instrument(skip_all)]
 pub(crate) async fn verify_person_in_community(
-  person_id: &Url,
-  community_id: &Url,
-  context: &LemmyContext,
-  request_counter: &mut i32,
+  person_id: &ObjectId<ApubPerson>,
+  community: &ApubCommunity,
+  context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
-  let community = get_or_fetch_and_upsert_community(community_id, context, request_counter).await?;
-  let person = get_or_fetch_and_upsert_person(person_id, context, request_counter).await?;
-  check_community_or_site_ban(&person, community.id, context.pool()).await
+  let person = person_id.dereference(context).await?;
+  if person.banned {
+    return Err(LemmyErrorType::PersonIsBannedFromSite)?;
+  }
+  let person_id = person.id;
+  let community_id = community.id;
+  let is_banned = CommunityPersonBanView::get(&mut context.pool(), person_id, community_id)
+    .await
+    .is_ok();
+  if is_banned {
+    return Err(LemmyErrorType::PersonIsBannedFromCommunity)?;
+  }
+
+  Ok(())
 }
 
-/// Simply check that the url actually refers to a valid group.
-async fn verify_community(
-  community_id: &Url,
-  context: &LemmyContext,
-  request_counter: &mut i32,
+/// Verify that mod action in community was performed by a moderator.
+///
+/// * `mod_id` - Activitypub ID of the mod or admin who performed the action
+/// * `object_id` - Activitypub ID of the actor or object that is being moderated
+/// * `community` - The community inside which moderation is happening
+#[tracing::instrument(skip_all)]
+pub(crate) async fn verify_mod_action(
+  mod_id: &ObjectId<ApubPerson>,
+  object_id: &Url,
+  community_id: CommunityId,
+  context: &Data<LemmyContext>,
 ) -> Result<(), LemmyError> {
-  get_or_fetch_and_upsert_community(community_id, context, request_counter).await?;
-  Ok(())
+  let mod_ = mod_id.dereference(context).await?;
+
+  let is_mod_or_admin =
+    CommunityView::is_mod_or_admin(&mut context.pool(), mod_.id, community_id).await?;
+  if is_mod_or_admin {
+    return Ok(());
+  }
+
+  // mod action comes from the same instance as the moderated object, so it was presumably done
+  // by an instance admin.
+  // TODO: federate instance admin status and check it here
+  if mod_id.inner().domain() == object_id.domain() {
+    return Ok(());
+  }
+
+  Err(LemmyErrorType::NotAModerator)?
 }
 
-fn verify_activity(common: &ActivityCommonFields) -> Result<(), LemmyError> {
-  check_is_apub_id_valid(&common.actor, false)?;
-  verify_domains_match(common.id_unchecked(), &common.actor)?;
+pub(crate) fn verify_is_public(to: &[Url], cc: &[Url]) -> Result<(), LemmyError> {
+  if ![to, cc].iter().any(|set| set.contains(&public())) {
+    Err(LemmyErrorType::ObjectIsNotPublic)?;
+  }
   Ok(())
 }
 
-/// Verify that the actor is a community mod. This check is only run if the community is local,
-/// because in case of remote communities, admins can also perform mod actions. As admin status
-/// is not federated, we cant verify their actions remotely.
-pub(crate) async fn verify_mod_action(
-  actor_id: &Url,
-  community: Url,
-  context: &LemmyContext,
-) -> Result<(), LemmyError> {
-  let community = blocking(context.pool(), move |conn| {
-    Community::read_from_apub_id(conn, &community.into())
-  })
-  .await??;
-
-  if community.local {
-    let actor_id: DbUrl = actor_id.clone().into();
-    let actor = blocking(context.pool(), move |conn| {
-      Person::read_from_apub_id(conn, &actor_id)
-    })
-    .await??;
-
-    // Note: this will also return true for admins in addition to mods, but as we dont know about
-    //       remote admins, it doesnt make any difference.
-    let community_id = community.id;
-    let actor_id = actor.id;
-    let is_mod_or_admin = blocking(context.pool(), move |conn| {
-      CommunityView::is_mod_or_admin(conn, actor_id, community_id)
-    })
-    .await?;
-    if !is_mod_or_admin {
-      return Err(anyhow!("Not a mod").into());
-    }
+pub(crate) fn verify_community_matches<T>(
+  a: &ObjectId<ApubCommunity>,
+  b: T,
+) -> Result<(), LemmyError>
+where
+  T: Into<ObjectId<ApubCommunity>>,
+{
+  let b: ObjectId<ApubCommunity> = b.into();
+  if a != &b {
+    Err(LemmyErrorType::InvalidCommunity)?;
   }
   Ok(())
 }
 
-/// For Add/Remove community moderator activities, check that the target field actually contains
-/// /c/community/moderators. Any different values are unsupported.
-fn verify_add_remove_moderator_target(target: &Url, community: Url) -> Result<(), LemmyError> {
-  if target != &generate_moderators_url(&community.into())?.into_inner() {
-    return Err(anyhow!("Unkown target url").into());
+pub(crate) fn check_community_deleted_or_removed(community: &Community) -> Result<(), LemmyError> {
+  if community.deleted || community.removed {
+    Err(LemmyErrorType::CannotCreatePostOrCommentInDeletedOrRemovedCommunity)?
+  } else {
+    Ok(())
   }
-  Ok(())
 }
 
 /// Generate a unique ID for an activity, in the format:
 /// `http(s)://example.com/receive/create/202daf0a-1489-45df-8d2e-c8a3173fed36`
-fn generate_activity_id<T>(kind: T) -> Result<Url, ParseError>
+fn generate_activity_id<T>(kind: T, protocol_and_hostname: &str) -> Result<Url, ParseError>
 where
   T: ToString,
 {
   let id = format!(
     "{}/activities/{}/{}",
-    Settings::get().get_protocol_and_hostname(),
+    protocol_and_hostname,
     kind.to_string().to_lowercase(),
     Uuid::new_v4()
   );
   Url::parse(&id)
 }
+
+#[tracing::instrument(skip_all)]
+async fn send_lemmy_activity<Activity, ActorT>(
+  data: &Data<LemmyContext>,
+  activity: Activity,
+  actor: &ActorT,
+  mut inbox: Vec<Url>,
+  sensitive: bool,
+) -> Result<(), LemmyError>
+where
+  Activity: ActivityHandler + Serialize + Send + Sync + Clone,
+  ActorT: Actor,
+  Activity: ActivityHandler<Error = LemmyError>,
+{
+  static CACHE: Lazy<Cache<(), Arc<Vec<String>>>> = Lazy::new(|| {
+    Cache::builder()
+      .max_capacity(1)
+      .time_to_live(DEAD_INSTANCE_LIST_CACHE_DURATION)
+      .build()
+  });
+  let dead_instances = CACHE
+    .try_get_with((), async {
+      Ok::<_, diesel::result::Error>(Arc::new(Instance::dead_instances(&mut data.pool()).await?))
+    })
+    .await?;
+
+  inbox.retain(|i| {
+    let domain = i.domain().expect("has domain").to_string();
+    !dead_instances.contains(&domain)
+  });
+  info!("Sending activity {}", activity.id().to_string());
+  let activity = WithContext::new(activity, CONTEXT.deref().clone());
+
+  let form = SentActivityForm {
+    ap_id: activity.id().clone().into(),
+    data: serde_json::to_value(activity.clone())?,
+    sensitive,
+  };
+  SentActivity::create(&mut data.pool(), form).await?;
+  send_activity(activity, actor, inbox, data).await?;
+
+  Ok(())
+}
+
+pub async fn handle_outgoing_activities(context: Data<LemmyContext>) -> LemmyResult<()> {
+  while let Some(data) = ActivityChannel::retrieve_activity().await {
+    match_outgoing_activities(data, &context.reset_request_count()).await?
+  }
+  Ok(())
+}
+
+pub async fn match_outgoing_activities(
+  data: SendActivityData,
+  context: &Data<LemmyContext>,
+) -> LemmyResult<()> {
+  let context = context.reset_request_count();
+  let fed_task = async {
+    use SendActivityData::*;
+    match data {
+      CreatePost(post) => {
+        let creator_id = post.creator_id;
+        CreateOrUpdatePage::send(post, creator_id, CreateOrUpdateType::Create, context).await
+      }
+      UpdatePost(post) => {
+        let creator_id = post.creator_id;
+        CreateOrUpdatePage::send(post, creator_id, CreateOrUpdateType::Update, context).await
+      }
+      DeletePost(post, person, data) => {
+        send_apub_delete_in_community_new(
+          person,
+          post.community_id,
+          DeletableObjects::Post(post.into()),
+          None,
+          data.deleted,
+          context,
+        )
+        .await
+      }
+      RemovePost(post, person, data) => {
+        send_apub_delete_in_community_new(
+          person,
+          post.community_id,
+          DeletableObjects::Post(post.into()),
+          data.reason.or_else(|| Some(String::new())),
+          data.removed,
+          context,
+        )
+        .await
+      }
+      LockPost(post, actor, locked) => send_lock_post(post, actor, locked, context).await,
+      FeaturePost(post, actor, featured) => send_feature_post(post, actor, featured, context).await,
+      CreateComment(comment) => {
+        let creator_id = comment.creator_id;
+        CreateOrUpdateNote::send(comment, creator_id, CreateOrUpdateType::Create, context).await
+      }
+      UpdateComment(comment) => {
+        let creator_id = comment.creator_id;
+        CreateOrUpdateNote::send(comment, creator_id, CreateOrUpdateType::Update, context).await
+      }
+      DeleteComment(comment, actor, community) => {
+        let is_deleted = comment.deleted;
+        let deletable = DeletableObjects::Comment(comment.into());
+        send_apub_delete_in_community(actor, community, deletable, None, is_deleted, &context).await
+      }
+      RemoveComment(comment, actor, community, reason) => {
+        let is_removed = comment.removed;
+        let deletable = DeletableObjects::Comment(comment.into());
+        send_apub_delete_in_community(actor, community, deletable, reason, is_removed, &context)
+          .await
+      }
+      LikePostOrComment(object_id, person, community, score) => {
+        send_like_activity(object_id, person, community, score, context).await
+      }
+      FollowCommunity(community, person, follow) => {
+        send_follow_community(community, person, follow, &context).await
+      }
+      UpdateCommunity(actor, community) => send_update_community(community, actor, context).await,
+      DeleteCommunity(actor, community, removed) => {
+        let deletable = DeletableObjects::Community(community.clone().into());
+        send_apub_delete_in_community(actor, community, deletable, None, removed, &context).await
+      }
+      RemoveCommunity(actor, community, reason, removed) => {
+        let deletable = DeletableObjects::Community(community.clone().into());
+        send_apub_delete_in_community(
+          actor,
+          community,
+          deletable,
+          reason.clone().or_else(|| Some(String::new())),
+          removed,
+          &context,
+        )
+        .await
+      }
+      AddModToCommunity(actor, community_id, updated_mod_id, added) => {
+        send_add_mod_to_community(actor, community_id, updated_mod_id, added, context).await
+      }
+      BanFromCommunity(mod_, community_id, target, data) => {
+        send_ban_from_community(mod_, community_id, target, data, context).await
+      }
+      BanFromSite(mod_, target, data) => send_ban_from_site(mod_, target, data, context).await,
+      CreatePrivateMessage(pm) => {
+        send_create_or_update_pm(pm, CreateOrUpdateType::Create, context).await
+      }
+      UpdatePrivateMessage(pm) => {
+        send_create_or_update_pm(pm, CreateOrUpdateType::Update, context).await
+      }
+      DeletePrivateMessage(person, pm, deleted) => {
+        send_apub_delete_private_message(&person.into(), pm, deleted, context).await
+      }
+      DeleteUser(person) => delete_user(person, context).await,
+      CreateReport(url, actor, community, reason) => {
+        Report::send(ObjectId::from(url), actor, community, reason, context).await
+      }
+    }
+  };
+  if *SYNCHRONOUS_FEDERATION {
+    fed_task.await?;
+  } else {
+    spawn_try_task(fed_task);
+  }
+  Ok(())
+}