Compare commits
2 commits
17c1765295
...
3a6e632448
Author | SHA1 | Date | |
---|---|---|---|
3a6e632448 | |||
d8b53c7c93 |
3 changed files with 93 additions and 36 deletions
src
|
@ -73,6 +73,12 @@ pub struct SecurityConfig {
|
|||
|
||||
#[serde(default)]
|
||||
pub allow_login_refresh: bool,
|
||||
|
||||
#[serde_inline_default(2)]
|
||||
pub max_id_redirects: u32,
|
||||
|
||||
#[serde_inline_default(20)]
|
||||
pub thread_crawl_depth: u32,
|
||||
}
|
||||
|
||||
|
||||
|
|
|
@ -9,12 +9,21 @@ use crate::{errors::UpubError, model, VERSION};
|
|||
|
||||
use super::{addresser::Addresser, httpsign::HttpSignature, normalizer::Normalizer, Context};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum PullResult<T> {
|
||||
Actor(T),
|
||||
Activity(T),
|
||||
Object(T),
|
||||
}
|
||||
|
||||
impl<T> PullResult<T> {
|
||||
pub fn inner(self) -> T {
|
||||
match self {
|
||||
Self::Actor(x) | Self::Activity(x) | Self::Object(x) => x
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl PullResult<serde_json::Value> {
|
||||
pub fn actor(self) -> crate::Result<serde_json::Value> {
|
||||
match self {
|
||||
|
@ -39,11 +48,23 @@ impl PullResult<serde_json::Value> {
|
|||
Self::Object(x) => Ok(x),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn resolve(self, ctx: &(impl Fetcher + Sync)) -> crate::Result<()> {
|
||||
match self {
|
||||
Self::Actor(x) => { ctx.resolve_user(x).await?; },
|
||||
Self::Object(x) => { ctx.resolve_object(x).await?; },
|
||||
Self::Activity(x) => { ctx.resolve_activity(x).await?; },
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[axum::async_trait]
|
||||
pub trait Fetcher {
|
||||
async fn pull(&self, id: &str) -> crate::Result<PullResult<serde_json::Value>>;
|
||||
async fn pull(&self, id: &str) -> crate::Result<PullResult<serde_json::Value>> { self.pull_r(id, 0).await }
|
||||
async fn pull_r(&self, id: &str, depth: i32) -> crate::Result<PullResult<serde_json::Value>>;
|
||||
|
||||
|
||||
async fn webfinger(&self, user: &str, host: &str) -> crate::Result<String>;
|
||||
|
||||
async fn fetch_domain(&self, domain: &str) -> crate::Result<model::instance::Model>;
|
||||
|
@ -57,8 +78,8 @@ pub trait Fetcher {
|
|||
async fn fetch_object(&self, id: &str) -> crate::Result<model::object::Model> { self.fetch_object_r(id, 0).await }
|
||||
#[allow(unused)] async fn resolve_object(&self, object: serde_json::Value) -> crate::Result<model::object::Model> { self.resolve_object_r(object, 0).await }
|
||||
|
||||
async fn fetch_object_r(&self, id: &str, depth: i32) -> crate::Result<model::object::Model>;
|
||||
async fn resolve_object_r(&self, object: serde_json::Value, depth: i32) -> crate::Result<model::object::Model>;
|
||||
async fn fetch_object_r(&self, id: &str, depth: u32) -> crate::Result<model::object::Model>;
|
||||
async fn resolve_object_r(&self, object: serde_json::Value, depth: u32) -> crate::Result<model::object::Model>;
|
||||
|
||||
|
||||
async fn fetch_thread(&self, id: &str) -> crate::Result<()>;
|
||||
|
@ -122,7 +143,7 @@ pub trait Fetcher {
|
|||
|
||||
#[axum::async_trait]
|
||||
impl Fetcher for Context {
|
||||
async fn pull(&self, id: &str) -> crate::Result<PullResult<serde_json::Value>> {
|
||||
async fn pull_r(&self, id: &str, depth: u32) -> crate::Result<PullResult<serde_json::Value>> {
|
||||
let _domain = self.fetch_domain(&Context::server(id)).await?;
|
||||
|
||||
let document = Self::request(
|
||||
|
@ -133,6 +154,14 @@ impl Fetcher for Context {
|
|||
.json::<serde_json::Value>()
|
||||
.await?;
|
||||
|
||||
let doc_id = document.id().ok_or_else(|| UpubError::field("id"))?;
|
||||
if id != doc_id {
|
||||
if depth >= self.cfg().security.max_id_redirects {
|
||||
return Err(UpubError::unprocessable());
|
||||
}
|
||||
return self.pull(doc_id).await;
|
||||
}
|
||||
|
||||
match document.object_type() {
|
||||
None => Err(UpubError::bad_request()),
|
||||
Some(apb::ObjectType::Collection(_)) => Err(UpubError::unprocessable()),
|
||||
|
@ -321,7 +350,7 @@ impl Fetcher for Context {
|
|||
todo!()
|
||||
}
|
||||
|
||||
async fn fetch_object_r(&self, id: &str, depth: i32) -> crate::Result<model::object::Model> {
|
||||
async fn fetch_object_r(&self, id: &str, depth: u32) -> crate::Result<model::object::Model> {
|
||||
if let Some(x) = model::object::Entity::find_by_ap_id(id).one(self.db()).await? {
|
||||
return Ok(x); // already in db, easy
|
||||
}
|
||||
|
@ -331,7 +360,7 @@ impl Fetcher for Context {
|
|||
self.resolve_object_r(object, depth).await
|
||||
}
|
||||
|
||||
async fn resolve_object_r(&self, object: serde_json::Value, depth: i32) -> crate::Result<model::object::Model> {
|
||||
async fn resolve_object_r(&self, object: serde_json::Value, depth: u32) -> crate::Result<model::object::Model> {
|
||||
let id = object.id().ok_or_else(|| UpubError::field("id"))?.to_string();
|
||||
|
||||
if let Some(oid) = object.id() {
|
||||
|
@ -351,10 +380,10 @@ impl Fetcher for Context {
|
|||
let addressed = object.addressed();
|
||||
|
||||
if let Some(reply) = object.in_reply_to().id() {
|
||||
if depth <= 16 {
|
||||
if depth <= self.cfg().security.thread_crawl_depth {
|
||||
self.fetch_object_r(&reply, depth + 1).await?;
|
||||
} else {
|
||||
tracing::warn!("thread deeper than 16, giving up fetching more replies");
|
||||
tracing::warn!("thread deeper than {}, giving up fetching more replies", self.cfg().security.thread_crawl_depth);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -1,9 +1,10 @@
|
|||
use apb::{target::Addressed, Activity, Base, Object};
|
||||
use reqwest::StatusCode;
|
||||
use sea_orm::{sea_query::Expr, ActiveValue::{Set, NotSet}, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, SelectColumns};
|
||||
|
||||
use crate::{errors::{LoggableError, UpubError}, model, server::{addresser::Addresser, builders::AnyQuery, normalizer::Normalizer}};
|
||||
|
||||
use super::{fetcher::Fetcher, Context, side_effects::SideEffects};
|
||||
use super::{fetcher::{Fetcher, PullResult}, side_effects::SideEffects, Context};
|
||||
|
||||
|
||||
#[axum::async_trait]
|
||||
|
@ -261,12 +262,16 @@ impl apb::server::Inbox for Context {
|
|||
let actor = self.fetch_user(&uid).await?;
|
||||
let internal_uid = model::actor::Entity::ap_to_internal(&uid, self.db()).await?;
|
||||
let announced_id = activity.object().id().ok_or_else(|| UpubError::field("object"))?;
|
||||
let activity_model = self.insert_activity(activity.clone(), Some(server)).await?;
|
||||
|
||||
let announced = self.fetch_object(&announced_id).await?;
|
||||
// relays send us activities as Announce, but we don't really want to count those towards the
|
||||
match self.pull(&announced_id).await? {
|
||||
PullResult::Actor(_) => Err(UpubError::unprocessable()),
|
||||
PullResult::Object(object) => {
|
||||
let object_model = self.resolve_object(object).await?;
|
||||
let activity_model = self.insert_activity(activity.clone(), Some(server.clone())).await?;
|
||||
|
||||
// relays send us objects as Announce, but we don't really want to count those towards the
|
||||
// total shares count of an object, so just fetch the object and be done with it
|
||||
if matches!(actor.actor_type, apb::ActorType::Person) {
|
||||
if !matches!(actor.actor_type, apb::ActorType::Person) {
|
||||
tracing::info!("relay {} broadcasted {}", activity_model.actor, announced_id);
|
||||
return Ok(())
|
||||
}
|
||||
|
@ -274,7 +279,7 @@ impl apb::server::Inbox for Context {
|
|||
let share = model::announce::ActiveModel {
|
||||
internal: NotSet,
|
||||
actor: Set(internal_uid),
|
||||
object: Set(announced.internal),
|
||||
object: Set(object_model.internal),
|
||||
published: Set(activity.published().unwrap_or(chrono::Utc::now())),
|
||||
};
|
||||
|
||||
|
@ -284,11 +289,28 @@ impl apb::server::Inbox for Context {
|
|||
.exec(self.db()).await?;
|
||||
model::object::Entity::update_many()
|
||||
.col_expr(model::object::Column::Announces, Expr::col(model::object::Column::Announces).add(1))
|
||||
.filter(model::object::Column::Internal.eq(announced.internal))
|
||||
.filter(model::object::Column::Internal.eq(object_model.internal))
|
||||
.exec(self.db())
|
||||
.await?;
|
||||
|
||||
tracing::info!("{} shared {}", activity_model.actor, announced.id);
|
||||
tracing::info!("{} shared {}", activity_model.actor, announced_id);
|
||||
Ok(())
|
||||
},
|
||||
PullResult::Activity(activity) => {
|
||||
// groups update all members of other things that happen inside, process those
|
||||
let server = Context::server(activity.id().unwrap_or_default());
|
||||
match activity.activity_type().ok_or_else(UpubError::bad_request)? {
|
||||
apb::ActivityType::Like | apb::ActivityType::EmojiReact => Ok(self.like(server, activity).await?),
|
||||
apb::ActivityType::Create => Ok(self.create(server, activity).await?),
|
||||
apb::ActivityType::Undo => Ok(self.undo(server, activity).await?),
|
||||
apb::ActivityType::Delete => Ok(self.delete(server, activity).await?),
|
||||
apb::ActivityType::Update => Ok(self.update(server, activity).await?),
|
||||
x => {
|
||||
tracing::warn!("ignoring unhandled announced activity of type {x:?}");
|
||||
Err(StatusCode::NOT_IMPLEMENTED.into())
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue