diff --git a/Cargo.lock b/Cargo.lock index 2bffcef4..8569b69c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3496,6 +3496,7 @@ dependencies = [ "sieve-rs", "smtp", "smtp-proto", + "spam-filter", "store", "tokio", "tokio-tungstenite 0.24.0", diff --git a/Cargo.toml b/Cargo.toml index fba76680..bd71ca41 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,14 +2,14 @@ resolver = "2" members = [ "crates/main", -# "crates/jmap", -# "crates/jmap-proto", -# "crates/imap", -# "crates/imap-proto", -# "crates/smtp", -# "crates/managesieve", -# "crates/pop3", -# "crates/spam-filter", + "crates/jmap", + "crates/jmap-proto", + "crates/imap", + "crates/imap-proto", + "crates/smtp", + "crates/managesieve", + "crates/pop3", + "crates/spam-filter", "crates/nlp", "crates/store", "crates/directory", diff --git a/crates/common/src/config/smtp/session.rs b/crates/common/src/config/smtp/session.rs index 525bc6bd..464b5ae3 100644 --- a/crates/common/src/config/smtp/session.rs +++ b/crates/common/src/config/smtp/session.rs @@ -130,6 +130,7 @@ pub enum AddressMapping { #[derive(Clone)] pub struct Data { pub script: IfBlock, + pub spam_filter: IfBlock, // Limits pub max_messages: IfBlock, @@ -143,6 +144,7 @@ pub struct Data { pub add_auth_results: IfBlock, pub add_message_id: IfBlock, pub add_date: IfBlock, + pub add_delivered_to: bool, } #[derive(Clone)] @@ -411,6 +413,11 @@ impl SessionConfig { "session.data.limits.received-headers", &has_rcpt_vars, ), + ( + &mut session.data.spam_filter, + "session.data.spam-filter", + &has_rcpt_vars, + ), ( &mut session.data.add_received, "session.data.add-headers.received", @@ -446,7 +453,9 @@ impl SessionConfig { *value = if_block; } } - + session.data.add_delivered_to = config + .property_or_default("session.data.add-headers.delivered-to", "true") + .unwrap_or(true); session } } @@ -773,6 +782,7 @@ impl Default for SessionConfig { }, data: Data { script: IfBlock::empty("session.data.script"), + spam_filter: IfBlock::new::<()>("session.data.spam-filter", [], "true"), max_messages: IfBlock::new::<()>("session.data.limits.messages", [], "10"), max_message_size: IfBlock::new::<()>("session.data.limits.size", [], "104857600"), max_received_headers: IfBlock::new::<()>( @@ -810,6 +820,7 @@ impl Default for SessionConfig { [("local_port == 25", "true")], "false", ), + add_delivered_to: false, }, extensions: Extensions { pipelining: IfBlock::new::<()>("session.extensions.pipelining", [], "true"), diff --git a/crates/common/src/config/spamfilter.rs b/crates/common/src/config/spamfilter.rs index 05164e06..1b2be048 100644 --- a/crates/common/src/config/spamfilter.rs +++ b/crates/common/src/config/spamfilter.rs @@ -34,6 +34,7 @@ pub struct SpamFilterConfig { pub struct SpamFilterHeaderConfig { pub status: Option, pub result: Option, + pub bayes_result: Option, pub llm: Option, } @@ -81,7 +82,9 @@ pub struct BayesConfig { pub auto_learn_ham_threshold: f64, pub score_spam: f64, pub score_ham: f64, - pub enabled_account: bool, + pub account_score_spam: f64, + pub account_score_ham: f64, + pub account_classify: bool, } #[derive(Debug, Clone, Default)] @@ -319,6 +322,7 @@ impl SpamFilterHeaderConfig { ("status", &mut header.status), ("result", &mut header.result), ("llm", &mut header.llm), + ("bayes", &mut header.bayes_result), ] { if config .property_or_default(("spam-filter.header", typ, "enable"), "true") @@ -549,9 +553,15 @@ impl BayesConfig { score_ham: config .property_or_default("spam-filter.bayes.score.ham", "0.5") .unwrap_or(0.5), - enabled_account: config - .property_or_default("spam-filter.bayes.enable-account", "false") + account_classify: config + .property_or_default("spam-filter.bayes.account.enable", "false") .unwrap_or(false), + account_score_spam: config + .property_or_default("spam-filter.bayes.account.score.spam", "0.7") + .unwrap_or(0.7), + account_score_ham: config + .property_or_default("spam-filter.bayes.account.score.ham", "0.5") + .unwrap_or(0.5), } .into() } @@ -633,6 +643,7 @@ impl Default for SpamFilterHeaderConfig { SpamFilterHeaderConfig { status: "X-Spam-Status".to_string().into(), result: "X-Spam-Result".to_string().into(), + bayes_result: "X-Spam-Bayes".to_string().into(), llm: "X-Spam-LLM".to_string().into(), } } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 10c04dfc..30018611 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -81,7 +81,7 @@ pub const KV_TRUSTED_REPLY: u8 = 19; pub const KV_LOCK_PURGE_ACCOUNT: u8 = 20; pub const KV_LOCK_QUEUE_MESSAGE: u8 = 21; pub const KV_LOCK_QUEUE_REPORT: u8 = 22; -pub const KV_LOCK_FTS: u8 = 23; +pub const KV_LOCK_EMAIL_TASK: u8 = 23; pub const KV_LOCK_HOUSEKEEPER: u8 = 24; #[derive(Clone)] diff --git a/crates/common/src/manager/restore.rs b/crates/common/src/manager/restore.rs index 236a7a2c..3677e5b5 100644 --- a/crates/common/src/manager/restore.rs +++ b/crates/common/src/manager/restore.rs @@ -15,7 +15,7 @@ use store::{ roaring::RoaringBitmap, write::{ key::DeserializeBigEndian, BatchBuilder, BitmapClass, BitmapHash, BlobOp, DirectoryClass, - FtsQueueClass, LookupClass, MaybeDynamicId, MaybeDynamicValue, Operation, TagValue, + LookupClass, MaybeDynamicId, MaybeDynamicValue, Operation, TagValue, TaskQueueClass, ValueClass, }, BlobStore, Serialize, Store, U32_LEN, @@ -145,7 +145,7 @@ async fn restore_file(store: Store, blob_store: BlobStore, path: &Path) { if account_id != u32::MAX && document_id != u32::MAX { if reader.version == 1 && collection == email_collection { batch.set( - ValueClass::FtsQueue(FtsQueueClass { + ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq, hash: hash.clone(), }), diff --git a/crates/directory/src/core/mod.rs b/crates/directory/src/core/mod.rs index 2683eb5e..2bd8a2b8 100644 --- a/crates/directory/src/core/mod.rs +++ b/crates/directory/src/core/mod.rs @@ -79,10 +79,11 @@ impl Permission { Permission::Undelete => "Restore deleted items", Permission::DkimSignatureCreate => "Create DKIM signatures for email authentication", Permission::DkimSignatureGet => "Retrieve DKIM signature information", - Permission::UpdateSpamFilter => "Modify spam filter settings", - Permission::UpdateWebadmin => "Modify web admin interface settings", + Permission::SpamFilterUpdate => "Modify spam filter settings", + Permission::WebadminUpdate => "Modify web admin interface settings", Permission::LogsView => "Access system logs", - Permission::SieveRun => "Execute Sieve scripts from the REST API", + Permission::SpamFilterTrain => "Train the spam filter", + Permission::SpamFilterClassify => "Classify emails with the spam filter", 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 ead910b4..79f6ecd5 100644 --- a/crates/directory/src/core/principal.rs +++ b/crates/directory/src/core/principal.rs @@ -1088,6 +1088,8 @@ impl Permission { | Permission::SieveRenameScript | Permission::SieveCheckScript | Permission::SieveHaveSpace + | Permission::SpamFilterClassify + | Permission::SpamFilterTrain ) } diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index a6c1280e..230257c0 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -128,10 +128,10 @@ pub enum Permission { Undelete, DkimSignatureCreate, DkimSignatureGet, - UpdateSpamFilter, - UpdateWebadmin, + SpamFilterUpdate, + WebadminUpdate, LogsView, - SieveRun, + SpamFilterTrain, Restart, TracingList, TracingGet, @@ -264,7 +264,9 @@ pub enum Permission { OauthClientOverride, AiModelInteract, - Troubleshoot, // WARNING: add new ids at the end (TODO: use static ids) + Troubleshoot, + SpamFilterClassify, + // WARNING: add new ids at the end (TODO: use static ids) } pub type Permissions = Bitset<{ Permission::COUNT.div_ceil(std::mem::size_of::()) }>; diff --git a/crates/imap/src/op/append.rs b/crates/imap/src/op/append.rs index 284b5c7d..ee335abc 100644 --- a/crates/imap/src/op/append.rs +++ b/crates/imap/src/op/append.rs @@ -19,7 +19,10 @@ use crate::{ }; use common::{listener::SessionStream, MailboxId}; use jmap::{ - email::ingest::{EmailIngest, IngestEmail, IngestSource}, + email::{ + bayes::EmailBayesTrain, + ingest::{EmailIngest, IngestEmail, IngestSource}, + }, services::state::StateManager, }; use jmap_proto::types::{acl::Acl, keyword::Keyword, state::StateChange, type_state::DataType}; @@ -91,12 +94,13 @@ impl SessionData { } // Obtain quota - let resource_token = self + let access_token = self .server .get_cached_access_token(mailbox.account_id) .await - .imap_ctx(&arguments.tag, trc::location!())? - .as_resource_token(); + .imap_ctx(&arguments.tag, trc::location!())?; + let resource_token = access_token.as_resource_token(); + let spam_train = self.server.email_bayes_can_train(&access_token); // Append messages let mut response = StatusResponse::completed(Command::Append); @@ -113,7 +117,8 @@ impl SessionData { keywords: message.flags.into_iter().map(Keyword::from).collect(), received_at: message.received_at.map(|d| d as u64), source: IngestSource::Imap, - encrypt: self.server.core.jmap.encrypt && self.server.core.jmap.encrypt_append, + spam_classify: false, + spam_train, 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 697748fa..396f4af2 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -19,9 +19,9 @@ use crate::{ use common::{listener::SessionStream, MailboxId}; use jmap::{ changes::write::ChangeLog, - email::{copy::EmailCopy, ingest::EmailIngest, set::TagManager}, - mailbox::UidMailbox, - services::state::StateManager, + email::{bayes::EmailBayesTrain, copy::EmailCopy, ingest::EmailIngest, set::TagManager}, + mailbox::{UidMailbox, JUNK_ID}, + services::{index::Indexer, state::StateManager}, JmapMethods, }; use jmap_proto::{ @@ -33,7 +33,7 @@ use jmap_proto::{ }; use store::{ roaring::RoaringBitmap, - write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_VALUE}, + write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, ValueClass, F_VALUE}, }; use super::ImapContext; @@ -171,10 +171,19 @@ impl SessionData { let mut changelog = ChangeLogBuilder::new(); let mut did_move = false; let mut copied_ids = Vec::with_capacity(ids.len()); + let access_token = self + .server + .get_cached_access_token(dest_mailbox.account_id) + .await + .imap_ctx(&arguments.tag, trc::location!())?; + if src_mailbox.id.account_id == dest_mailbox.account_id { // 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; + for (id, imap_id) in ids { // Obtain mailbox tags let (mut mailboxes, thread_id) = if let Some(result) = self @@ -216,7 +225,7 @@ impl SessionData { } } - // Write changes + // Perepare write batch let mut batch = BatchBuilder::new(); batch .with_account_id(account_id) @@ -231,10 +240,41 @@ impl SessionData { .imap_ctx(&arguments.tag, trc::location!())?; } batch.value(Property::Cid, changelog.change_id, F_VALUE); + + // Add bayes train task + if can_spam_train { + if dest_mailbox_id.mailbox_id == JUNK_ID { + batch.set( + ValueClass::TaskQueue( + self.server + .email_bayes_queue_task_build(account_id, id, true) + .await + .imap_ctx(&arguments.tag, trc::location!())?, + ), + vec![], + ); + has_spam_train_tasks = true; + } else if src_mailbox.id.mailbox_id == JUNK_ID { + batch.set( + ValueClass::TaskQueue( + self.server + .email_bayes_queue_task_build(account_id, id, false) + .await + .imap_ctx(&arguments.tag, trc::location!())?, + ), + vec![], + ); + has_spam_train_tasks = true; + } + } + + // Write changes self.server .write_batch(batch) .await .imap_ctx(&arguments.tag, trc::location!())?; + + // Update changelog changelog.log_update(Collection::Email, Id::from_parts(thread_id, id)); changelog.log_child_update(Collection::Mailbox, dest_mailbox_id.mailbox_id); if is_move { @@ -242,17 +282,17 @@ impl SessionData { did_move = true; } } + + // Trigger Bayes training + if has_spam_train_tasks { + self.server.notify_task_queue(); + } } else { // Obtain quota for target account let src_account_id = src_mailbox.id.account_id; let mut dest_change_id = None; let dest_account_id = dest_mailbox.account_id; - let resource_token = self - .server - .get_cached_access_token(dest_account_id) - .await - .imap_ctx(&arguments.tag, trc::location!())? - .as_resource_token(); + let resource_token = access_token.as_resource_token(); let mut destroy_ids = RoaringBitmap::new(); for (id, imap_id) in ids { match self diff --git a/crates/imap/src/op/store.rs b/crates/imap/src/op/store.rs index 14a5626a..bf918279 100644 --- a/crates/imap/src/op/store.rs +++ b/crates/imap/src/op/store.rs @@ -24,9 +24,9 @@ use imap_proto::{ }; use jmap::{ changes::{get::ChangesLookup, write::ChangeLog}, - email::set::TagManager, + email::{bayes::EmailBayesTrain, set::TagManager}, mailbox::UidMailbox, - services::state::StateManager, + services::{index::Indexer, state::StateManager}, JmapMethods, }; use jmap_proto::types::{ @@ -35,7 +35,7 @@ use jmap_proto::types::{ }; use store::{ query::log::{Change, Query}, - write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, F_VALUE}, + write::{assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, ValueClass, F_VALUE}, }; use super::{FromModSeq, ImapContext}; @@ -193,6 +193,14 @@ impl SessionData { .collect::>(); let mut changelog = ChangeLogBuilder::new(); let mut changed_mailboxes = AHashSet::new(); + let access_token = self + .server + .get_cached_access_token(account_id) + .await + .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; + let can_spam_train = self.server.email_bayes_can_train(&access_token); + let mut has_spam_train_tasks = false; + 'outer: for (id, imap_id) in &ids { let mut try_count = 0; loop { @@ -235,6 +243,28 @@ impl SessionData { } if keywords.has_changes() { + // Train spam filter + let mut train_spam = None; + if can_spam_train { + for keyword in keywords.added() { + 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 keywords.removed() { + if keyword == &Keyword::Junk { + train_spam = Some(false); + break; + } + } + } + }; + // Convert keywords to flags let seen_changed = keywords .changed_tags() @@ -265,6 +295,21 @@ impl SessionData { .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())? } batch.value(Property::Cid, changelog.change_id, F_VALUE); + + // Add spam train task + if let Some(learn_spam) = train_spam { + batch.set( + ValueClass::TaskQueue( + self.server + .email_bayes_queue_task_build(account_id, *id, learn_spam) + .await + .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?, + ), + vec![], + ); + has_spam_train_tasks = true; + } + match self.server.write_batch(batch).await { Ok(_) => { // Set all current mailboxes as changed if the Seen tag changed @@ -285,6 +330,8 @@ impl SessionData { } } } + + // Update changelog changelog.log_update(Collection::Email, Id::from_parts(thread_id, *id)); // Add item to response @@ -338,6 +385,11 @@ impl SessionData { changelog.log_child_update(Collection::Mailbox, *mailbox_id); } + // Trigger Bayes training + if has_spam_train_tasks { + self.server.notify_task_queue(); + } + // Write changes if !changelog.is_empty() { let change_id = self diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index d4dbc48d..76c3006e 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -13,6 +13,7 @@ utils = { path = "../utils" } common = { path = "../common" } directory = { path = "../directory" } trc = { path = "../trc" } +spam-filter = { path = "../spam-filter" } smtp-proto = { version = "0.1" } mail-parser = { version = "0.9", features = ["full_encoding", "serde_support", "ludicrous_mode"] } mail-builder = { version = "0.3", features = ["ludicrous_mode"] } diff --git a/crates/jmap/src/api/management/enterprise/undelete.rs b/crates/jmap/src/api/management/enterprise/undelete.rs index e3a40cef..c873d541 100644 --- a/crates/jmap/src/api/management/enterprise/undelete.rs +++ b/crates/jmap/src/api/management/enterprise/undelete.rs @@ -193,8 +193,9 @@ impl UndeleteApi for Server { mailbox_ids: vec![INBOX_ID], keywords: vec![], received_at: (request.time as u64).into(), - source: IngestSource::Smtp, - encrypt: false, + source: IngestSource::Restore, + spam_classify: false, + spam_train: false, session_id: session.session_id, }) .await diff --git a/crates/jmap/src/api/management/reload.rs b/crates/jmap/src/api/management/reload.rs index 4c4fe110..727f3165 100644 --- a/crates/jmap/src/api/management/reload.rs +++ b/crates/jmap/src/api/management/reload.rs @@ -121,7 +121,7 @@ impl ManageReload for Server { match (path.get(1).copied(), req.method()) { (Some("spam-filter"), &Method::GET) => { // Validate the access token - access_token.assert_has_permission(Permission::UpdateSpamFilter)?; + access_token.assert_has_permission(Permission::SpamFilterUpdate)?; Ok(JsonResponse::new(json!({ "data": self @@ -135,7 +135,7 @@ impl ManageReload for Server { } (Some("webadmin"), &Method::GET) => { // Validate the access token - access_token.assert_has_permission(Permission::UpdateWebadmin)?; + access_token.assert_has_permission(Permission::WebadminUpdate)?; self.inner .data diff --git a/crates/jmap/src/api/management/sieve.rs b/crates/jmap/src/api/management/sieve.rs index bccffc19..f73b4987 100644 --- a/crates/jmap/src/api/management/sieve.rs +++ b/crates/jmap/src/api/management/sieve.rs @@ -53,7 +53,7 @@ impl SieveHandler for Server { access_token: &AccessToken, ) -> trc::Result { // Validate the access token - access_token.assert_has_permission(Permission::SieveRun)?; + access_token.assert_has_permission(Permission::SpamFilterTrain)?; let (script, script_id) = match ( path.get(1).and_then(|name| { diff --git a/crates/jmap/src/email/bayes.rs b/crates/jmap/src/email/bayes.rs new file mode 100644 index 00000000..f0027e42 --- /dev/null +++ b/crates/jmap/src/email/bayes.rs @@ -0,0 +1,92 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use std::future::Future; + +use common::{auth::AccessToken, Server}; +use directory::Permission; +use jmap_proto::types::{collection::Collection, property::Property}; +use mail_parser::Message; +use spam_filter::{ + analysis::init::SpamFilterInit, modules::bayes::BayesClassifier, SpamFilterInput, +}; +use store::write::{Bincode, TaskQueueClass}; +use trc::StoreEvent; + +use crate::{changes::write::ChangeLog, JmapMethods}; + +use super::metadata::MessageMetadata; + +pub trait EmailBayesTrain: Sync + Send { + fn email_bayes_train( + &self, + account_id: u32, + span_id: u64, + message: Message<'_>, + learn_spam: bool, + ) -> impl Future + Send; + + fn email_bayes_queue_task_build( + &self, + account_id: u32, + document_id: u32, + learn_spam: bool, + ) -> impl Future> + Send; + + fn email_bayes_can_train(&self, access_token: &AccessToken) -> bool; +} + +impl EmailBayesTrain for Server { + async fn email_bayes_train( + &self, + account_id: u32, + span_id: u64, + message: Message<'_>, + learn_spam: bool, + ) { + self.bayes_train_if_balanced( + &self.spam_filter_init(SpamFilterInput::from_account_message( + &message, account_id, span_id, + )), + learn_spam, + ) + .await + } + + async fn email_bayes_queue_task_build( + &self, + account_id: u32, + document_id: u32, + learn_spam: bool, + ) -> trc::Result { + let metadata = self + .get_property::>( + account_id, + Collection::Email, + document_id, + Property::BodyStructure, + ) + .await? + .ok_or_else(|| { + StoreEvent::NotFound + .into_err() + .account_id(account_id) + .document_id(document_id) + })?; + + Ok(TaskQueueClass::BayesTrain { + seq: self.generate_snowflake_id()?, + hash: metadata.inner.blob_hash, + learn_spam, + }) + } + + fn email_bayes_can_train(&self, access_token: &AccessToken) -> bool { + self.core.spam.bayes.as_ref().map_or(false, |bayes| { + bayes.account_classify && access_token.has_permission(Permission::SpamFilterTrain) + }) + } +} diff --git a/crates/jmap/src/email/copy.rs b/crates/jmap/src/email/copy.rs index bc7aa75b..c71435fe 100644 --- a/crates/jmap/src/email/copy.rs +++ b/crates/jmap/src/email/copy.rs @@ -37,10 +37,10 @@ use mail_parser::{parsers::fields::thread::thread_name, HeaderName, HeaderValue} use store::{ write::{ log::{Changes, LogInsert}, - BatchBuilder, Bincode, FtsQueueClass, MaybeDynamicId, TagValue, ValueClass, F_BITMAP, + BatchBuilder, Bincode, MaybeDynamicId, TagValue, TaskQueueClass, ValueClass, F_BITMAP, F_VALUE, }, - BlobClass, Serialize, + BlobClass, }; use trc::AddContext; use utils::map::vec_map::VecMap; @@ -441,11 +441,11 @@ impl EmailCopy for Server { .value(Property::Keywords, keywords, F_VALUE | F_BITMAP) .value(Property::Cid, change_id, F_VALUE) .set( - ValueClass::FtsQueue(FtsQueueClass { + ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: self.generate_snowflake_id()?, hash: metadata.blob_hash.clone(), }), - 0u64.serialize(), + vec![], ); EmailIndexBuilder::set(metadata).build( &mut batch, @@ -468,7 +468,7 @@ impl EmailCopy for Server { let document_id = ids.last_document_id().caused_by(trc::location!())?; // Request FTS index - self.request_fts_index(); + self.notify_task_queue(); // Update response email.id = Id::from_parts(thread_id, document_id); diff --git a/crates/jmap/src/email/import.rs b/crates/jmap/src/email/import.rs index 1d9b441d..8ddc972f 100644 --- a/crates/jmap/src/email/import.rs +++ b/crates/jmap/src/email/import.rs @@ -25,7 +25,10 @@ use crate::{ changes::state::StateManager, mailbox::set::MailboxSet, JmapMethods, }; -use super::ingest::{EmailIngest, IngestEmail, IngestSource}; +use super::{ + bayes::EmailBayesTrain, + ingest::{EmailIngest, IngestEmail, IngestSource}, +}; use std::future::Future; pub trait EmailImport: Sync + Send { @@ -70,6 +73,7 @@ impl EmailImport for Server { not_created: VecMap::new(), state_change: None, }; + let can_train_spam = self.email_bayes_can_train(access_token); 'outer: for (id, email) in request.emails { // Validate mailboxIds @@ -135,7 +139,8 @@ impl EmailImport for Server { keywords: email.keywords, received_at: email.received_at.map(|r| r.into()), source: IngestSource::Jmap, - encrypt: self.core.jmap.encrypt && self.core.jmap.encrypt_append, + spam_classify: false, + spam_train: can_train_spam, session_id: session.session_id, }) .await diff --git a/crates/jmap/src/email/ingest.rs b/crates/jmap/src/email/ingest.rs index 9303b7cb..447418c2 100644 --- a/crates/jmap/src/email/ingest.rs +++ b/crates/jmap/src/email/ingest.rs @@ -6,6 +6,7 @@ use std::{ borrow::Cow, + fmt::Write, time::{Duration, Instant}, }; @@ -22,14 +23,17 @@ use mail_parser::{ }; use rand::Rng; +use spam_filter::{ + analysis::init::SpamFilterInit, modules::bayes::BayesClassifier, SpamFilterInput, +}; use std::future::Future; use store::{ ahash::AHashSet, query::Filter, write::{ log::{ChangeLogBuilder, Changes, LogInsert}, - now, AssignedIds, BatchBuilder, BitmapClass, FtsQueueClass, MaybeDynamicId, - MaybeDynamicValue, SerializeWithId, TagValue, ValueClass, F_BITMAP, F_CLEAR, F_VALUE, + now, AssignedIds, BatchBuilder, BitmapClass, MaybeDynamicId, MaybeDynamicValue, + SerializeWithId, TagValue, TaskQueueClass, ValueClass, F_BITMAP, F_CLEAR, F_VALUE, }, BitmapKey, BlobClass, Serialize, }; @@ -67,16 +71,18 @@ pub struct IngestEmail<'x> { pub mailbox_ids: Vec, pub keywords: Vec, pub received_at: Option, - pub source: IngestSource, - pub encrypt: bool, + pub source: IngestSource<'x>, + pub spam_classify: bool, + pub spam_train: bool, pub session_id: u64, } #[derive(Clone, Copy, PartialEq, Eq, Debug)] -pub enum IngestSource { - Smtp, +pub enum IngestSource<'x> { + Smtp { deliver_to: &'x str }, Jmap, Imap, + Restore, } const MAX_RETRIES: u32 = 10; @@ -119,23 +125,94 @@ impl EmailIngest for Server { .ctx(trc::Key::Reason, "Failed to parse e-mail message.") })?; - // Check for Spam headers let mut is_spam = false; - if let (IngestSource::Smtp, Some(header_name)) = - (params.source, &self.core.spam.headers.status) - { - if params.mailbox_ids == [INBOX_ID] - && message.root_part().headers().iter().any(|header| { - header.name() == header_name - && header - .value() - .as_text() - .map_or(false, |value| value.contains("Yes")) - }) - { - params.mailbox_ids[0] = JUNK_ID; - is_spam = true; + let mut train_spam = None; + let mut extra_headers = String::new(); + match params.source { + IngestSource::Smtp { deliver_to } => { + // Add delivered to header + if self.core.smtp.session.data.add_delivered_to { + extra_headers = format!("Delivered-To: {deliver_to}\r\n"); + } + + // Spam classification and training + if params.spam_classify + && self.core.spam.enabled + && params.mailbox_ids == [INBOX_ID] + { + // Set the spam filter result + is_spam = self + .core + .spam + .headers + .status + .as_ref() + .and_then(|name| message.header(name.as_str()).and_then(|v| v.as_text())) + .map_or(false, |v| v.contains("Yes")); + + // 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) + { + // 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 _ = write!( + &mut extra_headers, + "{header}: {result}, {score:.2}\r\n", + ); + } + } + Ok(None) => (), + Err(err) => { + trc::error!(err.caused_by(trc::location!())); + } + } + } + + if is_spam { + params.mailbox_ids[0] = JUNK_ID; + params.keywords.push(Keyword::Junk); + } + } } + IngestSource::Jmap | IngestSource::Imap + if params.spam_train && self.core.spam.enabled => + { + if params.keywords.contains(&Keyword::Junk) { + train_spam = Some(true); + } else if params.keywords.contains(&Keyword::NotJunk) { + train_spam = Some(false); + } else if params.mailbox_ids[0] == JUNK_ID { + train_spam = Some(true); + } else if params.mailbox_ids[0] == INBOX_ID { + train_spam = Some(false); + } + } + + _ => (), } // Obtain message references and thread name @@ -177,7 +254,7 @@ impl EmailIngest for Server { } // Check for duplicates - if params.source == IngestSource::Smtp + if params.source.is_smtp() && !message_id.is_empty() && !self .core @@ -223,8 +300,24 @@ impl EmailIngest for Server { } }; + // Add additional headers to message + if !extra_headers.is_empty() { + raw_message_len += extra_headers.len() as u64; + let mut new_message = Vec::with_capacity(raw_message_len as usize); + new_message.extend_from_slice(extra_headers.as_bytes()); + new_message.extend_from_slice(raw_message.as_ref()); + raw_message = Cow::from(new_message); + } + // Encrypt message - if params.encrypt && !message.is_encrypted() { + let do_encrypt = match params.source { + 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 && !message.is_encrypted() { if let Some(encrypt_params) = self .get_property::( account_id, @@ -340,13 +433,25 @@ impl EmailIngest for Server { .set(Property::ThreadId, maybe_thread_id) .tag(Property::ThreadId, TagValue::Id(maybe_thread_id), 0) .set( - ValueClass::FtsQueue(FtsQueueClass { + ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: self.generate_snowflake_id().caused_by(trc::location!())?, hash: blob_id.hash.clone(), }), - 0u64.serialize(), + vec![], ); + // Request spam training + if let Some(learn_spam) = train_spam { + batch.set( + ValueClass::TaskQueue(TaskQueueClass::BayesTrain { + seq: self.generate_snowflake_id()?, + hash: blob_id.hash.clone(), + learn_spam, + }), + vec![], + ); + } + // Insert and obtain ids let ids = self .core @@ -363,17 +468,17 @@ impl EmailIngest for Server { let id = Id::from_parts(thread_id, document_id); // Request FTS index - self.request_fts_index(); + self.notify_task_queue(); trc::event!( MessageIngest(match params.source { - IngestSource::Smtp => + IngestSource::Smtp { .. } => if !is_spam { MessageIngestEvent::Ham } else { MessageIngestEvent::Spam }, - IngestSource::Jmap => MessageIngestEvent::JmapAppend, + IngestSource::Jmap | IngestSource::Restore => MessageIngestEvent::JmapAppend, IngestSource::Imap => MessageIngestEvent::ImapAppend, }), SpanId = params.session_id, @@ -569,6 +674,12 @@ impl LogEmailInsert { } } +impl IngestSource<'_> { + pub fn is_smtp(&self) -> bool { + matches!(self, Self::Smtp { .. }) + } +} + impl SerializeWithId for LogEmailInsert { fn serialize_with_id(&self, ids: &AssignedIds) -> trc::Result> { let thread_id = match self.0 { diff --git a/crates/jmap/src/email/mod.rs b/crates/jmap/src/email/mod.rs index 7c45b756..c2ddb5da 100644 --- a/crates/jmap/src/email/mod.rs +++ b/crates/jmap/src/email/mod.rs @@ -4,6 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +pub mod bayes; pub mod body; pub mod cache; pub mod copy; diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index a1a0f719..d502bd86 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -52,6 +52,7 @@ use crate::{ use std::future::Future; use super::{ + bayes::EmailBayesTrain, delete::EmailDeletion, headers::{BuildHeader, ValueToHeader}, ingest::{EmailIngest, IngestEmail, IngestSource}, @@ -78,6 +79,7 @@ impl EmailSet for Server { let mut response = self .prepare_set_response(&request, Collection::Email) .await?; + let can_train_spam = self.email_bayes_can_train(access_token); // Obtain mailboxIds let mailbox_ids = self.mailbox_get_or_create(account_id).await?; @@ -738,7 +740,8 @@ impl EmailSet for Server { keywords, received_at, source: IngestSource::Jmap, - encrypt: self.core.jmap.encrypt && self.core.jmap.encrypt_append, + spam_classify: false, + spam_train: can_train_spam, session_id: session.session_id, }) .await diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index bd75a9fe..578619ff 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -21,8 +21,8 @@ use jmap_proto::{ types::{collection::Collection, property::Property}, }; use services::{ - delivery::spawn_delivery_manager, housekeeper::spawn_housekeeper, index::spawn_index_task, - state::spawn_state_manager, + delivery::spawn_delivery_manager, housekeeper::spawn_housekeeper, + index::spawn_email_queue_task, state::spawn_state_manager, }; use store::{ @@ -98,7 +98,7 @@ impl SpawnServices for IpcReceivers { spawn_housekeeper(inner.clone(), self.housekeeper_rx.take().unwrap()); // Spawn index task - spawn_index_task(inner); + spawn_email_queue_task(inner); } } diff --git a/crates/jmap/src/services/gossip/ping.rs b/crates/jmap/src/services/gossip/ping.rs index ba2b6346..b2e67de7 100644 --- a/crates/jmap/src/services/gossip/ping.rs +++ b/crates/jmap/src/services/gossip/ping.rs @@ -74,7 +74,7 @@ impl Gossiper { tokio::spawn(async move { trc::event!(Cluster(ClusterEvent::OneOrMorePeersOffline)); - server.request_fts_index(); + server.notify_task_queue(); let _ = server .inner .ipc diff --git a/crates/jmap/src/services/index.rs b/crates/jmap/src/services/index.rs index fa5516f5..1ebb6af7 100644 --- a/crates/jmap/src/services/index.rs +++ b/crates/jmap/src/services/index.rs @@ -6,7 +6,7 @@ use std::{sync::Arc, time::Instant}; -use common::{core::BuildServer, Inner, Server, KV_LOCK_FTS}; +use common::{core::BuildServer, Inner, Server, KV_LOCK_EMAIL_TASK}; use directory::{ backend::internal::{manage::ManageDirectory, PrincipalField}, Type, @@ -17,34 +17,42 @@ use store::{ fts::index::FtsDocument, roaring::RoaringBitmap, write::{ - key::DeserializeBigEndian, BatchBuilder, Bincode, BlobOp, FtsQueueClass, MaybeDynamicId, - ValueClass, + key::{DeserializeBigEndian, KeySerializer}, + BatchBuilder, Bincode, BlobOp, MaybeDynamicId, TaskQueueClass, ValueClass, }, IterateParams, Serialize, ValueKey, U32_LEN, U64_LEN, }; use std::future::Future; -use trc::{AddContext, FtsIndexEvent}; +use trc::{AddContext, TaskQueueEvent}; use utils::{BlobHash, BLOB_HASH_LEN}; use crate::{ blob::download::BlobDownload, changes::write::ChangeLog, - email::{index::IndexMessageText, metadata::MessageMetadata}, + email::{bayes::EmailBayesTrain, index::IndexMessageText, metadata::MessageMetadata}, JmapMethods, }; -#[derive(Debug)] -pub struct IndexEmail { +#[derive(Debug, Clone)] +pub struct EmailTask { account_id: u32, document_id: u32, seq: u64, - insert_hash: BlobHash, + hash: BlobHash, + action: EmailTaskAction, } -const INDEX_LOCK_EXPIRY: u64 = 60 * 5; +#[derive(Debug, Clone, Copy)] +pub enum EmailTaskAction { + Index, + BayesTrain { learn_spam: bool }, +} -pub fn spawn_index_task(inner: Arc) { +const FTS_LOCK_EXPIRY: u64 = 60 * 5; +const BAYES_LOCK_EXPIRY: u64 = 60 * 30; + +pub fn spawn_email_queue_task(inner: Arc) { tokio::spawn(async move { let rx = inner.ipc.index_tx.clone(); let mut locked_seq_ids = AHashMap::new(); @@ -52,7 +60,7 @@ pub fn spawn_index_task(inner: Arc) { // Index any queued messages inner .build_server() - .fts_index_queued(&mut locked_seq_ids) + .email_task_queued(&mut locked_seq_ids) .await; // Wait for a signal to index more messages @@ -62,27 +70,27 @@ pub fn spawn_index_task(inner: Arc) { } pub trait Indexer: Sync + Send { - fn fts_index_queued( + fn email_task_queued( &self, locked_seq_ids: &mut AHashMap, ) -> impl Future + Send; - fn try_lock_index(&self, event: &IndexEmail) -> impl Future + Send; - fn remove_index_lock(&self, seq_id: u64) -> impl Future + Send; + fn try_lock_index(&self, event: &EmailTask) -> impl Future + Send; + fn remove_index_lock(&self, event: &EmailTask) -> impl Future + Send; fn reindex( &self, account_id: Option, tenant_id: Option, ) -> impl Future> + Send; - fn request_fts_index(&self); + fn notify_task_queue(&self); } impl Indexer for Server { - async fn fts_index_queued(&self, locked_seq_ids: &mut AHashMap) { + async fn email_task_queued(&self, locked_seq_ids: &mut AHashMap) { let from_key = ValueKey::> { account_id: 0, collection: 0, document_id: 0, - class: ValueClass::FtsQueue(FtsQueueClass { + class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: 0, hash: BlobHash::default(), }), @@ -91,7 +99,7 @@ impl Indexer for Server { account_id: u32::MAX, collection: u8::MAX, document_id: u32::MAX, - class: ValueClass::FtsQueue(FtsQueueClass { + class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: u64::MAX, hash: BlobHash::default(), }), @@ -107,7 +115,7 @@ impl Indexer for Server { .iterate( IterateParams::new(from_key, to_key).ascending().no_values(), |key, _| { - let entry = IndexEmail::deserialize(key)?; + let entry = EmailTask::deserialize(key)?; if locked_seq_ids .get(&entry.seq) .map_or(true, |expires| now >= *expires) @@ -126,18 +134,21 @@ impl Indexer for Server { }); // Add entries to the index - let mut unlock_seq_ids = Vec::with_capacity(entries.len()); + let mut unlock_events = Vec::with_capacity(entries.len()); for event in entries { let op_start = Instant::now(); // Lock index if !self.try_lock_index(&event).await { locked_seq_ids.insert( event.seq, - Instant::now() + std::time::Duration::from_secs(INDEX_LOCK_EXPIRY + 1), + Instant::now() + std::time::Duration::from_secs(event.lock_expiry() + 1), ); continue; } - unlock_seq_ids.push(event.seq); + + if event.remove_lock() { + unlock_events.push(event.clone()); + } match self .get_property::>( @@ -149,7 +160,7 @@ impl Indexer for Server { .await { Ok(Some(metadata)) - if metadata.inner.blob_hash.as_slice() == event.insert_hash.as_slice() => + if metadata.inner.blob_hash.as_slice() == event.hash.as_slice() => { // Obtain raw message let raw_message = if let Ok(Some(raw_message)) = self @@ -159,7 +170,7 @@ impl Indexer for Server { raw_message } else { trc::event!( - FtsIndex(FtsIndexEvent::BlobNotFound), + TaskQueue(TaskQueueEvent::BlobNotFound), AccountId = event.account_id, DocumentId = event.document_id, BlobId = metadata.inner.blob_hash.to_hex(), @@ -168,29 +179,46 @@ impl Indexer for Server { }; let message = metadata.inner.contents.into_message(&raw_message); - // Index message - let document = - FtsDocument::with_default_language(self.core.jmap.default_language) - .with_account_id(event.account_id) - .with_collection(Collection::Email) - .with_document_id(event.document_id) - .index_message(&message); - if let Err(err) = self.core.storage.fts.index(document).await { - trc::error!(err - .account_id(event.account_id) - .document_id(event.document_id) - .details("Failed to index email in FTS index")); + match event.action { + EmailTaskAction::Index => { + // Index message + let document = + FtsDocument::with_default_language(self.core.jmap.default_language) + .with_account_id(event.account_id) + .with_collection(Collection::Email) + .with_document_id(event.document_id) + .index_message(&message); + if let Err(err) = self.core.storage.fts.index(document).await { + trc::error!(err + .account_id(event.account_id) + .document_id(event.document_id) + .details("Failed to index email in FTS index")); - continue; + continue; + } + + trc::event!( + TaskQueue(TaskQueueEvent::Index), + AccountId = event.account_id, + Collection = Collection::Email, + DocumentId = event.document_id, + Elapsed = op_start.elapsed(), + ); + } + EmailTaskAction::BayesTrain { learn_spam } => { + // Train bayes classifier for account + self.email_bayes_train(event.account_id, 0, message, learn_spam) + .await; + + trc::event!( + TaskQueue(TaskQueueEvent::BayesTrain), + AccountId = event.account_id, + Collection = Collection::Email, + DocumentId = event.document_id, + Elapsed = op_start.elapsed(), + ); + } } - - trc::event!( - FtsIndex(FtsIndexEvent::Index), - AccountId = event.account_id, - Collection = Collection::Email, - DocumentId = event.document_id, - Elapsed = op_start.elapsed(), - ); } Err(err) => { @@ -200,12 +228,12 @@ impl Indexer for Server { .caused_by(trc::location!()) .details("Failed to retrieve email metadata")); - break; + continue; } _ => { // The message was probably deleted or overwritten trc::event!( - FtsIndex(FtsIndexEvent::MetadataNotFound), + TaskQueue(TaskQueueEvent::MetadataNotFound), AccountId = event.account_id, DocumentId = event.document_id, ); @@ -231,14 +259,12 @@ impl Indexer for Server { .account_id(event.account_id) .document_id(event.document_id) .details("Failed to remove index email from queue.")); - - break; } } // Unlock entries - for seq_id in unlock_seq_ids { - self.remove_index_lock(seq_id).await; + for event in unlock_events { + self.remove_index_lock(&event).await; } // Delete expired locks @@ -246,19 +272,19 @@ impl Indexer for Server { locked_seq_ids.retain(|_, expires| *expires > now); } - async fn try_lock_index(&self, event: &IndexEmail) -> bool { + async fn try_lock_index(&self, event: &EmailTask) -> bool { match self .in_memory_store() - .try_lock(KV_LOCK_FTS, &event.seq.to_be_bytes(), INDEX_LOCK_EXPIRY) + .try_lock(KV_LOCK_EMAIL_TASK, &event.lock_key(), event.lock_expiry()) .await { Ok(result) => { if !result { trc::event!( - FtsIndex(FtsIndexEvent::Locked), + TaskQueue(TaskQueueEvent::Locked), AccountId = event.account_id, DocumentId = event.document_id, - Expires = trc::Value::Timestamp(INDEX_LOCK_EXPIRY), + Expires = trc::Value::Timestamp(event.lock_expiry()), ); } result @@ -267,27 +293,27 @@ impl Indexer for Server { trc::error!(err .account_id(event.account_id) .document_id(event.document_id) - .details("Failed to lock FTS index")); + .details("Failed to lock email task")); false } } } - async fn remove_index_lock(&self, seq_id: u64) { + async fn remove_index_lock(&self, event: &EmailTask) { if let Err(err) = self .in_memory_store() - .remove_lock(KV_LOCK_FTS, &seq_id.to_be_bytes()) + .remove_lock(KV_LOCK_EMAIL_TASK, &event.lock_key()) .await { trc::error!(err - .details("Failed to unlock FTS index") - .ctx(trc::Key::Key, seq_id) + .details("Failed to unlock email task") + .ctx(trc::Key::Key, event.seq) .caused_by(trc::location!())); } } - fn request_fts_index(&self) { + fn notify_task_queue(&self) { self.inner.ipc.index_tx.notify_one(); } @@ -376,7 +402,7 @@ impl Indexer for Server { for (document_id, hash) in hashes { batch.update_document(document_id).set( - ValueClass::FtsQueue(FtsQueueClass { hash, seq }), + ValueClass::TaskQueue(TaskQueueClass::IndexEmail { hash, seq }), 0u64.serialize(), ); seq += 1; @@ -396,26 +422,64 @@ impl Indexer for Server { } // Request indexing - self.request_fts_index(); + self.notify_task_queue(); Ok(()) } } -impl IndexEmail { +impl EmailTask { + fn remove_lock(&self) -> bool { + matches!(self.action, EmailTaskAction::Index) + } + + fn lock_key(&self) -> Vec { + match self.action { + EmailTaskAction::Index => KeySerializer::new(U64_LEN + 1) + .write(0u8) + .write(self.seq) + .finalize(), + EmailTaskAction::BayesTrain { .. } => KeySerializer::new((U32_LEN * 2) + 1) + .write(1u8) + .write_leb128(self.account_id) + .write_leb128(self.document_id) + .finalize(), + } + } + + fn lock_expiry(&self) -> u64 { + match self.action { + EmailTaskAction::Index => FTS_LOCK_EXPIRY, + EmailTaskAction::BayesTrain { .. } => BAYES_LOCK_EXPIRY, + } + } + fn value_class(&self) -> ValueClass { - ValueClass::FtsQueue(FtsQueueClass { - hash: self.insert_hash.clone(), - seq: self.seq, + ValueClass::TaskQueue(match self.action { + EmailTaskAction::Index => TaskQueueClass::IndexEmail { + hash: self.hash.clone(), + seq: self.seq, + }, + EmailTaskAction::BayesTrain { learn_spam } => TaskQueueClass::BayesTrain { + hash: self.hash.clone(), + seq: self.seq, + learn_spam, + }, }) } fn deserialize(key: &[u8]) -> trc::Result { - Ok(IndexEmail { + Ok(EmailTask { seq: key.deserialize_be_u64(0)?, account_id: key.deserialize_be_u32(U64_LEN)?, document_id: key.deserialize_be_u32(U64_LEN + U32_LEN + 1)?, - insert_hash: key + action: match key.get(U64_LEN + U32_LEN) { + Some(0) => EmailTaskAction::Index, + Some(1) => EmailTaskAction::BayesTrain { learn_spam: true }, + Some(2) => EmailTaskAction::BayesTrain { learn_spam: false }, + _ => return Err(trc::Error::corrupted_key(key, None, trc::location!())), + }, + hash: key .get( U64_LEN + U32_LEN + U32_LEN + 1 ..U64_LEN + U32_LEN + U32_LEN + BLOB_HASH_LEN + 1, diff --git a/crates/jmap/src/services/ingest.rs b/crates/jmap/src/services/ingest.rs index 71c91e3a..da65194f 100644 --- a/crates/jmap/src/services/ingest.rs +++ b/crates/jmap/src/services/ingest.rs @@ -15,7 +15,10 @@ use std::future::Future; use store::ahash::AHashMap; use crate::{ - email::ingest::{EmailIngest, IngestEmail, IngestSource}, + email::{ + bayes::EmailBayesTrain, + ingest::{EmailIngest, IngestEmail, IngestSource}, + }, mailbox::INBOX_ID, sieve::{get::SieveScriptGet, ingest::SieveScriptIngest}, }; @@ -131,8 +134,10 @@ impl MailDelivery for Server { mailbox_ids: vec![INBOX_ID], keywords: vec![], received_at: None, - source: IngestSource::Smtp, - encrypt: self.core.jmap.encrypt, + source: IngestSource::Smtp { deliver_to: &rcpt }, + spam_classify: access_token + .has_permission(Permission::SpamFilterClassify), + spam_train: self.email_bayes_can_train(&access_token), session_id: message.session_id, }) .await diff --git a/crates/jmap/src/sieve/ingest.rs b/crates/jmap/src/sieve/ingest.rs index c05917db..b4731610 100644 --- a/crates/jmap/src/sieve/ingest.rs +++ b/crates/jmap/src/sieve/ingest.rs @@ -9,7 +9,7 @@ use std::borrow::Cow; use common::{ auth::AccessToken, listener::stream::NullIo, scripts::plugins::PluginContext, Server, }; -use directory::{backend::internal::PrincipalField, QueryBy}; +use directory::{backend::internal::PrincipalField, Permission, QueryBy}; use jmap_proto::types::{collection::Collection, id::Id, keyword::Keyword, property::Property}; use mail_parser::MessageParser; use sieve::{Envelope, Event, Input, Mailbox, Recipient}; @@ -21,7 +21,10 @@ use store::{ use trc::{AddContext, SieveEvent}; use crate::{ - email::ingest::{EmailIngest, IngestEmail, IngestSource, IngestedEmail}, + email::{ + bayes::EmailBayesTrain, + ingest::{EmailIngest, IngestEmail, IngestSource, IngestedEmail}, + }, mailbox::{get::MailboxGet, set::MailboxSet, INBOX_ID, TRASH_ID}, sieve::SeenIdHash, JmapMethods, @@ -456,6 +459,7 @@ 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 @@ -484,8 +488,11 @@ impl SieveScriptIngest for Server { mailbox_ids: sieve_message.file_into, keywords: sieve_message.flags, received_at: None, - source: IngestSource::Smtp, - encrypt: self.core.jmap.encrypt, + source: IngestSource::Smtp { + deliver_to: envelope_to, + }, + spam_classify: access_token.has_permission(Permission::SpamFilterClassify), + spam_train: can_spam_train, session_id, }) .await diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index e4826057..8a57f987 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -421,7 +421,13 @@ impl Session { } // Run SPAM filter - if self.server.core.spam.enabled { + if self.server.core.spam.enabled + && self + .server + .eval_if(&dc.spam_filter, self, self.data.session_id) + .await + .unwrap_or(true) + { match self .spam_classify( &parsed_message, diff --git a/crates/smtp/src/reporting/mod.rs b/crates/smtp/src/reporting/mod.rs index 8ca7f15f..f4e68b55 100644 --- a/crates/smtp/src/reporting/mod.rs +++ b/crates/smtp/src/reporting/mod.rs @@ -250,6 +250,7 @@ pub trait AggregateTimestamp { fn to_timestamp(&self) -> u64; fn to_timestamp_(&self, dt: DateTime) -> u64; fn as_secs(&self) -> u64; + fn due(&self) -> u64; } impl AggregateTimestamp for AggregateFrequency { @@ -293,6 +294,10 @@ impl AggregateTimestamp for AggregateFrequency { AggregateFrequency::Never => 0, } } + + fn due(&self) -> u64 { + self.to_timestamp() + self.as_secs() + } } pub struct SerializedSize { diff --git a/crates/smtp/src/reporting/scheduler.rs b/crates/smtp/src/reporting/scheduler.rs index 3be933d8..f15320c2 100644 --- a/crates/smtp/src/reporting/scheduler.rs +++ b/crates/smtp/src/reporting/scheduler.rs @@ -20,82 +20,103 @@ use tokio::sync::mpsc; use crate::queue::spool::LOCK_EXPIRY; -use super::{dmarc::DmarcReporting, tls::TlsReporting, ReportLock}; +use super::{dmarc::DmarcReporting, tls::TlsReporting, AggregateTimestamp, ReportLock}; pub const REPORT_REFRESH: Duration = Duration::from_secs(86400); impl SpawnReport for mpsc::Receiver { fn spawn(mut self, inner: Arc) { tokio::spawn(async move { - let mut next_wake_up; + let mut next_wake_up = REPORT_REFRESH; + let mut refresh_queue = true; loop { - // Read events - let now = now(); - let events = next_report_event(inner.shared_core.load().storage.data.clone()).await; - next_wake_up = events - .last() - .and_then(|e| match e { - QueueClass::DmarcReportHeader(e) | QueueClass::TlsReportHeader(e) - if e.due > now => - { - Duration::from_secs(e.due - now).into() - } - _ => None, - }) - .unwrap_or(REPORT_REFRESH); - let server = inner.build_server(); - let server_ = server.clone(); - tokio::spawn(async move { - let mut tls_reports = AHashMap::new(); - for report_event in events { - match report_event { - QueueClass::DmarcReportHeader(event) if event.due <= now => { - let lock_name = event.dmarc_lock(); - if server.try_lock_report(&lock_name).await { - server.send_dmarc_aggregate_report(event).await; - server.unlock_report(&lock_name).await; + + if refresh_queue { + // Read events + let events = next_report_event(server.store()).await; + let now = now(); + next_wake_up = events + .last() + .and_then(|e| { + e.due() + .filter(|due| *due > now) + .map(|due| Duration::from_secs(due - now)) + }) + .unwrap_or(REPORT_REFRESH); + + if events + .first() + .and_then(|e| e.due()) + .map_or(false, |due| due <= now) + { + let server_ = server.clone(); + tokio::spawn(async move { + let mut tls_reports = AHashMap::new(); + for report_event in events { + match report_event { + QueueClass::DmarcReportHeader(event) if event.due <= now => { + let lock_name = event.dmarc_lock(); + if server_.try_lock_report(&lock_name).await { + server_.send_dmarc_aggregate_report(event).await; + server_.unlock_report(&lock_name).await; + } + } + QueueClass::TlsReportHeader(event) if event.due <= now => { + tls_reports + .entry(event.domain.clone()) + .or_insert_with(Vec::new) + .push(event); + } + _ => (), } } - QueueClass::TlsReportHeader(event) if event.due <= now => { - tls_reports - .entry(event.domain.clone()) - .or_insert_with(Vec::new) - .push(event); - } - _ => (), - } - } - for (_, tls_report) in tls_reports { - let lock_name = tls_report.first().unwrap().tls_lock(); - if server.try_lock_report(&lock_name).await { - server.send_tls_aggregate_report(tls_report).await; - server.unlock_report(&lock_name).await; - } + for (_, tls_report) in tls_reports { + let lock_name = tls_report.first().unwrap().tls_lock(); + if server_.try_lock_report(&lock_name).await { + server_.send_tls_aggregate_report(tls_report).await; + server_.unlock_report(&lock_name).await; + } + } + }); } - }); + } match tokio::time::timeout(next_wake_up, self.recv()).await { - Ok(Some(event)) => match event { - ReportingEvent::Dmarc(event) => { - server_.schedule_dmarc(event).await; + Ok(Some(event)) => { + refresh_queue = false; + + match event { + ReportingEvent::Dmarc(event) => { + next_wake_up = std::cmp::min( + next_wake_up, + Duration::from_secs(event.interval.due().saturating_sub(now())), + ); + server.schedule_dmarc(event).await; + } + ReportingEvent::Tls(event) => { + next_wake_up = std::cmp::min( + next_wake_up, + Duration::from_secs(event.interval.due().saturating_sub(now())), + ); + server.schedule_tls(event).await; + } + ReportingEvent::Stop => break, } - ReportingEvent::Tls(event) => { - server_.schedule_tls(event).await; - } - ReportingEvent::Stop => break, - }, + } Ok(None) => break, - Err(_) => {} + Err(_) => { + refresh_queue = true; + } } } }); } } -async fn next_report_event(store: Store) -> Vec { +async fn next_report_event(store: &Store) -> Vec { let now = now(); let from_key = ValueKey::from(ValueClass::Queue(QueueClass::DmarcReportHeader( ReportEvent { @@ -182,7 +203,7 @@ impl LockReport for Server { if !result { trc::event!( OutgoingReport(trc::OutgoingReportEvent::Locked), - Expires = trc::Value::Timestamp(LOCK_EXPIRY), + Expires = trc::Value::Timestamp(now() + LOCK_EXPIRY), Key = key ); } diff --git a/crates/spam-filter/src/analysis/bayes.rs b/crates/spam-filter/src/analysis/bayes.rs index ed8837b6..a2a82798 100644 --- a/crates/spam-filter/src/analysis/bayes.rs +++ b/crates/spam-filter/src/analysis/bayes.rs @@ -8,7 +8,7 @@ use std::future::Future; use common::Server; -use crate::{modules::bayes::bayes_classify, SpamFilterContext}; +use crate::{modules::bayes::BayesClassifier, SpamFilterContext}; pub trait SpamFilterAnalyzeBayes: Sync + Send { fn spam_filter_analyze_bayes_classify( @@ -26,7 +26,7 @@ impl SpamFilterAnalyzeBayes for Server { async fn spam_filter_analyze_bayes_classify(&self, ctx: &mut SpamFilterContext<'_>) { if let Some(config) = &self.core.spam.bayes { if !ctx.result.has_tag("SPAM_TRAP") && !ctx.result.has_tag("TRUSTED_REPLY") { - match bayes_classify(self, ctx).await { + match self.bayes_classify(ctx).await { Ok(Some(score)) => { if score > config.score_spam { ctx.result.add_tag("BAYES_SPAM"); diff --git a/crates/spam-filter/src/analysis/from.rs b/crates/spam-filter/src/analysis/from.rs index 93b14479..34b861fd 100644 --- a/crates/spam-filter/src/analysis/from.rs +++ b/crates/spam-filter/src/analysis/from.rs @@ -122,15 +122,15 @@ impl SpamFilterAnalyzeFrom for Server { ctx.result.add_tag("FROM_BOUNCE"); } - if (!env_from_empty && ctx.output.env_from_addr.address == from_addr.address) - || (ctx.output.env_from_postmaster - && from_addr_is_valid - && from_addr.domain_part.sld == ctx.output.ehlo_host.sld) - { + if !env_from_empty && ctx.output.env_from_addr.address == from_addr.address { ctx.result.add_tag("FROM_EQ_ENVFROM"); } else if from_addr_is_valid { - ctx.result.add_tag("FORGED_SENDER"); - ctx.result.add_tag("FROM_NEQ_ENVFROM"); + if from_addr.domain_part.sld == ctx.output.ehlo_host.sld { + ctx.result.add_tag("FROMTLD_EQ_ENVFROMTLD"); + } else if !ctx.output.env_from_postmaster { + ctx.result.add_tag("FORGED_SENDER"); + ctx.result.add_tag("FROM_NEQ_ENVFROM"); + } } // Validate FROM/TO relationship diff --git a/crates/spam-filter/src/analysis/score.rs b/crates/spam-filter/src/analysis/score.rs index f490b968..172a53cf 100644 --- a/crates/spam-filter/src/analysis/score.rs +++ b/crates/spam-filter/src/analysis/score.rs @@ -7,7 +7,7 @@ use common::{config::spamfilter::SpamFilterAction, Server}; use std::{fmt::Write, future::Future, vec}; -use crate::{modules::bayes::bayes_train_if_balanced, SpamFilterContext}; +use crate::{modules::bayes::BayesClassifier, SpamFilterContext}; pub trait SpamFilterAnalyzeScore: Sync + Send { fn spam_filter_score( @@ -78,11 +78,11 @@ impl SpamFilterAnalyzeScore for Server { if ctx.result.has_tag("SPAM_TRAP") || (ctx.result.score >= config.auto_learn_spam_threshold && !was_classified) { - bayes_train_if_balanced(self, ctx, true).await; + 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) { - bayes_train_if_balanced(self, ctx, false).await; + self.bayes_train_if_balanced(ctx, false).await; } } diff --git a/crates/spam-filter/src/analysis/trusted_reply.rs b/crates/spam-filter/src/analysis/trusted_reply.rs index 7995ea4d..3515e0fa 100644 --- a/crates/spam-filter/src/analysis/trusted_reply.rs +++ b/crates/spam-filter/src/analysis/trusted_reply.rs @@ -10,7 +10,7 @@ use common::{Server, KV_TRUSTED_REPLY}; use mail_parser::{HeaderName, HeaderValue}; use store::dispatch::lookup::KeyValue; -use crate::{modules::bayes::bayes_train_if_balanced, SpamFilterContext}; +use crate::{modules::bayes::BayesClassifier, SpamFilterContext}; pub trait SpamFilterAnalyzeTrustedReply: Sync + Send { fn spam_filter_analyze_reply_in( @@ -84,7 +84,7 @@ impl SpamFilterAnalyzeTrustedReply for Server { .as_ref() .map_or(false, |config| config.auto_learn_reply_ham) { - bayes_train_if_balanced(self, ctx, false).await; + self.bayes_train_if_balanced(ctx, false).await; } } } diff --git a/crates/spam-filter/src/analysis/url.rs b/crates/spam-filter/src/analysis/url.rs index e1f14b64..6a5c66ef 100644 --- a/crates/spam-filter/src/analysis/url.rs +++ b/crates/spam-filter/src/analysis/url.rs @@ -162,6 +162,7 @@ impl SpamFilterAnalyzeUrl for Server { parts: url_parsed, }, _ => { + let c = println!("URL {}", url.element); // URL could not be parsed ctx.output.urls.insert(ElementLocation::new( UrlParts::new(url.element), diff --git a/crates/spam-filter/src/lib.rs b/crates/spam-filter/src/lib.rs index 8119f540..469dbc88 100644 --- a/crates/spam-filter/src/lib.rs +++ b/crates/spam-filter/src/lib.rs @@ -9,7 +9,7 @@ pub mod modules; use std::collections::HashSet; use std::hash::{Hash, Hasher}; -use std::net::IpAddr; +use std::net::{IpAddr, Ipv4Addr}; use analysis::url::UrlParts; use analysis::ElementLocation; @@ -128,6 +128,33 @@ pub struct Recipient { pub name: Option, } +impl<'x> SpamFilterInput<'x> { + 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, + } + } +} + impl PartialEq for Hostname { fn eq(&self, other: &Self) -> bool { self.fqdn.eq(&other.fqdn) diff --git a/crates/spam-filter/src/modules/bayes.rs b/crates/spam-filter/src/modules/bayes.rs index df83b46e..2705eea4 100644 --- a/crates/spam-filter/src/modules/bayes.rs +++ b/crates/spam-filter/src/modules/bayes.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::collections::HashSet; +use std::{collections::HashSet, future::Future}; use common::{ip_to_bytes, Server, KV_BAYES_MODEL_GLOBAL, KV_BAYES_MODEL_USER}; use mail_auth::DmarcResult; @@ -23,329 +23,362 @@ use trc::AddContext; use crate::{SpamFilterContext, TextPart}; -pub(crate) async fn bayes_train( - server: &Server, - ctx: &SpamFilterContext<'_>, - is_spam: bool, - is_train: bool, -) -> trc::Result<()> { - // Train the model - let mut model = BayesModel::default(); +pub trait BayesClassifier { + fn bayes_train( + &self, + ctx: &SpamFilterContext<'_>, + is_spam: bool, + is_train: bool, + ) -> impl Future> + Send; - // Train metadata tokens - for token in ctx.spam_tokens() { - model.train_token(TokenHash::from(Gram::Uni { t1: &token }), is_spam); + fn bayes_classify( + &self, + ctx: &SpamFilterContext<'_>, + ) -> impl Future>> + Send; + + fn bayes_is_balanced( + &self, + ctx: &SpamFilterContext<'_>, + learn_spam: bool, + ) -> impl Future> + Send; + + fn bayes_train_if_balanced( + &self, + ctx: &SpamFilterContext<'_>, + learn_spam: bool, + ) -> impl Future + Send; +} + +impl BayesClassifier for Server { + async fn bayes_train( + &self, + ctx: &SpamFilterContext<'_>, + is_spam: bool, + is_train: bool, + ) -> trc::Result<()> { + // Train the model + let mut model = BayesModel::default(); + + // Train metadata tokens + for token in ctx.spam_tokens() { + model.train_token(TokenHash::from(Gram::Uni { t1: &token }), is_spam); + } + + // Train the subject + model.train( + OsbTokenizer::new( + BayesTokenizer::new( + &ctx.output.subject_thread, + ctx.output.subject_tokens.iter().filter_map(to_bayes_token), + ), + 5, + ), + is_spam, + ); + + // Train the body + match ctx + .input + .message + .html_body + .first() + .or_else(|| ctx.input.message.text_body.first()) + .and_then(|idx| ctx.output.text_parts.get(*idx)) + { + Some(TextPart::Html { + text_body, tokens, .. + }) => { + model.train( + OsbTokenizer::new( + BayesTokenizer::new( + text_body, + tokens.iter().filter_map(to_bayes_token_owned), + ), + 5, + ), + is_spam, + ); + } + Some(TextPart::Plain { text_body, tokens }) => { + model.train( + OsbTokenizer::new( + BayesTokenizer::new(text_body, tokens.iter().filter_map(to_bayes_token)), + 5, + ), + is_spam, + ); + } + _ => {} + } + + if model.weights.is_empty() { + trc::bail!(trc::SpamEvent::TrainError + .into_err() + .reason("No weights found")); + } + + trc::event!( + Spam(trc::SpamEvent::Train), + SpanId = ctx.input.span_id, + Details = is_spam, + Total = model.weights.len(), + ); + + // Update weight and invalidate cache + if is_train { + let prefix = if ctx.input.account_id.is_none() { + KV_BAYES_MODEL_GLOBAL + } else { + KV_BAYES_MODEL_USER + }; + for (hash, weights) in model.weights { + self.in_memory_store() + .counter_incr(KeyValue::new( + hash.serialize(prefix, ctx.input.account_id), + i64::from(weights), + )) + .await + .caused_by(trc::location!())?; + } + + // Update training counts + let weights = if is_spam { + Weights { spam: 1, ham: 0 } + } else { + Weights { spam: 0, ham: 1 } + }; + self.in_memory_store() + .counter_incr(KeyValue::new( + TokenHash::serialize_index(prefix, ctx.input.account_id), + i64::from(weights), + )) + .await + .caused_by(trc::location!()) + .map(|_| ()) + } else { + //TODO: Implement untrain + Ok(()) + } } - // Train the subject - model.train( - OsbTokenizer::new( + async fn bayes_classify(&self, ctx: &SpamFilterContext<'_>) -> trc::Result> { + let classifier = if let Some(config) = &self.core.spam.bayes { + &config.classifier + } else { + return Ok(None); + }; + + // Obtain training counts + let prefix = if ctx.input.account_id.is_none() { + KV_BAYES_MODEL_GLOBAL + } else { + KV_BAYES_MODEL_USER + }; + let (spam_learns, ham_learns) = self + .in_memory_store() + .counter_get(TokenHash::serialize_index(prefix, ctx.input.account_id)) + .await + .map(|w| { + let w = Weights::from(w); + (w.spam, w.ham) + })?; + + // Make sure we have enough training data + if spam_learns < classifier.min_learns || ham_learns < classifier.min_learns { + trc::event!( + Spam(trc::SpamEvent::ClassifyError), + SpanId = ctx.input.span_id, + AccountId = ctx.input.account_id, + Reason = "Not enough training data", + Details = vec![ + trc::Value::from(spam_learns), + trc::Value::from(ham_learns), + trc::Value::from(classifier.min_learns) + ], + ); + return Ok(None); + } + + // Classify the text + let mut osb_tokens = Vec::new(); + + // Classify metadata tokens + for token in ctx.spam_tokens() { + let weights = self + .in_memory_store() + .counter_get( + TokenHash::from(Gram::Uni { t1: &token }) + .serialize(prefix, ctx.input.account_id), + ) + .await + .map(Weights::from)?; + osb_tokens.push(OsbToken { + inner: weights, + idx: 1, + }); + } + + // Classify the subject + for token in OsbTokenizer::<_, TokenHash>::new( BayesTokenizer::new( &ctx.output.subject_thread, ctx.output.subject_tokens.iter().filter_map(to_bayes_token), ), 5, - ), - is_spam, - ); + ) { + let weights = self + .in_memory_store() + .counter_get(token.inner.serialize(prefix, ctx.input.account_id)) + .await + .map(Weights::from)?; + osb_tokens.push(OsbToken { + inner: weights, + idx: token.idx, + }); + } - // Train the body - match ctx - .input - .message - .html_body - .first() - .or_else(|| ctx.input.message.text_body.first()) - .and_then(|idx| ctx.output.text_parts.get(*idx)) - { - Some(TextPart::Html { - text_body, tokens, .. - }) => { - model.train( - OsbTokenizer::new( + // Classify the body + match ctx + .input + .message + .html_body + .first() + .or_else(|| ctx.input.message.text_body.first()) + .and_then(|idx| ctx.output.text_parts.get(*idx)) + { + Some(TextPart::Html { + text_body, tokens, .. + }) => { + for token in OsbTokenizer::<_, TokenHash>::new( BayesTokenizer::new(text_body, tokens.iter().filter_map(to_bayes_token_owned)), 5, - ), - is_spam, - ); - } - Some(TextPart::Plain { text_body, tokens }) => { - model.train( - OsbTokenizer::new( + ) { + let weights = self + .in_memory_store() + .counter_get(token.inner.serialize(prefix, ctx.input.account_id)) + .await + .map(Weights::from)?; + osb_tokens.push(OsbToken { + inner: weights, + idx: token.idx, + }); + } + } + Some(TextPart::Plain { text_body, tokens }) => { + for token in OsbTokenizer::<_, TokenHash>::new( BayesTokenizer::new(text_body, tokens.iter().filter_map(to_bayes_token)), 5, - ), - is_spam, - ); - } - _ => {} - } - - if model.weights.is_empty() { - trc::bail!(trc::SpamEvent::TrainError - .into_err() - .reason("No weights found")); - } - - trc::event!( - Spam(trc::SpamEvent::Train), - SpanId = ctx.input.span_id, - Details = is_spam, - Total = model.weights.len(), - ); - - // Update weight and invalidate cache - let prefix = if ctx.input.account_id.is_some() { - KV_BAYES_MODEL_GLOBAL - } else { - KV_BAYES_MODEL_USER - }; - if is_train { - for (hash, weights) in model.weights { - server - .in_memory_store() - .counter_incr(KeyValue::new( - hash.serialize(prefix, ctx.input.account_id), - i64::from(weights), - )) - .await - .caused_by(trc::location!())?; + ) { + let weights = self + .in_memory_store() + .counter_get(token.inner.serialize(prefix, ctx.input.account_id)) + .await + .map(Weights::from)?; + osb_tokens.push(OsbToken { + inner: weights, + idx: token.idx, + }); + } + } + _ => {} } - // Update training counts - let weights = if is_spam { - Weights { spam: 1, ham: 0 } - } else { - Weights { spam: 0, ham: 1 } - }; - server - .in_memory_store() - .counter_incr(KeyValue::new( - TokenHash::serialize_index(prefix, ctx.input.account_id), - i64::from(weights), - )) - .await - .caused_by(trc::location!()) - .map(|_| ()) - } else { - //TODO: Implement untrain - Ok(()) - } -} + let result = classifier.classify(osb_tokens.into_iter(), ham_learns, spam_learns); -pub(crate) async fn bayes_classify( - server: &Server, - ctx: &SpamFilterContext<'_>, -) -> trc::Result> { - let classifier = if let Some(config) = &server.core.spam.bayes { - &config.classifier - } else { - return Ok(None); - }; - - // Obtain training counts - let prefix = if ctx.input.account_id.is_some() { - KV_BAYES_MODEL_GLOBAL - } else { - KV_BAYES_MODEL_USER - }; - let (spam_learns, ham_learns) = server - .in_memory_store() - .counter_get(TokenHash::serialize_index(prefix, ctx.input.account_id)) - .await - .map(|w| { - let w = Weights::from(w); - (w.spam, w.ham) - })?; - - // Make sure we have enough training data - if spam_learns < classifier.min_learns || ham_learns < classifier.min_learns { trc::event!( - Spam(trc::SpamEvent::ClassifyError), + Spam(trc::SpamEvent::Classify), SpanId = ctx.input.span_id, + AccountId = ctx.input.account_id, Details = vec![ trc::Value::from(spam_learns), trc::Value::from(ham_learns), trc::Value::from(classifier.min_learns) ], + Result = result.map(trc::Value::from).unwrap_or_default() ); - return Ok(None); + + Ok(result) } - // Classify the text - let mut osb_tokens = Vec::new(); + async fn bayes_is_balanced( + &self, + ctx: &SpamFilterContext<'_>, + learn_spam: bool, + ) -> trc::Result { + let min_balance = self + .core + .spam + .bayes + .as_ref() + .map_or(0.0, |c| c.classifier.min_balance); - // Classify metadata tokens - for token in ctx.spam_tokens() { - let weights = server - .in_memory_store() - .counter_get( - TokenHash::from(Gram::Uni { t1: &token }).serialize(prefix, ctx.input.account_id), - ) - .await - .map(Weights::from)?; - osb_tokens.push(OsbToken { - inner: weights, - idx: 1, - }); - } - - // Classify the subject - for token in OsbTokenizer::<_, TokenHash>::new( - BayesTokenizer::new( - &ctx.output.subject_thread, - ctx.output.subject_tokens.iter().filter_map(to_bayes_token), - ), - 5, - ) { - let weights = server - .in_memory_store() - .counter_get(token.inner.serialize(prefix, ctx.input.account_id)) - .await - .map(Weights::from)?; - osb_tokens.push(OsbToken { - inner: weights, - idx: token.idx, - }); - } - - // Classify the body - match ctx - .input - .message - .html_body - .first() - .or_else(|| ctx.input.message.text_body.first()) - .and_then(|idx| ctx.output.text_parts.get(*idx)) - { - Some(TextPart::Html { - text_body, tokens, .. - }) => { - for token in OsbTokenizer::<_, TokenHash>::new( - BayesTokenizer::new(text_body, tokens.iter().filter_map(to_bayes_token_owned)), - 5, - ) { - let weights = server - .in_memory_store() - .counter_get(token.inner.serialize(prefix, ctx.input.account_id)) - .await - .map(Weights::from)?; - osb_tokens.push(OsbToken { - inner: weights, - idx: token.idx, - }); - } + if min_balance == 0.0 { + return Ok(true); } - Some(TextPart::Plain { text_body, tokens }) => { - for token in OsbTokenizer::<_, TokenHash>::new( - BayesTokenizer::new(text_body, tokens.iter().filter_map(to_bayes_token)), - 5, - ) { - let weights = server - .in_memory_store() - .counter_get(token.inner.serialize(prefix, ctx.input.account_id)) - .await - .map(Weights::from)?; - osb_tokens.push(OsbToken { - inner: weights, - idx: token.idx, - }); - } - } - _ => {} - } - let result = classifier.classify(osb_tokens.into_iter(), ham_learns, spam_learns); - - trc::event!( - Spam(trc::SpamEvent::Classify), - SpanId = ctx.input.span_id, - Details = vec![ - trc::Value::from(spam_learns), - trc::Value::from(ham_learns), - trc::Value::from(classifier.min_learns) - ], - Result = result.map(trc::Value::from).unwrap_or_default() - ); - - Ok(result) -} - -pub(crate) async fn bayes_is_balanced( - server: &Server, - ctx: &SpamFilterContext<'_>, - learn_spam: bool, -) -> trc::Result { - let min_balance = server - .core - .spam - .bayes - .as_ref() - .map_or(0.0, |c| c.classifier.min_balance); - - if min_balance == 0.0 { - return Ok(true); - } - - // Obtain training counts - let prefix = if ctx.input.account_id.is_some() { - KV_BAYES_MODEL_GLOBAL - } else { - KV_BAYES_MODEL_USER - }; - let (spam_learns, ham_learns) = server - .in_memory_store() - .counter_get(TokenHash::serialize_index(prefix, ctx.input.account_id)) - .await - .map(|w| { - let w = Weights::from(w); - (w.spam as f64, w.ham as f64) - })?; - - let result = if spam_learns > 0.0 || ham_learns > 0.0 { - if learn_spam { - (spam_learns / (ham_learns + 1.0)) <= 1.0 / min_balance + // Obtain training counts + let prefix = if ctx.input.account_id.is_none() { + KV_BAYES_MODEL_GLOBAL } else { - (ham_learns / (spam_learns + 1.0)) <= 1.0 / min_balance - } - } else { - true - }; + KV_BAYES_MODEL_USER + }; + let (spam_learns, ham_learns) = self + .in_memory_store() + .counter_get(TokenHash::serialize_index(prefix, ctx.input.account_id)) + .await + .map(|w| { + let w = Weights::from(w); + (w.spam as f64, w.ham as f64) + })?; - trc::event!( - Spam(trc::SpamEvent::TrainBalance), - SpanId = ctx.input.span_id, - Details = vec![ - trc::Value::from(learn_spam), - trc::Value::from(min_balance), - trc::Value::from(spam_learns), - trc::Value::from(ham_learns), - ], - Result = result - ); + let result = if spam_learns > 0.0 || ham_learns > 0.0 { + if learn_spam { + (spam_learns / (ham_learns + 1.0)) <= 1.0 / min_balance + } else { + (ham_learns / (spam_learns + 1.0)) <= 1.0 / min_balance + } + } else { + true + }; - Ok(result) -} + trc::event!( + Spam(trc::SpamEvent::TrainBalance), + SpanId = ctx.input.span_id, + Details = vec![ + trc::Value::from(learn_spam), + trc::Value::from(min_balance), + trc::Value::from(spam_learns), + trc::Value::from(ham_learns), + ], + Result = result + ); -pub(crate) async fn bayes_train_if_balanced( - server: &Server, - ctx: &SpamFilterContext<'_>, - learn_spam: bool, -) { - let err = match bayes_is_balanced(server, ctx, learn_spam).await { - Ok(true) => match bayes_train(server, ctx, learn_spam, true).await { - Ok(_) => { + Ok(result) + } + + async fn bayes_train_if_balanced(&self, ctx: &SpamFilterContext<'_>, learn_spam: bool) { + let err = match self.bayes_is_balanced(ctx, learn_spam).await { + Ok(true) => match self.bayes_train(ctx, learn_spam, true).await { + Ok(_) => { + return; + } + Err(err) => err, + }, + Ok(false) => { return; } Err(err) => err, - }, - Ok(false) => { - return; - } - Err(err) => err, - }; + }; - trc::error!(err.span_id(ctx.input.span_id).caused_by(trc::location!())); + if let Some(account_id) = ctx.input.account_id { + trc::error!(err + .span_id(ctx.input.span_id) + .account_id(account_id) + .caused_by(trc::location!())); + } else { + trc::error!(err.span_id(ctx.input.span_id).caused_by(trc::location!())); + } + } } const P_FROM_NAME: u8 = 0; @@ -357,7 +390,9 @@ const P_REMOTE_IP: u8 = 4; impl SpamFilterContext<'_> { pub fn spam_tokens(&self) -> HashSet> { let mut tokens = HashSet::new(); - if matches!(self.input.dmarc_result, Some(DmarcResult::Pass)) { + if matches!(self.input.dmarc_result, Some(DmarcResult::Pass)) + || self.input.account_id.is_some() + { for addr in [&self.output.env_from_addr, &self.output.from.email] { if !addr.address.is_empty() { tokens.insert(add_prefix(P_FROM_EMAIL, addr.address.as_bytes())); @@ -376,7 +411,9 @@ impl SpamFilterContext<'_> { if let Some(asn) = self.input.asn { tokens.insert(add_prefix(P_ASN, &asn.to_be_bytes())); } - tokens.insert(add_prefix(P_REMOTE_IP, &ip_to_bytes(&self.input.remote_ip))); + if !self.input.remote_ip.is_loopback() { + tokens.insert(add_prefix(P_REMOTE_IP, &ip_to_bytes(&self.input.remote_ip))); + } tokens } } diff --git a/crates/store/src/backend/mysql/main.rs b/crates/store/src/backend/mysql/main.rs index c28a271b..46fb7647 100644 --- a/crates/store/src/backend/mysql/main.rs +++ b/crates/store/src/backend/mysql/main.rs @@ -93,7 +93,7 @@ impl MysqlStore { for table in [ SUBSPACE_ACL, SUBSPACE_DIRECTORY, - SUBSPACE_FTS_QUEUE, + SUBSPACE_TASK_QUEUE, SUBSPACE_BLOB_RESERVE, SUBSPACE_BLOB_LINK, SUBSPACE_LOOKUP_VALUE, diff --git a/crates/store/src/backend/postgres/main.rs b/crates/store/src/backend/postgres/main.rs index 3f140d87..5780f77c 100644 --- a/crates/store/src/backend/postgres/main.rs +++ b/crates/store/src/backend/postgres/main.rs @@ -80,7 +80,7 @@ impl PostgresStore { for table in [ SUBSPACE_ACL, SUBSPACE_DIRECTORY, - SUBSPACE_FTS_QUEUE, + SUBSPACE_TASK_QUEUE, SUBSPACE_BLOB_RESERVE, SUBSPACE_BLOB_LINK, SUBSPACE_LOOKUP_VALUE, diff --git a/crates/store/src/backend/rocksdb/main.rs b/crates/store/src/backend/rocksdb/main.rs index c933d281..7eaeecb3 100644 --- a/crates/store/src/backend/rocksdb/main.rs +++ b/crates/store/src/backend/rocksdb/main.rs @@ -74,7 +74,7 @@ impl RocksDbStore { SUBSPACE_INDEXES, SUBSPACE_ACL, SUBSPACE_DIRECTORY, - SUBSPACE_FTS_QUEUE, + SUBSPACE_TASK_QUEUE, SUBSPACE_BLOB_RESERVE, SUBSPACE_BLOB_LINK, SUBSPACE_LOOKUP_VALUE, diff --git a/crates/store/src/backend/sqlite/main.rs b/crates/store/src/backend/sqlite/main.rs index 716670f8..d3edbd1d 100644 --- a/crates/store/src/backend/sqlite/main.rs +++ b/crates/store/src/backend/sqlite/main.rs @@ -91,7 +91,7 @@ impl SqliteStore { for table in [ SUBSPACE_ACL, SUBSPACE_DIRECTORY, - SUBSPACE_FTS_QUEUE, + SUBSPACE_TASK_QUEUE, SUBSPACE_BLOB_RESERVE, SUBSPACE_BLOB_LINK, SUBSPACE_LOOKUP_VALUE, diff --git a/crates/store/src/dispatch/store.rs b/crates/store/src/dispatch/store.rs index 3931ded2..d4eaf240 100644 --- a/crates/store/src/dispatch/store.rs +++ b/crates/store/src/dispatch/store.rs @@ -590,7 +590,7 @@ impl Store { SUBSPACE_BITMAP_TAG, SUBSPACE_BITMAP_TEXT, SUBSPACE_DIRECTORY, - SUBSPACE_FTS_QUEUE, + SUBSPACE_TASK_QUEUE, SUBSPACE_INDEXES, SUBSPACE_BLOB_RESERVE, SUBSPACE_BLOB_LINK, @@ -772,7 +772,7 @@ impl Store { for (subspace, with_values) in [ (SUBSPACE_ACL, true), //(SUBSPACE_DIRECTORY, true), - (SUBSPACE_FTS_QUEUE, true), + (SUBSPACE_TASK_QUEUE, true), (SUBSPACE_LOOKUP_VALUE, true), (SUBSPACE_PROPERTY, true), (SUBSPACE_SETTINGS, true), diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 3d5e9f85..41c8a41b 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -135,7 +135,7 @@ pub const SUBSPACE_BITMAP_ID: u8 = b'b'; pub const SUBSPACE_BITMAP_TAG: u8 = b'c'; pub const SUBSPACE_BITMAP_TEXT: u8 = b'v'; pub const SUBSPACE_DIRECTORY: u8 = b'd'; -pub const SUBSPACE_FTS_QUEUE: u8 = b'f'; +pub const SUBSPACE_TASK_QUEUE: u8 = b'f'; pub const SUBSPACE_INDEXES: u8 = b'i'; pub const SUBSPACE_BLOB_RESERVE: u8 = b'j'; pub const SUBSPACE_BLOB_LINK: u8 = b'k'; diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 0918e278..e1522dc3 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -11,15 +11,15 @@ use crate::{ BitmapKey, Deserialize, IndexKey, IndexKeyPrefix, Key, LogKey, ValueKey, SUBSPACE_ACL, SUBSPACE_BITMAP_ID, SUBSPACE_BITMAP_TAG, SUBSPACE_BITMAP_TEXT, SUBSPACE_BLOB_LINK, SUBSPACE_BLOB_RESERVE, SUBSPACE_COUNTER, SUBSPACE_DIRECTORY, SUBSPACE_FTS_INDEX, - SUBSPACE_FTS_QUEUE, SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_LOOKUP_VALUE, SUBSPACE_PROPERTY, + SUBSPACE_INDEXES, SUBSPACE_LOGS, SUBSPACE_LOOKUP_VALUE, SUBSPACE_PROPERTY, SUBSPACE_QUEUE_EVENT, SUBSPACE_QUEUE_MESSAGE, SUBSPACE_QUOTA, SUBSPACE_REPORT_IN, - SUBSPACE_REPORT_OUT, SUBSPACE_SETTINGS, SUBSPACE_TELEMETRY_INDEX, SUBSPACE_TELEMETRY_METRIC, - SUBSPACE_TELEMETRY_SPAN, U32_LEN, U64_LEN, WITH_SUBSPACE, + SUBSPACE_REPORT_OUT, SUBSPACE_SETTINGS, SUBSPACE_TASK_QUEUE, SUBSPACE_TELEMETRY_INDEX, + SUBSPACE_TELEMETRY_METRIC, SUBSPACE_TELEMETRY_SPAN, U32_LEN, U64_LEN, WITH_SUBSPACE, }; use super::{ AnyKey, AssignedIds, BitmapClass, BlobOp, DirectoryClass, LookupClass, QueueClass, ReportClass, - ReportEvent, ResolveId, TagValue, TelemetryClass, ValueClass, + ReportEvent, ResolveId, TagValue, TaskQueueClass, TelemetryClass, ValueClass, }; pub struct KeySerializer { @@ -266,12 +266,24 @@ impl ValueClass { .write(account_id) .write(collection) .write(document_id), - ValueClass::FtsQueue(queue) => serializer - .write(queue.seq) - .write(account_id) - .write(collection) - .write(document_id) - .write::<&[u8]>(queue.hash.as_ref()), + ValueClass::TaskQueue(task) => match task { + TaskQueueClass::IndexEmail { seq, hash } => serializer + .write(*seq) + .write(account_id) + .write(0u8) + .write(document_id) + .write::<&[u8]>(hash.as_ref()), + TaskQueueClass::BayesTrain { + seq, + hash, + learn_spam, + } => serializer + .write(*seq) + .write(account_id) + .write(if *learn_spam { 1u8 } else { 2u8 }) + .write(document_id) + .write::<&[u8]>(hash.as_ref()), + }, ValueClass::Blob(op) => match op { BlobOp::Reserve { hash, until } => serializer .write(account_id) @@ -542,7 +554,7 @@ impl ValueClass { BLOB_HASH_LEN + U32_LEN * 2 + 2 } }, - ValueClass::FtsQueue { .. } => BLOB_HASH_LEN + U64_LEN * 2, + ValueClass::TaskQueue { .. } => BLOB_HASH_LEN + U64_LEN * 2, ValueClass::Queue(q) => match q { QueueClass::Message(_) => U64_LEN, QueueClass::MessageEvent(_) => U64_LEN * 2, @@ -575,7 +587,7 @@ impl ValueClass { } ValueClass::Acl(_) => SUBSPACE_ACL, ValueClass::FtsIndex(_) => SUBSPACE_FTS_INDEX, - ValueClass::FtsQueue { .. } => SUBSPACE_FTS_QUEUE, + ValueClass::TaskQueue { .. } => SUBSPACE_TASK_QUEUE, ValueClass::Blob(op) => match op { BlobOp::Reserve { .. } => SUBSPACE_BLOB_RESERVE, BlobOp::Commit { .. } | BlobOp::Link { .. } | BlobOp::LinkId { .. } => { diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 53bb87fe..edf6f125 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -146,7 +146,7 @@ pub enum ValueClass { Acl(u32), Lookup(LookupClass), FtsIndex(BitmapHash), - FtsQueue(FtsQueueClass), + TaskQueue(TaskQueueClass), Directory(DirectoryClass), Blob(BlobOp), Config(Vec), @@ -157,9 +157,16 @@ pub enum ValueClass { } #[derive(Debug, PartialEq, Clone, Eq, Hash)] -pub struct FtsQueueClass { - pub seq: u64, - pub hash: BlobHash, +pub enum TaskQueueClass { + IndexEmail { + seq: u64, + hash: BlobHash, + }, + BayesTrain { + seq: u64, + hash: BlobHash, + learn_spam: bool, + }, } #[derive(Debug, PartialEq, Clone, Eq, Hash)] @@ -892,3 +899,13 @@ impl RandomAvailableId for RoaringBitmap { available_ids[rand::thread_rng().gen_range(0..available_ids.len())] } } + +impl QueueClass { + pub fn due(&self) -> Option { + match self { + QueueClass::DmarcReportHeader(report_event) => report_event.due.into(), + QueueClass::TlsReportHeader(report_event) => report_event.due.into(), + _ => None, + } + } +} diff --git a/crates/trc/src/event/description.rs b/crates/trc/src/event/description.rs index 64b5ab54..c7fc3e2b 100644 --- a/crates/trc/src/event/description.rs +++ b/crates/trc/src/event/description.rs @@ -39,7 +39,7 @@ impl EventType { EventType::PushSubscription(event) => event.description(), EventType::Cluster(event) => event.description(), EventType::Housekeeper(event) => event.description(), - EventType::FtsIndex(event) => event.description(), + EventType::TaskQueue(event) => event.description(), EventType::Milter(event) => event.description(), EventType::MtaHook(event) => event.description(), EventType::Delivery(event) => event.description(), @@ -87,7 +87,7 @@ impl EventType { EventType::PushSubscription(event) => event.explain(), EventType::Cluster(event) => event.explain(), EventType::Housekeeper(event) => event.explain(), - EventType::FtsIndex(event) => event.explain(), + EventType::TaskQueue(event) => event.explain(), EventType::Milter(event) => event.explain(), EventType::MtaHook(event) => event.explain(), EventType::Delivery(event) => event.explain(), @@ -188,22 +188,24 @@ impl HousekeeperEvent { } } -impl FtsIndexEvent { +impl TaskQueueEvent { pub fn description(&self) -> &'static str { match self { - FtsIndexEvent::Index => "Full-text search index done", - FtsIndexEvent::Locked => "Full-text search index is locked by another process", - FtsIndexEvent::BlobNotFound => "Blob not found for full-text indexing", - FtsIndexEvent::MetadataNotFound => "Metadata not found for full-text indexing", + TaskQueueEvent::Index => "Full-text search indexing completed", + TaskQueueEvent::Locked => "Task is locked by another process", + TaskQueueEvent::BlobNotFound => "Blob not found for task", + TaskQueueEvent::MetadataNotFound => "Metadata not found for task", + TaskQueueEvent::BayesTrain => "Bayesian training completed", } } pub fn explain(&self) -> &'static str { match self { - FtsIndexEvent::Index => "The full-text search index has been updated", - FtsIndexEvent::Locked => "The full-text search index is locked by another process", - FtsIndexEvent::BlobNotFound => "The blob was not found for full-text indexing", - FtsIndexEvent::MetadataNotFound => "The metadata was not found for full-text indexing", + TaskQueueEvent::Index => "The full-text search index has been updated", + TaskQueueEvent::Locked => "The task id is locked by another process", + TaskQueueEvent::BlobNotFound => "The requested blob was not found for task", + TaskQueueEvent::MetadataNotFound => "The metadata was not found for task", + TaskQueueEvent::BayesTrain => "Bayesian training has been completed", } } } @@ -799,8 +801,8 @@ impl OutgoingReportEvent { OutgoingReportEvent::DkimRateLimited => "DKIM report rate limited", OutgoingReportEvent::DmarcReport => "DMARC report sent", OutgoingReportEvent::DmarcRateLimited => "DMARC report rate limited", - OutgoingReportEvent::DmarcAggregateReport => "DMARC aggregate report sent", - OutgoingReportEvent::TlsAggregate => "TLS aggregate report sent", + OutgoingReportEvent::DmarcAggregateReport => "DMARC aggregate is being prepared", + OutgoingReportEvent::TlsAggregate => "TLS aggregate report is being prepared", OutgoingReportEvent::HttpSubmission => "Report submitted via HTTP", OutgoingReportEvent::UnauthorizedReportingAddress => "Unauthorized reporting address", OutgoingReportEvent::ReportingAddressValidationError => { @@ -821,8 +823,8 @@ impl OutgoingReportEvent { OutgoingReportEvent::DkimRateLimited => "The DKIM report was rate limited", OutgoingReportEvent::DmarcReport => "A DMARC report has been sent", OutgoingReportEvent::DmarcRateLimited => "The DMARC report was rate limited", - OutgoingReportEvent::DmarcAggregateReport => "A DMARC aggregate report has been sent", - OutgoingReportEvent::TlsAggregate => "A TLS aggregate report has been sent", + OutgoingReportEvent::DmarcAggregateReport => "A DMARC aggregate report will be sent", + OutgoingReportEvent::TlsAggregate => "A TLS aggregate report will be sent", OutgoingReportEvent::HttpSubmission => "The report was submitted via HTTP", OutgoingReportEvent::UnauthorizedReportingAddress => { "The reporting address is not authorized to send reports" diff --git a/crates/trc/src/event/level.rs b/crates/trc/src/event/level.rs index 6e23f94d..d0a47e93 100644 --- a/crates/trc/src/event/level.rs +++ b/crates/trc/src/event/level.rs @@ -373,11 +373,12 @@ impl EventType { HousekeeperEvent::Start | HousekeeperEvent::Stop => Level::Info, HousekeeperEvent::Run | HousekeeperEvent::Schedule => Level::Debug, }, - EventType::FtsIndex(event) => match event { - FtsIndexEvent::Index => Level::Info, - FtsIndexEvent::BlobNotFound - | FtsIndexEvent::Locked - | FtsIndexEvent::MetadataNotFound => Level::Debug, + EventType::TaskQueue(event) => match event { + TaskQueueEvent::Index => Level::Info, + TaskQueueEvent::BlobNotFound + | TaskQueueEvent::Locked + | TaskQueueEvent::BayesTrain + | TaskQueueEvent::MetadataNotFound => Level::Debug, }, EventType::Dmarc(_) => Level::Debug, EventType::Spf(_) => Level::Debug, diff --git a/crates/trc/src/ipc/metrics.rs b/crates/trc/src/ipc/metrics.rs index 44890edd..ca13b63c 100644 --- a/crates/trc/src/ipc/metrics.rs +++ b/crates/trc/src/ipc/metrics.rs @@ -178,7 +178,7 @@ impl Collector { EventType::Queue(QueueEvent::QueueAutogenerated | QueueEvent::QueueDsn) => { QUEUE_COUNT.increment(); } - EventType::FtsIndex(FtsIndexEvent::Index) => { + EventType::TaskQueue(TaskQueueEvent::Index) => { MESSAGE_INDEX_TIME.observe(elapsed); } EventType::Store(StoreEvent::BlobWrite) => { @@ -595,10 +595,10 @@ impl EventType { | ClusterEvent::Error, ) => true, EventType::Housekeeper(_) => false, - EventType::FtsIndex( - FtsIndexEvent::Index - | FtsIndexEvent::BlobNotFound - | FtsIndexEvent::MetadataNotFound, + EventType::TaskQueue( + TaskQueueEvent::Index + | TaskQueueEvent::BlobNotFound + | TaskQueueEvent::MetadataNotFound, ) => true, EventType::Milter( MilterEvent::ActionAccept diff --git a/crates/trc/src/lib.rs b/crates/trc/src/lib.rs index f0f8c929..3927caf0 100644 --- a/crates/trc/src/lib.rs +++ b/crates/trc/src/lib.rs @@ -172,7 +172,7 @@ pub enum EventType { PushSubscription(PushSubscriptionEvent), Cluster(ClusterEvent), Housekeeper(HousekeeperEvent), - FtsIndex(FtsIndexEvent), + TaskQueue(TaskQueueEvent), Milter(MilterEvent), MtaHook(MtaHookEvent), Delivery(DeliveryEvent), @@ -233,8 +233,9 @@ pub enum HousekeeperEvent { } #[event_type] -pub enum FtsIndexEvent { +pub enum TaskQueueEvent { Index, + BayesTrain, Locked, BlobNotFound, MetadataNotFound, diff --git a/crates/trc/src/serializers/binary.rs b/crates/trc/src/serializers/binary.rs index 21c26333..bd41386b 100644 --- a/crates/trc/src/serializers/binary.rs +++ b/crates/trc/src/serializers/binary.rs @@ -446,10 +446,11 @@ impl EventType { EventType::Eval(EvalEvent::Error) => 138, EventType::Eval(EvalEvent::Result) => 139, EventType::Eval(EvalEvent::StoreNotFound) => 140, - EventType::FtsIndex(FtsIndexEvent::BlobNotFound) => 141, - EventType::FtsIndex(FtsIndexEvent::Index) => 142, - EventType::FtsIndex(FtsIndexEvent::Locked) => 144, - EventType::FtsIndex(FtsIndexEvent::MetadataNotFound) => 145, + EventType::TaskQueue(TaskQueueEvent::BlobNotFound) => 141, + EventType::TaskQueue(TaskQueueEvent::Index) => 142, + EventType::TaskQueue(TaskQueueEvent::BayesTrain) => 143, + EventType::TaskQueue(TaskQueueEvent::Locked) => 144, + EventType::TaskQueue(TaskQueueEvent::MetadataNotFound) => 145, EventType::Housekeeper(HousekeeperEvent::Run) => 146, EventType::Housekeeper(HousekeeperEvent::Schedule) => 149, EventType::Housekeeper(HousekeeperEvent::Start) => 150, @@ -1008,10 +1009,11 @@ impl EventType { 138 => Some(EventType::Eval(EvalEvent::Error)), 139 => Some(EventType::Eval(EvalEvent::Result)), 140 => Some(EventType::Eval(EvalEvent::StoreNotFound)), - 141 => Some(EventType::FtsIndex(FtsIndexEvent::BlobNotFound)), - 142 => Some(EventType::FtsIndex(FtsIndexEvent::Index)), - 144 => Some(EventType::FtsIndex(FtsIndexEvent::Locked)), - 145 => Some(EventType::FtsIndex(FtsIndexEvent::MetadataNotFound)), + 141 => Some(EventType::TaskQueue(TaskQueueEvent::BlobNotFound)), + 142 => Some(EventType::TaskQueue(TaskQueueEvent::Index)), + 143 => Some(EventType::TaskQueue(TaskQueueEvent::BayesTrain)), + 144 => Some(EventType::TaskQueue(TaskQueueEvent::Locked)), + 145 => Some(EventType::TaskQueue(TaskQueueEvent::MetadataNotFound)), 146 => Some(EventType::Housekeeper(HousekeeperEvent::Run)), 149 => Some(EventType::Housekeeper(HousekeeperEvent::Schedule)), 150 => Some(EventType::Housekeeper(HousekeeperEvent::Start)), @@ -1469,6 +1471,8 @@ impl EventType { } } +// 147 148 335 336 376 458 459 + impl Key { fn code(&self) -> u64 { match self { diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 74cb017f..a92fb932 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -6,7 +6,7 @@ resolver = "2" [features] #default = ["sqlite", "postgres", "mysql", "rocks", "elastic", "s3", "redis", "azure", "foundationdb"] -default = ["rocks"] +default = ["rocks", "sqlite"] sqlite = ["store/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres"] diff --git a/tests/resources/smtp/antispam/from.test b/tests/resources/smtp/antispam/from.test index 2f75a125..ad49442f 100644 --- a/tests/resources/smtp/antispam/from.test +++ b/tests/resources/smtp/antispam/from.test @@ -48,14 +48,14 @@ From: "hello@other.domain.co.uk" Test helo_domain mx.domain.co.uk -expect FROM_EQ_ENVFROM FROM_NEQ_DISPLAY_NAME FROM_HAS_DN FROM_BOUNCE +expect FROMTLD_EQ_ENVFROMTLD FROM_NEQ_DISPLAY_NAME FROM_HAS_DN FROM_BOUNCE From: "postmaster@mx.domain.co.uk" Test helo_domain mx.domain.co.uk -expect FROM_EQ_ENVFROM FROM_HAS_DN FROM_BOUNCE +expect FROMTLD_EQ_ENVFROMTLD FROM_HAS_DN FROM_BOUNCE From: "Mailer Daemon" diff --git a/tests/resources/smtp/antispam/url.test b/tests/resources/smtp/antispam/url.test index 7256e9fd..e6da8d67 100644 --- a/tests/resources/smtp/antispam/url.test +++ b/tests/resources/smtp/antispam/url.test @@ -28,7 +28,7 @@ Subject: test my site is https://www.xn--80ak6aa92e.com/ -expect R_SUSPICIOUS_URL +expect R_UNPARSABLE_URL Subject: test diff --git a/tests/src/imap/bayes.rs b/tests/src/imap/bayes.rs new file mode 100644 index 00000000..dd011b93 --- /dev/null +++ b/tests/src/imap/bayes.rs @@ -0,0 +1,678 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use common::KV_BAYES_MODEL_USER; +use directory::backend::internal::manage::ManageDirectory; +use imap_proto::ResponseType; +use nlp::bayes::{TokenHash, Weights}; + +use crate::{ + imap::Type, + jmap::{delivery::SmtpConnection, wait_for_index}, + smtp::session::VerifyResponse, +}; + +use super::{IMAPTest, ImapConnection}; + +pub async fn test(handle: &IMAPTest) { + println!("Running Bayes tests..."); + let mut imap = ImapConnection::connect(b"_x ").await; + imap.assert_read(Type::Untagged, ResponseType::Ok).await; + imap.send("AUTHENTICATE PLAIN AGJheWVzQGV4YW1wbGUuY29tAHNlY3JldA==") + .await; + imap.assert_read(Type::Tagged, ResponseType::Ok).await; + + // Make sure the bayes classifier is empty + let account_id = handle + .server + .store() + .get_principal_id("bayes@example.com") + .await + .unwrap() + .unwrap(); + let w = handle.spam_weights(account_id).await; + assert_eq!(w.ham, 0); + assert_eq!(w.spam, 0); + + // Train the classifier via APPEND + imap.append("INBOX", HAM[0]).await; + imap.append("Junk Mail", SPAM[0]).await; + let w = handle.spam_weights(account_id).await; + assert_eq!(w.ham, 1); + assert_eq!(w.spam, 1); + + // Append two spam samples to "Drafts", then train the classifier via STORE and MOVE + imap.append("Drafts", SPAM[1]).await; + imap.append("Drafts", SPAM[2]).await; + imap.send_ok("SELECT Drafts").await; + imap.send_ok("STORE 1 +FLAGS ($Junk)").await; + imap.send_ok("MOVE 2 \"Junk Mail\"").await; + let w = handle.spam_weights(account_id).await; + assert_eq!(w.ham, 1); + assert_eq!(w.spam, 3); + + // Add the remaining messages via APPEND + for message in HAM.iter().skip(1) { + imap.append("INBOX", message).await; + } + for message in SPAM.iter().skip(3) { + imap.append("Junk Mail", message).await; + } + let w = handle.spam_weights(account_id).await; + assert_eq!(w.ham, 10); + assert_eq!(w.spam, 10); + + // Send 3 test emails + for message in TEST { + let mut lmtp = SmtpConnection::connect_port(11201).await; + lmtp.ingest("bill@example.com", &["bayes@example.com"], message) + .await; + } + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + imap.send_ok("SELECT INBOX").await; + imap.send("FETCH 11 (FLAGS RFC822.TEXT)").await; + imap.assert_read(Type::Tagged, ResponseType::Ok) + .await + .assert_not_contains("FLAGS ($Junk") + .assert_contains("Subject: can someone explain") + .assert_contains("X-Spam-Bayes: No"); + imap.send("FETCH 12 (FLAGS RFC822.TEXT)").await; + imap.assert_read(Type::Tagged, ResponseType::Ok) + .await + .assert_not_contains("FLAGS ($Junk") + .assert_contains("Subject: classifier test") + .assert_not_contains("X-Spam-Bayes: "); + imap.send_ok("SELECT \"Junk Mail\"").await; + imap.send("FETCH 10 (FLAGS RFC822.TEXT)").await; + imap.assert_read(Type::Tagged, ResponseType::Ok) + .await + .assert_contains("FLAGS ($Junk") + .assert_contains("Subject: save up to") + .assert_contains("X-Spam-Bayes: Yes"); + imap.send_ok("MOVE 10 INBOX").await; + let w = handle.spam_weights(account_id).await; + assert_eq!(w.ham, 11); + assert_eq!(w.spam, 10); +} + +impl ImapConnection { + async fn append(&mut self, mailbox: &str, message: &str) { + self.send_ok(&format!( + "APPEND {:?} {{{}+}}\r\n{}", + mailbox, + message.len(), + message + )) + .await; + } + + async fn send_ok(&mut self, cmd: &str) { + self.send(cmd).await; + self.assert_read(Type::Tagged, ResponseType::Ok).await; + } +} + +impl IMAPTest { + async fn spam_weights(&self, account_id: u32) -> Weights { + wait_for_index(&self.server).await; + + self.server + .in_memory_store() + .counter_get(TokenHash::serialize_index( + KV_BAYES_MODEL_USER, + account_id.into(), + )) + .await + .map(Weights::from) + .unwrap() + } +} + +const SPAM: [&str; 10] = [ + concat!( + "Subject: save up to NUMBER on life insurance\r\n\r\n wh", + "y spend more than you have to life quote savings e", + "nsuring your family s financial security is very i", + "mportant life quote savings makes buying life insu", + "rance simple and affordable we provide free access", + " to the very best companies and the lowest rates l", + "ife quote savings is fast easy and saves you money", + " let us help you get started with the best values ", + "in the country on new coverage you can save hundre", + "ds or even thousands of dollars by requesting a fr", + "ee quote from lifequote savings our service will t", + "ake you less than NUMBER minutes to complete shop ", + "and compare save up to NUMBER on all types of life", + " insurance hyperlink click here for your free quot", + "e protecting your family is the best investment yo", + "u ll ever make if you are in receipt of this email", + " in error and or wish to be removed from our list ", + "hyperlink please click here and type remove if you", + " reside in any state which prohibits e mail solici", + "tations for insurance please disregard this email\r\n", + " \r\n" + ), + concat!( + "Subject: a powerhouse gifting program\r\n\r\nyou don t ", + "want to miss get in with the founders the major pl", + "ayers are on this one for once be where the player", + "s are this is your private invitation experts are ", + "calling this the fastest way to huge cash flow eve", + "r conceived leverage NUMBER NUMBER into NUMBER NUM", + "BER over and over again the question here is you e", + "ither want to be wealthy or you don t which one ar", + "e you i am tossing you a financial lifeline and fo", + "r your sake i hope you grab onto it and hold on ti", + "ght for the ride of your life testimonials hear wh", + "at average people are doing their first few days w", + "e ve received NUMBER NUMBER in NUMBER day and we a", + "re doing that over and over again q s in al i m a ", + "single mother in fl and i ve received NUMBER NUMBE", + "R in the last NUMBER days d s in fl i was not sure", + " about this when i sent off my NUMBER NUMBER pledg", + "e but i got back NUMBER NUMBER the very next day l", + " l in ky i didn t have the money so i found myself", + " a partner to work this with we have received NUMB", + "ER NUMBER over the last NUMBER days i think i made", + " the right decision don t you k c in fl i pick up ", + "NUMBER NUMBER my first day and i they gave me free", + " leads and all the training you can too j w in ca ", + "announcing we will close your sales for you and he", + "lp you get a fax blast immediately upon your entry", + " you make the money free leads training don t wait", + " call now fax back to NUMBER NUMBER NUMBER NUMBER ", + "or call NUMBER NUMBER NUMBER NUMBER name__________", + "________________________phone_____________________", + "______________________ fax________________________", + "_____________email________________________________", + "____________ best time to call____________________", + "_____time zone____________________________________", + "____ this message is sent in compliance of the new", + " e mail bill per section NUMBER paragraph a NUMBER", + " c of s NUMBER further transmissions by the sender", + " of this email may be stopped at no cost to you by", + " sending a reply to this email address with the wo", + "rd remove in the subject line errors omissions and", + " exceptions excluded this is not spam i have compi", + "led this list from our replicate database relative", + " to seattle marketing group the gigt or turbo team", + " for the sole purpose of these communications your", + " continued inclusion is only by your gracious perm", + "ission if you wish to not receive this mail from m", + "e please send an email to tesrewinter URL with rem", + "ove in the subject and you will be deleted immedia", + "tely\r\n\r\n" + ), + concat!( + "Subject: help wanted \r\n\r\nwe are a NUMBER year old f", + "ortune NUMBER company that is growing at a tremend", + "ous rate we are looking for individuals who want t", + "o work from home this is an opportunity to make an", + " excellent income no experience is required we wil", + "l train you so if you are looking to be employed f", + "rom home with a career that has vast opportunities", + " then go URL we are looking for energetic and self", + " motivated people if that is you than click on the", + " link and fill out the form and one of our employe", + "ment specialist will contact you to be removed fro", + "m our link simple go to URL \r\n\r\n" + ), + concat!( + "Subject: tired of the bull out there\r\n\r\n want to st", + "op losing money want a real money maker receive NU", + "MBER NUMBER NUMBER NUMBER today experts are callin", + "g this the fastest way to huge cash flow ever conc", + "eived a powerhouse gifting program you don t want ", + "to miss we work as a team this is your private inv", + "itation get in with the founders this is where the", + " big boys play the major players are on this one f", + "or once be where the players are this is a system ", + "that will drive NUMBER NUMBER s to your doorstep i", + "n a short period of time leverage NUMBER NUMBER in", + "to NUMBER NUMBER over and over again the question ", + "here is you either want to be wealthy or you don t", + " which one are you i am tossing you a financial li", + "feline and for your sake i hope you grab onto it a", + "nd hold on tight for the ride of your life testimo", + "nials hear what average people are doing their fir", + "st few days we ve received NUMBER NUMBER in NUMBER", + " day and we are doing that over and over again q s", + " in al i m a single mother in fl and i ve received", + " NUMBER NUMBER in the last NUMBER days d s in fl i", + " was not sure about this when i sent off my NUMBER", + " NUMBER pledge but i got back NUMBER NUMBER the ve", + "ry next day l l in ky i didn t have the money so i", + " found myself a partner to work this with we have ", + "received NUMBER NUMBER over the last NUMBER days i", + " think i made the right decision don t you k c in ", + "fl i pick up NUMBER NUMBER my first day and i they", + " gave me free leads and all the training you can t", + "oo j w in ca this will be the most important call ", + "you make this year free leads training announcing ", + "we will close your sales for you and help you get ", + "a fax blast immediately upon your entry you make t", + "he money free leads training don t wait call now N", + "UMBER NUMBER NUMBER NUMBER print and fax to NUMBER", + " NUMBER NUMBER NUMBER or send an email requesting ", + "more information to successleads URL please includ", + "e your name and telephone number receive NUMBER NU", + "MBER free leads just for responding a NUMBER NUMBE", + "R value name___________________________________ ph", + "one___________________________________ fax________", + "_____________________________ email_______________", + "____________________ this message is sent in compl", + "iance of the new e mail bill per section NUMBER pa", + "ragraph a NUMBER c of s NUMBER further transmissio", + "ns by the sender of this email may be stopped at n", + "o cost to you by sending a reply to this email add", + "ress with the word remove in the subject line erro", + "rs omissions and exceptions excluded this is not s", + "pam i have compiled this list from our replicate d", + "atabase relative to seattle marketing group the gi", + "gt or turbo team for the sole purpose of these com", + "munications your continued inclusion is only by yo", + "ur gracious permission if you wish to not receive ", + "this mail from me please send an email to tesrewin", + "ter URL with remove in the subject and you will be", + " deleted immediately\r\n\r\n" + ), + concat!( + "Subject: cellular phone accessories \r\n\r\n all at bel", + "ow wholesale prices http NUMBER NUMBER NUMBER NUMB", + "ER NUMBER sites merchant sales hands free ear buds", + " NUMBER NUMBER phone holsters NUMBER NUMBER booste", + "r antennas only NUMBER NUMBER phone cases NUMBER N", + "UMBER car chargers NUMBER NUMBER face plates as lo", + "w as NUMBER NUMBER lithium ion batteries as low as", + " NUMBER NUMBER http NUMBER NUMBER NUMBER NUMBER NU", + "MBER sites merchant sales click below for accessor", + "ies on all nokia motorola lg nextel samsung qualco", + "mm ericsson audiovox phones at below wholesale pri", + "ces http NUMBER NUMBER NUMBER NUMBER NUMBER sites ", + "merchant sales if you need assistance please call ", + "us NUMBER NUMBER NUMBER to be removed from future ", + "mailings please send your remove request to remove", + " me now NUMBER URL thank you and have a super day\r\n", + " \r\n" + ), + concat!( + "Subject: conferencing made easy\r\n\r\n only NUMBER cen", + "ts per minute including long distance no setup fee", + "s no contracts or monthly fees call anytime from a", + "nywhere to anywhere connects up to NUMBER particip", + "ants simplicity in set up and administration opera", + "tor help available NUMBER NUMBER the highest quali", + "ty service for the lowest rate in the industry fil", + "l out the form below to find out how you can lower", + " your phone bill every month required input field ", + "name web address company name state business phone", + " home phone email address type of business to be r", + "emoved from our distribution lists please hyperlin", + "k click here\r\n\r\n" + ), + concat!( + "Subject: dear friend\r\n\r\n i am mrs sese seko widow o", + "f late president mobutu sese seko of zaire now kno", + "wn as democratic republic of congo drc i am moved ", + "to write you this letter this was in confidence co", + "nsidering my presentcircumstance and situation i e", + "scaped along with my husband and two of our sons g", + "eorge kongolo and basher out of democratic republi", + "c of congo drc to abidjan cote d ivoire where my f", + "amily and i settled while we later moved to settle", + "d in morroco where my husband later died of cancer", + " disease however due to this situation we decided ", + "to changed most of my husband s billions of dollar", + "s deposited in swiss bank and other countries into", + " other forms of money coded for safe purpose becau", + "se the new head of state of dr mr laurent kabila h", + "as made arrangement with the swiss government and ", + "other european countries to freeze all my late hus", + "band s treasures deposited in some european countr", + "ies hence my children and i decided laying low in ", + "africa to study the situation till when things get", + "s better like now that president kabila is dead an", + "d the son taking over joseph kabila one of my late", + " husband s chateaux in southern france was confisc", + "ated by the french government and as such i had to", + " change my identity so that my investment will not", + " be traced and confiscated i have deposited the su", + "m eighteen million united state dollars us NUMBER ", + "NUMBER NUMBER NUMBER with a security company for s", + "afekeeping the funds are security coded to prevent", + " them from knowing the content what i want you to ", + "do is to indicate your interest that you will assi", + "st us by receiving the money on our behalf acknowl", + "edge this message so that i can introduce you to m", + "y son kongolo who has the out modalities for the c", + "laim of the said funds i want you to assist in inv", + "esting this money but i will not want my identity ", + "revealed i will also want to buy properties and st", + "ock in multi national companies and to engage in o", + "ther safe and non speculative investments may i at", + " this point emphasise the high level of confidenti", + "ality which this business demands and hope you wil", + "l not betray the trust and confidence which i repo", + "se in you in conclusion if you want to assist us m", + "y son shall put you in the picture of the business", + " tell you where the funds are currently being main", + "tained and also discuss other modalities including", + " remunerationfor your services for this reason kin", + "dly furnish us your contact information that is yo", + "ur personal telephone and fax number for confident", + "ial URL regards mrs m sese seko\r\n\r\n" + ), + concat!( + "Subject: lowest rates available for term life insu", + "rance\r\n\r\n take a moment and fill out our online for", + "m to see the low rate you qualify for save up to N", + "UMBER from regular rates smokers accepted URL repr", + "esenting quality nationwide carriers act now to ea", + "sily remove your address from the list go to URL p", + "lease allow NUMBER NUMBER hours for removal\r\n\r\n" + ), + concat!( + "Subject: central bank of nigeria foreign remittanc", + "e \r\n\r\n dept tinubu square lagos nigeria email smith", + "_j URL NUMBERth of august NUMBER attn president ce", + "o strictly private business proposal i am mr johns", + "on s abu the bills and exchange director at the fo", + "reignremittance department of the central bank of ", + "nigeria i am writingyou this letter to ask for you", + "r support and cooperation to carrying thisbusiness", + " opportunity in my department we discovered abando", + "ned the sumof us NUMBER NUMBER NUMBER NUMBER thirt", + "y seven million four hundred thousand unitedstates", + " dollars in an account that belong to one of our f", + "oreign customers an american late engr john creek ", + "junior an oil merchant with the federal government", + " of nigeria who died along with his entire family ", + "of a wifeand two children in kenya airbus aNUMBER ", + "NUMBER flight kqNUMBER in novemberNUMBER since we ", + "heard of his death we have been expecting his next", + " of kin tocome over and put claims for his money a", + "s the heir because we cannotrelease the fund from ", + "his account unless someone applies for claims asth", + "e next of kin to the deceased as indicated in our ", + "banking guidelines unfortunately neither their fam", + "ily member nor distant relative hasappeared to cla", + "im the said fund upon this discovery i and other o", + "fficialsin my department have agreed to make busin", + "ess with you release the totalamount into your acc", + "ount as the heir of the fund since no one came for", + "it or discovered either maintained account with ou", + "r bank other wisethe fund will be returned to the ", + "bank treasury as unclaimed fund we have agreed tha", + "t our ratio of sharing will be as stated thus NUMB", + "ER for you as foreign partner and NUMBER for us th", + "e officials in my department upon the successful c", + "ompletion of this transfer my colleague and i will", + "come to your country and mind our share it is from", + " our NUMBER we intendto import computer accessorie", + "s into my country as way of recycling thefund to c", + "ommence this transaction we require you to immedia", + "tely indicateyour interest by calling me or sendin", + "g me a fax immediately on the abovetelefax and enc", + "lose your private contact telephone fax full namea", + "nd address and your designated banking co ordinate", + "s to enable us fileletter of claim to the appropri", + "ate department for necessary approvalsbefore the t", + "ransfer can be made note also this transaction mus", + "t be kept strictly confidential becauseof its natu", + "re nb please remember to give me your phone and fa", + "x no mr johnson smith abu irish linux users group ", + "ilug URL URL for un subscription information list ", + "maintainer listmaster URL\r\n\r\n" + ), + concat!( + "Subject: dear stuart\r\n\r\n are you tired of searching", + " for love in all the wrong places find love now at", + " URL URL browse through thousands of personals in ", + "your area join for free URL search e mail chat use", + " URL to meet cool guys and hot girls go NUMBER on ", + "NUMBER or use our private chat rooms click on the ", + "link to get started URL find love now you have rec", + "eived this email because you have registerd with e", + "mailrewardz or subscribed through one of our marke", + "ting partners if you have received this message in", + " error or wish to stop receiving these great offer", + "s please click the remove link above to unsubscrib", + "e from these mailings please click here URL\r\n\r\n" + ), +]; + +const HAM: [&str; 10] = [ + concat!( + "Message-ID: \r\nSubject: i have been", + " trying to research via sa mirrors and search engi", + "nes\r\n\r\nif a canned script exists giving clients acce", + "ss to their user_prefs options via a web based cgi", + " interface numerous isps provide this feature to c", + "lients but so far i can find nothing our configura", + "tion uses amavis postfix and clamav for virus filt", + "ering and procmail with spamassassin for spam filt", + "ering i would prefer not to have to write a script", + " myself but will appreciate any suggestions this U", + "RL email is sponsored by osdn tired of that same o", + "ld cell phone get a new here for free URL ________", + "_______________________________________ spamassass", + "in talk mailing list spamassassin talk URL URL\r\n\r\n" + ), + concat!( + "Message-ID: mid2@foobar.org\r\nSubject: hello\r\n\r\nhave y", + "ou seen and discussed this article and his approac", + "h thank you URL hell there are no rules here we re", + " trying to accomplish something thomas alva edison", + " this URL email is sponsored by osdn tired of that", + " same old cell phone get a new here for free URL _", + "______________________________________________ spa", + "massassin devel mailing list spamassassin devel UR", + "L URL \r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: hi all apol", + "ogies for the possible silly question\r\n\r\ni don t thi", + "nk it is but but is eircom s adsl service nat ed a", + "nd what implications would that have for voip i kn", + "ow there are difficulties with voip or connecting ", + "to clients connected to a nat ed network from the ", + "internet wild i e machines with static real ips an", + "y help pointers would be helpful cheers rgrds bern", + "ard bernard tyers national centre for sensor resea", + "rch p NUMBER NUMBER NUMBER NUMBER e bernard tyers ", + "URL w URL l nNUMBER ______________________________", + "_________________ iiu mailing list iiu URL URL \r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: can someone", + " explain\r\n\r\nwhat type of operating system solaris is", + " as ive never seen or used it i dont know wheather", + " to get a server from sun or from dell i would pre", + "fer a linux based server and sun seems to be the o", + "ne for that but im not sure if solaris is a distro", + " of linux or a completely different operating syst", + "em can someone explain kiall mac innes irish linux", + " users group ilug URL URL for un subscription info", + "rmation list maintainer listmaster URL \r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: folks my fi", + "rst time posting\r\n\r\nhave a bit of unix experience bu", + "t am new to linux just got a new pc at home dell b", + "ox with windows xp added a second hard disk for li", + "nux partitioned the disk and have installed suse N", + "UMBER NUMBER from cd which went fine except it did", + "n t pick up my monitor i have a dell branded eNUMB", + "ERfpp NUMBER lcd flat panel monitor and a nvidia g", + "eforceNUMBER tiNUMBER video card both of which are", + " probably too new to feature in suse s default set", + " i downloaded a driver from the nvidia website and", + " installed it using rpm then i ran saxNUMBER as wa", + "s recommended in some postings i found on the net ", + "but it still doesn t feature my video card in the ", + "available list what next another problem i have a ", + "dell branded keyboard and if i hit caps lock twice", + " the whole machine crashes in linux not windows ev", + "en the on off switch is inactive leaving me to rea", + "ch for the power cable instead if anyone can help ", + "me in any way with these probs i d be really grate", + "ful i ve searched the net but have run out of idea", + "s or should i be going for a different version of ", + "linux such as redhat opinions welcome thanks a lot", + " peter irish linux users group ilug URL URL for un", + " subscription information list maintainer listmast", + "er URL\r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: has anyone\r\n", + "\r\nseen heard of used some package that would let a ", + "random person go to a webpage create a mailing lis", + "t then administer that list also of course let ppl", + " sign up for the lists and manage their subscripti", + "ons similar to the old URL but i d like to have it", + " running on my server not someone elses chris URL ", + "\r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: hi thank yo", + "u for the useful replies\r\n\r\ni have found some intere", + "sting tutorials in the ibm developer connection UR", + "L and URL registration is needed i will post the s", + "ame message on the web application security list a", + "s suggested by someone for now i thing i will use ", + "mdNUMBER for password checking i will use the appr", + "oach described in secure programmin fo linux and u", + "nix how to i will separate the authentication modu", + "le so i can change its implementation at anytime t", + "hank you again mario torre please avoid sending me", + " word or powerpoint attachments see URL \r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: hehe sorry\r\n", + "\r\nbut if you hit caps lock twice the computer crash", + "es theres one ive never heard before have you trye", + "d dell support yet i think dell computers prefer r", + "edhat dell provide some computers pre loaded with ", + "red hat i dont know for sure tho so get someone el", + "ses opnion as well as mine original message from i", + "lug admin URL mailto ilug admin URL on behalf of p", + "eter staunton sent NUMBER august NUMBER NUMBER NUM", + "BER to ilug URL subject ilug newbie seeks advice s", + "use NUMBER NUMBER folks my first time posting have", + " a bit of unix experience but am new to linux just", + " got a new pc at home dell box with windows xp add", + "ed a second hard disk for linux partitioned the di", + "sk and have installed suse NUMBER NUMBER from cd w", + "hich went fine except it didn t pick up my monitor", + " i have a dell branded eNUMBERfpp NUMBER lcd flat ", + "panel monitor and a nvidia geforceNUMBER tiNUMBER ", + "video card both of which are probably too new to f", + "eature in suse s default set i downloaded a driver", + " from the nvidia website and installed it using rp", + "m then i ran saxNUMBER as was recommended in some ", + "postings i found on the net but it still doesn t f", + "eature my video card in the available list what ne", + "xt another problem i have a dell branded keyboard ", + "and if i hit caps lock twice the whole machine cra", + "shes in linux not windows even the on off switch i", + "s inactive leaving me to reach for the power cable", + " instead if anyone can help me in any way with the", + "se probs i d be really grateful i ve searched the ", + "net but have run out of ideas or should i be going", + " for a different version of linux such as redhat o", + "pinions welcome thanks a lot peter irish linux use", + "rs group ilug URL URL for un subscription informat", + "ion list maintainer listmaster URL irish linux use", + "rs group ilug URL URL for un subscription informat", + "ion list maintainer listmaster URL\r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: it will fun", + "ction as a router\r\n\r\nif that is what you wish it eve", + "n looks like the modem s embedded os is some kind ", + "of linux being that it has interesting interfaces ", + "like ethNUMBER i don t use it as a router though i", + " just have it do the absolute minimum dsl stuff an", + "d do all the really fun stuff like pppoe on my lin", + "ux box also the manual tells you what the default ", + "password is don t forget to run pppoe over the alc", + "atel speedtouch NUMBERi as in my case you have to ", + "have a bridge configured in the router modem s sof", + "tware this lists your vci values etc also does any", + "one know if the high end speedtouch with NUMBER et", + "hernet ports can act as a full router or do i stil", + "l need to run a pppoe stack on the linux box regar", + "ds vin irish linux users group ilug URL URL for un", + " subscription information list maintainer listmast", + "er URL irish linux users group ilug URL URL for un", + " subscription information list maintainer listmast", + "er URL \r\n\r\n" + ), + concat!( + "Message-ID: \r\nSubject: all is it ", + "just me\r\n\r\nor has there been a massive increase in t", + "he amount of email being falsely bounced around th", + "e place i ve already received email from a number ", + "of people i don t know asking why i am sending the", + "m email these can be explained by servers from rus", + "sia and elsewhere coupled with the false emails i ", + "received myself it s really starting to annoy me a", + "m i the only one seeing an increase in recent week", + "s martin martin whelan déise design URL tel NUMBE", + "R NUMBER our core product déiseditor allows organ", + "isations to publish information to their web site ", + "in a fast and cost effective manner there is no ne", + "ed for a full time web developer as the site can b", + "e easily updated by the organisations own staff in", + "stant updates to keep site information fresh sites", + " which are updated regularly bring users back visi", + "t URL for a demonstration déiseditor managing you", + "r information ____________________________________", + "___________ iiu mailing list iiu URL URL ,0\r\n" + ), +]; + +const TEST: [&str; 3] = [ + concat!( + "Subject: save up to NUMBER on life insurance\r\n\r\nwhy ", + "spend more than you have to life quote savings ens", + "uring your family s financial security is very imp", + "ortant life quote savings makes buying life insura", + "nce simple and affordable we provide free access t", + "o the very best companies and the lowest rates lif", + "e quote savings is fast easy and saves you money l", + "et us help you get started with the best values in", + " the country on new coverage you can save hundreds", + " or even thousands of dollars by requesting a free", + " quote from lifequote savings our service will tak", + "e you less than NUMBER minutes to complete shop an", + "d compare save up to NUMBER on all types of life i", + "nsurance hyperlink click here for your free quote ", + "protecting your family is the best investment you ", + "ll ever make if you are in receipt of this email i", + "n error and or wish to be removed from our list hy", + "perlink please click here and type remove if you r", + "eside in any state which prohibits e mail solicita", + "tions for insurance please disregard this email\r\n" + ), + concat!( + "Subject: can someone explain\r\n\r\nwhat type of operati", + "ng system solaris is as ive never seen or used it ", + "i dont know wheather to get a server from sun or f", + "rom dell i would prefer a linux based server and s", + "un seems to be the one for that but im not sure if", + " solaris is a distro of linux or a completely diff", + "erent operating system can someone explain kiall m", + "ac innes irish linux users group ilug URL URL for ", + "un subscription information list maintainer listma", + "ster URL \r\n" + ), + concat!( + "Subject: classifier test\r\n\r\nthis is a novel text tha", + "t the bayes classifier has never seen before, it s", + "hould be classified as ham or non-ham\r\n" + ), +]; diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index b23462d6..3467391a 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -7,6 +7,7 @@ pub mod acl; pub mod append; pub mod basic; +pub mod bayes; pub mod body_structure; pub mod condstore; pub mod copy_move; @@ -108,7 +109,14 @@ total = 5 wait = "1ms" [spam-filter] -enable = false +enable = true + +[spam-filter.bayes.account] +enable = true + +[spam-filter.bayes.classify] +balance = "0.0" +learns = 10 [queue] path = "{TMP}" @@ -135,6 +143,12 @@ protocol = "smtp" enable = false allow-invalid-certs = true +[session.data] +spam-filter = "recipients[0] != 'popper@example.com'" + +[session.data.add-headers] +delivered-to = false + [session.extensions] future-release = [ { if = "!is_empty(authenticated_as)", then = "99999999d"}, { else = false } ] @@ -412,6 +426,14 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { &["popper@example.com"], ) .await; + store + .create_test_user( + "bayes@example.com", + "secret", + "Thomas Bayes", + &["bayes@example.com"], + ) + .await; store .create_test_group( "support@example.com", @@ -485,6 +507,9 @@ pub async fn imap_tests() { imap.assert_read(Type::Untagged, ResponseType::Bye).await; } + // Bayes training + bayes::test(&handle).await; + // Run ManageSieve tests managesieve::test().await; diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index fd384a42..2e9d1472 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -42,7 +42,7 @@ use smtp::{core::SmtpSessionManager, SpawnQueueManager}; use store::{ roaring::RoaringBitmap, - write::{key::DeserializeBigEndian, AnyKey, FtsQueueClass, ValueClass}, + write::{key::DeserializeBigEndian, AnyKey, TaskQueueClass, ValueClass}, IterateParams, Stores, ValueKey, SUBSPACE_PROPERTY, }; use tokio::sync::watch; @@ -455,7 +455,7 @@ pub async fn wait_for_index(server: &Server) { account_id: 0, collection: 0, document_id: 0, - class: ValueClass::FtsQueue(FtsQueueClass { + class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: 0, hash: BlobHash::default(), }), @@ -464,7 +464,7 @@ pub async fn wait_for_index(server: &Server) { account_id: u32::MAX, collection: u8::MAX, document_id: u32::MAX, - class: ValueClass::FtsQueue(FtsQueueClass { + class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { seq: u64::MAX, hash: BlobHash::default(), }), diff --git a/tests/src/jmap/thread_merge.rs b/tests/src/jmap/thread_merge.rs index 56322061..87c5faec 100644 --- a/tests/src/jmap/thread_merge.rs +++ b/tests/src/jmap/thread_merge.rs @@ -250,8 +250,9 @@ async fn test_multi_thread(params: &mut JMAPTest) { mailbox_ids: vec![mailbox_id], keywords: vec![], received_at: None, - source: IngestSource::Smtp, - encrypt: false, + source: IngestSource::Smtp { deliver_to: "" }, + spam_classify: false, + spam_train: false, session_id: 0, }) .await diff --git a/tests/src/store/import_export.rs b/tests/src/store/import_export.rs index 752e90cb..8b0d45f3 100644 --- a/tests/src/store/import_export.rs +++ b/tests/src/store/import_export.rs @@ -279,7 +279,7 @@ impl Snapshot { (SUBSPACE_BITMAP_TAG, false), (SUBSPACE_BITMAP_TEXT, false), (SUBSPACE_DIRECTORY, true), - (SUBSPACE_FTS_QUEUE, true), + (SUBSPACE_TASK_QUEUE, true), (SUBSPACE_INDEXES, false), (SUBSPACE_BLOB_RESERVE, true), (SUBSPACE_BLOB_LINK, true),