From 33f288d0eaf2c761f3cafa04372485ce9074a1f8 Mon Sep 17 00:00:00 2001 From: Maurus Decimus <11444311+mdecimus@users.noreply.github.com> Date: Wed, 10 Jun 2026 13:51:36 +0200 Subject: [PATCH] Fix JMAP: `*/changes` methods leak ids of non-shared objects --- CHANGELOG.md | 11 + Cargo.lock | 62 +- crates/common/Cargo.toml | 2 +- crates/coordinator/Cargo.toml | 2 +- crates/dav-proto/Cargo.toml | 2 +- crates/dav/Cargo.toml | 2 +- crates/directory/Cargo.toml | 2 +- crates/email/Cargo.toml | 2 +- crates/groupware/Cargo.toml | 2 +- crates/http-proto/Cargo.toml | 2 +- crates/http/Cargo.toml | 2 +- crates/imap-proto/Cargo.toml | 2 +- crates/imap/Cargo.toml | 2 +- crates/jmap-proto/Cargo.toml | 2 +- crates/jmap/Cargo.toml | 2 +- crates/jmap/src/changes/get.rs | 89 ++- crates/main/Cargo.toml | 2 +- crates/managesieve/Cargo.toml | 2 +- crates/migration/Cargo.toml | 2 +- crates/nlp/Cargo.toml | 2 +- crates/pop3/Cargo.toml | 2 +- crates/registry/Cargo.toml | 2 +- crates/services/Cargo.toml | 2 +- crates/smtp/Cargo.toml | 2 +- crates/spam-filter/Cargo.toml | 2 +- crates/store/Cargo.toml | 2 +- crates/trc/Cargo.toml | 2 +- crates/trc/event-macro/Cargo.toml | 2 +- crates/types/Cargo.toml | 2 +- crates/utils/Cargo.toml | 2 +- crates/utils/proc-macros/Cargo.toml | 2 +- tests/Cargo.toml | 2 +- tests/resources/scripts/smtp_stress_test.py | 627 ++++++++++++++++++ tests/resources/scripts/stress_test.py | 524 ++++++++++++--- .../resources/scripts/stress_test_prepare.py | 275 ++++++-- tests/src/jmap/mail/acl.rs | 19 + 36 files changed, 1475 insertions(+), 190 deletions(-) create mode 100644 tests/resources/scripts/smtp_stress_test.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e81d80b9..7a9cc7cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ All notable changes to this project will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org/). +## [0.16.9] - 2026-06-XX + +If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If you are upgrading from v0.15.x and below, please read the [upgrading documentation](https://github.com/stalwartlabs/stalwart/blob/main/UPGRADING/v0_16.md) for more information on how to upgrade from previous versions. + +## Added + +## Changed + +## Fixed +- JMAP: `*/changes` methods leak ids of non-shared objects + ## [0.16.8] - 2026-06-06 If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If you are upgrading from v0.15.x and below, please read the [upgrading documentation](https://github.com/stalwartlabs/stalwart/blob/main/UPGRADING/v0_16.md) for more information on how to upgrade from previous versions. diff --git a/Cargo.lock b/Cargo.lock index 8900f6a2..6d8989e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1102,7 +1102,7 @@ dependencies = [ [[package]] name = "common" -version = "0.16.8" +version = "0.16.9" dependencies = [ "aes-gcm-siv", "ahash", @@ -1298,7 +1298,7 @@ checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" [[package]] name = "coordinator" -version = "0.16.8" +version = "0.16.9" dependencies = [ "async-nats", "futures", @@ -1651,7 +1651,7 @@ checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" [[package]] name = "dav" -version = "0.16.8" +version = "0.16.9" dependencies = [ "calcard", "chrono", @@ -1674,7 +1674,7 @@ dependencies = [ [[package]] name = "dav-proto" -version = "0.16.8" +version = "0.16.9" dependencies = [ "calcard", "chrono", @@ -1873,7 +1873,7 @@ dependencies = [ [[package]] name = "directory" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "argon2", @@ -2114,7 +2114,7 @@ dependencies = [ [[package]] name = "email" -version = "0.16.8" +version = "0.16.9" dependencies = [ "aes 0.9.1", "aes-gcm", @@ -2211,7 +2211,7 @@ dependencies = [ [[package]] name = "event_macro" -version = "0.16.8" +version = "0.16.9" dependencies = [ "proc-macro2", "quote", @@ -2708,7 +2708,7 @@ dependencies = [ [[package]] name = "groupware" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "calcard", @@ -2990,7 +2990,7 @@ dependencies = [ [[package]] name = "http" -version = "0.16.8" +version = "0.16.9" dependencies = [ "async-stream", "base64 0.22.1", @@ -3089,7 +3089,7 @@ dependencies = [ [[package]] name = "http_proto" -version = "0.16.8" +version = "0.16.9" dependencies = [ "common", "compact_str", @@ -3361,7 +3361,7 @@ checksum = "09e54e57b4c48b40f7aec75635392b12b3421fa26fe8b4332e63138ed278459c" [[package]] name = "imap" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "common", @@ -3388,7 +3388,7 @@ dependencies = [ [[package]] name = "imap_proto" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "chrono", @@ -3601,7 +3601,7 @@ dependencies = [ [[package]] name = "jmap" -version = "0.16.8" +version = "0.16.9" dependencies = [ "aes-gcm", "aes-gcm-siv", @@ -3689,7 +3689,7 @@ dependencies = [ [[package]] name = "jmap_proto" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "calcard", @@ -4198,7 +4198,7 @@ dependencies = [ [[package]] name = "managesieve" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "common", @@ -4311,7 +4311,7 @@ checksum = "c797b9d6bb23aab2fc369c65f871be49214f5c759af65bde26ffaaa2b646b492" [[package]] name = "migration" -version = "0.16.8" +version = "0.16.9" dependencies = [ "base64 0.22.1", "common", @@ -4528,7 +4528,7 @@ dependencies = [ [[package]] name = "nlp" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "bincode 1.3.3", @@ -5262,7 +5262,7 @@ dependencies = [ [[package]] name = "pop3" -version = "0.16.8" +version = "0.16.9" dependencies = [ "common", "directory", @@ -5434,7 +5434,7 @@ dependencies = [ [[package]] name = "proc_macros" -version = "0.16.8" +version = "0.16.9" dependencies = [ "proc-macro2", "quote", @@ -6066,7 +6066,7 @@ checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" [[package]] name = "registry" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "hashify", @@ -6980,7 +6980,7 @@ dependencies = [ [[package]] name = "services" -version = "0.16.8" +version = "0.16.9" dependencies = [ "aes-gcm", "aes-gcm-siv", @@ -7238,7 +7238,7 @@ checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" [[package]] name = "smtp" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "base64 0.22.1", @@ -7339,7 +7339,7 @@ dependencies = [ [[package]] name = "spam-filter" -version = "0.16.8" +version = "0.16.9" dependencies = [ "common", "compact_str", @@ -7449,7 +7449,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" [[package]] name = "stalwart" -version = "0.16.8" +version = "0.16.9" dependencies = [ "common", "coordinator", @@ -7457,7 +7457,7 @@ dependencies = [ "directory", "email", "groupware", - "http 0.16.8", + "http 0.16.9", "http_proto", "imap", "jemallocator", @@ -7486,7 +7486,7 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "store" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "arc-swap", @@ -7683,7 +7683,7 @@ dependencies = [ [[package]] name = "tests" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "async-trait", @@ -7707,7 +7707,7 @@ dependencies = [ "form_urlencoded", "futures", "groupware", - "http 0.16.8", + "http 0.16.9", "http-body-util", "http_proto", "hyper", @@ -8243,7 +8243,7 @@ dependencies = [ [[package]] name = "trc" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "base64 0.22.1", @@ -8336,7 +8336,7 @@ checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" [[package]] name = "types" -version = "0.16.8" +version = "0.16.9" dependencies = [ "blake3", "compact_str", @@ -8501,7 +8501,7 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" [[package]] name = "utils" -version = "0.16.8" +version = "0.16.9" dependencies = [ "ahash", "arcstr", diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml index 4e2d96ac..c07c0172 100644 --- a/crates/common/Cargo.toml +++ b/crates/common/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "common" -version = "0.16.8" +version = "0.16.9" edition = "2024" build = "build.rs" diff --git a/crates/coordinator/Cargo.toml b/crates/coordinator/Cargo.toml index e3e212fc..dff272ca 100644 --- a/crates/coordinator/Cargo.toml +++ b/crates/coordinator/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "coordinator" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/dav-proto/Cargo.toml b/crates/dav-proto/Cargo.toml index ed4ce04b..cb96d915 100644 --- a/crates/dav-proto/Cargo.toml +++ b/crates/dav-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dav-proto" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/dav/Cargo.toml b/crates/dav/Cargo.toml index e487e4e2..d4bc13a3 100644 --- a/crates/dav/Cargo.toml +++ b/crates/dav/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "dav" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/directory/Cargo.toml b/crates/directory/Cargo.toml index 328020d4..3820c542 100644 --- a/crates/directory/Cargo.toml +++ b/crates/directory/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "directory" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/email/Cargo.toml b/crates/email/Cargo.toml index 69ba53f1..a656e9b7 100644 --- a/crates/email/Cargo.toml +++ b/crates/email/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "email" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/groupware/Cargo.toml b/crates/groupware/Cargo.toml index 2beb56e3..f29b471b 100644 --- a/crates/groupware/Cargo.toml +++ b/crates/groupware/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "groupware" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/http-proto/Cargo.toml b/crates/http-proto/Cargo.toml index bc39b7db..c61f7fe3 100644 --- a/crates/http-proto/Cargo.toml +++ b/crates/http-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "http_proto" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/http/Cargo.toml b/crates/http/Cargo.toml index 42fe9a8b..7667f30c 100644 --- a/crates/http/Cargo.toml +++ b/crates/http/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "http" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/imap-proto/Cargo.toml b/crates/imap-proto/Cargo.toml index e82d7fea..2184fb2a 100644 --- a/crates/imap-proto/Cargo.toml +++ b/crates/imap-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "imap_proto" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/imap/Cargo.toml b/crates/imap/Cargo.toml index f1b3af60..66d026a7 100644 --- a/crates/imap/Cargo.toml +++ b/crates/imap/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "imap" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/jmap-proto/Cargo.toml b/crates/jmap-proto/Cargo.toml index 33b6198f..aac986c7 100644 --- a/crates/jmap-proto/Cargo.toml +++ b/crates/jmap-proto/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "jmap_proto" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index d8f0d996..164a42e8 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "jmap" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/jmap/src/changes/get.rs b/crates/jmap/src/changes/get.rs index 65fe9482..821b1e20 100644 --- a/crates/jmap/src/changes/get.rs +++ b/crates/jmap/src/changes/get.rs @@ -6,7 +6,7 @@ use crate::{api::auth::JmapAuthorization, changes::state::JmapCacheState}; use common::{Server, auth::AccessToken}; -use email::cache::MessageCacheFetch; +use email::cache::{MessageCacheFetch, email::MessageCacheAccess, mailbox::MailboxCacheAccess}; use groupware::cache::GroupwareCache; use jmap_proto::{ method::changes::{ChangesRequest, ChangesResponse}, @@ -16,9 +16,15 @@ use jmap_proto::{ types::state::State, }; use std::future::Future; -use store::query::log::{Change, Query}; +use store::{ + query::log::{Change, Query}, + roaring::RoaringBitmap, +}; use trc::AddContext; -use types::collection::{Collection, SyncCollection}; +use types::{ + acl::Acl, + collection::{Collection, SyncCollection}, +}; pub trait ChangesLookup: Sync + Send { fn changes( @@ -126,6 +132,73 @@ impl ChangesLookup for Server { }; let account_id = request.account_id.document_id(); + let allowed_ids: Option = if access_token.is_member(account_id) { + None + } else { + Some(match object { + MethodObject::Email => self + .get_cached_messages(account_id) + .await? + .shared_messages(access_token, Acl::ReadItems), + MethodObject::Mailbox => self + .get_cached_messages(account_id) + .await? + .shared_mailboxes(access_token, Acl::Read), + MethodObject::Thread => { + let cache = self.get_cached_messages(account_id).await?; + let shared = cache.shared_messages(access_token, Acl::ReadItems); + let mut threads = RoaringBitmap::new(); + for item in &cache.emails.items { + if shared.contains(item.document_id) { + threads.insert(item.thread_id); + } + } + threads + } + MethodObject::AddressBook => self + .fetch_dav_resources( + access_token.account_id(), + account_id, + SyncCollection::AddressBook, + ) + .await? + .shared_containers(access_token, [Acl::Read, Acl::ReadItems], true), + MethodObject::ContactCard => self + .fetch_dav_resources( + access_token.account_id(), + account_id, + SyncCollection::AddressBook, + ) + .await? + .shared_items(access_token, [Acl::ReadItems], true), + MethodObject::Calendar => self + .fetch_dav_resources( + access_token.account_id(), + account_id, + SyncCollection::Calendar, + ) + .await? + .shared_containers(access_token, [Acl::Read, Acl::ReadItems], true), + MethodObject::CalendarEvent => self + .fetch_dav_resources( + access_token.account_id(), + account_id, + SyncCollection::Calendar, + ) + .await? + .shared_items(access_token, [Acl::ReadItems], true), + MethodObject::FileNode => self + .fetch_dav_resources( + access_token.account_id(), + account_id, + SyncCollection::FileNode, + ) + .await? + .shared_documents(access_token, [Acl::Read, Acl::ReadItems], true), + _ => RoaringBitmap::new(), + }) + }; + let (items_sent, changelog) = match &request.since_state { State::Initial => { let changelog = self @@ -226,6 +299,16 @@ impl ChangesLookup for Server { (is_container && change.is_container_change()) || (!is_container && change.is_item_change()) }) + .filter(|change| { + allowed_ids.as_ref().is_none_or(|allowed| { + let id = if is_container { + change.container_id() + } else { + change.item_id() + }; + id.is_some_and(|id| allowed.contains(id as u32)) + }) + }) .skip(items_sent) .peekable(); diff --git a/crates/main/Cargo.toml b/crates/main/Cargo.toml index 6810d3da..1ea8a72d 100644 --- a/crates/main/Cargo.toml +++ b/crates/main/Cargo.toml @@ -7,7 +7,7 @@ homepage = "https://stalw.art" keywords = ["imap", "jmap", "smtp", "email", "mail", "webdav", "server"] categories = ["email"] license = "AGPL-3.0-only OR LicenseRef-SEL" -version = "0.16.8" +version = "0.16.9" edition = "2024" [[bin]] diff --git a/crates/managesieve/Cargo.toml b/crates/managesieve/Cargo.toml index 7f8c0b6d..fa8c1b0b 100644 --- a/crates/managesieve/Cargo.toml +++ b/crates/managesieve/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "managesieve" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/migration/Cargo.toml b/crates/migration/Cargo.toml index 7045c7ca..1014fc08 100644 --- a/crates/migration/Cargo.toml +++ b/crates/migration/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "migration" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/nlp/Cargo.toml b/crates/nlp/Cargo.toml index 3865969c..750dad09 100644 --- a/crates/nlp/Cargo.toml +++ b/crates/nlp/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "nlp" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/pop3/Cargo.toml b/crates/pop3/Cargo.toml index d70b8f21..29d2edb6 100644 --- a/crates/pop3/Cargo.toml +++ b/crates/pop3/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "pop3" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/registry/Cargo.toml b/crates/registry/Cargo.toml index ec08eb84..796f9541 100644 --- a/crates/registry/Cargo.toml +++ b/crates/registry/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "registry" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/services/Cargo.toml b/crates/services/Cargo.toml index 16f67db0..ef366844 100644 --- a/crates/services/Cargo.toml +++ b/crates/services/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "services" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/smtp/Cargo.toml b/crates/smtp/Cargo.toml index 46ee71fa..ba1d4d27 100644 --- a/crates/smtp/Cargo.toml +++ b/crates/smtp/Cargo.toml @@ -7,7 +7,7 @@ homepage = "https://stalw.art/smtp" keywords = ["smtp", "email", "mail", "server"] categories = ["email"] license = "AGPL-3.0-only OR LicenseRef-SEL" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/spam-filter/Cargo.toml b/crates/spam-filter/Cargo.toml index a931788d..b03aeab9 100644 --- a/crates/spam-filter/Cargo.toml +++ b/crates/spam-filter/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "spam-filter" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index ae579e3a..79e356a1 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "store" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/trc/Cargo.toml b/crates/trc/Cargo.toml index 0b868c3d..7af258bd 100644 --- a/crates/trc/Cargo.toml +++ b/crates/trc/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "trc" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/trc/event-macro/Cargo.toml b/crates/trc/event-macro/Cargo.toml index 07294dd2..f83fd798 100644 --- a/crates/trc/event-macro/Cargo.toml +++ b/crates/trc/event-macro/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "event_macro" -version = "0.16.8" +version = "0.16.9" edition = "2024" [lib] diff --git a/crates/types/Cargo.toml b/crates/types/Cargo.toml index 2d77d629..b25c08be 100644 --- a/crates/types/Cargo.toml +++ b/crates/types/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "types" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/utils/Cargo.toml b/crates/utils/Cargo.toml index 4db0da65..deb9eb80 100644 --- a/crates/utils/Cargo.toml +++ b/crates/utils/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "utils" -version = "0.16.8" +version = "0.16.9" edition = "2024" [dependencies] diff --git a/crates/utils/proc-macros/Cargo.toml b/crates/utils/proc-macros/Cargo.toml index 0a23144f..77e31fad 100644 --- a/crates/utils/proc-macros/Cargo.toml +++ b/crates/utils/proc-macros/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "proc_macros" -version = "0.16.8" +version = "0.16.9" edition = "2024" [lib] diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 762aacf8..cdfa3962 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tests" -version = "0.16.8" +version = "0.16.9" edition = "2024" [features] diff --git a/tests/resources/scripts/smtp_stress_test.py b/tests/resources/scripts/smtp_stress_test.py new file mode 100644 index 00000000..b03ee192 --- /dev/null +++ b/tests/resources/scripts/smtp_stress_test.py @@ -0,0 +1,627 @@ +import argparse +import math +import multiprocessing +import os +import queue +import random +import re +import shutil +import smtplib +import ssl +import sys +import tempfile +import threading +import time +from email.utils import formatdate, make_msgid + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 25 +DEFAULT_THREADS = 5 +DEFAULT_PROCESSES = 1 +DEFAULT_MESSAGES = 100 +DEFAULT_MIN_SIZE = 1024 +DEFAULT_MAX_SIZE = 51200 +DEFAULT_POOL_SIZE = 64 +DEFAULT_SENDER = "stress-test@example.com" +DEFAULT_USERS_FILE = "users.txt" +DEFAULT_DICTIONARY = "/usr/share/dict/words" +DEFAULT_TIMEOUT = 60 +LINE_WIDTH = 72 + +FALLBACK_WORDS = ( + "lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiusmod " + "tempor incididunt ut labore et dolore magna aliqua enim ad minim veniam " + "quis nostrud exercitation ullamco laboris nisi aliquip ex ea commodo" +).split() + +WORDS = FALLBACK_WORDS + +DOT_LINE = re.compile(br"(?m)^\.") +STOP_EVENT = threading.Event() + + +class SmtpError(Exception): + pass + + +class AsyncLogger: + def __init__(self, enabled): + self.enabled = enabled + self._queue = queue.Queue() if enabled else None + self._thread = None + + def start(self): + if not self.enabled: + return + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + + def log(self, message): + if self.enabled: + self._queue.put(message) + + def _run(self): + while True: + message = self._queue.get() + if message is None: + break + print(message, file=sys.stderr, flush=True) + + def stop(self): + if not self.enabled: + return + self._queue.put(None) + if self._thread is not None: + self._thread.join() + + +def load_words(path): + try: + with open(path, "r", encoding="utf-8", errors="ignore") as file: + words = [w.strip() for w in file if w.strip().isalpha()] + except OSError: + words = [] + if not words: + print( + f"WARNING: word list '{path}' not found or empty; " + f"falling back to built-in lorem ipsum words. " + f"Override with --dict .", + file=sys.stderr, + ) + return list(FALLBACK_WORDS) + return words + + +def random_subject(): + return " ".join(random.choices(WORDS, k=random.randint(3, 10))) + + +def random_body(target_size): + lines = [] + total = 0 + line = "" + while total < target_size: + word = random.choice(WORDS) + if line and len(line) + 1 + len(word) > LINE_WIDTH: + lines.append(line) + total += len(line) + 2 + line = word + elif line: + line = f"{line} {word}" + else: + line = word + if line: + lines.append(line) + return "\r\n".join(lines) + "\r\n" + + +def quote_periods(data): + return DOT_LINE.sub(b"..", data) + + +def build_headers(sender, recipient): + return ( + f"From: {sender}\r\n" + f"To: {recipient}\r\n" + f"Subject: {random_subject()}\r\n" + f"Date: {formatdate(localtime=True)}\r\n" + f"Message-ID: {make_msgid(domain='stress.test')}\r\n" + f"MIME-Version: 1.0\r\n" + f"Content-Type: text/plain; charset=us-ascii\r\n" + f"\r\n" + ).encode("ascii", "replace") + + +def build_body(size): + body = quote_periods(random_body(size).encode("ascii", "replace")) + if not body.endswith(b"\r\n"): + body += b"\r\n" + return body + + +class MemoryStore: + backend = "memory" + + def __init__(self): + self._items = [] + + def add(self, data): + self._items.append(data) + + def get(self, index): + return self._items[index] + + def __len__(self): + return len(self._items) + + def cleanup(self): + self._items = [] + + +class DiskStore: + def __init__(self, root): + self._dir = tempfile.mkdtemp(prefix="smtp_stress_", dir=root) + self.backend = self._dir + self._paths = [] + + def add(self, data): + path = os.path.join(self._dir, f"msg_{len(self._paths):09d}.eml") + with open(path, "wb") as handle: + handle.write(data) + self._paths.append(path) + + def get(self, index): + with open(self._paths[index], "rb") as handle: + return handle.read() + + def __len__(self): + return len(self._paths) + + def cleanup(self): + shutil.rmtree(self._dir, ignore_errors=True) + + +def build_body_store(ctx): + count = min(ctx.pool_size, ctx.messages) + if ctx.spool_dir is not None: + store = DiskStore(ctx.spool_dir) + else: + store = MemoryStore() + for _ in range(count): + if ctx.fixed_size is not None: + size = ctx.fixed_size + else: + size = random.randint(ctx.min_size, ctx.max_size) + store.add(build_body(size)) + return store + + +def make_tls_context(): + context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + return context + + +class Stats: + def __init__(self): + self._lock = threading.Lock() + self.latencies = [] + self.errors = 0 + self.bytes = 0 + + def record(self, latency_ms, num_bytes): + with self._lock: + self.latencies.append(latency_ms) + self.bytes += num_bytes + + def record_error(self): + with self._lock: + self.errors += 1 + + def snapshot(self): + with self._lock: + return list(self.latencies), self.errors, self.bytes + + +class Counter: + def __init__(self, total): + self._lock = threading.Lock() + self._remaining = total + + def claim(self): + with self._lock: + if self._remaining <= 0: + return False + self._remaining -= 1 + return True + + +def read_recipients(file_path): + recipients = [] + try: + with open(file_path, "r") as file: + for line in file: + line = line.strip() + if not line: + continue + recipients.append(line.split(":", 1)[0]) + except OSError as e: + raise SystemExit(f"Could not read recipients from '{file_path}': {e}") + if not recipients: + raise SystemExit(f"No recipients found in '{file_path}'.") + return recipients + + +def connect(ctx): + server = smtplib.SMTP(ctx.host, ctx.port, timeout=ctx.timeout) + server.ehlo() + if ctx.starttls: + if not server.has_extn("starttls"): + server.quit() + raise SmtpError("server does not advertise STARTTLS") + server.starttls(context=ctx.tls_context) + server.ehlo() + return server + + +def send_one(server, sender, recipient, header, body): + code, resp = server.mail(sender) + if code != 250: + server.rset() + raise SmtpError(f"MAIL FROM rejected: {code} {resp!r}") + code, resp = server.rcpt(recipient) + if code not in (250, 251): + server.rset() + raise SmtpError(f"RCPT TO rejected: {code} {resp!r}") + code, resp = server.docmd("DATA") + if code != 354: + raise SmtpError(f"DATA rejected: {code} {resp!r}") + server.send(header) + server.send(body) + start = time.monotonic() + server.send(b".\r\n") + code, resp = server.getreply() + elapsed_ms = (time.monotonic() - start) * 1000 + if code != 250: + raise SmtpError(f"message rejected: {code} {resp!r}") + return elapsed_ms + + +def worker(ctx, counter, recipients, stats, store, logger): + pool_len = len(store) + while not STOP_EVENT.is_set() and counter.claim(): + server = None + try: + server = connect(ctx) + recipient = random.choice(recipients) + body = store.get(random.randrange(pool_len)) + header = build_headers(ctx.sender, recipient) + elapsed_ms = send_one(server, ctx.sender, recipient, header, body) + num_bytes = len(header) + len(body) + stats.record(elapsed_ms, num_bytes) + if logger.enabled: + logger.log(f"OK {elapsed_ms:9.2f}ms {num_bytes:>9}B -> {recipient}") + except (SmtpError, smtplib.SMTPException, OSError) as e: + stats.record_error() + if logger.enabled: + logger.log(f"ERR {e}") + finally: + if server is not None: + try: + server.quit() + except Exception: + try: + server.close() + except Exception: + pass + + +def percentile(sorted_values, pct): + if not sorted_values: + return 0.0 + if len(sorted_values) == 1: + return sorted_values[0] + rank = (len(sorted_values) - 1) * (pct / 100.0) + low = math.floor(rank) + high = math.ceil(rank) + if low == high: + return sorted_values[int(rank)] + return sorted_values[low] * (high - rank) + sorted_values[high] * (rank - low) + + +def stddev(values, mean): + if len(values) < 2: + return 0.0 + variance = sum((v - mean) ** 2 for v in values) / (len(values) - 1) + return math.sqrt(variance) + + +def print_report( + latencies, + errors, + total_bytes, + send_seconds, + gen_seconds, + pool_count, + workers, + storage, + report_header=None, +): + count = len(latencies) + mb = total_bytes / (1024 * 1024) + throughput = count / send_seconds if send_seconds > 0 else 0.0 + mb_per_sec = mb / send_seconds if send_seconds > 0 else 0.0 + + line = "-" * 60 + print() + if report_header: + print(report_header) + print(line) + print("SMTP ingestion stress test report") + print(line) + print(f"{'Workers':<26}{workers}") + print(f"{'Message store':<26}{storage}") + print(f"{'Messages OK':<26}{count}") + print(f"{'Messages failed':<26}{errors}") + print(f"{'Bodies pregenerated':<26}{pool_count}") + print(f"{'Pool gen time (s)':<26}{gen_seconds:.2f}") + print(f"{'Send wall time (s)':<26}{send_seconds:.2f}") + print(f"{'Throughput (msg/s)':<26}{throughput:.2f}") + print(f"{'Data sent (MB)':<26}{mb:.2f}") + print(f"{'Data rate (MB/s)':<26}{mb_per_sec:.2f}") + print(line) + print("Ingestion time (DATA terminator to server OK), milliseconds") + print(line) + if count: + ordered = sorted(latencies) + mean = sum(ordered) / count + rows = [ + ("min", ordered[0]), + ("max", ordered[-1]), + ("avg", mean), + ("median", percentile(ordered, 50)), + ("p95", percentile(ordered, 95)), + ("p99", percentile(ordered, 99)), + ("stddev", stddev(ordered, mean)), + ] + for name, value in rows: + print(f"{name:<22}{value:.2f}") + else: + print("no messages were ingested") + print(line) + sys.stdout.flush() + + +class Context: + def __init__(self, args, messages): + self.host = args.host + self.port = args.port + self.threads = args.threads + self.sender = args.sender + self.starttls = not args.no_starttls + self.timeout = args.timeout + self.min_size = args.min_size + self.max_size = args.max_size + self.fixed_size = args.size + self.pool_size = args.pool_size + self.spool_dir = args.spool_dir + self.messages = messages + self.tls_context = make_tls_context() if self.starttls else None + + +def run_threads(ctx, recipients, message_count, store, logger): + stats = Stats() + counter = Counter(message_count) + threads = [ + threading.Thread( + target=worker, + args=(ctx, counter, recipients, stats, store, logger), + daemon=True, + ) + for _ in range(ctx.threads) + ] + for thread in threads: + thread.start() + try: + while any(t.is_alive() for t in threads): + for t in threads: + t.join(timeout=0.2) + except KeyboardInterrupt: + logger.log("Stopping...") + STOP_EVENT.set() + for t in threads: + t.join() + return stats + + +def child_main(args, recipients, message_count, barrier, result_queue): + global WORDS + WORDS = load_words(args.dict) + ctx = Context(args, message_count) + logger = AsyncLogger(not args.quiet) + store = build_body_store(ctx) + try: + logger.start() + try: + barrier.wait() + except threading.BrokenBarrierError: + result_queue.put(([], 0, 0, len(store))) + return + stats = run_threads(ctx, recipients, message_count, store, logger) + logger.stop() + latencies, errors, total_bytes = stats.snapshot() + result_queue.put((latencies, errors, total_bytes, len(store))) + finally: + store.cleanup() + + +def distribute(total, parts): + base, remainder = divmod(total, parts) + return [base + (1 if i < remainder else 0) for i in range(parts)] + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Concurrent SMTP ingestion stress test over port 25 with STARTTLS." + ) + parser.add_argument("--host", default=DEFAULT_HOST) + parser.add_argument("--port", type=int, default=DEFAULT_PORT) + parser.add_argument("--threads", type=int, default=DEFAULT_THREADS) + parser.add_argument( + "--processes", + type=int, + default=DEFAULT_PROCESSES, + help="Worker processes to spawn (each runs --threads threads). Scales past the GIL.", + ) + parser.add_argument( + "--messages", + type=int, + default=DEFAULT_MESSAGES, + help="Total messages to send, distributed across the threads.", + ) + parser.add_argument("--sender", default=DEFAULT_SENDER, help="Envelope MAIL FROM address.") + parser.add_argument("--users-file", default=DEFAULT_USERS_FILE) + parser.add_argument( + "--size", + type=int, + help="Fixed message body size in bytes; overrides --min-size/--max-size.", + ) + parser.add_argument("--min-size", type=int, default=DEFAULT_MIN_SIZE) + parser.add_argument("--max-size", type=int, default=DEFAULT_MAX_SIZE) + parser.add_argument( + "--pool-size", + type=int, + default=DEFAULT_POOL_SIZE, + help="Distinct message bodies pregenerated before timing (reused at random). " + "Each sent message gets a fresh unique Message-ID regardless of this.", + ) + parser.add_argument( + "--spool-dir", + nargs="?", + const=tempfile.gettempdir(), + default=None, + help="Spool pregenerated messages to disk instead of memory. " + "With no value uses the system temp dir; pass a path to override.", + ) + parser.add_argument("--dict", default=DEFAULT_DICTIONARY) + parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT) + parser.add_argument( + "--header", + help="Optional header line printed at the top of the final report on stdout.", + ) + parser.add_argument( + "--no-starttls", + action="store_true", + help="Send over plaintext instead of upgrading with STARTTLS.", + ) + parser.add_argument("--quiet", action="store_true") + args = parser.parse_args() + if args.threads < 1: + parser.error("--threads must be at least 1") + if args.processes < 1: + parser.error("--processes must be at least 1") + if args.messages < 1: + parser.error("--messages must be at least 1") + if args.pool_size < 1: + parser.error("--pool-size must be at least 1") + if args.size is None and args.min_size > args.max_size: + parser.error("--min-size must not exceed --max-size") + return args + + +def run_single_process(args, recipients): + global WORDS + WORDS = load_words(args.dict) + ctx = Context(args, args.messages) + logger = AsyncLogger(not args.quiet) + + gen_start = time.monotonic() + store = build_body_store(ctx) + gen_seconds = time.monotonic() - gen_start + + try: + logger.start() + start = time.monotonic() + stats = run_threads(ctx, recipients, args.messages, store, logger) + send_seconds = time.monotonic() - start + logger.stop() + latencies, errors, total_bytes = stats.snapshot() + workers = f"1 process x {args.threads} threads" + print_report( + latencies, errors, total_bytes, send_seconds, gen_seconds, len(store), + workers, store.backend, args.header, + ) + finally: + store.cleanup() + + +def run_multi_process(args, recipients): + nproc = min(args.processes, args.messages) + shares = distribute(args.messages, nproc) + barrier = multiprocessing.Barrier(nproc + 1) + result_queue = multiprocessing.Queue() + procs = [] + for share in shares: + proc = multiprocessing.Process( + target=child_main, + args=(args, recipients, share, barrier, result_queue), + daemon=False, + ) + proc.start() + procs.append(proc) + + gen_start = time.monotonic() + interrupted = False + try: + barrier.wait() + except KeyboardInterrupt: + interrupted = True + barrier.abort() + gen_seconds = time.monotonic() - gen_start + + start = time.monotonic() + results = [] + try: + for _ in procs: + results.append(result_queue.get()) + except KeyboardInterrupt: + interrupted = True + for proc in procs: + proc.terminate() + send_seconds = time.monotonic() - start + + for proc in procs: + proc.join() + + latencies = [] + errors = 0 + total_bytes = 0 + pool_count = 0 + for lat, err, nbytes, pool_len in results: + latencies.extend(lat) + errors += err + total_bytes += nbytes + pool_count += pool_len + + if interrupted: + print("Interrupted.", file=sys.stderr, flush=True) + workers = f"{nproc} processes x {args.threads} threads" + storage = "memory" if args.spool_dir is None else f"disk ({args.spool_dir})" + print_report( + latencies, errors, total_bytes, send_seconds, gen_seconds, pool_count, + workers, storage, args.header, + ) + + +def main(): + args = parse_args() + recipients = read_recipients(args.users_file) + if args.processes == 1: + run_single_process(args, recipients) + else: + run_multi_process(args, recipients) + + +if __name__ == "__main__": + main() diff --git a/tests/resources/scripts/stress_test.py b/tests/resources/scripts/stress_test.py index 02b4cc7b..3ad608b5 100644 --- a/tests/resources/scripts/stress_test.py +++ b/tests/resources/scripts/stress_test.py @@ -1,132 +1,490 @@ -import smtplib +import argparse import imaplib +import math +import os +import random +import smtplib import ssl import threading -import random import time -import string +from collections import defaultdict from email.mime.text import MIMEText -smtp_server = "127.0.0.1" -smtp_port = 465 -imap_server = "127.0.0.1" -imap_port = 993 -num_threads = 5 -runs = 10 # Set to None for infinite loop +DEFAULT_SMTP_SERVER = "127.0.0.1" +DEFAULT_SMTP_PORT = 465 +DEFAULT_IMAP_SERVER = "127.0.0.1" +DEFAULT_IMAP_PORT = 993 +DEFAULT_THREADS = 5 +DEFAULT_RUNS = 10 +DEFAULT_DICTIONARY = "/usr/share/dict/words" + +FALLBACK_WORDS = ( + "lorem ipsum dolor sit amet consectetur adipiscing elit sed do eiusmod " + "tempor incididunt ut labore et dolore magna aliqua enim ad minim veniam " + "quis nostrud exercitation ullamco laboris nisi aliquip ex ea commodo" +).split() + +WORDS = FALLBACK_WORDS + +SMTP_SEND = "SMTP SEND" +IMAP_APPEND = "IMAP APPEND" +IMAP_FETCH = "IMAP FETCH" +IMAP_DELETE = "IMAP DELETE" +ACTIONS = (SMTP_SEND, IMAP_APPEND, IMAP_FETCH, IMAP_DELETE) + + +class Stats: + def __init__(self): + self._lock = threading.Lock() + self._latencies = defaultdict(list) + self._errors = defaultdict(int) + self._skips = defaultdict(int) + self._bytes = defaultdict(int) + + def record(self, action, latency_ms, num_bytes=0): + with self._lock: + self._latencies[action].append(latency_ms) + self._bytes[action] += num_bytes + + def record_error(self, action): + with self._lock: + self._errors[action] += 1 + + def record_skip(self, action): + with self._lock: + self._skips[action] += 1 + + def snapshot(self): + with self._lock: + return ( + {k: list(v) for k, v in self._latencies.items()}, + dict(self._errors), + dict(self._skips), + dict(self._bytes), + ) + + +PRINT_LOCK = threading.Lock() +STOP_EVENT = threading.Event() + def read_credentials(file_path): + if not os.path.exists(file_path): + raise SystemExit( + f"Credentials file '{file_path}' not found. " + f"Run stress_test_prepare.py first to create users." + ) + credentials = [] with open(file_path, "r") as file: - credentials = [line.strip().split(':') for line in file if line.strip()] + for line in file: + line = line.strip() + if not line: + continue + parts = line.split(":", 1) + if len(parts) != 2: + continue + credentials.append((parts[0], parts[1])) + if not credentials: + raise SystemExit(f"No valid credentials found in '{file_path}'.") return credentials + def allow_invalid_certificates(): - # Create an SSL context context = ssl.create_default_context() context.check_hostname = False - context.verify_mode = ssl.CERT_NONE + context.verify_mode = ssl.CERT_NONE return context -def generate_random_string(min_size, max_size): - """Generates a random string of a size between min_size and max_size.""" - size = random.randint(min_size, max_size) - chars = string.ascii_letters + string.digits + ' ' - return ''.join(random.choice(chars) for _ in range(size)) -def generate_email(username, recipient): - """Generate random subject and content for email.""" - subject = generate_random_string(10, 100) # Random subject between 10 and 100 characters - content_size = random.randint(100, 1048576) # Random content size between 100 bytes and ~1MB - content = generate_random_string(content_size, content_size) +def load_words(path): + try: + with open(path, "r", encoding="utf-8", errors="ignore") as file: + words = [w.strip() for w in file if w.strip().isalpha()] + except OSError: + words = [] + if not words: + print( + f"WARNING: word list '{path}' not found or empty; " + f"falling back to built-in lorem ipsum words. " + f"Override with --dict ." + ) + return list(FALLBACK_WORDS) + return words + + +def random_words(min_size, max_size): + target = random.randint(min_size, max_size) + parts = [] + length = 0 + while length < target: + for word in random.choices(WORDS, k=64): + parts.append(word) + length += len(word) + 1 + if length >= target: + break + return " ".join(parts) + + +def generate_email(username, recipient, max_content_size): + subject = random_words(10, 100) + content = random_words(100, max_content_size) message = MIMEText(content) - message['Subject'] = subject - message['From'] = username - message['To'] = recipient + message["Subject"] = subject + message["From"] = username + message["To"] = recipient return message.as_string() -def smtp_send_message(username, password, recipient): + +def log_ok(stats, action, latency_ms, detail="", num_bytes=0, verbose=True): + stats.record(action, latency_ms, num_bytes) + if verbose: + with PRINT_LOCK: + print(f"OK {latency_ms:9.2f}ms {action} {detail}") + + +def log_err(stats, action, error, verbose=True): + stats.record_error(action) + if verbose: + with PRINT_LOCK: + print(f"ERR {action} {error}") + + +def smtp_send_message(ctx, username, password, recipient): try: - with smtplib.SMTP_SSL(smtp_server, smtp_port, context=allow_invalid_certificates()) as server: + with smtplib.SMTP_SSL( + ctx.smtp_server, ctx.smtp_port, context=allow_invalid_certificates() + ) as server: server.login(username, password) - start_time = time.time() - server.sendmail(username, recipient, generate_email(username, recipient)) - elapsed_time_ms = (time.time() - start_time) * 1000 - print(f"OK {elapsed_time_ms} SMTP {username} -> {recipient}") + payload = generate_email(username, recipient, ctx.max_content_size) + start_time = time.monotonic() + server.sendmail(username, recipient, payload) + elapsed_ms = (time.monotonic() - start_time) * 1000 + log_ok( + ctx.stats, + SMTP_SEND, + elapsed_ms, + f"{username} -> {recipient}", + len(payload), + ctx.verbose, + ) except Exception as e: - print(f"ERR SMTP {e}") + log_err(ctx.stats, SMTP_SEND, e, ctx.verbose) -def imap_append_message(username, password, recipient): + +def imap_append_message(ctx, username, password, recipient): try: - with imaplib.IMAP4_SSL(imap_server, imap_port, ssl_context=allow_invalid_certificates()) as imap: + with imaplib.IMAP4_SSL( + ctx.imap_server, ctx.imap_port, ssl_context=allow_invalid_certificates() + ) as imap: imap.login(username, password) - start_time = time.time() - imap.append('INBOX', None, imaplib.Time2Internaldate(time.time()), generate_email(username, recipient).encode('utf-8')) - elapsed_time_ms = (time.time() - start_time) * 1000 - print(f"OK {elapsed_time_ms} IMAP APPEND {username}") + payload = generate_email(username, recipient, ctx.max_content_size).encode("utf-8") + start_time = time.monotonic() + imap.append("INBOX", None, imaplib.Time2Internaldate(time.time()), payload) + elapsed_ms = (time.monotonic() - start_time) * 1000 + log_ok(ctx.stats, IMAP_APPEND, elapsed_ms, username, len(payload), ctx.verbose) except Exception as e: - print(f"ERR IMAP {e}") + log_err(ctx.stats, IMAP_APPEND, e, ctx.verbose) -def imap_list_fetch(username, password): + +def imap_list_fetch(ctx, username, password, recipient): try: - with imaplib.IMAP4_SSL(imap_server, imap_port, ssl_context=allow_invalid_certificates()) as imap: + with imaplib.IMAP4_SSL( + ctx.imap_server, ctx.imap_port, ssl_context=allow_invalid_certificates() + ) as imap: imap.login(username, password) - imap.select('INBOX') - start_time = time.time() - typ, data = imap.search(None, 'ALL') - if data[0]: + imap.select("INBOX") + start_time = time.monotonic() + typ, data = imap.search(None, "ALL") + if data and data[0]: messages = data[0].split() random_msg_num = random.choice(messages) - typ, msg_data = imap.fetch(random_msg_num, '(RFC822)') - elapsed_time_ms = (time.time() - start_time) * 1000 - print(f"OK {elapsed_time_ms} IMAP FETCH {username} {random_msg_num}") + imap.fetch(random_msg_num, "(RFC822)") + elapsed_ms = (time.monotonic() - start_time) * 1000 + log_ok( + ctx.stats, + IMAP_FETCH, + elapsed_ms, + f"{username} {random_msg_num.decode()}", + verbose=ctx.verbose, + ) + else: + ctx.stats.record_skip(IMAP_FETCH) except Exception as e: - print(f"ERR IMAP {e}") + log_err(ctx.stats, IMAP_FETCH, e, ctx.verbose) -def imap_delete_message(username, password): + +def imap_delete_message(ctx, username, password, recipient): try: - with imaplib.IMAP4_SSL(imap_server, imap_port, ssl_context=allow_invalid_certificates()) as imap: + with imaplib.IMAP4_SSL( + ctx.imap_server, ctx.imap_port, ssl_context=allow_invalid_certificates() + ) as imap: imap.login(username, password) - imap.select('INBOX') - start_time = time.time() - typ, data = imap.search(None, 'ALL') - if data[0]: + imap.select("INBOX") + start_time = time.monotonic() + typ, data = imap.search(None, "ALL") + if data and data[0]: messages = data[0].split() random_msg_num = random.choice(messages) - imap.store(random_msg_num, '+FLAGS', '\\Deleted') + imap.store(random_msg_num, "+FLAGS", "\\Deleted") imap.expunge() - elapsed_time_ms = (time.time() - start_time) * 1000 - print(f"OK {elapsed_time_ms} IMAP DELETE {username} {random_msg_num}") + elapsed_ms = (time.monotonic() - start_time) * 1000 + log_ok( + ctx.stats, + IMAP_DELETE, + elapsed_ms, + f"{username} {random_msg_num.decode()}", + verbose=ctx.verbose, + ) + else: + ctx.stats.record_skip(IMAP_DELETE) except Exception as e: - print(f"ERR IMAP {e}") + log_err(ctx.stats, IMAP_DELETE, e, ctx.verbose) -def perform_random_action(credentials): + +ACTION_FUNCS = ( + smtp_send_message, + imap_append_message, + imap_list_fetch, + imap_delete_message, +) + + +def pick_recipient(credentials, sender): + if len(credentials) == 1: + return credentials[0][0] + while True: + recipient = random.choice(credentials)[0] + if recipient != sender: + return recipient + + +def perform_random_action(ctx, credentials): username, password = random.choice(credentials) - recipient, _ = random.choice(credentials) - action = random.choice([smtp_send_message, imap_append_message, imap_list_fetch, imap_delete_message]) - - if action == smtp_send_message or action == imap_append_message: - action(username, password, recipient) - else: - action(username, password) + recipient = pick_recipient(credentials, username) + action = random.choice(ACTION_FUNCS) + action(ctx, username, password, recipient) + + +def thread_function(ctx, credentials): + count = 0 + while not STOP_EVENT.is_set(): + if ctx.runs is not None and count >= ctx.runs: + break + perform_random_action(ctx, credentials) + count += 1 + + +def percentile(sorted_values, pct): + if not sorted_values: + return 0.0 + if len(sorted_values) == 1: + return sorted_values[0] + rank = (len(sorted_values) - 1) * (pct / 100.0) + low = math.floor(rank) + high = math.ceil(rank) + if low == high: + return sorted_values[int(rank)] + return sorted_values[low] * (high - rank) + sorted_values[high] * (rank - low) + + +def stddev(values, mean): + if len(values) < 2: + return 0.0 + variance = sum((v - mean) ** 2 for v in values) / (len(values) - 1) + return math.sqrt(variance) + + +def summarize(action, latencies, errors, skips, total_bytes, wall_seconds): + count = len(latencies) + summary = { + "action": action, + "count": count, + "errors": errors, + "skips": skips, + "mb": total_bytes / (1024 * 1024), + } + if count == 0: + for key in ("min", "max", "avg", "median", "p95", "p99", "stddev", "ops"): + summary[key] = 0.0 + return summary + ordered = sorted(latencies) + mean = sum(ordered) / count + summary.update( + { + "min": ordered[0], + "max": ordered[-1], + "avg": mean, + "median": percentile(ordered, 50), + "p95": percentile(ordered, 95), + "p99": percentile(ordered, 99), + "stddev": stddev(ordered, mean), + "ops": count / wall_seconds if wall_seconds > 0 else 0.0, + } + ) + return summary + + +def print_report(stats, wall_seconds): + latencies, errors, skips, byte_counts = stats.snapshot() + + rows = [] + all_latencies = [] + total_errors = 0 + total_skips = 0 + total_bytes = 0 + for action in ACTIONS: + action_latencies = latencies.get(action, []) + all_latencies.extend(action_latencies) + total_errors += errors.get(action, 0) + total_skips += skips.get(action, 0) + total_bytes += byte_counts.get(action, 0) + rows.append( + summarize( + action, + action_latencies, + errors.get(action, 0), + skips.get(action, 0), + byte_counts.get(action, 0), + wall_seconds, + ) + ) + rows.append( + summarize( + "TOTAL", + all_latencies, + total_errors, + total_skips, + total_bytes, + wall_seconds, + ) + ) + + headers = [ + "Action", + "OK", + "Err", + "Skip", + "Min ms", + "Max ms", + "Avg ms", + "Med ms", + "P95 ms", + "P99 ms", + "Std ms", + "Ops/s", + "MB", + ] + fmt = "{:<12} {:>7} {:>5} {:>5} {:>9} {:>9} {:>9} {:>9} {:>9} {:>9} {:>9} {:>8} {:>9}" + line = "-" * 122 + + print() + print(line) + print(f"Stress test report (wall time: {wall_seconds:.2f}s)") + print(line) + print(fmt.format(*headers)) + print(line) + for r in rows: + if r["action"] == "TOTAL": + print(line) + print( + fmt.format( + r["action"], + r["count"], + r["errors"], + r["skips"], + f"{r['min']:.2f}", + f"{r['max']:.2f}", + f"{r['avg']:.2f}", + f"{r['median']:.2f}", + f"{r['p95']:.2f}", + f"{r['p99']:.2f}", + f"{r['stddev']:.2f}", + f"{r['ops']:.1f}", + f"{r['mb']:.1f}", + ) + ) + print(line) + + +class Context: + def __init__(self, args, stats): + self.smtp_server = args.smtp_server + self.smtp_port = args.smtp_port + self.imap_server = args.imap_server + self.imap_port = args.imap_port + self.runs = args.runs + self.max_content_size = args.max_content_size + self.verbose = not args.quiet + self.stats = stats + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Concurrent SMTP/IMAP stress test for Stalwart." + ) + parser.add_argument("--smtp-server", default=DEFAULT_SMTP_SERVER) + parser.add_argument("--smtp-port", type=int, default=DEFAULT_SMTP_PORT) + parser.add_argument("--imap-server", default=DEFAULT_IMAP_SERVER) + parser.add_argument("--imap-port", type=int, default=DEFAULT_IMAP_PORT) + parser.add_argument("--threads", type=int, default=DEFAULT_THREADS) + parser.add_argument( + "--runs", + type=int, + default=DEFAULT_RUNS, + help="Actions per thread. Use 0 for an infinite loop (stop with Ctrl-C).", + ) + parser.add_argument("--credentials", default="users.txt") + parser.add_argument( + "--max-content-size", + type=int, + default=1048576, + help="Maximum random message body size in bytes.", + ) + parser.add_argument( + "--dict", + default=DEFAULT_DICTIONARY, + help="Word list used to generate message text, one word per line.", + ) + parser.add_argument( + "--quiet", + action="store_true", + help="Suppress per-operation logging; print only the final report.", + ) + return parser.parse_args() -def thread_function(credentials): - if runs: - for _ in range(runs): - perform_random_action(credentials) - else: - while True: - perform_random_action(credentials) def main(): - credentials = read_credentials("users.txt") - threads = [] + global WORDS + args = parse_args() + if args.runs == 0: + args.runs = None + WORDS = load_words(args.dict) + credentials = read_credentials(args.credentials) + stats = Stats() + ctx = Context(args, stats) - for _ in range(num_threads): - thread = threading.Thread(target=thread_function, args=(credentials,)) - threads.append(thread) - thread.start() - + threads = [ + threading.Thread(target=thread_function, args=(ctx, credentials), daemon=True) + for _ in range(args.threads) + ] + + start = time.monotonic() for thread in threads: - thread.join() + thread.start() -if __name__ == '__main__': + try: + while any(t.is_alive() for t in threads): + for t in threads: + t.join(timeout=0.2) + except KeyboardInterrupt: + print("\nStopping...") + STOP_EVENT.set() + for t in threads: + t.join() + + wall_seconds = time.monotonic() - start + print_report(stats, wall_seconds) + + +if __name__ == "__main__": main() diff --git a/tests/resources/scripts/stress_test_prepare.py b/tests/resources/scripts/stress_test_prepare.py index 4af37735..426a5eec 100644 --- a/tests/resources/scripts/stress_test_prepare.py +++ b/tests/resources/scripts/stress_test_prepare.py @@ -1,57 +1,244 @@ -import requests +import argparse +import base64 +import json import random +import ssl import string -import urllib3 +import urllib.error +import urllib.request -# Configuration Variables -HOSTNAME = '127.0.0.1' # Replace with the actual hostname -DOMAIN = 'test.org' # Replace with your domain name -USERNAME = 'admin' # Basic auth username -PASSWORD = 'secret' # Basic auth password -NUM_USERS = 1000 # Number of test user accounts to create +CORE = "urn:ietf:params:jmap:core" +STALWART = "urn:stalwart:jmap" +USING = [CORE, STALWART] -# Suppress InsecureRequestWarning -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +DEFAULT_BASE_URL = "https://127.0.0.1" +DEFAULT_NUM_USERS = 1000 +DEFAULT_OUTPUT = "users.txt" +DEFAULT_PASSWORD_LENGTH = 16 +DEFAULT_PREFIX = "test" -# Generate SHA512 password hash -def generate_password(): - return ''.join(random.choices(string.ascii_letters + string.digits, k=10)) +CREATE_RETRIES = 5 -# Create Domain -def create_domain(): - url = f"https://{HOSTNAME}/api/domain/{DOMAIN}" - response = requests.post(url, auth=(USERNAME, PASSWORD), verify=False) - if response.status_code == 200: - print(f"Domain '{DOMAIN}' created successfully.") - else: - print(f"Failed to create domain '{DOMAIN}'. Status Code: {response.status_code}") - print(response.text) -# Create User Accounts -def create_user_accounts(): - with open('users.txt', 'w') as file: - for i in range(1, NUM_USERS + 1): - username = f"test{i}@{DOMAIN}" - password = generate_password() - data = { - "type": "individual", - "name": username, - "secrets": [password], - "emails": [username], - "description": f"Tester {i}" +class AccountError(Exception): + pass + + +def generate_password(length): + return "".join(random.choices(string.ascii_letters + string.digits, k=length)) + + +def primary_account(session): + accounts = session.get("primaryAccounts") or {} + if "urn:ietf:params:jmap:mail" in accounts: + return accounts["urn:ietf:params:jmap:mail"] + all_accounts = session.get("accounts") or {} + return next(iter(all_accounts), None) + + +def build_opener(verify): + context = ssl.create_default_context() + if not verify: + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + handler = urllib.request.HTTPSHandler(context=context) + return urllib.request.build_opener(handler) + + +class JmapClient: + def __init__(self, base_url, auth_header, verify): + self.base_url = base_url.rstrip("/") + self.auth_header = auth_header + self.opener = build_opener(verify) + self.api_url = None + self.account_id = None + self._discover() + + def _http(self, url, method, body=None): + headers = {"Authorization": self.auth_header} + data = None + if body is not None: + data = json.dumps(body).encode("utf-8") + headers["Content-Type"] = "application/json" + request = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with self.opener.open(request, timeout=60) as response: + return response.status, response.read().decode("utf-8", "replace") + except urllib.error.HTTPError as e: + return e.code, e.read().decode("utf-8", "replace") + except urllib.error.URLError as e: + raise SystemExit(f"Request to {url} failed: {e.reason}") + + def _discover(self): + if "/.well-known/jmap" in self.base_url: + candidates = [self.base_url] + else: + candidates = [self.base_url, f"{self.base_url}/.well-known/jmap"] + last = "" + for url in candidates: + status, text = self._http(url, "GET") + if status == 200: + try: + data = json.loads(text) + except ValueError: + last = f"{url}: invalid JSON" + continue + if data.get("apiUrl") and data.get("accounts"): + self.api_url = data["apiUrl"] + self.account_id = primary_account(data) + if not self.account_id: + raise SystemExit(f"JMAP session at {url} has no accounts.") + return + last = f"{url}: status {status}" + raise SystemExit(f"Could not discover JMAP session ({last}).") + + def request(self, method_calls): + body = {"using": USING, "methodCalls": method_calls} + status, text = self._http(self.api_url, "POST", body) + if status != 200: + raise SystemExit(f"JMAP request failed: status {status}: {text}") + return json.loads(text) + + def call(self, method, args): + args = dict(args) + args["accountId"] = self.account_id + parsed = self.request([[method, args, "c0"]]) + responses = parsed.get("methodResponses") or [] + if not responses: + raise SystemExit(f"{method}: empty methodResponses") + name, payload = responses[0][0], responses[0][1] + if name == "error": + raise SystemExit(f"{method} error: {payload}") + return payload + + def domain_id(self, name): + calls = [ + [ + "x:Domain/query", + {"accountId": self.account_id, "filter": {"name": name}}, + "q", + ], + [ + "x:Domain/get", + { + "accountId": self.account_id, + "#ids": { + "resultOf": "q", + "name": "x:Domain/query", + "path": "/ids", + }, + "properties": ["id", "name"], + }, + "g", + ], + ] + parsed = self.request(calls) + responses = parsed.get("methodResponses") or [] + if len(responses) < 2: + raise SystemExit(f"Domain lookup failed: {parsed}") + get = responses[1][1] + for entry in get.get("list") or []: + if entry.get("name") == name: + return entry.get("id") + return None + + def create_account(self, localpart, domain_id, password): + create = { + "a": { + "@type": "User", + "name": localpart, + "domainId": domain_id, + "credentials": {"0": {"@type": "Password", "secret": password}}, + "encryptionAtRest": {"@type": "Disabled"}, + "permissions": {"@type": "Inherit"}, + "roles": {"@type": "User"}, + "locale": "en_US", } - url = f"https://{HOSTNAME}/api/principal" - response = requests.post(url, json=data, auth=(USERNAME, PASSWORD), verify=False) - if response.status_code == 200: - file.write(f"{username}:{password}\n") - print(f"User account '{username}' created successfully.") - else: - print(f"Failed to create user account '{username}'. Status Code: {response.status_code}") - print(response.text) + } + response = self.call("x:Account/set", {"create": create}) + not_created = response.get("notCreated") or {} + if not_created: + raise AccountError(f"{not_created.get('a', not_created)}") + created = (response.get("created") or {}).get("a") or {} + return created.get("id") + + def invalidate_caches(self): + self.call( + "x:Action/set", + {"create": {"c": {"@type": "InvalidateCaches"}}}, + ) + + +def build_auth_header(args): + if args.token: + return f"Bearer {args.token}" + raw = f"{args.user}:{args.password}".encode("utf-8") + return "Basic " + base64.b64encode(raw).decode("ascii") + + +def parse_args(): + parser = argparse.ArgumentParser( + description="Provision test accounts for the stress test using the JMAP API." + ) + parser.add_argument("--base-url", default=DEFAULT_BASE_URL) + parser.add_argument( + "--domain", + required=True, + help="Existing domain name; resolved to a domain id via JMAP.", + ) + parser.add_argument("--token", help="OAuth bearer token for the JMAP API.") + parser.add_argument("--user", help="Basic auth username for the JMAP API.") + parser.add_argument("--password", help="Basic auth password for the JMAP API.") + parser.add_argument("--num-users", type=int, default=DEFAULT_NUM_USERS) + parser.add_argument("--prefix", default=DEFAULT_PREFIX) + parser.add_argument("--password-length", type=int, default=DEFAULT_PASSWORD_LENGTH) + parser.add_argument("--output", default=DEFAULT_OUTPUT) + parser.add_argument( + "--verify", + action="store_true", + help="Verify TLS certificates (disabled by default for self-signed servers).", + ) + args = parser.parse_args() + if not args.token and not (args.user and args.password): + parser.error("provide either --token or both --user and --password") + return args + def main(): - create_domain() - create_user_accounts() + args = parse_args() + client = JmapClient(args.base_url, build_auth_header(args), args.verify) + domain_id = client.domain_id(args.domain) + if not domain_id: + raise SystemExit(f"Domain '{args.domain}' not found via JMAP.") + + created = 0 + failed = 0 + with open(args.output, "w") as file: + for i in range(1, args.num_users + 1): + localpart = f"{args.prefix}{i}" + email = f"{localpart}@{args.domain}" + password = None + last_error = None + for _ in range(CREATE_RETRIES): + password = generate_password(args.password_length) + try: + client.create_account(localpart, domain_id, password) + last_error = None + break + except AccountError as e: + last_error = e + if last_error is not None: + failed += 1 + print(f"FAIL {email}: {last_error}") + continue + file.write(f"{email}:{password}\n") + file.flush() + created += 1 + print(f"OK {email}") + + client.invalidate_caches() + print(f"\nCreated {created} accounts ({failed} failed). Written to {args.output}.") + if __name__ == "__main__": main() diff --git a/tests/src/jmap/mail/acl.rs b/tests/src/jmap/mail/acl.rs index 99404301..0ab4b4c0 100644 --- a/tests/src/jmap/mail/acl.rs +++ b/tests/src/jmap/mail/acl.rs @@ -183,6 +183,25 @@ pub async fn test(test: &TestServer) { .is_none() ); + // Email/changes must not leak ids of emails in folders John cannot read + let jane_inbox_email = email_ids.get("jane").unwrap().first().unwrap().clone(); + let jane_trash_email = email_ids.get("jane").unwrap().last().unwrap().clone(); + let changed_ids = john_client + .set_default_account_id(jane.id_string()) + .email_changes("n", None) + .await + .unwrap() + .created() + .to_vec(); + assert!( + changed_ids.contains(&jane_inbox_email), + "Email/changes should report the shared Inbox email" + ); + assert!( + !changed_ids.contains(&jane_trash_email), + "Email/changes leaked the id of a non-shared Trash email" + ); + // John should only be able to copy blobs he has access to let blob_id = jane_client .email_get(