crates/apub/src/lib.rs

   1 use crate::fetcher::post_or_comment::PostOrComment;
   2 use activitypub_federation::config::{Data, UrlVerifier};
   3 use async_trait::async_trait;
   4 use lemmy_api_common::context::LemmyContext;
   5 use lemmy_db_schema::{
   6   source::{activity::ReceivedActivity, instance::Instance, local_site::LocalSite},
   7   utils::{ActualDbPool, DbPool},
   8 };
   9 use lemmy_utils::error::{LemmyError, LemmyErrorType, LemmyResult};
  10 use moka::future::Cache;
  11 use once_cell::sync::Lazy;
  12 use std::{sync::Arc, time::Duration};
  13 use url::Url;
  14
  15 pub mod activities;
  16 pub(crate) mod activity_lists;
  17 pub mod api;
  18 pub(crate) mod collections;
  19 pub mod fetcher;
  20 pub mod http;
  21 pub(crate) mod mentions;
  22 pub mod objects;
  23 pub mod protocol;
  24
  25 pub const FEDERATION_HTTP_FETCH_LIMIT: u32 = 50;
  26 /// All incoming and outgoing federation actions read the blocklist/allowlist and slur filters
  27 /// multiple times. This causes a huge number of database reads if we hit the db directly. So we
  28 /// cache these values for a short time, which will already make a huge difference and ensures that
  29 /// changes take effect quickly.
  30 const BLOCKLIST_CACHE_DURATION: Duration = Duration::from_secs(60);
  31
  32 static CONTEXT: Lazy<Vec<serde_json::Value>> = Lazy::new(|| {
  33   serde_json::from_str(include_str!("../assets/lemmy/context.json")).expect("parse context")
  34 });
  35
  36 #[derive(Clone)]
  37 pub struct VerifyUrlData(pub ActualDbPool);
  38
  39 #[async_trait]
  40 impl UrlVerifier for VerifyUrlData {
  41   async fn verify(&self, url: &Url) -> Result<(), &'static str> {
  42     let local_site_data = local_site_data_cached(&mut (&self.0).into())
  43       .await
  44       .expect("read local site data");
  45     check_apub_id_valid(url, &local_site_data).map_err(|err| match err {
  46       LemmyError {
  47         error_type: LemmyErrorType::FederationDisabled,
  48         ..
  49       } => "Federation disabled",
  50       LemmyError {
  51         error_type: LemmyErrorType::DomainBlocked(_),
  52         ..
  53       } => "Domain is blocked",
  54       LemmyError {
  55         error_type: LemmyErrorType::DomainNotInAllowList(_),
  56         ..
  57       } => "Domain is not in allowlist",
  58       _ => "Failed validating apub id",
  59     })?;
  60     Ok(())
  61   }
  62 }
  63
  64 /// Checks if the ID is allowed for sending or receiving.
  65 ///
  66 /// In particular, it checks for:
  67 /// - federation being enabled (if its disabled, only local URLs are allowed)
  68 /// - the correct scheme (either http or https)
  69 /// - URL being in the allowlist (if it is active)
  70 /// - URL not being in the blocklist (if it is active)
  71 #[tracing::instrument(skip(local_site_data))]
  72 fn check_apub_id_valid(apub_id: &Url, local_site_data: &LocalSiteData) -> Result<(), LemmyError> {
  73   let domain = apub_id.domain().expect("apud id has domain").to_string();
  74
  75   if !local_site_data
  76     .local_site
  77     .as_ref()
  78     .map(|l| l.federation_enabled)
  79     .unwrap_or(true)
  80   {
  81     Err(LemmyErrorType::FederationDisabled)?;
  82   }
  83
  84   if local_site_data
  85     .blocked_instances
  86     .iter()
  87     .any(|i| domain.eq(&i.domain))
  88   {
  89     Err(LemmyErrorType::DomainBlocked(domain.clone()))?;
  90   }
  91
  92   // Only check this if there are instances in the allowlist
  93   if !local_site_data.allowed_instances.is_empty()
  94     && !local_site_data
  95       .allowed_instances
  96       .iter()
  97       .any(|i| domain.eq(&i.domain))
  98   {
  99     Err(LemmyErrorType::DomainNotInAllowList(domain))?;
 100   }
 101
 102   Ok(())
 103 }
 104
 105 #[derive(Clone)]
 106 pub(crate) struct LocalSiteData {
 107   local_site: Option<LocalSite>,
 108   allowed_instances: Vec<Instance>,
 109   blocked_instances: Vec<Instance>,
 110 }
 111
 112 pub(crate) async fn local_site_data_cached(
 113   pool: &mut DbPool<'_>,
 114 ) -> LemmyResult<Arc<LocalSiteData>> {
 115   static CACHE: Lazy<Cache<(), Arc<LocalSiteData>>> = Lazy::new(|| {
 116     Cache::builder()
 117       .max_capacity(1)
 118       .time_to_live(BLOCKLIST_CACHE_DURATION)
 119       .build()
 120   });
 121   Ok(
 122     CACHE
 123       .try_get_with((), async {
 124         let (local_site, allowed_instances, blocked_instances) =
 125           lemmy_db_schema::try_join_with_pool!(pool => (
 126             // LocalSite may be missing
 127             |pool| async {
 128               Ok(LocalSite::read(pool).await.ok())
 129             },
 130             Instance::allowlist,
 131             Instance::blocklist
 132           ))?;
 133
 134         Ok::<_, diesel::result::Error>(Arc::new(LocalSiteData {
 135           local_site,
 136           allowed_instances,
 137           blocked_instances,
 138         }))
 139       })
 140       .await?,
 141   )
 142 }
 143
 144 pub(crate) async fn check_apub_id_valid_with_strictness(
 145   apub_id: &Url,
 146   is_strict: bool,
 147   context: &LemmyContext,
 148 ) -> Result<(), LemmyError> {
 149   let domain = apub_id.domain().expect("apud id has domain").to_string();
 150   let local_instance = context
 151     .settings()
 152     .get_hostname_without_port()
 153     .expect("local hostname is valid");
 154   if domain == local_instance {
 155     return Ok(());
 156   }
 157
 158   let local_site_data = local_site_data_cached(&mut context.pool()).await?;
 159   check_apub_id_valid(apub_id, &local_site_data)?;
 160
 161   // Only check allowlist if this is a community, and there are instances in the allowlist
 162   if is_strict && !local_site_data.allowed_instances.is_empty() {
 163     // need to allow this explicitly because apub receive might contain objects from our local
 164     // instance.
 165     let mut allowed_and_local = local_site_data
 166       .allowed_instances
 167       .iter()
 168       .map(|i| i.domain.clone())
 169       .collect::<Vec<String>>();
 170     let local_instance = context
 171       .settings()
 172       .get_hostname_without_port()
 173       .expect("local hostname is valid");
 174     allowed_and_local.push(local_instance);
 175
 176     let domain = apub_id.domain().expect("apud id has domain").to_string();
 177     if !allowed_and_local.contains(&domain) {
 178       return Err(LemmyErrorType::FederationDisabledByStrictAllowList)?;
 179     }
 180   }
 181   Ok(())
 182 }
 183
 184 /// Store received activities in the database.
 185 ///
 186 /// This ensures that the same activity doesnt get received and processed more than once, which
 187 /// would be a waste of resources.
 188 #[tracing::instrument(skip(data))]
 189 async fn insert_received_activity(
 190   ap_id: &Url,
 191   data: &Data<LemmyContext>,
 192 ) -> Result<(), LemmyError> {
 193   ReceivedActivity::create(&mut data.pool(), &ap_id.clone().into()).await?;
 194   Ok(())
 195 }
 196
 197 #[async_trait::async_trait]
 198 pub trait SendActivity: Sync {
 199   type Response: Sync + Send + Clone;
 200
 201   async fn send_activity(
 202     _request: &Self,
 203     _response: &Self::Response,
 204     _context: &Data<LemmyContext>,
 205   ) -> Result<(), LemmyError> {
 206     Ok(())
 207   }
 208 }