crates/apub/src/activities/community/announce.rs

   1 use crate::{
   2   activities::{
   3     generate_activity_id,
   4     send_lemmy_activity,
   5     verify_is_public,
   6     verify_person_in_community,
   7   },
   8   activity_lists::AnnouncableActivities,
   9   insert_activity,
  10   objects::community::ApubCommunity,
  11   protocol::{
  12     activities::community::announce::{AnnounceActivity, RawAnnouncableActivities},
  13     Id,
  14     IdOrNestedObject,
  15   },
  16   ActorType,
  17 };
  18 use activitypub_federation::{core::object_id::ObjectId, data::Data, traits::ActivityHandler};
  19 use activitystreams_kinds::{activity::AnnounceType, public};
  20 use lemmy_utils::error::LemmyError;
  21 use lemmy_websocket::LemmyContext;
  22 use serde_json::Value;
  23 use tracing::debug;
  24 use url::Url;
  25
  26 #[async_trait::async_trait(?Send)]
  27 impl ActivityHandler for RawAnnouncableActivities {
  28   type DataType = LemmyContext;
  29   type Error = LemmyError;
  30
  31   fn id(&self) -> &Url {
  32     &self.id
  33   }
  34
  35   fn actor(&self) -> &Url {
  36     &self.actor
  37   }
  38
  39   #[tracing::instrument(skip_all)]
  40   async fn verify(
  41     &self,
  42     _data: &Data<Self::DataType>,
  43     _request_counter: &mut i32,
  44   ) -> Result<(), Self::Error> {
  45     Ok(())
  46   }
  47
  48   #[tracing::instrument(skip_all)]
  49   async fn receive(
  50     self,
  51     data: &Data<Self::DataType>,
  52     request_counter: &mut i32,
  53   ) -> Result<(), Self::Error> {
  54     let activity: AnnouncableActivities = self.clone().try_into()?;
  55     // This is only for sending, not receiving so we reject it.
  56     if let AnnouncableActivities::Page(_) = activity {
  57       return Err(LemmyError::from_message("Cant receive page"));
  58     }
  59     let community = activity.get_community(data, &mut 0).await?;
  60     let actor_id = ObjectId::new(activity.actor().clone());
  61
  62     // verify and receive activity
  63     activity.verify(data, request_counter).await?;
  64     activity.receive(data, request_counter).await?;
  65
  66     // send to community followers
  67     if community.local {
  68       verify_person_in_community(&actor_id, &community, data, &mut 0).await?;
  69       AnnounceActivity::send(self, &community, data).await?;
  70     }
  71     Ok(())
  72   }
  73 }
  74
  75 #[async_trait::async_trait(?Send)]
  76 pub(crate) trait GetCommunity {
  77   async fn get_community(
  78     &self,
  79     context: &LemmyContext,
  80     request_counter: &mut i32,
  81   ) -> Result<ApubCommunity, LemmyError>;
  82 }
  83
  84 impl AnnounceActivity {
  85   pub(crate) fn new(
  86     object: RawAnnouncableActivities,
  87     community: &ApubCommunity,
  88     context: &LemmyContext,
  89   ) -> Result<AnnounceActivity, LemmyError> {
  90     Ok(AnnounceActivity {
  91       actor: ObjectId::new(community.actor_id()),
  92       to: vec![public()],
  93       object: IdOrNestedObject::NestedObject(object),
  94       cc: vec![community.followers_url.clone().into()],
  95       kind: AnnounceType::Announce,
  96       id: generate_activity_id(
  97         &AnnounceType::Announce,
  98         &context.settings().get_protocol_and_hostname(),
  99       )?,
 100     })
 101   }
 102
 103   #[tracing::instrument(skip_all)]
 104   pub async fn send(
 105     object: RawAnnouncableActivities,
 106     community: &ApubCommunity,
 107     context: &LemmyContext,
 108   ) -> Result<(), LemmyError> {
 109     let announce = AnnounceActivity::new(object.clone(), community, context)?;
 110     let inboxes = community.get_follower_inboxes(context).await?;
 111     send_lemmy_activity(context, announce, community, inboxes.clone(), false).await?;
 112
 113     // Pleroma and Mastodon can't handle activities like Announce/Create/Page. So for
 114     // compatibility, we also send Announce/Page so that they can follow Lemmy communities.
 115     let object_parsed = object.try_into()?;
 116     if let AnnouncableActivities::CreateOrUpdatePost(c) = object_parsed {
 117       // Hack: need to convert Page into a format which can be sent as activity, which requires
 118       //       adding actor field.
 119       let announcable_page = RawAnnouncableActivities {
 120         id: c.object.id.clone().into_inner(),
 121         actor: c.actor.clone().into_inner(),
 122         other: serde_json::to_value(c.object)?
 123           .as_object()
 124           .expect("is object")
 125           .clone(),
 126       };
 127       let announce_compat = AnnounceActivity::new(announcable_page, community, context)?;
 128       send_lemmy_activity(context, announce_compat, community, inboxes, false).await?;
 129     }
 130     Ok(())
 131   }
 132 }
 133
 134 #[async_trait::async_trait(?Send)]
 135 impl ActivityHandler for AnnounceActivity {
 136   type DataType = LemmyContext;
 137   type Error = LemmyError;
 138
 139   fn id(&self) -> &Url {
 140     &self.id
 141   }
 142
 143   fn actor(&self) -> &Url {
 144     self.actor.inner()
 145   }
 146
 147   #[tracing::instrument(skip_all)]
 148   async fn verify(
 149     &self,
 150     _context: &Data<LemmyContext>,
 151     _request_counter: &mut i32,
 152   ) -> Result<(), LemmyError> {
 153     verify_is_public(&self.to, &self.cc)?;
 154     Ok(())
 155   }
 156
 157   #[tracing::instrument(skip_all)]
 158   async fn receive(
 159     self,
 160     context: &Data<LemmyContext>,
 161     request_counter: &mut i32,
 162   ) -> Result<(), LemmyError> {
 163     let object: AnnouncableActivities = self
 164       .object
 165       .object(context, request_counter)
 166       .await?
 167       .try_into()?;
 168     // This is only for sending, not receiving so we reject it.
 169     if let AnnouncableActivities::Page(_) = object {
 170       return Err(LemmyError::from_message("Cant receive page"));
 171     }
 172
 173     // we have to verify this here in order to avoid fetching the object twice over http
 174     object.verify(context, request_counter).await?;
 175
 176     let object_value = serde_json::to_value(&object)?;
 177     let insert = insert_activity(object.id(), object_value, false, true, context.pool()).await?;
 178     if !insert {
 179       debug!(
 180         "Received duplicate activity in announce {}",
 181         object.id().to_string()
 182       );
 183       return Ok(());
 184     }
 185     object.receive(context, request_counter).await
 186   }
 187 }
 188
 189 impl Id for RawAnnouncableActivities {
 190   fn object_id(&self) -> &Url {
 191     ActivityHandler::id(self)
 192   }
 193 }
 194
 195 impl TryFrom<RawAnnouncableActivities> for AnnouncableActivities {
 196   type Error = serde_json::error::Error;
 197
 198   fn try_from(value: RawAnnouncableActivities) -> Result<Self, Self::Error> {
 199     let mut map = value.other.clone();
 200     map.insert("id".to_string(), Value::String(value.id.to_string()));
 201     map.insert("actor".to_string(), Value::String(value.actor.to_string()));
 202     serde_json::from_value(Value::Object(map))
 203   }
 204 }
 205
 206 impl TryFrom<AnnouncableActivities> for RawAnnouncableActivities {
 207   type Error = serde_json::error::Error;
 208
 209   fn try_from(value: AnnouncableActivities) -> Result<Self, Self::Error> {
 210     serde_json::from_value(serde_json::to_value(value)?)
 211   }
 212 }