diff --git a/Cargo.lock b/Cargo.lock index 48aa87d9..b07d389a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4620,7 +4620,6 @@ version = "0.14.1" dependencies = [ "ahash", "bincode 1.3.3", - "farmhash", "hashify", "jieba-rs", "lru-cache", @@ -4629,9 +4628,9 @@ dependencies = [ "parking_lot", "psl", "radix_trie 0.3.0", + "rand 0.9.2", "rust-stemmers", "serde", - "siphasher", "tokio", "utils", "whatlang", diff --git a/crates/common/src/config/spamfilter.rs b/crates/common/src/config/spamfilter.rs index 67cffc21..e0789038 100644 --- a/crates/common/src/config/spamfilter.rs +++ b/crates/common/src/config/spamfilter.rs @@ -51,21 +51,13 @@ pub struct SpamFilterConfig { pub pyzor: Option, pub classifier: Option, pub scores: SpamFilterScoreConfig, - pub headers: SpamFilterHeaderConfig, -} - -#[derive(Debug, Clone)] -pub struct SpamFilterHeaderConfig { - pub status: Option, - pub result: Option, - pub llm: Option, } #[derive(Debug, Clone, Default)] pub struct SpamFilterScoreConfig { - pub reject_threshold: f64, - pub discard_threshold: f64, - pub spam_threshold: f64, + pub reject_threshold: f32, + pub discard_threshold: f32, + pub spam_threshold: f32, } #[derive(Debug, Clone, Default)] @@ -80,7 +72,7 @@ pub struct DnsBlConfig { #[derive(Debug, Clone, Default)] pub struct SpamFilterLists { pub file_extensions: GlobMap, - pub scores: GlobMap>, + pub scores: GlobMap>, } #[derive(Debug, Clone)] @@ -88,6 +80,7 @@ pub enum SpamFilterAction { Allow(T), Discard, Reject, + Disabled, } #[derive(Debug, Clone, Default)] @@ -97,8 +90,8 @@ pub struct ClassifierConfig { pub alpha: f32, pub auto_learn_reply_ham: bool, pub auto_learn_card_is_ham: bool, - pub score_spam: f64, - pub score_ham: f64, + pub score_spam: f32, + pub score_ham: f32, } #[derive(Debug, Clone)] @@ -185,7 +178,6 @@ impl SpamFilterConfig { pyzor: PyzorConfig::parse(config).await, classifier: ClassifierConfig::parse(config), scores: SpamFilterScoreConfig::parse(config), - headers: SpamFilterHeaderConfig::parse(config), grey_list_expiry: config .property::>("spam-filter.grey-list.duration") .unwrap_or_default() @@ -316,31 +308,6 @@ impl DnsBlServer { } } -impl SpamFilterHeaderConfig { - pub fn parse(config: &mut Config) -> Self { - let mut header = SpamFilterHeaderConfig::default(); - - for (typ, var) in [ - ("status", &mut header.status), - ("result", &mut header.result), - ("llm", &mut header.llm), - ] { - if config - .property_or_default(("spam-filter.header", typ, "enable"), "true") - .unwrap_or(true) - && let Some(value) = config.value(("spam-filter.header", typ, "name")) - { - let value = value.trim(); - if !value.is_empty() { - *var = value.to_string().into(); - } - } - } - - header - } -} - impl SpamFilterLists { pub fn parse(config: &mut Config) -> Self { let mut lists = SpamFilterLists { @@ -571,16 +538,6 @@ impl Location { } } -impl Default for SpamFilterHeaderConfig { - fn default() -> Self { - SpamFilterHeaderConfig { - status: "X-Spam-Status".to_string().into(), - result: "X-Spam-Result".to_string().into(), - llm: "X-Spam-LLM".to_string().into(), - } - } -} - pub const V_SPAM_REMOTE_IP: u32 = 100; pub const V_SPAM_REMOTE_IP_PTR: u32 = 101; pub const V_SPAM_EHLO_DOMAIN: u32 = 102; @@ -809,3 +766,12 @@ impl CacheItemWeight for IpResolver { (std::mem::size_of::() + self.ip_string.len() + self.reverse.len()) as u64 } } + +impl SpamFilterAction { + pub fn as_score(&self) -> Option<&T> { + match self { + SpamFilterAction::Allow(value) => Some(value), + _ => None, + } + } +} diff --git a/crates/common/src/expr/mod.rs b/crates/common/src/expr/mod.rs index 5e07c278..86dfab25 100644 --- a/crates/common/src/expr/mod.rs +++ b/crates/common/src/expr/mod.rs @@ -4,12 +4,16 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use self::tokenizer::TokenMap; +use compact_str::CompactString; +use regex::Regex; use std::{ borrow::Cow, fmt::{Display, Formatter}, net::{IpAddr, Ipv4Addr, Ipv6Addr}, time::Duration, }; +use utils::config::{Rate, utils::ParseValue}; pub const V_RECIPIENT: u32 = 0; pub const V_RECIPIENT_DOMAIN: u32 = 1; @@ -81,12 +85,6 @@ pub const VARIABLES_MAP: &[(&str, u32)] = &[ ("queue_age", V_QUEUE_AGE), ]; -use compact_str::CompactString; -use regex::Regex; -use utils::config::{Rate, utils::ParseValue}; - -use self::tokenizer::TokenMap; - pub mod eval; pub mod functions; pub mod if_block; diff --git a/crates/directory/src/core/mod.rs b/crates/directory/src/core/mod.rs index 790f6c9e..8723c940 100644 --- a/crates/directory/src/core/mod.rs +++ b/crates/directory/src/core/mod.rs @@ -18,7 +18,7 @@ impl Permission { Permission::Impersonate => "Act on behalf of another user", Permission::UnlimitedRequests => "Perform unlimited requests", Permission::UnlimitedUploads => "Upload unlimited data", - Permission::DeleteSystemFolders => "Delete of system folders", + Permission::DeleteSystemFolders_ => "", Permission::MessageQueueList => "View message queue", Permission::MessageQueueGet => "Retrieve specific messages from the queue", Permission::MessageQueueUpdate => "Modify queued messages", @@ -82,8 +82,8 @@ impl Permission { Permission::SpamFilterUpdate => "Modify spam filter settings", Permission::WebadminUpdate => "Modify web admin interface settings", Permission::LogsView => "Access system logs", - Permission::SpamFilterTrain => "Train the spam filter", - Permission::SpamFilterClassify => "Classify emails with the spam filter", + Permission::SpamFilterTrain_ => "", + Permission::SpamFilterClassify_ => "", Permission::Restart => "Restart the email server", Permission::TracingList => "View stored traces", Permission::TracingGet => "Retrieve specific trace information", diff --git a/crates/directory/src/core/principal.rs b/crates/directory/src/core/principal.rs index e1a749e2..e90cc728 100644 --- a/crates/directory/src/core/principal.rs +++ b/crates/directory/src/core/principal.rs @@ -1577,8 +1577,6 @@ impl Permission { | Permission::SieveRenameScript | Permission::SieveCheckScript | Permission::SieveHaveSpace - | Permission::SpamFilterClassify - | Permission::SpamFilterTrain | Permission::DavSyncCollection | Permission::DavExpandProperty | Permission::DavPrincipalAcl diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index ddb8577e..283e9ff6 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -125,7 +125,7 @@ pub enum Permission { Impersonate, UnlimitedRequests, UnlimitedUploads, - DeleteSystemFolders, + DeleteSystemFolders_, MessageQueueList, MessageQueueGet, MessageQueueUpdate, @@ -187,7 +187,7 @@ pub enum Permission { SpamFilterUpdate, WebadminUpdate, LogsView, - SpamFilterTrain, + SpamFilterTrain_, Restart, TracingList, TracingGet, @@ -321,7 +321,7 @@ pub enum Permission { AiModelInteract, Troubleshoot, - SpamFilterClassify, + SpamFilterClassify_, // WebDAV permissions DavSyncCollection, @@ -423,7 +423,7 @@ pub enum Permission { JmapParticipantIdentityGet, JmapParticipantIdentitySet, JmapParticipantIdentityChanges, - // TODO: Reuse DeleteSystemFolders position for new permission + // TODO: Reuse _ suffixes for new permissions // WARNING: add new ids at the end (TODO: use static ids) } diff --git a/crates/email/src/cache/email.rs b/crates/email/src/cache/email.rs index 3a28729e..edfe07af 100644 --- a/crates/email/src/cache/email.rs +++ b/crates/email/src/cache/email.rs @@ -157,6 +157,8 @@ pub trait MessageCacheAccess { fn in_mailbox(&self, mailbox_id: u32) -> impl Iterator; + fn in_mailboxes(&self, mailbox_ids: &[u32]) -> impl Iterator; + fn in_thread(&self, thread_id: u32) -> impl Iterator; fn with_keyword(&self, keyword: &Keyword) -> impl Iterator; @@ -196,6 +198,14 @@ impl MessageCacheAccess for MessageStoreCache { .filter(move |m| m.mailboxes.iter().any(|m| m.mailbox_id == mailbox_id)) } + fn in_mailboxes(&self, mailbox_ids: &[u32]) -> impl Iterator { + self.emails.items.iter().filter(move |m| { + m.mailboxes + .iter() + .any(|mb| mailbox_ids.contains(&mb.mailbox_id)) + }) + } + fn in_thread(&self, thread_id: u32) -> impl Iterator { self.emails .items diff --git a/crates/email/src/message/crypto.rs b/crates/email/src/message/crypto.rs index 93bdc53f..ab74cfb4 100644 --- a/crates/email/src/message/crypto.rs +++ b/crates/email/src/message/crypto.rs @@ -79,11 +79,16 @@ pub enum EncryptionMethod { serde::Deserialize, )] pub struct EncryptionParams { - pub method: EncryptionMethod, - pub algo: Algorithm, - pub certs: Vec>, + pub certs: Box<[Box<[u8]>]>, + pub flags: u64, } +pub const ENCRYPT_TRAIN_SPAM_FILTER: u64 = 1; +pub const ENCRYPT_METHOD_SMIME: u64 = 1 << 1; +pub const ENCRYPT_METHOD_PGP: u64 = 1 << 2; +pub const ENCRYPT_ALGO_AES256: u64 = 1 << 3; +pub const ENCRYPT_ALGO_AES128: u64 = 1 << 4; + #[derive( rkyv::Serialize, rkyv::Deserialize, @@ -142,8 +147,8 @@ impl EncryptMessage for Message<'_> { inner_message.extend_from_slice(&raw_message[root.raw_body_offset() as usize..]); // Encrypt inner message - match params.method { - ArchivedEncryptionMethod::PGP => { + match params.method() { + EncryptionMethod::PGP => { // Prepare encrypted message let boundary = make_boundary("_"); outer_message.extend_from_slice( @@ -193,7 +198,7 @@ impl EncryptMessage for Message<'_> { })?; // Encrypt contents (TODO: use rayon) - let algo = params.algo; + let algo = params.algo(); let encrypted_contents = tokio::task::spawn_blocking(move || { // Parse public key let mut keys = Vec::with_capacity(certs.len()); @@ -224,8 +229,8 @@ impl EncryptMessage for Message<'_> { })?; let message = stream::Encryptor::for_recipients(message, keys) .symmetric_algo(match algo { - ArchivedAlgorithm::Aes128 => SymmetricAlgorithm::AES128, - ArchivedAlgorithm::Aes256 => SymmetricAlgorithm::AES256, + Algorithm::Aes128 => SymmetricAlgorithm::AES128, + Algorithm::Aes256 => SymmetricAlgorithm::AES256, }) .build() .map_err(|err| { @@ -269,18 +274,18 @@ impl EncryptMessage for Message<'_> { outer_message.extend_from_slice(boundary.as_bytes()); outer_message.extend_from_slice(b"--\r\n"); } - ArchivedEncryptionMethod::SMIME => { + EncryptionMethod::SMIME => { // Generate random IV let mut rng = StdRng::from_entropy(); let mut iv = vec![0u8; 16]; rng.fill_bytes(&mut iv); // Generate random key - let mut key = vec![0u8; params.algo.key_size()]; + let mut key = vec![0u8; params.key_size()]; rng.fill_bytes(&mut key); // Encrypt contents (TODO: use rayon) - let algo = params.algo; + let algo = params.algo(); let (encrypted_contents, key, iv) = tokio::task::spawn_blocking(move || { (algo.encrypt(&key, &iv, &inner_message), key, iv) }) @@ -354,7 +359,7 @@ impl EncryptMessage for Message<'_> { encrypted_content_info: EncryptedContentInfo { content_type: CONTENT_DATA.into(), content_encryption_algorithm: AlgorithmIdentifier { - algorithm: params.algo.to_algorithm_identifier(), + algorithm: params.to_algorithm_identifier(), parameters: Some( rasn::der::encode(&OctetString::from(iv)) .map_err(|err| { @@ -457,43 +462,74 @@ impl EncryptMessage for Message<'_> { } } -impl ArchivedAlgorithm { +impl ArchivedEncryptionParams { + pub fn method(&self) -> EncryptionMethod { + if self.flags & ENCRYPT_METHOD_PGP != 0 { + EncryptionMethod::PGP + } else { + EncryptionMethod::SMIME + } + } + + pub fn algo(&self) -> Algorithm { + if self.flags & ENCRYPT_ALGO_AES256 != 0 { + Algorithm::Aes256 + } else { + Algorithm::Aes128 + } + } + fn key_size(&self) -> usize { - match self { - ArchivedAlgorithm::Aes128 => 16, - ArchivedAlgorithm::Aes256 => 32, + if self.flags & ENCRYPT_ALGO_AES256 != 0 { + 32 + } else { + 16 } } - fn to_algorithm_identifier(self) -> ObjectIdentifier { - match self { - ArchivedAlgorithm::Aes128 => AES128_CBC.into(), - ArchivedAlgorithm::Aes256 => AES256_CBC.into(), + fn to_algorithm_identifier(&self) -> ObjectIdentifier { + if self.flags & ENCRYPT_ALGO_AES256 != 0 { + AES256_CBC.into() + } else { + AES128_CBC.into() } } + pub fn can_train_spam_filter(&self) -> bool { + self.flags & ENCRYPT_TRAIN_SPAM_FILTER != 0 + } +} + +impl Algorithm { fn encrypt(&self, key: &[u8], iv: &[u8], contents: &[u8]) -> Vec { match self { - ArchivedAlgorithm::Aes128 => cbc::Encryptor::::new(key.into(), iv.into()) + Algorithm::Aes128 => cbc::Encryptor::::new(key.into(), iv.into()) .encrypt_padded_vec_mut::(contents), - ArchivedAlgorithm::Aes256 => cbc::Encryptor::::new(key.into(), iv.into()) + Algorithm::Aes256 => cbc::Encryptor::::new(key.into(), iv.into()) .encrypt_padded_vec_mut::(contents), } } } +#[allow(clippy::type_complexity)] pub fn try_parse_certs( expected_method: EncryptionMethod, cert: Vec, -) -> Result>, Cow<'static, str>> { +) -> Result]>, Cow<'static, str>> { // Check if it's a PEM file - let (method, certs) = if let Some(result) = try_parse_pem(&cert)? { - result + let (flags, certs) = if let Some(result) = try_parse_pem(&cert)? { + (result.flags, result.certs) } else if rasn::der::decode::(&cert[..]).is_ok() { - (EncryptionMethod::SMIME, vec![cert]) + ( + ENCRYPT_METHOD_SMIME, + Box::from_iter([cert.into_boxed_slice()]), + ) } else if let Ok(cert_) = openpgp::Cert::from_bytes(&cert[..]) { if !has_pgp_keys(cert_) { - (EncryptionMethod::PGP, vec![cert]) + ( + ENCRYPT_METHOD_PGP, + Box::from_iter([cert.into_boxed_slice()]), + ) } else { return Err("Could not find any suitable keys in certificate".into()); } @@ -501,7 +537,7 @@ pub fn try_parse_certs( return Err("Could not find any valid certificates".into()); }; - if method == expected_method { + if expected_method.flags() & flags != 0 { Ok(certs) } else { Err("No valid certificates found for the selected encryption".into()) @@ -520,9 +556,7 @@ fn has_pgp_keys(cert: openpgp::Cert) -> bool { } #[allow(clippy::type_complexity)] -fn try_parse_pem( - bytes_: &[u8], -) -> Result>)>, Cow<'static, str>> { +fn try_parse_pem(bytes_: &[u8]) -> Result, Cow<'static, str>> { if let Some(internal) = std::str::from_utf8(bytes_) .ok() .and_then(|cert| cert.strip_prefix("-----STALWART CERTIFICATE-----")) @@ -534,13 +568,13 @@ fn try_parse_pem( .and_then(|arch| arch.deserialize::()) .map_err(|_| Cow::from("Failed to deserialize internal certificate")) }) - .map(|params| Some((params.method, params.certs))); + .map(Some); } let mut bytes = bytes_.iter().enumerate(); let mut buf = vec![]; let mut method = None; - let mut certs = vec![]; + let mut certs: Vec> = vec![]; loop { // Find start of PEM block @@ -626,8 +660,9 @@ fn try_parse_pem( } // Decode base64 - let cert = - base64_decode(&buf).ok_or_else(|| Cow::from("Failed to decode base64 certificate."))?; + let cert = base64_decode(&buf) + .ok_or_else(|| Cow::from("Failed to decode base64 certificate."))? + .into_boxed_slice(); match method.unwrap() { EncryptionMethod::PGP => match openpgp::Cert::from_bytes(bytes_) { Ok(cert) => { @@ -638,7 +673,7 @@ fn try_parse_pem( bytes_ .get(start_pos..end_pos + 1) .unwrap_or_default() - .to_vec(), + .into(), ); } Err(err) => { @@ -655,7 +690,28 @@ fn try_parse_pem( buf.clear(); } - Ok(method.map(|method| (method, certs))) + Ok(method.map(|method| EncryptionParams { + flags: method.flags(), + certs: certs.into_boxed_slice(), + })) +} + +impl EncryptionMethod { + pub fn flags(&self) -> u64 { + match self { + EncryptionMethod::PGP => ENCRYPT_METHOD_PGP, + EncryptionMethod::SMIME => ENCRYPT_METHOD_SMIME, + } + } +} + +impl Algorithm { + pub fn flags(&self) -> u64 { + match self { + Algorithm::Aes128 => ENCRYPT_ALGO_AES128, + Algorithm::Aes256 => ENCRYPT_ALGO_AES256, + } + } } impl Display for EncryptionMethod { diff --git a/crates/email/src/message/delivery.rs b/crates/email/src/message/delivery.rs index b2717b79..35e6a125 100644 --- a/crates/email/src/message/delivery.rs +++ b/crates/email/src/message/delivery.rs @@ -20,12 +20,18 @@ use types::blob_hash::BlobHash; pub struct IngestMessage { pub sender_address: String, pub sender_authenticated: bool, - pub recipients: Vec, + pub recipients: Vec, pub message_blob: BlobHash, pub message_size: u64, pub session_id: u64, } +#[derive(Debug)] +pub struct IngestRecipient { + pub address: String, + pub is_spam: bool, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub enum LocalDeliveryStatus { Success, @@ -112,7 +118,11 @@ impl MailDelivery for Server { for rcpt in message.recipients { let account_id = match self - .email_to_id(&self.core.storage.directory, &rcpt, message.session_id) + .email_to_id( + &self.core.storage.directory, + &rcpt.address, + message.session_id, + ) .await { Ok(Some(account_id)) => account_id, @@ -127,7 +137,7 @@ impl MailDelivery for Server { Err(err) => { trc::error!( err.details("Failed to lookup recipient.") - .ctx(trc::Key::To, rcpt) + .ctx(trc::Key::To, rcpt.address.to_string()) .span_id(message.session_id) .caused_by(trc::location!()) ); @@ -165,12 +175,10 @@ impl MailDelivery for Server { keywords: vec![], received_at: None, source: IngestSource::Smtp { - deliver_to: &rcpt, + deliver_to: &rcpt.address, is_sender_authenticated: message.sender_authenticated, + is_spam: rcpt.is_spam, }, - spam_classify: access_token - .has_permission(Permission::SpamFilterClassify), - spam_train: self.email_bayes_can_train(&access_token), session_id: message.session_id, }) .await @@ -250,7 +258,7 @@ impl MailDelivery for Server { }; trc::error!( - err.ctx(trc::Key::To, rcpt.to_string()) + err.ctx(trc::Key::To, rcpt.address.to_string()) .span_id(message.session_id) ); diff --git a/crates/email/src/message/ingest.rs b/crates/email/src/message/ingest.rs index a59251bf..ee14bbec 100644 --- a/crates/email/src/message/ingest.rs +++ b/crates/email/src/message/ingest.rs @@ -6,8 +6,8 @@ use super::crypto::{EncryptMessage, EncryptMessageError}; use crate::{ - cache::{MessageCacheFetch, email::MessageCacheAccess}, - mailbox::{INBOX_ID, JUNK_ID, UidMailbox}, + cache::{MessageCacheFetch, email::MessageCacheAccess, mailbox::MailboxCacheAccess}, + mailbox::{INBOX_ID, JUNK_ID, SENT_ID, UidMailbox}, message::{ crypto::EncryptionParams, index::{IndexMessage, extractors::VisitText}, @@ -24,17 +24,14 @@ use mail_parser::{ Header, HeaderName, HeaderValue, Message, MessageParser, MimeHeaders, PartType, parsers::fields::thread::thread_name, }; -use spam_filter::{ - SpamFilterInput, analysis::init::SpamFilterInit, modules::bayes::BayesClassifier, -}; use std::{borrow::Cow, cmp::Ordering, fmt::Write, time::Instant}; use std::{future::Future, hash::Hasher}; use store::{ - IndexKeyPrefix, IterateParams, U32_LEN, ValueKey, + IndexKeyPrefix, IterateParams, SerializeInfallible, U32_LEN, ValueKey, ahash::{AHashMap, AHashSet}, write::{ - AssignedId, AssignedIds, BatchBuilder, IndexPropertyClass, SearchIndex, TaskEpoch, - TaskQueueClass, ValueClass, key::DeserializeBigEndian, now, + AssignedId, AssignedIds, BatchBuilder, BlobLink, BlobOp, IndexPropertyClass, SearchIndex, + TaskEpoch, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now, }, }; use trc::{AddContext, MessageIngestEvent}; @@ -44,6 +41,7 @@ use types::{ collection::{Collection, SyncCollection}, field::{ContactField, EmailField, MailboxField, PrincipalField}, keyword::Keyword, + special_use::SpecialUse, }; use utils::{cheeky_hash::CheekyHash, sanitize_email}; @@ -66,8 +64,6 @@ pub struct IngestEmail<'x> { pub keywords: Vec, pub received_at: Option, pub source: IngestSource<'x>, - pub spam_classify: bool, - pub spam_train: bool, pub session_id: u64, } @@ -76,9 +72,14 @@ pub enum IngestSource<'x> { Smtp { deliver_to: &'x str, is_sender_authenticated: bool, + is_spam: bool, + }, + Jmap { + train_classifier: bool, + }, + Imap { + train_classifier: bool, }, - Jmap, - Imap, Restore, } @@ -99,7 +100,6 @@ pub trait EmailIngest: Sync + Send { mailbox_ids: impl IntoIterator + Sync + Send, generate_email_id: bool, ) -> impl Future + 'static>> + Send; - fn email_bayes_can_train(&self, access_token: &AccessToken) -> bool; } pub struct ThreadResult { @@ -130,15 +130,92 @@ impl EmailIngest for Server { .ctx(trc::Key::Reason, "Failed to parse e-mail message.") })?; - let mut is_spam = false; + // Obtain message references and thread name + let mut message_id = None; + let mut message_ids = Vec::new(); + let thread_result = { + let mut subject = ""; + for header in message.root_part().headers().iter().rev() { + match &header.name { + HeaderName::MessageId => header.value.visit_text(|id| { + if !id.is_empty() { + if message_id.is_none() { + message_id = id.to_string().into(); + } + message_ids.push(CheekyHash::new(id.as_bytes())); + } + }), + HeaderName::InReplyTo + | HeaderName::References + | HeaderName::ResentMessageId => { + header.value.visit_text(|id| { + if !id.is_empty() { + message_ids.push(CheekyHash::new(id.as_bytes())); + } + }); + } + HeaderName::Subject if subject.is_empty() => { + subject = thread_name(match &header.value { + HeaderValue::Text(text) => text.as_ref(), + HeaderValue::TextList(list) if !list.is_empty() => { + list.first().unwrap().as_ref() + } + _ => "", + }); + } + _ => (), + } + } + + message_ids.sort_unstable(); + message_ids.dedup(); + + self.find_thread_id(account_id, subject, &message_ids) + .await? + }; + + // Skip duplicate messages for SMTP ingestion + if !thread_result.duplicate_ids.is_empty() && params.source.is_smtp() { + // Fetch cached messages + let cache = self + .get_cached_messages(account_id) + .await + .caused_by(trc::location!())?; + + // Skip duplicate messages + let target_mailbox_id = params.mailbox_ids.first().copied().unwrap_or(INBOX_ID); + if !cache + .in_mailboxes(&[target_mailbox_id, JUNK_ID]) + .any(|m| thread_result.duplicate_ids.contains(&m.document_id)) + { + trc::event!( + MessageIngest(MessageIngestEvent::Duplicate), + SpanId = params.session_id, + AccountId = account_id, + MessageId = message_id, + ); + + return Ok(IngestedEmail { + document_id: 0, + thread_id: 0, + change_id: u64::MAX, + blob_id: BlobId::default(), + imap_uids: Vec::new(), + size: 0, + }); + } + } + + // Spam classification and training let mut train_spam = None; let mut extra_headers = String::new(); let mut extra_headers_parsed = Vec::new(); let mut itip_messages = Vec::new(); - match params.source { + let is_spam = match params.source { IngestSource::Smtp { deliver_to, is_sender_authenticated, + mut is_spam, } => { // Add delivered to header if self.core.smtp.session.data.add_delivered_to { @@ -152,52 +229,11 @@ impl EmailIngest for Server { }); } - // Spam classification and training - if params.spam_classify - && self.core.spam.enabled - && params.mailbox_ids == [INBOX_ID] - { - // Set the spam filter result - #[cfg(not(feature = "test_mode"))] - { - is_spam = self - .core - .spam - .headers - .status - .as_ref() - .and_then(|name| { - message - .root_part() - .headers - .iter() - .find(|h| h.name.as_str().eq_ignore_ascii_case(name.as_str())) - .and_then(|v| v.value.as_text()) - }) - .is_some_and(|v| v.contains("Yes")); - } - - #[cfg(feature = "test_mode")] - { - is_spam = self - .core - .spam - .headers - .status - .as_ref() - .and_then(|name| { - message - .root_part() - .headers - .iter() - .rev() - .find(|h| h.name.as_str().eq_ignore_ascii_case(name.as_str())) - .and_then(|v| v.value.as_text()) - }) - .is_some_and(|v| v.contains("Yes")); - } - - // If the message is classified as spam, check whether the sender address is present in the user's address book + // Spam training on confirmed false positives + if self.core.spam.enabled { + let mut overridden = None; + // If the message is classified as spam, check whether the + // sender address is present in the user's address book. if is_spam && self.core.spam.card_is_ham && let Some(sender) = message @@ -218,72 +254,58 @@ impl EmailIngest for Server { .caused_by(trc::location!())? { is_spam = false; - if self - .core - .spam - .bayes - .as_ref() - .is_some_and(|config| config.auto_learn_card_is_ham) + train_spam = Some(false); + overridden = Some("card-found"); + } + + // Check if the message is a trusted reply to a previous message + if is_spam && let Some(thread_id) = thread_result.thread_id { + let cache = self + .get_cached_messages(account_id) + .await + .caused_by(trc::location!())?; + let sent_folder_id = cache + .mailbox_by_role(&SpecialUse::Sent) + .map(|m| m.document_id) + .unwrap_or(SENT_ID); + + if cache + .in_thread(thread_id) + .any(|m| m.mailboxes.iter().any(|mb| mb.mailbox_id == sent_folder_id)) { + is_spam = false; train_spam = Some(false); + overridden = Some("trusted-reply"); } } - // Classify the message with user's model - if let Some(bayes_config) = self.core.spam.bayes.as_ref().filter(|config| { - config.account_classify && params.spam_train && train_spam.is_none() - }) { - // Initialize spam filter - let ctx = self.spam_filter_init(SpamFilterInput::from_account_message( - &message, - account_id, - params.session_id, - )); - - // Bayes classify - match self.bayes_classify(&ctx).await { - Ok(Some(score)) => { - let result = if score > bayes_config.score_spam { - is_spam = true; - "Yes" - } else if score < bayes_config.score_ham { - is_spam = false; - "No" - } else { - "Unknown" - }; - - if let Some(header) = &self.core.spam.headers.bayes_result { - let offset_field = extra_headers.len(); - let offset_start = offset_field + header.len() + 1; - - let _ = write!( - &mut extra_headers, - "{header}: {result}, {score:.2}\r\n", - ); - - extra_headers_parsed.push(Header { - name: HeaderName::Other(header.into()), - value: HeaderValue::Text( - extra_headers - [offset_start + 1..extra_headers.len() - 2] - .to_string() - .into(), - ), - offset_field: offset_field as u32, - offset_start: offset_start as u32, - offset_end: extra_headers.len() as u32, - }); - } - } - Ok(None) => (), - Err(err) => { - trc::error!(err.caused_by(trc::location!())); - } - } + // Add Spam-Result header + const HEADER: &str = "X-Spam-Result"; + let offset_field = extra_headers.len(); + let offset_start = offset_field + HEADER.len() + 1; + let result = if is_spam { "Yes" } else { "No" }; + if let Some(reason) = overridden { + let _ = write!( + &mut extra_headers, + "{HEADER}: {result}, reason={reason}\r\n", + ); + } else { + let _ = write!(&mut extra_headers, "{HEADER}: {result}\r\n",); } - if is_spam { + extra_headers_parsed.push(Header { + name: HeaderName::Other(HEADER.into()), + value: HeaderValue::Text( + extra_headers[offset_start + 1..extra_headers.len() - 2] + .to_string() + .into(), + ), + offset_field: offset_field as u32, + offset_start: offset_start as u32, + offset_end: extra_headers.len() as u32, + }); + + if is_spam && params.mailbox_ids == [INBOX_ID] { params.mailbox_ids[0] = JUNK_ID; params.keywords.push(Keyword::Junk); } @@ -291,11 +313,11 @@ impl EmailIngest for Server { // iMIP processing if self.core.groupware.itip_enabled + && !is_spam + && is_sender_authenticated && params .access_token .has_permission(Permission::CalendarSchedulingReceive) - && is_sender_authenticated - && !is_spam { let mut sender = None; for part in &message.parts { @@ -376,10 +398,15 @@ impl EmailIngest for Server { } } } + + is_spam } - IngestSource::Jmap | IngestSource::Imap - if params.spam_train && self.core.spam.enabled => - { + IngestSource::Jmap { + train_classifier: true, + } + | IngestSource::Imap { + train_classifier: true, + } if self.core.spam.enabled => { if params.keywords.contains(&Keyword::Junk) { train_spam = Some(true); } else if params.keywords.contains(&Keyword::NotJunk) { @@ -389,95 +416,20 @@ impl EmailIngest for Server { } else if params.mailbox_ids[0] == INBOX_ID { train_spam = Some(false); } + false } - - _ => (), - } - - // Obtain message references and thread name - let mut message_id = None; - let mut message_ids = Vec::new(); - let thread_result = { - let mut subject = ""; - for header in message.root_part().headers().iter().rev() { - match &header.name { - HeaderName::MessageId => header.value.visit_text(|id| { - if !id.is_empty() { - if message_id.is_none() { - message_id = id.to_string().into(); - } - message_ids.push(CheekyHash::new(id.as_bytes())); - } - }), - HeaderName::InReplyTo - | HeaderName::References - | HeaderName::ResentMessageId => { - header.value.visit_text(|id| { - if !id.is_empty() { - message_ids.push(CheekyHash::new(id.as_bytes())); - } - }); - } - HeaderName::Subject if subject.is_empty() => { - subject = thread_name(match &header.value { - HeaderValue::Text(text) => text.as_ref(), - HeaderValue::TextList(list) if !list.is_empty() => { - list.first().unwrap().as_ref() - } - _ => "", - }); - } - _ => (), - } - } - - message_ids.sort_unstable(); - message_ids.dedup(); - - self.find_thread_id(account_id, subject, &message_ids) - .await? + _ => false, }; - // Skip duplicate messages for SMTP ingestion - if !thread_result.duplicate_ids.is_empty() && params.source.is_smtp() { - // Fetch cached messages - let cache = self - .get_cached_messages(account_id) - .await - .caused_by(trc::location!())?; - - // Skip duplicate messages - if !cache - .in_mailbox(params.mailbox_ids.first().copied().unwrap_or(INBOX_ID)) - .any(|m| thread_result.duplicate_ids.contains(&m.document_id)) - { - trc::event!( - MessageIngest(MessageIngestEvent::Duplicate), - SpanId = params.session_id, - AccountId = account_id, - MessageId = message_id, - ); - - return Ok(IngestedEmail { - document_id: 0, - thread_id: 0, - change_id: u64::MAX, - blob_id: BlobId::default(), - imap_uids: Vec::new(), - size: 0, - }); - } - } - // Encrypt message let do_encrypt = match params.source { - IngestSource::Jmap | IngestSource::Imap => { + IngestSource::Jmap { .. } | IngestSource::Imap { .. } => { self.core.jmap.encrypt && self.core.jmap.encrypt_append } IngestSource::Smtp { .. } => self.core.jmap.encrypt, IngestSource::Restore => false, }; - if do_encrypt + let is_encrypted = if do_encrypt && !message.is_encrypted() && let Some(encrypt_params_) = self .archive_by_property( @@ -506,7 +458,11 @@ impl EmailIngest for Server { "Failed to parse encrypted e-mail message.", ) })?; - params.blob_hash = None; + + // Disable spam training if requested + if !encrypt_params.can_train_spam_filter() { + train_spam = None; + } // Remove contents from parsed message for part in &mut message.parts { @@ -523,6 +479,8 @@ impl EmailIngest for Server { PartType::Multipart(_) => (), } } + + true } Err(EncryptMessageError::Error(err)) => { trc::bail!( @@ -534,10 +492,12 @@ impl EmailIngest for Server { } _ => unreachable!(), } - } + } else { + false + }; // Store blob - let (blob_hash, blob_hold) = if let Some(blob_hash) = params.blob_hash { + let (blob_hash, blob_hold) = if !is_encrypted && let Some(blob_hash) = params.blob_hash { (blob_hash.clone(), None) } else { self.put_temporary_blob(account_id, raw_message.as_ref(), 60) @@ -630,12 +590,28 @@ impl EmailIngest for Server { // Request spam training if let Some(learn_spam) = train_spam { + let blob_hash = params.blob_hash.unwrap_or(&blob_hash).clone(); + let blob_expiry = if is_encrypted { + let hold_for = now() + 86400; // 24 hours + batch.set( + BlobOp::Link { + hash: blob_hash.clone(), + to: BlobLink::Temporary { until: hold_for }, + }, + vec![], + ); + hold_for.serialize() + } else { + vec![] + }; + batch.set( - ValueClass::TaskQueue(TaskQueueClass::BayesTrain { + ValueClass::TaskQueue(TaskQueueClass::SpamTrain { due: TaskEpoch::now(), + blob_hash, learn_spam, }), - vec![], + blob_expiry, ); } @@ -665,8 +641,8 @@ impl EmailIngest for Server { } else { MessageIngestEvent::Spam }, - IngestSource::Jmap | IngestSource::Restore => MessageIngestEvent::JmapAppend, - IngestSource::Imap => MessageIngestEvent::ImapAppend, + IngestSource::Jmap { .. } | IngestSource::Restore => MessageIngestEvent::JmapAppend, + IngestSource::Imap { .. } => MessageIngestEvent::ImapAppend, }), SpanId = params.session_id, AccountId = account_id, @@ -834,12 +810,6 @@ impl EmailIngest for Server { .ctx(trc::Key::Reason, "No all document ids were generated")) } } - - fn email_bayes_can_train(&self, access_token: &AccessToken) -> bool { - self.core.spam.bayes.as_ref().is_some_and(|bayes| { - bayes.account_classify && access_token.has_permission(Permission::SpamFilterTrain) - }) - } } fn has_message_id(a: &[CheekyHash], b: &[u8]) -> bool { diff --git a/crates/email/src/sieve/ingest.rs b/crates/email/src/sieve/ingest.rs index fe4df68a..f4c79e95 100644 --- a/crates/email/src/sieve/ingest.rs +++ b/crates/email/src/sieve/ingest.rs @@ -9,12 +9,12 @@ use crate::{ cache::{MessageCacheFetch, mailbox::MailboxCacheAccess}, mailbox::{INBOX_ID, TRASH_ID, manage::MailboxFnc}, message::{ - delivery::AutogeneratedMessage, + delivery::{AutogeneratedMessage, IngestRecipient}, ingest::{EmailIngest, IngestEmail, IngestSource, IngestedEmail}, }, }; use common::{Server, auth::AccessToken, scripts::plugins::PluginContext}; -use directory::{Permission, QueryParams}; +use directory::QueryParams; use mail_parser::MessageParser; use sieve::{Envelope, Event, Input, Mailbox, Recipient, Sieve}; use std::{borrow::Cow, sync::Arc}; @@ -54,7 +54,7 @@ pub trait SieveScriptIngest: Sync + Send { raw_message: &[u8], envelope_from: &str, envelope_from_authenticated: bool, - envelope_to: &str, + envelope_to: &IngestRecipient, session_id: u64, active_script: ActiveScript, autogenerated: &mut Vec, @@ -92,7 +92,7 @@ impl SieveScriptIngest for Server { raw_message: &[u8], envelope_from: &str, envelope_from_authenticated: bool, - envelope_to: &str, + envelope_to: &IngestRecipient, session_id: u64, active_script: ActiveScript, autogenerated: &mut Vec, @@ -132,12 +132,12 @@ impl SieveScriptIngest for Server { }); // Set account address - let mail_from = mail_from.unwrap_or_else(|| envelope_to.into()); + let mail_from = mail_from.unwrap_or_else(|| envelope_to.address.as_str().into()); instance.set_user_address(&mail_from); // Set envelope instance.set_envelope(Envelope::From, envelope_from); - instance.set_envelope(Envelope::To, envelope_to); + instance.set_envelope(Envelope::To, envelope_to.address.as_str()); let mut input = Input::script( active_script.script_name.to_string(), @@ -499,7 +499,6 @@ impl SieveScriptIngest for Server { // Deliver messages let mut last_temp_error = None; let mut has_delivered = false; - let can_spam_train = self.email_bayes_can_train(access_token); for (message_id, sieve_message) in messages.into_iter().enumerate() { if !sieve_message.file_into.is_empty() { // Parse message if needed @@ -530,12 +529,10 @@ impl SieveScriptIngest for Server { keywords: sieve_message.flags, received_at: None, source: IngestSource::Smtp { - deliver_to: envelope_to, + deliver_to: envelope_to.address.as_str(), is_sender_authenticated: envelope_from_authenticated, + is_spam: envelope_to.is_spam, }, - spam_classify: access_token.has_permission(Permission::SpamFilterClassify) - && !sieve_message.did_file_into, - spam_train: can_spam_train, session_id, }) .await diff --git a/crates/http/src/form/mod.rs b/crates/http/src/form/mod.rs index 74532d2a..80556099 100644 --- a/crates/http/src/form/mod.rs +++ b/crates/http/src/form/mod.rs @@ -11,7 +11,7 @@ use common::{ config::network::{ContactForm, FieldOrDefault}, ip_to_bytes, psl, }; -use email::message::delivery::{IngestMessage, LocalDeliveryStatus, MailDelivery}; +use email::message::delivery::{IngestMessage, IngestRecipient, LocalDeliveryStatus, MailDelivery}; use http_proto::*; use hyper::StatusCode; use mail_auth::common::cache::NoCache; @@ -179,7 +179,14 @@ impl FormHandler for Server { .deliver_message(IngestMessage { sender_address: from_email, sender_authenticated: false, - recipients: form.rcpt_to.clone(), + recipients: form + .rcpt_to + .iter() + .map(|address| IngestRecipient { + address: address.clone(), + is_spam: false, + }) + .collect(), message_blob, message_size: message.len() as u64, session_id: session.session_id, diff --git a/crates/http/src/management/crypto.rs b/crates/http/src/management/crypto.rs index 3feae081..ffd17f74 100644 --- a/crates/http/src/management/crypto.rs +++ b/crates/http/src/management/crypto.rs @@ -7,8 +7,8 @@ use common::{Server, auth::AccessToken}; use directory::backend::internal::manage; use email::message::crypto::{ - Algorithm, ArchivedAlgorithm, ArchivedEncryptionMethod, EncryptMessage, EncryptMessageError, - EncryptionMethod, EncryptionParams, EncryptionType, try_parse_certs, + EncryptMessage, EncryptMessageError, EncryptionMethod, EncryptionParams, EncryptionType, + try_parse_certs, }; use http_proto::*; use mail_builder::encoders::base64::base64_encode_mime; @@ -49,14 +49,8 @@ impl CryptoHandler for Server { let params = params_ .unarchive::() .caused_by(trc::location!())?; - let algo = match ¶ms.algo { - ArchivedAlgorithm::Aes128 => Algorithm::Aes128, - ArchivedAlgorithm::Aes256 => Algorithm::Aes256, - }; - let method = match ¶ms.method { - ArchivedEncryptionMethod::PGP => EncryptionMethod::PGP, - ArchivedEncryptionMethod::SMIME => EncryptionMethod::SMIME, - }; + let algo = params.algo(); + let method = params.method(); let mut certs = Vec::new(); certs.extend_from_slice(b"-----STALWART CERTIFICATE-----\r\n"); let _ = base64_encode_mime(¶ms_.into_inner(), &mut certs, false); @@ -115,12 +109,12 @@ impl CryptoHandler for Server { } // Parse certificates + let todo = "fetch privacy spam train"; let certs = try_parse_certs(method, certs.into_bytes()) .map_err(|err| manage::error(err, None::))?; let num_certs = certs.len(); let params = Archiver::new(EncryptionParams { - method, - algo, + flags: method.flags() | algo.flags(), certs, }) .serialize() diff --git a/crates/http/src/management/enterprise/undelete.rs b/crates/http/src/management/enterprise/undelete.rs index 5092e1b7..45618b34 100644 --- a/crates/http/src/management/enterprise/undelete.rs +++ b/crates/http/src/management/enterprise/undelete.rs @@ -280,8 +280,6 @@ impl UndeleteApi for Server { keywords: vec![], received_at: request.time.into(), source: IngestSource::Restore, - spam_classify: false, - spam_train: false, session_id: session.session_id, }) .await diff --git a/crates/http/src/management/queue.rs b/crates/http/src/management/queue.rs index 3abeac15..ad40267b 100644 --- a/crates/http/src/management/queue.rs +++ b/crates/http/src/management/queue.rs @@ -384,8 +384,8 @@ impl QueueManagement for Server { for rcpt in &mut message.message.recipients { if rcpt.address().contains(item) { rcpt.status = Status::PermanentFailure(ErrorDetails { - entity: "localhost".to_string(), - details: queue::Error::Io("Delivery canceled.".to_string()), + entity: "localhost".into(), + details: queue::Error::Io("Delivery canceled.".into()), }); found = true; } diff --git a/crates/http/src/management/spam.rs b/crates/http/src/management/spam.rs index aec4be35..0e2fb3f1 100644 --- a/crates/http/src/management/spam.rs +++ b/crates/http/src/management/spam.rs @@ -23,7 +23,6 @@ use serde_json::json; use spam_filter::{ SpamFilterInput, analysis::{init::SpamFilterInit, score::SpamFilterAnalyzeScore}, - modules::bayes::BayesClassifier, }; use std::future::Future; use store::ahash::AHashMap; @@ -67,7 +66,7 @@ pub struct SpamClassifyRequest { #[serde(rename_all = "camelCase")] pub struct SpamClassifyResponse { pub score: f64, - pub tags: AHashMap>, + pub tags: AHashMap>, pub disposition: SpamFilterDisposition, } @@ -90,11 +89,12 @@ impl ManageSpamHandler for Server { access_token: &AccessToken, ) -> trc::Result { // Validate the access token - access_token.assert_has_permission(Permission::SpamFilterTrain)?; + //access_token.assert_has_permission(Permission::SpamFilterTrain)?; match (path.get(1).copied(), path.get(2).copied(), req.method()) { (Some("train"), Some(class @ ("ham" | "spam")), &Method::POST) => { - let message = parse_message_or_err(body.as_deref().unwrap_or_default())?; + let todo = "fix"; + /*let message = parse_message_or_err(body.as_deref().unwrap_or_default())?; let input = if let Some(account) = path.get(3).copied().filter(|a| !a.is_empty()) { let account_id = self .store() @@ -106,7 +106,7 @@ impl ManageSpamHandler for Server { SpamFilterInput::from_message(&message, session.session_id) }; self.bayes_train(&self.spam_filter_init(input), class == "spam", true) - .await?; + .await?; */ Ok(JsonResponse::new(json!({ "data": (), @@ -242,16 +242,16 @@ impl ManageSpamHandler for Server { env_from: &request.env_from, env_from_flags: request.env_from_flags, env_rcpt_to: request.env_rcpt_to.iter().map(String::as_str).collect(), - account_id: None, is_test: true, }; // Classify let mut ctx = self.spam_filter_init(input); let result = self.spam_filter_classify(&mut ctx).await; + let todo = "fix"; // Build response - let mut response = SpamClassifyResponse { + /* let mut response = SpamClassifyResponse { score: ctx.result.score, tags: AHashMap::with_capacity(ctx.result.tags.len()), disposition: match result { @@ -275,7 +275,8 @@ impl ManageSpamHandler for Server { Ok(JsonResponse::new(json!({ "data": response, })) - .into_http_response()) + .into_http_response())*/ + todo!() } _ => Err(trc::ResourceEvent::NotFound.into_err()), } diff --git a/crates/http/src/management/stores.rs b/crates/http/src/management/stores.rs index 11519872..bfd4a589 100644 --- a/crates/http/src/management/stores.rs +++ b/crates/http/src/management/stores.rs @@ -171,31 +171,7 @@ impl ManageStore for Server { } Some("rate-http-anonymous") => vec![KV_RATE_LIMIT_HTTP_ANONYMOUS].into(), Some("rate-imap") => vec![KV_RATE_LIMIT_IMAP].into(), - Some("reputation-ip") => vec![KV_REPUTATION_IP].into(), - Some("reputation-from") => vec![KV_REPUTATION_FROM].into(), - Some("reputation-domain") => vec![KV_REPUTATION_DOMAIN].into(), - Some("reputation-asn") => vec![KV_REPUTATION_ASN].into(), Some("greylist") => vec![KV_GREYLIST].into(), - Some("bayes-account") => { - if let Some(account) = path.get(5).copied() { - let account_id = self - .core - .storage - .data - .get_principal_id(decode_path_element(account).as_ref()) - .await? - .ok_or_else(|| trc::ManageEvent::NotFound.into_err())?; - - let mut key = Vec::with_capacity(std::mem::size_of::() + 1); - key.push(KV_BAYES_MODEL_USER); - key.extend_from_slice(&account_id.to_be_bytes()); - key.into() - } else { - vec![KV_BAYES_MODEL_USER].into() - } - } - Some("bayes-global") => vec![KV_BAYES_MODEL_GLOBAL].into(), - Some("trusted-reply") => vec![KV_TRUSTED_REPLY].into(), Some("lock-purge-account") => vec![KV_LOCK_PURGE_ACCOUNT].into(), Some("lock-queue-message") => vec![KV_LOCK_QUEUE_MESSAGE].into(), Some("lock-queue-report") => vec![KV_LOCK_QUEUE_REPORT].into(), @@ -514,23 +490,6 @@ pub async fn destroy_account_data( trc::error!(err.details("Failed to delete FTS index")); } } - - // Delete bayes model - if server - .core - .spam - .bayes - .as_ref() - .is_some_and(|c| c.account_classify) - { - let mut key = Vec::with_capacity(std::mem::size_of::() + 1); - key.push(KV_BAYES_MODEL_USER); - key.extend_from_slice(&account_id.to_be_bytes()); - - if let Err(err) = server.in_memory_store().key_delete_prefix(&key).await { - trc::error!(err.details("Failed to delete user bayes model")); - } - } } Ok(()) diff --git a/crates/imap/src/op/append.rs b/crates/imap/src/op/append.rs index 0ece0a01..159c53bf 100644 --- a/crates/imap/src/op/append.rs +++ b/crates/imap/src/op/append.rs @@ -94,7 +94,6 @@ impl SessionData { .get_access_token(mailbox.account_id) .await .imap_ctx(&arguments.tag, trc::location!())?; - let spam_train = self.server.email_bayes_can_train(&access_token); // Append messages let mut response = StatusResponse::completed(Command::Append); @@ -111,9 +110,9 @@ impl SessionData { mailbox_ids: vec![mailbox_id], keywords: message.flags.into_iter().map(Keyword::from).collect(), received_at: message.received_at.map(|d| d as u64), - source: IngestSource::Imap, - spam_classify: false, - spam_train, + source: IngestSource::Imap { + train_classifier: true, + }, session_id: self.session_id, }) .await diff --git a/crates/imap/src/op/copy_move.rs b/crates/imap/src/op/copy_move.rs index ebe2f38a..9fe14269 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -204,7 +204,6 @@ impl SessionData { // Mailboxes are in the same account let account_id = src_mailbox.id.account_id; let dest_mailbox_id = UidMailbox::new_unassigned(dest_mailbox_id); - let can_spam_train = self.server.email_bayes_can_train(&access_token); let mut has_spam_train_tasks = false; let mut batch = BatchBuilder::new(); @@ -321,28 +320,28 @@ impl SessionData { } // Add bayes train task - if can_spam_train { - if dest_mailbox_id.mailbox_id == JUNK_ID { - batch.set( - ValueClass::TaskQueue(TaskQueueClass::BayesTrain { - due: TaskEpoch::now(), - learn_spam: true, - }), - vec![], - ); - has_spam_train_tasks = true; - } else if src_mailbox.id.mailbox_id == JUNK_ID - && dest_mailbox_id.mailbox_id != TRASH_ID - { - batch.set( - ValueClass::TaskQueue(TaskQueueClass::BayesTrain { - due: TaskEpoch::now(), - learn_spam: false, - }), - vec![], - ); - has_spam_train_tasks = true; - } + if dest_mailbox_id.mailbox_id == JUNK_ID { + batch.set( + ValueClass::TaskQueue(TaskQueueClass::SpamTrain { + due: TaskEpoch::now(), + blob_hash: Default::default(), + learn_spam: true, + }), + vec![], + ); + has_spam_train_tasks = true; + } else if src_mailbox.id.mailbox_id == JUNK_ID + && dest_mailbox_id.mailbox_id != TRASH_ID + { + batch.set( + ValueClass::TaskQueue(TaskQueueClass::SpamTrain { + due: TaskEpoch::now(), + blob_hash: Default::default(), + learn_spam: false, + }), + vec![], + ); + has_spam_train_tasks = true; } batch.commit_point(); diff --git a/crates/imap/src/op/store.rs b/crates/imap/src/op/store.rs index 51ba5374..ceee118e 100644 --- a/crates/imap/src/op/store.rs +++ b/crates/imap/src/op/store.rs @@ -12,7 +12,7 @@ use crate::{ use ahash::AHashSet; use common::{listener::SessionStream, storage::index::ObjectIndexBuilder}; use directory::Permission; -use email::message::{ingest::EmailIngest, metadata::MessageData}; +use email::message::metadata::MessageData; use imap_proto::{ Command, ResponseCode, ResponseType, StatusResponse, protocol::{ @@ -188,13 +188,7 @@ impl SessionData { .iter() .map(|k| Keyword::from(k.clone())) .collect::>(); - let access_token = self - .server - .get_access_token(account_id) - .await - .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; let mut changed_mailboxes = AHashSet::new(); - let can_spam_train = self.server.email_bayes_can_train(&access_token); let mut has_spam_train_tasks = false; let mut batch = BatchBuilder::new(); @@ -247,25 +241,23 @@ impl SessionData { // Train spam filter let mut train_spam = None; - if can_spam_train { - for keyword in new_data.added_keywords(data.inner) { + for keyword in new_data.added_keywords(data.inner) { + if keyword == &Keyword::Junk { + train_spam = Some(true); + break; + } else if keyword == &Keyword::NotJunk { + train_spam = Some(false); + break; + } + } + if train_spam.is_none() { + for keyword in new_data.removed_keywords(data.inner) { if keyword == &Keyword::Junk { - train_spam = Some(true); - break; - } else if keyword == &Keyword::NotJunk { train_spam = Some(false); break; } } - if train_spam.is_none() { - for keyword in new_data.removed_keywords(data.inner) { - if keyword == &Keyword::Junk { - train_spam = Some(false); - break; - } - } - } - }; + } // Convert keywords to flags let flags = if !arguments.is_silent { @@ -301,8 +293,9 @@ impl SessionData { // Add spam train task if let Some(learn_spam) = train_spam { batch.set( - ValueClass::TaskQueue(TaskQueueClass::BayesTrain { + ValueClass::TaskQueue(TaskQueueClass::SpamTrain { due: TaskEpoch::now(), + blob_hash: Default::default(), learn_spam, }), vec![], diff --git a/crates/jmap/src/email/import.rs b/crates/jmap/src/email/import.rs index 2040ddf8..59ba5031 100644 --- a/crates/jmap/src/email/import.rs +++ b/crates/jmap/src/email/import.rs @@ -77,7 +77,6 @@ impl EmailImport for Server { created: VecMap::with_capacity(request.emails.len()), not_created: VecMap::new(), }; - let can_train_spam = self.email_bayes_can_train(access_token); 'outer: for (id, email) in request.emails { // Validate mailboxIds @@ -153,9 +152,9 @@ impl EmailImport for Server { mailbox_ids, keywords: email.keywords, received_at: email.received_at.map(|r| r.into()), - source: IngestSource::Jmap, - spam_classify: false, - spam_train: can_train_spam, + source: IngestSource::Jmap { + train_classifier: true, + }, session_id: session.session_id, }) .await diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index 2ef1916a..c3f37b6e 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -74,7 +74,6 @@ impl EmailSet for Server { let cache = self.get_cached_messages(account_id).await?; let mut response = SetResponse::from_request(&request, self.core.jmap.set_max_objects)? .with_state(cache.assert_state(false, &request.if_in_state)?); - let can_train_spam = self.email_bayes_can_train(access_token); // Obtain mailboxIds let (can_add_mailbox_ids, can_delete_mailbox_ids, can_modify_mailbox_ids) = @@ -758,9 +757,9 @@ impl EmailSet for Server { mailbox_ids: mailboxes, keywords, received_at, - source: IngestSource::Jmap, - spam_classify: false, - spam_train: can_train_spam, + source: IngestSource::Jmap { + train_classifier: true, + }, session_id: session.session_id, }) .await @@ -874,6 +873,7 @@ impl EmailSet for Server { } // Process keywords + let todo = "train spam classifier"; if has_keyword_changes { // Verify permissions on shared accounts if can_modify_mailbox_ids.as_ref().is_some_and(|ids| { diff --git a/crates/jmap/src/submission/get.rs b/crates/jmap/src/submission/get.rs index 16284675..25379e74 100644 --- a/crates/jmap/src/submission/get.rs +++ b/crates/jmap/src/submission/get.rs @@ -292,7 +292,7 @@ fn build_address( .into() } -fn format_archived_response(response: &ArchivedResponse) -> String { +fn format_archived_response(response: &ArchivedResponse>) -> String { format!( "Code: {}, Enhanced code: {}.{}.{}, Message: {}", response.code, diff --git a/crates/services/src/task_manager/bayes.rs b/crates/services/src/task_manager/bayes.rs index ec4c8dd9..6365bed2 100644 --- a/crates/services/src/task_manager/bayes.rs +++ b/crates/services/src/task_manager/bayes.rs @@ -7,14 +7,12 @@ use common::Server; use email::message::metadata::MessageMetadata; use mail_parser::MessageParser; -use spam_filter::{ - SpamFilterInput, analysis::init::SpamFilterInit, modules::bayes::BayesClassifier, -}; +use spam_filter::{SpamFilterInput, analysis::init::SpamFilterInit}; use std::time::Instant; use trc::{SpamEvent, TaskQueueEvent}; use types::{collection::Collection, field::EmailField}; -pub trait BayesTrainTask: Sync + Send { +pub trait SpamTrainTask: Sync + Send { fn bayes_train( &self, account_id: u32, @@ -23,7 +21,7 @@ pub trait BayesTrainTask: Sync + Send { ) -> impl Future + Send; } -impl BayesTrainTask for Server { +impl SpamTrainTask for Server { async fn bayes_train(&self, account_id: u32, document_id: u32, learn_spam: bool) -> bool { let op_start = Instant::now(); // Obtain metadata @@ -68,7 +66,8 @@ impl BayesTrainTask for Server { { Ok(Some(raw_message)) => { // Train bayes classifier for account - self.bayes_train_if_balanced( + let todo = "fix"; + /*self.bayes_train_if_balanced( &self.spam_filter_init(SpamFilterInput::from_account_message( &MessageParser::new().parse(&raw_message).unwrap_or_default(), account_id, @@ -76,7 +75,7 @@ impl BayesTrainTask for Server { )), learn_spam, ) - .await; + .await;*/ trc::event!( Spam(SpamEvent::TrainAccount), diff --git a/crates/services/src/task_manager/lock.rs b/crates/services/src/task_manager/lock.rs index befc429d..b7a941e6 100644 --- a/crates/services/src/task_manager/lock.rs +++ b/crates/services/src/task_manager/lock.rs @@ -139,8 +139,11 @@ impl TaskLock for Task { } fn value_classes(&self) -> impl Iterator { - std::iter::once(ValueClass::TaskQueue(TaskQueueClass::BayesTrain { + let todo = "fix"; + + std::iter::once(ValueClass::TaskQueue(TaskQueueClass::SpamTrain { due: self.due, + blob_hash: Default::default(), learn_spam: self.action, })) } @@ -260,7 +263,7 @@ impl Task { pub(crate) fn lock_expiry(&self) -> u64 { match &self.action { TaskAction::UpdateIndex(_) => INDEX_EXPIRY, - TaskAction::BayesTrain(_) => BAYES_LOCK_EXPIRY, + TaskAction::SpamTrain(_) => BAYES_LOCK_EXPIRY, TaskAction::SendAlarm(_) => ALARM_EXPIRY, _ => ALARM_EXPIRY, } @@ -282,7 +285,7 @@ impl Task { .ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?, is_insert: *v == 7, }), - Some(v @ (1 | 2)) => TaskAction::BayesTrain(*v == 1), + Some(v @ (1 | 2)) => TaskAction::SpamTrain(*v == 1), Some(3) => TaskAction::SendAlarm(CalendarAlarm { event_id: key.deserialize_be_u16(U64_LEN + U32_LEN + U32_LEN + 1)?, alarm_id: key.deserialize_be_u16(U64_LEN + U32_LEN + U32_LEN + U16_LEN + 1)?, diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index 726bce0a..48a6e757 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::task_manager::bayes::BayesTrainTask; +use crate::task_manager::bayes::SpamTrainTask; use crate::task_manager::imip::SendImipTask; use crate::task_manager::index::SearchIndexTask; use crate::task_manager::lock::{TaskLock, TaskLockManager}; @@ -56,7 +56,7 @@ pub struct Task { #[derive(Debug, Clone, Hash, PartialEq, Eq)] pub enum TaskAction { UpdateIndex(IndexAction), - BayesTrain(bool), + SpamTrain(bool), SendAlarm(CalendarAlarm), SendImip, MergeThreads(MergeThreadIds>), @@ -475,7 +475,7 @@ impl TaskQueueManager for Server { ); } } - TaskAction::BayesTrain(learn_spam) + TaskAction::SpamTrain(learn_spam) if roles.bayes_training.is_enabled_for_hash(&event) => { if ipc @@ -608,7 +608,7 @@ impl TaskAction { pub fn name(&self) -> &'static str { match self { TaskAction::UpdateIndex(_) => "UpdateIndex", - TaskAction::BayesTrain(_) => "BayesTrain", + TaskAction::SpamTrain(_) => "SpamTrain", TaskAction::SendAlarm(_) => "SendAlarm", TaskAction::SendImip => "SendImip", TaskAction::MergeThreads(_) => "MergeThreads", diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index 00763140..80ddb229 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -8,7 +8,10 @@ use super::{ArcSeal, AuthResult, DkimSign}; use crate::{ core::{Session, SessionAddress, State}, inbound::milter::Modification, - queue::{self, Message, MessageSource, MessageWrapper, QueueEnvelope, quota::HasQueueQuota}, + queue::{ + self, Message, MessageSource, MessageWrapper, QueueEnvelope, RCPT_SPAM_PAYLOAD, + quota::HasQueueQuota, + }, reporting::analysis::AnalyzeReport, scripts::ScriptResult, }; @@ -422,6 +425,7 @@ impl Session { } // Run SPAM filter + let mut train_as_spam = false; if self.server.core.spam.enabled && self .server @@ -439,9 +443,18 @@ impl Session { ) .await { - SpamFilterAction::Allow(spam_headers) => { - if !spam_headers.is_empty() { - headers.extend_from_slice(spam_headers.as_bytes()); + SpamFilterAction::Allow(score) => { + // Add headers + headers.extend_from_slice(score.headers.as_bytes()); + train_as_spam = score.spam_trap; + + // Add scores for local recipients + for (is_spam, recipient) in + score.results.into_iter().zip(self.data.rcpt_to.iter_mut()) + { + if is_spam { + recipient.flags |= RCPT_SPAM_PAYLOAD; + } } } SpamFilterAction::Discard => { @@ -453,6 +466,7 @@ impl Session { return (b"550 5.7.1 Message rejected due to excessive spam score.\r\n"[..]) .into(); } + SpamFilterAction::Disabled => {} } } @@ -665,19 +679,22 @@ impl Session { // Queue message let source = if !self.is_authenticated() { - let is_dmarc_authenticated = - dmarc_result.is_some_and(|result| result == DmarcResult::Pass); + let dmarc_pass = dmarc_result.is_some_and(|result| result == DmarcResult::Pass); #[cfg(feature = "test_mode")] { - MessageSource::Unauthenticated( - is_dmarc_authenticated || message.message.return_path.starts_with("dmarc-"), - ) + MessageSource::Unauthenticated { + dmarc_pass: dmarc_pass || message.message.return_path.starts_with("dmarc-"), + train_as_spam, + } } #[cfg(not(feature = "test_mode"))] { - MessageSource::Unauthenticated(is_dmarc_authenticated) + MessageSource::Unauthenticated { + dmarc_pass, + train_as_spam, + } } } else { MessageSource::Authenticated @@ -718,12 +735,12 @@ impl Session { .map_or(0, |d| d.as_secs()); let mut message = Message { created, - return_path: mail_from.address.to_lowercase_domain(), + return_path: mail_from.address.to_lowercase_domain().into_boxed_str(), recipients: Vec::with_capacity(rcpt_to.len()), flags: mail_from.flags, priority: self.data.priority, size: 0, - env_id: mail_from.dsn_info, + env_id: mail_from.dsn_info.map(|i| i.into_boxed_str()), blob_hash: Default::default(), quota_keys: Vec::new(), received_from_ip: self.data.remote_ip, @@ -749,7 +766,7 @@ impl Session { rcpt.flags | RCPT_NOTIFY_DELAY | RCPT_NOTIFY_FAILURE }, ) - .with_orcpt(rcpt.dsn_info), + .with_orcpt(rcpt.dsn_info.map(|v| v.into_boxed_str())), ); let envelope = QueueEnvelope::new(&message, message.recipients.last().unwrap()); diff --git a/crates/smtp/src/inbound/rcpt.rs b/crates/smtp/src/inbound/rcpt.rs index 0c4f6ead..532f889f 100644 --- a/crates/smtp/src/inbound/rcpt.rs +++ b/crates/smtp/src/inbound/rcpt.rs @@ -290,8 +290,7 @@ impl Session { .server .core .spam - .expiry - .grey_list + .grey_list_expiry .filter(|_| self.data.authenticated_as.is_none()) { let from_addr = self diff --git a/crates/smtp/src/inbound/spam.rs b/crates/smtp/src/inbound/spam.rs index 5b980dc1..bcd1f0c3 100644 --- a/crates/smtp/src/inbound/spam.rs +++ b/crates/smtp/src/inbound/spam.rs @@ -10,8 +10,8 @@ use mail_parser::Message; use spam_filter::{ SpamFilterInput, analysis::{ - init::SpamFilterInit, score::SpamFilterAnalyzeScore, - trusted_reply::SpamFilterAnalyzeTrustedReply, + init::SpamFilterInit, + score::{SpamFilterAnalyzeScore, SpamFilterScore}, }, }; @@ -25,7 +25,7 @@ impl Session { arc_result: Option<&'x ArcOutput<'x>>, dmarc_result: Option<&'x DmarcResult>, dmarc_policy: Option<&'x Policy>, - ) -> SpamFilterAction { + ) -> SpamFilterAction { let server = &self.server; let mut ctx = server.spam_filter_init(self.build_spam_input( message, @@ -39,9 +39,8 @@ impl Session { // Spam classification server.spam_filter_classify(&mut ctx).await } else { - // Trusted reply tracking - server.spam_filter_analyze_reply_out(&mut ctx).await; - SpamFilterAction::Allow(String::new()) + // Do not classify authenticated sessions + SpamFilterAction::Disabled } } @@ -87,7 +86,6 @@ impl Session { .iter() .map(|r| r.address_lcase.as_str()) .collect(), - account_id: None, is_test: false, } } diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index 90f63faf..afc3f44b 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -6,14 +6,13 @@ #![warn(clippy::large_futures)] -use std::sync::Arc; - use common::{ Inner, manager::boot::{BootManager, IpcReceivers}, }; use queue::manager::SpawnQueue; use reporting::scheduler::SpawnReport; +use std::sync::Arc; pub mod core; pub mod inbound; diff --git a/crates/smtp/src/outbound/client.rs b/crates/smtp/src/outbound/client.rs index a031c21b..e86b857a 100644 --- a/crates/smtp/src/outbound/client.rs +++ b/crates/smtp/src/outbound/client.rs @@ -128,7 +128,7 @@ impl SmtpClient { pub async fn read_greeting( &mut self, hostname: &str, - ) -> Result<(), Status, ErrorDetails>> { + ) -> Result<(), Status>, ErrorDetails>> { tokio::time::timeout(self.timeout, self.read()) .await .map_err(|_| Status::timeout(hostname, "reading greeting"))? @@ -140,7 +140,7 @@ impl SmtpClient { &mut self, hostname: &str, bdat_cmd: &Option, - ) -> Result, Status, ErrorDetails>> { + ) -> Result, Status>, ErrorDetails>> { tokio::time::timeout(self.timeout, self.read()) .await .map_err(|_| Status::timeout(hostname, "reading SMTP DATA response"))? @@ -153,7 +153,7 @@ impl SmtpClient { &mut self, hostname: &str, num_responses: usize, - ) -> Result>, Status, ErrorDetails>> { + ) -> Result>>, Status>, ErrorDetails>> { tokio::time::timeout(self.timeout, async { self.read_many(num_responses).await }) .await .map_err(|_| Status::timeout(hostname, "reading LMTP DATA responses"))? @@ -175,7 +175,7 @@ impl SmtpClient { message: &MessageWrapper, bdat_cmd: &Option, params: &SessionParams<'_>, - ) -> Result<(), Status, ErrorDetails>> { + ) -> Result<(), Status>, ErrorDetails>> { match params .server .blob_store() @@ -227,7 +227,7 @@ impl SmtpClient { CausedBy = trc::location!() ); Err(Status::TemporaryFailure(ErrorDetails { - entity: "localhost".to_string(), + entity: "localhost".into(), details: Error::Io("Queue system error.".into()), })) } @@ -239,7 +239,7 @@ impl SmtpClient { ); Err(Status::TemporaryFailure(ErrorDetails { - entity: "localhost".to_string(), + entity: "localhost".into(), details: Error::Io("Queue system error.".into()), })) } @@ -249,7 +249,7 @@ impl SmtpClient { pub async fn say_helo( &mut self, params: &SessionParams<'_>, - ) -> Result, Status, ErrorDetails>> { + ) -> Result, Status>, ErrorDetails>> { let cmd = if params.is_smtp { format!("EHLO {}\r\n", params.local_hostname) } else { @@ -377,7 +377,7 @@ impl SmtpClient { } } - pub async fn read_many(&mut self, num: usize) -> mail_send::Result>> { + pub async fn read_many(&mut self, num: usize) -> mail_send::Result>>> { let mut buf = vec![0u8; 1024]; let mut response = Vec::with_capacity(num); let mut parser = ResponseReceiver::default(); @@ -398,7 +398,7 @@ impl SmtpClient { loop { match parser.parse(&mut iter) { Ok(reply) => { - response.push(reply); + response.push(reply.into_box()); if response.len() != num { parser.reset(); } else { @@ -590,7 +590,7 @@ impl SmtpClient { } } else { StartTlsResult::Unavailable { - response: response.into(), + response: response.into_box().into(), smtp_client: self, } } @@ -621,11 +621,25 @@ pub enum StartTlsResult { error: mail_send::Error, }, Unavailable { - response: Option>, + response: Option>>, smtp_client: SmtpClient, }, } +pub(crate) trait BoxResponse { + fn into_box(self) -> Response>; +} + +impl BoxResponse for Response { + fn into_box(self) -> Response> { + Response { + code: self.code, + esc: self.esc, + message: self.message.into_boxed_str(), + } + } +} + pub(crate) fn from_mail_send_error(error: &mail_send::Error) -> trc::Error { let event = trc::EventType::Smtp(trc::SmtpEvent::Error).into_err(); match error { @@ -654,7 +668,7 @@ pub(crate) fn from_mail_send_error(error: &mail_send::Error) -> trc::Error { } } -pub(crate) fn from_error_status(err: &Status, ErrorDetails>) -> trc::Error { +pub(crate) fn from_error_status(err: &Status>, ErrorDetails>) -> trc::Error { match err { Status::Scheduled | Status::Completed(_) => { trc::EventType::Smtp(trc::SmtpEvent::Error).into_err() diff --git a/crates/smtp/src/outbound/dane/verify.rs b/crates/smtp/src/outbound/dane/verify.rs index 68feb461..0921ad9e 100644 --- a/crates/smtp/src/outbound/dane/verify.rs +++ b/crates/smtp/src/outbound/dane/verify.rs @@ -19,7 +19,7 @@ pub trait TlsaVerify { session_id: u64, hostname: &str, certificates: Option<&[CertificateDer<'_>]>, - ) -> Result<(), Status, ErrorDetails>>; + ) -> Result<(), Status>, ErrorDetails>>; } impl TlsaVerify for Tlsa { @@ -28,7 +28,7 @@ impl TlsaVerify for Tlsa { session_id: u64, hostname: &str, certificates: Option<&[CertificateDer<'_>]>, - ) -> Result<(), Status, ErrorDetails>> { + ) -> Result<(), Status>, ErrorDetails>> { let certificates = if let Some(certificates) = certificates { certificates } else { diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index 5a2a02a3..408f2ddc 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -59,7 +59,7 @@ impl QueuedMessage { QueueId = message.queue_id, QueueName = message.queue_name.to_string(), From = if !message.message.return_path.is_empty() { - trc::Value::String(message.message.return_path.as_str().into()) + trc::Value::String(message.message.return_path.as_ref().into()) } else { trc::Value::String("<>".into()) }, @@ -202,7 +202,7 @@ impl QueuedMessage { { rcpt.retry.due = retry_at; rcpt.status = Status::TemporaryFailure(ErrorDetails { - entity: "localhost".to_string(), + entity: "localhost".into(), details: Error::RateLimited, }); } @@ -553,7 +553,7 @@ impl QueuedMessage { delivery_results.push(DeliveryResult::domain( Status::PermanentFailure(ErrorDetails { - entity: domain.to_string(), + entity: domain.into(), details: Error::DnsError( "Domain does not accept messages (null MX)".into(), ), @@ -565,7 +565,7 @@ impl QueuedMessage { } // Try delivering message - let mut last_status: Status, ErrorDetails> = Status::Scheduled; + let mut last_status: Status>, ErrorDetails> = Status::Scheduled; 'next_host: for remote_host in &remote_hosts { // Validate MTA-STS envelope.mx = remote_host.hostname(); @@ -603,11 +603,11 @@ impl QueuedMessage { if strict { last_status = Status::PermanentFailure(ErrorDetails { - entity: envelope.mx.to_string(), - details: Error::MtaStsError(format!( - "MX {:?} not authorized by policy.", - envelope.mx - )), + entity: envelope.mx.into(), + details: Error::MtaStsError( + format!("MX {:?} not authorized by policy.", envelope.mx) + .into_boxed_str(), + ), }); continue 'next_host; } @@ -721,7 +721,7 @@ impl QueuedMessage { if strict { last_status = Status::PermanentFailure(ErrorDetails { - entity: envelope.mx.to_string(), + entity: envelope.mx.into(), details: Error::DaneError( "No valid TLSA records were found".into(), ), @@ -1103,7 +1103,7 @@ impl QueuedMessage { Code = response.as_ref().map(|r| r.code()), Details = response .as_ref() - .map(|r| r.message().as_str()) + .map(|r| r.message().as_ref()) .unwrap_or("STARTTLS was not advertised by host") .to_string(), Elapsed = time.elapsed(), @@ -1356,7 +1356,7 @@ impl MessageWrapper { ); rcpt.status = Status::PermanentFailure(ErrorDetails { - entity: rcpt.domain_part().to_string(), + entity: rcpt.domain_part().into(), details: Error::Io( "Message expired without any delivery attempts made.".into(), ), @@ -1379,7 +1379,7 @@ impl MessageWrapper { pub async fn set_rcpt_status( &mut self, - status: Status, ErrorDetails>, + status: Status>, ErrorDetails>, rcpt_idx: usize, server: &Server, ) { @@ -1408,7 +1408,7 @@ impl MessageWrapper { let rcpt = &mut self.message.recipients[rcpt_idx]; rcpt.retry.due = retry_at; rcpt.status = Status::TemporaryFailure(ErrorDetails { - entity: "localhost".to_string(), + entity: "localhost".into(), details: Error::RateLimited, }); } diff --git a/crates/smtp/src/outbound/local.rs b/crates/smtp/src/outbound/local.rs index 79930264..dfc96ad7 100644 --- a/crates/smtp/src/outbound/local.rs +++ b/crates/smtp/src/outbound/local.rs @@ -8,13 +8,13 @@ use crate::{ outbound::DeliveryResult, queue::{ Error, ErrorDetails, FROM_AUTHENTICATED, FROM_UNAUTHENTICATED_DMARC, HostResponse, - MessageSource, MessageWrapper, Status, UnexpectedResponse, quota::HasQueueQuota, - spool::SmtpSpool, + MessageSource, MessageWrapper, RCPT_SPAM_PAYLOAD, Status, UnexpectedResponse, + quota::HasQueueQuota, spool::SmtpSpool, }, reporting::SmtpReporting, }; use common::Server; -use email::message::delivery::{IngestMessage, LocalDeliveryStatus, MailDelivery}; +use email::message::delivery::{IngestMessage, IngestRecipient, LocalDeliveryStatus, MailDelivery}; use smtp_proto::Response; use trc::SieveEvent; @@ -27,21 +27,25 @@ impl MessageWrapper { ) { // Prepare recipients list let mut pending_recipients = Vec::new(); - let mut recipient_addresses = Vec::new(); + let mut recipients = Vec::new(); for &rcpt_idx in rcpt_idxs { - let rcpt_addr = self.message.recipients[rcpt_idx].address(); - recipient_addresses.push(rcpt_addr.to_lowercase()); + let rcpt = &self.message.recipients[rcpt_idx]; + let rcpt_addr = rcpt.address(); + recipients.push(IngestRecipient { + address: rcpt_addr.to_lowercase(), + is_spam: rcpt.flags & RCPT_SPAM_PAYLOAD != 0, + }); pending_recipients.push((rcpt_idx, rcpt_addr)); } // Deliver message let delivery_result = server .deliver_message(IngestMessage { - sender_address: self.message.return_path.clone(), + sender_address: self.message.return_path.to_string(), sender_authenticated: self.message.flags & (FROM_UNAUTHENTICATED_DMARC | FROM_AUTHENTICATED) != 0, - recipients: recipient_addresses, + recipients, message_blob: self.message.blob_hash.clone(), message_size: self.message.size, session_id: self.span_id, @@ -65,7 +69,7 @@ impl MessageWrapper { Status::TemporaryFailure(ErrorDetails { entity: "localhost".into(), details: Error::UnexpectedResponse(UnexpectedResponse { - command: format!("RCPT TO:<{rcpt_addr}>"), + command: format!("RCPT TO:<{rcpt_addr}>").into_boxed_str(), response: Response { code: 451, esc: [4, 3, 0], @@ -78,7 +82,7 @@ impl MessageWrapper { Status::PermanentFailure(ErrorDetails { entity: "localhost".into(), details: Error::UnexpectedResponse(UnexpectedResponse { - command: format!("RCPT TO:<{rcpt_addr}>"), + command: format!("RCPT TO:<{rcpt_addr}>").into_boxed_str(), response: Response { code: 550, esc: code, diff --git a/crates/smtp/src/outbound/lookup.rs b/crates/smtp/src/outbound/lookup.rs index e33e722e..f7e5c998 100644 --- a/crates/smtp/src/outbound/lookup.rs +++ b/crates/smtp/src/outbound/lookup.rs @@ -31,7 +31,7 @@ pub trait DnsLookup: Sync + Send { &self, remote_host: &NextHop<'_>, envelope: &impl ResolveVariable, - ) -> impl Future, ErrorDetails>>> + Send; + ) -> impl Future>, ErrorDetails>>> + Send; } impl DnsLookup for Server { @@ -109,7 +109,7 @@ impl DnsLookup for Server { &self, remote_host: &NextHop<'_>, envelope: &impl ResolveVariable, - ) -> Result, ErrorDetails>> { + ) -> Result>, ErrorDetails>> { let mut remote_ips = self .ip_lookup( remote_host.fqdn_hostname().as_ref(), @@ -139,7 +139,9 @@ impl DnsLookup for Server { } else { Status::TemporaryFailure(ErrorDetails { entity: remote_host.hostname().into(), - details: Error::ConnectionError(format!("lookup error: {err}")), + details: Error::ConnectionError( + format!("lookup error: {err}").into_boxed_str(), + ), }) } })?; @@ -160,10 +162,13 @@ impl DnsLookup for Server { } else { Err(Status::TemporaryFailure(ErrorDetails { entity: remote_host.hostname().into(), - details: Error::DnsError(format!( - "No IP addresses found for {:?}.", - envelope.resolve_variable(V_MX).to_string() - )), + details: Error::DnsError( + format!( + "No IP addresses found for {:?}.", + envelope.resolve_variable(V_MX).to_string() + ) + .into_boxed_str(), + ), })) } } diff --git a/crates/smtp/src/outbound/mod.rs b/crates/smtp/src/outbound/mod.rs index 504955ae..978c2f4b 100644 --- a/crates/smtp/src/outbound/mod.rs +++ b/crates/smtp/src/outbound/mod.rs @@ -4,7 +4,10 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::queue::{Error, ErrorDetails, HostResponse, Status, UnexpectedResponse}; +use crate::{ + outbound::client::BoxResponse, + queue::{Error, ErrorDetails, HostResponse, Status, UnexpectedResponse}, +}; use common::config::{ server::ServerProtocol, smtp::queue::{MxConfig, RelayConfig}, @@ -24,11 +27,11 @@ pub mod session; pub(super) enum DeliveryResult { Domain { - status: Status, ErrorDetails>, + status: Status>, ErrorDetails>, rcpt_idxs: Vec, }, Account { - status: Status, ErrorDetails>, + status: Status>, ErrorDetails>, rcpt_idx: usize, }, RateLimited { @@ -37,7 +40,7 @@ pub(super) enum DeliveryResult { }, } -impl Status, ErrorDetails> { +impl Status>, ErrorDetails> { pub fn from_smtp_error(hostname: &str, command: &str, err: mail_send::Error) -> Self { match err { mail_send::Error::Io(_) @@ -50,7 +53,7 @@ impl Status, ErrorDetails> { | mail_send::Error::MissingRcptTo | mail_send::Error::Timeout => Status::TemporaryFailure(ErrorDetails { entity: hostname.into(), - details: Error::ConnectionError(err.to_string()), + details: Error::ConnectionError(err.to_string().into_boxed_str()), }), mail_send::Error::UnexpectedReply(response) => { @@ -59,7 +62,7 @@ impl Status, ErrorDetails> { entity: hostname.into(), details: Error::UnexpectedResponse(UnexpectedResponse { command: command.trim().into(), - response, + response: response.into_box(), }), }) } else { @@ -67,7 +70,7 @@ impl Status, ErrorDetails> { entity: hostname.into(), details: Error::UnexpectedResponse(UnexpectedResponse { command: command.trim().into(), - response, + response: response.into_box(), }), }) } @@ -78,12 +81,12 @@ impl Status, ErrorDetails> { | mail_send::Error::InvalidTLSName | mail_send::Error::MissingStartTls => Status::PermanentFailure(ErrorDetails { entity: hostname.into(), - details: Error::ConnectionError(err.to_string()), + details: Error::ConnectionError(err.to_string().into_boxed_str()), }), } } - pub fn from_starttls_error(hostname: &str, response: Option>) -> Self { + pub fn from_starttls_error(hostname: &str, response: Option>>) -> Self { let entity = hostname.into(); if let Some(response) = response { if response.severity() == Severity::PermanentNegativeCompletion { @@ -123,11 +126,11 @@ impl Status, ErrorDetails> { }), mail_send::Error::Tls(err) => Status::TemporaryFailure(ErrorDetails { entity: hostname.into(), - details: Error::TlsError(format!("Handshake failed: {err}")), + details: Error::TlsError(format!("Handshake failed: {err}").into_boxed_str()), }), mail_send::Error::Io(err) => Status::TemporaryFailure(ErrorDetails { entity: hostname.into(), - details: Error::TlsError(format!("I/O error: {err}")), + details: Error::TlsError(format!("I/O error: {err}").into_boxed_str()), }), _ => Status::PermanentFailure(ErrorDetails { entity: hostname.into(), @@ -139,7 +142,7 @@ impl Status, ErrorDetails> { pub fn timeout(hostname: &str, stage: &str) -> Self { Status::TemporaryFailure(ErrorDetails { entity: hostname.into(), - details: Error::ConnectionError(format!("Timeout while {stage}")), + details: Error::ConnectionError(format!("Timeout while {stage}").into_boxed_str()), }) } @@ -153,12 +156,12 @@ impl Status, ErrorDetails> { pub fn from_mail_auth_error(entity: &str, err: mail_auth::Error) -> Self { match &err { mail_auth::Error::DnsRecordNotFound(code) => Status::PermanentFailure(ErrorDetails { - entity: entity.to_string(), - details: Error::DnsError(format!("Domain not found: {code:?}")), + entity: entity.into(), + details: Error::DnsError(format!("Domain not found: {code:?}").into_boxed_str()), }), _ => Status::TemporaryFailure(ErrorDetails { - entity: entity.to_string(), - details: Error::DnsError(err.to_string()), + entity: entity.into(), + details: Error::DnsError(err.to_string().into_boxed_str()), }), } } @@ -168,28 +171,32 @@ impl Status, ErrorDetails> { mta_sts::Error::Dns(err) => match err { mail_auth::Error::DnsRecordNotFound(code) => { Status::PermanentFailure(ErrorDetails { - entity: entity.to_string(), - details: Error::MtaStsError(format!("Record not found: {code:?}")), + entity: entity.into(), + details: Error::MtaStsError( + format!("Record not found: {code:?}").into_boxed_str(), + ), }) } mail_auth::Error::InvalidRecordType => Status::PermanentFailure(ErrorDetails { - entity: entity.to_string(), + entity: entity.into(), details: Error::MtaStsError("Failed to parse MTA-STS DNS record.".into()), }), _ => Status::TemporaryFailure(ErrorDetails { - entity: entity.to_string(), - details: Error::MtaStsError(format!("DNS lookup error: {err}")), + entity: entity.into(), + details: Error::MtaStsError( + format!("DNS lookup error: {err}").into_boxed_str(), + ), }), }, mta_sts::Error::Http(err) => { if err.is_timeout() { Status::TemporaryFailure(ErrorDetails { - entity: entity.to_string(), + entity: entity.into(), details: Error::MtaStsError("Timeout fetching policy.".into()), }) } else if err.is_connect() { Status::TemporaryFailure(ErrorDetails { - entity: entity.to_string(), + entity: entity.into(), details: Error::MtaStsError("Could not reach policy host.".into()), }) } else if err.is_status() @@ -198,19 +205,21 @@ impl Status, ErrorDetails> { .is_some_and(|s| s == reqwest::StatusCode::NOT_FOUND) { Status::PermanentFailure(ErrorDetails { - entity: entity.to_string(), + entity: entity.into(), details: Error::MtaStsError("Policy not found.".into()), }) } else { Status::TemporaryFailure(ErrorDetails { - entity: entity.to_string(), + entity: entity.into(), details: Error::MtaStsError("Failed to fetch policy.".into()), }) } } mta_sts::Error::InvalidPolicy(err) => Status::PermanentFailure(ErrorDetails { - entity: entity.to_string(), - details: Error::MtaStsError(format!("Failed to parse policy: {err}")), + entity: entity.into(), + details: Error::MtaStsError( + format!("Failed to parse policy: {err}").into_boxed_str(), + ), }), } } @@ -322,7 +331,7 @@ impl NextHop<'_> { impl DeliveryResult { pub fn domain( - status: Status, ErrorDetails>, + status: Status>, ErrorDetails>, rcpt_idxs: Vec, ) -> Self { DeliveryResult::Domain { status, rcpt_idxs } @@ -335,7 +344,7 @@ impl DeliveryResult { } } - pub fn account(status: Status, ErrorDetails>, rcpt_idx: usize) -> Self { + pub fn account(status: Status>, ErrorDetails>, rcpt_idx: usize) -> Self { DeliveryResult::Account { status, rcpt_idx } } } diff --git a/crates/smtp/src/outbound/session.rs b/crates/smtp/src/outbound/session.rs index b47589e7..55d0b2b4 100644 --- a/crates/smtp/src/outbound/session.rs +++ b/crates/smtp/src/outbound/session.rs @@ -6,7 +6,7 @@ use super::client::SmtpClient; use crate::outbound::DeliveryResult; -use crate::outbound::client::{from_error_status, from_mail_send_error}; +use crate::outbound::client::{BoxResponse, from_error_status, from_mail_send_error}; use crate::queue::{Error, MessageWrapper, Recipient, Status}; use crate::queue::{ErrorDetails, HostResponse, UnexpectedResponse}; use common::Server; @@ -189,7 +189,7 @@ impl MessageWrapper { rcpt_idx, Status::Completed(HostResponse { hostname: params.hostname.into(), - response, + response: response.into_box(), }), )); } @@ -208,7 +208,7 @@ impl MessageWrapper { entity: params.hostname.into(), details: Error::UnexpectedResponse(UnexpectedResponse { command: cmd.trim().into(), - response, + response: response.into_box(), }), }; statuses.push(DeliveryResult::account( @@ -331,48 +331,54 @@ impl MessageWrapper { for ((rcpt, rcpt_idx, _), response) in accepted_rcpts.into_iter().zip(responses) { - let status = match response.severity() { - Severity::PositiveCompletion => { - trc::event!( - Delivery(DeliveryEvent::Delivered), - SpanId = params.session_id, - Hostname = params.hostname.to_string(), - To = rcpt.address().to_string(), - Code = response.code, - Details = response.message.to_string(), - Elapsed = time.elapsed(), - ); + let status: Status>, ErrorDetails> = + match response.severity() { + Severity::PositiveCompletion => { + trc::event!( + Delivery(DeliveryEvent::Delivered), + SpanId = params.session_id, + Hostname = params.hostname.to_string(), + To = rcpt.address().to_string(), + Code = response.code, + Details = response.message.to_string(), + Elapsed = time.elapsed(), + ); - Status::Completed(HostResponse { - hostname: params.hostname.to_string(), - response, - }) - } - severity => { - trc::event!( - Delivery(DeliveryEvent::RcptToRejected), - SpanId = params.session_id, - Hostname = params.hostname.to_string(), - To = rcpt.address().to_string(), - Code = response.code, - Details = response.message.to_string(), - Elapsed = time.elapsed(), - ); - - let response = ErrorDetails { - entity: params.hostname.into(), - details: Error::UnexpectedResponse(UnexpectedResponse { - command: bdat_cmd.as_deref().unwrap_or("DATA").into(), + Status::Completed(HostResponse { + hostname: params.hostname.into(), response, - }), - }; - if severity == Severity::PermanentNegativeCompletion { - Status::PermanentFailure(response) - } else { - Status::TemporaryFailure(response) + }) } - } - }; + severity => { + trc::event!( + Delivery(DeliveryEvent::RcptToRejected), + SpanId = params.session_id, + Hostname = params.hostname.to_string(), + To = rcpt.address().to_string(), + Code = response.code, + Details = response.message.to_string(), + Elapsed = time.elapsed(), + ); + + let response = ErrorDetails { + entity: params.hostname.into(), + details: Error::UnexpectedResponse( + UnexpectedResponse { + command: bdat_cmd + .as_deref() + .unwrap_or("DATA") + .into(), + response, + }, + ), + }; + if severity == Severity::PermanentNegativeCompletion { + Status::PermanentFailure(response) + } else { + Status::TemporaryFailure(response) + } + } + }; statuses.push(DeliveryResult::account(status, *rcpt_idx)); } diff --git a/crates/smtp/src/queue/dsn.rs b/crates/smtp/src/queue/dsn.rs index 99a87c45..8985e4a0 100644 --- a/crates/smtp/src/queue/dsn.rs +++ b/crates/smtp/src/queue/dsn.rs @@ -7,7 +7,7 @@ use super::spool::SmtpSpool; use super::{ Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope, RCPT_DSN_SENT, - RCPT_STATUS_CHANGED, Recipient, Status, + Recipient, Status, }; use crate::queue::{MessageWrapper, UnexpectedResponse}; use crate::reporting::SmtpReporting; @@ -39,7 +39,7 @@ impl SendDsn for Server { if let Some(dsn) = message.build_dsn(self).await { let mut dsn_message = self.new_message("", message.span_id); dsn_message - .add_recipient(message.message.return_path.as_str(), self) + .add_recipient(message.message.return_path.as_ref(), self) .await; // Sign message @@ -144,7 +144,7 @@ impl MessageWrapper { } match &rcpt.status { Status::Completed(response) => { - rcpt.flags |= RCPT_DSN_SENT | RCPT_STATUS_CHANGED; + rcpt.flags |= RCPT_DSN_SENT; if !rcpt.has_flag(RCPT_NOTIFY_SUCCESS) { continue; } @@ -161,7 +161,7 @@ impl MessageWrapper { response.write_dsn_text(&rcpt.address, &mut txt_delay); } Status::PermanentFailure(response) => { - rcpt.flags |= RCPT_DSN_SENT | RCPT_STATUS_CHANGED; + rcpt.flags |= RCPT_DSN_SENT; if !rcpt.has_flag(RCPT_NOTIFY_FAILURE) { continue; } @@ -369,7 +369,7 @@ impl MessageWrapper { .from((from_name.as_str(), from_addr.as_str())) .header( "To", - HeaderType::Text(self.message.return_path.as_str().into()), + HeaderType::Text(self.message.return_path.as_ref().into()), ) .header("Auto-Submitted", HeaderType::Text("auto-generated".into())) .message_id(format!("<{}@{}>", make_boundary("."), reporting_mta)) @@ -425,7 +425,7 @@ impl MessageWrapper { } } -impl HostResponse { +impl HostResponse> { fn write_dsn_text(&self, addr: &str, dsn: &mut String) { let _ = write!( dsn, @@ -464,7 +464,7 @@ impl UnexpectedResponse { impl ErrorDetails { fn write_dsn_text(&self, addr: &str, dsn: &mut String) { - let entity = self.entity.as_str(); + let entity = self.entity.as_ref(); match &self.details { Error::UnexpectedResponse(response) => { response.write_dsn_text(entity, addr, dsn); @@ -571,7 +571,7 @@ impl Status { } } -impl Status, ErrorDetails> { +impl Status>, ErrorDetails> { fn write_dsn(&self, dsn: &mut String) { self.write_dsn_action(dsn); self.write_dsn_status(dsn); @@ -634,7 +634,7 @@ impl Status, ErrorDetails> { } } -impl WriteDsn for Response { +impl WriteDsn for Response> { fn write_dsn_status(&self, dsn: &mut String) { if self.esc[0] > 0 { let _ = write!(dsn, "{}.{}.{}", self.esc[0], self.esc[1], self.esc[2]); diff --git a/crates/smtp/src/queue/mod.rs b/crates/smtp/src/queue/mod.rs index 33003674..07a8d2db 100644 --- a/crates/smtp/src/queue/mod.rs +++ b/crates/smtp/src/queue/mod.rs @@ -43,7 +43,10 @@ pub struct QueuedMessage { #[derive(Debug, Clone, Copy)] pub enum MessageSource { Authenticated, - Unauthenticated(bool), + Unauthenticated { + dmarc_pass: bool, + train_as_spam: bool, + }, Dsn, Report, Autogenerated, @@ -54,14 +57,14 @@ pub struct Message { pub created: u64, pub blob_hash: BlobHash, - pub return_path: String, + pub return_path: Box, pub recipients: Vec, pub received_from_ip: IpAddr, pub received_via_port: u16, pub flags: u64, - pub env_id: Option, + pub env_id: Option>, pub priority: i16, pub size: u64, @@ -103,16 +106,16 @@ pub enum QuotaKey { serde::Deserialize, )] pub struct Recipient { - pub address: String, + pub address: Box, pub retry: Schedule, pub notify: Schedule, pub expires: QueueExpiry, pub queue: QueueName, - pub status: Status, ErrorDetails>, + pub status: Status>, ErrorDetails>, pub flags: u64, - pub orcpt: Option, + pub orcpt: Option>, } pub const FROM_AUTHENTICATED: u64 = 1 << 32; @@ -123,7 +126,8 @@ pub const FROM_REPORT: u64 = 1 << 36; pub const FROM_AUTOGENERATED: u64 = 1 << 37; pub const RCPT_DSN_SENT: u64 = 1 << 32; -pub const RCPT_STATUS_CHANGED: u64 = 1 << 33; +//pub const RCPT_STATUS_CHANGED: u64 = 1 << 33; +pub const RCPT_SPAM_PAYLOAD: u64 = 1 << 34; #[derive( Debug, @@ -159,7 +163,7 @@ pub enum Status { )] pub struct HostResponse { pub hostname: T, - pub response: Response, + pub response: Response>, } #[derive( @@ -174,16 +178,16 @@ pub struct HostResponse { Default, )] pub enum Error { - DnsError(String), + DnsError(Box), UnexpectedResponse(UnexpectedResponse), - ConnectionError(String), - TlsError(String), - DaneError(String), - MtaStsError(String), + ConnectionError(Box), + TlsError(Box), + DaneError(Box), + MtaStsError(Box), RateLimited, #[default] ConcurrencyLimited, - Io(String), + Io(Box), } #[derive( @@ -197,8 +201,8 @@ pub enum Error { serde::Deserialize, )] pub struct UnexpectedResponse { - pub command: String, - pub response: Response, + pub command: Box, + pub response: Response>, } #[derive( @@ -213,7 +217,7 @@ pub struct UnexpectedResponse { serde::Deserialize, )] pub struct ErrorDetails { - pub entity: String, + pub entity: Box, pub details: Error, } @@ -278,15 +282,15 @@ impl<'x> QueueEnvelope<'x> { impl<'x> ResolveVariable for QueueEnvelope<'x> { fn resolve_variable(&self, variable: u32) -> expr::Variable<'x> { match variable { - V_SENDER => self.message.return_path.as_str().into(), + V_SENDER => self.message.return_path.as_ref().into(), V_SENDER_DOMAIN => self.message.return_path.domain_part().into(), V_RECIPIENT_DOMAIN => self.domain.into(), - V_RECIPIENT => self.rcpt.address.as_str().into(), + V_RECIPIENT => self.rcpt.address.as_ref().into(), V_RECIPIENTS => self .message .recipients .iter() - .map(|r| Variable::from(r.address.as_str())) + .map(|r| Variable::from(r.address.as_ref())) .collect::>() .into(), V_QUEUE_RETRY_NUM => self.rcpt.retry.inner.into(), @@ -353,12 +357,12 @@ impl<'x> ResolveVariable for QueueEnvelope<'x> { impl ResolveVariable for Message { fn resolve_variable(&self, variable: u32) -> expr::Variable<'_> { match variable { - V_SENDER => self.return_path.as_str().into(), + V_SENDER => self.return_path.as_ref().into(), V_SENDER_DOMAIN => self.return_path.domain_part().into(), V_RECIPIENTS => self .recipients .iter() - .map(|r| Variable::from(r.address.as_str())) + .map(|r| Variable::from(r.address.as_ref())) .collect::>() .into(), V_PRIORITY => self.priority.into(), @@ -403,7 +407,7 @@ pub fn instant_to_timestamp(now: Instant, time: Instant) -> u64 { impl Recipient { pub fn new(address: impl AsRef) -> Self { Recipient { - address: address.to_lowercase_domain(), + address: address.to_lowercase_domain().into_boxed_str(), status: Status::Scheduled, flags: 0, orcpt: None, @@ -419,7 +423,7 @@ impl Recipient { self } - pub fn with_orcpt(mut self, orcpt: Option) -> Self { + pub fn with_orcpt(mut self, orcpt: Option>) -> Self { self.orcpt = orcpt; self } @@ -435,7 +439,7 @@ impl Recipient { impl ArchivedRecipient { pub fn address(&self) -> &str { - self.address.as_str() + self.address.as_ref() } pub fn domain_part(&self) -> &str { @@ -537,7 +541,7 @@ impl Display for ArchivedError { } } -impl Display for Status, ErrorDetails> { +impl Display for Status>, ErrorDetails> { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Status::Scheduled => write!(f, "Scheduled"), @@ -564,7 +568,7 @@ pub trait DisplayArchivedResponse { fn to_string(&self) -> String; } -impl DisplayArchivedResponse for ArchivedResponse { +impl DisplayArchivedResponse for ArchivedResponse> { fn to_string(&self) -> String { format!( "Code: {}, Enhanced code: {}.{}.{}, Message: {}", diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 59b891f1..c65fc2ef 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -25,9 +25,9 @@ use store::write::key::DeserializeBigEndian; use store::write::serialize::rkyv_deserialize; use store::write::{ AlignedBytes, Archive, Archiver, BatchBuilder, BlobLink, BlobOp, MergeResult, Params, - QueueClass, ValueClass, now, + QueueClass, TaskEpoch, TaskQueueClass, ValueClass, now, }; -use store::{Deserialize, IterateParams, Serialize, U64_LEN, ValueKey}; +use store::{Deserialize, IterateParams, Serialize, SerializeInfallible, U64_LEN, ValueKey}; use trc::{AddContext, ServerEvent}; use types::blob_hash::BlobHash; use utils::DomainPart; @@ -83,7 +83,7 @@ impl SmtpSpool for Server { span_id, message: Message { created, - return_path: return_path.to_lowercase_domain(), + return_path: return_path.to_lowercase_domain().into_boxed_str(), recipients: Vec::with_capacity(1), flags: 0, env_id: None, @@ -302,22 +302,35 @@ impl MessageWrapper { source: MessageSource, ) -> bool { // Set flags - let (flags, event) = match source { + let (flags, event, train_spam) = match source { MessageSource::Authenticated => ( FROM_AUTHENTICATED, trc::QueueEvent::QueueMessageAuthenticated, + false, + ), + MessageSource::Unauthenticated { + dmarc_pass: true, + train_as_spam, + } => ( + FROM_UNAUTHENTICATED_DMARC, + trc::QueueEvent::QueueMessage, + train_as_spam, + ), + MessageSource::Unauthenticated { + dmarc_pass: false, + train_as_spam, + } => ( + FROM_UNAUTHENTICATED, + trc::QueueEvent::QueueMessage, + train_as_spam, + ), + MessageSource::Dsn => (FROM_DSN, trc::QueueEvent::QueueDsn, false), + MessageSource::Report => (FROM_REPORT, trc::QueueEvent::QueueReport, false), + MessageSource::Autogenerated => ( + FROM_AUTOGENERATED, + trc::QueueEvent::QueueAutogenerated, + false, ), - MessageSource::Unauthenticated(true) => { - (FROM_UNAUTHENTICATED_DMARC, trc::QueueEvent::QueueMessage) - } - MessageSource::Unauthenticated(false) => { - (FROM_UNAUTHENTICATED, trc::QueueEvent::QueueMessage) - } - MessageSource::Dsn => (FROM_DSN, trc::QueueEvent::QueueDsn), - MessageSource::Report => (FROM_REPORT, trc::QueueEvent::QueueReport), - MessageSource::Autogenerated => { - (FROM_AUTOGENERATED, trc::QueueEvent::QueueAutogenerated) - } }; self.message.flags |= flags; @@ -339,7 +352,7 @@ impl MessageWrapper { // Reserve and write blob let mut batch = BatchBuilder::new(); - let reserve_until = now() + 120; + let reserve_until = now() + if !train_spam { 120 } else { 86400 }; batch.set( BlobOp::Link { hash: self.message.blob_hash.clone(), @@ -377,7 +390,7 @@ impl MessageWrapper { SpanId = session_id, QueueId = self.queue_id, From = if !self.message.return_path.is_empty() { - trc::Value::String(self.message.return_path.as_str().into()) + trc::Value::String(self.message.return_path.as_ref().into()) } else { trc::Value::String("<>".into()) }, @@ -385,7 +398,7 @@ impl MessageWrapper { .message .recipients .iter() - .map(|r| trc::Value::String(r.address.as_str().into())) + .map(|r| trc::Value::String(r.address.as_ref().into())) .collect::>(), Size = self.message.size, NextRetry = self @@ -425,13 +438,25 @@ impl MessageWrapper { ); } - batch - .clear(BlobOp::Link { + if !train_spam { + batch.clear(BlobOp::Link { hash: self.message.blob_hash.clone(), to: BlobLink::Temporary { until: reserve_until, }, - }) + }); + } else { + batch.set( + ValueClass::TaskQueue(TaskQueueClass::SpamTrain { + due: TaskEpoch::now(), + blob_hash: self.message.blob_hash.clone(), + learn_spam: true, + }), + reserve_until.serialize(), + ); + } + + batch .set( BlobOp::Link { hash: self.message.blob_hash.clone(), diff --git a/crates/spam-filter/src/analysis/bayes.rs b/crates/spam-filter/src/analysis/classifier.rs similarity index 72% rename from crates/spam-filter/src/analysis/bayes.rs rename to crates/spam-filter/src/analysis/classifier.rs index a2da8a13..b279630e 100644 --- a/crates/spam-filter/src/analysis/bayes.rs +++ b/crates/spam-filter/src/analysis/classifier.rs @@ -4,14 +4,12 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::{SpamFilterContext, modules::classifier::SpamClassifier}; +use common::Server; use std::future::Future; -use common::Server; - -use crate::{SpamFilterContext, modules::bayes::BayesClassifier}; - -pub trait SpamFilterAnalyzeBayes: Sync + Send { - fn spam_filter_analyze_bayes_classify( +pub trait SpamFilterAnalyzeClassify: Sync + Send { + fn spam_filter_analyze_classify( &self, ctx: &mut SpamFilterContext<'_>, ) -> impl Future + Send; @@ -22,19 +20,18 @@ pub trait SpamFilterAnalyzeBayes: Sync + Send { ) -> impl Future + Send; } -impl SpamFilterAnalyzeBayes for Server { - async fn spam_filter_analyze_bayes_classify(&self, ctx: &mut SpamFilterContext<'_>) { - if let Some(config) = &self.core.spam.bayes - && !ctx.result.has_tag("SPAM_TRAP") - && !ctx.result.has_tag("TRUSTED_REPLY") +impl SpamFilterAnalyzeClassify for Server { + async fn spam_filter_analyze_classify(&self, ctx: &mut SpamFilterContext<'_>) { + if self.core.spam.classifier.is_some() && !ctx.result.has_tag("SPAM_TRAP") + //&& !ctx.result.has_tag("TRUSTED_REPLY") { - match self.bayes_classify(ctx).await { + match self.spam_classify(ctx).await { Ok(Some(score)) => { - if score > config.score_spam { + /*if score > config.score_spam { ctx.result.add_tag("BAYES_SPAM"); } else if score < config.score_ham { ctx.result.add_tag("BAYES_HAM"); - } + }*/ } Ok(None) => (), Err(err) => { @@ -50,6 +47,7 @@ impl SpamFilterAnalyzeBayes for Server { match store.key_exists(addr.address.as_str()).await { Ok(true) => { ctx.result.add_tag("SPAM_TRAP"); + ctx.result.spam_trap = true; return true; } Ok(false) => (), diff --git a/crates/spam-filter/src/analysis/domain.rs b/crates/spam-filter/src/analysis/domain.rs index 1fa0c057..c783d70b 100644 --- a/crates/spam-filter/src/analysis/domain.rs +++ b/crates/spam-filter/src/analysis/domain.rs @@ -4,17 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::{collections::HashSet, future::Future}; - -use common::{ - Server, - config::spamfilter::{Element, Location}, -}; -use compact_str::CompactString; -use mail_auth::DkimResult; -use mail_parser::{HeaderName, HeaderValue, Host, parsers::MessageStream}; -use nlp::tokenizers::types::TokenType; - +use super::{ElementLocation, is_trusted_domain}; use crate::{ Email, Hostname, Recipient, SpamFilterContext, TextPart, modules::{ @@ -23,8 +13,14 @@ use crate::{ html::{A, HREF, HtmlToken}, }, }; - -use super::{ElementLocation, is_trusted_domain}; +use common::{ + Server, + config::spamfilter::{Element, Location}, +}; +use mail_auth::DkimResult; +use mail_parser::{HeaderName, HeaderValue, Host, parsers::MessageStream}; +use nlp::tokenizers::types::TokenType; +use std::{collections::HashSet, future::Future}; pub trait SpamFilterAnalyzeDomain: Sync + Send { fn spam_filter_analyze_domain( @@ -36,7 +32,7 @@ pub trait SpamFilterAnalyzeDomain: Sync + Send { impl SpamFilterAnalyzeDomain for Server { async fn spam_filter_analyze_domain(&self, ctx: &mut SpamFilterContext<'_>) { // Obtain email addresses and domains - let mut domains: HashSet> = HashSet::new(); + let mut domains: HashSet> = HashSet::new(); let mut emails: HashSet> = HashSet::new(); // Add DKIM domains @@ -45,7 +41,7 @@ impl SpamFilterAnalyzeDomain for Server { && let Some(domain) = dkim.signature().map(|s| &s.d) { domains.insert(ElementLocation::new( - CompactString::from_str_to_lowercase(domain), + domain.to_lowercase(), Location::HeaderDkimPass, )); } diff --git a/crates/spam-filter/src/analysis/ehlo.rs b/crates/spam-filter/src/analysis/ehlo.rs index f66861e4..2c2de143 100644 --- a/crates/spam-filter/src/analysis/ehlo.rs +++ b/crates/spam-filter/src/analysis/ehlo.rs @@ -32,7 +32,7 @@ impl SpamFilterAnalyzeEhlo for Server { .output .iprev_ptr .as_ref() - .is_some_and(|ptr| ptr != ctx.output.ehlo_host.fqdn) + .is_some_and(|ptr| *ptr != ctx.output.ehlo_host.fqdn) { // Helo does not match reverse IP ctx.result.add_tag("HELO_IPREV_MISMATCH"); diff --git a/crates/spam-filter/src/analysis/init.rs b/crates/spam-filter/src/analysis/init.rs index 91edc534..ed669627 100644 --- a/crates/spam-filter/src/analysis/init.rs +++ b/crates/spam-filter/src/analysis/init.rs @@ -5,7 +5,7 @@ */ use common::Server; -use compact_str::CompactString; + use mail_parser::{HeaderName, PartType, parsers::fields::thread::thread_name}; use nlp::tokenizers::types::{TokenType, TypesTokenizer}; @@ -42,7 +42,7 @@ impl SpamFilterInit for Server { name: addr.name().and_then(|s| { let s = s.trim(); if !s.is_empty() { - Some(CompactString::from_str_to_lowercase(s)) + Some(s.to_lowercase()) } else { None } @@ -69,7 +69,7 @@ impl SpamFilterInit for Server { name: addr.name().and_then(|s| { let s = s.trim(); if !s.is_empty() { - Some(CompactString::from_str_to_lowercase(s)) + Some(s.to_lowercase()) } else { None } @@ -246,9 +246,10 @@ impl SpamFilterInit for Server { output: SpamFilterOutput { ehlo_host: Hostname::new(input.ehlo_domain.unwrap_or("unknown")), iprev_ptr: input.iprev_result.and_then(|r| { - r.ptr.as_ref().and_then(|ptr| ptr.first()).map(|ptr| { - CompactString::from_str_to_lowercase(ptr.strip_suffix('.').unwrap_or(ptr)) - }) + r.ptr + .as_ref() + .and_then(|ptr| ptr.first()) + .map(|ptr| (ptr.strip_suffix('.').unwrap_or(ptr)).to_lowercase()) }), env_from_postmaster: env_from_addr.address.is_empty() || POSTMASTER_ADDRESSES.contains(&env_from_addr.local_part.as_str()), @@ -260,9 +261,7 @@ impl SpamFilterInit for Server { .collect(), from: Recipient { email: Email::new(from.and_then(|f| f.address()).unwrap_or_default()), - name: from - .and_then(|f| f.name()) - .map(CompactString::from_str_to_lowercase), + name: from.and_then(|f| f.name()).map(|name| name.to_lowercase()), }, reply_to, subject_thread_lc: subject_thread.trim().to_lowercase(), diff --git a/crates/spam-filter/src/analysis/llm.rs b/crates/spam-filter/src/analysis/llm.rs index fc3f40fd..f7771ba7 100644 --- a/crates/spam-filter/src/analysis/llm.rs +++ b/crates/spam-filter/src/analysis/llm.rs @@ -99,11 +99,8 @@ impl SpamFilterAnalyzeLlm for Server { _ => return, }; - if let (Some(header), Some(explanation)) = - (&self.core.spam.headers.llm, explanation) - { - ctx.result.header = - format!("{header}: {category} ({explanation})\r\n",).into(); + if let Some(explanation) = explanation { + ctx.result.llm_result = Some((category, explanation)); } } Err(err) => { diff --git a/crates/spam-filter/src/analysis/mod.rs b/crates/spam-filter/src/analysis/mod.rs index 730ee365..55fe41f2 100644 --- a/crates/spam-filter/src/analysis/mod.rs +++ b/crates/spam-filter/src/analysis/mod.rs @@ -10,14 +10,14 @@ use std::{ }; use common::{Server, config::spamfilter::Location}; -use compact_str::CompactString; + use mail_parser::{Header, parsers::MessageStream}; use crate::{ Recipient, SpamFilterContext, SpamFilterInput, SpamFilterOutput, SpamFilterResult, TextPart, }; -pub mod bayes; +pub mod classifier; pub mod date; pub mod dmarc; pub mod domain; @@ -33,7 +33,6 @@ pub mod pyzor; pub mod received; pub mod recipient; pub mod replyto; -pub mod reputation; pub mod rules; pub mod score; pub mod subject; @@ -84,7 +83,7 @@ impl SpamFilterContext<'_> { } impl SpamFilterResult { - pub fn add_tag(&mut self, tag: impl Into) { + pub fn add_tag(&mut self, tag: impl Into) { self.tags.insert(tag.into()); } diff --git a/crates/spam-filter/src/analysis/recipient.rs b/crates/spam-filter/src/analysis/recipient.rs index f180048a..04c64c63 100644 --- a/crates/spam-filter/src/analysis/recipient.rs +++ b/crates/spam-filter/src/analysis/recipient.rs @@ -153,7 +153,7 @@ impl SpamFilterAnalyzeRecipient for Server { for rcpt in &unique_recipients { // Validate name if let Some(rcpt_name) = &rcpt.name { - if rcpt_name == rcpt.email.address { + if *rcpt_name == rcpt.email.address { to_dn_eq_addr_count += 1; } else { to_dn_count += 1; diff --git a/crates/spam-filter/src/analysis/reputation.rs b/crates/spam-filter/src/analysis/reputation.rs deleted file mode 100644 index 226bf8c6..00000000 --- a/crates/spam-filter/src/analysis/reputation.rs +++ /dev/null @@ -1,103 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use crate::SpamFilterContext; -use common::{ - Server, - config::spamfilter::{ReputationCount, ReputationType}, -}; -use mail_auth::DmarcResult; -use std::future::Future; - -pub trait SpamFilterAnalyzeReputation: Sync + Send { - fn spam_filter_analyze_reputation( - &self, - ctx: &mut SpamFilterContext<'_>, - ) -> impl Future + Send; -} - -impl SpamFilterAnalyzeReputation for Server { - async fn spam_filter_analyze_reputation(&self, ctx: &mut SpamFilterContext<'_>) { - // Do not penalize forged domains - let reputation = self.inner.data.spam_reputation.load(); - if reputation.total.ham > 0 - && reputation.total.spam > 0 - && reputation.total.ham + reputation.total.spam < 100 - { - if matches!(ctx.input.dmarc_result, Some(DmarcResult::Pass)) { - // Obtain sender address - let sender = if !ctx.output.env_from_addr.address.is_empty() { - &ctx.output.env_from_addr - } else { - &ctx.output.from.email - }; - - if let Some(count) = reputation.items.get(&ReputationType::Domain( - sender.domain_part.sld_or_default().into(), - )) && let Some(tag) = reputation_tag("REP_DOMAIN", count, &reputation.total) - { - ctx.result.add_tag(tag); - } - } - - // Add ASN - if let Some(asn_id) = &ctx.input.asn { - ctx.result.add_tag(format!("SOURCE_ASN_{asn_id}")); - if let Some(count) = reputation.items.get(&ReputationType::Asn(*asn_id)) - && let Some(tag) = reputation_tag("REP_ASN", count, &reputation.total) - { - ctx.result.add_tag(tag); - } - } - - // Add IP - if let Some(count) = reputation - .items - .get(&ReputationType::Ip(ctx.input.remote_ip)) - && let Some(tag) = reputation_tag("REP_IP", count, &reputation.total) - { - ctx.result.add_tag(tag); - } - } else { - // Add ASN - if let Some(asn_id) = &ctx.input.asn { - ctx.result.add_tag(format!("SOURCE_ASN_{asn_id}")); - } - } - - if let Some(country) = &ctx.input.country { - ctx.result.add_tag(format!("SOURCE_COUNTRY_{country}")); - } - } -} - -fn reputation_tag( - prefix: &str, - token_count: &ReputationCount, - total_count: &ReputationCount, -) -> Option { - let total_token_occurrences = token_count.spam + token_count.ham; - if total_token_occurrences < 10 { - return None; - } - let prob_token_given_spam = token_count.spam as f64 / total_count.spam as f64; - let prob_token_given_ham = token_count.ham as f64 / total_count.ham as f64; - if prob_token_given_spam + prob_token_given_ham == 0.0 { - return None; - } - let spam_probability = prob_token_given_spam / (prob_token_given_spam + prob_token_given_ham); - if spam_probability >= 0.90 { - Some(format!("{prefix}_VERY_BAD",)) - } else if spam_probability >= 0.75 { - Some(format!("{prefix}_BAD",)) - } else if spam_probability <= 0.10 { - Some(format!("{prefix}_VERY_GOOD",)) - } else if spam_probability <= 0.25 { - Some(format!("{prefix}_GOOD",)) - } else { - None - } -} diff --git a/crates/spam-filter/src/analysis/rules.rs b/crates/spam-filter/src/analysis/rules.rs index bf8f6e3f..6f65712c 100644 --- a/crates/spam-filter/src/analysis/rules.rs +++ b/crates/spam-filter/src/analysis/rules.rs @@ -10,7 +10,6 @@ use common::{ Server, config::spamfilter::{IpResolver, Location}, }; -use compact_str::CompactString; use crate::{ SpamFilterContext, TextPart, @@ -30,7 +29,7 @@ impl SpamFilterAnalyzeRules for Server { for url in &ctx.output.urls { for rule in &self.core.spam.rules.url { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &url.element, url.location), ctx.input.span_id, @@ -49,7 +48,7 @@ impl SpamFilterAnalyzeRules for Server { for rule in &self.core.spam.rules.domain { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &resolver, domain.location), ctx.input.span_id, @@ -66,7 +65,7 @@ impl SpamFilterAnalyzeRules for Server { for email in &ctx.output.emails { for rule in &self.core.spam.rules.email { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &email.element, email.location), ctx.input.span_id, @@ -86,7 +85,7 @@ impl SpamFilterAnalyzeRules for Server { for email in rcpt { for rule in &self.core.spam.rules.email { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, email, location), ctx.input.span_id, @@ -106,7 +105,7 @@ impl SpamFilterAnalyzeRules for Server { for rule in &self.core.spam.rules.ip { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &ip_resolver, ip.location), ctx.input.span_id, @@ -135,7 +134,7 @@ impl SpamFilterAnalyzeRules for Server { for rule in &self.core.spam.rules.header { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &header_resolver, Location::BodyText), ctx.input.span_id, @@ -167,7 +166,7 @@ impl SpamFilterAnalyzeRules for Server { for rule in &self.core.spam.rules.body { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &string_resolver, location), ctx.input.span_id, @@ -184,7 +183,7 @@ impl SpamFilterAnalyzeRules for Server { let dummy_resolver = StringResolver(""); for rule in &self.core.spam.rules.any { if let Some(tag) = self - .eval_if::( + .eval_if::( rule, &SpamFilterResolver::new(ctx, &dummy_resolver, Location::BodyText), ctx.input.span_id, diff --git a/crates/spam-filter/src/analysis/score.rs b/crates/spam-filter/src/analysis/score.rs index 61f6f974..c0d75c68 100644 --- a/crates/spam-filter/src/analysis/score.rs +++ b/crates/spam-filter/src/analysis/score.rs @@ -7,16 +7,16 @@ use crate::{ SpamFilterContext, analysis::{ - bayes::SpamFilterAnalyzeBayes, date::SpamFilterAnalyzeDate, dmarc::SpamFilterAnalyzeDmarc, - domain::SpamFilterAnalyzeDomain, ehlo::SpamFilterAnalyzeEhlo, from::SpamFilterAnalyzeFrom, + classifier::SpamFilterAnalyzeClassify, date::SpamFilterAnalyzeDate, + dmarc::SpamFilterAnalyzeDmarc, domain::SpamFilterAnalyzeDomain, + ehlo::SpamFilterAnalyzeEhlo, from::SpamFilterAnalyzeFrom, headers::SpamFilterAnalyzeHeaders, html::SpamFilterAnalyzeHtml, ip::SpamFilterAnalyzeIp, messageid::SpamFilterAnalyzeMid, mime::SpamFilterAnalyzeMime, pyzor::SpamFilterAnalyzePyzor, received::SpamFilterAnalyzeReceived, recipient::SpamFilterAnalyzeRecipient, replyto::SpamFilterAnalyzeReplyTo, - reputation::SpamFilterAnalyzeReputation, rules::SpamFilterAnalyzeRules, - subject::SpamFilterAnalyzeSubject, url::SpamFilterAnalyzeUrl, + rules::SpamFilterAnalyzeRules, subject::SpamFilterAnalyzeSubject, + url::SpamFilterAnalyzeUrl, }, - modules::bayes::BayesClassifier, }; use common::{Server, config::spamfilter::SpamFilterAction}; use std::{fmt::Write, future::Future, vec}; @@ -29,24 +29,30 @@ use crate::analysis::llm::SpamFilterAnalyzeLlm; // SPDX-SnippetEnd pub trait SpamFilterAnalyzeScore: Sync + Send { - fn spam_filter_score( - &self, - ctx: &mut SpamFilterContext<'_>, - ) -> impl Future> + Send; - fn spam_filter_finalize( &self, ctx: &mut SpamFilterContext<'_>, - ) -> impl Future> + Send; + ) -> impl Future> + Send; fn spam_filter_classify( &self, ctx: &mut SpamFilterContext<'_>, - ) -> impl Future> + Send; + ) -> impl Future> + Send; +} + +#[derive(Debug, Default)] +pub struct SpamFilterScore { + pub results: Vec, + pub headers: String, + pub spam_trap: bool, } impl SpamFilterAnalyzeScore for Server { - async fn spam_filter_score(&self, ctx: &mut SpamFilterContext<'_>) -> SpamFilterAction<()> { + async fn spam_filter_finalize( + &self, + ctx: &mut SpamFilterContext<'_>, + ) -> SpamFilterAction { + // Calculate final score let mut results = vec![]; let mut header_len = 60; @@ -59,7 +65,7 @@ impl SpamFilterAnalyzeScore for Server { Some(SpamFilterAction::Reject) => { return SpamFilterAction::Reject; } - None => 0.0, + None | Some(SpamFilterAction::Disabled) => 0.0, }; ctx.result.score += score; header_len += tag.len() + 10; @@ -68,85 +74,101 @@ impl SpamFilterAnalyzeScore for Server { } } - // Write results header sorted by score - if let Some(header_name) = &self.core.spam.headers.result { - let mut header = ctx - .result - .header - .get_or_insert_with(|| String::with_capacity(header_name.len() + header_len + 2)); - results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap().then_with(|| a.0.cmp(b.0))); - header.push_str(header_name); - header.push_str(": "); - for (idx, (tag, score)) in results.into_iter().enumerate() { - if idx > 0 { - header.push_str(",\r\n\t"); + let mut final_score = ctx.result.score; + let mut avg_confidence: f32 = 0.0; + let mut user_results = vec![false; ctx.result.classifier_confidence.len()]; + if !ctx.result.classifier_confidence.is_empty() { + for (idx, &confidence) in ctx.result.classifier_confidence.iter().enumerate() { + if confidence != 0.0 { + avg_confidence += confidence; + let user_score = self + .core + .spam + .lists + .scores + .get(confidence.spam_tag()) + .and_then(|v| v.as_score()) + .copied() + .unwrap_or_default(); + + if ctx.result.score + user_score >= self.core.spam.scores.spam_threshold { + user_results[idx] = true; + } } - let _ = write!(&mut header, "{} ({:.2})", tag, score); } - header.push_str("\r\n"); - SpamFilterAction::Allow(()) - } else { - SpamFilterAction::Allow(()) - } - } + avg_confidence /= ctx.result.classifier_confidence.len() as f32; - async fn spam_filter_finalize( - &self, - ctx: &mut SpamFilterContext<'_>, - ) -> SpamFilterAction { - // Train Bayes classifier - if let Some(config) = self - .core - .spam - .bayes - .as_ref() - .filter(|c| c.auto_learn && !ctx.input.is_test) - { - let was_classified = - ctx.result.has_tag("BAYES_SPAM") || ctx.result.has_tag("BAYES_HAM"); - if ctx.result.has_tag("SPAM_TRAP") - || (ctx.result.score >= config.auto_learn_spam_threshold && !was_classified) - { - self.bayes_train_if_balanced(ctx, true).await; - } else if ctx.result.has_tag("TRUSTED_REPLY") - || (ctx.result.score <= config.auto_learn_ham_threshold && !was_classified) - { - self.bayes_train_if_balanced(ctx, false).await; + if avg_confidence != 0.0 { + let tag = avg_confidence.spam_tag(); + let score = self + .core + .spam + .lists + .scores + .get(tag) + .and_then(|v| v.as_score()) + .copied() + .unwrap_or_default(); + results.push((tag, score)); + final_score += score; } } if self.core.spam.scores.reject_threshold > 0.0 - && ctx.result.score >= self.core.spam.scores.reject_threshold + && final_score >= self.core.spam.scores.reject_threshold { SpamFilterAction::Reject } else if self.core.spam.scores.discard_threshold > 0.0 - && ctx.result.score >= self.core.spam.scores.discard_threshold + && final_score >= self.core.spam.scores.discard_threshold { SpamFilterAction::Discard } else { - let mut header = std::mem::take(&mut ctx.result.header).unwrap_or_default(); - if let Some(header_name) = &self.core.spam.headers.status { + let mut headers = String::with_capacity(header_len + 40); + results.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap().then_with(|| a.0.cmp(b.0))); + headers.push_str("X-Spam-Status: "); + for (idx, (tag, score)) in results.into_iter().enumerate() { + if idx > 0 { + headers.push_str(",\r\n\t"); + } + let _ = write!(&mut headers, "{} ({:.2})", tag, score); + } + headers.push_str("\r\n"); + + if let Some((category, explanation)) = &ctx.result.llm_result { + let _ = write!(&mut headers, "X-Spam-LLM: {category} ({explanation})\r\n",); + } + + let class = if final_score >= self.core.spam.scores.spam_threshold { + "spam" + } else { + "ham" + }; + + if avg_confidence != 0.0 { let _ = write!( - &mut header, - "{}: {}, score={:.2}\r\n", - header_name, - if ctx.result.score >= self.core.spam.scores.spam_threshold { - "Yes" - } else { - "No" - }, - ctx.result.score + &mut headers, + "X-Spam-Score: {class}, score={final_score:.2}, avg_confidence={avg_confidence:.2}\r\n", + ); + } else { + let _ = write!( + &mut headers, + "X-Spam-Score: {class}, score={final_score:.2}\r\n", ); } - SpamFilterAction::Allow(header) + + SpamFilterAction::Allow(SpamFilterScore { + results: user_results, + headers, + spam_trap: ctx.result.spam_trap, + }) } } async fn spam_filter_classify( &self, ctx: &mut SpamFilterContext<'_>, - ) -> SpamFilterAction { + ) -> SpamFilterAction { // IP address analysis self.spam_filter_analyze_ip(ctx).await; @@ -202,29 +224,53 @@ impl SpamFilterAnalyzeScore for Server { // SPDX-SnippetEnd - // Reputation tracking and adjust score - self.spam_filter_analyze_reputation(ctx).await; - // Spam trap self.spam_filter_analyze_spam_trap(ctx).await; // Pyzor checks self.spam_filter_analyze_pyzor(ctx).await; - // Bayes classification - self.spam_filter_analyze_bayes_classify(ctx).await; + // Model classification + self.spam_filter_analyze_classify(ctx).await; // User-defined rules self.spam_filter_analyze_rules(ctx).await; - // Calculate score - match self.spam_filter_score(ctx).await { - SpamFilterAction::Allow(_) => (), - SpamFilterAction::Discard => return SpamFilterAction::Discard, - SpamFilterAction::Reject => return SpamFilterAction::Reject, - } - // Final score calculation self.spam_filter_finalize(ctx).await } } + +pub trait ConfidenceStore { + fn spam_tag(&self) -> &'static str; + fn is_certain(&self) -> Option; +} + +impl ConfidenceStore for f32 { + fn spam_tag(&self) -> &'static str { + match *self { + p if p < 0.10 => "PROB_HAM_HIGH", + p if p < 0.25 => "PROB_HAM_MEDIUM", + p if p < 0.40 => "PROB_HAM_LOW", + p if p < 0.50 => "PROB_HAM_UNCERTAIN", + p if p < 0.60 => "PROB_SPAM_UNCERTAIN", + p if p < 0.75 => "PROB_SPAM_LOW", + p if p < 0.90 => "PROB_SPAM_MEDIUM", + p => { + if p.is_finite() { + "PROB_SPAM_HIGH" + } else { + "PROB_SPAM_UNCERTAIN" + } + } + } + } + + fn is_certain(&self) -> Option { + match *self { + p if p < 0.40 => Some(false), // certain ham + p if p > 0.60 => Some(true), // certain spam + _ => None, // uncertain + } + } +} diff --git a/crates/spam-filter/src/lib.rs b/crates/spam-filter/src/lib.rs index 59da4abf..8e75d609 100644 --- a/crates/spam-filter/src/lib.rs +++ b/crates/spam-filter/src/lib.rs @@ -7,18 +7,17 @@ pub mod analysis; pub mod modules; -use std::borrow::Cow; -use std::collections::HashSet; -use std::hash::{Hash, Hasher}; -use std::net::{IpAddr, Ipv4Addr}; - use analysis::ElementLocation; use analysis::url::UrlParts; -use compact_str::CompactString; + use mail_auth::{ArcOutput, DkimOutput, DmarcResult, IprevOutput, SpfOutput, dmarc::Policy}; use mail_parser::Message; use modules::html::HtmlToken; use nlp::tokenizers::types::TokenType; +use std::borrow::Cow; +use std::collections::HashSet; +use std::hash::{Hash, Hasher}; +use std::net::{IpAddr, Ipv4Addr}; use store::ahash::AHashSet; pub struct SpamFilterInput<'x> { @@ -49,13 +48,12 @@ pub struct SpamFilterInput<'x> { pub env_from_flags: u64, pub env_rcpt_to: Vec<&'x str>, - pub account_id: Option, pub is_test: bool, } pub struct SpamFilterOutput<'x> { pub ehlo_host: Hostname, - pub iprev_ptr: Option, + pub iprev_ptr: Option, pub env_from_addr: Email, pub env_from_postmaster: bool, @@ -75,7 +73,7 @@ pub struct SpamFilterOutput<'x> { pub ips: AHashSet>, pub urls: HashSet>>, pub emails: HashSet>, - pub domains: HashSet>, + pub domains: HashSet>, pub text_parts: Vec>, } @@ -101,13 +99,15 @@ pub enum TextPart<'x> { #[derive(Debug, Default)] pub struct SpamFilterResult { - pub tags: AHashSet, - pub score: f64, + pub tags: AHashSet, + pub classifier_confidence: Vec, + pub score: f32, pub rbl_ip_checks: usize, pub rbl_domain_checks: usize, pub rbl_url_checks: usize, pub rbl_email_checks: usize, - pub header: Option, + pub llm_result: Option<(String, String)>, + pub spam_trap: bool, } pub struct SpamFilterContext<'x> { @@ -118,22 +118,22 @@ pub struct SpamFilterContext<'x> { #[derive(Debug, Clone)] pub struct Hostname { - pub fqdn: CompactString, + pub fqdn: String, pub ip: Option, - pub sld: Option, + pub sld: Option, } #[derive(Debug, Clone)] pub struct Email { - pub address: CompactString, - pub local_part: CompactString, + pub address: String, + pub local_part: String, pub domain_part: Hostname, } #[derive(Debug, Clone)] pub struct Recipient { pub email: Email, - pub name: Option, + pub name: Option, } impl<'x> SpamFilterInput<'x> { @@ -157,32 +157,6 @@ impl<'x> SpamFilterInput<'x> { env_from: "", env_from_flags: 0, env_rcpt_to: vec![], - account_id: None, - is_test: false, - } - } - - pub fn from_account_message(message: &'x Message<'x>, account_id: u32, span_id: u64) -> Self { - Self { - message, - span_id, - arc_result: None, - spf_ehlo_result: None, - spf_mail_from_result: None, - dkim_result: &[], - dmarc_result: None, - dmarc_policy: None, - iprev_result: None, - remote_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), - ehlo_domain: None, - authenticated_as: None, - asn: None, - country: None, - is_tls: true, - env_from: "", - env_from_flags: 0, - env_rcpt_to: vec![], - account_id: Some(account_id), is_test: false, } } diff --git a/crates/spam-filter/src/modules/classifier.rs b/crates/spam-filter/src/modules/classifier.rs new file mode 100644 index 00000000..c6516ff6 --- /dev/null +++ b/crates/spam-filter/src/modules/classifier.rs @@ -0,0 +1,31 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use common::Server; + +use crate::SpamFilterContext; + +pub trait SpamClassifier { + fn spam_classify( + &self, + ctx: &SpamFilterContext<'_>, + ) -> impl Future>> + Send; + + fn spam_train( + &self, + ctx: &SpamFilterContext<'_>, + ) -> impl Future> + Send; +} + +impl SpamClassifier for Server { + async fn spam_train(&self, ctx: &SpamFilterContext<'_>) -> trc::Result<()> { + todo!() + } + + async fn spam_classify(&self, ctx: &SpamFilterContext<'_>) -> trc::Result> { + todo!() + } +} diff --git a/crates/spam-filter/src/modules/dnsbl.rs b/crates/spam-filter/src/modules/dnsbl.rs index 03888887..24ed05b9 100644 --- a/crates/spam-filter/src/modules/dnsbl.rs +++ b/crates/spam-filter/src/modules/dnsbl.rs @@ -15,7 +15,7 @@ use common::{ config::spamfilter::{DnsBlServer, Element, IpResolver, Location}, expr::functions::ResolveVariable, }; -use compact_str::CompactString; + use mail_auth::{Error, common::resolver::IntoFqdn}; use trc::SpamEvent; @@ -81,10 +81,10 @@ async fn is_dnsbl( resolver: SpamFilterResolver<'_, impl ResolveVariable>, element: Element, checks: &mut usize, -) -> Option { +) -> Option { let time = Instant::now(); let zone = server - .eval_if::(&config.zone, &resolver, resolver.ctx.input.span_id) + .eval_if::(&config.zone, &resolver, resolver.ctx.input.span_id) .await?; #[cfg(feature = "test_mode")] diff --git a/crates/spam-filter/src/modules/html.rs b/crates/spam-filter/src/modules/html.rs index 81b8c6a0..3b3d758e 100644 --- a/crates/spam-filter/src/modules/html.rs +++ b/crates/spam-filter/src/modules/html.rs @@ -4,24 +4,24 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use compact_str::CompactString; + use mail_parser::decoders::html::add_html_token; #[derive(Debug, Eq, PartialEq, Clone)] pub enum HtmlToken { StartTag { name: u64, - attributes: Vec<(u64, Option)>, + attributes: Vec<(u64, Option)>, is_self_closing: bool, }, EndTag { name: u64, }, Comment { - text: CompactString, + text: String, }, Text { - text: CompactString, + text: String, }, } @@ -131,7 +131,7 @@ pub fn html_to_tokens(input: &str) -> Vec { last_ch = ch; } tags.push(HtmlToken::Comment { - text: CompactString::from_utf8(comment).unwrap_or_default(), + text: String::from_utf8(comment).unwrap_or_default(), }); } else { let mut is_end_tag = false; @@ -157,7 +157,7 @@ pub fn html_to_tokens(input: &str) -> Vec { let mut shift = 0; let mut tag = 0; - let mut attributes: Vec<(u64, Option)> = vec![]; + let mut attributes: Vec<(u64, Option)> = vec![]; 'outer: while let Some((_, &ch)) = iter.next() { match ch { @@ -203,7 +203,7 @@ pub fn html_to_tokens(input: &str) -> Vec { match ch { b'>' if !in_quote => { if !value.is_empty() { - let value = CompactString::from_utf8(value) + let value = String::from_utf8(value) .unwrap_or_default(); if let Some((_, v)) = attributes.last_mut() { *v = value.into(); @@ -232,7 +232,7 @@ pub fn html_to_tokens(input: &str) -> Vec { } if !value.is_empty() { - let value = CompactString::from_utf8(value).unwrap_or_default(); + let value = String::from_utf8(value).unwrap_or_default(); if let Some((_, v)) = attributes.last_mut() { *v = value.into(); } else { diff --git a/crates/spam-filter/src/modules/mod.rs b/crates/spam-filter/src/modules/mod.rs index 40ae65ab..8b0e17bd 100644 --- a/crates/spam-filter/src/modules/mod.rs +++ b/crates/spam-filter/src/modules/mod.rs @@ -4,31 +4,9 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::Server; -use store::{ - Deserialize, Value, - dispatch::lookup::{KeyValue, LookupKey}, -}; - -pub mod bayes; +pub mod classifier; pub mod dnsbl; pub mod expression; pub mod html; pub mod pyzor; pub mod sanitize; - -pub(crate) async fn key_get> + std::fmt::Debug + 'static>( - server: &Server, - span_id: u64, - key: impl Into>, -) -> Result, ()> { - server.in_memory_store().key_get(key).await.map_err(|err| { - trc::error!(err.span_id(span_id).caused_by(trc::location!())); - }) -} - -pub(crate) async fn key_set(server: &Server, span_id: u64, kv: KeyValue>) { - if let Err(err) = server.in_memory_store().key_set(kv).await { - trc::error!(err.span_id(span_id).caused_by(trc::location!())); - } -} diff --git a/crates/spam-filter/src/modules/sanitize.rs b/crates/spam-filter/src/modules/sanitize.rs index 27baa1cf..b778e13b 100644 --- a/crates/spam-filter/src/modules/sanitize.rs +++ b/crates/spam-filter/src/modules/sanitize.rs @@ -6,17 +6,15 @@ use std::net::IpAddr; -use compact_str::CompactString; - use crate::{Email, Hostname}; impl Hostname { pub fn new(host: &str) -> Self { - let mut fqdn = CompactString::from_str_to_lowercase(host.trim_end_matches('.')); + let mut fqdn = host.trim_end_matches('.').to_lowercase(); // Decode punycode if fqdn.contains("xn--") { - let mut decoded = CompactString::with_capacity(fqdn.len()); + let mut decoded = String::with_capacity(fqdn.len()); for part in fqdn.split('.') { if !decoded.is_empty() { @@ -63,7 +61,7 @@ impl Hostname { impl Email { pub fn new(address: &str) -> Self { - let address = CompactString::from_str_to_lowercase(address); + let address = address.to_lowercase(); let (local_part, domain) = address.rsplit_once('@').unwrap_or_default(); Email { diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 496dd371..3cc48bc8 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -289,10 +289,15 @@ impl ValueClass { .write(if *is_insert { 7u8 } else { 8u8 }) .write(document_id) .write(index.to_u8()), - TaskQueueClass::BayesTrain { due, learn_spam } => serializer + TaskQueueClass::SpamTrain { + due, + blob_hash, + learn_spam, + } => serializer .write(due.inner()) .write(account_id) .write(if *learn_spam { 1u8 } else { 2u8 }) + .write(blob_hash.as_slice()) .write(document_id), TaskQueueClass::SendAlarm { due, @@ -580,7 +585,7 @@ impl ValueClass { }, ValueClass::TaskQueue(e) => match e { TaskQueueClass::UpdateIndex { .. } => (U64_LEN * 2) + 2, - TaskQueueClass::BayesTrain { .. } => (U64_LEN * 2) + 1, + TaskQueueClass::SpamTrain { .. } => BLOB_HASH_LEN + (U64_LEN * 2) + 1, TaskQueueClass::SendAlarm { .. } | TaskQueueClass::MergeThreads { .. } => { U64_LEN + (U32_LEN * 3) + 1 } diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index ad2ff534..fc12c7bd 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -226,8 +226,9 @@ pub enum TaskQueueClass { index: SearchIndex, is_insert: bool, }, - BayesTrain { + SpamTrain { due: TaskEpoch, + blob_hash: BlobHash, learn_spam: bool, }, SendAlarm { diff --git a/crates/trc/src/event/conv.rs b/crates/trc/src/event/conv.rs index 890952e9..2bd8ce26 100644 --- a/crates/trc/src/event/conv.rs +++ b/crates/trc/src/event/conv.rs @@ -35,6 +35,12 @@ impl From for Value { } } +impl From> for Value { + fn from(value: Box) -> Self { + Self::String(CompactString::from(value)) + } +} + impl From for Value { fn from(value: u64) -> Self { Self::UInt(value)