lemmy_apub/src/inbox/user_inbox.rs

   1 use crate::{
   2   activities::receive::{
   3     comment::{receive_create_comment, receive_update_comment},
   4     community::{
   5       receive_delete_community,
   6       receive_remove_community,
   7       receive_undo_delete_community,
   8       receive_undo_remove_community,
   9     },
  10     private_message::{
  11       receive_create_private_message,
  12       receive_delete_private_message,
  13       receive_undo_delete_private_message,
  14       receive_update_private_message,
  15     },
  16     receive_unhandled_activity,
  17     verify_activity_domains_valid,
  18   },
  19   check_is_apub_id_valid,
  20   fetcher::get_or_fetch_and_upsert_community,
  21   inbox::{
  22     get_activity_id,
  23     get_activity_to_and_cc,
  24     inbox_verify_http_signature,
  25     is_activity_already_known,
  26     is_addressed_to_public,
  27     receive_for_community::{
  28       receive_create_for_community,
  29       receive_delete_for_community,
  30       receive_dislike_for_community,
  31       receive_like_for_community,
  32       receive_remove_for_community,
  33       receive_undo_for_community,
  34       receive_update_for_community,
  35     },
  36   },
  37   insert_activity,
  38   ActorType,
  39 };
  40 use activitystreams::{
  41   activity::{Accept, ActorAndObject, Announce, Create, Delete, Follow, Undo, Update},
  42   base::AnyBase,
  43   prelude::*,
  44 };
  45 use actix_web::{web, HttpRequest, HttpResponse};
  46 use anyhow::{anyhow, Context};
  47 use diesel::NotFound;
  48 use lemmy_db::{
  49   community::{Community, CommunityFollower, CommunityFollowerForm},
  50   private_message::PrivateMessage,
  51   user::User_,
  52   Followable,
  53 };
  54 use lemmy_structs::blocking;
  55 use lemmy_utils::{location_info, LemmyError};
  56 use lemmy_websocket::LemmyContext;
  57 use log::debug;
  58 use serde::{Deserialize, Serialize};
  59 use std::fmt::Debug;
  60 use url::Url;
  61
  62 /// Allowed activities for user inbox.
  63 #[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd, Deserialize, Serialize)]
  64 #[serde(rename_all = "PascalCase")]
  65 pub enum UserValidTypes {
  66   Accept,   // community accepted our follow request
  67   Create,   // create private message
  68   Update,   // edit private message
  69   Delete,   // private message or community deleted by creator
  70   Undo,     // private message or community restored
  71   Remove,   // community removed by admin
  72   Announce, // post, comment or vote in community
  73 }
  74
  75 pub type UserAcceptedActivities = ActorAndObject<UserValidTypes>;
  76
  77 /// Handler for all incoming activities to user inboxes.
  78 pub async fn user_inbox(
  79   request: HttpRequest,
  80   input: web::Json<UserAcceptedActivities>,
  81   path: web::Path<String>,
  82   context: web::Data<LemmyContext>,
  83 ) -> Result<HttpResponse, LemmyError> {
  84   let activity = input.into_inner();
  85   // First of all check the http signature
  86   let request_counter = &mut 0;
  87   let actor = inbox_verify_http_signature(&activity, &context, request, request_counter).await?;
  88
  89   // Do nothing if we received the same activity before
  90   let activity_id = get_activity_id(&activity, &actor.actor_id()?)?;
  91   if is_activity_already_known(context.pool(), &activity_id).await? {
  92     return Ok(HttpResponse::Ok().finish());
  93   }
  94
  95   // Check if the activity is actually meant for us
  96   let username = path.into_inner();
  97   let user = blocking(&context.pool(), move |conn| {
  98     User_::read_from_name(&conn, &username)
  99   })
 100   .await??;
 101   let to_and_cc = get_activity_to_and_cc(&activity)?;
 102   if !to_and_cc.contains(&&user.actor_id()?) {
 103     return Err(anyhow!("Activity delivered to wrong user").into());
 104   }
 105
 106   insert_activity(&activity_id, activity.clone(), false, true, context.pool()).await?;
 107
 108   debug!(
 109     "User {} received activity {:?} from {}",
 110     user.name,
 111     &activity.id_unchecked(),
 112     &actor.actor_id_str()
 113   );
 114
 115   user_receive_message(
 116     activity.clone(),
 117     Some(user.clone()),
 118     actor.as_ref(),
 119     &context,
 120     request_counter,
 121   )
 122   .await
 123 }
 124
 125 /// Receives Accept/Follow, Announce, private messages and community (undo) remove, (undo) delete
 126 pub(crate) async fn user_receive_message(
 127   activity: UserAcceptedActivities,
 128   to_user: Option<User_>,
 129   actor: &dyn ActorType,
 130   context: &LemmyContext,
 131   request_counter: &mut i32,
 132 ) -> Result<HttpResponse, LemmyError> {
 133   // TODO: must be addressed to one or more local users, or to followers of a remote community
 134
 135   // TODO: if it is addressed to community followers, check that at least one local user is following it
 136
 137   let any_base = activity.clone().into_any_base()?;
 138   let kind = activity.kind().context(location_info!())?;
 139   let actor_url = actor.actor_id()?;
 140   match kind {
 141     UserValidTypes::Accept => {
 142       receive_accept(&context, any_base, actor, to_user.unwrap(), request_counter).await?;
 143     }
 144     UserValidTypes::Announce => {
 145       receive_announce(&context, any_base, actor, request_counter).await?
 146     }
 147     UserValidTypes::Create => {
 148       receive_create(&context, any_base, actor_url, request_counter).await?
 149     }
 150     UserValidTypes::Update => {
 151       receive_update(&context, any_base, actor_url, request_counter).await?
 152     }
 153     UserValidTypes::Delete => {
 154       receive_delete(context, any_base, &actor_url, request_counter).await?
 155     }
 156     UserValidTypes::Undo => receive_undo(context, any_base, &actor_url, request_counter).await?,
 157     UserValidTypes::Remove => receive_remove_community(&context, any_base, &actor_url).await?,
 158   };
 159
 160   // TODO: would be logical to move websocket notification code here
 161
 162   Ok(HttpResponse::Ok().finish())
 163 }
 164
 165 /// Handle accepted follows.
 166 async fn receive_accept(
 167   context: &LemmyContext,
 168   activity: AnyBase,
 169   actor: &dyn ActorType,
 170   user: User_,
 171   request_counter: &mut i32,
 172 ) -> Result<(), LemmyError> {
 173   let accept = Accept::from_any_base(activity)?.context(location_info!())?;
 174   verify_activity_domains_valid(&accept, &actor.actor_id()?, false)?;
 175
 176   // TODO: we should check that we actually sent this activity, because the remote instance
 177   //       could just put a fake Follow
 178   let object = accept.object().to_owned().one().context(location_info!())?;
 179   let follow = Follow::from_any_base(object)?.context(location_info!())?;
 180   verify_activity_domains_valid(&follow, &user.actor_id()?, false)?;
 181
 182   let community_uri = accept
 183     .actor()?
 184     .to_owned()
 185     .single_xsd_any_uri()
 186     .context(location_info!())?;
 187
 188   let community =
 189     get_or_fetch_and_upsert_community(&community_uri, context, request_counter).await?;
 190
 191   // Now you need to add this to the community follower
 192   let community_follower_form = CommunityFollowerForm {
 193     community_id: community.id,
 194     user_id: user.id,
 195   };
 196
 197   // This will fail if they're already a follower
 198   blocking(&context.pool(), move |conn| {
 199     CommunityFollower::follow(conn, &community_follower_form).ok()
 200   })
 201   .await?;
 202
 203   Ok(())
 204 }
 205
 206 /// Takes an announce and passes the inner activity to the appropriate handler.
 207 async fn receive_announce(
 208   context: &LemmyContext,
 209   activity: AnyBase,
 210   actor: &dyn ActorType,
 211   request_counter: &mut i32,
 212 ) -> Result<(), LemmyError> {
 213   let announce = Announce::from_any_base(activity)?.context(location_info!())?;
 214   verify_activity_domains_valid(&announce, &actor.actor_id()?, false)?;
 215   is_addressed_to_public(&announce)?;
 216
 217   let kind = announce.object().as_single_kind_str();
 218   let inner_activity = announce
 219     .object()
 220     .to_owned()
 221     .one()
 222     .context(location_info!())?;
 223
 224   let inner_id = inner_activity.id().context(location_info!())?.to_owned();
 225   check_is_apub_id_valid(&inner_id)?;
 226   if is_activity_already_known(context.pool(), &inner_id).await? {
 227     return Ok(());
 228   }
 229
 230   match kind {
 231     Some("Create") => {
 232       receive_create_for_community(context, inner_activity, &inner_id, request_counter).await
 233     }
 234     Some("Update") => {
 235       receive_update_for_community(context, inner_activity, &inner_id, request_counter).await
 236     }
 237     Some("Like") => {
 238       receive_like_for_community(context, inner_activity, &inner_id, request_counter).await
 239     }
 240     Some("Dislike") => {
 241       receive_dislike_for_community(context, inner_activity, &inner_id, request_counter).await
 242     }
 243     Some("Delete") => receive_delete_for_community(context, inner_activity, &inner_id).await,
 244     Some("Remove") => receive_remove_for_community(context, inner_activity, &inner_id).await,
 245     Some("Undo") => {
 246       receive_undo_for_community(context, inner_activity, &inner_id, request_counter).await
 247     }
 248     _ => receive_unhandled_activity(inner_activity),
 249   }
 250 }
 251
 252 async fn receive_create(
 253   context: &LemmyContext,
 254   activity: AnyBase,
 255   expected_domain: Url,
 256   request_counter: &mut i32,
 257 ) -> Result<(), LemmyError> {
 258   let create = Create::from_any_base(activity)?.context(location_info!())?;
 259   verify_activity_domains_valid(&create, &expected_domain, true)?;
 260   if is_addressed_to_public(&create).is_ok() {
 261     receive_create_comment(create, context, request_counter).await
 262   } else {
 263     receive_create_private_message(&context, create, expected_domain, request_counter).await
 264   }
 265 }
 266
 267 async fn receive_update(
 268   context: &LemmyContext,
 269   activity: AnyBase,
 270   expected_domain: Url,
 271   request_counter: &mut i32,
 272 ) -> Result<(), LemmyError> {
 273   let update = Update::from_any_base(activity)?.context(location_info!())?;
 274   verify_activity_domains_valid(&update, &expected_domain, true)?;
 275   if is_addressed_to_public(&update).is_ok() {
 276     receive_update_comment(update, context, request_counter).await
 277   } else {
 278     receive_update_private_message(&context, update, expected_domain, request_counter).await
 279   }
 280 }
 281
 282 async fn receive_delete(
 283   context: &LemmyContext,
 284   any_base: AnyBase,
 285   expected_domain: &Url,
 286   request_counter: &mut i32,
 287 ) -> Result<(), LemmyError> {
 288   use CommunityOrPrivateMessage::*;
 289
 290   let delete = Delete::from_any_base(any_base.clone())?.context(location_info!())?;
 291   verify_activity_domains_valid(&delete, expected_domain, true)?;
 292   let object_uri = delete
 293     .object()
 294     .to_owned()
 295     .single_xsd_any_uri()
 296     .context(location_info!())?;
 297
 298   match find_community_or_private_message_by_id(context, object_uri).await? {
 299     Community(c) => receive_delete_community(context, c).await,
 300     PrivateMessage(p) => receive_delete_private_message(context, delete, p, request_counter).await,
 301   }
 302 }
 303
 304 async fn receive_undo(
 305   context: &LemmyContext,
 306   any_base: AnyBase,
 307   expected_domain: &Url,
 308   request_counter: &mut i32,
 309 ) -> Result<(), LemmyError> {
 310   use CommunityOrPrivateMessage::*;
 311   let undo = Undo::from_any_base(any_base)?.context(location_info!())?;
 312   verify_activity_domains_valid(&undo, expected_domain, true)?;
 313
 314   let inner_activity = undo.object().to_owned().one().context(location_info!())?;
 315   let kind = inner_activity.kind_str();
 316   match kind {
 317     Some("Delete") => {
 318       let delete = Delete::from_any_base(inner_activity)?.context(location_info!())?;
 319       verify_activity_domains_valid(&delete, expected_domain, true)?;
 320       let object_uri = delete
 321         .object()
 322         .to_owned()
 323         .single_xsd_any_uri()
 324         .context(location_info!())?;
 325       match find_community_or_private_message_by_id(context, object_uri).await? {
 326         Community(c) => receive_undo_delete_community(context, undo, c, expected_domain).await,
 327         PrivateMessage(p) => {
 328           receive_undo_delete_private_message(context, undo, expected_domain, p, request_counter)
 329             .await
 330         }
 331       }
 332     }
 333     Some("Remove") => receive_undo_remove_community(context, undo, expected_domain).await,
 334     _ => receive_unhandled_activity(undo),
 335   }
 336 }
 337 enum CommunityOrPrivateMessage {
 338   Community(Community),
 339   PrivateMessage(PrivateMessage),
 340 }
 341
 342 async fn find_community_or_private_message_by_id(
 343   context: &LemmyContext,
 344   apub_id: Url,
 345 ) -> Result<CommunityOrPrivateMessage, LemmyError> {
 346   let ap_id = apub_id.to_string();
 347   let community = blocking(context.pool(), move |conn| {
 348     Community::read_from_actor_id(conn, &ap_id)
 349   })
 350   .await?;
 351   if let Ok(c) = community {
 352     return Ok(CommunityOrPrivateMessage::Community(c));
 353   }
 354
 355   let ap_id = apub_id.to_string();
 356   let private_message = blocking(context.pool(), move |conn| {
 357     PrivateMessage::read_from_apub_id(conn, &ap_id)
 358   })
 359   .await?;
 360   if let Ok(p) = private_message {
 361     return Ok(CommunityOrPrivateMessage::PrivateMessage(p));
 362   }
 363
 364   return Err(NotFound.into());
 365 }