diff --git a/Cargo.lock b/Cargo.lock index 66e1dd23..197ae579 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1748,6 +1748,7 @@ dependencies = [ "serde_json", "sha2", "sieve-rs", + "smtp", "sqlx", "store", "tokio", @@ -3798,6 +3799,7 @@ dependencies = [ "async-trait", "base64 0.21.0", "bytes", + "chrono", "csv", "dashmap", "ece", diff --git a/crates/jmap-proto/src/types/keyword.rs b/crates/jmap-proto/src/types/keyword.rs index 9dd12f88..a1e5c6bf 100644 --- a/crates/jmap-proto/src/types/keyword.rs +++ b/crates/jmap-proto/src/types/keyword.rs @@ -102,6 +102,42 @@ impl JsonObjectParser for Keyword { } } +impl From for Keyword { + fn from(value: String) -> Self { + if value.starts_with('$') { + let mut hash = 0; + let mut shift = 0; + + for &ch in value.as_bytes() { + if shift < 128 { + hash |= (ch as u128) << shift; + shift += 8; + } else { + break; + } + } + + match hash { + 0x6e65_6573 => return Keyword::Seen, + 0x0074_6661_7264 => return Keyword::Draft, + 0x0064_6567_6761_6c66 => return Keyword::Flagged, + 0x6465_7265_7773_6e61 => return Keyword::Answered, + 0x746e_6563_6572 => return Keyword::Recent, + 0x0074_6e61_7472_6f70_6d69 => return Keyword::Important, + 0x676e_6968_7369_6870 => return Keyword::Phishing, + 0x6b6e_756a => return Keyword::Junk, + 0x006b_6e75_6a74_6f6e => return Keyword::NotJunk, + 0x0064_6574_656c_6564 => return Keyword::Deleted, + 0x0064_6564_7261_7772_6f66 => return Keyword::Forwarded, + 0x0074_6e65_736e_646d => return Keyword::MdnSent, + _ => (), + } + } + + Keyword::Other(value) + } +} + impl Display for Keyword { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index 1606c33a..2450163a 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -8,6 +8,7 @@ resolver = "2" store = { path = "../store" } jmap_proto = { path = "../jmap-proto" } utils = { path = "../utils" } +smtp = { path = "../smtp" } mail-parser = { git = "https://github.com/stalwartlabs/mail-parser", features = ["full_encoding", "serde_support", "ludicrous_mode"] } mail-builder = { git = "https://github.com/stalwartlabs/mail-builder", features = ["ludicrous_mode"] } mail-send = { git = "https://github.com/stalwartlabs/mail-send" } diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index ff2a4d47..13e82b32 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -132,7 +132,7 @@ impl JMAP { get::RequestArguments::SieveScript => { acl_token.assert_is_member(req.account_id)?; - self.sieve_script_get(req, acl_token).await?.into() + self.sieve_script_get(req).await?.into() } get::RequestArguments::VacationResponse => { acl_token.assert_is_member(req.account_id)?; @@ -190,7 +190,7 @@ impl JMAP { set::RequestArguments::VacationResponse => { acl_token.assert_is_member(req.account_id)?; - self.vacation_response_set(req, acl_token).await?.into() + self.vacation_response_set(req).await?.into() } }, RequestMethod::Changes(req) => self.changes(req, acl_token).await?.into(), diff --git a/crates/jmap/src/auth/account.rs b/crates/jmap/src/auth/account.rs index b9db690d..7495e67a 100644 --- a/crates/jmap/src/auth/account.rs +++ b/crates/jmap/src/auth/account.rs @@ -42,6 +42,20 @@ impl JMAP { } } + pub async fn get_account_name(&self, account_id: u32) -> Option { + match &self.auth_db { + AuthDatabase::Sql { + db, + query_name_by_uid, + .. + } => { + db.fetch_uid_to_string(query_name_by_uid, account_id as i64) + .await + } + AuthDatabase::Ldap => None, + } + } + pub async fn get_account_id(&self, account: &str) -> Option { match &self.auth_db { AuthDatabase::Sql { diff --git a/crates/jmap/src/auth/mod.rs b/crates/jmap/src/auth/mod.rs index 0af311be..a031dc8f 100644 --- a/crates/jmap/src/auth/mod.rs +++ b/crates/jmap/src/auth/mod.rs @@ -24,6 +24,7 @@ pub enum AuthDatabase { query_uid_by_login: String, query_login_by_uid: String, query_secret_by_uid: String, + query_name_by_uid: String, query_gids_by_uid: String, query_uids_by_address: String, query_addresses_by_uid: String, diff --git a/crates/jmap/src/blob/copy.rs b/crates/jmap/src/blob/copy.rs index 7df1a33d..9685840f 100644 --- a/crates/jmap/src/blob/copy.rs +++ b/crates/jmap/src/blob/copy.rs @@ -30,7 +30,14 @@ impl JMAP { let dest_blob_id = BlobId::temporary(account_id); match self .store - .copy_blob(&blob_id.kind, &dest_blob_id.kind) + .copy_blob( + &blob_id.kind, + &dest_blob_id.kind, + blob_id + .section + .as_ref() + .map(|s| (s.offset_start as u32)..((s.offset_start + s.size) as u32)), + ) .await { Ok(success) => { diff --git a/crates/jmap/src/email/copy.rs b/crates/jmap/src/email/copy.rs index 4da4d4ed..ec95bc9b 100644 --- a/crates/jmap/src/email/copy.rs +++ b/crates/jmap/src/email/copy.rs @@ -280,6 +280,7 @@ impl JMAP { document_id: from_message_id, }, &email.blob_id.kind, + None, ) .await .map_err(|err| { diff --git a/crates/jmap/src/email/import.rs b/crates/jmap/src/email/import.rs index 4fcc14ca..48d5f807 100644 --- a/crates/jmap/src/email/import.rs +++ b/crates/jmap/src/email/import.rs @@ -99,7 +99,7 @@ impl JMAP { // Import message match self .email_ingest( - &raw_message, + (&raw_message).into(), account_id, mailbox_ids, email.keywords, @@ -111,7 +111,7 @@ impl JMAP { Ok(email) => { response.created.append(id, email.into()); } - Err(MaybeError::Permanent(reason)) => { + Err(MaybeError::Permanent { reason, .. }) => { response.not_created.append( id, SetError::new(SetErrorType::InvalidEmail).with_description(reason), diff --git a/crates/jmap/src/email/ingest.rs b/crates/jmap/src/email/ingest.rs index eaff4344..4808a723 100644 --- a/crates/jmap/src/email/ingest.rs +++ b/crates/jmap/src/email/ingest.rs @@ -31,11 +31,16 @@ pub struct IngestedEmail { pub size: usize, } +pub struct IngestEmail<'x> { + pub raw_message: &'x [u8], + pub message: Option>, +} + impl JMAP { #[allow(clippy::blocks_in_if_conditions)] pub async fn email_ingest( &self, - raw_message: &[u8], + ingest_email: IngestEmail<'_>, account_id: u32, mailbox_ids: Vec, keywords: Vec, @@ -43,8 +48,11 @@ impl JMAP { skip_duplicates: bool, ) -> Result { // Parse message - let message = Message::parse(raw_message) - .ok_or_else(|| MaybeError::Permanent("Failed to parse e-mail message.".to_string()))?; + let raw_message = ingest_email.raw_message; + let message = ingest_email.message.ok_or_else(|| MaybeError::Permanent { + code: [5, 5, 0], + reason: "Failed to parse e-mail message.".to_string(), + })?; // Obtain message references and thread name let mut references = Vec::with_capacity(5); @@ -398,6 +406,33 @@ impl JMAP { } } +impl<'x> From<&'x [u8]> for IngestEmail<'x> { + fn from(raw_message: &'x [u8]) -> Self { + IngestEmail { + raw_message, + message: Message::parse(raw_message), + } + } +} + +impl<'x> From<&'x Vec> for IngestEmail<'x> { + fn from(raw_message: &'x Vec) -> Self { + IngestEmail { + raw_message, + message: Message::parse(raw_message), + } + } +} + +impl<'x> IngestEmail<'x> { + pub fn new(raw_message: &'x [u8], message: Message<'x>) -> Self { + IngestEmail { + raw_message, + message: message.into(), + } + } +} + impl From for Object { fn from(email: IngestedEmail) -> Self { Object::with_capacity(3) diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index fa5a9307..9542a9cd 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -697,7 +697,7 @@ impl JMAP { response.created.insert( id, self.email_ingest( - &raw_message, + (&raw_message).into(), account_id, mailboxes, keywords, diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index 2df06e6b..c6935b13 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -20,13 +20,14 @@ use services::{ delivery::spawn_delivery_manager, state::{self, init_state_manager, spawn_state_manager}, }; +use smtp::{core::SMTP, queue}; use sqlx::{mysql::MySqlPoolOptions, postgres::PgPoolOptions, sqlite::SqlitePoolOptions}; use store::{ fts::Language, parking_lot::Mutex, query::{sort::Pagination, Comparator, Filter, ResultSet, SortedResultSet}, roaring::RoaringBitmap, - write::{BatchBuilder, BitmapFamily}, + write::{BatchBuilder, BitmapFamily, ToBitmaps}, BitmapKey, Deserialize, Serialize, Store, ValueKey, }; use tokio::sync::mpsc; @@ -61,6 +62,7 @@ pub struct JMAP { pub auth_db: AuthDatabase, pub state_tx: mpsc::Sender, + pub smtp: Arc, pub sieve_compiler: Compiler, pub sieve_runtime: Runtime, @@ -115,13 +117,14 @@ pub struct Bincode { pub enum MaybeError { Temporary, - Permanent(String), + Permanent { code: [u8; 3], reason: String }, } impl JMAP { pub async fn init( config: &utils::config::Config, delivery_rx: mpsc::Receiver, + smtp: Arc, ) -> Result, String> { let auth_db = match config.value_require("jmap.auth.database.type")? { "ldap" => AuthDatabase::Ldap, @@ -186,6 +189,9 @@ impl JMAP { query_secret_by_uid: config .value_require("jmap.auth.database.query.secret-by-uid")? .to_string(), + query_name_by_uid: config + .value_require("jmap.auth.database.query.name-by-uid")? + .to_string(), query_gids_by_uid: config .value_require("jmap.auth.database.query.gids-by-uid")? .to_string(), @@ -233,6 +239,7 @@ impl JMAP { ), auth_db, state_tx, + smtp, sieve_compiler: Compiler::new() .with_max_script_size( config @@ -700,6 +707,12 @@ impl De } } +impl ToBitmaps for Bincode { + fn to_bitmaps(&self, _ops: &mut Vec, _field: u8, _set: bool) { + unreachable!() + } +} + trait UpdateResults: Sized { fn update_results(&mut self, sorted_results: SortedResultSet) -> Result<(), MethodError>; } diff --git a/crates/jmap/src/mailbox/get.rs b/crates/jmap/src/mailbox/get.rs index 2bfa651c..315ab6fc 100644 --- a/crates/jmap/src/mailbox/get.rs +++ b/crates/jmap/src/mailbox/get.rs @@ -4,7 +4,7 @@ use jmap_proto::{ object::Object, types::{acl::Acl, collection::Collection, keyword::Keyword, property::Property, value::Value}, }; -use store::{ahash::AHashSet, roaring::RoaringBitmap}; +use store::{ahash::AHashSet, query::Filter, roaring::RoaringBitmap}; use crate::{ auth::{acl::EffectiveAcl, AclToken}, @@ -294,4 +294,115 @@ impl JMAP { Ok(None) } } + + pub async fn mailbox_expand_path<'x>( + &self, + account_id: u32, + path: &'x str, + exact_match: bool, + ) -> Result>, MethodError> { + let path = path + .split('/') + .filter_map(|p| { + let p = p.trim(); + if !p.is_empty() { + p.into() + } else { + None + } + }) + .collect::>(); + if path.is_empty() || path.len() > self.config.mailbox_max_depth { + return Ok(None); + } + + let mut filter = Vec::with_capacity(path.len() + 2); + filter.push(Filter::Or); + for &item in &path { + filter.push(Filter::eq(Property::Name, item)); + } + filter.push(Filter::End); + + let document_ids = self + .filter(account_id, Collection::Mailbox, filter) + .await? + .results; + if exact_match && (document_ids.len() as usize) < path.len() { + return Ok(None); + } + + let mut found_names = Vec::new(); + for document_id in document_ids { + if let Some(mut obj) = self + .get_property::>( + account_id, + Collection::Mailbox, + document_id, + Property::Value, + ) + .await? + { + if let Some(Value::Text(value)) = obj.properties.remove(&Property::Name) { + found_names.push(( + value, + if let Some(Value::Id(value)) = obj.properties.remove(&Property::ParentId) { + value.document_id() + } else { + 0 + }, + document_id + 1, + )); + } else { + return Ok(None); + } + } else { + return Ok(None); + } + } + + Ok(Some(ExpandPath { path, found_names })) + } + + pub async fn mailbox_get_by_name( + &self, + account_id: u32, + path: &str, + ) -> Result, MethodError> { + Ok(self + .mailbox_expand_path(account_id, path, true) + .await? + .and_then(|ep| { + let mut next_parent_id = 0; + 'outer: for name in ep.path { + for (part, parent_id, document_id) in &ep.found_names { + if part.eq(name) && *parent_id == next_parent_id { + next_parent_id = *document_id; + continue 'outer; + } + } + return None; + } + Some(next_parent_id - 1) + })) + } + + pub async fn mailbox_get_by_role( + &self, + account_id: u32, + role: &str, + ) -> Result, MethodError> { + self.filter( + account_id, + Collection::Mailbox, + vec![Filter::eq(Property::Role, role)], + ) + .await + .map(|r| r.results.min()) + } +} + +#[derive(Debug)] +pub struct ExpandPath<'x> { + pub path: Vec<&'x str>, + pub found_names: Vec<(String, u32, u32)>, } diff --git a/crates/jmap/src/mailbox/set.rs b/crates/jmap/src/mailbox/set.rs index ffdcc44e..9e038a63 100644 --- a/crates/jmap/src/mailbox/set.rs +++ b/crates/jmap/src/mailbox/set.rs @@ -796,4 +796,65 @@ impl JMAP { Ok(mailbox_ids) } + + pub async fn mailbox_create_path( + &self, + account_id: u32, + path: &str, + ) -> Result)>, MethodError> { + let expanded_path = + if let Some(expand_path) = self.mailbox_expand_path(account_id, path, false).await? { + expand_path + } else { + return Ok(None); + }; + + let mut next_parent_id = 0; + let mut path = expanded_path.path.into_iter().peekable(); + 'outer: while let Some(name) = path.peek() { + for (part, parent_id, document_id) in &expanded_path.found_names { + if part.eq(name) && *parent_id == next_parent_id { + next_parent_id = *document_id; + path.next(); + continue 'outer; + } + } + break; + } + + // Create missing folders + if path.peek().is_some() { + let mut batch = BatchBuilder::new(); + let mut changes = self.begin_changes(account_id).await?; + batch + .with_account_id(account_id) + .with_collection(Collection::Mailbox); + + for name in path { + if name.len() > self.config.mailbox_name_max_len { + return Ok(None); + } + + let document_id = self + .assign_document_id(account_id, Collection::Mailbox) + .await?; + batch.create_document(document_id).custom( + ObjectIndexBuilder::new(SCHEMA).with_changes( + Object::with_capacity(2) + .with_property(Property::Name, name) + .with_property(Property::ParentId, Value::Id(Id::from(next_parent_id))), + ), + ); + changes.log_insert(Collection::Mailbox, document_id); + next_parent_id = document_id + 1; + } + let change_id = changes.change_id; + batch.custom(changes); + self.write_batch(batch).await?; + + Ok(Some((next_parent_id - 1, Some(change_id)))) + } else { + Ok(Some((next_parent_id - 1, None))) + } + } } diff --git a/crates/jmap/src/services/ingest.rs b/crates/jmap/src/services/ingest.rs index 4828762d..73b1be2c 100644 --- a/crates/jmap/src/services/ingest.rs +++ b/crates/jmap/src/services/ingest.rs @@ -21,20 +21,48 @@ impl JMAP { // Obtain the UIDs for each recipient let mut recipients = Vec::with_capacity(message.recipients.len()); let mut deliver_uids = AHashMap::with_capacity(message.recipients.len()); - for rcpt in message.recipients { - let uids = self.get_uids_by_address(&rcpt).await; + for rcpt in &message.recipients { + let uids = self.get_uids_by_address(rcpt).await; for uid in &uids { - deliver_uids.insert(*uid, DeliveryResult::Success); + deliver_uids.insert(*uid, (DeliveryResult::Success, rcpt)); } recipients.push(uids); } // Deliver to each recipient - for (uid, status) in &mut deliver_uids { - match self - .email_ingest(&raw_message, *uid, vec![INBOX_ID], vec![], None, true) - .await - { + for (uid, (status, rcpt)) in &mut deliver_uids { + // Check if there is an active sieve script + let result = match self.sieve_script_get_active(*uid).await { + Ok(Some(active_script)) => { + self.sieve_script_ingest( + &raw_message, + &message.sender_address, + rcpt, + *uid, + active_script, + ) + .await + } + Ok(None) => { + self.email_ingest( + (&raw_message).into(), + *uid, + vec![INBOX_ID], + vec![], + None, + true, + ) + .await + } + Err(_) => { + *status = DeliveryResult::TemporaryFailure { + reason: "Transient server failure.".into(), + }; + continue; + } + }; + + match result { Ok(ingested_message) => { // Notify state change if ingested_message.change_id != u64::MAX { @@ -54,9 +82,9 @@ impl JMAP { reason: "Transient server failure.".into(), } } - MaybeError::Permanent(reason) => { + MaybeError::Permanent { code, reason } => { *status = DeliveryResult::PermanentFailure { - code: [5, 5, 0], + code, reason: reason.into(), } } @@ -71,7 +99,7 @@ impl JMAP { match uids.len() { 1 => { // Delivery to single recipient - deliver_uids.get(&uids[0]).unwrap().clone() + deliver_uids.get(&uids[0]).unwrap().0.clone() } 0 => { // Something went wrong @@ -84,7 +112,7 @@ impl JMAP { let mut success = 0; let mut temp_failures = 0; for uid in uids { - match deliver_uids.get(&uid).unwrap() { + match deliver_uids.get(&uid).unwrap().0 { DeliveryResult::Success => success += 1, DeliveryResult::TemporaryFailure { .. } => temp_failures += 1, DeliveryResult::PermanentFailure { .. } => {} diff --git a/crates/jmap/src/sieve/get.rs b/crates/jmap/src/sieve/get.rs index 8cb924c3..c1413199 100644 --- a/crates/jmap/src/sieve/get.rs +++ b/crates/jmap/src/sieve/get.rs @@ -9,7 +9,7 @@ use jmap_proto::{ use sieve::Sieve; use store::{query::Filter, BlobKind, Deserialize, Serialize}; -use crate::{auth::AclToken, sieve::SeenIds, Bincode, JMAP}; +use crate::{sieve::SeenIds, Bincode, JMAP}; use super::ActiveScript; @@ -17,12 +17,11 @@ 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 account_id = request.account_id.document_id(); let push_ids = self .get_document_ids(account_id, Collection::SieveScript) .await? @@ -113,9 +112,16 @@ impl JMAP { .results .min() { + let (script, mut script_object) = + self.sieve_script_compile(account_id, document_id).await?; Ok(Some(ActiveScript { document_id, - script: Arc::new(self.sieve_script_compile(account_id, document_id).await?), + script: Arc::new(script), + script_name: script_object + .properties + .remove(&Property::Name) + .and_then(|name| name.try_unwrap_string()) + .unwrap_or_else(|| account_id.to_string()), seen_ids: self .get_property::>( account_id, @@ -150,7 +156,7 @@ impl JMAP { { self.sieve_script_compile(account_id, document_id) .await - .map(Some) + .map(|(sieve, _)| Some(sieve)) } else { Ok(None) } @@ -160,9 +166,9 @@ impl JMAP { &self, account_id: u32, document_id: u32, - ) -> Result { - // Obtain the sieve script length - let script_offset = self + ) -> Result<(Sieve, Object), MethodError> { + // Obtain script object + let script_object = self .get_property::>( account_id, Collection::SieveScript, @@ -170,7 +176,22 @@ impl JMAP { Property::Value, ) .await? - .and_then(|mut object| object.properties.remove(&Property::BlobId)) + .ok_or_else(|| { + tracing::warn!( + context = "sieve_script_compile", + event = "error", + account_id = account_id, + document_id = document_id, + "Failed to obtain sieve script object" + ); + + MethodError::ServerPartialFail + })?; + + // Obtain the sieve script length + let script_offset = script_object + .properties + .get(&Property::BlobId) .and_then(|value| value.as_uint()) .ok_or_else(|| { tracing::warn!( @@ -202,7 +223,7 @@ impl JMAP { .get(script_offset..) .and_then(|bytes| Bincode::::deserialize(bytes).ok()) { - Ok(sieve.inner) + Ok((sieve.inner, script_object)) } else { // Deserialization failed, probably because the script compiler version changed match self @@ -237,7 +258,7 @@ impl JMAP { ) .await; - Ok(sieve.inner) + Ok((sieve.inner, script_object)) } Err(error) => { tracing::warn!( diff --git a/crates/jmap/src/sieve/ingest.rs b/crates/jmap/src/sieve/ingest.rs new file mode 100644 index 00000000..fdbb2c1d --- /dev/null +++ b/crates/jmap/src/sieve/ingest.rs @@ -0,0 +1,440 @@ +use std::borrow::Cow; + +use jmap_proto::types::{collection::Collection, id::Id, keyword::Keyword, property::Property}; +use mail_parser::Message; +use sieve::{Envelope, Event, Input, Mailbox, Recipient}; +use store::{ + ahash::AHashSet, + write::{now, BatchBuilder, F_VALUE}, +}; + +use crate::{ + email::ingest::{IngestEmail, IngestedEmail}, + mailbox::{INBOX_ID, TRASH_ID}, + sieve::SeenIdHash, + Bincode, MaybeError, JMAP, +}; + +use super::ActiveScript; + +struct OutgoingMessage { + pub mail_from: String, + pub rcpt_to: Vec, + pub message: Vec, +} + +struct SieveMessage<'x> { + pub raw_message: Cow<'x, [u8]>, + pub file_into: Vec, + pub flags: Vec, +} + +impl JMAP { + #[allow(clippy::blocks_in_if_conditions)] + pub async fn sieve_script_ingest( + &self, + raw_message: &[u8], + envelope_from: &str, + envelope_to: &str, + account_id: u32, + mut active_script: ActiveScript, + ) -> Result { + // Parse message + let message = if let Some(message) = Message::parse(raw_message) { + message + } else { + return Err(MaybeError::Permanent { + code: [5, 5, 0], + reason: "Failed to parse message.".to_string(), + }); + }; + + // Obtain mailboxIds + let mailbox_ids = self + .mailbox_get_or_create(account_id) + .await + .map_err(|_| MaybeError::Temporary)?; + + // Create Sieve instance + let mut instance = self.sieve_runtime.filter_parsed(message); + + // Obtain mail from address + let mail_from = if let Some(email) = self + .get_addresses_by_uid(account_id) + .await + .into_iter() + .next() + { + email + } else { + envelope_to.to_string() + }; + + // Set account address + instance.set_user_address(&mail_from); + + // Set account name + if let Some(name) = self.get_account_name(account_id).await { + instance.set_user_full_name(&name); + } + + // Set envelope + instance.set_envelope(Envelope::From, envelope_from); + instance.set_envelope(Envelope::To, envelope_to); + + let mut input = Input::script(active_script.script_name, active_script.script.clone()); + + let mut do_discard = false; + let mut do_deliver = false; + + let mut new_ids = AHashSet::new(); + let mut reject_reason = None; + let mut messages: Vec = vec![SieveMessage { + raw_message: raw_message.into(), + file_into: Vec::new(), + flags: Vec::new(), + }]; + let mut outgoing_messages = Vec::new(); + let now = now(); + let mut ingested_message = IngestedEmail { + id: Id::default(), + change_id: u64::MAX, + blob_id: Default::default(), + size: raw_message.len(), + }; + + while let Some(event) = instance.run(input) { + match event { + Ok(event) => match event { + Event::IncludeScript { name, .. } => { + if let Ok(Some(script)) = + self.sieve_script_get_by_name(account_id, &name).await + { + input = Input::script(name, script); + } else { + input = false.into(); + } + } + Event::MailboxExists { + mailboxes, + special_use, + } => { + if !mailboxes.is_empty() { + let mut special_use_ids = Vec::with_capacity(special_use.len()); + for role in special_use { + special_use_ids.push(if role.eq_ignore_ascii_case("inbox") { + INBOX_ID + } else if role.eq_ignore_ascii_case("trash") { + TRASH_ID + } else { + let mut mailbox_id = u32::MAX; + let role = role.to_ascii_lowercase(); + if is_valid_role(&role) { + if let Ok(Some(mailbox_id_)) = + self.mailbox_get_by_role(account_id, &role).await + { + mailbox_id = mailbox_id_; + } + } + mailbox_id + }); + } + + let mut result = true; + for mailbox in mailboxes { + match mailbox { + Mailbox::Name(name) => { + if !matches!( + self.mailbox_get_by_name(account_id, &name).await, + Ok(Some(document_id)) if special_use_ids.is_empty() || + special_use_ids.contains(&document_id) + ) { + result = false; + break; + } + } + Mailbox::Id(id) => { + if !matches!(Id::from_bytes(id.as_bytes()), Some(id) if + mailbox_ids.contains(id.document_id()) && + (special_use_ids.is_empty() || + special_use_ids.contains(&id.document_id()))) + { + result = false; + break; + } + } + } + } + input = result.into(); + } else if !special_use.is_empty() { + let mut result = true; + + for role in special_use { + if !role.eq_ignore_ascii_case("inbox") + && !role.eq_ignore_ascii_case("trash") + { + let role = role.to_ascii_lowercase(); + if !is_valid_role(&role) + || !matches!( + self.mailbox_get_by_role(account_id, &role).await, + Ok(Some(_)) + ) + { + result = false; + break; + } + } + } + input = result.into(); + } else { + input = false.into(); + } + } + Event::DuplicateId { id, expiry, last } => { + let id_hash = SeenIdHash::new(&id, expiry + now); + let seen_id = active_script.seen_ids.ids.contains(&id_hash); + if !seen_id || last { + new_ids.insert(id_hash); + } + + input = seen_id.into(); + } + Event::Discard => { + do_discard = true; + input = true.into(); + } + Event::Reject { reason, .. } => { + reject_reason = reason.into(); + do_discard = true; + input = true.into(); + } + Event::Keep { flags, message_id } => { + if let Some(message) = messages.get_mut(message_id) { + message.flags = flags.into_iter().map(Keyword::from).collect(); + if !message.file_into.contains(&INBOX_ID) { + message.file_into.push(INBOX_ID); + } + do_deliver = true; + } else { + tracing::error!( + "Sieve filter failed: Unknown message id {}.", + message_id + ); + } + input = true.into(); + } + Event::FileInto { + folder, + flags, + mailbox_id, + special_use, + create, + message_id, + } => { + let mut target_id = u32::MAX; + + // Find mailbox by Id + if let Some(mailbox_id) = + mailbox_id.and_then(|m| Id::from_bytes(m.as_bytes())) + { + let mailbox_id = mailbox_id.document_id(); + if mailbox_ids.contains(mailbox_id) { + target_id = mailbox_id; + } + } + + // Find mailbox by role + if let Some(special_use) = special_use { + if target_id == u32::MAX { + if special_use.eq_ignore_ascii_case("inbox") { + target_id = INBOX_ID; + } else if special_use.eq_ignore_ascii_case("trash") { + target_id = TRASH_ID; + } else { + let role = special_use.to_ascii_lowercase(); + if is_valid_role(&role) { + if let Ok(Some(mailbox_id_)) = + self.mailbox_get_by_role(account_id, &role).await + { + target_id = mailbox_id_; + } + } + } + } + } + + // Find mailbox by name + if target_id == u32::MAX { + if !create { + if let Ok(Some(document_id)) = + self.mailbox_get_by_name(account_id, &folder).await + { + target_id = document_id; + } + } else if let Ok(Some((document_id, changes))) = + self.mailbox_create_path(account_id, &folder).await + { + target_id = document_id; + if let Some(change_id) = changes { + ingested_message.change_id = change_id; + } + } + } + + // Default to Inbox + if target_id == u32::MAX { + target_id = INBOX_ID; + } + + if let Some(message) = messages.get_mut(message_id) { + message.flags = flags.into_iter().map(Keyword::from).collect(); + if !message.file_into.contains(&target_id) { + message.file_into.push(target_id); + } + do_deliver = true; + } else { + tracing::error!( + "Sieve filter failed: Unknown message id {}.", + message_id + ); + } + input = true.into(); + } + Event::SendMessage { + recipient, + message_id, + .. + } => { + input = true.into(); + + outgoing_messages.push(OutgoingMessage { + mail_from: mail_from.clone(), + rcpt_to: match recipient { + Recipient::Address(rcpt) => vec![rcpt], + Recipient::Group(rcpts) => rcpts, + Recipient::List(_) => { + // Not yet implemented + continue; + } + }, + message: if let Some(message) = messages.get(message_id) { + message.raw_message.to_vec() + } else { + tracing::error!( + "Sieve filter failed: Unknown message id {}.", + message_id + ); + continue; + }, + }); + } + Event::ListContains { .. } | Event::Execute { .. } | Event::Notify { .. } => { + // Not allowed + input = false.into(); + } + Event::CreatedMessage { message, .. } => { + messages.push(SieveMessage { + raw_message: message.into(), + file_into: Vec::new(), + flags: Vec::new(), + }); + input = true.into(); + } + #[allow(unreachable_patterns)] + _ => unreachable!(), + }, + + #[cfg(feature = "test_mode")] + Err(sieve::runtime::RuntimeError::ScriptErrorMessage(err)) => { + panic!("Sieve test failed: {}", err); + } + + Err(err) => { + tracing::debug!("Sieve script runtime error: {}", err); + input = true.into(); + } + } + } + + let coco = "send outgoing"; + + // Fail-safe, no discard and no keep seen, assume that something went wrong and file anyway. + if !do_deliver && !do_discard { + messages[0].file_into.push(INBOX_ID); + } + + // Deliver messages + let mut has_temp_errors = false; + let mut has_delivered = false; + for (message_id, sieve_message) in messages.into_iter().enumerate() { + if !sieve_message.file_into.is_empty() { + // Parse message if needed + let message = if message_id == 0 && !instance.has_message_changed() { + instance.take_message() + } else if let Some(message) = Message::parse(&sieve_message.raw_message) { + message + } else { + tracing::debug!("Failed to parse Sieve generated message."); + continue; + }; + + // Deliver message + match self + .email_ingest( + IngestEmail::new(&sieve_message.raw_message, message), + account_id, + sieve_message.file_into, + sieve_message.flags, + None, + true, + ) + .await + { + Ok(ingested_message_) => { + has_delivered = true; + ingested_message = ingested_message_; + } + Err(_) => { + has_temp_errors = true; + } + } + } + } + + // Save new ids script changes + if !new_ids.is_empty() || active_script.seen_ids.has_changes { + active_script.seen_ids.ids.extend(new_ids); + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::SieveScript) + .update_document(active_script.document_id) + .value( + Property::EmailIds, + Bincode::new(active_script.seen_ids), + F_VALUE, + ); + let _ = self.write_batch(batch).await; + } + + if let Some(reject_reason) = reject_reason { + Err(MaybeError::Permanent { + code: [5, 7, 1], + reason: reject_reason, + }) + } else if has_delivered || !has_temp_errors { + Ok(ingested_message) + } else { + // There were problems during delivery + Err(MaybeError::Temporary) + } + } +} + +#[inline(always)] +pub fn is_valid_role(role: &str) -> bool { + [ + "inbox", "trash", "spam", "junk", "drafts", "archive", "sent", + ] + .contains(&role) +} diff --git a/crates/jmap/src/sieve/mod.rs b/crates/jmap/src/sieve/mod.rs index d729675e..6880d098 100644 --- a/crates/jmap/src/sieve/mod.rs +++ b/crates/jmap/src/sieve/mod.rs @@ -5,12 +5,14 @@ use sieve::Sieve; use store::{ahash::AHashSet, blake3, write::now}; pub mod get; +pub mod ingest; pub mod query; pub mod set; pub mod validate; pub struct ActiveScript { pub document_id: u32, + pub script_name: String, pub script: Arc, pub seen_ids: SeenIds, } diff --git a/crates/jmap/src/sieve/query.rs b/crates/jmap/src/sieve/query.rs index 769d8593..33b08012 100644 --- a/crates/jmap/src/sieve/query.rs +++ b/crates/jmap/src/sieve/query.rs @@ -28,7 +28,7 @@ impl JMAP { Language::None, )), Filter::IsActive(is_active) => { - filters.push(query::Filter::lt(Property::IsActive, is_active as u32)) + filters.push(query::Filter::eq(Property::IsActive, is_active as u32)) } other => return Err(MethodError::UnsupportedFilter(other.to_string())), } @@ -46,7 +46,7 @@ impl JMAP { for comparator in request .sort .and_then(|s| if !s.is_empty() { s.into() } else { None }) - .unwrap_or_else(|| vec![Comparator::descending(SortProperty::ReceivedAt)]) + .unwrap_or_else(|| vec![Comparator::descending(SortProperty::Name)]) { comparators.push(match comparator.property { SortProperty::Name => { diff --git a/crates/jmap/src/sieve/set.rs b/crates/jmap/src/sieve/set.rs index 1c9bde1e..7dc34e50 100644 --- a/crates/jmap/src/sieve/set.rs +++ b/crates/jmap/src/sieve/set.rs @@ -52,7 +52,7 @@ impl JMAP { mut request: SetRequest, acl_token: &AclToken, ) -> Result { - let account_id = acl_token.primary_id(); + let account_id = request.account_id.document_id(); let mut sieve_ids = self .get_document_ids(account_id, Collection::SieveScript) .await? diff --git a/crates/jmap/src/vacation/set.rs b/crates/jmap/src/vacation/set.rs index 9c31c914..591d1238 100644 --- a/crates/jmap/src/vacation/set.rs +++ b/crates/jmap/src/vacation/set.rs @@ -22,15 +22,14 @@ use store::{ BlobKind, }; -use crate::{auth::AclToken, sieve::set::SCHEMA, JMAP}; +use crate::{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 account_id = request.account_id.document_id(); let mut response = self .prepare_set_response(&request, Collection::SieveScript) .await?; diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index a942d618..6bf85b1f 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -28,7 +28,7 @@ async fn main() -> std::io::Result<()> { let smtp = SMTP::init(&config, &servers, delivery_tx) .await .failed("Invalid configuration file"); - let jmap = JMAP::init(&config, delivery_rx) + let jmap = JMAP::init(&config, delivery_rx, smtp.clone()) .await .failed("Invalid configuration file"); diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index a568e34c..bc7c1113 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -353,3 +353,94 @@ impl PartialOrd for SessionAddress { } } } + +#[cfg(feature = "local_delivery")] +pub struct NullIo(); + +#[cfg(feature = "local_delivery")] +impl AsyncWrite for NullIo { + fn poll_write( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &[u8], + ) -> std::task::Poll> { + unreachable!() + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + unreachable!() + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + unreachable!() + } +} + +#[cfg(feature = "local_delivery")] +impl AsyncRead for NullIo { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + unreachable!() + } +} + +#[cfg(feature = "local_delivery")] +impl Session { + pub fn local( + core: std::sync::Arc, + instance: std::sync::Arc, + data: SessionData, + ) -> Self { + Session { + state: State::None, + instance, + core, + span: tracing::info_span!( + "local_delivery", + "return_path" = if let Some(mail_from) = &data.mail_from { + mail_from.address_lcase.as_str() + } else { + "<>" + }, + "nrcpt" = data.rcpt_to.len(), + "size" = data.message.len(), + ), + stream: NullIo(), + data, + params: SessionParameters { + timeout: Default::default(), + ehlo_require: Default::default(), + ehlo_reject_non_fqdn: Default::default(), + auth_lookup: Default::default(), + auth_require: Default::default(), + auth_errors_max: Default::default(), + auth_errors_wait: Default::default(), + rcpt_script: Default::default(), + rcpt_relay: Default::default(), + rcpt_errors_max: Default::default(), + rcpt_errors_wait: Default::default(), + rcpt_max: Default::default(), + rcpt_dsn: Default::default(), + rcpt_lookup_domain: Default::default(), + rcpt_lookup_addresses: Default::default(), + rcpt_lookup_expn: Default::default(), + rcpt_lookup_vrfy: Default::default(), + max_message_size: Default::default(), + iprev: crate::config::VerifyStrategy::Disable, + spf_ehlo: crate::config::VerifyStrategy::Disable, + spf_mail_from: crate::config::VerifyStrategy::Disable, + dnsbl_policy: 0, + }, + in_flight: vec![], + } + } +} diff --git a/crates/store/src/backend/foundationdb/read.rs b/crates/store/src/backend/foundationdb/read.rs index 676b55ea..0aa73204 100644 --- a/crates/store/src/backend/foundationdb/read.rs +++ b/crates/store/src/backend/foundationdb/read.rs @@ -174,10 +174,22 @@ impl ReadTransaction<'_> { let mut bm = RoaringBitmap::new(); let mut range_stream = self.trx.get_ranges(opt, true); - while let Some(values) = range_stream.next().await { - for value in values? { - let key = value.key(); - bm.insert(key.deserialize_be_u32(key.len() - std::mem::size_of::())?); + if op != Operator::Equal { + while let Some(values) = range_stream.next().await { + for value in values? { + let key = value.key(); + bm.insert(key.deserialize_be_u32(key.len() - std::mem::size_of::())?); + } + } + } else { + let key_len = begin.len(); + while let Some(values) = range_stream.next().await { + for value in values? { + let key = value.key(); + if key.len() == key_len { + bm.insert(key.deserialize_be_u32(key.len() - std::mem::size_of::())?); + } + } } } diff --git a/crates/store/src/backend/sqlite/read.rs b/crates/store/src/backend/sqlite/read.rs index 818e99b1..b0747aa9 100644 --- a/crates/store/src/backend/sqlite/read.rs +++ b/crates/store/src/backend/sqlite/read.rs @@ -180,9 +180,19 @@ impl ReadTransaction<'_> { let mut query = self.conn.prepare_cached(query)?; let mut rows = query.query([&begin, &end])?; - while let Some(row) = rows.next()? { - let key = row.get_ref(0)?.as_bytes()?; - bm.insert(key.deserialize_be_u32(key.len() - std::mem::size_of::())?); + if op != Operator::Equal { + while let Some(row) = rows.next()? { + let key = row.get_ref(0)?.as_bytes()?; + bm.insert(key.deserialize_be_u32(key.len() - std::mem::size_of::())?); + } + } else { + let key_len = begin.len(); + while let Some(row) = rows.next()? { + let key = row.get_ref(0)?.as_bytes()?; + if key.len() == key_len { + bm.insert(key.deserialize_be_u32(key.len() - std::mem::size_of::())?); + } + } } Ok(Some(bm)) diff --git a/crates/store/src/blob/write.rs b/crates/store/src/blob/write.rs index 9b43124f..f2c1b875 100644 --- a/crates/store/src/blob/write.rs +++ b/crates/store/src/blob/write.rs @@ -1,3 +1,5 @@ +use std::ops::Range; + use tokio::{ fs::{self, File}, io::AsyncWriteExt, @@ -31,20 +33,34 @@ impl Store { } } - pub async fn copy_blob(&self, src: &BlobKind, dest: &BlobKind) -> crate::Result { + pub async fn copy_blob( + &self, + src: &BlobKind, + dest: &BlobKind, + range: Option>, + ) -> crate::Result { match &self.blob { BlobStore::Local(base_path) => { - let src_path = get_path(base_path, src)?; let dest_path = get_path(base_path, dest)?; - if fs::metadata(&src_path).await.is_err() { - return Ok(false); + if let Some(range) = range { + if let Some(bytes) = self.get_blob(src, range).await? { + fs::create_dir_all(dest_path.parent().unwrap()).await?; + fs::write(dest_path, bytes).await?; + Ok(true) + } else { + Ok(false) + } + } else { + let src_path = get_path(base_path, src)?; + if fs::metadata(&src_path).await.is_ok() { + fs::create_dir_all(dest_path.parent().unwrap()).await?; + fs::copy(src_path, dest_path).await?; + Ok(true) + } else { + Ok(false) + } } - - fs::create_dir_all(dest_path.parent().unwrap()).await?; - fs::copy(src_path, dest_path).await?; - - Ok(true) } BlobStore::Remote(_) => todo!(), } diff --git a/crates/store/src/query/mod.rs b/crates/store/src/query/mod.rs index b96ad480..6e43f8cc 100644 --- a/crates/store/src/query/mod.rs +++ b/crates/store/src/query/mod.rs @@ -9,7 +9,7 @@ use crate::{ fts::Language, write::BitmapFamily, BitmapKey, Deserialize, Serialize, BM_DOCUMENT_IDS, }; -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum Operator { LowerThan, LowerEqualThan, diff --git a/crates/utils/src/lib.rs b/crates/utils/src/lib.rs index 84896458..a88d7c01 100644 --- a/crates/utils/src/lib.rs +++ b/crates/utils/src/lib.rs @@ -30,6 +30,7 @@ pub mod config; pub mod ipc; pub mod listener; pub mod map; +//pub mod queue; use opentelemetry::{ sdk::{ diff --git a/crates/utils/src/queue.rs b/crates/utils/src/queue.rs new file mode 100644 index 00000000..f12f07c3 --- /dev/null +++ b/crates/utils/src/queue.rs @@ -0,0 +1,276 @@ +use std::{ + fmt::Display, + path::PathBuf, + sync::{atomic::AtomicUsize, Arc}, + time::{Duration, Instant}, +}; + +use serde::{Deserialize, Serialize}; +use smtp_proto::Response; +use tokio::sync::oneshot; + +use crate::listener::limiter::ConcurrencyLimiter; + +pub type QueueId = u64; + +#[derive(Debug)] +pub enum Event { + Queue(Schedule>), + Manage(QueueRequest), + Done(WorkerResult), + Stop, +} + +#[derive(Debug)] +pub enum QueueRequest { + List { + from: Option, + to: Option, + before: Option, + after: Option, + result_tx: oneshot::Sender>, + }, + Status { + queue_ids: Vec, + result_tx: oneshot::Sender>>, + }, + Cancel { + queue_ids: Vec, + item: Option, + result_tx: oneshot::Sender>, + }, + Retry { + queue_ids: Vec, + item: Option, + time: Instant, + result_tx: oneshot::Sender>, + }, +} + +#[derive(Debug)] +pub enum WorkerResult { + Done, + Retry(Schedule>), + OnHold(OnHold>), +} + +#[derive(Debug)] +pub struct OnHold { + pub next_due: Option, + pub limiters: Vec, + pub message: T, +} + +#[derive(Debug)] +pub struct Schedule { + pub due: Instant, + pub inner: T, +} + +#[derive(Debug)] +pub struct Message { + pub id: QueueId, + pub created: u64, + pub path: PathBuf, + + pub return_path: String, + pub return_path_lcase: String, + pub return_path_domain: String, + pub recipients: Vec, + pub domains: Vec, + + pub flags: u64, + pub env_id: Option, + pub priority: i16, + + pub size: usize, + pub queue_refs: Vec, +} + +#[derive(Debug, PartialEq, Eq)] +pub struct Domain { + pub domain: String, + pub retry: Schedule, + pub notify: Schedule, + pub expires: Instant, + pub status: Status<(), Error>, + pub changed: bool, +} + +#[derive(Debug, PartialEq, Eq)] +pub struct Recipient { + pub domain_idx: usize, + pub address: String, + pub address_lcase: String, + pub status: Status, HostResponse>, + pub flags: u64, + pub orcpt: Option, +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Status { + #[serde(rename = "scheduled")] + Scheduled, + #[serde(rename = "completed")] + Completed(T), + #[serde(rename = "temp_fail")] + TemporaryFailure(E), + #[serde(rename = "perm_fail")] + PermanentFailure(E), +} + +#[derive(Debug, PartialEq, Eq)] +pub struct HostResponse { + pub hostname: T, + pub response: Response, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum Error { + DnsError(String), + UnexpectedResponse(HostResponse), + ConnectionError(ErrorDetails), + TlsError(ErrorDetails), + DaneError(ErrorDetails), + MtaStsError(String), + RateLimited, + ConcurrencyLimited, + Io(String), +} + +#[derive(Debug, PartialEq, Eq)] +pub struct ErrorDetails { + pub entity: String, + pub details: String, +} + +#[derive(Debug)] +pub struct UsedQuota { + pub id: u64, + pub size: usize, + pub limiter: Arc, +} + +#[derive(Debug)] +pub struct QuotaLimiter { + pub max_size: usize, + pub max_messages: usize, + pub size: AtomicUsize, + pub messages: AtomicUsize, +} + +impl PartialEq for UsedQuota { + fn eq(&self, other: &Self) -> bool { + self.id == other.id && self.size == other.size + } +} + +impl Eq for UsedQuota {} + +impl Ord for Schedule { + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + other.due.cmp(&self.due) + } +} + +impl PartialOrd for Schedule { + fn partial_cmp(&self, other: &Self) -> Option { + other.due.partial_cmp(&self.due) + } +} + +impl PartialEq for Schedule { + fn eq(&self, other: &Self) -> bool { + self.due == other.due + } +} + +impl Eq for Schedule {} + +impl Schedule { + pub fn now() -> Self { + Schedule { + due: Instant::now(), + inner: T::default(), + } + } + + pub fn later(duration: Duration) -> Self { + Schedule { + due: Instant::now() + duration, + inner: T::default(), + } + } +} + +impl Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Error::UnexpectedResponse(response) => { + write!( + f, + "Unexpected response from '{}': {}", + response.hostname.entity, response.response + ) + } + Error::DnsError(err) => { + write!(f, "DNS lookup failed: {err}") + } + Error::ConnectionError(details) => { + write!( + f, + "Connection to '{}' failed: {}", + details.entity, details.details + ) + } + Error::TlsError(details) => { + write!( + f, + "TLS error from '{}': {}", + details.entity, details.details + ) + } + Error::DaneError(details) => { + write!( + f, + "DANE failed to authenticate '{}': {}", + details.entity, details.details + ) + } + Error::MtaStsError(details) => { + write!(f, "MTA-STS auth failed: {details}") + } + Error::RateLimited => { + write!(f, "Rate limited") + } + Error::ConcurrencyLimited => { + write!(f, "Too many concurrent connections to remote server") + } + Error::Io(err) => { + write!(f, "Queue error: {err}") + } + } + } +} + +impl Display for Status<(), Error> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Status::Scheduled => write!(f, "Scheduled"), + Status::Completed(_) => write!(f, "Completed"), + Status::TemporaryFailure(err) => write!(f, "Temporary Failure: {err}"), + Status::PermanentFailure(err) => write!(f, "Permanent Failure: {err}"), + } + } +} + +impl Display for Status, HostResponse> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Status::Scheduled => write!(f, "Scheduled"), + Status::Completed(response) => write!(f, "Delivered: {}", response.response), + Status::TemporaryFailure(err) => write!(f, "Temporary Failure: {}", err.response), + Status::PermanentFailure(err) => write!(f, "Permanent Failure: {}", err.response), + } + } +} diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 4c332d33..574bffe9 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -41,3 +41,4 @@ serial_test = "2.0.0" sqlx = { version = "0.7.0-alpha.3", features = [ "runtime-tokio-rustls", "postgres", "mysql", "sqlite" ] } num_cpus = "1.15.0" async-trait = "0.1.68" +chrono = "0.4" diff --git a/tests/src/jmap/delivery.rs b/tests/src/jmap/delivery.rs index 3764bb9c..71b44902 100644 --- a/tests/src/jmap/delivery.rs +++ b/tests/src/jmap/delivery.rs @@ -47,7 +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( @@ -78,7 +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( @@ -120,7 +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 @@ -154,7 +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 @@ -192,7 +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 @@ -236,14 +236,16 @@ impl SmtpConnection { self.rcpt_to(recipient, 2).await; } self.data(3).await; - self.data_bytes(message, recipients.len(), code).await + let result = self.data_bytes(message, recipients.len(), code).await; + tokio::time::sleep(Duration::from_millis(200)).await; + result } pub async fn ingest(&mut self, from: &str, recipients: &[&str], message: &str) { self.ingest_with_code(from, recipients, message, 2).await; } - pub async fn ingest_chunked( + async fn ingest_chunked( &mut self, from: &str, recipients: &[&str], @@ -258,6 +260,7 @@ impl SmtpConnection { self.bdat(std::str::from_utf8(chunk).unwrap(), 2).await; } self.bdat_last("", recipients.len(), 2).await; + tokio::time::sleep(Duration::from_millis(200)).await; } pub async fn connect() -> Self { diff --git a/tests/src/jmap/email_submission.rs b/tests/src/jmap/email_submission.rs new file mode 100644 index 00000000..09568a14 --- /dev/null +++ b/tests/src/jmap/email_submission.rs @@ -0,0 +1,628 @@ +use std::{sync::Arc, time::Duration}; + +use ahash::AHashMap; +use jmap::{JMAP, SUPERUSER_ID}; +use jmap_client::{ + client::Client, + core::set::{SetError, SetErrorType, SetObject}, + email_submission::{Address, Delivered, DeliveryStatus, Displayed, UndoStatus}, + mailbox::Role, + Error, +}; +use jmap_proto::types::id::Id; +use mail_parser::DateTime; +use store::parking_lot::Mutex; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::TcpListener, + sync::mpsc, +}; + +use crate::jmap::email_set::assert_email_properties; + +#[derive(Default, Debug, PartialEq, Eq)] +pub struct MockMessage { + pub mail_from: String, + pub rcpt_to: Vec, + pub message: String, +} + +impl MockMessage { + pub fn new(mail_from: T, rcpt_to: U, message: T) -> Self + where + T: Into, + U: IntoIterator, + { + Self { + mail_from: mail_from.into(), + rcpt_to: rcpt_to.into_iter().map(|s| s.into()).collect(), + message: message.into(), + } + } +} + +#[derive(Default)] +pub struct MockSMTPSettings { + pub fail_mail_from: bool, + pub fail_rcpt_to: bool, + pub fail_message: bool, + pub do_stop: bool, +} + +const TEST_DKIM_KEY: &str = r#"-----BEGIN RSA PRIVATE KEY----- +MIICXwIBAAKBgQDwIRP/UC3SBsEmGqZ9ZJW3/DkMoGeLnQg1fWn7/zYtIxN2SnFC +jxOCKG9v3b4jYfcTNh5ijSsq631uBItLa7od+v/RtdC2UzJ1lWT947qR+Rcac2gb +to/NMqJ0fzfVjH4OuKhitdY9tf6mcwGjaNBcWToIMmPSPDdQPNUYckcQ2QIDAQAB +AoGBALmn+XwWk7akvkUlqb+dOxyLB9i5VBVfje89Teolwc9YJT36BGN/l4e0l6QX +/1//6DWUTB3KI6wFcm7TWJcxbS0tcKZX7FsJvUz1SbQnkS54DJck1EZO/BLa5ckJ +gAYIaqlA9C0ZwM6i58lLlPadX/rtHb7pWzeNcZHjKrjM461ZAkEA+itss2nRlmyO +n1/5yDyCluST4dQfO8kAB3toSEVc7DeFeDhnC1mZdjASZNvdHS4gbLIA1hUGEF9m +3hKsGUMMPwJBAPW5v/U+AWTADFCS22t72NUurgzeAbzb1HWMqO4y4+9Hpjk5wvL/ +eVYizyuce3/fGke7aRYw/ADKygMJdW8H/OcCQQDz5OQb4j2QDpPZc0Nc4QlbvMsj +7p7otWRO5xRa6SzXqqV3+F0VpqvDmshEBkoCydaYwc2o6WQ5EBmExeV8124XAkEA +qZzGsIxVP+sEVRWZmW6KNFSdVUpk3qzK0Tz/WjQMe5z0UunY9Ax9/4PVhp/j61bf +eAYXunajbBSOLlx4D+TunwJBANkPI5S9iylsbLs6NkaMHV6k5ioHBBmgCak95JGX +GMot/L2x0IYyMLAz6oLWh2hm7zwtb0CgOrPo1ke44hFYnfc= +-----END RSA PRIVATE KEY-----"#; + +#[allow(clippy::disallowed_types)] +pub async fn test(server: Arc, client: &mut Client) { + println!("Running E-mail submissions tests..."); + // Start mock SMTP server + let (mut smtp_rx, smtp_settings) = spawn_mock_smtp_server(); + + // Create an identity without using a valid address should fail + match client + .set_default_account_id(Id::new(1).to_string()) + .identity_create("John Doe", "jdoe@example.com") + .await + .unwrap_err() + { + Error::Set(err) => assert_eq!(err.error(), &SetErrorType::InvalidProperties), + err => panic!("Unexpected error: {:?}", err), + } + + // Create a domain and a test account + let domain_id = client + .set_default_account_id(Id::new(0)) + .domain_create("example.com") + .await + .unwrap() + .take_id(); + let account_id = client + .individual_create("jdoe@example.com", "12345", "John Doe") + .await + .unwrap() + .take_id(); + let identity_id = client + .set_default_account_id(&account_id) + .identity_create("John Doe", "jdoe@example.com") + .await + .unwrap() + .take_id(); + + // Create test mailboxes + let mailbox_id = client + .mailbox_create("JMAP EmailSubmission", None::, Role::None) + .await + .unwrap() + .take_id(); + let mailbox_id_2 = client + .mailbox_create("JMAP EmailSubmission 2", None::, Role::None) + .await + .unwrap() + .take_id(); + + // Import an email without any recipients + let email_id = client + .email_import( + b"From: jdoe@example.com\nSubject: hey\n\ntest".to_vec(), + [&mailbox_id], + None::>, + None, + ) + .await + .unwrap() + .take_id(); + + // Submission without a valid emailId or identityId should fail + assert!(matches!( + client + .email_submission_create(Id::new(123456).to_string(), &identity_id) + .await, + Err(Error::Set(SetError { + type_: SetErrorType::InvalidProperties, + .. + })) + )); + assert!(matches!( + client + .email_submission_create(&email_id, Id::new(123456).to_string()) + .await, + Err(Error::Set(SetError { + type_: SetErrorType::InvalidProperties, + .. + })) + )); + + // Submissions of e-mails without any recipients should fail + assert!(matches!( + client + .email_submission_create(&email_id, &identity_id) + .await, + Err(Error::Set(SetError { + type_: SetErrorType::InvalidProperties, + .. + })) + )); + + // Submissions with an envelope that does not match + // the identity from address should fail + assert!(matches!( + client + .email_submission_create_envelope( + &email_id, + &identity_id, + "other_address@example.com", + Vec::
::new(), + ) + .await, + Err(Error::Set(SetError { + type_: SetErrorType::InvalidProperties, + .. + })) + )); + + // Submit a valid message submission + let email_body = + "From: jdoe@example.com\r\nTo: jane_smith@example.com\r\nSubject: hey\r\n\r\ntest"; + let email_id = client + .email_import( + email_body.as_bytes().to_vec(), + [&mailbox_id], + None::>, + None, + ) + .await + .unwrap() + .take_id(); + client + .email_submission_create(&email_id, &identity_id) + .await + .unwrap(); + + // Confirm that the message has been delivered + assert_message_delivery( + &mut smtp_rx, + MockMessage::new( + "", + [""], + email_body, + ), + false, + ) + .await; + + // Manually add recipients to the envelope and confirm submission + let email_submission_id = client + .email_submission_create_envelope( + &email_id, + &identity_id, + "jdoe@example.com", + [ + "tim@foobar.com", // Should be de-duplicated + "tim@foobar.com", + "tim@foobar.com ", + " james@other_domain.com ", // Should be sanitized + " secret_rcpt@test.com ", + ], + ) + .await + .unwrap() + .take_id(); + + assert_message_delivery( + &mut smtp_rx, + MockMessage::new( + "", + [ + "", + "", + "", + ], + email_body, + ), + false, + ) + .await; + + // Confirm that the email submission status was updated + tokio::time::sleep(Duration::from_millis(100)).await; + let email_submission = client + .email_submission_get(&email_submission_id, None) + .await + .unwrap() + .unwrap(); + assert_eq!(email_submission.undo_status().unwrap(), &UndoStatus::Final); + assert_eq!( + email_submission.delivery_status().unwrap(), + &AHashMap::from_iter([ + ( + "tim@foobar.com".to_string(), + DeliveryStatus::new("250 OK", Delivered::Queued, Displayed::Unknown) + ), + ( + "secret_rcpt@test.com".to_string(), + DeliveryStatus::new("250 OK", Delivered::Queued, Displayed::Unknown) + ), + ( + "james@other_domain.com".to_string(), + DeliveryStatus::new("250 OK", Delivered::Queued, Displayed::Unknown) + ), + ]) + ); + + // SMTP rejects some of the recipients + smtp_settings.lock().fail_rcpt_to = true; + let email_submission_id = client + .email_submission_create_envelope( + &email_id, + &identity_id, + "jdoe@example.com", + ["tim@foobar.com", "james@other_domain.com", "jane@test.com"], + ) + .await + .unwrap() + .take_id(); + assert_message_delivery( + &mut smtp_rx, + MockMessage::new("", [""], email_body), + false, + ) + .await; + + // Confirm that all delivery failures were included + tokio::time::sleep(Duration::from_millis(100)).await; + let email_submission = client + .email_submission_get(&email_submission_id, None) + .await + .unwrap() + .unwrap(); + assert_eq!(email_submission.undo_status().unwrap(), &UndoStatus::Final); + assert_eq!( + email_submission.delivery_status().unwrap(), + &AHashMap::from_iter([ + ( + "james@other_domain.com".to_string(), + DeliveryStatus::new( + "550 I refuse to accept that recipient.", + Delivered::No, + Displayed::Unknown + ) + ), + ( + "jane@test.com".to_string(), + DeliveryStatus::new( + "550 I refuse to accept that recipient.", + Delivered::No, + Displayed::Unknown + ) + ), + ( + "tim@foobar.com".to_string(), + DeliveryStatus::new("250 OK", Delivered::Queued, Displayed::Unknown) + ), + ]) + ); + smtp_settings.lock().fail_rcpt_to = false; + + // SMTP rejects the message + smtp_settings.lock().fail_message = true; + let email_submission_id = client + .email_submission_create_envelope( + &email_id, + &identity_id, + "jdoe@example.com", + ["tim@foobar.com", "james@other_domain.com", "jane@test.com"], + ) + .await + .unwrap() + .take_id(); + expect_nothing(&mut smtp_rx).await; + + // Confirm that all delivery failures were included + tokio::time::sleep(Duration::from_millis(100)).await; + let email_submission = client + .email_submission_get(&email_submission_id, None) + .await + .unwrap() + .unwrap(); + assert_eq!( + email_submission.undo_status().unwrap(), + &UndoStatus::Canceled + ); + assert_eq!( + email_submission.delivery_status().unwrap(), + &AHashMap::from_iter([ + ( + "james@other_domain.com".to_string(), + DeliveryStatus::new( + "503 Thank you but I am saving myself for dessert.", + Delivered::No, + Displayed::Unknown + ) + ), + ( + "jane@test.com".to_string(), + DeliveryStatus::new( + "503 Thank you but I am saving myself for dessert.", + Delivered::No, + Displayed::Unknown + ) + ), + ( + "tim@foobar.com".to_string(), + DeliveryStatus::new( + "503 Thank you but I am saving myself for dessert.", + Delivered::No, + Displayed::Unknown + ) + ), + ]) + ); + smtp_settings.lock().fail_message = false; + + // Enable DKIM for the domain + client + .set_default_account_id(Id::from(SUPERUSER_ID)) + .domain_enable_dkim(&domain_id, TEST_DKIM_KEY, "my-selector", None) + .await + .unwrap(); + client.set_default_account_id(&account_id); + + // Confirm that the sendAt property is updated when using FUTURERELEASE + let email_submission_id = client + .email_submission_create_envelope( + &email_id, + &identity_id, + Address::new("jdoe@example.com").parameter("HOLDUNTIL", Some("2079-11-20T05:00:00Z")), + ["jane_smith@example.com"], + ) + .await + .unwrap() + .take_id(); + assert_message_delivery( + &mut smtp_rx, + MockMessage::new( + " HOLDUNTIL=2079-11-20T05:00:00Z", + [""], + email_body, + ), + true, + ) + .await; + tokio::time::sleep(Duration::from_millis(100)).await; + let email_submission = client + .email_submission_get(&email_submission_id, None) + .await + .unwrap() + .unwrap(); + assert_eq!( + email_submission.send_at().unwrap(), + DateTime::parse_rfc3339("2079-11-20T05:00:00Z") + .unwrap() + .to_timestamp() + ); + + // Verify onSuccessUpdateEmail action + let mut request = client.build(); + let set_request = request.set_email_submission(); + let create_id = set_request + .create() + .email_id(&email_id) + .identity_id(&identity_id) + .create_id() + .unwrap(); + set_request + .arguments() + .on_success_update_email(&create_id) + .keyword("$draft", true) + .mailbox_id(&mailbox_id, false) + .mailbox_id(&mailbox_id_2, true); + request.send().await.unwrap().unwrap_method_responses(); + + assert_email_properties(client, &email_id, &[&mailbox_id_2], &["$draft"]).await; + + // Verify onSuccessDestroyEmail action + smtp_settings.lock().do_stop = true; + let mut request = client.build(); + let set_request = request.set_email_submission(); + let create_id = set_request + .create() + .email_id(&email_id) + .identity_id(&identity_id) + .create_id() + .unwrap(); + set_request.arguments().on_success_destroy_email(&create_id); + request.send().await.unwrap().unwrap_method_responses(); + + assert!(client + .email_get(&email_id, None::>) + .await + .unwrap() + .is_none()); + + // Destroy the created mailbox, identity and all submissions + let todo = "true"; + /*client + .set_default_account_id(Id::from(SUPERUSER_ID)) + .principal_destroy(&account_id) + .await + .unwrap(); + client.principal_destroy(&domain_id).await.unwrap(); + server.store.principal_purge().unwrap(); + server.store.assert_is_empty();*/ +} + +pub fn spawn_mock_smtp_server() -> (mpsc::Receiver, Arc>) { + // Create channels + let (event_tx, event_rx) = mpsc::channel::(100); + let _settings = Arc::new(Mutex::new(MockSMTPSettings::default())); + let settings = _settings.clone(); + + // Start mock SMTP server + tokio::spawn(async move { + let listener = TcpListener::bind("127.0.0.1:9999") + .await + .unwrap_or_else(|e| { + panic!("Failed to bind mock SMTP server to 127.0.0.1:9999: {}", e); + }); + + while let Ok((mut stream, _)) = listener.accept().await { + let (rx, mut tx) = stream.split(); + let mut rx = BufReader::new(rx); + let mut buf = String::with_capacity(128); + let mut message = MockMessage::default(); + + tx.write_all(b"220 [127.0.0.1] Clueless host service ready\r\n") + .await + .unwrap(); + + while rx.read_line(&mut buf).await.is_ok() { + print!("-> {}", buf); + if buf.starts_with("EHLO") { + tx.write_all(b"250 Hi there, but I have no extensions to offer :-(\r\n") + .await + .unwrap(); + } else if buf.starts_with("MAIL FROM") { + if settings.lock().fail_mail_from { + tx.write_all("552-I do not\r\n552 like that MAIL FROM.\r\n".as_bytes()) + .await + .unwrap(); + } else { + message.mail_from = buf.split_once(':').unwrap().1.trim().to_string(); + tx.write_all(b"250 OK\r\n").await.unwrap(); + } + } else if buf.starts_with("RCPT TO") { + if settings.lock().fail_rcpt_to && !buf.contains("foobar.com") { + tx.write_all( + "550-I refuse to\r\n550 accept that recipient.\r\n".as_bytes(), + ) + .await + .unwrap(); + } else { + message + .rcpt_to + .push(buf.split(':').nth(1).unwrap().trim().to_string()); + tx.write_all(b"250 OK\r\n").await.unwrap(); + } + } else if buf.starts_with("DATA") { + if settings.lock().fail_message { + tx.write_all( + "503-Thank you but I am\r\n503 saving myself for dessert.\r\n" + .as_bytes(), + ) + .await + .unwrap(); + } else if !message.mail_from.is_empty() && !message.rcpt_to.is_empty() { + tx.write_all(b"354 Start feeding me now some quality content please\r\n") + .await + .unwrap(); + buf.clear(); + while rx.read_line(&mut buf).await.is_ok() { + if buf.starts_with('.') { + message.message = message.message.trim().to_string(); + break; + } else { + message.message += &buf; + buf.clear(); + } + } + tx.write_all(b"250 Great success!\r\n").await.unwrap(); + message.rcpt_to.sort_unstable(); + event_tx.send(message).await.unwrap(); + message = MockMessage::default(); + } else { + tx.write_all("554 You forgot to tell me a few things.\r\n".as_bytes()) + .await + .unwrap(); + } + } else if buf.starts_with("QUIT") { + tx.write_all("250 Arrivederci!\r\n".as_bytes()) + .await + .unwrap(); + break; + } else if buf.starts_with("RSET") { + tx.write_all("250 Your wish is my command.\r\n".as_bytes()) + .await + .unwrap(); + message = MockMessage::default(); + } else { + println!("Unknown command: {}", buf.trim()); + } + buf.clear(); + } + + if settings.lock().do_stop { + println!("Mock SMTP server stopped."); + break; + } + } + }); + + (event_rx, _settings) +} + +pub async fn assert_message_delivery( + event_rx: &mut mpsc::Receiver, + expected_message: MockMessage, + expect_dkim: bool, +) { + match tokio::time::timeout(Duration::from_millis(3000), event_rx.recv()).await { + Ok(Some(message)) => { + assert_eq!(message.mail_from, expected_message.mail_from); + assert_eq!(message.rcpt_to, expected_message.rcpt_to); + + println!("Got message [{}]", message.message); + + if let Some(needle) = expected_message.message.strip_prefix('@') { + assert!( + message.message.contains(needle), + "[{}] needle = {:?}", + message.message, + needle + ); + } else { + let message = if expect_dkim { + if message.message.starts_with("DKIM-Signature:") { + message.message.split_once('\n').unwrap().1 + } else { + panic!( + "Expected DKIM-Signature header but got: {}", + message.message + ); + } + } else { + &message.message + }; + + assert_eq!(message, expected_message.message); + } + } + result => { + panic!( + "Timeout waiting for message {:?}: {:?}", + expected_message, result + ); + } + } +} + +pub async fn expect_nothing(event_rx: &mut mpsc::Receiver) { + match tokio::time::timeout(Duration::from_millis(500), event_rx.recv()).await { + Err(_) => {} + message => { + panic!("Received a message when expecting nothing: {:?}", message); + } + } +} diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 66af6d50..ca13be54 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -21,11 +21,14 @@ pub mod email_query; pub mod email_query_changes; pub mod email_search_snippet; pub mod email_set; +pub mod email_submission; pub mod event_source; pub mod mailbox; pub mod push_subscription; +pub mod sieve_script; pub mod thread_get; pub mod thread_merge; +pub mod vacation_response; const SERVER: &str = r#" [server] @@ -123,6 +126,7 @@ address = "sqlite::memory:" 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 = ?" +name-by-uid = "SELECT name FROM users WHERE ROWID - 1 = ?" gids-by-uid = "SELECT gid FROM groups WHERE uid = ?" uids-by-address = "SELECT uid FROM emails WHERE email = ?" addresses-by-uid = "SELECT email FROM emails WHERE uid = ?" @@ -166,8 +170,11 @@ pub async fn jmap_tests() { //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; + sieve_script::test(params.server.clone(), &mut params.client).await; + + let websockets = "todo"; if delete { params.temp_dir.delete(); @@ -197,7 +204,7 @@ async fn init_jmap_tests(delete_if_exists: bool) -> JMAPTest { let smtp = SMTP::init(&config, &servers, delivery_tx) .await .failed("Invalid configuration file"); - let jmap = JMAP::init(&config, delivery_rx) + let jmap = JMAP::init(&config, delivery_rx, smtp.clone()) .await .failed("Invalid configuration file"); let shutdown_tx = servers.spawn(|server, shutdown_rx| { diff --git a/tests/src/jmap/sieve_script.rs b/tests/src/jmap/sieve_script.rs new file mode 100644 index 00000000..45252e3f --- /dev/null +++ b/tests/src/jmap/sieve_script.rs @@ -0,0 +1,460 @@ +use std::{fs, path::PathBuf, sync::Arc, time::Duration}; + +use jmap::{JMAP, SUPERUSER_ID}; +use jmap_client::{ + client::Client, + core::set::{SetError, SetErrorType}, + email, mailbox, + sieve::query::{Comparator, Filter}, + Error, +}; +use jmap_proto::types::id::Id; + +use crate::jmap::{ + delivery::SmtpConnection, + email_submission::{assert_message_delivery, spawn_mock_smtp_server, MockMessage}, + test_account_create, +}; +use crate::smtp::session::VerifyResponse; + +pub async fn test(server: Arc, client: &mut Client) { + println!("Running Sieve tests..."); + + // Create test account + let account_id = test_account_create(&server, "jdoe@example.com", "12345", "John Doe") + .await + .to_string(); + client.set_default_account_id(&account_id); + + // Validate scripts + client + .sieve_script_validate(get_script("validate_ok")) + .await + .unwrap(); + assert!(matches!( + client + .sieve_script_validate(get_script("validate_error")) + .await, + Err(Error::Set(SetError { + type_: SetErrorType::InvalidScript, + .. + })) + )); + + // Create 5 Sieve scripts, all deactivated. + let mut script_ids = Vec::new(); + for i in 0..5 { + script_ids.push( + client + .sieve_script_create( + format!("script_{}", i + 1), + format!("require \"fileinto\"; fileinto \"{}\";", i + 1).into_bytes(), + false, + ) + .await + .unwrap() + .take_id(), + ); + } + + let response = client + .sieve_script_query(Filter::is_active(false).into(), [Comparator::name()].into()) + .await + .unwrap(); + assert_eq!(response.ids().len(), 5); + for (pos, id) in response.ids().iter().enumerate() { + let script = client + .sieve_script_get(id, None::>) + .await + .unwrap() + .unwrap(); + assert_eq!(script.name().unwrap(), format!("script_{}", pos + 1)); + assert_eq!( + String::from_utf8(client.download(script.blob_id().unwrap()).await.unwrap()).unwrap(), + format!("require \"fileinto\"; fileinto \"{}\";", pos + 1) + ); + } + + // Activate last script twice and then the first script + for _ in 0..2 { + client + .sieve_script_activate(script_ids.last().unwrap()) + .await + .unwrap(); + assert_eq!( + client + .sieve_script_query(Filter::is_active(true).into(), [Comparator::name()].into()) + .await + .unwrap() + .ids(), + vec![script_ids.last().unwrap().to_string()] + ); + } + client + .sieve_script_activate(script_ids.first().unwrap()) + .await + .unwrap(); + assert_eq!( + client + .sieve_script_query(Filter::is_active(true).into(), [Comparator::name()].into()) + .await + .unwrap() + .ids(), + vec![script_ids.first().unwrap().to_string()] + ); + + // Destroying an active script should not work + assert!(matches!( + client + .sieve_script_destroy(script_ids.first().unwrap()) + .await, + Err(Error::Set(SetError { + type_: SetErrorType::ScriptIsActive, + .. + })) + )); + + // Deactivate all scripts + client.sieve_script_deactivate().await.unwrap(); + assert_eq!( + client + .sieve_script_query(Filter::is_active(true).into(), [Comparator::name()].into()) + .await + .unwrap() + .ids(), + Vec::::new() + ); + + // Connect to LMTP service + let mut lmtp = SmtpConnection::connect().await; + + // Run mailbox, fileinto, flags tests + client + .sieve_script_create("test_mailbox", get_script("test_mailbox"), true) + .await + .unwrap(); + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: TPS Report\r\n", + "\r\n", + "I'm going to need those TPS reports ASAP. ", + "So, if you could do that, that'd be great." + ), + ) + .await; + + // Make sure all folders were created + let mailbox_names = "My/Nested/Mailbox/with/multiple/levels/Folder" + .split('/') + .collect::>(); + let mut mailbox_ids = Vec::new(); + for &mailbox in &mailbox_names { + let mut response = client + .mailbox_query(mailbox::query::Filter::name(mailbox).into(), None::>) + .await + .unwrap(); + assert!( + !response.ids().is_empty(), + "Mailbox {} was not created.", + mailbox + ); + mailbox_ids.extend(response.take_ids()); + } + assert_eq!(mailbox_ids.len(), mailbox_names.len()); + + // Make sure the message was delivered to the right folders + let message_ids = client + .email_query(None::, None::>) + .await + .unwrap() + .take_ids(); + assert_eq!(message_ids.len(), 1, "too many messages {:?}", message_ids); + let email = client + .email_get( + message_ids.last().unwrap(), + [email::Property::MailboxIds, email::Property::Keywords].into(), + ) + .await + .unwrap() + .unwrap(); + assert_eq!( + email.keywords().len(), + 2, + "Expected 2 keywords, found {:?}.", + email.keywords() + ); + for keyword in ["$important", "$seen"] { + if !email.keywords().contains(&keyword) { + panic!("Keyword {} not found in {:?}.", keyword, email.keywords()); + } + } + assert_eq!( + email.mailbox_ids().len(), + 2, + "Expected 2 mailbox ids, found {:?}.", + email.mailbox_ids() + ); + for mailbox_pos in [mailbox_ids.len() - 1, mailbox_ids.len() - 2] { + if !email + .mailbox_ids() + .contains(&mailbox_ids[mailbox_pos].as_str()) + { + panic!( + "Mailbox {} ({}) not found in {:?}.", + mailbox_names[mailbox_pos], + mailbox_ids[mailbox_pos], + email.keywords() + ); + } + } + + // Run discard and duplicate tests + client + .sieve_script_create( + "test_discard_reject", + get_script("test_discard_reject"), + true, + ) + .await + .unwrap(); + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "Bcc: Undisclosed recipients;\r\n", + "Message-ID: <1234@example.com>\r\n", + "Subject: Holidays\r\n", + "\r\n", + "Remember to file your TPS reports before ", + "going on holidays." + ), + ) + .await; + assert_eq!( + client + .email_query(None::, None::>) + .await + .unwrap() + .ids() + .len(), + 1, + "Discard failed." + ); + + // Let one sec duplicate ids expire + tokio::time::sleep(Duration::from_millis(1100)).await; + + // Run reject and duplicate check tests + let test = "fd"; + /*lmtp.ingest_with_code( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "Bcc: Undisclosed recipients;\r\n", + "Message-ID: <1234@example.com>\r\n", + "Subject: Holidays\r\n", + "\r\n", + "Remember to file your T.P.S. reports before ", + "going on holidays." + ), + 5, + ) + .await + .assert_contains("No soup for you"); + assert_eq!( + client + .email_query(None::, None::>) + .await + .unwrap() + .ids() + .len(), + 1, + "Reject failed." + );*/ + + // Run include tests + client + .sieve_script_create("test_include_this", get_script("test_include_this"), false) + .await + .unwrap(); + client + .sieve_script_create("test_include", get_script("test_include"), true) + .await + .unwrap(); + lmtp.ingest_with_code( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "Bcc: Undisclosed recipients;\r\n", + "Message-ID: <1234@example.com>\r\n", + "Subject: Holidays\r\n", + "\r\n", + "Remember to file your T.P.S. reports before ", + "going on holidays." + ), + 5, + ) + .await + .assert_contains("Rejected from an included script"); + + // Start mock SMTP server + let coco = "fd"; + /*let (mut smtp_rx, smtp_settings) = spawn_mock_smtp_server(); + + // Run enclose + redirect tests + client + .sieve_script_create( + "test_redirect_enclose", + get_script("test_redirect_enclose"), + true, + ) + .await + .unwrap(); + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: TPS Report\r\n", + "\r\n", + "I'm going to need those TPS reports ASAP. ", + "So, if you could do that, that'd be great." + ), + ) + .await; + assert_message_delivery( + &mut smtp_rx, + MockMessage::new( + "", + [""], + "@Attached you'll find", + ), + false, + ) + .await; + assert_eq!( + client + .email_query(None::, None::>) + .await + .unwrap() + .ids() + .len(), + 1, + "Redirected message was stored." + ); + + // Run notify + editheader + notify + fcc tests + client + .sieve_script_create("test_notify_fcc", get_script("test_notify_fcc"), true) + .await + .unwrap(); + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: Urgently I need those TPS Reports\r\n", + "\r\n", + "I'm going to need those TPS reports ASAP. ", + "So, if you could do that, that'd be great." + ), + ) + .await; + + assert_message_delivery( + &mut smtp_rx, + MockMessage::new( + "", + [""], + "@It's TPS-o-clock", + ), + false, + ) + .await; + + let mut request = client.build(); + request.get_email().properties([ + email::Property::MailboxIds, + email::Property::Keywords, + email::Property::Subject, + ]); + let emails = request.send_get_email().await.unwrap().take_list(); + + assert_eq!( + emails.len(), + 3, + "Two new messages were expected: {:#?}.", + emails + ); + + 'outer: for (subject, folder, keywords) in [ + ("It's TPS-o-clock", "Notifications", ""), + ( + "Urgently I need those **censored** Reports", + "Inbox", + "$seen", + ), + ] { + for email in &emails { + if email.subject().unwrap().eq(subject) { + if !keywords.is_empty() && !email.keywords().contains(&keywords) { + panic!("Keyword {:?} not found in: {:#?}", keywords, email); + } + + let mailbox_id = client + .mailbox_query( + mailbox::query::Filter::name(folder.to_string()).into(), + None::>, + ) + .await + .unwrap() + .take_ids() + .pop() + .unwrap_or_else(|| panic!("Mailbox {:?} not found", folder)); + + if !email.mailbox_ids().contains(&mailbox_id.as_str()) { + panic!( + "Mailbox {:?} ({}) not found in: {:#?}", + folder, mailbox_id, email + ); + } + + continue 'outer; + } + } + panic!("Email {:?} not found in: {:#?}", subject, emails); + } + + smtp_settings.lock().do_stop = true; + + */ + + // Remove test data + let todo = "fix"; + /*for account_id in [&account_id, &domain_id] { + client + .set_default_account_id(Id::new(SUPERUSER_ID as u64)) + .principal_destroy(account_id) + .await + .unwrap(); + } + server.store.principal_purge().unwrap(); + server.store.assert_is_empty();*/ +} + +fn get_script(name: &str) -> Vec { + let mut script_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + script_path.push("resources"); + script_path.push("jmap_sieve"); + script_path.push(format!("{}.sieve", name)); + fs::read(script_path).unwrap() +} diff --git a/tests/src/jmap/vacation_response.rs b/tests/src/jmap/vacation_response.rs new file mode 100644 index 00000000..265fd760 --- /dev/null +++ b/tests/src/jmap/vacation_response.rs @@ -0,0 +1,167 @@ +use std::sync::Arc; + +use chrono::{Duration, Utc}; +use jmap::{JMAP, SUPERUSER_ID}; +use jmap_client::client::Client; +use jmap_proto::types::id::Id; + +use crate::jmap::{ + delivery::SmtpConnection, + email_submission::{ + assert_message_delivery, expect_nothing, spawn_mock_smtp_server, MockMessage, + }, +}; + +pub async fn test(server: Arc, client: &mut Client) { + println!("Running Vacation Response tests..."); + + // Create INBOX + let domain_id = client + .set_default_account_id(Id::new(SUPERUSER_ID as u64)) + .domain_create("example.com") + .await + .unwrap() + .take_id(); + let account_id = client + .individual_create("jdoe@example.com", "12345", "John Doe") + .await + .unwrap() + .take_id(); + + // Start mock SMTP server + let (mut smtp_rx, smtp_settings) = spawn_mock_smtp_server(); + + // Let people know that we'll be down in Kokomo + client + .set_default_account_id(&account_id) + .vacation_response_create( + "Off the Florida Keys there's a place called Kokomo", + "That's where you wanna go to get away from it all".into(), + "That's where you wanna go to get away from it all".into(), + ) + .await + .unwrap(); + + // Connect to LMTP service + let mut lmtp = SmtpConnection::connect().await; + + // Send a message + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: TPS Report\r\n", + "\r\n", + "I'm going to need those TPS reports ASAP. ", + "So, if you could do that, that'd be great." + ), + ) + .await; + + // Await vacation response + assert_message_delivery( + &mut smtp_rx, + MockMessage::new("", [""], "@Kokomo"), + false, + ) + .await; + + // Further messages from the same recipient should not + // trigger a vacation response + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: TPS Report -- friendly reminder\r\n", + "\r\n", + "Listen, are you gonna have those TPS reports for us this afternoon?", + ), + ) + .await; + + expect_nothing(&mut smtp_rx).await; + + // Messages from MAILER-DAEMON should not + // trigger a vacation response + lmtp.ingest( + "MAILER-DAEMON@example.com", + &["jdoe@example.com"], + concat!( + "From: MAILER-DAEMON@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: Delivery Failure\r\n", + "\r\n", + "I tried so hard and got so far but in the end it wasn't delivered.", + ), + ) + .await; + + expect_nothing(&mut smtp_rx).await; + + // Vacation responses should honor the configured date ranges + client + .vacation_response_set_dates((Utc::now() + Duration::days(1)).timestamp().into(), None) + .await + .unwrap(); + lmtp.ingest( + "jane_smith@example.com", + &["jdoe@example.com"], + concat!( + "From: jane_smith@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: When were you going on holidays?\r\n", + "\r\n", + "I'm asking because Bill really wants those TPS reports.", + ), + ) + .await; + + expect_nothing(&mut smtp_rx).await; + + client + .vacation_response_set_dates((Utc::now() - Duration::days(1)).timestamp().into(), None) + .await + .unwrap(); + smtp_settings.lock().do_stop = true; + lmtp.ingest( + "jane_smith@example.com", + &["jdoe@example.com"], + concat!( + "From: jane_smith@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: When were you going on holidays?\r\n", + "\r\n", + "I'm asking because Bill really wants those TPS reports.", + ), + ) + .await; + lmtp.quit().await; + + assert_message_delivery( + &mut smtp_rx, + MockMessage::new( + "", + [""], + "@Kokomo", + ), + false, + ) + .await; + + // Remove test data + let implement = "true"; + /*for account_id in [&account_id, &domain_id] { + client + .set_default_account_id(Id::new(SUPERUSER_ID as u64)) + .principal_destroy(account_id) + .await + .unwrap(); + } + server.store.principal_purge().unwrap(); + server.store.assert_is_empty(); + */ +}