diff --git a/crates/jmap-proto/src/method/get.rs b/crates/jmap-proto/src/method/get.rs index 2a65f328..5e5ca174 100644 --- a/crates/jmap-proto/src/method/get.rs +++ b/crates/jmap-proto/src/method/get.rs @@ -28,7 +28,6 @@ pub enum RequestArguments { PushSubscription, SieveScript, VacationResponse, - Principal, } #[derive(Debug, Clone, serde::Serialize)] @@ -61,7 +60,6 @@ impl JsonObjectParser for GetRequest { MethodObject::PushSubscription => RequestArguments::PushSubscription, MethodObject::SieveScript => RequestArguments::SieveScript, MethodObject::VacationResponse => RequestArguments::VacationResponse, - MethodObject::Principal => RequestArguments::Principal, _ => { return Err(Error::Method(MethodError::UnknownMethod(format!( "{}/get", @@ -125,7 +123,7 @@ impl RequestPropertyParser for RequestArguments { impl GetRequest { pub fn take_arguments(&mut self) -> RequestArguments { - std::mem::replace(&mut self.arguments, RequestArguments::Principal) + std::mem::replace(&mut self.arguments, RequestArguments::VacationResponse) } pub fn with_arguments(self, arguments: T) -> GetRequest { diff --git a/crates/jmap-proto/src/method/query.rs b/crates/jmap-proto/src/method/query.rs index 66d17646..56711f14 100644 --- a/crates/jmap-proto/src/method/query.rs +++ b/crates/jmap-proto/src/method/query.rs @@ -135,7 +135,6 @@ pub enum RequestArguments { Mailbox(mailbox::QueryArguments), EmailSubmission, SieveScript, - Principal, } impl JsonObjectParser for QueryRequest { @@ -149,7 +148,6 @@ impl JsonObjectParser for QueryRequest { MethodObject::Mailbox => RequestArguments::Mailbox(Default::default()), MethodObject::EmailSubmission => RequestArguments::EmailSubmission, MethodObject::SieveScript => RequestArguments::SieveScript, - MethodObject::Principal => RequestArguments::Principal, _ => { return Err(Error::Method(MethodError::UnknownMethod(format!( "{}/query", @@ -714,7 +712,7 @@ impl Comparator { impl QueryRequest { pub fn take_arguments(&mut self) -> RequestArguments { - std::mem::replace(&mut self.arguments, RequestArguments::Principal) + std::mem::replace(&mut self.arguments, RequestArguments::SieveScript) } } diff --git a/crates/jmap-proto/src/method/set.rs b/crates/jmap-proto/src/method/set.rs index e160f99c..c387ce2f 100644 --- a/crates/jmap-proto/src/method/set.rs +++ b/crates/jmap-proto/src/method/set.rs @@ -47,7 +47,6 @@ pub enum RequestArguments { PushSubscription, SieveScript(sieve::SetArguments), VacationResponse, - Principal, } #[derive(Debug, Clone, Default, serde::Serialize)] @@ -108,7 +107,6 @@ impl JsonObjectParser for SetRequest { MethodObject::PushSubscription => RequestArguments::PushSubscription, MethodObject::VacationResponse => RequestArguments::VacationResponse, MethodObject::SieveScript => RequestArguments::SieveScript(Default::default()), - MethodObject::Principal => RequestArguments::Principal, _ => { return Err(Error::Method(MethodError::UnknownMethod(format!( "{}/set", @@ -410,7 +408,7 @@ impl SetRequest { impl SetRequest { pub fn take_arguments(&mut self) -> RequestArguments { - std::mem::replace(&mut self.arguments, RequestArguments::Principal) + std::mem::replace(&mut self.arguments, RequestArguments::VacationResponse) } pub fn with_arguments(self, arguments: T) -> SetRequest { @@ -500,6 +498,22 @@ impl SetResponse { } } + pub fn get_object_by_id(&mut self, id: Id) -> Option<&mut Object> { + if let Some(obj) = self.updated.get_mut(&id) { + if let Some(obj) = obj { + return Some(obj); + } else { + *obj = Some(Object::with_capacity(1)); + return obj.as_mut().unwrap().into(); + } + } + + (&mut self.created) + .into_iter() + .map(|(_, obj)| obj) + .find(|obj| obj.properties.get(&Property::Id) == Some(&Value::Id(id))) + } + pub fn has_changes(&self) -> bool { !self.created.is_empty() || !self.updated.is_empty() || !self.destroyed.is_empty() } diff --git a/crates/jmap-proto/src/object/index.rs b/crates/jmap-proto/src/object/index.rs index 01d219db..e32fad07 100644 --- a/crates/jmap-proto/src/object/index.rs +++ b/crates/jmap-proto/src/object/index.rs @@ -83,6 +83,12 @@ impl ObjectIndexBuilder { .unwrap_or(&Value::Null) } + pub fn set(&mut self, property: Property, value: Value) { + if let Some(changes) = &mut self.changes { + changes.properties.set(property, value); + } + } + pub fn validate(self) -> Result { for item in self.index { if item.required || item.max_size > 0 { @@ -485,6 +491,13 @@ fn build_batch( set, }); } + (Value::Bool(boolean), IndexAs::Integer) => { + batch.ops.push(Operation::Index { + field: (&item.property).into(), + key: (*boolean as u32).serialize(), + set, + }); + } (Value::Id(id), IndexAs::Integer | IndexAs::LongInteger) => { batch.ops.push(Operation::Index { field: (&item.property).into(), diff --git a/crates/jmap-proto/src/object/mod.rs b/crates/jmap-proto/src/object/mod.rs index a5ed393b..4c8c4594 100644 --- a/crates/jmap-proto/src/object/mod.rs +++ b/crates/jmap-proto/src/object/mod.rs @@ -94,10 +94,11 @@ const BOOL_FALSE: u8 = 3; const ID: u8 = 4; const DATE: u8 = 5; const BLOB_ID: u8 = 6; -const KEYWORD: u8 = 7; -const LIST: u8 = 8; -const OBJECT: u8 = 9; -const NULL: u8 = 10; +const BLOB: u8 = 7; +const KEYWORD: u8 = 8; +const LIST: u8 = 9; +const OBJECT: u8 = 10; +const NULL: u8 = 11; impl Serialize for Value { fn serialize(self) -> Vec { @@ -198,6 +199,10 @@ impl SerializeInto for Value { buf.push(OBJECT); v.serialize_into(buf); } + Value::Blob(v) => { + buf.push(BLOB); + v.serialize_into(buf); + } Value::Null => { buf.push(NULL); } @@ -227,6 +232,7 @@ impl DeserializeFrom for Value { Some(Value::List(items)) } OBJECT => Some(Value::Object(Object::deserialize_from(bytes)?)), + BLOB => Some(Value::Blob(Vec::deserialize_from(bytes)?)), NULL => Some(Value::Null), _ => None, } diff --git a/crates/jmap-proto/src/types/blob.rs b/crates/jmap-proto/src/types/blob.rs index b876900b..68858046 100644 --- a/crates/jmap-proto/src/types/blob.rs +++ b/crates/jmap-proto/src/types/blob.rs @@ -35,7 +35,7 @@ use utils::codec::{ use crate::parser::{base32::JsonBase32Reader, json::Parser, JsonObjectParser}; -use super::date::UTCDate; +use super::{collection::Collection, date::UTCDate}; const B_LINKED: u8 = 0x10; const B_LINKED_MAILDIR: u8 = 0x20; @@ -47,7 +47,7 @@ pub struct BlobId { pub section: Option, } -#[derive(Clone, Debug, PartialEq, Eq, Hash)] +#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)] pub struct BlobSection { pub offset_start: usize, pub size: usize, @@ -65,6 +65,17 @@ impl BlobId { } } + pub fn linked(account_id: u32, collection: Collection, document_id: u32) -> Self { + Self { + kind: BlobKind::Linked { + account_id, + collection: collection.into(), + document_id, + }, + section: None, + } + } + pub fn temporary(account_id: u32) -> Self { let now_secs = now(); let now = UTCDate::from_timestamp(now_secs as i64); @@ -89,6 +100,11 @@ impl BlobId { BlobKind::Temporary { account_id, .. } => *account_id, } } + + pub fn with_section_size(mut self, size: usize) -> Self { + self.section.get_or_insert_with(Default::default).size = size; + self + } } impl JsonObjectParser for BlobId { diff --git a/crates/jmap-proto/src/types/collection.rs b/crates/jmap-proto/src/types/collection.rs index c6f1d8e8..dd30c071 100644 --- a/crates/jmap-proto/src/types/collection.rs +++ b/crates/jmap-proto/src/types/collection.rs @@ -14,7 +14,7 @@ pub enum Collection { EmailSubmission = 4, SieveScript = 5, PushSubscription = 6, - None = 8, + None = 7, } impl From for Collection { diff --git a/crates/jmap-proto/src/types/value.rs b/crates/jmap-proto/src/types/value.rs index 2f4a4765..460d60d4 100644 --- a/crates/jmap-proto/src/types/value.rs +++ b/crates/jmap-proto/src/types/value.rs @@ -30,6 +30,7 @@ pub enum Value { Keyword(Keyword), List(Vec), Object(Object), + Blob(Vec), #[default] Null, } @@ -283,6 +284,7 @@ impl Value { match self { Value::UnsignedInt(u) => Some(*u), Value::Id(id) => Some(id.id()), + Value::Bool(b) => Some(*b as u64), _ => None, } } diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index 13788486..7d458bc1 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -64,7 +64,7 @@ impl JMAP { }; } ("download", &Method::GET) => { - if let (Some(account_id), Some(blob_id), Some(name)) = ( + if let (Some(_), Some(blob_id), Some(name)) = ( path.next().and_then(|p| Id::from_bytes(p.as_bytes())), path.next().and_then(BlobId::from_base32), path.next(), @@ -85,13 +85,7 @@ impl JMAP { } .into_http_response(), Ok(None) => RequestError::not_found().into_http_response(), - Err(err) => { - tracing::error!(event = "error", - context = "blob_store", - account_id = account_id.document_id(), - blob_id = ?blob_id, - error = ?err, - "Failed to download blob"); + Err(_) => { RequestError::internal_server_error().into_http_response() } }; diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index 5d44d86c..ff2a4d47 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -129,9 +129,16 @@ impl JMAP { get::RequestArguments::PushSubscription => { self.push_subscription_get(req, acl_token).await?.into() } - get::RequestArguments::SieveScript => todo!(), - get::RequestArguments::VacationResponse => todo!(), - get::RequestArguments::Principal => todo!(), + get::RequestArguments::SieveScript => { + acl_token.assert_is_member(req.account_id)?; + + self.sieve_script_get(req, acl_token).await?.into() + } + get::RequestArguments::VacationResponse => { + acl_token.assert_is_member(req.account_id)?; + + self.vacation_response_get(req).await?.into() + } }, RequestMethod::Query(mut req) => match req.take_arguments() { query::RequestArguments::Email(arguments) => { @@ -149,8 +156,11 @@ impl JMAP { .into() } query::RequestArguments::EmailSubmission => todo!(), - query::RequestArguments::SieveScript => todo!(), - query::RequestArguments::Principal => todo!(), + query::RequestArguments::SieveScript => { + acl_token.assert_is_member(req.account_id)?; + + self.sieve_script_query(req).await?.into() + } }, RequestMethod::Set(mut req) => match req.take_arguments() { set::RequestArguments::Email => { @@ -170,9 +180,18 @@ impl JMAP { set::RequestArguments::PushSubscription => { self.push_subscription_set(req, acl_token).await?.into() } - set::RequestArguments::SieveScript(_) => todo!(), - set::RequestArguments::VacationResponse => todo!(), - set::RequestArguments::Principal => todo!(), + set::RequestArguments::SieveScript(arguments) => { + acl_token.assert_is_member(req.account_id)?; + + self.sieve_script_set(req.with_arguments(arguments), acl_token) + .await? + .into() + } + set::RequestArguments::VacationResponse => { + acl_token.assert_is_member(req.account_id)?; + + self.vacation_response_set(req, acl_token).await?.into() + } }, RequestMethod::Changes(req) => self.changes(req, acl_token).await?.into(), RequestMethod::Copy(req) => { @@ -199,7 +218,11 @@ impl JMAP { self.email_search_snippet(req, acl_token).await?.into() } - RequestMethod::ValidateScript(_) => todo!(), + RequestMethod::ValidateScript(req) => { + acl_token.assert_is_member(req.account_id)?; + + self.sieve_script_validate(req, acl_token).await?.into() + } RequestMethod::Echo(req) => req.into(), RequestMethod::Error(error) => return Err(error), }) diff --git a/crates/jmap/src/api/session.rs b/crates/jmap/src/api/session.rs index ccffd34e..9f6ab56d 100644 --- a/crates/jmap/src/api/session.rs +++ b/crates/jmap/src/api/session.rs @@ -90,6 +90,8 @@ struct WebSocketCapabilities { #[derive(Debug, Clone, serde::Serialize)] struct SieveCapabilities { + #[serde(rename(serialize = "implementation"))] + implementation: &'static str, #[serde(rename(serialize = "maxSizeScriptName"))] max_script_name: usize, #[serde(rename(serialize = "maxSizeScript"))] @@ -378,6 +380,7 @@ impl SieveCapabilities { None }, ext_lists: None, + implementation: concat!("Stalwart JMAP v", env!("CARGO_PKG_VERSION"),), } } } diff --git a/crates/jmap/src/auth/account.rs b/crates/jmap/src/auth/account.rs index c8aa1ea2..b9db690d 100644 --- a/crates/jmap/src/auth/account.rs +++ b/crates/jmap/src/auth/account.rs @@ -90,10 +90,10 @@ impl JMAP { match &self.auth_db { AuthDatabase::Sql { db, - query_gids_by_uid, + query_uids_by_address, .. } => db - .fetch_string_to_uids(query_gids_by_uid, address) + .fetch_string_to_uids(query_uids_by_address, address) .await .into_iter() .map(|id| id as u32) diff --git a/crates/jmap/src/blob/copy.rs b/crates/jmap/src/blob/copy.rs index a8c7382a..7df1a33d 100644 --- a/crates/jmap/src/blob/copy.rs +++ b/crates/jmap/src/blob/copy.rs @@ -4,9 +4,9 @@ use jmap_proto::{ set::{SetError, SetErrorType}, }, method::copy::{CopyBlobRequest, CopyBlobResponse}, - types::{acl::Acl, blob::BlobId}, + types::blob::BlobId, }; -use store::BlobKind; + use utils::map::vec_map::VecMap; use crate::{auth::AclToken, JMAP}; @@ -26,38 +26,7 @@ impl JMAP { let account_id = request.account_id.document_id(); for blob_id in request.blob_ids { - let has_access = match &blob_id.kind { - BlobKind::Linked { - account_id, - collection, - document_id, - } => { - acl_token.is_member(*account_id) - || (acl_token.has_access(*account_id, *collection) - && self - .has_access_to_document( - acl_token, - *account_id, - *collection, - *document_id, - Acl::Read, - ) - .await?) - } - BlobKind::LinkedMaildir { - account_id, - document_id, - } => { - acl_token.is_member(*account_id) - || self - .shared_messages(acl_token, *account_id, Acl::ReadItems) - .await? - .contains(*document_id) - } - BlobKind::Temporary { account_id, .. } => acl_token.is_member(*account_id), - }; - - if has_access { + if self.has_access_blob(&blob_id, acl_token).await? { let dest_blob_id = BlobId::temporary(account_id); match self .store diff --git a/crates/jmap/src/blob/download.rs b/crates/jmap/src/blob/download.rs index 548d4678..c665ca4f 100644 --- a/crates/jmap/src/blob/download.rs +++ b/crates/jmap/src/blob/download.rs @@ -17,7 +17,7 @@ impl JMAP { &self, blob_id: &BlobId, acl_token: &AclToken, - ) -> store::Result>> { + ) -> Result>, MethodError> { if !acl_token.is_member(blob_id.account_id()) { match &blob_id.kind { BlobKind::Linked { @@ -57,7 +57,6 @@ impl JMAP { if let Some(section) = &blob_id.section { Ok(self - .store .get_blob( &blob_id.kind, (section.offset_start as u32) @@ -70,7 +69,7 @@ impl JMAP { Encoding::QuotedPrintable => quoted_printable_decode(&bytes), })) } else { - self.store.get_blob(&blob_id.kind, 0..u32::MAX).await + self.get_blob(&blob_id.kind, 0..u32::MAX).await } } @@ -91,4 +90,41 @@ impl JMAP { } } } + + pub async fn has_access_blob( + &self, + blob_id: &BlobId, + acl_token: &AclToken, + ) -> Result { + Ok(match &blob_id.kind { + BlobKind::Linked { + account_id, + collection, + document_id, + } => { + acl_token.is_member(*account_id) + || (acl_token.has_access(*account_id, *collection) + && self + .has_access_to_document( + acl_token, + *account_id, + *collection, + *document_id, + Acl::Read, + ) + .await?) + } + BlobKind::LinkedMaildir { + account_id, + document_id, + } => { + acl_token.is_member(*account_id) + || self + .shared_messages(acl_token, *account_id, Acl::ReadItems) + .await? + .contains(*document_id) + } + BlobKind::Temporary { account_id, .. } => acl_token.is_member(*account_id), + }) + } } diff --git a/crates/jmap/src/blob/upload.rs b/crates/jmap/src/blob/upload.rs index 9ef74172..6773349c 100644 --- a/crates/jmap/src/blob/upload.rs +++ b/crates/jmap/src/blob/upload.rs @@ -1,7 +1,8 @@ use jmap_proto::{ - error::request::RequestError, + error::{method::MethodError, request::RequestError}, types::{blob::BlobId, id::Id}, }; +use store::BlobKind; use crate::JMAP; @@ -46,4 +47,28 @@ impl JMAP { } } } + + pub async fn put_blob(&self, kind: &BlobKind, data: &[u8]) -> Result { + self.store.put_blob(kind, data).await.map_err(|err| { + tracing::error!( + event = "error", + context = "blob_put", + kind = ?kind, + error = ?err, + "Failed to store blob."); + MethodError::ServerPartialFail + }) + } + + pub async fn delete_blob(&self, kind: &BlobKind) -> Result { + self.store.delete_blob(kind).await.map_err(|err| { + tracing::error!( + event = "error", + context = "delete_blob", + kind = ?kind, + error = ?err, + "Failed to delete blob."); + MethodError::ServerPartialFail + }) + } } diff --git a/crates/jmap/src/email/import.rs b/crates/jmap/src/email/import.rs index be84dbd7..4fcc14ca 100644 --- a/crates/jmap/src/email/import.rs +++ b/crates/jmap/src/email/import.rs @@ -84,9 +84,9 @@ impl JMAP { } // Fetch raw message to import - let raw_message = match self.blob_download(&email.blob_id, acl_token).await { - Ok(Some(raw_message)) => raw_message, - Ok(None) => { + let raw_message = match self.blob_download(&email.blob_id, acl_token).await? { + Some(raw_message) => raw_message, + None => { response.not_created.append( id, SetError::new(SetErrorType::BlobNotFound) @@ -94,15 +94,6 @@ impl JMAP { ); continue; } - Err(err) => { - tracing::error!(event = "error", - context = "store", - account_id = account_id, - blob_id = ?email.blob_id, - error = ?err, - "Failed to retrieve blob"); - return Err(MethodError::ServerPartialFail); - } }; // Import message diff --git a/crates/jmap/src/email/ingest.rs b/crates/jmap/src/email/ingest.rs index 8f939cf5..eaff4344 100644 --- a/crates/jmap/src/email/ingest.rs +++ b/crates/jmap/src/email/ingest.rs @@ -88,6 +88,7 @@ impl JMAP { // Check for duplicates if !skip_duplicates + && !references.is_empty() && !self .store .filter( diff --git a/crates/jmap/src/email/parse.rs b/crates/jmap/src/email/parse.rs index a120d8a8..a16c77e1 100644 --- a/crates/jmap/src/email/parse.rs +++ b/crates/jmap/src/email/parse.rs @@ -26,7 +26,6 @@ impl JMAP { if request.blob_ids.len() > self.config.mail_parse_max_items { return Err(MethodError::RequestTooLarge); } - let account_id = request.account_id.document_id(); let properties = request.properties.unwrap_or_else(|| { vec![ Property::BlobId, @@ -79,21 +78,12 @@ impl JMAP { for blob_id in request.blob_ids { // Fetch raw message to parse - let raw_message = match self.blob_download(&blob_id, acl_token).await { - Ok(Some(raw_message)) => raw_message, - Ok(None) => { + let raw_message = match self.blob_download(&blob_id, acl_token).await? { + Some(raw_message) => raw_message, + None => { response.not_found.push(blob_id); continue; } - Err(err) => { - tracing::error!(event = "error", - context = "store", - account_id = account_id, - blob_id = ?blob_id, - error = ?err, - "Failed to retrieve blob"); - return Err(MethodError::ServerPartialFail); - } }; let message = if let Some(message) = Message::parse(&raw_message) { message diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index 6f635f59..fa5a9307 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -521,11 +521,11 @@ impl JMAP { headers, contents: if !is_multipart { if let Some(blob_id) = blob_id { - match self.blob_download(&blob_id, acl_token).await { - Ok(Some(contents)) => { + match self.blob_download(&blob_id, acl_token).await? { + Some(contents) => { BodyPart::Binary(contents.into()) } - Ok(None) => { + None => { response.not_created.append( id, SetError::new(SetErrorType::BlobNotFound).with_description( @@ -534,15 +534,6 @@ impl JMAP { ); continue 'create; } - Err(err) => { - tracing::error!(event = "error", - context = "email_set", - account_id = account_id, - blob_id = ?blob_id, - error = ?err, - "Failed to retrieve blob while creating message"); - return Err(MethodError::ServerPartialFail); - } } } else if let Some(part_id) = part_id { if let Some(contents) = diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index 6aa6292e..2df06e6b 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -1,5 +1,6 @@ use std::{sync::Arc, time::Duration}; +use ::sieve::{Compiler, Runtime}; use api::session::BaseCapabilities; use auth::{ oauth::OAuthCode, @@ -39,18 +40,30 @@ pub mod email; pub mod mailbox; pub mod push; pub mod services; +pub mod sieve; pub mod thread; +pub mod vacation; + +pub const SUPERUSER_ID: u32 = 0; +pub const LONG_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24); pub struct JMAP { pub store: Store, pub config: Config, + pub sessions: LruCache, pub acl_tokens: LruCache>, + pub rate_limit_auth: LruCache>>, pub rate_limit_unauth: LruCache>>, + pub oauth_codes: LruCache>, pub auth_db: AuthDatabase, + pub state_tx: mpsc::Sender, + + pub sieve_compiler: Compiler, + pub sieve_runtime: Runtime, } pub struct Config { @@ -96,8 +109,9 @@ pub struct Config { pub capabilities: BaseCapabilities, } -pub const SUPERUSER_ID: u32 = 0; -pub const LONG_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24); +pub struct Bincode { + pub inner: T, +} pub enum MaybeError { Temporary, @@ -108,27 +122,18 @@ impl JMAP { pub async fn init( config: &utils::config::Config, delivery_rx: mpsc::Receiver, - ) -> Arc { - let auth_db = match config - .value_require("jmap.auth.database.type") - .failed("Invalid property") - { + ) -> Result, String> { + let auth_db = match config.value_require("jmap.auth.database.type")? { "ldap" => AuthDatabase::Ldap, "sql" => { - let address = config - .value_require("jmap.auth.database.address") - .failed("Invalid property"); + let address = config.value_require("jmap.auth.database.address")?; let max_connections = config - .property("jmap.auth.database.max-connections") - .failed("Invalid property") + .property("jmap.auth.database.max-connections")? .unwrap_or(10); let min_connections = config - .property("jmap.auth.database.min-connections") - .failed("Invalid property") + .property("jmap.auth.database.min-connections")? .unwrap_or(0); - let idle_timeout = config - .property("jmap.auth.database.idle-timeout") - .failed("Invalid property"); + let idle_timeout = config.property("jmap.auth.database.idle-timeout")?; let db = if address.starts_with("postgres:") { SqlDatabase::Postgres( @@ -173,36 +178,28 @@ impl JMAP { AuthDatabase::Sql { db, query_uid_by_login: config - .value_require("jmap.auth.database.query.uid-by-login") - .failed("Invalid property") + .value_require("jmap.auth.database.query.uid-by-login")? .to_string(), query_login_by_uid: config - .value_require("jmap.auth.database.query.login-by-uid") - .failed("Invalid property") + .value_require("jmap.auth.database.query.login-by-uid")? .to_string(), query_secret_by_uid: config - .value_require("jmap.auth.database.query.secret-by-uid") - .failed("Invalid property") + .value_require("jmap.auth.database.query.secret-by-uid")? .to_string(), query_gids_by_uid: config - .value_require("jmap.auth.database.query.gids-by-uid") - .failed("Invalid property") + .value_require("jmap.auth.database.query.gids-by-uid")? .to_string(), query_uids_by_address: config - .value_require("jmap.auth.database.query.uids-by-address") - .failed("Invalid property") + .value_require("jmap.auth.database.query.uids-by-address")? .to_string(), query_addresses_by_uid: config - .value_require("jmap.auth.database.query.addresses-by-uid") - .failed("Invalid property") + .value_require("jmap.auth.database.query.addresses-by-uid")? .to_string(), query_vrfy: config - .value_require("jmap.auth.database.query.vrfy") - .failed("Invalid property") + .value_require("jmap.auth.database.query.vrfy")? .to_string(), query_expn: config - .value_require("jmap.auth.database.query.expn") - .failed("Invalid property") + .value_require("jmap.auth.database.query.expn")? .to_string(), } } @@ -216,37 +213,161 @@ impl JMAP { store: Store::open(config).await.failed("Unable to open database"), config: Config::new(config).failed("Invalid configuration file"), sessions: LruCache::with_capacity( - config - .property("jmap.session.cache.size") - .failed("Invalid property") - .unwrap_or(100), + config.property("jmap.session.cache.size")?.unwrap_or(100), ), acl_tokens: LruCache::with_capacity( - config - .property("jmap.session.cache.size") - .failed("Invalid property") - .unwrap_or(100), + config.property("jmap.session.cache.size")?.unwrap_or(100), ), rate_limit_auth: LruCache::with_capacity( config - .property("jmap.rate-limit.account.size") - .failed("Invalid property") + .property("jmap.rate-limit.account.size")? .unwrap_or(1024), ), rate_limit_unauth: LruCache::with_capacity( config - .property("jmap.rate-limit.anonymous.size") - .failed("Invalid property") + .property("jmap.rate-limit.anonymous.size")? .unwrap_or(2048), ), oauth_codes: LruCache::with_capacity( - config - .property("oauth.code.cache-size") - .failed("Invalid property") - .unwrap_or(128), + config.property("oauth.code.cache-size")?.unwrap_or(128), ), auth_db, state_tx, + sieve_compiler: Compiler::new() + .with_max_script_size( + config + .property("jmap.sieve.limits.script-size")? + .unwrap_or(1024 * 1024), + ) + .with_max_string_size( + config + .property("jmap.sieve.limits.string-size")? + .unwrap_or(4096), + ) + .with_max_variable_name_size( + config + .property("jmap.sieve.limits.variable-name-size")? + .unwrap_or(32), + ) + .with_max_nested_blocks( + config + .property("jmap.sieve.limits.nested-blocks")? + .unwrap_or(15), + ) + .with_max_nested_tests( + config + .property("jmap.sieve.limits.nested-tests")? + .unwrap_or(15), + ) + .with_max_nested_foreverypart( + config + .property("jmap.sieve.limits.nested-foreverypart")? + .unwrap_or(3), + ) + .with_max_match_variables( + config + .property("jmap.sieve.limits.match-variables")? + .unwrap_or(30), + ) + .with_max_local_variables( + config + .property("jmap.sieve.limits.local-variables")? + .unwrap_or(128), + ) + .with_max_header_size( + config + .property("jmap.sieve.limits.header-size")? + .unwrap_or(1024), + ) + .with_max_includes(config.property("jmap.sieve.limits.includes")?.unwrap_or(3)), + sieve_runtime: Runtime::new() + .with_max_nested_includes( + config + .property("jmap.sieve.limits.nested-includes")? + .unwrap_or(3), + ) + .with_cpu_limit(config.property("jmap.sieve.cpu-limit")?.unwrap_or(5000)) + .with_max_variable_size( + config + .property("jmap.sieve.limits.variable-size")? + .unwrap_or(4096), + ) + .with_max_redirects(config.property("jmap.sieve.limits.redirects")?.unwrap_or(1)) + .with_max_received_headers( + config + .property("jmap.sieve.limits.received-headers")? + .unwrap_or(10), + ) + .with_max_header_size( + config + .property("jmap.sieve.limits.header-size")? + .unwrap_or(1024), + ) + .with_max_out_messages( + config + .property("jmap.sieve.limits.outgoing-messages")? + .unwrap_or(3), + ) + .with_default_vacation_expiry( + config + .property::("jmap.sieve.default-expiry.vacation")? + .unwrap_or(Duration::from_secs(30 * 86400)) + .as_secs(), + ) + .with_default_duplicate_expiry( + config + .property::("jmap.sieve.default-expiry.duplicate")? + .unwrap_or(Duration::from_secs(7 * 86400)) + .as_secs(), + ) + .without_capabilities( + config + .values("jmap.sieve.disable-capabilities") + .map(|(_, v)| v), + ) + .with_valid_notification_uris({ + let values = config + .values("jmap.sieve.notification-uris") + .map(|(_, v)| v.to_string()) + .collect::>(); + if !values.is_empty() { + values + } else { + vec!["mailto".to_string()] + } + }) + .with_protected_headers({ + let values = config + .values("jmap.sieve.protected-headers") + .map(|(_, v)| v.to_string()) + .collect::>(); + if !values.is_empty() { + values + } else { + vec![ + "Original-Subject".to_string(), + "Original-From".to_string(), + "Received".to_string(), + "Auto-Submitted".to_string(), + ] + } + }) + .with_vacation_default_subject( + config + .value("jmap.sieve.vacation.default-subject") + .unwrap_or("Automated reply") + .to_string(), + ) + .with_vacation_subject_prefix( + config + .value("jmap.sieve.vacation.subject-prefix") + .unwrap_or("Auto: ") + .to_string(), + ) + .with_env_variable("name", "Stalwart JMAP") + .with_env_variable("version", env!("CARGO_PKG_VERSION")) + .with_env_variable("location", "MS") + .with_env_variable("phase", "during"), }); // Spawn delivery manager @@ -255,7 +376,7 @@ impl JMAP { // Spawn state manager spawn_state_manager(jmap_server.clone(), config, state_rx); - jmap_server + Ok(jmap_server) } pub async fn assign_document_id( @@ -549,6 +670,36 @@ impl JMAP { } } +impl Bincode { + pub fn new(inner: T) -> Self { + Self { inner } + } +} + +impl Serialize for &Bincode { + fn serialize(self) -> Vec { + bincode::serialize(&self.inner).unwrap_or_default() + } +} + +impl Serialize for Bincode { + fn serialize(self) -> Vec { + bincode::serialize(&self.inner).unwrap_or_default() + } +} + +impl Deserialize + for Bincode +{ + fn deserialize(bytes: &[u8]) -> store::Result { + bincode::deserialize(bytes) + .map(|inner| Self { inner }) + .map_err(|err| { + store::Error::InternalError(format!("Bincode deserialization failed: {err}")) + }) + } +} + trait UpdateResults: Sized { fn update_results(&mut self, sorted_results: SortedResultSet) -> Result<(), MethodError>; } diff --git a/crates/jmap/src/mailbox/set.rs b/crates/jmap/src/mailbox/set.rs index 82ce6bd2..ffdcc44e 100644 --- a/crates/jmap/src/mailbox/set.rs +++ b/crates/jmap/src/mailbox/set.rs @@ -37,7 +37,7 @@ struct SetContext<'x> { account_id: u32, acl_token: &'x AclToken, is_shared: bool, - set_response: SetResponse, + response: SetResponse, mailbox_ids: RoaringBitmap, will_destroy: Vec, } @@ -74,7 +74,7 @@ impl JMAP { account_id, is_shared: acl_token.is_shared(account_id), acl_token, - set_response: self + response: self .prepare_set_response(&request, Collection::Mailbox) .await?, mailbox_ids: self.mailbox_get_or_create(account_id).await?, @@ -98,10 +98,10 @@ impl JMAP { changes.log_insert(Collection::Mailbox, document_id); ctx.mailbox_ids.insert(document_id); self.write_batch(batch).await?; - ctx.set_response.created(id, document_id); + ctx.response.created(id, document_id); } Err(err) => { - ctx.set_response.not_created.append(id, err); + ctx.response.not_created.append(id, err); continue 'create; } } @@ -111,7 +111,7 @@ impl JMAP { 'update: for (id, object) in request.unwrap_update() { // Make sure id won't be destroyed if ctx.will_destroy.contains(&id) { - ctx.set_response + ctx.response .not_updated .append(id, SetError::will_destroy()); continue 'update; @@ -132,7 +132,7 @@ impl JMAP { if ctx.is_shared { let acl = mailbox.inner.effective_acl(acl_token); if !acl.contains(Acl::Modify) { - ctx.set_response.not_updated.append( + ctx.response.not_updated.append( id, SetError::forbidden() .with_description("You are not allowed to modify this mailbox."), @@ -141,7 +141,7 @@ impl JMAP { } else if object.properties.contains_key(&Property::Acl) && !acl.contains(Acl::Administer) { - ctx.set_response.not_updated.append( + ctx.response.not_updated.append( id, SetError::forbidden().with_description( "You are not allowed to change the permissions of this mailbox.", @@ -168,7 +168,7 @@ impl JMAP { match self.store.write(batch.build()).await { Ok(_) => (), Err(store::Error::AssertValueFailed) => { - ctx.set_response.not_updated.append(id, SetError::forbidden().with_description( + ctx.response.not_updated.append(id, SetError::forbidden().with_description( "Another process modified this mailbox, please try again.", )); continue 'update; @@ -184,17 +184,15 @@ impl JMAP { } } } - ctx.set_response.updated.append(id, None); + ctx.response.updated.append(id, None); } Err(err) => { - ctx.set_response.not_updated.append(id, err); + ctx.response.not_updated.append(id, err); continue 'update; } } } else { - ctx.set_response - .not_updated - .append(id, SetError::not_found()); + ctx.response.not_updated.append(id, SetError::not_found()); } } @@ -206,7 +204,7 @@ impl JMAP { if (document_id == INBOX_ID || document_id == TRASH_ID) && !acl_token.is_member(SUPERUSER_ID) { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::forbidden() .with_description("You are not allowed to delete Inbox or Trash folders."), @@ -225,7 +223,7 @@ impl JMAP { .results .is_empty() { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::new(SetErrorType::MailboxHasChild) .with_description("Mailbox has at least one children."), @@ -295,7 +293,7 @@ impl JMAP { Id::from_parts(thread_id, message_id), ), Err(store::Error::AssertValueFailed) => { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::forbidden().with_description( concat!("Another process modified a message in this mailbox ", @@ -347,7 +345,7 @@ impl JMAP { } } } else { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::new(SetErrorType::MailboxHasEmail) .with_description("Mailbox is not empty."), @@ -371,7 +369,7 @@ impl JMAP { let acl = mailbox.inner.effective_acl(acl_token); if !acl.contains(Acl::Administer) { if !acl.contains(Acl::Delete) { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::forbidden().with_description( "You are not allowed to delete this mailbox.", @@ -379,7 +377,7 @@ impl JMAP { ); continue 'destroy; } else if on_destroy_remove_emails && !acl.contains(Acl::RemoveItems) { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::forbidden().with_description( "You are not allowed to delete emails from this mailbox.", @@ -401,10 +399,10 @@ impl JMAP { match self.store.write(batch.build()).await { Ok(_) => { changes.log_delete(Collection::Mailbox, document_id); - ctx.set_response.destroyed.push(id); + ctx.response.destroyed.push(id); } Err(store::Error::AssertValueFailed) => { - ctx.set_response.not_destroyed.append( + ctx.response.not_destroyed.append( id, SetError::forbidden().with_description(concat!( "Another process modified this mailbox ", @@ -424,9 +422,7 @@ impl JMAP { } } } else { - ctx.set_response - .not_destroyed - .append(id, SetError::not_found()); + ctx.response.not_destroyed.append(id, SetError::not_found()); } } @@ -434,7 +430,7 @@ impl JMAP { if !changes.is_empty() { let state_change = StateChange::new(account_id).with_change(TypeState::Mailbox, changes.change_id); - ctx.set_response.state_change = if did_remove_emails { + ctx.response.state_change = if did_remove_emails { state_change .with_change(TypeState::Email, changes.change_id) .with_change(TypeState::Thread, changes.change_id) @@ -442,10 +438,10 @@ impl JMAP { state_change } .into(); - ctx.set_response.new_state = self.commit_changes(account_id, changes).await?.into(); + ctx.response.new_state = self.commit_changes(account_id, changes).await?.into(); } - Ok(ctx.set_response) + Ok(ctx.response) } #[allow(clippy::blocks_in_if_conditions)] @@ -458,7 +454,7 @@ impl JMAP { // Parse properties let mut changes = Object::with_capacity(changes_.properties.len()); for (property, value) in changes_.properties { - let value = match ctx.set_response.eval_object_references(value) { + let value = match ctx.response.eval_object_references(value) { Ok(value) => value, Err(err) => { return Ok(Err(err)); diff --git a/crates/jmap/src/sieve/get.rs b/crates/jmap/src/sieve/get.rs new file mode 100644 index 00000000..8cb924c3 --- /dev/null +++ b/crates/jmap/src/sieve/get.rs @@ -0,0 +1,255 @@ +use std::sync::Arc; + +use jmap_proto::{ + error::method::MethodError, + method::get::{GetRequest, GetResponse, RequestArguments}, + object::Object, + types::{blob::BlobId, collection::Collection, property::Property, value::Value}, +}; +use sieve::Sieve; +use store::{query::Filter, BlobKind, Deserialize, Serialize}; + +use crate::{auth::AclToken, sieve::SeenIds, Bincode, JMAP}; + +use super::ActiveScript; + +impl JMAP { + pub async fn sieve_script_get( + &self, + mut request: GetRequest, + acl_token: &AclToken, + ) -> Result { + let ids = request.unwrap_ids(self.config.get_max_objects)?; + let properties = + request.unwrap_properties(&[Property::Id, Property::Name, Property::BlobId]); + let account_id = acl_token.primary_id(); + let push_ids = self + .get_document_ids(account_id, Collection::SieveScript) + .await? + .unwrap_or_default(); + let ids = if let Some(ids) = ids { + ids + } else { + push_ids + .iter() + .take(self.config.get_max_objects) + .map(Into::into) + .collect::>() + }; + let mut response = GetResponse { + account_id: Some(request.account_id), + state: self + .get_state(account_id, Collection::SieveScript) + .await? + .into(), + list: Vec::with_capacity(ids.len()), + not_found: vec![], + }; + + for id in ids { + // Obtain the sieve script object + let document_id = id.document_id(); + if !push_ids.contains(document_id) { + response.not_found.push(id); + continue; + } + let mut push = if let Some(push) = self + .get_property::>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + { + push + } else { + response.not_found.push(id); + continue; + }; + let mut result = Object::with_capacity(properties.len()); + for property in &properties { + match property { + Property::Id => { + result.append(Property::Id, Value::Id(id)); + } + Property::BlobId => { + if let Some(Value::UnsignedInt(blob_size)) = + push.properties.remove(&Property::BlobId) + { + result.append( + Property::BlobId, + BlobId::linked(account_id, Collection::SieveScript, document_id) + .with_section_size(blob_size as usize), + ); + } + } + Property::Name | Property::IsActive => { + result.append(property.clone(), push.remove(property)); + } + property => { + result.append(property.clone(), Value::Null); + } + } + } + response.list.push(result); + } + + Ok(response) + } + + pub async fn sieve_script_get_active( + &self, + account_id: u32, + ) -> Result, MethodError> { + // Find the currently active script + if let Some(document_id) = self + .filter( + account_id, + Collection::SieveScript, + vec![Filter::eq(Property::IsActive, 1u32)], + ) + .await? + .results + .min() + { + Ok(Some(ActiveScript { + document_id, + script: Arc::new(self.sieve_script_compile(account_id, document_id).await?), + seen_ids: self + .get_property::>( + account_id, + Collection::SieveScript, + document_id, + Property::EmailIds, + ) + .await? + .map(|seen_ids| seen_ids.inner) + .unwrap_or_default(), + })) + } else { + Ok(None) + } + } + + pub async fn sieve_script_get_by_name( + &self, + account_id: u32, + name: &str, + ) -> Result, MethodError> { + // Find the script by name + if let Some(document_id) = self + .filter( + account_id, + Collection::SieveScript, + vec![Filter::eq(Property::Name, name)], + ) + .await? + .results + .min() + { + self.sieve_script_compile(account_id, document_id) + .await + .map(Some) + } else { + Ok(None) + } + } + + async fn sieve_script_compile( + &self, + account_id: u32, + document_id: u32, + ) -> Result { + // Obtain the sieve script length + let script_offset = self + .get_property::>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + .and_then(|mut object| object.properties.remove(&Property::BlobId)) + .and_then(|value| value.as_uint()) + .ok_or_else(|| { + tracing::warn!( + context = "sieve_script_compile", + event = "error", + account_id = account_id, + document_id = document_id, + "Failed to obtain sieve script offset" + ); + + MethodError::ServerPartialFail + })? as usize; + + // Obtain the sieve script blob + let script_bytes = self + .get_blob( + &BlobKind::Linked { + account_id, + collection: Collection::SieveScript.into(), + document_id, + }, + 0..u32::MAX, + ) + .await? + .ok_or(MethodError::ServerPartialFail)?; + + // Obtain the precompiled script + if let Some(sieve) = script_bytes + .get(script_offset..) + .and_then(|bytes| Bincode::::deserialize(bytes).ok()) + { + Ok(sieve.inner) + } else { + // Deserialization failed, probably because the script compiler version changed + match self + .sieve_compiler + .compile(script_bytes.get(0..script_offset).ok_or_else(|| { + tracing::warn!( + context = "sieve_script_compile", + event = "error", + account_id = account_id, + document_id = document_id, + "Invalid sieve script offset" + ); + + MethodError::ServerPartialFail + })?) { + Ok(sieve) => { + // Store updated compiled sieve script + let sieve = Bincode::new(sieve); + let compiled_bytes = (&sieve).serialize(); + let mut updated_sieve_bytes = + Vec::with_capacity(script_offset + compiled_bytes.len()); + updated_sieve_bytes.extend_from_slice(&script_bytes[0..script_offset]); + updated_sieve_bytes.extend_from_slice(&compiled_bytes); + let _ = self + .put_blob( + &BlobKind::Linked { + account_id, + collection: Collection::SieveScript.into(), + document_id, + }, + &updated_sieve_bytes, + ) + .await; + + Ok(sieve.inner) + } + Err(error) => { + tracing::warn!( + context = "sieve_script_compile", + event = "error", + account_id = account_id, + document_id = document_id, + reason = %error, + "Failed to compile sieve script"); + Err(MethodError::ServerPartialFail) + } + } + } + } +} diff --git a/crates/jmap/src/sieve/mod.rs b/crates/jmap/src/sieve/mod.rs new file mode 100644 index 00000000..d729675e --- /dev/null +++ b/crates/jmap/src/sieve/mod.rs @@ -0,0 +1,132 @@ +use std::sync::Arc; + +use serde::ser::SerializeSeq; +use sieve::Sieve; +use store::{ahash::AHashSet, blake3, write::now}; + +pub mod get; +pub mod query; +pub mod set; +pub mod validate; + +pub struct ActiveScript { + pub document_id: u32, + pub script: Arc, + pub seen_ids: SeenIds, +} + +#[derive(Debug, Clone)] +pub struct SeenIdHash { + hash: [u8; 32], + expiry: u64, +} + +#[derive(Debug, Clone, Default)] +pub struct SeenIds { + pub ids: AHashSet, + pub has_changes: bool, +} + +impl SeenIdHash { + pub fn new(id: &str, expiry: u64) -> Self { + let mut hasher = blake3::Hasher::new(); + hasher.update(id.as_bytes()); + SeenIdHash { + hash: hasher.finalize().into(), + expiry, + } + } +} + +impl PartialOrd for SeenIdHash { + fn partial_cmp(&self, other: &Self) -> Option { + self.expiry.partial_cmp(&other.expiry) + } +} + +impl Ord for SeenIdHash { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + self.expiry.cmp(&other.expiry) + } +} + +impl std::hash::Hash for SeenIdHash { + fn hash(&self, state: &mut H) { + self.hash.hash(state); + } +} + +impl PartialEq for SeenIdHash { + fn eq(&self, other: &Self) -> bool { + self.hash == other.hash + } +} + +impl Eq for SeenIdHash {} + +// SeenIds serializer +impl serde::Serialize for SeenIds { + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + let mut seq = serializer.serialize_seq((self.ids.len() * 2).into())?; + for id in &self.ids { + seq.serialize_element(&id.expiry)?; + seq.serialize_element(&id.hash)?; + } + + seq.end() + } +} + +impl<'de> serde::Deserialize<'de> for SeenIds { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_seq(SeenIdsVisitor) + } +} + +struct SeenIdsVisitor; + +impl<'de> serde::de::Visitor<'de> for SeenIdsVisitor { + type Value = SeenIds; + + fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + formatter.write_str("invalid SeenIds") + } + + fn visit_seq(self, mut seq: A) -> Result + where + A: serde::de::SeqAccess<'de>, + { + let num_entries = seq.size_hint().unwrap_or(0) / 2; + let mut seen_ids = SeenIds { + ids: AHashSet::with_capacity(num_entries), + has_changes: false, + }; + let now = now(); + + for _ in 0..num_entries { + let expiry = seq + .next_element::()? + .ok_or_else(|| serde::de::Error::custom("Expected expiry."))?; + if expiry > now { + seen_ids.ids.insert(SeenIdHash { + hash: seq + .next_element()? + .ok_or_else(|| serde::de::Error::custom("Expected hash."))?, + expiry, + }); + } else { + seq.next_element::<[u8; 32]>()? + .ok_or_else(|| serde::de::Error::custom("Expected hash."))?; + seen_ids.has_changes = true; + } + } + + Ok(seen_ids) + } +} diff --git a/crates/jmap/src/sieve/query.rs b/crates/jmap/src/sieve/query.rs new file mode 100644 index 00000000..769d8593 --- /dev/null +++ b/crates/jmap/src/sieve/query.rs @@ -0,0 +1,68 @@ +use jmap_proto::{ + error::method::MethodError, + method::query::{ + Comparator, Filter, QueryRequest, QueryResponse, RequestArguments, SortProperty, + }, + types::{collection::Collection, property::Property}, +}; +use store::{ + fts::Language, + query::{self}, +}; + +use crate::JMAP; + +impl JMAP { + pub async fn sieve_script_query( + &self, + mut request: QueryRequest, + ) -> Result { + let account_id = request.account_id.document_id(); + let mut filters = Vec::with_capacity(request.filter.len()); + + for cond in std::mem::take(&mut request.filter) { + match cond { + Filter::Name(name) => filters.push(query::Filter::has_text( + Property::Name, + &name, + Language::None, + )), + Filter::IsActive(is_active) => { + filters.push(query::Filter::lt(Property::IsActive, is_active as u32)) + } + other => return Err(MethodError::UnsupportedFilter(other.to_string())), + } + } + + let result_set = self + .filter(account_id, Collection::SieveScript, filters) + .await?; + + let (response, paginate) = self.build_query_response(&result_set, &request).await?; + + if let Some(paginate) = paginate { + // Parse sort criteria + let mut comparators = Vec::with_capacity(request.sort.as_ref().map_or(1, |s| s.len())); + for comparator in request + .sort + .and_then(|s| if !s.is_empty() { s.into() } else { None }) + .unwrap_or_else(|| vec![Comparator::descending(SortProperty::ReceivedAt)]) + { + comparators.push(match comparator.property { + SortProperty::Name => { + query::Comparator::field(Property::Name, comparator.is_ascending) + } + SortProperty::IsActive => { + query::Comparator::field(Property::IsActive, comparator.is_ascending) + } + other => return Err(MethodError::UnsupportedSort(other.to_string())), + }); + } + + // Sort results + self.sort(result_set, comparators, paginate, response).await + } else { + Ok(response) + } + } +} diff --git a/crates/jmap/src/sieve/set.rs b/crates/jmap/src/sieve/set.rs new file mode 100644 index 00000000..1c9bde1e --- /dev/null +++ b/crates/jmap/src/sieve/set.rs @@ -0,0 +1,551 @@ +use jmap_proto::{ + error::{ + method::MethodError, + set::{SetError, SetErrorType}, + }, + method::set::{SetRequest, SetResponse}, + object::{ + index::{IndexAs, IndexProperty, ObjectIndexBuilder}, + sieve::SetArguments, + Object, + }, + request::reference::MaybeReference, + response::references::EvalObjectReferences, + types::{ + blob::BlobId, + collection::Collection, + id::Id, + property::Property, + value::{MaybePatchValue, SetValue, Value}, + }, +}; +use sieve::compiler::ErrorType; +use store::{ + query::Filter, + rand::{distributions::Alphanumeric, thread_rng, Rng}, + write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_CLEAR, F_VALUE}, + BlobKind, +}; + +use crate::{auth::AclToken, JMAP}; + +struct SetContext<'x> { + account_id: u32, + acl_token: &'x AclToken, + response: SetResponse, +} + +pub static SCHEMA: &[IndexProperty] = &[ + IndexProperty::new(Property::Name) + .index_as(IndexAs::Text { + tokenize: true, + index: true, + }) + .max_size(255) + .required(), + IndexProperty::new(Property::IsActive).index_as(IndexAs::Integer), +]; + +impl JMAP { + pub async fn sieve_script_set( + &self, + mut request: SetRequest, + acl_token: &AclToken, + ) -> Result { + let account_id = acl_token.primary_id(); + let mut sieve_ids = self + .get_document_ids(account_id, Collection::SieveScript) + .await? + .unwrap_or_default(); + let mut ctx = SetContext { + account_id, + acl_token, + response: self + .prepare_set_response(&request, Collection::SieveScript) + .await?, + }; + let will_destroy = request.unwrap_destroy(); + + // Process creates + let mut changes = ChangeLogBuilder::new(); + for (id, object) in request.unwrap_create() { + if sieve_ids.len() as usize <= self.config.sieve_max_scripts { + match self.sieve_set_item(object, None, &ctx).await? { + Ok((builder, Some(blob))) => { + // Obtain document id + let document_id = self + .assign_document_id(account_id, Collection::SieveScript) + .await?; + + // Store blob + let blob_id = + BlobId::linked(account_id, Collection::SieveScript, document_id); + self.put_blob(&blob_id.kind, &blob).await?; + + // Write record + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::SieveScript) + .create_document(document_id) + .custom(builder); + sieve_ids.insert(document_id); + self.write_batch(batch).await?; + changes.log_insert(Collection::SieveScript, document_id); + + // Add result with updated blobId + ctx.response.created.insert( + id, + Object::with_capacity(1) + .with_property(Property::Id, Value::Id(document_id.into())) + .with_property( + Property::BlobId, + blob_id.with_section_size(blob.len()), + ), + ); + } + Err(err) => { + ctx.response.not_created.append(id, err); + } + _ => unreachable!(), + } + } else { + ctx.response.not_created.append(id, SetError::new(SetErrorType::OverQuota).with_description( + "There are too many sieve scripts, please delete some before adding a new one.", + )); + } + } + + // Process updates + 'update: for (id, object) in request.unwrap_update() { + // Make sure id won't be destroyed + if will_destroy.contains(&id) { + ctx.response + .not_updated + .append(id, SetError::will_destroy()); + continue 'update; + } + + // Obtain sieve script + let document_id = id.document_id(); + if let Some(mut sieve) = self + .get_property::>>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + { + match self + .sieve_set_item(object, (document_id, sieve.take()).into(), &ctx) + .await? + { + Ok((builder, blob)) => { + // Store blob + let blob_id = if let Some(blob) = blob { + let blob_id = + BlobId::linked(account_id, Collection::SieveScript, document_id); + self.put_blob(&blob_id.kind, &blob).await?; + Some(blob_id.with_section_size(blob.len())) + } else { + None + }; + + // Write record + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::SieveScript) + .update_document(document_id) + .assert_value(Property::Value, &sieve) + .custom(builder); + if !batch.is_empty() { + changes.log_update(Collection::SieveScript, document_id); + match self.store.write(batch.build()).await { + Ok(_) => (), + Err(store::Error::AssertValueFailed) => { + ctx.response.not_updated.append(id, SetError::forbidden().with_description( + "Another process modified this sieve, please try again.", + )); + continue 'update; + } + Err(err) => { + tracing::error!( + event = "error", + context = "sieve_set", + account_id = account_id, + error = ?err, + "Failed to update sieve script(s)."); + return Err(MethodError::ServerPartialFail); + } + } + } + + // Add result with updated blobId + ctx.response.updated.append( + id, + blob_id.map(|blob_id| { + Object::with_capacity(1).with_property(Property::BlobId, blob_id) + }), + ); + } + Err(err) => { + ctx.response.not_updated.append(id, err); + continue 'update; + } + } + } else { + ctx.response.not_updated.append(id, SetError::not_found()); + } + } + + // Process deletions + for id in will_destroy { + let document_id = id.document_id(); + if sieve_ids.contains(document_id) { + // Make sure the script is not active + if matches!( + self.get_property::>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + .and_then(|mut obj| obj.properties.remove(&Property::IsActive)), + Some(Value::Bool(true)) + ) { + ctx.response.not_destroyed.append( + id, + SetError::new(SetErrorType::ScriptIsActive) + .with_description("Deactivate Sieve script before deletion."), + ); + continue; + } + self.sieve_script_delete(account_id, document_id).await?; + changes.log_delete(Collection::SieveScript, document_id); + ctx.response.destroyed.push(id); + } else { + ctx.response.not_destroyed.append(id, SetError::not_found()); + } + } + + // Write changes + if !changes.is_empty() { + ctx.response.new_state = self.commit_changes(account_id, changes).await?.into(); + } + + // Activate / deactivate scripts + if ctx.response.not_created.is_empty() + && ctx.response.not_updated.is_empty() + && ctx.response.not_destroyed.is_empty() + && (request.arguments.on_success_activate_script.is_some() + || request + .arguments + .on_success_deactivate_script + .unwrap_or(false)) + { + let changed_ids = if let Some(id) = request.arguments.on_success_activate_script { + self.sieve_activate_script( + account_id, + match id { + MaybeReference::Value(id) => id.document_id(), + MaybeReference::Reference(id_ref) => match ctx.response.get_id(&id_ref) { + Some(id) => id.document_id(), + None => return Ok(ctx.response), + }, + } + .into(), + ) + .await? + } else { + self.sieve_activate_script(account_id, None).await? + }; + + for (document_id, is_active) in changed_ids { + if let Some(obj) = ctx.response.get_object_by_id(Id::from(document_id)) { + obj.append(Property::IsActive, Value::Bool(is_active)); + } + } + } + + Ok(ctx.response) + } + + pub async fn sieve_script_delete( + &self, + account_id: u32, + document_id: u32, + ) -> Result<(), MethodError> { + // Delete record + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::SieveScript) + .delete_document(document_id) + .value(Property::Value, (), F_VALUE | F_CLEAR) + .value(Property::EmailIds, (), F_VALUE | F_CLEAR); + self.write_batch(batch).await?; + let _ = self + .delete_blob(&BlobKind::Linked { + account_id, + collection: Collection::SieveScript.into(), + document_id, + }) + .await; + Ok(()) + } + + #[allow(clippy::blocks_in_if_conditions)] + async fn sieve_set_item( + &self, + changes_: Object, + update: Option<(u32, Object)>, + ctx: &SetContext<'_>, + ) -> Result>), SetError>, MethodError> { + // Vacation script cannot be modified + if matches!(update.as_ref().and_then(|(_, obj)| obj.properties.get(&Property::Name)), Some(Value::Text ( value )) if value.eq_ignore_ascii_case("vacation")) + { + return Ok(Err(SetError::forbidden().with_description( + "The 'vacation' script cannot be modified, use VacationResponse/set instead.", + ))); + } + + // Parse properties + let mut changes = Object::with_capacity(changes_.properties.len()); + let mut blob_id = None; + for (property, value) in changes_.properties { + let value = match ctx.response.eval_object_references(value) { + Ok(value) => value, + Err(err) => { + return Ok(Err(err)); + } + }; + let value = match (&property, value) { + (Property::Name, MaybePatchValue::Value(Value::Text(value))) => { + if value.len() > self.config.sieve_max_script_name { + return Ok(Err(SetError::invalid_properties() + .with_property(property) + .with_description("Script name is too long."))); + } else if value.eq_ignore_ascii_case("vacation") { + return Ok(Err(SetError::forbidden() + .with_property(property) + .with_description( + "The 'vacation' name is reserved, please use a different name.", + ))); + } else if update + .as_ref() + .and_then(|(_, obj)| obj.properties.get(&Property::Name)) + .map_or( + true, + |p| matches!(p, Value::Text (prev_value ) if prev_value != &value), + ) + { + if let Some(id) = self + .filter( + ctx.account_id, + Collection::SieveScript, + vec![Filter::eq(Property::Name, &value)], + ) + .await? + .results + .min() + { + return Ok(Err(SetError::already_exists() + .with_existing_id(id.into()) + .with_description(format!( + "A sieve script with name '{}' already exists.", + value + )))); + } + } + + Value::Text(value) + } + (Property::BlobId, MaybePatchValue::Value(Value::BlobId(value))) => { + blob_id = value.into(); + continue; + } + (Property::Name, MaybePatchValue::Value(Value::Null)) => { + continue; + } + _ => { + return Ok(Err(SetError::invalid_properties() + .with_property(property) + .with_description("Invalid property or value.".to_string()))) + } + }; + changes.append(property, value); + } + + if update.is_none() { + // Add name if missing + if !matches!(changes.properties.get(&Property::Name), Some(Value::Text ( value )) if !value.is_empty()) + { + changes.set( + Property::Name, + Value::Text( + thread_rng() + .sample_iter(Alphanumeric) + .take(15) + .map(char::from) + .collect::(), + ), + ); + } + + // Set script as inactive + changes.set(Property::IsActive, Value::Bool(false)); + } + + let blob_update = if let Some(blob_id) = blob_id { + if update.as_ref().map_or(true, |(document_id, _)| { + !blob_id + .kind + .is_document(ctx.account_id, Collection::SieveScript, *document_id) + }) { + // Check access + if let Some(mut bytes) = self.blob_download(&blob_id, ctx.acl_token).await? { + // Compile script + match self.sieve_compiler.compile(&bytes) { + Ok(script) => { + changes.set(Property::BlobId, Value::UnsignedInt(bytes.len() as u64)); + bytes.extend(bincode::serialize(&script).unwrap_or_default()); + bytes.into() + } + Err(err) => { + return Ok(Err(SetError::new( + if let ErrorType::ScriptTooLong = &err.error_type() { + SetErrorType::TooLarge + } else { + SetErrorType::InvalidScript + }, + ) + .with_description(err.to_string()))); + } + } + } else { + return Ok(Err(SetError::new(SetErrorType::BlobNotFound) + .with_property(Property::BlobId) + .with_description("Blob does not exist."))); + } + } else { + None + } + } else if update.is_none() { + return Ok(Err(SetError::invalid_properties() + .with_property(Property::BlobId) + .with_description("Missing blobId."))); + } else { + None + }; + + // Validate + Ok(ObjectIndexBuilder::new(SCHEMA) + .with_changes(changes) + .with_current_opt(update.map(|(_, current)| current)) + .validate() + .map(|obj| (obj, blob_update))) + } + + pub async fn sieve_activate_script( + &self, + account_id: u32, + activate_id: Option, + ) -> Result, MethodError> { + let mut changed_ids = Vec::new(); + // Find the currently active script + let active_ids = self + .filter( + account_id, + Collection::SieveScript, + vec![Filter::eq(Property::IsActive, 1u32)], + ) + .await? + .results; + + // Check if script is already active + if activate_id.map_or(false, |id| active_ids.contains(id)) { + return Ok(changed_ids); + } + + // Prepare batch + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::SieveScript); + + // Deactivate scripts + for document_id in active_ids { + if let Some(sieve) = self + .get_property::>>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + { + batch + .update_document(document_id) + .value(Property::EmailIds, (), F_VALUE | F_CLEAR) + .assert_value(Property::Value, &sieve) + .custom( + ObjectIndexBuilder::new(SCHEMA) + .with_changes( + Object::with_capacity(1).with_property(Property::IsActive, false), + ) + .with_current(sieve.inner), + ); + changed_ids.push((document_id, false)); + } + } + + // Activate script + if let Some(document_id) = activate_id { + if let Some(sieve) = self + .get_property::>>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + { + batch + .update_document(document_id) + .assert_value(Property::Value, &sieve) + .custom( + ObjectIndexBuilder::new(SCHEMA) + .with_changes( + Object::with_capacity(1).with_property(Property::IsActive, true), + ) + .with_current(sieve.inner), + ); + changed_ids.push((document_id, true)); + } + } + + // Write changes + if !changed_ids.is_empty() { + match self.store.write(batch.build()).await { + Ok(_) => (), + Err(store::Error::AssertValueFailed) => { + return Ok(vec![]); + } + Err(err) => { + tracing::error!( + event = "error", + context = "sieve_activate_script", + account_id = account_id, + error = ?err, + "Failed to activate sieve script(s)."); + return Err(MethodError::ServerPartialFail); + } + } + } + + Ok(changed_ids) + } +} diff --git a/crates/jmap/src/sieve/validate.rs b/crates/jmap/src/sieve/validate.rs new file mode 100644 index 00000000..ea9778be --- /dev/null +++ b/crates/jmap/src/sieve/validate.rs @@ -0,0 +1,32 @@ +use jmap_proto::{ + error::{ + method::MethodError, + set::{SetError, SetErrorType}, + }, + method::validate::{ValidateSieveScriptRequest, ValidateSieveScriptResponse}, +}; + +use crate::{auth::AclToken, JMAP}; + +impl JMAP { + pub async fn sieve_script_validate( + &self, + request: ValidateSieveScriptRequest, + acl_token: &AclToken, + ) -> Result { + Ok(ValidateSieveScriptResponse { + account_id: request.account_id, + error: match self + .blob_download(&request.blob_id, acl_token) + .await? + .map(|bytes| self.sieve_compiler.compile(&bytes)) + { + Some(Ok(_)) => None, + Some(Err(err)) => SetError::new(SetErrorType::InvalidScript) + .with_description(err.to_string()) + .into(), + None => SetError::new(SetErrorType::BlobNotFound).into(), + }, + }) + } +} diff --git a/crates/jmap/src/vacation/get.rs b/crates/jmap/src/vacation/get.rs new file mode 100644 index 00000000..f76a31e8 --- /dev/null +++ b/crates/jmap/src/vacation/get.rs @@ -0,0 +1,106 @@ +use jmap_proto::{ + error::method::MethodError, + method::get::{GetRequest, GetResponse, RequestArguments}, + object::Object, + request::reference::MaybeReference, + types::{collection::Collection, id::Id, property::Property, value::Value}, +}; +use store::query::Filter; + +use crate::JMAP; + +impl JMAP { + pub async fn vacation_response_get( + &self, + mut request: GetRequest, + ) -> Result { + let account_id = request.account_id.document_id(); + let properties = request.unwrap_properties(&[ + Property::Id, + Property::IsEnabled, + Property::FromDate, + Property::ToDate, + Property::Subject, + Property::TextBody, + Property::HtmlBody, + ]); + let mut response = GetResponse { + account_id: Some(request.account_id), + state: self + .get_state(account_id, Collection::SieveScript) + .await? + .into(), + list: Vec::with_capacity(1), + not_found: vec![], + }; + + let do_get = if let Some(MaybeReference::Value(ids)) = request.ids { + let mut do_get = false; + for id in ids { + if id.is_singleton() { + do_get = true; + } else { + response.not_found.push(id); + } + } + do_get + } else { + true + }; + if do_get { + if let Some(document_id) = self.get_vacation_sieve_script_id(account_id).await? { + if let Some(mut obj) = self + .get_property::>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + { + let mut result = Object::with_capacity(properties.len()); + for property in &properties { + match property { + Property::Id => { + result.append(Property::Id, Value::Id(Id::singleton())); + } + Property::IsEnabled => { + result.append(Property::IsEnabled, obj.remove(&Property::IsActive)); + } + Property::FromDate + | Property::ToDate + | Property::Subject + | Property::TextBody + | Property::HtmlBody => { + result.append(property.clone(), obj.remove(property)); + } + property => { + result.append(property.clone(), Value::Null); + } + } + } + response.list.push(result); + } else { + response.not_found.push(Id::singleton()); + } + } else { + response.not_found.push(Id::singleton()); + } + } + + Ok(response) + } + + pub async fn get_vacation_sieve_script_id( + &self, + account_id: u32, + ) -> Result, MethodError> { + self.filter( + account_id, + Collection::SieveScript, + vec![Filter::eq(Property::Name, "vacation")], + ) + .await + .map(|r| r.results.min()) + } +} diff --git a/crates/jmap/src/vacation/mod.rs b/crates/jmap/src/vacation/mod.rs new file mode 100644 index 00000000..e7537b1c --- /dev/null +++ b/crates/jmap/src/vacation/mod.rs @@ -0,0 +1,2 @@ +pub mod get; +pub mod set; diff --git a/crates/jmap/src/vacation/set.rs b/crates/jmap/src/vacation/set.rs new file mode 100644 index 00000000..9c31c914 --- /dev/null +++ b/crates/jmap/src/vacation/set.rs @@ -0,0 +1,394 @@ +use std::borrow::Cow; + +use jmap_proto::{ + error::{ + method::MethodError, + set::{SetError, SetErrorType}, + }, + method::set::{RequestArguments, SetRequest, SetResponse}, + object::{index::ObjectIndexBuilder, Object}, + response::references::EvalObjectReferences, + types::{ + collection::Collection, + id::Id, + property::Property, + value::{MaybePatchValue, Value}, + }, +}; +use mail_builder::MessageBuilder; +use mail_parser::decoders::html::html_to_text; +use store::{ + write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_CLEAR, F_VALUE}, + BlobKind, +}; + +use crate::{auth::AclToken, sieve::set::SCHEMA, JMAP}; + +impl JMAP { + pub async fn vacation_response_set( + &self, + mut request: SetRequest, + acl_token: &AclToken, + ) -> Result { + let account_id = acl_token.primary_id(); + let mut response = self + .prepare_set_response(&request, Collection::SieveScript) + .await?; + let will_destroy = request.unwrap_destroy(); + + // Process set or update requests + let mut create_id = None; + let mut changes = None; + match (request.create, request.update) { + (Some(create), Some(update)) if !create.is_empty() && !update.is_empty() => { + return Err(MethodError::InvalidArguments( + "Creating and updating on the same request is not allowed.".into(), + )); + } + (Some(create), _) if !create.is_empty() => { + for (id, obj) in create { + if will_destroy.contains(&Id::singleton()) { + response.not_created.append( + id, + SetError::new(SetErrorType::WillDestroy) + .with_description("ID will be destroyed."), + ); + } else if create_id.is_some() { + response.not_created.append( + id, + SetError::forbidden() + .with_description("Only one object can be created."), + ); + } else { + create_id = Some(id); + changes = Some(obj); + } + } + } + (_, Some(update)) if !update.is_empty() => { + for (id, obj) in update { + if id.is_singleton() { + if !will_destroy.contains(&id) { + changes = Some(obj); + } else { + response.not_updated.append( + id, + SetError::new(SetErrorType::WillDestroy) + .with_description("ID will be destroyed."), + ); + } + } else { + response.not_updated.append( + id, + SetError::new(SetErrorType::NotFound).with_description("ID not found."), + ); + } + } + } + _ => { + return Ok(response); + } + } + + // Process changes + let mut change_log = ChangeLogBuilder::new(); + if let Some(changes_) = changes { + // Parse properties + let mut changes = Object::with_capacity(changes_.properties.len()); + let mut is_active = false; + let mut build_script = create_id.is_some(); + + for (property, value) in changes_.properties { + let value = match response.eval_object_references(value) { + Ok(value) => value, + Err(err) => { + return Ok(set_error(response, create_id, err)); + } + }; + match (&property, value) { + (Property::Subject, MaybePatchValue::Value(Value::Text(value))) + if value.len() < 512 => + { + build_script = true; + changes.append(property, Value::Text(value)); + } + ( + Property::HtmlBody | Property::TextBody, + MaybePatchValue::Value(Value::Text(value)), + ) if value.len() < 2048 => { + build_script = true; + + changes.append(property, Value::Text(value)); + } + ( + Property::ToDate | Property::FromDate, + MaybePatchValue::Value(value @ Value::Date(_)), + ) => { + build_script = true; + changes.append(property, value); + } + (Property::IsEnabled, MaybePatchValue::Value(Value::Bool(value))) => { + is_active = value; + changes.append(Property::IsActive, value); + } + (Property::IsEnabled, MaybePatchValue::Value(Value::Null)) => { + changes.append(Property::IsActive, Value::Bool(false)); + } + ( + Property::Subject + | Property::HtmlBody + | Property::TextBody + | Property::ToDate + | Property::FromDate, + MaybePatchValue::Value(Value::Null), + ) => { + if create_id.is_none() { + build_script = true; + + changes.append(property, Value::Null); + } + } + _ => { + return Ok(set_error( + response, + create_id, + SetError::invalid_properties() + .with_property(property) + .with_description("Field could not be set."), + )); + } + } + } + + // Add name and isActive + if create_id.is_some() { + changes.append(Property::Name, Value::Text("vacation".into())); + if !changes.properties.contains_key(&Property::IsActive) { + changes.append(Property::IsActive, Value::Bool(false)); + } + } + + // Prepare write batch + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::SieveScript); + + // Obtain current script + let document_id = self.get_vacation_sieve_script_id(account_id).await?; + let mut was_active = false; + let mut obj = ObjectIndexBuilder::new(SCHEMA).with_current_opt( + if let Some(document_id) = document_id { + self.get_property::>>( + account_id, + Collection::SieveScript, + document_id, + Property::Value, + ) + .await? + .map(|value| { + batch.assert_value(Property::Value, &value); + was_active = value.inner.properties.get(&Property::IsActive) + == Some(&Value::Bool(true)); + value.inner + }) + .ok_or(MethodError::ServerPartialFail)? + .into() + } else { + None + }, + ); + + // Create sieve script only if there are changes + let script_blob = if build_script { + self.build_script(&mut obj)?.into() + } else { + None + }; + + // Write changes + let document_id = if let Some(document_id) = document_id { + batch + .update_document(document_id) + .value(Property::EmailIds, (), F_VALUE | F_CLEAR) + .custom(obj); + change_log.log_insert(Collection::SieveScript, document_id); + document_id + } else { + let document_id = self + .assign_document_id(account_id, Collection::SieveScript) + .await?; + batch.create_document(document_id).custom(obj); + change_log.log_update(Collection::SieveScript, document_id); + document_id + }; + if !batch.is_empty() { + self.write_batch(batch).await?; + } + + // Write blob + if let Some(script_blob) = script_blob { + self.put_blob( + &BlobKind::Linked { + account_id, + collection: Collection::SieveScript.into(), + document_id, + }, + &script_blob, + ) + .await?; + } + + // Deactivate other sieve scripts + if !was_active && is_active { + self.sieve_activate_script(account_id, document_id.into()) + .await?; + } + + // Add result + if let Some(create_id) = create_id { + response.created.insert( + create_id, + Object::with_capacity(1).with_property(Property::Id, Id::singleton()), + ); + } else { + response.updated.append(Id::singleton(), None); + } + } else if !will_destroy.is_empty() { + for id in will_destroy { + if id.is_singleton() { + if let Some(document_id) = self.get_vacation_sieve_script_id(account_id).await? + { + self.sieve_script_delete(account_id, document_id).await?; + change_log.log_delete(Collection::SieveScript, document_id); + response.destroyed.push(id); + continue; + } + } + + response.not_destroyed.append(id, SetError::not_found()); + } + } + + // Write changes + if !change_log.is_empty() { + response.new_state = self.commit_changes(account_id, change_log).await?.into(); + } + + Ok(response) + } + + fn build_script(&self, obj: &mut ObjectIndexBuilder) -> Result, MethodError> { + // Build Sieve script + let mut script = Vec::with_capacity(1024); + script.extend_from_slice(b"require [\"vacation\", \"relational\", \"date\"];\r\n\r\n"); + let mut num_blocks = 0; + + // Add start date + if let Value::Date(value) = obj.get(&Property::FromDate) { + script.extend_from_slice(b"if currentdate :value \"ge\" \"iso8601\" \""); + script.extend_from_slice(value.to_string().as_bytes()); + script.extend_from_slice(b"\" {\r\n"); + num_blocks += 1; + } + + // Add end date + if let Value::Date(value) = obj.get(&Property::ToDate) { + script.extend_from_slice(b"if currentdate :value \"le\" \"iso8601\" \""); + script.extend_from_slice(value.to_string().as_bytes()); + script.extend_from_slice(b"\" {\r\n"); + num_blocks += 1; + } + + script.extend_from_slice(b"vacation :mime "); + if let Value::Text(value) = obj.get(&Property::Subject) { + script.extend_from_slice(b":subject \""); + for &ch in value.as_bytes().iter() { + match ch { + b'\\' | b'\"' => { + script.push(b'\\'); + } + b'\r' | b'\n' => { + continue; + } + _ => (), + } + script.push(ch); + } + script.extend_from_slice(b"\" "); + } + + let mut text_body = if let Value::Text(value) = obj.get(&Property::TextBody) { + Cow::from(value.as_str()).into() + } else { + None + }; + let html_body = if let Value::Text(value) = obj.get(&Property::HtmlBody) { + Cow::from(value.as_str()).into() + } else { + None + }; + match (&html_body, &text_body) { + (Some(html_body), None) => { + text_body = Cow::from(html_to_text(html_body.as_ref())).into(); + } + (None, None) => { + text_body = Cow::from("I am away.").into(); + } + _ => (), + } + + let mut builder = MessageBuilder::new(); + let mut body_len = 0; + if let Some(html_body) = html_body { + body_len = html_body.len(); + builder = builder.html_body(html_body); + } + if let Some(text_body) = text_body { + body_len += text_body.len(); + builder = builder.text_body(text_body); + } + let mut message_body = Vec::with_capacity(body_len + 128); + builder.write_body(&mut message_body).ok(); + + script.push(b'\"'); + for ch in message_body { + if [b'\\', b'\"'].contains(&ch) { + script.push(b'\\'); + } + script.push(ch); + } + script.extend_from_slice(b"\";\r\n"); + + // Close blocks + for _ in 0..num_blocks { + script.extend_from_slice(b"}\r\n"); + } + + // Compile script + match self.sieve_compiler.compile(&script) { + Ok(compiled_script) => { + // Update blob length + obj.set(Property::BlobId, Value::UnsignedInt(script.len() as u64)); + + // Serialize script + script.extend(bincode::serialize(&compiled_script).unwrap_or_default()); + + Ok(script) + } + Err(err) => { + tracing::error!("Vacation Sieve Script failed to compile: {}", err); + Err(MethodError::ServerPartialFail) + } + } + } +} + +fn set_error(mut response: SetResponse, id: Option, err: SetError) -> SetResponse { + if let Some(id) = id { + response.not_created.append(id, err); + } else { + response.not_updated.append(Id::singleton(), err); + } + response +} diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index 900ddd64..a942d618 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -1,10 +1,7 @@ use std::time::Duration; use jmap::{api::JmapSessionManager, services::IPC_CHANNEL_BUFFER, JMAP}; -use smtp::{ - core::{SmtpAdminSessionManager, SmtpSessionManager, SMTP}, - outbound::delivery, -}; +use smtp::core::{SmtpAdminSessionManager, SmtpSessionManager, SMTP}; use tokio::sync::mpsc; use utils::{ config::{Config, ServerProtocol}, @@ -28,8 +25,12 @@ async fn main() -> std::io::Result<()> { // Init servers let (delivery_tx, delivery_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); - let smtp = SMTP::init(&config, &servers, delivery_tx).await; - let jmap = JMAP::init(&config, delivery_rx).await; + let smtp = SMTP::init(&config, &servers, delivery_tx) + .await + .failed("Invalid configuration file"); + let jmap = JMAP::init(&config, delivery_rx) + .await + .failed("Invalid configuration file"); // Spawn servers let shutdown_tx = servers.spawn(|server, shutdown_rx| { diff --git a/crates/smtp/src/core/scripts.rs b/crates/smtp/src/core/scripts.rs index fa625f36..f2d85eb7 100644 --- a/crates/smtp/src/core/scripts.rs +++ b/crates/smtp/src/core/scripts.rs @@ -49,6 +49,7 @@ pub enum ScriptResult { Accept, Replace(Vec), Reject(String), + Discard, } impl Session { @@ -239,7 +240,7 @@ impl SMTP { input = true.into(); } Event::Discard => { - reject_reason = "503 5.5.3 Message rejected.\r\n".to_string().into(); + keep_id = usize::MAX - 1; input = true.into(); } Event::Reject { reason, .. } => { @@ -449,6 +450,11 @@ impl SMTP { } } + // Keep id + // 0 = use original message + // MAX = implicit keep + // MAX - 1 = discard message + if keep_id == 0 { ScriptResult::Accept } else if let Some(mut reject_reason) = reject_reason { @@ -465,12 +471,14 @@ impl SMTP { } else { ScriptResult::Reject(format!("503 5.5.3 {reject_reason}")) } - } else { + } else if keep_id != usize::MAX - 1 { messages .into_iter() .nth(keep_id - 1) .map(ScriptResult::Replace) .unwrap_or(ScriptResult::Accept) + } else { + ScriptResult::Discard } } } diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index f065128c..35f243df 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -393,6 +393,9 @@ impl Session { return message.into_bytes().into(); } + ScriptResult::Discard => { + return (b"250 2.0.0 Message queued for delivery.\r\n"[..]).into(); + } } } diff --git a/crates/smtp/src/inbound/ehlo.rs b/crates/smtp/src/inbound/ehlo.rs index 0b776e75..37e7d974 100644 --- a/crates/smtp/src/inbound/ehlo.rs +++ b/crates/smtp/src/inbound/ehlo.rs @@ -96,20 +96,17 @@ impl Session { // Sieve filtering if let Some(script) = self.core.session.config.ehlo.script.eval(self).await { - match self.run_script(script.clone(), None).await { - ScriptResult::Accept | ScriptResult::Replace(_) => (), - ScriptResult::Reject(message) => { - tracing::debug!(parent: &self.span, + if let ScriptResult::Reject(message) = self.run_script(script.clone(), None).await { + tracing::debug!(parent: &self.span, context = "ehlo", event = "sieve-reject", domain = &self.data.helo_domain, reason = message); - self.data.mail_from = None; - self.data.helo_domain = prev_helo_domain; - self.data.spf_ehlo = None; - return self.write(message.as_bytes()).await; - } + self.data.mail_from = None; + self.data.helo_domain = prev_helo_domain; + self.data.spf_ehlo = None; + return self.write(message.as_bytes()).await; } } diff --git a/crates/smtp/src/inbound/mail.rs b/crates/smtp/src/inbound/mail.rs index 1e981e07..bac6e4bf 100644 --- a/crates/smtp/src/inbound/mail.rs +++ b/crates/smtp/src/inbound/mail.rs @@ -139,17 +139,14 @@ impl Session { // Sieve filtering if let Some(script) = self.core.session.config.mail.script.eval(self).await { - match self.run_script(script.clone(), None).await { - ScriptResult::Accept | ScriptResult::Replace(_) => (), - ScriptResult::Reject(message) => { - tracing::debug!(parent: &self.span, + if let ScriptResult::Reject(message) = self.run_script(script.clone(), None).await { + tracing::debug!(parent: &self.span, context = "mail-from", event = "sieve-reject", address = &self.data.mail_from.as_ref().unwrap().address, reason = message); - self.data.mail_from = None; - return self.write(message.as_bytes()).await; - } + self.data.mail_from = None; + return self.write(message.as_bytes()).await; } } diff --git a/crates/smtp/src/inbound/rcpt.rs b/crates/smtp/src/inbound/rcpt.rs index be1785f3..d7e90bf0 100644 --- a/crates/smtp/src/inbound/rcpt.rs +++ b/crates/smtp/src/inbound/rcpt.rs @@ -133,17 +133,14 @@ impl Session { // Sieve filtering if let Some(script) = &self.params.rcpt_script { - match self.run_script(script.clone(), None).await { - ScriptResult::Accept | ScriptResult::Replace(_) => (), - ScriptResult::Reject(message) => { - tracing::debug!(parent: &self.span, + if let ScriptResult::Reject(message) = self.run_script(script.clone(), None).await { + tracing::debug!(parent: &self.span, context = "rcpt", event = "sieve-reject", address = &self.data.rcpt_to.last().unwrap().address, reason = message); - self.data.rcpt_to.pop(); - return self.write(message.as_bytes()).await; - } + self.data.rcpt_to.pop(); + return self.write(message.as_bytes()).await; } } diff --git a/crates/smtp/src/inbound/spawn.rs b/crates/smtp/src/inbound/spawn.rs index 677cb769..d78c7a64 100644 --- a/crates/smtp/src/inbound/spawn.rs +++ b/crates/smtp/src/inbound/spawn.rs @@ -146,17 +146,14 @@ impl Session { // Sieve filtering if let Some(script) = self.core.session.config.connect.script.eval(self).await { - match self.run_script(script.clone(), None).await { - ScriptResult::Accept | ScriptResult::Replace(_) => (), - ScriptResult::Reject(message) => { - tracing::debug!(parent: &self.span, + if let ScriptResult::Reject(message) = self.run_script(script.clone(), None).await { + tracing::debug!(parent: &self.span, context = "connect", event = "sieve-reject", reason = message); - let _ = self.write(message.as_bytes()).await; - return false; - } + let _ = self.write(message.as_bytes()).await; + return false; } } diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index d52434ad..79a2ef6c 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -29,7 +29,7 @@ use std::sync::Arc; use config::{ auth::ConfigAuth, database::ConfigDatabase, list::ConfigList, queue::ConfigQueue, remote::ConfigHost, report::ConfigReport, resolver::ConfigResolver, scripts::ConfigSieve, - session::ConfigSession, ConfigContext, + session::ConfigSession, ConfigContext, Host, }; use dashmap::DashMap; use lookup::Lookup; @@ -38,7 +38,7 @@ use queue::manager::SpawnQueue; use reporting::scheduler::SpawnReport; use tokio::sync::mpsc; use utils::{ - config::{Config, Servers}, + config::{Config, ServerProtocol, Servers}, UnwrapFailure, }; @@ -57,43 +57,50 @@ impl SMTP { config: &Config, servers: &Servers, #[cfg(feature = "local_delivery")] delivery_tx: mpsc::Sender, - ) -> Arc { + ) -> Result, String> { // Read configuration parameters let mut config_ctx = ConfigContext::new(&servers.inner); #[cfg(feature = "local_delivery")] - config_ctx.lookup.insert( - "local".to_string(), - Arc::new(Lookup::Local(delivery_tx.clone())), - ); + { + config_ctx.lookup.insert( + "local".to_string(), + Arc::new(Lookup::Local(delivery_tx.clone())), + ); + let (channel_tx, channel_rx) = mpsc::channel(1024); + config_ctx.hosts.insert( + "local".to_string(), + Host { + address: String::new(), + port: 0, + protocol: ServerProtocol::Jmap, + concurrency: Default::default(), + timeout: Default::default(), + tls_implicit: Default::default(), + tls_allow_invalid_certs: Default::default(), + username: Default::default(), + secret: Default::default(), + max_errors: Default::default(), + max_requests: Default::default(), + cache_entries: Default::default(), + cache_ttl_positive: Default::default(), + cache_ttl_negative: Default::default(), + channel_tx, + channel_rx, + lookup: false, + }, + ); + } - config - .parse_remote_hosts(&mut config_ctx) - .failed("Configuration error"); - config - .parse_databases(&mut config_ctx) - .failed("Configuration error"); - config - .parse_lists(&mut config_ctx) - .failed("Configuration error"); - config - .parse_signatures(&mut config_ctx) - .failed("Configuration error"); - let sieve_config = config - .parse_sieve(&mut config_ctx) - .failed("Configuration error"); - let session_config = config - .parse_session_config(&config_ctx) - .failed("Configuration error"); - let queue_config = config - .parse_queue(&config_ctx) - .failed("Configuration error"); - let mail_auth_config = config - .parse_mail_auth(&config_ctx) - .failed("Configuration error"); - let report_config = config - .parse_reports(&config_ctx) - .failed("Configuration error"); + config.parse_remote_hosts(&mut config_ctx)?; + config.parse_databases(&mut config_ctx)?; + config.parse_lists(&mut config_ctx)?; + config.parse_signatures(&mut config_ctx)?; + let sieve_config = config.parse_sieve(&mut config_ctx)?; + let session_config = config.parse_session_config(&config_ctx)?; + let queue_config = config.parse_queue(&config_ctx)?; + let mail_auth_config = config.parse_mail_auth(&config_ctx)?; + let report_config = config.parse_reports(&config_ctx)?; // Build core let (queue_tx, queue_rx) = mpsc::channel(1024); @@ -102,8 +109,7 @@ impl SMTP { worker_pool: rayon::ThreadPoolBuilder::new() .num_threads( config - .property::("global.thread-pool") - .failed("Failed to parse thread pool size") + .property::("global.thread-pool")? .filter(|v| *v > 0) .unwrap_or_else(num_cpus::get), ) @@ -113,14 +119,10 @@ impl SMTP { session: SessionCore { config: session_config, throttle: DashMap::with_capacity_and_hasher_and_shard_amount( - config - .property("global.shared-map.capacity") - .failed("Failed to parse shared map capacity") - .unwrap_or(2), + config.property("global.shared-map.capacity")?.unwrap_or(2), ThrottleKeyHasherBuilder::default(), config - .property::("global.shared-map.shard") - .failed("Failed to parse shared map shard amount") + .property::("global.shared-map.shard")? .unwrap_or(32) .next_power_of_two() as usize, ), @@ -128,27 +130,19 @@ impl SMTP { queue: QueueCore { config: queue_config, throttle: DashMap::with_capacity_and_hasher_and_shard_amount( - config - .property("global.shared-map.capacity") - .failed("Failed to parse shared map capacity") - .unwrap_or(2), + config.property("global.shared-map.capacity")?.unwrap_or(2), ThrottleKeyHasherBuilder::default(), config - .property::("global.shared-map.shard") - .failed("Failed to parse shared map shard amount") + .property::("global.shared-map.shard")? .unwrap_or(32) .next_power_of_two() as usize, ), id_seq: 0.into(), quota: DashMap::with_capacity_and_hasher_and_shard_amount( - config - .property("global.shared-map.capacity") - .failed("Failed to parse shared map capacity") - .unwrap_or(2), + config.property("global.shared-map.capacity")?.unwrap_or(2), ThrottleKeyHasherBuilder::default(), config - .property::("global.shared-map.shard") - .failed("Failed to parse shared map shard amount") + .property::("global.shared-map.shard")? .unwrap_or(32) .next_power_of_two() as usize, ), @@ -181,6 +175,6 @@ impl SMTP { } } - core + Ok(core) } } diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 7a990b62..dcccfd3e 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -137,6 +137,21 @@ pub enum BlobKind { }, } +impl BlobKind { + pub fn is_document( + &self, + account_id: u32, + collection: impl Into, + document_id: u32, + ) -> bool { + matches!(self, BlobKind::Linked { + account_id: a, + collection: c, + document_id: d, + } if *a == account_id && *c == collection.into() && *d == document_id) + } +} + pub type Result = std::result::Result; #[derive(Debug)] diff --git a/crates/store/src/write/batch.rs b/crates/store/src/write/batch.rs index 1512af52..102a16cd 100644 --- a/crates/store/src/write/batch.rs +++ b/crates/store/src/write/batch.rs @@ -133,6 +133,15 @@ impl BatchBuilder { pub fn is_empty(&self) -> bool { self.ops.is_empty() + || self.ops.iter().any(|op| { + !matches!( + op, + Operation::AccountId { .. } + | Operation::Collection { .. } + | Operation::DocumentId { .. } + | Operation::AssertValue { .. } + ) + }) } } diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 07df7a1b..89f6e63f 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -169,6 +169,15 @@ impl SerializeInto for String { } } +impl SerializeInto for Vec { + fn serialize_into(&self, buf: &mut Vec) { + buf.push_leb128(self.len()); + if !self.is_empty() { + buf.extend_from_slice(self.as_slice()); + } + } +} + impl SerializeInto for u32 { fn serialize_into(&self, buf: &mut Vec) { buf.push_leb128(*self); @@ -194,13 +203,19 @@ impl DeserializeFrom for u64 { } impl DeserializeFrom for String { + fn deserialize_from(bytes: &mut Iter<'_, u8>) -> Option { + >::deserialize_from(bytes).and_then(|s| String::from_utf8(s).ok()) + } +} + +impl DeserializeFrom for Vec { fn deserialize_from(bytes: &mut Iter<'_, u8>) -> Option { let len: usize = bytes.next_leb128()?; - let mut s = Vec::with_capacity(len); + let mut buf = Vec::with_capacity(len); for _ in 0..len { - s.push(*bytes.next()?); + buf.push(*bytes.next()?); } - String::from_utf8(s).ok() + buf.into() } } diff --git a/crates/utils/src/ipc.rs b/crates/utils/src/ipc.rs index 7dfd79b1..ec48e6cb 100644 --- a/crates/utils/src/ipc.rs +++ b/crates/utils/src/ipc.rs @@ -3,6 +3,7 @@ use std::{borrow::Cow, path::PathBuf}; use mail_send::Credentials; use tokio::{fs, io::AsyncReadExt, sync::oneshot}; +#[derive(Debug)] pub enum DeliveryEvent { Ingest { message: IngestMessage, @@ -12,6 +13,7 @@ pub enum DeliveryEvent { Stop, } +#[derive(Debug)] pub struct IngestMessage { pub sender_address: String, pub recipients: Vec, diff --git a/tests/src/jmap/delivery.rs b/tests/src/jmap/delivery.rs index 379c226b..3764bb9c 100644 --- a/tests/src/jmap/delivery.rs +++ b/tests/src/jmap/delivery.rs @@ -25,12 +25,12 @@ pub async fn test(server: Arc, client: &mut Client) { let account_id_3 = test_account_create(&server, "bill@example.com", "12345", "Bill Foobar") .await .to_string(); - test_alias_create(&server, "jdoe@example.com", "john.doe@example.com").await; + test_alias_create(&server, "jdoe@example.com", "john.doe@example.com", false).await; // Create a mailing list - test_alias_create(&server, "jdoe@example.com", "members@example.com").await; - test_alias_create(&server, "jane@example.com", "members@example.com").await; - test_alias_create(&server, "bill@example.com", "members@example.com").await; + test_alias_create(&server, "jdoe@example.com", "members@example.com", true).await; + test_alias_create(&server, "jane@example.com", "members@example.com", true).await; + test_alias_create(&server, "bill@example.com", "members@example.com", true).await; // Delivering to individuals let mut lmtp = SmtpConnection::connect().await; @@ -47,6 +47,7 @@ pub async fn test(server: Arc, client: &mut Client) { ), ) .await; + tokio::time::sleep(Duration::from_millis(200)).await; assert_eq!( server .get_document_ids( @@ -77,6 +78,7 @@ pub async fn test(server: Arc, client: &mut Client) { ), ) .await; + tokio::time::sleep(Duration::from_millis(200)).await; assert_eq!( server .get_document_ids( @@ -101,7 +103,7 @@ pub async fn test(server: Arc, client: &mut Client) { lmtp.expn("non_existant@example.com", 5).await; lmtp.expn("jdoe@example.com", 5).await; lmtp.vrfy("jdoe@example.com", 2).await; - lmtp.vrfy("members@example.com", 2).await; + lmtp.vrfy("members@example.com", 5).await; lmtp.vrfy("non_existant@example.com", 5).await; // Delivering to a mailing list @@ -118,6 +120,7 @@ pub async fn test(server: Arc, client: &mut Client) { ), ) .await; + tokio::time::sleep(Duration::from_millis(200)).await; for (account_id, num_messages) in [(&account_id_1, 3), (&account_id_2, 1), (&account_id_3, 1)] { assert_eq!( server @@ -151,6 +154,7 @@ pub async fn test(server: Arc, client: &mut Client) { 10, ) .await; + tokio::time::sleep(Duration::from_millis(200)).await; for (account_id, num_messages) in [(&account_id_1, 3), (&account_id_2, 2), (&account_id_3, 2)] { assert_eq!( server @@ -188,6 +192,7 @@ pub async fn test(server: Arc, client: &mut Client) { ), ) .await; + tokio::time::sleep(Duration::from_millis(200)).await; for (account_id, num_messages) in [(&account_id_1, 4), (&account_id_2, 3), (&account_id_3, 3)] { assert_eq!( server @@ -205,12 +210,6 @@ pub async fn test(server: Arc, client: &mut Client) { ); } - // Size checks - lmtp.send("MAIL FROM: SIZE=943718400").await; - lmtp.read(1, 5).await; - lmtp.send("BDAT 943718400").await; - lmtp.read(1, 5).await; - // Remove test data for account_id in [&account_id_1, &account_id_2, &account_id_3] { client.set_default_account_id(account_id); @@ -269,6 +268,7 @@ impl SmtpConnection { writer, }; conn.read(1, 2).await; + conn.lhlo().await; conn } diff --git a/tests/src/jmap/event_source.rs b/tests/src/jmap/event_source.rs index 8feaf1e9..8034d077 100644 --- a/tests/src/jmap/event_source.rs +++ b/tests/src/jmap/event_source.rs @@ -1,24 +1,22 @@ use std::{sync::Arc, time::Duration}; use futures::StreamExt; -use jmap::JMAP; -use jmap_client::{ - client::{Client, Credentials}, - event_source::Changes, - mailbox::Role, - TypeState, -}; +use jmap::{mailbox::INBOX_ID, JMAP}; +use jmap_client::{client::Client, event_source::Changes, mailbox::Role, TypeState}; use jmap_proto::types::id::Id; use store::ahash::AHashSet; use tokio::sync::mpsc; -use crate::jmap::{mailbox::destroy_all_mailboxes, test_account_create, test_account_login}; +use crate::jmap::{ + delivery::SmtpConnection, mailbox::destroy_all_mailboxes, test_account_create, + test_account_login, +}; pub async fn test(server: Arc, admin_client: &mut Client) { println!("Running EventSource tests..."); // Create test account - test_account_create(&server, "jdoe@example.com", "12345", "John Doe").await; + let account_id = test_account_create(&server, "jdoe@example.com", "12345", "John Doe").await; let mut client = test_account_login("jdoe@example.com", "12345").await; let mut changes = client @@ -59,8 +57,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { assert_ping(&mut event_rx).await; // Pings are only received in cfg(test) // Ingest email and expect state change - let implement = "true"; - /*let mut lmtp = SmtpConnection::connect().await; + let mut lmtp = SmtpConnection::connect().await; lmtp.ingest( "bill@example.com", &["jdoe@example.com"], @@ -86,18 +83,23 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ], ) .await; - assert_ping(&mut event_rx).await;*/ + assert_ping(&mut event_rx).await; // Destroy mailbox client.mailbox_destroy(&mailbox_id, true).await.unwrap(); + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; - /*assert_state( + // Destroy Inbox + admin_client.set_default_account_id(&account_id.to_string()); + admin_client + .mailbox_destroy(&Id::from(INBOX_ID).to_string(), true) + .await + .unwrap(); + assert_state( &mut event_rx, &[TypeState::Email, TypeState::Thread, TypeState::Mailbox], ) - .await;*/ - let fix = "true"; - assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + .await; assert_ping(&mut event_rx).await; assert_ping(&mut event_rx).await; diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 40bc7c69..66af6d50 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -5,7 +5,7 @@ use jmap_client::client::{Client, Credentials}; use jmap_proto::types::id::Id; use smtp::core::{SmtpSessionManager, SMTP}; use tokio::sync::{mpsc, watch}; -use utils::config::ServerProtocol; +use utils::{config::ServerProtocol, UnwrapFailure}; use crate::{add_test_certs, store::TempDir}; @@ -41,7 +41,7 @@ max-connections = 512 bind = ['127.0.0.1:11200'] greeting = 'Test LMTP instance' protocol = 'lmtp' -tls.implicit = true +tls.implicit = false [server.socket] reuse-addr = true @@ -51,6 +51,37 @@ enable = true implicit = false certificate = "default" +[session.ehlo] +reject-non-fqdn = false + +[session.rcpt.lookup] +domains = "list/domains" +addresses = "local" +vrfy = "local" +expn = "local" + +[session.rcpt.errors] +total = 5 +wait = "1ms" + +[list] +domains = ["example.com"] + +[queue] +path = "{TMP}" +hash = 64 + +[report] +path = "{TMP}" +hash = 64 + +[resolver] +type = "system" + +[queue.outbound] +next-hop = [ { if = "rcpt-domain", in-list = "list/domains", then = "local" }, + { else = false } ] + [store] db.path = "{TMP}/sqlite.db" blob.path = "{TMP}" @@ -93,10 +124,10 @@ uid-by-login = "SELECT ROWID - 1 FROM users WHERE login = ?" login-by-uid = "SELECT login FROM users WHERE ROWID - 1 = ?" secret-by-uid = "SELECT secret FROM users WHERE ROWID - 1 = ?" gids-by-uid = "SELECT gid FROM groups WHERE uid = ?" -uids-by-address = "SELECT uid FROM emails WHERE address = ?" -addresses-by-uid = "SELECT address FROM emails WHERE uid = ?" -vrfy = "SELECT address FROM emails WHERE address LIKE '%' || ? || '%' LIMIT 5" -expn = "SELECT address FROM emails WHERE address LIKE '%' || ? || '%' LIMIT 5" +uids-by-address = "SELECT uid FROM emails WHERE email = ?" +addresses-by-uid = "SELECT email FROM emails WHERE uid = ?" +vrfy = "SELECT email FROM emails WHERE email LIKE '%' || ? || '%' AND is_list = false LIMIT 5" +expn = "SELECT u.login FROM users u INNER JOIN emails e ON u.rowid -1 = e.uid WHERE e.email = ? AND e.is_list = true LIMIT 5" [oauth] key = "parerga_und_paralipomena" @@ -131,11 +162,11 @@ pub async fn jmap_tests() { //thread_get::test(params.server.clone(), &mut params.client).await; //thread_merge::test(params.server.clone(), &mut params.client).await; //mailbox::test(params.server.clone(), &mut params.client).await; - delivery::test(params.server.clone(), &mut params.client).await; + //delivery::test(params.server.clone(), &mut params.client).await; //auth_acl::test(params.server.clone(), &mut params.client).await; //auth_limits::test(params.server.clone(), &mut params.client).await; //auth_oauth::test(params.server.clone(), &mut params.client).await; - //event_source::test(params.server.clone(), &mut params.client).await; + event_source::test(params.server.clone(), &mut params.client).await; //push_subscription::test(params.server.clone(), &mut params.client).await; if delete { @@ -163,8 +194,12 @@ async fn init_jmap_tests(delete_if_exists: bool) -> JMAPTest { // Start JMAP and SMTP servers servers.bind(&config); let (delivery_tx, delivery_rx) = mpsc::channel(IPC_CHANNEL_BUFFER); - let smtp = SMTP::init(&config, &servers, delivery_tx).await; - let jmap = JMAP::init(&config, delivery_rx).await; + let smtp = SMTP::init(&config, &servers, delivery_tx) + .await + .failed("Invalid configuration file"); + let jmap = JMAP::init(&config, delivery_rx) + .await + .failed("Invalid configuration file"); let shutdown_tx = servers.spawn(|server, shutdown_rx| { match &server.protocol { ServerProtocol::Smtp | ServerProtocol::Lmtp => { @@ -181,7 +216,7 @@ async fn init_jmap_tests(delete_if_exists: bool) -> JMAPTest { for query in [ "CREATE TABLE users (login TEXT PRIMARY KEY, secret TEXT, name TEXT)", "CREATE TABLE groups (uid INTEGER, gid INTEGER, PRIMARY KEY (uid, gid))", - "CREATE TABLE emails (uid INTEGER NOT NULL, email TEXT NOT NULL, PRIMARY KEY (uid, email))", + "CREATE TABLE emails (uid INTEGER NOT NULL, email TEXT NOT NULL, is_list BOOLEAN DEFAULT 0, PRIMARY KEY (uid, email))", "INSERT INTO users (login, secret) VALUES ('admin', 'secret')", // RowID 0 is admin ] { assert!( @@ -290,14 +325,15 @@ pub async fn test_account_create(jmap: &JMAP, login: &str, secret: &str, name: & Id::new(uid) } -pub async fn test_alias_create(jmap: &JMAP, login: &str, alias: &str) { +pub async fn test_alias_create(jmap: &JMAP, login: &str, alias: &str, is_list: bool) { let uid = jmap.get_account_id(login).await.unwrap() as u64; assert!( jmap.auth_db .execute( &format!( - "INSERT OR REPLACE INTO emails (uid, email) VALUES ({}, ?)", - uid + "INSERT OR REPLACE INTO emails (uid, email, is_list) VALUES ({}, ?, {})", + uid, + if is_list { "true" } else { "false" } ), vec![alias.to_string()].into_iter() ) diff --git a/tests/src/smtp/inbound/scripts.rs b/tests/src/smtp/inbound/scripts.rs index 09a6ba9b..9a61696c 100644 --- a/tests/src/smtp/inbound/scripts.rs +++ b/tests/src/smtp/inbound/scripts.rs @@ -120,6 +120,13 @@ if envelope :domain :is "to" "foobar.org" { data = ''' require ["envelope", "reject", "variables", "replace", "mime", "foreverypart", "editheader", "extracttext", "enotify"]; +if envelope :localpart :is "to" "thomas" { + deleteheader "from"; + addheader "From" "no-reply@my.domain"; + redirect "redirect@here.email"; + discard; +} + if envelope :localpart :is "to" "bill" { reject "Bill cannot receive messages."; stop; @@ -313,6 +320,32 @@ async fn sieve_scripts() { .assert_contains("THIS IS A PIECE OF HTML TEXT") .assert_not_contains("X-My-Header: true"); + // Expect a modified redirected message + session + .send_message( + "test@example.net", + &["thomas@foobar.gov"], + "test:no_dkim", + "250", + ) + .await; + let redirect = qr.read_event().await.unwrap_message(); + assert_eq!(redirect.return_path, ""); + assert_eq!(redirect.recipients.len(), 1); + assert_eq!( + redirect.recipients.first().unwrap().address, + "redirect@here.email" + ); + + redirect + .read_lines() + .assert_contains("From: no-reply@my.domain") + .assert_contains("To: Suzie Q ") + .assert_contains("Subject: Is dinner ready?") + .assert_contains("Message-ID: <20030712040037.46341.5F8J@football.example.com>") + .assert_not_contains("From: Joe SixPack "); + qr.assert_empty_queue(); + // Test pipes session.data.remote_ip = "10.0.0.123".parse().unwrap(); session