diff --git a/Cargo.lock b/Cargo.lock index bf4db3ba..cd825528 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6511,6 +6511,7 @@ dependencies = [ "revolt_rocket_okapi", "rocket", "schemars", + "sentry", "serde", "serde_json", "utoipa", diff --git a/crates/bonfire/src/config.rs b/crates/bonfire/src/config.rs index 5642d3ba..f7d897cf 100644 --- a/crates/bonfire/src/config.rs +++ b/crates/bonfire/src/config.rs @@ -1,7 +1,7 @@ use async_tungstenite::tungstenite::{handshake, Message}; use futures::channel::oneshot::Sender; use revolt_database::events::client::ReadyPayloadFields; -use revolt_result::{create_error, Result}; +use revolt_result::{create_error, Result, ToRevoltError}; use serde::{Deserialize, Serialize}; /// Enumeration of supported protocol formats @@ -38,14 +38,14 @@ impl ProtocolConfiguration { match self.format { ProtocolFormat::Json => { if let Message::Text(text) = msg { - serde_json::from_str(text).map_err(|_| create_error!(InternalError)) + serde_json::from_str(text).to_internal_error() } else { Err(create_error!(InternalError)) } } ProtocolFormat::Msgpack => { if let Message::Binary(buf) = msg { - rmp_serde::from_slice(buf).map_err(|_| create_error!(InternalError)) + rmp_serde::from_slice(buf).to_internal_error() } else { Err(create_error!(InternalError)) } diff --git a/crates/core/database/src/models/users/model.rs b/crates/core/database/src/models/users/model.rs index 0c2b6cee..19f6d74b 100644 --- a/crates/core/database/src/models/users/model.rs +++ b/crates/core/database/src/models/users/model.rs @@ -705,6 +705,8 @@ impl User { duration_days: Option, reason: Option>, ) -> Result<()> { + // TODO: authifier Error should implement Error + let authifier = db.clone().to_authifier().await; let mut account = authifier .database diff --git a/crates/core/files/Cargo.toml b/crates/core/files/Cargo.toml index 7cc219f1..7cd15b10 100644 --- a/crates/core/files/Cargo.toml +++ b/crates/core/files/Cargo.toml @@ -14,7 +14,7 @@ imagesize = "0.13.0" tempfile = "3.12.0" base64 = "0.22.1" -aes-gcm = "0.10.3" +aes-gcm = { version = "0.10.3", features = ["std"] } typenum = "1.17.0" aws-config = "1.5.5" diff --git a/crates/core/files/src/lib.rs b/crates/core/files/src/lib.rs index dacc988e..05bb2ac2 100644 --- a/crates/core/files/src/lib.rs +++ b/crates/core/files/src/lib.rs @@ -6,7 +6,7 @@ use aes_gcm::{ }; use image::{DynamicImage, ImageBuffer}; use revolt_config::{config, report_internal_error, FilesS3}; -use revolt_result::{create_error, Result}; +use revolt_result::{create_error, Result, ToRevoltError}; use aws_sdk_s3::{ config::{Credentials, Region}, @@ -78,7 +78,7 @@ pub async fn fetch_from_s3(bucket_id: &str, path: &str, nonce: &str) -> Result Result axum::response::Response { - let status = match self.error_type { +impl Error { + pub fn axum_status(&self) -> StatusCode { + match self.error_type { ErrorType::LabelMe => StatusCode::INTERNAL_SERVER_ERROR, ErrorType::AlreadyOnboarded => StatusCode::FORBIDDEN, @@ -74,7 +73,10 @@ impl IntoResponse for Error { ErrorType::VosoUnavailable => StatusCode::BAD_REQUEST, ErrorType::NotFound => StatusCode::NOT_FOUND, ErrorType::NoEffect => StatusCode::OK, - ErrorType::FailedValidation { .. } => StatusCode::BAD_REQUEST, + ErrorType::IOError => StatusCode::BAD_REQUEST, + ErrorType::UnprocessableEntity => StatusCode::UNPROCESSABLE_ENTITY, + ErrorType::DeserializationError { .. } => StatusCode::UNPROCESSABLE_ENTITY, + ErrorType::FailedValidation { .. } => StatusCode::UNPROCESSABLE_ENTITY, ErrorType::InvalidFlagValue => StatusCode::BAD_REQUEST, ErrorType::FeatureDisabled { .. } => StatusCode::BAD_REQUEST, @@ -84,8 +86,13 @@ impl IntoResponse for Error { ErrorType::FileTypeNotAllowed => StatusCode::BAD_REQUEST, ErrorType::ImageProcessingFailed => StatusCode::INTERNAL_SERVER_ERROR, ErrorType::NoEmbedData => StatusCode::BAD_REQUEST, - }; - - (status, Json(&self)).into_response() + } + } +} + +/// HTTP response builder for Error enum +impl IntoResponse for Error { + fn into_response(self) -> axum::response::Response { + (self.axum_status(), Json(&self)).into_response() } } diff --git a/crates/core/result/src/lib.rs b/crates/core/result/src/lib.rs index 10b56d5d..729da276 100644 --- a/crates/core/result/src/lib.rs +++ b/crates/core/result/src/lib.rs @@ -1,3 +1,4 @@ +use std::panic::Location; use std::fmt::Display; #[cfg(feature = "serde")] @@ -154,6 +155,11 @@ pub enum ErrorType { DuplicateNonce, NotFound, NoEffect, + IOError, + UnprocessableEntity, + DeserializationError { + error: String, + }, FailedValidation { error: String, }, @@ -197,6 +203,53 @@ macro_rules! create_database_error { }; } +pub trait ToRevoltError: Sized { + fn capture_error(self) -> Self; + + #[track_caller] + fn to_internal_error(self) -> Result; +} + +impl ToRevoltError for Result { + fn capture_error(self) -> Self { + self.inspect_err(|e| { sentry::capture_error(e); }) + } + + #[track_caller] + fn to_internal_error(self) -> Result { + let loc = Location::caller(); + + self + .capture_error() + .map_err(|_| { + Error { + error_type: ErrorType::InternalError, + location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column()) + } + }) + } +} + +impl ToRevoltError for Option { + fn capture_error(self) -> Self { + self.inspect(|e| { sentry::capture_error(e); }) + } + + #[track_caller] + fn to_internal_error(self) -> Result { + let loc = Location::caller(); + + self + .capture_error() + .ok_or_else(|| { + Error { + error_type: ErrorType::InternalError, + location: format!("{}:{}:{}", loc.file(), loc.line(), loc.column()) + } + }) + } +} + #[cfg(test)] mod tests { use crate::ErrorType; diff --git a/crates/core/result/src/rocket.rs b/crates/core/result/src/rocket.rs index 1d996a17..bc127c9a 100644 --- a/crates/core/result/src/rocket.rs +++ b/crates/core/result/src/rocket.rs @@ -8,10 +8,9 @@ use rocket::{ use crate::{Error, ErrorType}; -/// HTTP response builder for Error enum -impl<'r> Responder<'r, 'static> for Error { - fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> { - let status = match self.error_type { +impl Error { + pub fn rocket_status(&self) -> Status { + match self.error_type { ErrorType::LabelMe => Status::InternalServerError, ErrorType::AlreadyOnboarded => Status::Forbidden, @@ -81,7 +80,10 @@ impl<'r> Responder<'r, 'static> for Error { ErrorType::VosoUnavailable => Status::BadRequest, ErrorType::NotFound => Status::NotFound, ErrorType::NoEffect => Status::Ok, - ErrorType::FailedValidation { .. } => Status::BadRequest, + ErrorType::IOError => Status::BadRequest, + ErrorType::UnprocessableEntity => Status::UnprocessableEntity, + ErrorType::DeserializationError { .. } => Status::UnprocessableEntity, + ErrorType::FailedValidation { .. } => Status::UnprocessableEntity, ErrorType::FeatureDisabled { .. } => Status::BadRequest, ErrorType::ProxyError => Status::BadRequest, @@ -90,8 +92,13 @@ impl<'r> Responder<'r, 'static> for Error { ErrorType::FileTypeNotAllowed => Status::BadRequest, ErrorType::ImageProcessingFailed => Status::InternalServerError, ErrorType::NoEmbedData => Status::BadRequest, - }; + } + } +} +/// HTTP response builder for Error enum +impl<'r> Responder<'r, 'static> for Error { + fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> { // Serialize the error data structure into JSON. let string = serde_json::to_string(&self).unwrap(); @@ -99,7 +106,7 @@ impl<'r> Responder<'r, 'static> for Error { Response::build() .sized_body(string.len(), Cursor::new(string)) .header(ContentType::new("application", "json")) - .status(status) + .status(self.rocket_status()) .ok() } } diff --git a/crates/daemons/crond/src/main.rs b/crates/daemons/crond/src/main.rs index 40af6a2b..ae8066a4 100644 --- a/crates/daemons/crond/src/main.rs +++ b/crates/daemons/crond/src/main.rs @@ -1,19 +1,32 @@ -use revolt_config::configure; -use revolt_database::DatabaseInfo; +use std::{future::Future, time::Duration}; + +use revolt_config::{configure, capture_error}; +use revolt_database::{Database, DatabaseInfo}; use revolt_result::Result; use tasks::{file_deletion, prune_dangling_files}; -use tokio::try_join; +use tokio::{join, time::sleep}; pub mod tasks; +pub async fn cron_task_wrapper>>(func: fn(Database) -> Fut, db: Database) { + loop { + if let Err(error) = func(db.clone()).await { + log::error!("cron task failed unexpectidly: {error:?}\nRetrying after 60s"); + capture_error(&error); + } + + sleep(Duration::from_secs(60)).await; + } +} + #[tokio::main] -async fn main() -> Result<()> { +async fn main() { configure!(crond); let db = DatabaseInfo::Auto.connect().await.expect("database"); - try_join!( - file_deletion::task(db.clone()), - prune_dangling_files::task(db) - ) - .map(|_| ()) + + join!( + cron_task_wrapper(file_deletion::task, db.clone()), + cron_task_wrapper(prune_dangling_files::task, db.clone()), + ); } diff --git a/crates/daemons/pushd/src/consumers/inbound/ack.rs b/crates/daemons/pushd/src/consumers/inbound/ack.rs index 77dd5d44..7fd7ef8e 100644 --- a/crates/daemons/pushd/src/consumers/inbound/ack.rs +++ b/crates/daemons/pushd/src/consumers/inbound/ack.rs @@ -123,24 +123,26 @@ impl AsyncConsumer for AckConsumer { token: session.subscription.as_ref().unwrap().auth.clone(), extras: Default::default(), }; - let raw_service_payload = serde_json::to_string(&service_payload); - if let Ok(p) = raw_service_payload { - let args = BasicPublishArguments::new( - config.pushd.exchange.as_str(), - config.pushd.apn.queue.as_str(), - ) - .finish(); + match serde_json::to_string(&service_payload) { + Ok(p) => { + let args = BasicPublishArguments::new( + config.pushd.exchange.as_str(), + config.pushd.apn.queue.as_str(), + ) + .finish(); - log::debug!( - "Publishing ack to apn session {}", - session.subscription.as_ref().unwrap().auth - ); + log::debug!( + "Publishing ack to apn session {}", + session.subscription.as_ref().unwrap().auth + ); - publish_message(self, p.into(), args).await; - } else { - log::warn!("Failed to serialize ack badge update payload!"); - revolt_config::capture_error(&raw_service_payload.unwrap_err()); + publish_message(self, p.into(), args).await; + }, + Err(e) => { + log::warn!("Failed to serialize ack badge update payload!"); + revolt_config::capture_error(&e); + } } } } diff --git a/crates/delta/src/main.rs b/crates/delta/src/main.rs index 9cce39ce..2ddddaa8 100644 --- a/crates/delta/src/main.rs +++ b/crates/delta/src/main.rs @@ -135,6 +135,7 @@ pub async fn web() -> Rocket { .manage(cors.clone()) .attach(util::ratelimiter::RatelimitFairing) .attach(cors) + .register("/", util::catchers::all_catchers()) .configure(rocket::Config { limits: rocket::data::Limits::default().limit("string", 5.megabytes()), address: Ipv4Addr::new(0, 0, 0, 0).into(), diff --git a/crates/delta/src/routes/channels/message_send.rs b/crates/delta/src/routes/channels/message_send.rs index cb05f94b..e20438a9 100644 --- a/crates/delta/src/routes/channels/message_send.rs +++ b/crates/delta/src/routes/channels/message_send.rs @@ -8,9 +8,9 @@ use revolt_models::v0; use revolt_permissions::PermissionQuery; use revolt_permissions::{calculate_channel_permissions, ChannelPermission}; use revolt_result::{create_error, Result}; -use rocket::serde::json::Json; +use crate::util::json::{Json, Validate}; use rocket::State; -use validator::Validate; +// use validator::Validate; /// # Send Message /// @@ -22,15 +22,10 @@ pub async fn message_send( amqp: &State, user: User, target: Reference<'_>, - data: Json, + data: Validate>, idempotency: IdempotencyKey, ) -> Result> { let data = data.into_inner(); - data.validate().map_err(|error| { - create_error!(FailedValidation { - error: error.to_string() - }) - })?; // Ensure we have permissions to send a message let channel = target.as_channel(db).await?; diff --git a/crates/delta/src/routes/webhooks/webhook_execute_github.rs b/crates/delta/src/routes/webhooks/webhook_execute_github.rs index f84ecc63..7f2f475a 100644 --- a/crates/delta/src/routes/webhooks/webhook_execute_github.rs +++ b/crates/delta/src/routes/webhooks/webhook_execute_github.rs @@ -701,7 +701,7 @@ fn safe_from_str Deserialize<'de>>(data: &str) -> Result { match serde_json::from_str(data) { Ok(output) => Ok(output), Err(err) => { - log::error!("{err:?}"); + revolt_config::capture_error(&err); Err(create_error!(InvalidOperation)) } } diff --git a/crates/delta/src/util/catchers.rs b/crates/delta/src/util/catchers.rs new file mode 100644 index 00000000..6738dcc7 --- /dev/null +++ b/crates/delta/src/util/catchers.rs @@ -0,0 +1,19 @@ +use rocket::{catch, Catcher, Request}; +use revolt_result::{create_error, Error, Result}; + +#[catch(404)] +pub fn not_found() -> Result<()> { + Err(create_error!(NotFound)) +} + +#[catch(422)] +pub fn unprocessable_entity(req: &Request) -> Result<()> { + match req.local_cache(|| None::) { + Some(e) => Err(e.clone()), + None => Err(create_error!(UnprocessableEntity)) + } +} + +pub fn all_catchers() -> Vec { + catchers![not_found, unprocessable_entity] +} \ No newline at end of file diff --git a/crates/delta/src/util/json.rs b/crates/delta/src/util/json.rs new file mode 100644 index 00000000..f2c5a32e --- /dev/null +++ b/crates/delta/src/util/json.rs @@ -0,0 +1,176 @@ +use std::fmt::Debug; + +use revolt_rocket_okapi::{ + r#gen::OpenApiGenerator, + response::OpenApiResponderInner, + revolt_okapi::openapi3::Responses, + util::add_schema_response, + request::OpenApiFromData, + revolt_okapi::{ + openapi3::{MediaType, RequestBody}, + Map, + }, +}; +use rocket::data::{Data, FromData, Limits, Outcome}; +use rocket::response::{self, Responder, content}; +use rocket::request::{local_cache, Request}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use revolt_result::{create_error, Error, ToRevoltError}; + +// A lot of this code is modified versions of rocket::serde::json so we can store +// the error so it can be passed to the error catcher. + +macro_rules! fn_request_body { + ($gen:ident, $ty:path, $mime_type:expr) => {{ + let schema = $gen.json_schema::<$ty>(); + Ok(RequestBody { + content: { + let mut map = Map::new(); + map.insert( + $mime_type.to_owned(), + MediaType { + schema: Some(schema), + ..MediaType::default() + }, + ); + map + }, + required: true, + ..revolt_rocket_okapi::revolt_okapi::openapi3::RequestBody::default() + }) + }}; +} + +#[derive(Debug, Clone)] +pub struct Json(pub T); + +impl<'r, T: Deserialize<'r>> Json { + #[inline] + pub fn into_inner(self) -> T { + self.0 + } + + fn from_str(s: &'r str) -> Result { + serde_json::from_str(s) + .map(Json) + .map_err(|e| create_error!(DeserializationError { error: e.to_string() })) + } + + async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Result { + let limit = req.limits().get("json").unwrap_or(Limits::JSON); + let string = match data.open(limit).into_string().await { + Ok(s) if s.is_complete() => s.into_inner(), + Ok(_) => { + return Err(create_error!(PayloadTooLarge)); + }, + Err(_) => return Err(create_error!(IOError)), + }; + + Self::from_str(local_cache!(req, string)) + } +} + +impl std::ops::Deref for Json { + type Target = T; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} + +#[async_trait] +impl<'r, T: Deserialize<'r> + std::fmt::Debug> FromData<'r> for Json { + type Error = Error; + + async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Outcome<'r, Self> { + let r = Self::from_data(req, data).await; + + match r { + Ok(value) => Outcome::Success(value), + Err(e) => { + req.local_cache(|| Some(e.clone())); + rocket::outcome::Outcome::Error((e.rocket_status(), e)) + } + } + } +} + +impl<'r, T: Serialize> Responder<'r, 'static> for Json { + fn respond_to(self, req: &'r Request<'_>) -> response::Result<'static> { + match serde_json::to_string(&self.0).capture_error() { + Ok(string) => content::RawJson(string).respond_to(req), + Err(_) => create_error!(InternalError).respond_to(req) + } + } +} + +impl<'r, T: JsonSchema + Deserialize<'r> + Debug> OpenApiFromData<'r> for Json { + fn request_body(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result { + fn_request_body!(gen, T, "application/json") + } +} + +impl OpenApiResponderInner for Json { + fn responses(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result { + let mut responses = Responses::default(); + let schema = gen.json_schema::(); + add_schema_response(&mut responses, 200, "application/json", schema)?; + Ok(responses) + } +} + +impl validator::Validate for Json { + fn validate(&self) -> Result<(), validator::ValidationErrors> { + self.0.validate() + } +} + +pub struct Validate(pub T); + +impl Validate> { + pub fn into_inner(self) -> T { + self.0.0 + } +} + +#[async_trait] +impl<'r, T: validator::Validate + Deserialize<'r> + Debug> FromData<'r> for Validate> { + type Error = Error; + + async fn from_data(req: &'r Request<'_>, data: Data<'r>) -> Outcome<'r, Self> { + as FromData<'r>>::from_data(req, data).await.and_then(|inner| { + if let Err(e) = inner.validate() { + let error = create_error!(FailedValidation { error: e.to_string() }); + req.local_cache(|| Some(error.clone())); + + return Outcome::Error((error.rocket_status(), error)) + }; + + Outcome::Success(Self(inner)) + }) + } +} + +impl<'r, T: JsonSchema + Deserialize<'r> + validator::Validate + Debug> OpenApiFromData<'r> for Validate> { + fn request_body(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result { + fn_request_body!(gen, T, "application/json") + } +} + +impl OpenApiResponderInner for Validate> { + fn responses(gen: &mut OpenApiGenerator) -> revolt_rocket_okapi::Result { + let mut responses = Responses::default(); + let schema = gen.json_schema::(); + add_schema_response(&mut responses, 200, "application/json", schema)?; + Ok(responses) + } +} + +impl std::ops::Deref for Validate { + type Target = T; + + fn deref(&self) -> &Self::Target { + &self.0 + } +} diff --git a/crates/delta/src/util/mod.rs b/crates/delta/src/util/mod.rs index 6ee3a7f6..5d2e1f76 100644 --- a/crates/delta/src/util/mod.rs +++ b/crates/delta/src/util/mod.rs @@ -1,2 +1,4 @@ pub mod ratelimiter; pub mod test; +pub mod catchers; +pub mod json; \ No newline at end of file