From c467ce07f1bade5f46b1ce66abc1dfc6804eac50 Mon Sep 17 00:00:00 2001 From: mdecimus <11444311+mdecimus@users.noreply.github.com> Date: Thu, 4 Dec 2025 17:46:38 +0100 Subject: [PATCH] Spam filter performance and accuracy improvements (part 5) --- crates/common/src/config/network.rs | 4 +- crates/common/src/config/spamfilter.rs | 5 +- crates/common/src/core.rs | 12 +- crates/directory/src/core/mod.rs | 4 +- crates/directory/src/core/principal.rs | 2 + crates/directory/src/lib.rs | 4 +- crates/email/src/message/delete.rs | 2 +- crates/email/src/message/ingest.rs | 44 +- crates/http/src/management/spam.rs | 114 ++-- crates/http/src/request.rs | 1 - crates/imap/src/op/copy_move.rs | 4 +- crates/imap/src/op/store.rs | 8 +- crates/jmap/src/email/set.rs | 79 ++- crates/nlp/src/classifier/feature.rs | 9 + crates/nlp/src/tokenizers/types.rs | 19 + crates/services/src/housekeeper/mod.rs | 12 +- crates/services/src/state_manager/push.rs | 4 +- crates/smtp/src/inbound/spam.rs | 1 + crates/smtp/src/queue/spool.rs | 9 +- crates/spam-filter/src/analysis/domain.rs | 75 ++- crates/spam-filter/src/analysis/init.rs | 29 +- crates/spam-filter/src/analysis/ip.rs | 5 + crates/spam-filter/src/analysis/mod.rs | 13 +- crates/spam-filter/src/analysis/score.rs | 9 +- crates/spam-filter/src/analysis/url.rs | 241 ++++----- crates/spam-filter/src/lib.rs | 7 + crates/spam-filter/src/modules/classifier.rs | 515 ++++++++++++++----- crates/store/src/write/blob.rs | 128 ++--- crates/trc/src/event/conv.rs | 6 + crates/trc/src/event/description.rs | 26 +- crates/trc/src/event/level.rs | 18 +- crates/trc/src/ipc/metrics.rs | 6 +- crates/trc/src/lib.rs | 12 +- crates/trc/src/serializers/binary.rs | 24 +- 34 files changed, 967 insertions(+), 484 deletions(-) diff --git a/crates/common/src/config/network.rs b/crates/common/src/config/network.rs index 9d251884..e77d6e77 100644 --- a/crates/common/src/config/network.rs +++ b/crates/common/src/config/network.rs @@ -363,14 +363,14 @@ impl ClusterRole { matches!(self, ClusterRole::Enabled | ClusterRole::Sharded { .. }) } - pub fn is_enabled_for_account(&self, account_id: u32) -> bool { + pub fn is_enabled_for_integer(&self, value: u32) -> bool { match self { ClusterRole::Enabled => true, ClusterRole::Disabled => false, ClusterRole::Sharded { shard_id, total_shards, - } => (account_id % total_shards) == *shard_id, + } => (value % total_shards) == *shard_id, } } diff --git a/crates/common/src/config/spamfilter.rs b/crates/common/src/config/spamfilter.rs index e12bdd34..8fca624e 100644 --- a/crates/common/src/config/spamfilter.rs +++ b/crates/common/src/config/spamfilter.rs @@ -84,7 +84,7 @@ pub struct ClassifierConfig { pub auto_learn_reply_ham: bool, pub auto_learn_card_is_ham: bool, pub hold_samples_for: u64, - pub train_frequency: Option, + pub train_frequency: Option, } #[derive(Debug, Clone)] @@ -486,7 +486,8 @@ impl ClassifierConfig { "spam-filter.classifier.training.frequency", "12h", ) - .unwrap_or(Some(Duration::from_secs(12 * 60 * 60))), + .unwrap_or(Some(Duration::from_secs(12 * 60 * 60))) + .map(|d| d.as_secs()), } .into() } diff --git a/crates/common/src/core.rs b/crates/common/src/core.rs index 9d1fc1c1..f9e0dbbf 100644 --- a/crates/common/src/core.rs +++ b/crates/common/src/core.rs @@ -36,7 +36,7 @@ use store::{ DirectoryClass, QueueClass, ValueClass, key::DeserializeBigEndian, now, }, }; -use trc::AddContext; +use trc::{AddContext, SpamEvent}; use types::{ blob::{BlobClass, BlobId}, blob_hash::BlobHash, @@ -1071,10 +1071,16 @@ impl Server { last_trained_at: model.last_trained_at, })); } else { - let todo = "log insufficient samples, keep existing model"; + trc::event!( + Spam(SpamEvent::ModelNotReady), + Details = vec![ + trc::Value::from(model.ham_count), + trc::Value::from(model.spam_count) + ], + ); } } else { - let todo = "log missing model, keep existing one"; + trc::event!(Spam(SpamEvent::ModelNotFound)); } } diff --git a/crates/directory/src/core/mod.rs b/crates/directory/src/core/mod.rs index 8723c940..905b003f 100644 --- a/crates/directory/src/core/mod.rs +++ b/crates/directory/src/core/mod.rs @@ -82,8 +82,8 @@ impl Permission { Permission::SpamFilterUpdate => "Modify spam filter settings", Permission::WebadminUpdate => "Modify web admin interface settings", Permission::LogsView => "Access system logs", - Permission::SpamFilterTrain_ => "", - Permission::SpamFilterClassify_ => "", + Permission::SpamFilterTrain => "Train the spam filter", + Permission::SpamFilterTest => "Test 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 e90cc728..dca7f9ec 100644 --- a/crates/directory/src/core/principal.rs +++ b/crates/directory/src/core/principal.rs @@ -1729,6 +1729,8 @@ impl Permission { | Permission::ApiKeyCreate | Permission::ApiKeyUpdate | Permission::ApiKeyDelete + | Permission::SpamFilterTrain + | Permission::SpamFilterTest ) || self.is_user_permission() } diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 283e9ff6..9dc75d19 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -187,7 +187,7 @@ pub enum Permission { SpamFilterUpdate, WebadminUpdate, LogsView, - SpamFilterTrain_, + SpamFilterTrain, Restart, TracingList, TracingGet, @@ -321,7 +321,7 @@ pub enum Permission { AiModelInteract, Troubleshoot, - SpamFilterClassify_, + SpamFilterTest, // WebDAV permissions DavSyncCollection, diff --git a/crates/email/src/message/delete.rs b/crates/email/src/message/delete.rs index 566f5f50..e4424fba 100644 --- a/crates/email/src/message/delete.rs +++ b/crates/email/src/message/delete.rs @@ -120,7 +120,7 @@ impl EmailDeletion for Server { .network .roles .purge_accounts - .is_enabled_for_account(*id) + .is_enabled_for_integer(*id) }) .collect(); diff --git a/crates/email/src/message/ingest.rs b/crates/email/src/message/ingest.rs index 7764b241..063d0f12 100644 --- a/crates/email/src/message/ingest.rs +++ b/crates/email/src/message/ingest.rs @@ -35,7 +35,7 @@ use store::{ TaskEpoch, TaskQueueClass, ValueClass, key::DeserializeBigEndian, now, }, }; -use trc::{AddContext, MessageIngestEvent}; +use trc::{AddContext, MessageIngestEvent, SpamEvent}; use types::{ blob::{BlobClass, BlobId}, blob_hash::BlobHash, @@ -107,6 +107,7 @@ pub trait EmailIngest: Sync + Send { account_id: u32, document_id: u32, is_spam: bool, + span_id: u64, ) -> impl Future> + Send; fn add_spam_sample( &self, @@ -114,6 +115,7 @@ pub trait EmailIngest: Sync + Send { hash: BlobHash, is_spam: bool, hold_sample: bool, + span_id: u64, ); } @@ -630,6 +632,7 @@ impl EmailIngest for Server { params.blob_hash.unwrap_or(&blob_hash).clone(), learn_spam, !is_encrypted, + params.session_id, ); } @@ -835,22 +838,30 @@ impl EmailIngest for Server { account_id: u32, document_id: u32, is_spam: bool, + span_id: u64, ) -> trc::Result<()> { - if let Some(archive) = self - .store() - .get_value::>(ValueKey::property( - account_id, - Collection::Email, - document_id, - EmailField::Metadata, - )) - .await - .caused_by(trc::location!())? + if self.core.spam.classifier.is_some() + && let Some(archive) = self + .store() + .get_value::>(ValueKey::property( + account_id, + Collection::Email, + document_id, + EmailField::Metadata, + )) + .await + .caused_by(trc::location!())? { let metadata = archive .to_unarchived::() .caused_by(trc::location!())?; - self.add_spam_sample(batch, (&metadata.inner.blob_hash).into(), is_spam, true); + self.add_spam_sample( + batch, + (&metadata.inner.blob_hash).into(), + is_spam, + true, + span_id, + ); } Ok(()) @@ -862,6 +873,7 @@ impl EmailIngest for Server { hash: BlobHash, is_spam: bool, hold_sample: bool, + span_id: u64, ) { if let Some(config) = &self.core.spam.classifier { let mut dt = DateTime::from_timestamp(now() as i64); @@ -882,6 +894,14 @@ impl EmailIngest for Server { BlobOp::SpamSample { hash, until }, vec![u8::from(is_spam), u8::from(hold_sample)], ); + + trc::event!( + Spam(SpamEvent::TrainSampleAdded), + AccountId = batch.last_account_id(), + Details = if is_spam { "spam" } else { "ham" }, + Expires = trc::Value::Timestamp(until), + SpanId = span_id, + ); } } } diff --git a/crates/http/src/management/spam.rs b/crates/http/src/management/spam.rs index 0e2fb3f1..32963312 100644 --- a/crates/http/src/management/spam.rs +++ b/crates/http/src/management/spam.rs @@ -4,20 +4,18 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::net::IpAddr; - use common::{Server, auth::AccessToken, config::spamfilter::SpamFilterAction, psl}; - -use compact_str::CompactString; use directory::{ Permission, backend::internal::manage::{self, ManageDirectory}, }; +use email::message::ingest::EmailIngest; +use http_proto::{request::decode_path_element, *}; use hyper::Method; use mail_auth::{ AuthenticatedMessage, DmarcResult, dmarc::verify::DmarcParameters, spf::verify::SpfParameters, }; -use mail_parser::{Message, MessageParser}; +use mail_parser::MessageParser; use serde::{Deserialize, Serialize}; use serde_json::json; use spam_filter::{ @@ -25,9 +23,8 @@ use spam_filter::{ analysis::{init::SpamFilterInit, score::SpamFilterAnalyzeScore}, }; use std::future::Future; -use store::ahash::AHashMap; - -use http_proto::{request::decode_path_element, *}; +use std::net::IpAddr; +use store::{ahash::AHashMap, write::BatchBuilder}; pub trait ManageSpamHandler: Sync + Send { fn handle_manage_spam( @@ -65,8 +62,8 @@ pub struct SpamClassifyRequest { #[derive(Debug, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct SpamClassifyResponse { - pub score: f64, - pub tags: AHashMap>, + pub score: f32, + pub tags: AHashMap>, pub disposition: SpamFilterDisposition, } @@ -88,25 +85,63 @@ impl ManageSpamHandler for Server { session: &HttpSessionData, access_token: &AccessToken, ) -> trc::Result { - // Validate the access token - //access_token.assert_has_permission(Permission::SpamFilterTrain)?; - match (path.get(1).copied(), path.get(2).copied(), req.method()) { - (Some("train"), Some(class @ ("ham" | "spam")), &Method::POST) => { - let todo = "fix"; - /*let message = parse_message_or_err(body.as_deref().unwrap_or_default())?; - let input = if let Some(account) = path.get(3).copied().filter(|a| !a.is_empty()) { - let account_id = self + (Some("sample"), Some(class @ ("ham" | "spam")), &Method::POST) => { + // Validate the access token + access_token.assert_has_permission(Permission::SpamFilterTrain)?; + + let message = + body.ok_or_else(|| manage::error("Failed to parse message.", None::))?; + let account_id = if let Some(account) = + path.get(3).copied().filter(|a| !a.is_empty()) + { + let principal = self .store() - .get_principal_id(decode_path_element(account).as_ref()) + .get_principal_info(decode_path_element(account).as_ref()) .await? .ok_or_else(|| manage::not_found(account.to_string()))?; - SpamFilterInput::from_account_message(&message, account_id, session.session_id) + if access_token.tenant.is_some() && principal.tenant != access_token.tenant_id() + { + return Err(manage::error( + "Account does not belong to this tenant.", + None::, + )); + } + + principal.id + } else if access_token.tenant.is_none() { + u32::MAX } else { - SpamFilterInput::from_message(&message, session.session_id) + return Err(manage::error( + "Account ID is required for tenants.", + None::, + )); }; - self.bayes_train(&self.spam_filter_init(input), class == "spam", true) - .await?; */ + + // Write sample + let (blob_hash, blob_hold) = + self.put_temporary_blob(account_id, &message, 60).await?; + let mut batch = BatchBuilder::new(); + batch.with_account_id(account_id).clear(blob_hold); + self.add_spam_sample( + &mut batch, + blob_hash, + class == "spam", + true, + session.session_id, + ); + self.store().write(batch.build_all()).await?; + + Ok(JsonResponse::new(json!({ + "data": (), + })) + .into_http_response()) + } + (Some("train"), _, &Method::GET) => { + // Validate the access token + access_token.assert_has_permission(Permission::SpamFilterTrain)?; + + let todo = "implement"; Ok(JsonResponse::new(json!({ "data": (), @@ -114,6 +149,9 @@ impl ManageSpamHandler for Server { .into_http_response()) } (Some("classify"), _, &Method::POST) => { + // Validate the access token + access_token.assert_has_permission(Permission::SpamFilterTest)?; + // Parse request let request = serde_json::from_slice::( body.as_deref().unwrap_or_default(), @@ -123,7 +161,10 @@ impl ManageSpamHandler for Server { })?; // Built spam filter input - let message = parse_message_or_err(request.message.as_bytes())?; + let message = MessageParser::new() + .parse(request.message.as_bytes()) + .filter(|m| m.root_part().headers().iter().any(|h| !h.name.is_other())) + .ok_or_else(|| manage::error("Failed to parse message.", None::))?; let remote_ip = request.remote_ip; let ehlo_domain = request.ehlo_domain.to_lowercase(); @@ -243,21 +284,26 @@ impl ManageSpamHandler for Server { env_from_flags: request.env_from_flags, env_rcpt_to: request.env_rcpt_to.iter().map(String::as_str).collect(), is_test: true, + is_train: false, }; // Classify let mut ctx = self.spam_filter_init(input); let result = self.spam_filter_classify(&mut ctx).await; - let todo = "fix"; // Build response - /* let mut response = SpamClassifyResponse { + let mut response = SpamClassifyResponse { score: ctx.result.score, tags: AHashMap::with_capacity(ctx.result.tags.len()), disposition: match result { - SpamFilterAction::Allow(value) => SpamFilterDisposition::Allow { value }, + SpamFilterAction::Allow(value) => SpamFilterDisposition::Allow { + value: value.headers, + }, SpamFilterAction::Discard => SpamFilterDisposition::Discard, SpamFilterAction::Reject => SpamFilterDisposition::Reject, + SpamFilterAction::Disabled => SpamFilterDisposition::Allow { + value: String::new(), + }, }, }; for tag in ctx.result.tags { @@ -267,7 +313,9 @@ impl ManageSpamHandler for Server { } Some(SpamFilterAction::Discard) => SpamFilterDisposition::Discard, Some(SpamFilterAction::Reject) => SpamFilterDisposition::Reject, - None => SpamFilterDisposition::Allow { value: 0.0 }, + Some(SpamFilterAction::Disabled) | None => { + SpamFilterDisposition::Allow { value: 0.0 } + } }; response.tags.insert(tag, disposition); } @@ -275,17 +323,9 @@ impl ManageSpamHandler for Server { Ok(JsonResponse::new(json!({ "data": response, })) - .into_http_response())*/ - todo!() + .into_http_response()) } _ => Err(trc::ResourceEvent::NotFound.into_err()), } } } - -fn parse_message_or_err(bytes: &[u8]) -> trc::Result> { - MessageParser::new() - .parse(bytes) - .filter(|m| m.root_part().headers().iter().any(|h| !h.name.is_other())) - .ok_or_else(|| manage::error("Failed to parse message.", None::)) -} diff --git a/crates/http/src/request.rs b/crates/http/src/request.rs index 1dc9f555..4006f3cb 100644 --- a/crates/http/src/request.rs +++ b/crates/http/src/request.rs @@ -85,7 +85,6 @@ impl ParseHttp for Server { } } - let todo = "hashify"; match path.next().unwrap_or_default() { "jmap" => { match (path.next().unwrap_or_default(), req.method()) { diff --git a/crates/imap/src/op/copy_move.rs b/crates/imap/src/op/copy_move.rs index 67aabd1a..1a49b2dd 100644 --- a/crates/imap/src/op/copy_move.rs +++ b/crates/imap/src/op/copy_move.rs @@ -322,14 +322,14 @@ impl SessionData { // Add message to training queue if dest_mailbox_id.mailbox_id == JUNK_ID { self.server - .add_account_spam_sample(&mut batch, account_id, id, true) + .add_account_spam_sample(&mut batch, account_id, id, true, self.session_id) .await .imap_ctx(&arguments.tag, trc::location!())?; } else if src_mailbox.id.mailbox_id == JUNK_ID && dest_mailbox_id.mailbox_id != TRASH_ID { self.server - .add_account_spam_sample(&mut batch, account_id, id, false) + .add_account_spam_sample(&mut batch, account_id, id, false, self.session_id) .await .imap_ctx(&arguments.tag, trc::location!())?; } diff --git a/crates/imap/src/op/store.rs b/crates/imap/src/op/store.rs index 96f89e07..472b9abd 100644 --- a/crates/imap/src/op/store.rs +++ b/crates/imap/src/op/store.rs @@ -298,7 +298,13 @@ impl SessionData { // Add spam train task if let Some(learn_spam) = train_spam { self.server - .add_account_spam_sample(&mut batch, account_id, *id, learn_spam) + .add_account_spam_sample( + &mut batch, + account_id, + *id, + learn_spam, + self.session_id, + ) .await .imap_ctx(response.tag.as_ref().unwrap(), trc::location!())?; } diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index 5c4cbfe0..032c55ee 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -15,7 +15,7 @@ use common::{ }; use email::{ cache::{MessageCacheFetch, email::MessageCacheAccess, mailbox::MailboxCacheAccess}, - mailbox::UidMailbox, + mailbox::{JUNK_ID, TRASH_ID, UidMailbox}, message::{ delete::EmailDeletion, ingest::{EmailIngest, IngestEmail, IngestSource}, @@ -43,13 +43,18 @@ use mail_builder::{ use mail_parser::MessageParser; use std::future::Future; use std::{borrow::Cow, collections::HashMap}; -use store::{ValueKey, ahash::AHashMap, roaring::RoaringBitmap, write::{AlignedBytes, Archive, BatchBuilder}}; +use store::{ + ValueKey, + ahash::AHashMap, + roaring::RoaringBitmap, + write::{AlignedBytes, Archive, BatchBuilder}, +}; use trc::AddContext; use types::{ acl::Acl, collection::{Collection, SyncCollection, VanishedCollection}, id::Id, - keyword::Keyword, + keyword::{ArchivedKeyword, Keyword}, type_state::{DataType, StateChange}, }; @@ -878,7 +883,7 @@ impl EmailSet for Server { } // Process keywords - let todo = "train spam classifier"; + let mut train_spam = None; if has_keyword_changes { // Verify permissions on shared accounts if can_modify_mailbox_ids.as_ref().is_some_and(|ids| { @@ -895,14 +900,36 @@ impl EmailSet for Server { continue 'update; } + // Process keyword changes + let mut changed_seen = false; + for keyword in new_data.added_keywords(data.inner) { + match keyword { + Keyword::Seen => { + changed_seen = true; + } + Keyword::Junk => { + train_spam = Some(true); + } + Keyword::NotJunk => { + train_spam = Some(false); + } + _ => {} + } + } + for keyword in new_data.removed_keywords(data.inner) { + match keyword { + ArchivedKeyword::Seen => { + changed_seen = true; + } + ArchivedKeyword::Junk if train_spam.is_none() => { + train_spam = Some(false); + } + _ => {} + } + } + // Set all current mailboxes as changed if the Seen tag changed - if new_data - .added_keywords(data.inner) - .any(|keyword| keyword == &Keyword::Seen) - || new_data - .removed_keywords(data.inner) - .any(|keyword| keyword == &Keyword::Seen) - { + if changed_seen { for mailbox_id in new_data.mailboxes.iter() { changed_mailboxes.insert(mailbox_id.mailbox_id, Vec::new()); } @@ -930,6 +957,10 @@ impl EmailSet for Server { .as_ref() .is_none_or(|ids| ids.contains(mailbox_id.mailbox_id)) { + if mailbox_id.mailbox_id == JUNK_ID { + train_spam = Some(true); + } + changed_mailboxes.insert(mailbox_id.mailbox_id, Vec::new()); } else { response.not_updated.append( @@ -962,6 +993,15 @@ impl EmailSet for Server { .as_ref() .is_none_or(|ids| ids.contains(u32::from(mailbox_id.mailbox_id))) { + if mailbox_id.mailbox_id == JUNK_ID + && !new_data + .mailboxes + .iter() + .any(|mb| mb.mailbox_id == TRASH_ID) + { + train_spam = Some(false); + } + changed_mailboxes .entry(mailbox_id.mailbox_id.to_native()) .or_default() @@ -1011,8 +1051,21 @@ impl EmailSet for Server { .with_current(data) .with_changes(new_data.seal()), ) - .caused_by(trc::location!())? - .commit_point(); + .caused_by(trc::location!())?; + + if let Some(train_spam) = train_spam { + self.add_account_spam_sample( + &mut batch, + account_id, + document_id, + train_spam, + session.session_id, + ) + .await + .caused_by(trc::location!())?; + } + + batch.commit_point(); will_update.push(id); } diff --git a/crates/nlp/src/classifier/feature.rs b/crates/nlp/src/classifier/feature.rs index 345a634d..0af8e93d 100644 --- a/crates/nlp/src/classifier/feature.rs +++ b/crates/nlp/src/classifier/feature.rs @@ -78,6 +78,15 @@ impl FeatureBuilder { } } +impl Sample { + pub fn new(features: Features, class: bool) -> Self { + Self { + features, + class: if class { 1.0 } else { 0.0 }, + } + } +} + impl AsRef for Sample { fn as_ref(&self) -> &Sample { self diff --git a/crates/nlp/src/tokenizers/types.rs b/crates/nlp/src/tokenizers/types.rs index 96b1b1d8..7818d3b6 100644 --- a/crates/nlp/src/tokenizers/types.rs +++ b/crates/nlp/src/tokenizers/types.rs @@ -2837,6 +2837,25 @@ mod test { TokenType::Punctuation('!'), ], ), + ( + "vοΌ₯ⓑ𝔂 π”½π•ŒΕ‡β„•ο½™ ţ乇𝕏𝓣 wWiIiIIttHh l133t5p3/-\\|<", + vec![ + TokenType::Alphabetic("vοΌ₯ⓑ𝔂"), + TokenType::Space, + TokenType::Alphabetic("π”½π•ŒΕ‡β„•ο½™"), + TokenType::Space, + TokenType::Alphabetic("ţ乇𝕏𝓣"), + TokenType::Space, + TokenType::Alphabetic("wWiIiIIttHh"), + TokenType::Space, + TokenType::Alphanumeric("l133t5p3"), + TokenType::Punctuation('/'), + TokenType::Punctuation('-'), + TokenType::Punctuation('\\'), + TokenType::Punctuation('|'), + TokenType::Punctuation('<'), + ], + ), ] { let result = TypesTokenizer::new(text) .map(|t| t.word) diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index 2daf12b5..d8105f47 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -109,9 +109,15 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver 0 { + now().saturating_sub(last_trained_at).min(train_frequency) + } else { + train_frequency + }; + queue.schedule( - Instant::now() + train_frequency, + Instant::now() + Duration::from_secs(next_train), ActionClass::TrainSpamClassifier, ); } @@ -577,7 +583,7 @@ pub fn spawn_housekeeper(inner: Arc, mut rx: mpsc::Receiver) -> mpsc::Sender { .network .roles .push_notifications - .is_enabled_for_account(account_id) + .is_enabled_for_integer(account_id) { // Load push subscriptions for account let (subscriptions, member_account_ids) = @@ -147,7 +147,7 @@ pub fn spawn_push_manager(inner: Arc) -> mpsc::Sender { .network .roles .push_notifications - .is_enabled_for_account(account_id) + .is_enabled_for_integer(account_id) { continue; } diff --git a/crates/smtp/src/inbound/spam.rs b/crates/smtp/src/inbound/spam.rs index bcd1f0c3..bb0eb45d 100644 --- a/crates/smtp/src/inbound/spam.rs +++ b/crates/smtp/src/inbound/spam.rs @@ -87,6 +87,7 @@ impl Session { .map(|r| r.address_lcase.as_str()) .collect(), is_test: false, + is_train: false, } } } diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 5af99828..0658c19b 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -28,7 +28,7 @@ use store::write::{ QueueClass, ValueClass, now, }; use store::{Deserialize, IterateParams, Serialize, U64_LEN, ValueKey}; -use trc::{AddContext, ServerEvent}; +use trc::{AddContext, ServerEvent, SpamEvent}; use types::blob_hash::BlobHash; use utils::DomainPart; @@ -457,6 +457,13 @@ impl MessageWrapper { }, vec![1, 1], ); + + trc::event!( + Spam(SpamEvent::TrainSampleAdded), + Details = "spam", + Expires = trc::Value::Timestamp(hold_period), + SpanId = self.span_id, + ); } batch diff --git a/crates/spam-filter/src/analysis/domain.rs b/crates/spam-filter/src/analysis/domain.rs index 99c87694..557ef9ab 100644 --- a/crates/spam-filter/src/analysis/domain.rs +++ b/crates/spam-filter/src/analysis/domain.rs @@ -57,7 +57,6 @@ impl SpamFilterAnalyzeDomain for Server { { if let Host::Name(name) = host && let Some(name) = Hostname::new(name.as_ref()).sld - && !is_trusted_domain(self, &name, ctx.input.span_id).await { domains.insert(ElementLocation::new(name, Location::HeaderReceived)); } @@ -71,8 +70,6 @@ impl SpamFilterAnalyzeDomain for Server { let host = Hostname::new(d); if host.sld.is_some() { Some(host) } else { None } }) - && !is_trusted_domain(self, mid_domain.sld_or_default(), ctx.input.span_id) - .await { domains.insert(ElementLocation::new(mid_domain.fqdn, Location::HeaderMid)); } @@ -188,7 +185,7 @@ impl SpamFilterAnalyzeDomain for Server { for token in tokens { if let TokenType::Email(email) = token { - if is_body && !ctx.result.has_tag("RCPT_IN_BODY") { + if !ctx.input.is_train && is_body && !ctx.result.has_tag("RCPT_IN_BODY") { for rcpt in ctx.output.all_recipients() { if rcpt.email.address == email.address { ctx.result.add_tag("RCPT_IN_BODY"); @@ -214,42 +211,44 @@ impl SpamFilterAnalyzeDomain for Server { } } - // Validate email - for email in &emails { - // Skip trusted domains - if !email.element.email.is_valid() - || is_trusted_domain( - self, - &email.element.email.domain_part.fqdn, - ctx.input.span_id, - ) - .await - { - continue; + if !ctx.input.is_train { + // Validate email + for email in &emails { + // Skip trusted domains + if !email.element.email.is_valid() + || is_trusted_domain( + self, + &email.element.email.domain_part.fqdn, + ctx.input.span_id, + ) + .await + { + continue; + } + + // Check Email DNSBL + check_dnsbl(self, ctx, &email.element, Element::Email, email.location).await; + + domains.insert(ElementLocation::new( + email.element.email.domain_part.fqdn.clone(), + email.location, + )); } - // Check Email DNSBL - check_dnsbl(self, ctx, &email.element, Element::Email, email.location).await; - - domains.insert(ElementLocation::new( - email.element.email.domain_part.fqdn.clone(), - email.location, - )); - } - - // Validate domains - for domain in &domains { - // Skip trusted domains - if !is_trusted_domain(self, &domain.element, ctx.input.span_id).await { - // Check Domain DNSBL - check_dnsbl( - self, - ctx, - &StringResolver(domain.element.as_str()), - Element::Domain, - domain.location, - ) - .await; + // Validate domains + for domain in &domains { + // Skip trusted domains + if !is_trusted_domain(self, &domain.element, ctx.input.span_id).await { + // Check Domain DNSBL + check_dnsbl( + self, + ctx, + &StringResolver(domain.element.as_str()), + Element::Domain, + domain.location, + ) + .await; + } } } ctx.output.emails = emails; diff --git a/crates/spam-filter/src/analysis/init.rs b/crates/spam-filter/src/analysis/init.rs index 0f6d792e..5378891a 100644 --- a/crates/spam-filter/src/analysis/init.rs +++ b/crates/spam-filter/src/analysis/init.rs @@ -6,6 +6,7 @@ use common::Server; +use mail_auth::DmarcResult; use mail_parser::{HeaderName, PartType, parsers::fields::thread::thread_name}; use nlp::tokenizers::types::{TokenType, TypesTokenizer}; @@ -24,13 +25,14 @@ pub trait SpamFilterInit { const POSTMASTER_ADDRESSES: [&str; 3] = ["postmaster", "mailer-daemon", "root"]; impl SpamFilterInit for Server { - fn spam_filter_init<'x>(&self, input: SpamFilterInput<'x>) -> SpamFilterContext<'x> { + fn spam_filter_init<'x>(&self, mut input: SpamFilterInput<'x>) -> SpamFilterContext<'x> { let mut subject = ""; let mut from = None; let mut reply_to = None; let mut recipients_to = Vec::new(); let mut recipients_cc = Vec::new(); let mut recipients_bcc = Vec::new(); + let mut found_spam_status = false; for header in input.message.headers() { match &header.name { @@ -83,6 +85,31 @@ impl SpamFilterInit for Server { HeaderName::From => { from = header.value().as_address().and_then(|addrs| addrs.first()); } + HeaderName::Other(name) + if input.is_train && !found_spam_status && name.eq("X-Spam-Status") => + { + for token in header + .value() + .as_text() + .unwrap_or_default() + .split_ascii_whitespace() + { + if let Some(dmarc) = token.strip_prefix("DMARC_") { + input.dmarc_result = if dmarc == "POLICY_ALLOW" { + Some(&DmarcResult::Pass) + } else { + Some(&DmarcResult::None) + }; + } else if let Some(asn) = token + .strip_prefix("SOURCE_ASN_") + .and_then(|v| v.parse().ok()) + { + input.asn = Some(asn); + } + } + + found_spam_status = true; + } _ => {} } } diff --git a/crates/spam-filter/src/analysis/ip.rs b/crates/spam-filter/src/analysis/ip.rs index 17b60416..3280726c 100644 --- a/crates/spam-filter/src/analysis/ip.rs +++ b/crates/spam-filter/src/analysis/ip.rs @@ -116,6 +116,11 @@ impl SpamFilterAnalyzeIp for Server { IprevResult::Pass | IprevResult::None => (), } } + + // Add ASN + if let Some(asn_id) = &ctx.input.asn { + ctx.result.add_tag(format!("SOURCE_ASN_{asn_id}")); + } } } diff --git a/crates/spam-filter/src/analysis/mod.rs b/crates/spam-filter/src/analysis/mod.rs index 55fe41f2..6f2ce58d 100644 --- a/crates/spam-filter/src/analysis/mod.rs +++ b/crates/spam-filter/src/analysis/mod.rs @@ -4,19 +4,16 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::{ + Recipient, SpamFilterContext, SpamFilterInput, SpamFilterOutput, SpamFilterResult, TextPart, +}; +use common::{Server, config::spamfilter::Location}; +use mail_parser::{Header, parsers::MessageStream}; use std::{ borrow::Cow, hash::{Hash, Hasher}, }; -use common::{Server, config::spamfilter::Location}; - -use mail_parser::{Header, parsers::MessageStream}; - -use crate::{ - Recipient, SpamFilterContext, SpamFilterInput, SpamFilterOutput, SpamFilterResult, TextPart, -}; - pub mod classifier; pub mod date; pub mod dmarc; diff --git a/crates/spam-filter/src/analysis/score.rs b/crates/spam-filter/src/analysis/score.rs index f3883b96..8af9d549 100644 --- a/crates/spam-filter/src/analysis/score.rs +++ b/crates/spam-filter/src/analysis/score.rs @@ -45,6 +45,7 @@ pub struct SpamFilterScore { pub results: Vec, pub headers: String, pub spam_trap: bool, + pub score: f32, } impl SpamFilterAnalyzeScore for Server { @@ -76,11 +77,14 @@ impl SpamFilterAnalyzeScore for Server { let mut final_score = ctx.result.score; let mut avg_confidence: f32 = 0.0; + let mut total_results = 0; let mut user_results = vec![false; ctx.result.classifier_confidence.len()]; if !ctx.result.classifier_confidence.is_empty() { for (idx, &confidence) in ctx.result.classifier_confidence.iter().enumerate() { if let Some(confidence) = confidence { avg_confidence += confidence; + total_results += 1; + let user_score = self .core .spam @@ -97,7 +101,9 @@ impl SpamFilterAnalyzeScore for Server { } } - avg_confidence /= ctx.result.classifier_confidence.len() as f32; + if total_results > 0 { + avg_confidence /= total_results as f32; + } if avg_confidence != 0.0 { let tag = avg_confidence.spam_tag(); @@ -161,6 +167,7 @@ impl SpamFilterAnalyzeScore for Server { results: user_results, headers, spam_trap: ctx.result.spam_trap, + score: final_score, }) } } diff --git a/crates/spam-filter/src/analysis/url.rs b/crates/spam-filter/src/analysis/url.rs index 30472800..b227dbda 100644 --- a/crates/spam-filter/src/analysis/url.rs +++ b/crates/spam-filter/src/analysis/url.rs @@ -4,18 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::collections::HashSet; -use std::hash::{Hash, Hasher}; -use std::{borrow::Cow, future::Future, time::Duration}; - -use common::Server; -use common::config::spamfilter::{Element, IpResolver, Location}; -use common::scripts::IsMixedCharset; -use common::scripts::functions::unicode::CharUtils; -use hyper::{Uri, header::LOCATION}; -use nlp::tokenizers::types::TokenType; -use reqwest::redirect::Policy; - +use super::{ElementLocation, is_trusted_domain, is_url_redirector}; use crate::modules::dnsbl::check_dnsbl; use crate::modules::expression::StringResolver; use crate::modules::html::SRC; @@ -23,8 +12,16 @@ use crate::{ Hostname, SpamFilterContext, TextPart, modules::html::{A, HREF, HtmlToken}, }; - -use super::{ElementLocation, is_trusted_domain, is_url_redirector}; +use common::Server; +use common::config::spamfilter::{Element, IpResolver, Location}; +use common::scripts::IsMixedCharset; +use common::scripts::functions::unicode::CharUtils; +use hyper::{Uri, header::LOCATION}; +use nlp::tokenizers::types::TokenType; +use reqwest::redirect::Policy; +use std::collections::HashSet; +use std::hash::{Hash, Hasher}; +use std::{borrow::Cow, future::Future, time::Duration}; pub trait SpamFilterAnalyzeUrl: Sync + Send { fn spam_filter_analyze_url( @@ -96,7 +93,8 @@ impl SpamFilterAnalyzeUrl for Server { for token in tokens { match token { TokenType::Url(url) | TokenType::UrlNoScheme(url) => { - if is_body + if !ctx.input.is_train + && is_body && !ctx.result.has_tag("RCPT_DOMAIN_IN_BODY") && let Some(url_parsed) = &url.url_parsed { @@ -122,7 +120,7 @@ impl SpamFilterAnalyzeUrl for Server { } } - if is_body { + if is_body && !ctx.input.is_train { let is_single = match part { TextPart::Plain { tokens, .. } => is_single_url(tokens), TextPart::Html { @@ -139,136 +137,141 @@ impl SpamFilterAnalyzeUrl for Server { } } - let mut redirected_urls = HashSet::new(); - for url in &urls { - for ch in url.element.url.chars() { - if ch.is_zwsp() { - ctx.result.add_tag("ZERO_WIDTH_SPACE_URL"); + if !ctx.input.is_train { + let mut redirected_urls = HashSet::new(); + for url in &urls { + for ch in url.element.url.chars() { + if ch.is_zwsp() { + ctx.result.add_tag("ZERO_WIDTH_SPACE_URL"); + } + + if ch.is_obscured() { + ctx.result.add_tag("SUSPICIOUS_URL"); + } } - if ch.is_obscured() { - ctx.result.add_tag("SUSPICIOUS_URL"); + // Skip non-URLs such as 'data:' and 'mailto:' + if !url.element.url.contains("://") { + continue; } - } - // Skip non-URLs such as 'data:' and 'mailto:' - if !url.element.url.contains("://") { - continue; - } + // Obtain parse url + let url_parsed = if let Some(url_parsed) = &url.element.url_parsed { + url_parsed + } else { + // URL could not be parsed + ctx.result.add_tag("UNPARSABLE_URL"); + continue; + }; + let host_sld = url_parsed.host.sld_or_default(); - // Obtain parse url - let url_parsed = if let Some(url_parsed) = &url.element.url_parsed { - url_parsed - } else { - // URL could not be parsed - ctx.result.add_tag("UNPARSABLE_URL"); - continue; - }; - let host_sld = url_parsed.host.sld_or_default(); + // Skip local and trusted domains + if is_trusted_domain(self, host_sld, ctx.input.span_id).await { + continue; + } - // Skip local and trusted domains - if is_trusted_domain(self, host_sld, ctx.input.span_id).await { - continue; - } + if let Some(ip) = url_parsed.host.ip { + // Check IP DNSBL + check_dnsbl(self, ctx, &IpResolver::new(ip), Element::Ip, url.location).await; + } else if is_url_redirector(self, host_sld, ctx.input.span_id).await { + // Check for redirectors + ctx.result.add_tag("REDIRECTOR_URL"); - if let Some(ip) = url_parsed.host.ip { - // Check IP DNSBL - check_dnsbl(self, ctx, &IpResolver::new(ip), Element::Ip, url.location).await; - } else if is_url_redirector(self, host_sld, ctx.input.span_id).await { - // Check for redirectors - ctx.result.add_tag("REDIRECTOR_URL"); + if !ctx.result.has_tag("URL_REDIRECTOR_NESTED") { + let mut redirect_count = 1; + let mut url_redirect = Cow::Borrowed(url.element.url.as_str()); - if !ctx.result.has_tag("URL_REDIRECTOR_NESTED") { - let mut redirect_count = 1; - let mut url_redirect = Cow::Borrowed(url.element.url.as_str()); - - while redirect_count <= 3 { - match http_get_header( - url_redirect.as_ref(), - LOCATION, - Duration::from_secs(5), - ) - .await - { - Ok(Some(location)) => { - let location = UrlParts::new(location); - if let Some(location_parsed) = &location.url_parsed { - if is_url_redirector( - self, - location_parsed.host.sld_or_default(), - ctx.input.span_id, - ) - .await - { - url_redirect = Cow::Owned(location.url); - redirect_count += 1; - continue; - } else { - redirected_urls - .insert(ElementLocation::new(location, url.location)); + while redirect_count <= 3 { + match http_get_header( + url_redirect.as_ref(), + LOCATION, + Duration::from_secs(5), + ) + .await + { + Ok(Some(location)) => { + let location = UrlParts::new(location); + if let Some(location_parsed) = &location.url_parsed { + if is_url_redirector( + self, + location_parsed.host.sld_or_default(), + ctx.input.span_id, + ) + .await + { + url_redirect = Cow::Owned(location.url); + redirect_count += 1; + continue; + } else { + redirected_urls.insert(ElementLocation::new( + location, + url.location, + )); + } } } + Ok(None) => {} + Err(err) => { + trc::error!(err.span_id(ctx.input.span_id)); + } } - Ok(None) => {} - Err(err) => { - trc::error!(err.span_id(ctx.input.span_id)); - } + break; } - break; - } - if redirect_count > 3 { - ctx.result.add_tag("URL_REDIRECTOR_NESTED"); + if redirect_count > 3 { + ctx.result.add_tag("URL_REDIRECTOR_NESTED"); + } } } } - } - urls.extend(redirected_urls); + urls.extend(redirected_urls); - for (el, url_parsed) in urls.iter().filter_map(|el| { - el.element - .url_parsed - .as_ref() - .map(|url_parsed| (el, url_parsed)) - }) { - let host = &url_parsed.host; + for (el, url_parsed) in urls.iter().filter_map(|el| { + el.element + .url_parsed + .as_ref() + .map(|url_parsed| (el, url_parsed)) + }) { + let host = &url_parsed.host; - if host.ip.is_none() { - if !host.fqdn.is_ascii() { - if let Ok(cured_host) = decancer::cure(&host.fqdn, decancer::Options::default()) - { - let cured_host = cured_host.to_string(); - if cured_host != host.fqdn - && matches!(self.dns_exists_ip(&cured_host).await, Ok(true)) + if host.ip.is_none() { + if !host.fqdn.is_ascii() { + if let Ok(cured_host) = + decancer::cure(&host.fqdn, decancer::Options::default()) { - ctx.result.add_tag("HOMOGRAPH_URL"); + let cured_host = cured_host.to_string(); + if cured_host != host.fqdn + && matches!(self.dns_exists_ip(&cured_host).await, Ok(true)) + { + ctx.result.add_tag("HOMOGRAPH_URL"); + } + } + + if host.fqdn.is_mixed_charset() { + ctx.result.add_tag("MIXED_CHARSET_URL"); } } - if host.fqdn.is_mixed_charset() { - ctx.result.add_tag("MIXED_CHARSET_URL"); + // Check Domain DNSBL + if let Some(sld) = &host.sld { + check_dnsbl( + self, + ctx, + &StringResolver(sld), + Element::Domain, + el.location, + ) + .await; } + } else { + // URL is an ip address + ctx.result.add_tag("SUSPICIOUS_URL"); } - // Check Domain DNSBL - if let Some(sld) = &host.sld { - check_dnsbl( - self, - ctx, - &StringResolver(sld), - Element::Domain, - el.location, - ) - .await; - } - } else { - // URL is an ip address - ctx.result.add_tag("SUSPICIOUS_URL"); + // Check URL DNSBL + check_dnsbl(self, ctx, &el.element, Element::Url, el.location).await; } - - // Check URL DNSBL - check_dnsbl(self, ctx, &el.element, Element::Url, el.location).await; } // Update context diff --git a/crates/spam-filter/src/lib.rs b/crates/spam-filter/src/lib.rs index 5531b7ec..ac1ae253 100644 --- a/crates/spam-filter/src/lib.rs +++ b/crates/spam-filter/src/lib.rs @@ -47,6 +47,7 @@ pub struct SpamFilterInput<'x> { pub env_from_flags: u64, pub env_rcpt_to: Vec<&'x str>, + pub is_train: bool, pub is_test: bool, } @@ -156,8 +157,14 @@ impl<'x> SpamFilterInput<'x> { env_from_flags: 0, env_rcpt_to: vec![], is_test: false, + is_train: false, } } + + pub fn train_mode(mut self) -> Self { + self.is_train = true; + self + } } impl PartialEq for Hostname { diff --git a/crates/spam-filter/src/modules/classifier.rs b/crates/spam-filter/src/modules/classifier.rs index 8bbb6a6d..1d60deb8 100644 --- a/crates/spam-filter/src/modules/classifier.rs +++ b/crates/spam-filter/src/modules/classifier.rs @@ -4,11 +4,17 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::analysis::domain::SpamFilterAnalyzeDomain; +use crate::analysis::init::SpamFilterInit; +use crate::analysis::is_trusted_domain; +use crate::analysis::url::SpamFilterAnalyzeUrl; use crate::{Email, IpParts, SpamFilterContext, TextPart, analysis::url::UrlParts}; +use crate::{Hostname, SpamFilterInput}; use common::config::spamfilter::SpamClassifierModel; use common::{Server, config::spamfilter::Location, ipc::BroadcastEvent}; use mail_auth::DmarcResult; -use mail_parser::MimeHeaders; +use mail_parser::{MessageParser, MimeHeaders}; +use nlp::classifier::feature::Sample; use nlp::{ classifier::{feature::Feature, sgd::TextClassifier}, tokenizers::{ @@ -16,6 +22,7 @@ use nlp::{ types::TokenType, }, }; +use std::time::Instant; use std::{ borrow::Cow, collections::{HashMap, hash_map::Entry}, @@ -30,8 +37,11 @@ use store::{ key::DeserializeBigEndian, }, }; -use trc::AddContext; +use tokio::sync::{mpsc, oneshot}; +use trc::{AddContext, SpamEvent}; use types::{blob_hash::BlobHash, collection::Collection, field::PrincipalField}; +use unicode_security::is_potential_mixed_script_confusable_char; +use unicode_security::mixed_script::AugmentedScriptSet; pub trait SpamClassifier { fn spam_train(&self, retrain: bool) -> impl Future> + Send; @@ -40,6 +50,11 @@ pub trait SpamClassifier { &self, ctx: &mut SpamFilterContext<'_>, ) -> impl Future> + Send; + + fn spam_build_tokens<'x>( + &self, + ctx: &'x SpamFilterContext<'_>, + ) -> impl Future> + Send; } struct TrainingSample { @@ -51,11 +66,13 @@ struct TrainingSample { impl SpamClassifier for Server { async fn spam_train(&self, retrain: bool) -> trc::Result<()> { - let todo = "parse ASN and other stuff, build context properly"; let Some(config) = &self.core.spam.classifier else { return Ok(()); }; + let started = Instant::now(); + trc::event!(Spam(SpamEvent::TrainStarted)); + // Fetch model let mut model = if !retrain && let Some(model) = self @@ -126,15 +143,23 @@ impl SpamClassifier for Server { }; let do_remove = *hold == 0; + let is_spam = *is_spam == 1; samples.push(TrainingSample { hash, account_id, - is_spam: *is_spam == 1, + is_spam, remove: do_remove.then_some(until), }); remove_entries |= do_remove; + + // Update model stats model.last_sample_expiry = until; + if is_spam { + model.spam_count += 1; + } else { + model.ham_count += 1; + } Ok(true) }, @@ -142,28 +167,114 @@ impl SpamClassifier for Server { .await .caused_by(trc::location!())?; - if !samples.is_empty() { - let todo = "log no new samples"; + if samples.is_empty() { + trc::event!( + Spam(SpamEvent::TrainCompleted), + Total = 0, + Elapsed = started.elapsed() + ); + return Ok(()); } + let num_samples = samples.len(); + + // Spawn training task + struct TrainJob { + samples: Vec, + done: oneshot::Sender<()>, + } + let builder = model.classifier.feature_builder(); + let n_epochs = config.epochs; + let alpha = config.alpha; + let (batch_tx, mut batch_rx) = mpsc::channel::(1); + let (model_tx, model_rx) = oneshot::channel(); + + std::thread::Builder::new() + .name("SGD Train Task".into()) + .spawn(move || { + while let Some(mut job) = batch_rx.blocking_recv() { + model.classifier.fit(&mut job.samples, n_epochs, alpha); + let _ = job.done.send(()); + } + // Send model back when done + let _ = model_tx.send(model); + }) + .map_err(|err| { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(err) + .details("Failed to spawn spam train task") + .caused_by(trc::location!()) + })?; // Train model for chunk in samples.chunks(config.train_batch_size.max(10)) { - let todo = "do magic here"; - let mut samples = Vec::with_capacity(chunk.len()); for sample in chunk { - if sample.is_spam { - model.spam_count += 1; + let account_id = if sample.account_id != u32::MAX { + Some(sample.account_id) } else { - model.ham_count += 1; - } - samples.push(sample); + None + }; + let Some(raw_message) = self + .blob_store() + .get_blob(sample.hash.as_slice(), 0..usize::MAX) + .await + .caused_by(trc::location!())? + else { + trc::event!( + Spam(SpamEvent::TrainSampleNotFound), + Reason = "Blob not found", + AccountId = account_id, + BlobId = sample.hash.to_hex(), + ); + continue; + }; + + // Build features + let message = MessageParser::new().parse(&raw_message).unwrap_or_default(); + let mut ctx = + self.spam_filter_init(SpamFilterInput::from_message(&message, 0).train_mode()); + self.spam_filter_analyze_domain(&mut ctx).await; + self.spam_filter_analyze_url(&mut ctx).await; + let mut tokens = self.spam_build_tokens(&ctx).await.0; + builder.scale(&mut tokens); + let features = builder.build(&tokens, account_id); + + samples.push(Sample::new(features, sample.is_spam)); } - let todo = "use blocking"; + // Send batch for training + let (done_tx, done_rx) = oneshot::channel(); + batch_tx + .send(TrainJob { + samples, + done: done_tx, + }) + .await + .map_err(|err| { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(err) + .details("Spam train task failed") + .caused_by(trc::location!()) + })?; + + done_rx.await.map_err(|err| { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(err) + .details("Spam train task failed") + .caused_by(trc::location!()) + })?; } + // Take ownership of model + drop(batch_tx); + let mut model = model_rx.await.map_err(|err| { + trc::EventType::Server(trc::ServerEvent::ThreadError) + .reason(err) + .details("Spam train task failed") + .caused_by(trc::location!()) + })?; + // Store updated model model.last_trained_at = now(); let archiver = Archiver::new(model); @@ -195,6 +306,15 @@ impl SpamClassifier for Server { self.cluster_broadcast(BroadcastEvent::ReloadSpamFilter) .await; } + trc::event!( + Spam(SpamEvent::TrainCompleted), + Total = num_samples, + Details = vec![ + trc::Value::from(model.ham_count), + trc::Value::from(model.spam_count) + ], + Elapsed = started.elapsed() + ); // Remove samples marked for deletion if remove_entries { @@ -236,9 +356,10 @@ impl SpamClassifier for Server { let model = &classifier.model; if model.is_active() { + let started = Instant::now(); let mut classifier_confidence = Vec::with_capacity(ctx.input.env_rcpt_to.len()); let mut has_prediction = false; - let mut tokens = ctx.classifier_tokens().0; + let mut tokens = self.spam_build_tokens(ctx).await.0; let feature_builder = model.feature_builder(); feature_builder.scale(&mut tokens); @@ -267,49 +388,45 @@ impl SpamClassifier for Server { ctx.result.classifier_confidence = vec![prediction.into(); ctx.input.env_rcpt_to.len()]; } + + trc::event!( + Spam(SpamEvent::Classify), + Result = ctx + .result + .classifier_confidence + .iter() + .zip(ctx.input.env_rcpt_to.iter()) + .map(|(v, rcpt)| trc::Value::Array(vec![ + trc::Value::from(rcpt.to_string()), + trc::Value::from(*v) + ])) + .collect::>(), + SpanId = ctx.input.span_id, + Elapsed = started.elapsed() + ); } Ok(()) } -} -const MAX_TOKEN_LENGTH: usize = 16; - -#[derive(Debug, Clone, PartialEq, Eq, Hash)] -pub enum Token<'x> { - // User types - Word { value: Cow<'x, str> }, - Number { code: [u8; 3] }, - Alphanumeric { code: [u8; 4] }, - Symbol { value: String }, - - // User and global types - Sender { value: Cow<'x, str> }, - Asn { number: [u8; 4] }, - Url { value: Cow<'x, str> }, - Email { value: Cow<'x, str> }, - Hostname { value: &'x str }, - Attachment { value: Cow<'x, str> }, - MimeType { value: String }, -} - -#[derive(Debug)] -struct Tokens<'x>(HashMap, f32, RandomState>); - -impl<'x> SpamFilterContext<'x> { - fn classifier_tokens(&'x self) -> Tokens<'x> { + async fn spam_build_tokens<'x>(&self, ctx: &'x SpamFilterContext<'_>) -> Tokens<'x> { let mut tokens = Tokens::default(); // Add From addresses - if !matches!(self.input.dmarc_result, Some(DmarcResult::Pass)) { + if ctx + .input + .dmarc_result + .as_ref() + .is_some_and(|result| **result != DmarcResult::Pass) + { tokens.insert(Token::Sender { value: "!".into() }); } - for email in [&self.output.env_from_addr, &self.output.from.email] { + for email in [&ctx.output.env_from_addr, &ctx.output.from.email] { tokens.insert_email(email, true); } // Add Email addresses - for email in &self.output.emails { + for email in &ctx.output.emails { let is_sender = match &email.location { Location::HeaderReplyTo | Location::HeaderDnt => true, Location::BodyText @@ -319,12 +436,23 @@ impl<'x> SpamFilterContext<'x> { _ => continue, }; - tokens.insert_email(&email.element.email, is_sender); + if is_sender + || !is_trusted_domain( + self, + email.element.email.domain_part.sld_or_default(), + ctx.input.span_id, + ) + .await + { + tokens.insert_email(&email.element.email, is_sender); + } } // Add URLs - for url in &self.output.urls { - if let Some(url) = &url.element.url_parsed { + for url in &ctx.output.urls { + if let Some(url) = &url.element.url_parsed + && !is_trusted_domain(self, url.host.sld_or_default(), ctx.input.span_id).await + { if let Some(host) = &url.host.sld { tokens.insert(Token::Url { value: host.into() }); if host != &url.host.fqdn { @@ -351,26 +479,37 @@ impl<'x> SpamFilterContext<'x> { } // Add hostnames - for domain in &self.output.domains { + for domain in &ctx.output.domains { if matches!( domain.location, Location::HeaderReceived | Location::HeaderMid | Location::Ehlo | Location::Tcp ) { - tokens.insert(Token::Hostname { - value: &domain.element, - }); + let host = Hostname::new(&domain.element); + let host_sld = host.sld_or_default(); + + if !is_trusted_domain(self, host_sld, ctx.input.span_id).await { + if host_sld != host.fqdn { + tokens.insert(Token::Hostname { + value: host_sld.to_string().into(), + }); + } + + tokens.insert(Token::Hostname { + value: host.fqdn.into(), + }); + } } } // Add ASN - if let Some(asn) = self.input.asn { + if let Some(asn) = ctx.input.asn { tokens.insert(Token::Asn { number: asn.to_be_bytes(), }); } // Add MIME and attachment indicators - for part in &self.input.message.parts { + for part in &ctx.input.message.parts { if let Some(name) = part.attachment_name() && let Some((name, ext)) = name.rsplit_once('.') { @@ -407,26 +546,26 @@ impl<'x> SpamFilterContext<'x> { } // Tokenize the subject - for token in &self.output.subject_tokens { + for token in &ctx.output.subject_tokens { tokens.insert_type( - &WordStemTokenizer::new(&self.output.subject_thread_lc), + &WordStemTokenizer::new(&ctx.output.subject_thread_lc), token, ); } // Tokenize the text parts - let body_idx = self + let body_idx = ctx .input .message .html_body .first() - .or_else(|| self.input.message.text_body.first()) + .or_else(|| ctx.input.message.text_body.first()) .map(|idx| *idx as usize); let mut alt_tokens = Tokens::default(); - for (idx, part) in self.output.text_parts.iter().enumerate() { + for (idx, part) in ctx.output.text_parts.iter().enumerate() { if Some(idx) == body_idx - || (!self.input.message.text_body.contains(&(idx as u32)) - && !self.input.message.html_body.contains(&(idx as u32))) + || (!ctx.input.message.text_body.contains(&(idx as u32)) + && !ctx.input.message.html_body.contains(&(idx as u32))) { tokens.insert_text_part(part); } else { @@ -445,6 +584,29 @@ impl<'x> SpamFilterContext<'x> { } } +const MAX_TOKEN_LENGTH: usize = 16; + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum Token<'x> { + // User types + Word { value: Cow<'x, str> }, + Number { code: [u8; 2] }, + Alphanumeric { code: [u8; 4] }, + Symbol { value: String }, + + // User and global types + Sender { value: Cow<'x, str> }, + Asn { number: [u8; 4] }, + Url { value: Cow<'x, str> }, + Email { value: Cow<'x, str> }, + Hostname { value: Cow<'x, str> }, + Attachment { value: Cow<'x, str> }, + MimeType { value: String }, +} + +#[derive(Debug)] +pub struct Tokens<'x>(HashMap, f32, RandomState>); + impl<'x> Tokens<'x> { fn insert_text_part(&mut self, part: &'x TextPart<'x>) { match part { @@ -459,6 +621,7 @@ impl<'x> Tokens<'x> { text_body, tokens, .. } => { let word_tokenizer = WordStemTokenizer::new(text_body); + for token in tokens { self.insert_type(&word_tokenizer, token); } @@ -474,7 +637,32 @@ impl<'x> Tokens<'x> { ) { match token { TokenType::Alphabetic(word) => { - if word.chars().all(|c| c.is_lowercase() || !c.is_uppercase()) { + let mut set: Option = None; + let mut has_confusables = false; + let mut is_lowercase = true; + for ch in word.chars() { + has_confusables |= + !ch.is_ascii() && is_potential_mixed_script_confusable_char(ch); + is_lowercase &= ch.is_lowercase() || !ch.is_uppercase(); + set.get_or_insert_default().intersect_with(ch.into()); + } + let is_mixed_script = set.is_some_and(|set| set.is_empty()); + + if (is_mixed_script || has_confusables) + && let Ok(word) = decancer::cure(word.as_ref(), decancer::Options::default()) + { + if word.len() > MAX_TOKEN_LENGTH { + self.insert(Token::Word { + value: truncate_word(word.as_str(), MAX_TOKEN_LENGTH) + .to_string() + .into(), + }); + } else { + self.insert(Token::Word { + value: String::from(word).into(), + }); + } + } else if is_lowercase { word_tokenizer.tokenize(word, |value| match value { Cow::Borrowed(value) => { self.insert(Token::Word { @@ -531,10 +719,14 @@ impl<'x> Tokens<'x> { TokenType::Float(word) => { self.insert(Token::from_number(true, word.as_ref())); } + TokenType::IpAddr(_) => { + self.insert(Token::Url { + value: "!ip".into(), + }); + } TokenType::Email(_) | TokenType::Url(_) | TokenType::UrlNoScheme(_) - | TokenType::IpAddr(_) | TokenType::Punctuation(_) | TokenType::Space => {} } @@ -579,90 +771,143 @@ impl<'x> Tokens<'x> { impl Token<'static> { fn from_alphanumeric(s: &str) -> Self { - // Character class counts - let mut upper = 0u32; - let mut lower = 0u32; - let mut digit = 0u32; - let mut len = 0; - let mut char_types = Vec::with_capacity(len); - for c in s.chars() { - let char_type = CharType::from_char(c); - char_types.push(char_type); - match char_type { - CharType::Upper => upper += 1, - CharType::Lower => lower += 1, - CharType::Digit => digit += 1, - CharType::Other => (), - } - len += 1; - } + let mut is_hex = true; + let mut is_ascii = true; + let mut digit_count = 0; - // Determine dominant composition - let composition = match (upper > 0, lower > 0, digit > 0) { - (true, false, false) => b'U', // UPPERCASE only - (false, true, false) => b'L', // lowercase only - (false, false, true) => b'D', // digits only - (true, true, false) => b'A', // Alphabetic mixed case - (true, false, true) => b'H', // Upper + digits (common in codes) - (false, true, true) => b'M', // lower + digits (common in identifiers) - (true, true, true) => b'X', // eXtreme mix - all three - (false, false, false) => b'E', // empty/invalid - }; - - // Length bucket (log-ish scale) - let len_code = match len { - 1 => b'1', - 2 => b'2', - 3 => b'3', - 4 => b'4', - 5..=6 => b'5', - 7..=8 => b'6', - 9..=12 => b'7', - 13..=16 => b'8', - 17..=32 => b'9', - _ => b'Z', - }; - - // Ratio encoding (which class dominates) - let max_count = upper.max(lower).max(digit); - let dominance = (max_count * 100) / len.min(1) as u32; - let ratio = match dominance { - 0..=50 => b'B', // Balanced - 51..=75 => b'P', // Partial dominance - 76..=99 => b'D', // Dominant - _ => b'O', // One class only (100%) - }; - - // Run code - let mut run_count = 0; - if len > 1 { - let mut prev_type = char_types[0]; - for ¤t_type in char_types.iter().skip(1) { - if current_type != prev_type { - run_count += 1; - prev_type = current_type; + for ch in s.chars() { + match ch { + 'a'..='f' | 'A'..='F' => {} + '0'..='9' => { + digit_count += 1; + } + _ => { + is_ascii &= ch.is_ascii(); + is_hex = false; } } } - let run_ratio = (run_count as f64) / ((len - 1) as f64); - let run_code = match run_ratio { - r if r <= 0.1 => b'0', // Very long runs (e.g., AAAABBBB) - r if r <= 0.3 => b'1', // Moderate runs - r if r <= 0.5 => b'2', // Balanced runs/alternation - r if r <= 0.7 => b'3', // High alternation - _ => b'4', // Near maximum alternation (e.g., A1A1A1) - }; - Token::Alphanumeric { - code: [composition, len_code, ratio, run_code], + if is_hex { + Token::Number { + code: [b'X', s.len().min(u8::MAX as usize) as u8], + } + } else if !is_ascii { + let word: String = if let Ok(cured) = decancer::cure(s, decancer::Options::default()) { + cured + .as_str() + .chars() + .filter(|ch| ch.is_alphabetic()) + .take(MAX_TOKEN_LENGTH) + .collect() + } else { + s.chars() + .filter(|ch| ch.is_alphabetic()) + .flat_map(|ch| ch.to_lowercase()) + .take(MAX_TOKEN_LENGTH) + .collect() + }; + + Token::Word { value: word.into() } + } else if s.len() > 3 && digit_count == 1 { + let word: String = s + .chars() + .filter(|ch| ch.is_alphabetic()) + .flat_map(|ch| ch.to_lowercase()) + .take(MAX_TOKEN_LENGTH) + .collect(); + Token::Word { value: word.into() } + } else { + // Character class counts + let mut upper = 0u32; + let mut lower = 0u32; + let mut digit = 0u32; + let mut len = 0; + let mut char_types = Vec::with_capacity(len); + for c in s.chars() { + let char_type = CharType::from_char(c); + char_types.push(char_type); + match char_type { + CharType::Upper => upper += 1, + CharType::Lower => lower += 1, + CharType::Digit => digit += 1, + CharType::Other => (), + } + len += 1; + } + + // Determine dominant composition + let composition = match (upper > 0, lower > 0, digit > 0) { + (true, false, false) => b'U', // UPPERCASE only + (false, true, false) => b'L', // lowercase only + (false, false, true) => b'D', // digits only + (true, true, false) => b'A', // Alphabetic mixed case + (true, false, true) => b'H', // Upper + digits (common in codes) + (false, true, true) => b'M', // lower + digits (common in identifiers) + (true, true, true) => b'X', // eXtreme mix - all three + (false, false, false) => b'E', // empty/invalid + }; + + // Length bucket (log-ish scale) + let len_code = match len { + 1 => b'1', + 2 => b'2', + 3 => b'3', + 4 => b'4', + 5..=6 => b'5', + 7..=8 => b'6', + 9..=12 => b'7', + 13..=16 => b'8', + 17..=32 => b'9', + _ => b'Z', + }; + + // Ratio encoding (which class dominates) + let max_count = upper.max(lower).max(digit); + let dominance = (max_count * 100) / len.min(1) as u32; + let ratio = match dominance { + 0..=50 => b'B', // Balanced + 51..=75 => b'P', // Partial dominance + 76..=99 => b'D', // Dominant + _ => b'O', // One class only (100%) + }; + + // Run code + let mut run_count = 0; + if len > 1 { + let mut prev_type = char_types[0]; + for ¤t_type in char_types.iter().skip(1) { + if current_type != prev_type { + run_count += 1; + prev_type = current_type; + } + } + } + let run_ratio = (run_count as f64) / ((len - 1) as f64); + let run_code = match run_ratio { + r if r <= 0.1 => b'0', // Very long runs (e.g., AAAABBBB) + r if r <= 0.3 => b'1', // Moderate runs + r if r <= 0.5 => b'2', // Balanced runs/alternation + r if r <= 0.7 => b'3', // High alternation + _ => b'4', // Near maximum alternation (e.g., A1A1A1) + }; + + Token::Alphanumeric { + code: [composition, len_code, ratio, run_code], + } } } fn from_number(is_float: bool, num: &str) -> Self { Token::Number { code: [ - u8::from(is_float), - u8::from(num.starts_with('-')), + if num.starts_with("-") { + if is_float { b'F' } else { b'I' } + } else if is_float { + b'f' + } else { + b'i' + }, num.as_bytes() .iter() .filter(|c| c.is_ascii_digit()) diff --git a/crates/store/src/write/blob.rs b/crates/store/src/write/blob.rs index e0fafa38..94e8b7e4 100644 --- a/crates/store/src/write/blob.rs +++ b/crates/store/src/write/blob.rs @@ -117,79 +117,83 @@ impl Store { } pub async fn purge_blobs(&self, blob_store: BlobStore) -> trc::Result<()> { - // Validate linked blobs - let from_key = ValueKey { - account_id: 0, - collection: 0, - document_id: 0, - class: ValueClass::Blob(BlobOp::Commit { - hash: BlobHash::default(), - }), - }; - let to_key = ValueKey { - account_id: u32::MAX, - collection: u8::MAX, - document_id: u32::MAX, - class: ValueClass::Blob(BlobOp::Link { - hash: BlobHash::new_max(), - to: BlobLink::Document, - }), - }; + for byte in 0..=u8::MAX { + // Validate linked blobs + let mut from_hash = BlobHash::default(); + let mut to_hash = BlobHash::new_max(); + from_hash.0[0] = byte; + to_hash.0[0] = byte; + let from_key = ValueKey { + account_id: 0, + collection: 0, + document_id: 0, + class: ValueClass::Blob(BlobOp::Commit { hash: from_hash }), + }; + let to_key = ValueKey { + account_id: u32::MAX, + collection: u8::MAX, + document_id: u32::MAX, + class: ValueClass::Blob(BlobOp::Link { + hash: to_hash, + to: BlobLink::Document, + }), + }; - let mut state = BlobPurgeState::new(); - self.iterate( - IterateParams::new(from_key, to_key).ascending(), - |key, value| { - let hash = - BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else(|| { - trc::Error::corrupted_key(key, value.into(), trc::location!()) - })?) - .unwrap(); + let mut state = BlobPurgeState::new(); + self.iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let hash = + BlobHash::try_from_hash_slice(key.get(0..BLOB_HASH_LEN).ok_or_else( + || trc::Error::corrupted_key(key, value.into(), trc::location!()), + )?) + .unwrap(); - state.update_hash(hash); - state.process_key(key, value)?; + state.update_hash(hash); + state.process_key(key, value)?; - Ok(true) - }, - ) - .await - .caused_by(trc::location!())?; + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; - state.finalize(BlobHash::default()); + state.finalize(BlobHash::default()); - // Delete expired or unlinked blobs - for (_, op) in &state.delete_keys { - if let BlobOp::Commit { hash } = op { - blob_store - .delete_blob(hash.as_ref()) - .await - .caused_by(trc::location!())?; + // Delete expired or unlinked blobs + for (_, op) in &state.delete_keys { + if let BlobOp::Commit { hash } = op { + blob_store + .delete_blob(hash.as_ref()) + .await + .caused_by(trc::location!())?; + } } - } - // Delete hashes - let mut batch = BatchBuilder::new(); - for (account_id, op) in state.delete_keys { - if batch.is_large_batch() { + // Delete hashes + let mut batch = BatchBuilder::new(); + for (account_id, op) in state.delete_keys { + if batch.is_large_batch() { + self.write(batch.build_all()) + .await + .caused_by(trc::location!())?; + batch = BatchBuilder::new(); + } + + if let Some(account_id) = account_id { + batch.with_account_id(account_id); + } + + batch.any_op(Operation::Value { + class: ValueClass::Blob(op), + op: ValueOp::Clear, + }); + } + if !batch.is_empty() { self.write(batch.build_all()) .await .caused_by(trc::location!())?; - batch = BatchBuilder::new(); } - - if let Some(account_id) = account_id { - batch.with_account_id(account_id); - } - - batch.any_op(Operation::Value { - class: ValueClass::Blob(op), - op: ValueOp::Clear, - }); - } - if !batch.is_empty() { - self.write(batch.build_all()) - .await - .caused_by(trc::location!())?; } Ok(()) diff --git a/crates/trc/src/event/conv.rs b/crates/trc/src/event/conv.rs index 2bd8ce26..41aaba96 100644 --- a/crates/trc/src/event/conv.rs +++ b/crates/trc/src/event/conv.rs @@ -59,6 +59,12 @@ impl From for Value { } } +impl From for Value { + fn from(value: f32) -> Self { + Self::Float(value.into()) + } +} + impl From for Value { fn from(value: u16) -> Self { Self::UInt(value.into()) diff --git a/crates/trc/src/event/description.rs b/crates/trc/src/event/description.rs index 119c6a50..3a6647f1 100644 --- a/crates/trc/src/event/description.rs +++ b/crates/trc/src/event/description.rs @@ -1013,29 +1013,35 @@ impl SpamEvent { match self { SpamEvent::Pyzor => "Pyzor success", SpamEvent::PyzorError => "Pyzor error", - SpamEvent::Train => "Training spam filter", - SpamEvent::TrainBalance => "Spam filter model balance verify", - SpamEvent::TrainError => "Error training spam filter", SpamEvent::Classify => "Classifying message for spam", - SpamEvent::ClassifyError => "Not enough training data for spam filter", SpamEvent::Dnsbl => "DNSBL query", SpamEvent::DnsblError => "Error querying DNSBL", - SpamEvent::TrainAccount => "Training spam filter for account", + SpamEvent::TrainStarted => "Spam classifier training started", + SpamEvent::TrainCompleted => "Spam classifier training completed", + SpamEvent::TrainSampleAdded => "New training sample added", + SpamEvent::TrainSampleNotFound => "Training sample not found", + SpamEvent::ModelLoaded => "Spam classifier model loaded", + SpamEvent::ModelNotReady => "Spam classifier model not ready", + SpamEvent::ModelNotFound => "Spam classifier model not found", } } pub fn explain(&self) -> &'static str { match self { SpamEvent::PyzorError => "An error occurred with Pyzor", - SpamEvent::Train => "The spam filter is being trained with the message", - SpamEvent::TrainBalance => "The spam filter training data is verified for balance", - SpamEvent::TrainError => "An error occurred while training the spam filter", SpamEvent::Classify => "The message is being classified for spam", - SpamEvent::ClassifyError => "There is not enough training data for the spam filter", SpamEvent::Pyzor => "Pyzor query successful", SpamEvent::Dnsbl => "The DNSBL query was successful", SpamEvent::DnsblError => "An error occurred while querying the DNSBL", - SpamEvent::TrainAccount => "The spam filter has been trained for the account", + SpamEvent::TrainStarted => "SGD logistic regression training has started", + SpamEvent::TrainCompleted => "SGD logistic regression training has completed", + SpamEvent::TrainSampleAdded => "A new training sample has been added", + SpamEvent::TrainSampleNotFound => "A training sample was not found", + SpamEvent::ModelLoaded => "The spam classifier model has been loaded", + SpamEvent::ModelNotReady => { + "The spam classifier model has not been trained with enough data" + } + SpamEvent::ModelNotFound => "The spam classifier model has not been trained yet", } } } diff --git a/crates/trc/src/event/level.rs b/crates/trc/src/event/level.rs index 2157c42b..6fbb406e 100644 --- a/crates/trc/src/event/level.rs +++ b/crates/trc/src/event/level.rs @@ -340,16 +340,18 @@ impl EventType { | SieveEvent::ActionReject => Level::Debug, }, EventType::Spam(event) => match event { - SpamEvent::PyzorError - | SpamEvent::TrainError + SpamEvent::Pyzor + | SpamEvent::PyzorError + | SpamEvent::Dnsbl | SpamEvent::DnsblError - | SpamEvent::Pyzor - | SpamEvent::Train - | SpamEvent::TrainAccount | SpamEvent::Classify - | SpamEvent::ClassifyError - | SpamEvent::TrainBalance - | SpamEvent::Dnsbl => Level::Debug, + | SpamEvent::TrainSampleAdded => Level::Debug, + SpamEvent::TrainSampleNotFound => Level::Warn, + SpamEvent::TrainStarted + | SpamEvent::TrainCompleted + | SpamEvent::ModelLoaded + | SpamEvent::ModelNotReady + | SpamEvent::ModelNotFound => Level::Info, }, EventType::Http(event) => match event { HttpEvent::ConnectionStart | HttpEvent::ConnectionEnd => Level::Debug, diff --git a/crates/trc/src/ipc/metrics.rs b/crates/trc/src/ipc/metrics.rs index 2458460e..a61ed235 100644 --- a/crates/trc/src/ipc/metrics.rs +++ b/crates/trc/src/ipc/metrics.rs @@ -578,10 +578,10 @@ impl EventType { ) => true, EventType::Spam( SpamEvent::PyzorError - | SpamEvent::Train - | SpamEvent::TrainError + | SpamEvent::TrainCompleted + | SpamEvent::TrainSampleAdded | SpamEvent::Classify - | SpamEvent::ClassifyError + | SpamEvent::ModelNotReady | SpamEvent::DnsblError, ) => true, EventType::PushSubscription(_) => true, diff --git a/crates/trc/src/lib.rs b/crates/trc/src/lib.rs index 1adaccd6..85e72989 100644 --- a/crates/trc/src/lib.rs +++ b/crates/trc/src/lib.rs @@ -605,12 +605,14 @@ pub enum SpamEvent { PyzorError, Dnsbl, DnsblError, - Train, - TrainBalance, - TrainError, + TrainStarted, + TrainCompleted, + TrainSampleAdded, + TrainSampleNotFound, Classify, - ClassifyError, - TrainAccount, + ModelLoaded, + ModelNotReady, + ModelNotFound, } #[event_type] diff --git a/crates/trc/src/serializers/binary.rs b/crates/trc/src/serializers/binary.rs index 35160beb..41319240 100644 --- a/crates/trc/src/serializers/binary.rs +++ b/crates/trc/src/serializers/binary.rs @@ -445,7 +445,7 @@ impl EventType { EventType::Eval(EvalEvent::StoreNotFound) => 140, EventType::TaskQueue(TaskQueueEvent::BlobNotFound) => 141, EventType::MessageIngest(MessageIngestEvent::FtsIndex) => 142, - EventType::Spam(SpamEvent::TrainAccount) => 143, + EventType::Spam(SpamEvent::TrainSampleAdded) => 143, EventType::TaskQueue(TaskQueueEvent::TaskLocked) => 144, EventType::TaskQueue(TaskQueueEvent::MetadataNotFound) => 145, EventType::Housekeeper(HousekeeperEvent::Run) => 146, @@ -786,13 +786,13 @@ impl EventType { EventType::Smtp(SmtpEvent::VrfyDisabled) => 488, EventType::Smtp(SmtpEvent::VrfyNotFound) => 489, EventType::Spam(SpamEvent::Classify) => 490, - EventType::Spam(SpamEvent::ClassifyError) => 491, + EventType::Spam(SpamEvent::TrainSampleNotFound) => 491, EventType::Store(StoreEvent::HttpStoreFetch) => 492, EventType::Store(StoreEvent::HttpStoreError) => 493, EventType::Spam(SpamEvent::PyzorError) => 494, - EventType::Spam(SpamEvent::Train) => 495, - EventType::Spam(SpamEvent::TrainBalance) => 496, - EventType::Spam(SpamEvent::TrainError) => 497, + EventType::Spam(SpamEvent::TrainCompleted) => 495, + EventType::Spam(SpamEvent::ModelNotReady) => 496, + EventType::Spam(SpamEvent::ModelNotFound) => 497, EventType::Spf(SpfEvent::Fail) => 498, EventType::Spf(SpfEvent::Neutral) => 499, EventType::Spf(SpfEvent::None) => 500, @@ -895,6 +895,8 @@ impl EventType { EventType::Calendar(CalendarEvent::ItipMessageError) => 585, EventType::TaskQueue(TaskQueueEvent::TaskIgnored) => 586, EventType::TaskQueue(TaskQueueEvent::TaskFailed) => 587, + EventType::Spam(SpamEvent::TrainStarted) => 588, + EventType::Spam(SpamEvent::ModelLoaded) => 589, } } @@ -1038,7 +1040,7 @@ impl EventType { 140 => Some(EventType::Eval(EvalEvent::StoreNotFound)), 141 => Some(EventType::TaskQueue(TaskQueueEvent::BlobNotFound)), 142 => Some(EventType::MessageIngest(MessageIngestEvent::FtsIndex)), - 143 => Some(EventType::Spam(SpamEvent::TrainAccount)), + 143 => Some(EventType::Spam(SpamEvent::TrainSampleAdded)), 144 => Some(EventType::TaskQueue(TaskQueueEvent::TaskLocked)), 145 => Some(EventType::TaskQueue(TaskQueueEvent::MetadataNotFound)), 146 => Some(EventType::Housekeeper(HousekeeperEvent::Run)), @@ -1415,13 +1417,13 @@ impl EventType { 488 => Some(EventType::Smtp(SmtpEvent::VrfyDisabled)), 489 => Some(EventType::Smtp(SmtpEvent::VrfyNotFound)), 490 => Some(EventType::Spam(SpamEvent::Classify)), - 491 => Some(EventType::Spam(SpamEvent::ClassifyError)), + 491 => Some(EventType::Spam(SpamEvent::TrainSampleNotFound)), 492 => Some(EventType::Store(StoreEvent::HttpStoreFetch)), 493 => Some(EventType::Store(StoreEvent::HttpStoreError)), 494 => Some(EventType::Spam(SpamEvent::PyzorError)), - 495 => Some(EventType::Spam(SpamEvent::Train)), - 496 => Some(EventType::Spam(SpamEvent::TrainBalance)), - 497 => Some(EventType::Spam(SpamEvent::TrainError)), + 495 => Some(EventType::Spam(SpamEvent::TrainCompleted)), + 496 => Some(EventType::Spam(SpamEvent::ModelNotReady)), + 497 => Some(EventType::Spam(SpamEvent::ModelNotFound)), 498 => Some(EventType::Spf(SpfEvent::Fail)), 499 => Some(EventType::Spf(SpfEvent::Neutral)), 500 => Some(EventType::Spf(SpfEvent::None)), @@ -1528,6 +1530,8 @@ impl EventType { 585 => Some(EventType::Calendar(CalendarEvent::ItipMessageError)), 586 => Some(EventType::TaskQueue(TaskQueueEvent::TaskIgnored)), 587 => Some(EventType::TaskQueue(TaskQueueEvent::TaskFailed)), + 588 => Some(EventType::Spam(SpamEvent::TrainStarted)), + 589 => Some(EventType::Spam(SpamEvent::ModelLoaded)), _ => None, } }