From 20ca18f9e315407ae23d5e18ecad660e8890a917 Mon Sep 17 00:00:00 2001 From: alemi Date: Mon, 8 Apr 2024 02:52:51 +0200 Subject: [PATCH] chore: also modularized APInbox --- src/activitypub/mod.rs | 16 ++- src/activitypub/user/inbox.rs | 174 +++++---------------------------- src/activitypub/user/outbox.rs | 4 +- src/errors.rs | 8 +- src/server.rs | 166 ++++++++++++++++++++++++++++++- 5 files changed, 205 insertions(+), 163 deletions(-) diff --git a/src/activitypub/mod.rs b/src/activitypub/mod.rs index 58a3fe65..0a103ab1 100644 --- a/src/activitypub/mod.rs +++ b/src/activitypub/mod.rs @@ -132,11 +132,23 @@ pub async fn auth(State(ctx): State, Json(login): Json) -> R #[axum::async_trait] pub trait APOutbox { - async fn post_note(&self, uid: String, object: serde_json::Value) -> crate::Result; - async fn post_activity(&self, uid: String, activity: serde_json::Value) -> crate::Result; + async fn create_note(&self, uid: String, object: serde_json::Value) -> crate::Result; + async fn create(&self, uid: String, activity: serde_json::Value) -> crate::Result; async fn like(&self, uid: String, activity: serde_json::Value) -> crate::Result; async fn follow(&self, uid: String, activity: serde_json::Value) -> crate::Result; async fn accept(&self, uid: String, activity: serde_json::Value) -> crate::Result; async fn reject(&self, _uid: String, _activity: serde_json::Value) -> crate::Result; async fn undo(&self, uid: String, activity: serde_json::Value) -> crate::Result; } + +#[axum::async_trait] +pub trait APInbox { + async fn create(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn like(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn follow(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn accept(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn reject(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn undo(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn delete(&self, activity: serde_json::Value) -> crate::Result<()>; + async fn update(&self, activity: serde_json::Value) -> crate::Result<()>; +} diff --git a/src/activitypub/user/inbox.rs b/src/activitypub/user/inbox.rs index 6ef7a04e..5059db2f 100644 --- a/src/activitypub/user/inbox.rs +++ b/src/activitypub/user/inbox.rs @@ -1,8 +1,8 @@ use axum::{extract::{Path, Query, State}, http::StatusCode, Json}; -use sea_orm::{sea_query::Expr, ColumnTrait, Condition, EntityTrait, IntoActiveModel, Order, QueryFilter, QueryOrder, QuerySelect, Set}; +use sea_orm::{ColumnTrait, Condition, EntityTrait, Order, QueryFilter, QueryOrder, QuerySelect}; -use apb::{Activity, ActivityType, Object, ObjectType, Base, BaseType}; -use crate::{activitypub::{activity::ap_activity, jsonld::LD, Addressed, JsonLD, Pagination}, auth::{AuthIdentity, Identity}, errors::{LoggableError, UpubError}, model, server::Context, url}; +use apb::{ActivityType, ObjectType, Base, BaseType}; +use crate::{activitypub::{activity::ap_activity, jsonld::LD, APInbox, JsonLD, Pagination}, auth::{AuthIdentity, Identity}, errors::UpubError, model, server::Context, url}; pub async fn get( State(ctx): State, @@ -68,177 +68,49 @@ pub async fn page( pub async fn post( State(ctx): State, Path(_id): Path, - Json(object): Json + Json(activity): Json ) -> Result<(), UpubError> { - match object.base_type() { + match activity.base_type() { None => { Err(StatusCode::BAD_REQUEST.into()) }, Some(BaseType::Link(_x)) => { - tracing::warn!("skipping remote activity: {}", serde_json::to_string_pretty(&object).unwrap()); + tracing::warn!("skipping remote activity: {}", serde_json::to_string_pretty(&activity).unwrap()); Err(StatusCode::UNPROCESSABLE_ENTITY.into()) // we could but not yet }, Some(BaseType::Object(ObjectType::Activity(ActivityType::Activity))) => { - tracing::warn!("skipping unprocessable base activity: {}", serde_json::to_string_pretty(&object).unwrap()); + tracing::warn!("skipping unprocessable base activity: {}", serde_json::to_string_pretty(&activity).unwrap()); Err(StatusCode::UNPROCESSABLE_ENTITY.into()) // won't ingest useless stuff }, - Some(BaseType::Object(ObjectType::Activity(ActivityType::Delete))) => { - // TODO verify the signature before just deleting lmao - let oid = object.object().id().ok_or(StatusCode::BAD_REQUEST)?; - // TODO maybe we should keep the tombstone? - model::user::Entity::delete_by_id(&oid).exec(ctx.db()).await.info_failed("failed deleting from users"); - model::activity::Entity::delete_by_id(&oid).exec(ctx.db()).await.info_failed("failed deleting from activities"); - model::object::Entity::delete_by_id(&oid).exec(ctx.db()).await.info_failed("failed deleting from objects"); - Ok(()) - }, + Some(BaseType::Object(ObjectType::Activity(ActivityType::Delete))) => + Ok(ctx.delete(activity).await?), - Some(BaseType::Object(ObjectType::Activity(ActivityType::Follow))) => { - let activity_targets = object.addressed(); - let activity_entity = model::activity::Model::new(&object)?; - let aid = activity_entity.id.clone(); - tracing::info!("{} wants to follow {}", activity_entity.actor, activity_entity.object.as_deref().unwrap_or("")); - model::activity::Entity::insert(activity_entity.into_active_model()) - .exec(ctx.db()).await?; - ctx.address_to(&aid, None, &activity_targets).await?; - Ok(()) - }, + Some(BaseType::Object(ObjectType::Activity(ActivityType::Follow))) => + Ok(ctx.follow(activity).await?), - Some(BaseType::Object(ObjectType::Activity(ActivityType::Accept(_)))) => { - // TODO what about TentativeAccept - let activity_model = model::activity::Model::new(&object)?; - let Some(follow_request_id) = activity_model.object else { - return Err(StatusCode::BAD_REQUEST.into()); - }; - let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id) - .one(ctx.db()).await? - else { - return Err(StatusCode::NOT_FOUND.into()); - }; - if follow_activity.object.unwrap_or("".into()) != activity_model.actor { - return Err(StatusCode::FORBIDDEN.into()); - } + Some(BaseType::Object(ObjectType::Activity(ActivityType::Accept(_)))) => + Ok(ctx.accept(activity).await?), - tracing::info!("{} accepted follow request by {}", activity_model.actor, follow_activity.actor); + Some(BaseType::Object(ObjectType::Activity(ActivityType::Reject(_)))) => + Ok(ctx.reject(activity).await?), - model::relation::Entity::insert( - model::relation::ActiveModel { - follower: Set(follow_activity.actor), - following: Set(activity_model.actor), - ..Default::default() - } - ).exec(ctx.db()).await?; + Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) => + Ok(ctx.like(activity).await?), - ctx.address_to(&activity_model.id, None, &object.addressed()).await?; - Ok(()) - }, + Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) => + Ok(ctx.create(activity).await?), - Some(BaseType::Object(ObjectType::Activity(ActivityType::Reject(_)))) => { - // TODO what about TentativeReject? - let activity_model = model::activity::Model::new(&object)?; - let Some(follow_request_id) = activity_model.object else { - return Err(StatusCode::BAD_REQUEST.into()); - }; - let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id) - .one(ctx.db()).await? - else { - return Err(StatusCode::NOT_FOUND.into()); - }; - if follow_activity.object.unwrap_or("".into()) != activity_model.actor { - return Err(StatusCode::FORBIDDEN.into()); - } - tracing::info!("{} rejected follow request by {}", activity_model.actor, follow_activity.actor); - ctx.address_to(&activity_model.id, None, &object.addressed()).await?; - Ok(()) - }, - - Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) => { - let aid = object.actor().id().ok_or(StatusCode::BAD_REQUEST)?; - let oid = object.object().id().ok_or(StatusCode::BAD_REQUEST)?; - let like = model::like::ActiveModel { - id: sea_orm::ActiveValue::NotSet, - actor: sea_orm::Set(aid.clone()), - likes: sea_orm::Set(oid.clone()), - date: sea_orm::Set(chrono::Utc::now()), - }; - match model::like::Entity::insert(like).exec(ctx.db()).await { - Err(sea_orm::DbErr::RecordNotInserted) => Err(StatusCode::NOT_MODIFIED.into()), - Err(sea_orm::DbErr::Exec(_)) => Err(StatusCode::NOT_MODIFIED.into()), // bad fix for sqlite - Err(e) => { - tracing::error!("unexpected error procesing like from {aid} to {oid}: {e}"); - Err(StatusCode::INTERNAL_SERVER_ERROR.into()) - } - Ok(_) => { - model::object::Entity::update_many() - .col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1)) - .filter(model::object::Column::Id.eq(oid.clone())) - .exec(ctx.db()) - .await?; - tracing::info!("{} liked {}", aid, oid); - Ok(()) - }, - } - }, - - Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) => { - let activity_model = model::activity::Model::new(&object)?; - let activity_targets = object.addressed(); - let Some(object_node) = object.object().extract() else { - // TODO we could process non-embedded activities or arrays but im lazy rn - tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&object).unwrap()); - return Err(StatusCode::UNPROCESSABLE_ENTITY.into()); - }; - let object_model = model::object::Model::new(&object_node)?; - let aid = activity_model.id.clone(); - let oid = object_model.id.clone(); - model::object::Entity::insert(object_model.into_active_model()).exec(ctx.db()).await?; - model::activity::Entity::insert(activity_model.into_active_model()).exec(ctx.db()).await?; - ctx.address_to(&aid, Some(&oid), &activity_targets).await?; - tracing::info!("{} posted {}", aid, oid); - Ok(()) - }, - - Some(BaseType::Object(ObjectType::Activity(ActivityType::Update))) => { - let activity_model = model::activity::Model::new(&object)?; - let activity_targets = object.addressed(); - let Some(object_node) = object.object().extract() else { - // TODO we could process non-embedded activities or arrays but im lazy rn - tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&object).unwrap()); - return Err(StatusCode::UNPROCESSABLE_ENTITY.into()); - }; - let aid = activity_model.id.clone(); - let Some(oid) = object_node.id().map(|x| x.to_string()) else { - return Err(StatusCode::BAD_REQUEST.into()); - }; - model::activity::Entity::insert(activity_model.into_active_model()).exec(ctx.db()).await?; - match object_node.object_type() { - Some(ObjectType::Actor(_)) => { - // TODO oof here is an example of the weakness of this model, we have to go all the way - // back up to serde_json::Value because impl Object != impl Actor - let actor_model = model::user::Model::new(&object_node)?; - model::user::Entity::update(actor_model.into_active_model()) - .exec(ctx.db()).await?; - }, - Some(ObjectType::Note) => { - let object_model = model::object::Model::new(&object_node)?; - model::object::Entity::update(object_model.into_active_model()) - .exec(ctx.db()).await?; - }, - Some(t) => tracing::warn!("no side effects implemented for update type {t:?}"), - None => tracing::warn!("empty type on embedded updated object"), - } - ctx.address_to(&aid, Some(&oid), &activity_targets).await?; - tracing::info!("{} updated {}", aid, oid); - Ok(()) - }, + Some(BaseType::Object(ObjectType::Activity(ActivityType::Update))) => + Ok(ctx.update(activity).await?), Some(BaseType::Object(ObjectType::Activity(_x))) => { - tracing::info!("received unimplemented activity on inbox: {}", serde_json::to_string_pretty(&object).unwrap()); + tracing::info!("received unimplemented activity on inbox: {}", serde_json::to_string_pretty(&activity).unwrap()); Err(StatusCode::NOT_IMPLEMENTED.into()) }, Some(_x) => { - tracing::warn!("ignoring non-activity object in inbox: {}", serde_json::to_string_pretty(&object).unwrap()); + tracing::warn!("ignoring non-activity object in inbox: {}", serde_json::to_string_pretty(&activity).unwrap()); Err(StatusCode::UNPROCESSABLE_ENTITY.into()) } } diff --git a/src/activitypub/user/outbox.rs b/src/activitypub/user/outbox.rs index 2a64ed60..81767ad1 100644 --- a/src/activitypub/user/outbox.rs +++ b/src/activitypub/user/outbox.rs @@ -79,10 +79,10 @@ pub async fn post( Some(BaseType::Link(_)) => Err(StatusCode::UNPROCESSABLE_ENTITY.into()), Some(BaseType::Object(ObjectType::Note)) => - Ok(CreationResult(ctx.post_note(uid, activity).await?)), + Ok(CreationResult(ctx.create_note(uid, activity).await?)), Some(BaseType::Object(ObjectType::Activity(ActivityType::Create))) => - Ok(CreationResult(ctx.post_activity(uid, activity).await?)), + Ok(CreationResult(ctx.create(uid, activity).await?)), Some(BaseType::Object(ObjectType::Activity(ActivityType::Like))) => Ok(CreationResult(ctx.like(uid, activity).await?)), diff --git a/src/errors.rs b/src/errors.rs index ef561eab..840a5881 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -1,5 +1,3 @@ -use reqwest::StatusCode; - #[derive(Debug, thiserror::Error)] pub enum UpubError { #[error("database error: {0}")] @@ -20,7 +18,11 @@ pub enum UpubError { impl UpubError { pub fn bad_request() -> Self { - Self::Status(StatusCode::BAD_REQUEST) + Self::Status(axum::http::StatusCode::BAD_REQUEST) + } + + pub fn unprocessable() -> Self { + Self::Status(axum::http::StatusCode::UNPROCESSABLE_ENTITY) } } diff --git a/src/server.rs b/src/server.rs index 32dd71a7..23debd0b 100644 --- a/src/server.rs +++ b/src/server.rs @@ -2,10 +2,10 @@ use std::{str::Utf8Error, sync::Arc}; use openssl::rsa::Rsa; use reqwest::StatusCode; -use sea_orm::{ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, IntoActiveModel, QueryFilter, QuerySelect, SelectColumns, Set}; +use sea_orm::{sea_query::Expr, ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, IntoActiveModel, QueryFilter, QuerySelect, SelectColumns, Set}; -use crate::{activitypub::{jsonld::LD, APOutbox, Addressed, CreationResult, PUBLIC_TARGET}, dispatcher::Dispatcher, errors::UpubError, fetcher::Fetcher, model}; -use apb::{Activity, ActivityMut, BaseMut, CollectionMut, CollectionPageMut, CollectionType, Node, ObjectMut}; +use crate::{activitypub::{jsonld::LD, APInbox, APOutbox, Addressed, PUBLIC_TARGET}, dispatcher::Dispatcher, errors::{LoggableError, UpubError}, fetcher::Fetcher, model}; +use apb::{Activity, ActivityMut, Base, BaseMut, CollectionMut, CollectionPageMut, CollectionType, Node, Object, ObjectMut}; #[derive(Clone)] pub struct Context(Arc); @@ -240,7 +240,7 @@ impl Context { #[axum::async_trait] impl APOutbox for Context { - async fn post_note(&self, uid: String, object: serde_json::Value) -> crate::Result { + async fn create_note(&self, uid: String, object: serde_json::Value) -> crate::Result { let oid = self.oid(uuid::Uuid::new_v4().to_string()); let aid = self.aid(uuid::Uuid::new_v4().to_string()); let activity_targets = object.addressed(); @@ -273,7 +273,7 @@ impl APOutbox for Context { Ok(aid) } - async fn post_activity(&self, uid: String, activity: serde_json::Value) -> crate::Result { + async fn create(&self, uid: String, activity: serde_json::Value) -> crate::Result { let Some(object) = activity.object().extract() else { return Err(UpubError::bad_request()); }; @@ -450,3 +450,159 @@ impl APOutbox for Context { Ok(aid) } } + +#[axum::async_trait] +impl APInbox for Context { + async fn create(&self, activity: serde_json::Value) -> crate::Result<()> { + let activity_model = model::activity::Model::new(&activity)?; + let activity_targets = activity.addressed(); + let Some(object_node) = activity.object().extract() else { + // TODO we could process non-embedded activities or arrays but im lazy rn + tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&activity).unwrap()); + return Err(StatusCode::UNPROCESSABLE_ENTITY.into()); + }; + let object_model = model::object::Model::new(&object_node)?; + let aid = activity_model.id.clone(); + let oid = object_model.id.clone(); + model::object::Entity::insert(object_model.into_active_model()).exec(self.db()).await?; + model::activity::Entity::insert(activity_model.into_active_model()).exec(self.db()).await?; + self.address_to(&aid, Some(&oid), &activity_targets).await?; + tracing::info!("{} posted {}", aid, oid); + Ok(()) + } + + async fn like(&self, activity: serde_json::Value) -> crate::Result<()> { + let aid = activity.actor().id().ok_or(StatusCode::BAD_REQUEST)?; + let oid = activity.object().id().ok_or(StatusCode::BAD_REQUEST)?; + let like = model::like::ActiveModel { + id: sea_orm::ActiveValue::NotSet, + actor: sea_orm::Set(aid.clone()), + likes: sea_orm::Set(oid.clone()), + date: sea_orm::Set(chrono::Utc::now()), + }; + match model::like::Entity::insert(like).exec(self.db()).await { + Err(sea_orm::DbErr::RecordNotInserted) => Err(StatusCode::NOT_MODIFIED.into()), + Err(sea_orm::DbErr::Exec(_)) => Err(StatusCode::NOT_MODIFIED.into()), // bad fix for sqlite + Err(e) => { + tracing::error!("unexpected error procesing like from {aid} to {oid}: {e}"); + Err(StatusCode::INTERNAL_SERVER_ERROR.into()) + } + Ok(_) => { + model::object::Entity::update_many() + .col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1)) + .filter(model::object::Column::Id.eq(oid.clone())) + .exec(self.db()) + .await?; + tracing::info!("{} liked {}", aid, oid); + Ok(()) + }, + } + } + + async fn follow(&self, activity: serde_json::Value) -> crate::Result<()> { + let activity_targets = activity.addressed(); + let activity_model = model::activity::Model::new(&activity)?; + let aid = activity_model.id.clone(); + tracing::info!("{} wants to follow {}", activity_model.actor, activity_model.object.as_deref().unwrap_or("")); + model::activity::Entity::insert(activity_model.into_active_model()) + .exec(self.db()).await?; + self.address_to(&aid, None, &activity_targets).await?; + Ok(()) + } + + async fn accept(&self, activity: serde_json::Value) -> crate::Result<()> { + // TODO what about TentativeAccept + let activity_model = model::activity::Model::new(&activity)?; + let Some(follow_request_id) = activity_model.object else { + return Err(StatusCode::BAD_REQUEST.into()); + }; + let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id) + .one(self.db()).await? + else { + return Err(StatusCode::NOT_FOUND.into()); + }; + if follow_activity.object.unwrap_or("".into()) != activity_model.actor { + return Err(StatusCode::FORBIDDEN.into()); + } + + tracing::info!("{} accepted follow request by {}", activity_model.actor, follow_activity.actor); + + model::relation::Entity::insert( + model::relation::ActiveModel { + follower: Set(follow_activity.actor), + following: Set(activity_model.actor), + ..Default::default() + } + ).exec(self.db()).await?; + + self.address_to(&activity_model.id, None, &activity.addressed()).await?; + Ok(()) + } + + async fn reject(&self, activity: serde_json::Value) -> crate::Result<()> { + // TODO what about TentativeReject? + let activity_model = model::activity::Model::new(&activity)?; + let Some(follow_request_id) = activity_model.object else { + return Err(StatusCode::BAD_REQUEST.into()); + }; + let Some(follow_activity) = model::activity::Entity::find_by_id(follow_request_id) + .one(self.db()).await? + else { + return Err(StatusCode::NOT_FOUND.into()); + }; + if follow_activity.object.unwrap_or("".into()) != activity_model.actor { + return Err(StatusCode::FORBIDDEN.into()); + } + tracing::info!("{} rejected follow request by {}", activity_model.actor, follow_activity.actor); + self.address_to(&activity_model.id, None, &activity.addressed()).await?; + Ok(()) + } + + async fn delete(&self, activity: serde_json::Value) -> crate::Result<()> { + // TODO verify the signature before just deleting lmao + let oid = activity.object().id().ok_or(StatusCode::BAD_REQUEST)?; + // TODO maybe we should keep the tombstone? + model::user::Entity::delete_by_id(&oid).exec(self.db()).await.info_failed("failed deleting from users"); + model::activity::Entity::delete_by_id(&oid).exec(self.db()).await.info_failed("failed deleting from activities"); + model::object::Entity::delete_by_id(&oid).exec(self.db()).await.info_failed("failed deleting from objects"); + Ok(()) + } + + async fn update(&self, activity: serde_json::Value) -> crate::Result<()> { + let activity_model = model::activity::Model::new(&activity)?; + let activity_targets = activity.addressed(); + let Some(object_node) = activity.object().extract() else { + // TODO we could process non-embedded activities or arrays but im lazy rn + tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&activity).unwrap()); + return Err(UpubError::unprocessable()); + }; + let aid = activity_model.id.clone(); + let Some(oid) = object_node.id().map(|x| x.to_string()) else { + return Err(UpubError::bad_request()); + }; + model::activity::Entity::insert(activity_model.into_active_model()).exec(self.db()).await?; + match object_node.object_type() { + Some(apb::ObjectType::Actor(_)) => { + // TODO oof here is an example of the weakness of this model, we have to go all the way + // back up to serde_json::Value because impl Object != impl Actor + let actor_model = model::user::Model::new(&object_node)?; + model::user::Entity::update(actor_model.into_active_model()) + .exec(self.db()).await?; + }, + Some(apb::ObjectType::Note) => { + let object_model = model::object::Model::new(&object_node)?; + model::object::Entity::update(object_model.into_active_model()) + .exec(self.db()).await?; + }, + Some(t) => tracing::warn!("no side effects implemented for update type {t:?}"), + None => tracing::warn!("empty type on embedded updated object"), + } + self.address_to(&aid, Some(&oid), &activity_targets).await?; + tracing::info!("{} updated {}", aid, oid); + Ok(()) + } + + async fn undo(&self, _activity: serde_json::Value) -> crate::Result<()> { + todo!() + } +}