From e451f037c873db5a7c9ce8997299d11102722f63 Mon Sep 17 00:00:00 2001 From: Maurus Decimus <11444311+mdecimus@users.noreply.github.com> Date: Fri, 27 Mar 2026 16:49:26 +0100 Subject: [PATCH] Registry testing - all tests passing --- crates/common/src/cache/directory.rs | 47 +- crates/common/src/cache/invalidate.rs | 9 +- crates/common/src/network/mta.rs | 49 +- crates/directory/src/backend/oidc/lookup.rs | 125 ++-- crates/jmap/src/api/request.rs | 24 +- crates/jmap/src/registry/set.rs | 1 + crates/registry/src/utils/account.rs | 7 + crates/services/src/broadcast/mod.rs | 9 +- tests/Cargo.toml | 2 +- tests/docker/ldap/50-users.ldif | 2 + tests/src/cluster/broadcast.rs | 290 ++++++-- tests/src/cluster/mod.rs | 359 ---------- tests/src/cluster/stress.rs | 51 +- tests/src/directory/integration.rs | 197 ++++++ tests/src/directory/ldap.rs | 59 +- tests/src/directory/mod.rs | 15 +- tests/src/directory/oidc.rs | 3 +- tests/src/directory/sql.rs | 698 ++++---------------- tests/src/directory/synchronization.rs | 271 ++++++++ tests/src/imap/idle.rs | 2 +- tests/src/imap/mod.rs | 2 +- tests/src/lib.rs | 3 - tests/src/utils/cleanup.rs | 5 +- tests/src/utils/jmap.rs | 42 +- tests/src/utils/server.rs | 137 ++-- tests/src/utils/storage.rs | 4 +- 26 files changed, 1174 insertions(+), 1239 deletions(-) create mode 100644 tests/src/directory/integration.rs create mode 100644 tests/src/directory/synchronization.rs diff --git a/crates/common/src/cache/directory.rs b/crates/common/src/cache/directory.rs index 21815f72..1462698d 100644 --- a/crates/common/src/cache/directory.rs +++ b/crates/common/src/cache/directory.rs @@ -4,9 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::sync::Arc; - -use crate::{Server, auth::DomainCache}; +use crate::{Server, auth::DomainCache, ipc::BroadcastEvent}; use registry::{ schema::{ prelude::{Object, ObjectType}, @@ -17,21 +15,23 @@ use registry::{ }, types::{datetime::UTCDateTime, id::ObjectId, list::List}, }; +use std::sync::Arc; use store::registry::write::{RegistryWrite, RegistryWriteResult}; use trc::AddContext; use types::id::Id; -pub(crate) struct AccountWithId { +pub struct AccountWithId { pub id: u32, pub account: Account, } impl Server { - pub(crate) async fn synchronize_account( + pub async fn synchronize_account( &self, account: directory::Account, ) -> trc::Result { let (local, domain) = self.validate_address(&account.email).await?; + match self .account_id_from_parts(local, domain.id) .await @@ -159,6 +159,8 @@ impl Server { enabled: true, description: None, }); + + self.invalidate_local_negative_account_cache(local, alias_domain.id); } } let mut member_group_ids = Vec::with_capacity(account.groups.len()); @@ -182,10 +184,13 @@ impl Server { member_group_ids: member_group_ids.into(), member_tenant_id: domain.id_tenant.map(Id::from), roles: UserRoles::User, - credentials: List::from_iter([Credential::Password(PasswordCredential { - secret: account.secret.unwrap_or_default(), - ..Default::default() - })]), + credentials: List::from_iter(account.secret.map(|secret| { + Credential::Password(PasswordCredential { + credential_id: 0u64.into(), + secret, + ..Default::default() + }) + })), ..Default::default() })); @@ -206,10 +211,16 @@ impl Server { .await .caused_by(trc::location!())? { - RegistryWriteResult::Success(id) => Ok(AccountWithId { - id: id.document_id(), - account: account.into(), - }), + RegistryWriteResult::Success(id) => { + self.invalidate_local_negative_account_cache(local, domain.id); + self.cluster_broadcast(BroadcastEvent::CacheInvalidateNegative) + .await; + + Ok(AccountWithId { + id: id.document_id(), + account: account.into(), + }) + } failure => Err(trc::AuthEvent::Error .into_err() .caused_by(trc::location!()) @@ -220,7 +231,7 @@ impl Server { } } - pub(crate) async fn synchronize_group(&self, group: directory::Group) -> trc::Result { + pub async fn synchronize_group(&self, group: directory::Group) -> trc::Result { let (local, domain) = self.validate_address(&group.email).await?; match self @@ -313,6 +324,8 @@ impl Server { enabled: true, description: None, }); + + self.invalidate_local_negative_account_cache(local, alias_domain.id); } } @@ -344,7 +357,11 @@ impl Server { .await .caused_by(trc::location!())? { - RegistryWriteResult::Success(id) => Ok(id.document_id()), + RegistryWriteResult::Success(id) => { + self.invalidate_local_negative_account_cache(local, domain.id); + + Ok(id.document_id()) + } failure => Err(trc::AuthEvent::Error .into_err() .caused_by(trc::location!()) diff --git a/crates/common/src/cache/invalidate.rs b/crates/common/src/cache/invalidate.rs index 6218fe98..f062337c 100644 --- a/crates/common/src/cache/invalidate.rs +++ b/crates/common/src/cache/invalidate.rs @@ -6,7 +6,7 @@ use crate::{ Server, - auth::EmailCache, + auth::{EmailAddressRef, EmailCache}, ipc::{BroadcastEvent, CacheInvalidation}, }; use ahash::AHashSet; @@ -283,6 +283,13 @@ impl Server { self.inner.cache.emails_negative.clear(); } + pub fn invalidate_local_negative_account_cache(&self, local_part: &str, domain_id: u32) { + self.inner + .cache + .emails_negative + .remove(&EmailAddressRef::new(local_part, domain_id)); + } + pub async fn invalidate_local_caches(&self, changes: &[CacheInvalidation]) { let cache = &self.inner.cache; diff --git a/crates/common/src/network/mta.rs b/crates/common/src/network/mta.rs index 5f642dd3..d7de05f4 100644 --- a/crates/common/src/network/mta.rs +++ b/crates/common/src/network/mta.rs @@ -103,13 +103,36 @@ impl Server { } // SPDX-SnippetEnd + // Obtain external directory, if configured + let directory = self + .get_directory_for_cached_domain(&domain) + .filter(|directory| directory.can_lookup_recipients()); + if let Some(directory) = directory { + let address = if local_part.as_ref() == local_part_orig { + Cow::Borrowed(rcpt) + } else { + Cow::Owned(format!("{local_part}@{domain_part}")) + }; + match directory.recipient(address.as_ref()).await? { + Recipient::Account(account) => { + self.synchronize_account(account).await?; + return Ok(RcptResolution::Accept); + } + Recipient::Group(group) => { + self.synchronize_group(group).await?; + return Ok(RcptResolution::Accept); + } + Recipient::Invalid => {} + } + } + // Try resolving address from registry if let Some(address_type) = self .rcpt_id_from_parts(local_part.as_ref(), domain.id) .await? { match address_type { - EmailCache::Account(id) => { + EmailCache::Account(id) if directory.is_none() => { if self.try_account(id).await?.is_some() { return if local_part.as_ref() == local_part_orig { Ok(RcptResolution::Accept) @@ -135,29 +158,7 @@ impl Server { .remove(&EmailAddressRef::new(local_part.as_ref(), domain.id)); } } - } - } - - // Obtain external directory, if configured - if let Some(directory) = self - .get_directory_for_cached_domain(&domain) - .filter(|directory| directory.can_lookup_recipients()) - { - let address = if local_part.as_ref() == local_part_orig { - Cow::Borrowed(rcpt) - } else { - Cow::Owned(format!("{local_part}@{domain_part}")) - }; - match directory.recipient(address.as_ref()).await? { - Recipient::Account(account) => { - self.synchronize_account(account).await?; - return Ok(RcptResolution::Accept); - } - Recipient::Group(group) => { - self.synchronize_group(group).await?; - return Ok(RcptResolution::Accept); - } - Recipient::Invalid => {} + _ => {} } } diff --git a/crates/directory/src/backend/oidc/lookup.rs b/crates/directory/src/backend/oidc/lookup.rs index 91563d4b..02bdd945 100644 --- a/crates/directory/src/backend/oidc/lookup.rs +++ b/crates/directory/src/backend/oidc/lookup.rs @@ -10,7 +10,7 @@ use crate::{ }; use ahash::AHashMap; use jsonwebtoken::{ - Algorithm, DecodingKey, Validation, decode, decode_header, + Algorithm, DecodingKey, Header, Validation, decode, decode_header, jwk::{self, JwkSet}, }; use reqwest::Client; @@ -21,38 +21,33 @@ use trc::AuthEvent; impl OpenIdDirectory { pub async fn authenticate(&self, credentials: &Credentials) -> trc::Result { match credentials { - Credentials::Bearer { token, .. } => { - if token.chars().filter(|&c| c == '.').count() == 2 { - self.authenticate_jwt(token).await - } else { - #[cfg(feature = "test_mode")] - let token = token.strip_prefix(".").unwrap_or(token); - self.authenticate_opaque(token).await - } - .map_err(|err| match err { - OidcError::AuthorizationFailed(reason) => { - AuthEvent::Failed.into_err().reason(reason) - } - err => AuthEvent::Error.into_err().reason(err), - }) + Credentials::Bearer { token, .. } => if let Ok(header) = decode_header(token) { + self.authenticate_jwt(token, header).await + } else { + #[cfg(feature = "test_mode")] + let token = token.strip_prefix(".").unwrap_or(token); + self.authenticate_opaque(token).await } + .map_err(|err| match err { + OidcError::AuthorizationFailed(reason) => { + AuthEvent::Failed.into_err().reason(reason) + } + err => AuthEvent::Error.into_err().reason(err), + }), _ => Err(AuthEvent::Error .into_err() .reason("Unsupported credentials type for OIDC backend")), } } - async fn authenticate_jwt(&self, token: &str) -> Result { - let header = decode_header(token) - .map_err(|e| OidcError::TokenValidation(format!("Failed to decode JWT header: {e}")))?; - - match header.alg { - Algorithm::HS256 | Algorithm::HS384 | Algorithm::HS512 => { - return Err(OidcError::TokenValidation( - "Unsupported algorithm".to_string(), - )); - } - _ => {} + async fn authenticate_jwt(&self, token: &str, header: Header) -> Result { + if matches!( + header.alg, + Algorithm::HS256 | Algorithm::HS384 | Algorithm::HS512 + ) { + return Err(OidcError::TokenValidation( + "Unsupported algorithm".to_string(), + )); } let candidates = self.get_key(header.kid.as_deref()).await?; @@ -196,28 +191,24 @@ impl OpenIdDirectory { } fn build_account(&self, claims: &serde_json::Value) -> Result { - let email = self.resolve_email(claims)?; - let description = self - .config - .claim_name - .as_ref() - .and_then(|name_claim| claims.get(name_claim)) - .and_then(|v| v.as_str()) - .map(|s| s.to_string()); - let groups = self - .config - .claim_groups - .as_ref() - .and_then(|groups_claim| claims.get(groups_claim)) - .map(extract_string_list) - .unwrap_or_default(); - Ok(Account { - email, + email: self.resolve_email(claims)?, email_aliases: Vec::new(), secret: None, - groups, - description, + groups: self + .config + .claim_groups + .as_ref() + .and_then(|groups_claim| claims.get(groups_claim)) + .map(extract_string_list) + .unwrap_or_default(), + description: self + .config + .claim_name + .as_ref() + .and_then(|name_claim| claims.get(name_claim)) + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), }) } @@ -318,23 +309,9 @@ pub(super) async fn fetch_jwks_keys( } }; - if matches!( - algorithm, - Algorithm::HS256 | Algorithm::HS384 | Algorithm::HS512 - ) { - trc::event!( - Auth(AuthEvent::Warning), - Url = jwks_uri.to_string(), - Reason = format!( - "HMAC algorithm {:?} in JWKS (kid={:?}) is not accepted, skipping", - algorithm, key.common.key_id - ) - ); - continue; - } - - let decoding_key = DecodingKey::from_jwk(key) - .map_err(|e| { + let decoding_key = match DecodingKey::from_jwk(key) { + Ok(decoding_key) => decoding_key, + Err(e) => { trc::event!( Auth(AuthEvent::Warning), Url = jwks_uri.to_string(), @@ -343,25 +320,19 @@ pub(super) async fn fetch_jwks_keys( key.common.key_id ) ); - }) - .ok(); - - let decoding_key = match decoding_key { - Some(dk) => dk, - None => continue, - }; - - let kid = match &key.common.key_id { - Some(id) => id.clone(), - None => { - let id = format!("_synthetic_{synthetic_id}"); - synthetic_id += 1; - id + continue; } }; map.insert( - kid, + match &key.common.key_id { + Some(id) => id.clone(), + None => { + let id = format!("_synthetic_{synthetic_id}"); + synthetic_id += 1; + id + } + }, CachedKey { decoding_key, algorithm, diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index eb91f534..9dbe1411 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -351,9 +351,13 @@ impl RequestHandler for Server { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; - self.registry_get(method_name.obj.unwrap_registry(), req, access_token) - .await? - .into() + Box::pin(self.registry_get( + method_name.obj.unwrap_registry(), + req, + access_token, + )) + .await? + .into() } }, RequestMethod::Query(req) => match req { @@ -426,9 +430,13 @@ impl RequestHandler for Server { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; - self.registry_query(method_name.obj.unwrap_registry(), req, access_token) - .await? - .into() + Box::pin(self.registry_query( + method_name.obj.unwrap_registry(), + req, + access_token, + )) + .await? + .into() } }, RequestMethod::Set(req) => match req { @@ -536,12 +544,12 @@ impl RequestHandler for Server { set_account_id_if_missing(&mut req.account_id, access_token); access_token.assert_is_member(req.account_id)?; - self.registry_set( + Box::pin(self.registry_set( method_name.obj.unwrap_registry(), req, access_token, session, - ) + )) .await? .into() } diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index 27b1ea4e..c24ec6ae 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -321,6 +321,7 @@ impl RegistrySet for Server { if is_create { // Patch object + match new_object.patch( JsonPointerPatch::new(&JsonPointer::new(vec![])) .with_create(true) diff --git a/crates/registry/src/utils/account.rs b/crates/registry/src/utils/account.rs index 775c4c8d..ef58f0f4 100644 --- a/crates/registry/src/utils/account.rs +++ b/crates/registry/src/utils/account.rs @@ -124,4 +124,11 @@ impl Credential { Credential::Password(_) => None, } } + + pub fn as_main_credential(&self) -> Option<&PasswordCredential> { + match self { + Credential::Password(credential) => Some(credential), + Credential::AppPassword(_) | Credential::ApiKey(_) => None, + } + } } diff --git a/crates/services/src/broadcast/mod.rs b/crates/services/src/broadcast/mod.rs index f245efc3..9abcb186 100644 --- a/crates/services/src/broadcast/mod.rs +++ b/crates/services/src/broadcast/mod.rs @@ -5,7 +5,7 @@ */ use common::ipc::{ - BroadcastEvent, CacheInvalidation, CalendarAlert, PushNotification, RegistryChange, + BroadcastEvent, CacheInvalidation, CalendarAlert, EmailPush, PushNotification, RegistryChange, }; use registry::{ schema::prelude::ObjectType, @@ -186,6 +186,13 @@ where }), ))) } + 2 => Ok(Some(BroadcastEvent::PushNotification( + PushNotification::EmailPush(EmailPush { + account_id: self.messages.next_leb128().ok_or(())?, + email_id: self.messages.next_leb128().ok_or(())?, + change_id: self.messages.next_leb128().ok_or(())?, + }), + ))), 3 => { let account_id = self.messages.next_leb128().ok_or(())?; Ok(Some(BroadcastEvent::PushServerUpdate(account_id))) diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 27ff36cf..8bbc3409 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -6,7 +6,7 @@ edition = "2024" [features] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "nats", "azure", "foundationdb"] #default = ["sqlite", "postgres", "mysql", "rocks", "s3", "redis", "foundationdb"] -default = ["rocks", "sqlite"] +default = ["rocks", "postgres", "redis", "nats"] sqlite = ["store/sqlite", "directory/sqlite"] foundationdb = ["store/foundation", "common/foundation"] postgres = ["store/postgres", "directory/postgres"] diff --git a/tests/docker/ldap/50-users.ldif b/tests/docker/ldap/50-users.ldif index ed0ce30b..795ae5f0 100644 --- a/tests/docker/ldap/50-users.ldif +++ b/tests/docker/ldap/50-users.ldif @@ -40,11 +40,13 @@ dn: uid=bill.foobar,ou=users,dc=stalwart,dc=test objectClass: inetOrgPerson objectClass: posixAccount objectClass: shadowAccount +objectClass: extensibleObject uid: bill.foobar cn: Bill Foobar sn: Foobar givenName: Bill mail: bill.foobar@example.org +mailAlias: bill@example.org userPassword: this is Bill's LDAP password uidNumber: 10003 gidNumber: 10003 diff --git a/tests/src/cluster/broadcast.rs b/tests/src/cluster/broadcast.rs index 2d0ecd3e..ad860c2e 100644 --- a/tests/src/cluster/broadcast.rs +++ b/tests/src/cluster/broadcast.rs @@ -4,80 +4,228 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::ClusterTest; -use crate::imap::idle; -use groupware::cache::GroupwareCache; -use std::net::IpAddr; -use types::collection::SyncCollection; +use crate::{ + imap::idle, + utils::{ + imap::{ImapConnection, Type}, + server::TestServerBuilder, + }, +}; +use imap_proto::ResponseType; +use registry::{ + schema::{ + enums::NetworkListenerProtocol, + prelude::{ObjectType, Property, SocketAddr}, + structs::{ + ClusterListenerGroup, ClusterListenerGroupProperties, ClusterRole, ClusterTaskGroup, + Coordinator, Expression, Http, NatsCoordinator, NetworkListener, RedisStore, + }, + }, + types::map::Map, +}; +use serde_json::json; +use std::str::FromStr; +use store::registry::RegistryQuery; +use types::id::Id; -pub async fn test(cluster: &ClusterTest) { - println!("Running cluster broadcast tests..."); +pub const NUM_NODES: usize = 3; - // Run IMAP idle tests across nodes - let server1 = cluster.server(1); - let server2 = cluster.server(2); - let mut node1_client = cluster.imap_client("john", 1).await; - let mut node2_client = cluster.imap_client("john", 2).await; - idle::test(&mut node1_client, &mut node2_client, true).await; +#[test] +fn cluster_tests() { + tokio::runtime::Builder::new_multi_thread() + .thread_stack_size(8 * 1024 * 1024) // 8MB stack + .enable_all() + .build() + .unwrap() + .block_on(async { + println!("Running cluster broadcast tests..."); + let mut servers = Vec::with_capacity(NUM_NODES); - // Test event broadcast - let test_ip: IpAddr = "8.8.8.8".parse().unwrap(); - assert!(!server1.is_ip_blocked(&test_ip)); - assert!(!server2.is_ip_blocked(&test_ip)); - server1.block_ip(test_ip).await.unwrap(); - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - assert!(server1.is_ip_blocked(&test_ip)); - assert!(server2.is_ip_blocked(&test_ip)); + let coordinator_id = std::env::var("COORDINATOR").expect(concat!( + "Missing coordinator type. Try running `STORE= ", + "COORDINATOR= cargo test`" + )); + let coordinator = match coordinator_id.as_str() { + "Nats" => Coordinator::Nats(NatsCoordinator { + addresses: Map::new(vec!["127.0.0.1:4222".to_string()]), + use_tls: false, + ..Default::default() + }), + "Redis" => Coordinator::Redis(RedisStore { + url: "redis://127.0.0.1".to_string(), + ..Default::default() + }), + _ => panic!("Unsupported coordinator type: {}", coordinator_id), + }; - // Change John's password and expect it to propagate - let account_id = cluster.account_id("john"); - assert!(server1.inner.cache.access_tokens.get(&account_id).is_some()); - assert!(server2.inner.cache.access_tokens.get(&account_id).is_some()); - let changes = server1 - .core - .storage - .data - .update_principal( - UpdatePrincipal::by_id(account_id).with_updates(vec![PrincipalUpdate { - action: PrincipalAction::AddItem, - field: PrincipalField::Secrets, - value: PrincipalValue::String("hello".into()), - }]), - ) - .await - .unwrap(); - server1.invalidate_principal_caches(changes).await; - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - assert!(server1.inner.cache.access_tokens.get(&account_id).is_none()); - assert!(server2.inner.cache.access_tokens.get(&account_id).is_none()); + // Create initial server + let test = TestServerBuilder::new("cluster_test_0") + .await + .with_object(Http { + base_url: Expression { + else_: "'https://127.0.0.1:' + local_port".to_string(), + ..Default::default() + }, + ..Default::default() + }) + .await + .with_object(coordinator) + .await + .with_listener(NetworkListenerProtocol::Http, "http_0", 11000, true) + .await + .with_imap_listener(12000) + .await + .with_listener(NetworkListenerProtocol::Lmtp, "lmtp_0", 11200, false) + .await + .build() + .await; + let admin = test.account("admin"); + admin.mta_no_auth().await; + let account = admin + .create_user_account( + "jdoe@example.com", + "this is john's secret", + "John's account", + &[], + vec![], + ) + .await; + admin.reload_settings().await; - // Rename John to Juan and expect DAV caches to be invalidated - let access_token = server1.get_access_token(account_id).await.unwrap(); - server1 - .fetch_dav_resources(&access_token, account_id, SyncCollection::Calendar) - .await - .unwrap(); - server2 - .fetch_dav_resources(&access_token, account_id, SyncCollection::Calendar) - .await - .unwrap(); - assert!(server1.inner.cache.events.get(&account_id).is_some()); - assert!(server2.inner.cache.events.get(&account_id).is_some()); - let changes = server1 - .core - .storage - .data - .update_principal( - UpdatePrincipal::by_id(account_id).with_updates(vec![PrincipalUpdate { - action: PrincipalAction::Set, - field: PrincipalField::Name, - value: PrincipalValue::String("juan".into()), - }]), - ) - .await - .unwrap(); - server1.invalidate_principal_caches(changes).await; - tokio::time::sleep(std::time::Duration::from_millis(200)).await; - assert!(server1.inner.cache.events.get(&account_id).is_none()); - assert!(server2.inner.cache.events.get(&account_id).is_none()); + // Create listeners + let mut listeners = vec![ + test.server + .registry() + .query::>(RegistryQuery::new(ObjectType::NetworkListener)) + .await + .unwrap(), + ]; + for node_id in 1..NUM_NODES { + let http_listener_id = admin + .registry_create_object(NetworkListener { + name: format!("http_{}", node_id), + bind: Map::new(vec![ + SocketAddr::from_str(&format!("127.0.0.1:1100{node_id}")).unwrap(), + ]), + protocol: NetworkListenerProtocol::Http, + tls_implicit: true, + use_tls: true, + ..Default::default() + }) + .await; + let imap_listener_id = admin + .registry_create_object(NetworkListener { + name: format!("imap_{}", node_id), + bind: Map::new(vec![ + SocketAddr::from_str(&format!("127.0.0.1:1200{node_id}")).unwrap(), + ]), + protocol: NetworkListenerProtocol::Imap, + tls_implicit: false, + use_tls: true, + ..Default::default() + }) + .await; + listeners.push(vec![http_listener_id, imap_listener_id]); + } + + // Create node roles + for (role_id, listener_ids) in listeners.into_iter().enumerate() { + admin + .registry_create_object(ClusterRole { + name: format!("role_{role_id}"), + listeners: ClusterListenerGroup::EnableSome( + ClusterListenerGroupProperties { + listener_ids: Map::new(listener_ids), + }, + ), + tasks: ClusterTaskGroup::EnableAll, + description: None, + }) + .await; + } + servers.push(test); + + // Build additional servers + for node_id in 1..NUM_NODES { + let test = TestServerBuilder::new_with_role( + &format!("cluster_test_{node_id}"), + format!("mail-{node_id}.example.com"), + Some(format!("role_{node_id}")), + false, + ) + .await + .build_with_opts(false) + .await; + + // Verify that the server was assigned the correct node id + assert_eq!(test.server.registry().node_id(), node_id as u16); + servers.push(test); + } + + // Verify cross-cluster cache invalidations + let admin = servers[0].account("admin"); + let server1 = &servers[1].server; + let server2 = &servers[2].server; + let account_id = account.id().document_id(); + assert_eq!( + server1 + .account(account_id) + .await + .unwrap() + .description + .as_deref(), + Some("John's account") + ); + assert_eq!( + server2 + .account(account_id) + .await + .unwrap() + .description + .as_deref(), + Some("John's account") + ); + admin + .registry_update_object( + ObjectType::Account, + account.id(), + json!({ + Property::Description: "John Doe" + }), + ) + .await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert_eq!( + server1 + .account(account_id) + .await + .unwrap() + .description + .as_deref(), + Some("John Doe") + ); + assert_eq!( + server2 + .account(account_id) + .await + .unwrap() + .description + .as_deref(), + Some("John Doe") + ); + + // Run IMAP idle tests across nodes + let mut node1_client = + imap_client("jdoe@example.com", "this is john's secret", 1).await; + let mut node2_client = + imap_client("jdoe@example.com", "this is john's secret", 2).await; + idle::test(&mut node1_client, &mut node2_client, true).await; + }); +} + +async fn imap_client(login: &str, secret: &str, node_id: u32) -> ImapConnection { + let mut conn = ImapConnection::connect_to(b"A1 ", format!("127.0.0.1:1200{node_id}")).await; + conn.assert_read(Type::Untagged, ResponseType::Ok).await; + conn.authenticate(login, secret).await; + conn } diff --git a/tests/src/cluster/mod.rs b/tests/src/cluster/mod.rs index 4116df4c..95c16049 100644 --- a/tests/src/cluster/mod.rs +++ b/tests/src/cluster/mod.rs @@ -4,364 +4,5 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ - AssertConfig, TEST_USERS, add_test_certs, - directory::internal::TestInternalDirectory, - imap::{ImapConnection, Type}, - jmap::server::enterprise::EnterpriseCore, - store::cleanup::store_destroy, -}; -use ahash::AHashMap; -use common::{ - Caches, Core, Data, Inner, Server, - config::{ - server::{Listeners, ServerProtocol}, - telemetry::Telemetry, - }, -}; -use http::HttpSessionManager; -use imap::core::ImapSessionManager; -use imap_proto::ResponseType; -use jmap_client::client::{Client, Credentials}; -use managesieve::core::ManageSieveSessionManager; -use pop3::Pop3SessionManager; -use services::{SpawnServices, broadcast::subscriber::spawn_broadcast_subscriber}; -use smtp::{SpawnQueueManager, core::SmtpSessionManager}; -use std::{path::PathBuf, sync::Arc, time::Duration}; -use tokio::sync::watch; - pub mod broadcast; pub mod stress; - -pub const NUM_NODES: usize = 3; - -#[tokio::test(flavor = "multi_thread")] -pub async fn cluster_tests() { - let params = init_cluster_tests(true).await; - //stress::test(params.server.clone(), params.client).await; - broadcast::test(¶ms).await; -} - -#[allow(dead_code)] -pub struct ClusterTest { - servers: Vec, - account_ids: AHashMap, - shutdown_txs: Vec>, -} - -async fn init_cluster_tests(delete_if_exists: bool) -> ClusterTest { - // Load and parse config - let store_id = std::env::var("STORE").expect( - "Missing store type. Try running `STORE= PUBSUB= cargo test`", - ); - let pubsub_id = std::env::var("PUBSUB").expect( - "Missing store type. Try running `STORE= PUBSUB= cargo test`", - ); - let mut pubsub_config = match pubsub_id.as_str() { - "nats" => Config::new(SERVER_NATS).unwrap(), - "redis" => Config::new(SERVER_REDIS).unwrap(), - _ => panic!("Unsupported pubsub type: {}", pubsub_id), - }; - - // Build configs - let mut configs = Vec::with_capacity(NUM_NODES); - for node_id in 0..NUM_NODES { - let mut config = Config::new( - add_test_certs(SERVER) - .replace("{STORE}", &store_id) - .replace("{PUBSUB}", &pubsub_id) - .replace("{NODE_ID}", &node_id.to_string()) - .replace( - "{LEVEL}", - &std::env::var("LOG").unwrap_or_else(|_| "disable".to_string()), - ), - ) - .unwrap(); - config.resolve_all_macros().await; - configs.push(config); - } - - // Build stores - let stores = Stores::parse_all(configs.first_mut().unwrap(), false).await; - - // Build servers - let mut servers = Vec::with_capacity(NUM_NODES); - let mut shutdown_txs = Vec::with_capacity(NUM_NODES); - for config in configs { - let mut stores = stores.clone(); - stores.pubsub_stores = Stores::parse(&mut pubsub_config).await.pubsub_stores; - let (server, shutdown_tx) = build_server(config, stores).await; - servers.push(server); - shutdown_txs.push(shutdown_tx); - } - - let store = servers.first().unwrap().store().clone(); - if delete_if_exists { - store_destroy(&store).await; - } - - // Create test users - let mut account_ids = AHashMap::new(); - for (account, secret, name, email) in TEST_USERS { - let account_id = store - .create_test_user(account, secret, name, &[email]) - .await; - account_ids.insert(account.to_string(), account_id); - } - - ClusterTest { - servers, - shutdown_txs, - account_ids, - } -} - -impl ClusterTest { - pub async fn jmap_client(&self, login: &str, node_id: u32) -> Client { - Client::new() - .credentials(Credentials::basic(login, find_account_secret(login))) - .timeout(Duration::from_secs(3600)) - .accept_invalid_certs(true) - .connect(&format!("https://127.0.0.1:1800{node_id}")) - .await - .unwrap() - } - - pub async fn imap_client(&self, login: &str, node_id: u32) -> ImapConnection { - let mut conn = ImapConnection::connect_to(b"A1 ", format!("127.0.0.1:1900{node_id}")).await; - conn.assert_read(Type::Untagged, ResponseType::Ok).await; - conn.authenticate(login, find_account_secret(login)).await; - conn - } - - pub fn server(&self, node_id: usize) -> &Server { - self.servers - .get(node_id) - .unwrap_or_else(|| panic!("No server found for node ID: {}", node_id)) - } - - pub fn account_id(&self, login: &str) -> u32 { - self.account_ids - .get(login) - .cloned() - .unwrap_or_else(|| panic!("No account ID found for login: {}", login)) - } -} - -fn find_account_secret(login: &str) -> &str { - TEST_USERS - .iter() - .find(|(account, _, _, _)| account == &login) - .map(|(_, secret, _, _)| secret) - .unwrap_or_else(|| panic!("No account found for login: {}", login)) -} - -async fn build_server(mut config: Config, stores: Stores) -> (Server, watch::Sender) { - // Parse servers - let mut servers = Listeners::parse(&mut config); - - // Bind ports and drop privileges - servers.bind_and_drop_priv(&mut config); - - // Parse core - let config_manager = ConfigManager { - cfg_local: Default::default(), - cfg_local_path: PathBuf::new(), - cfg_local_patterns: Patterns::parse(&mut config).into(), - cfg_store: config - .value("storage.data") - .and_then(|id| stores.stores.get(id)) - .cloned() - .unwrap_or_default(), - }; - let tracers = Telemetry::parse(&mut config, &stores); - let core = Core::parse(&mut config, stores, config_manager) - .await - .enable_enterprise(); - let data = Data::parse(&mut config); - let cache = Caches::parse(&mut config); - let (ipc, mut ipc_rxs) = build_ipc(true); - let inner = Arc::new(Inner { - shared_core: core.into_shared(), - data, - ipc, - cache, - }); - - // Parse acceptors - servers.parse_tcp_acceptors(&mut config, inner.clone()); - - // Enable tracing - tracers.enable(true); - - // Start services - config.assert_no_errors(); - ipc_rxs.spawn_queue_manager(inner.clone()); - ipc_rxs.spawn_services(inner.clone()); - - // Spawn servers - let (shutdown_tx, shutdown_rx) = servers.spawn(|server, acceptor, shutdown_rx| { - match &server.protocol { - ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( - SmtpSessionManager::new(inner.clone()), - inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Http => server.spawn( - HttpSessionManager::new(inner.clone()), - inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Imap => server.spawn( - ImapSessionManager::new(inner.clone()), - inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Pop3 => server.spawn( - Pop3SessionManager::new(inner.clone()), - inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::ManageSieve => server.spawn( - ManageSieveSessionManager::new(inner.clone()), - inner.clone(), - acceptor, - shutdown_rx, - ), - }; - }); - - // Start broadcast subscriber - spawn_broadcast_subscriber(inner.clone(), shutdown_rx); - - (inner.build_server(), shutdown_tx) -} - -const SERVER: &str = r#" -[server] -hostname = "'server{NODE_ID}.example.org'" - -[http] -url = "'https://127.0.0.1:800{NODE_ID}'" - -[cluster] -node-id = {NODE_ID} -coordinator = "{PUBSUB}" - -[server.listener.http] -bind = ["127.0.0.1:1800{NODE_ID}"] -protocol = "http" -max-connections = 81920 -tls.implicit = true - -[server.listener.imap] -bind = ["127.0.0.1:1900{NODE_ID}"] -protocol = "imap" -max-connections = 81920 - -[server.listener.lmtp] -bind = ['127.0.0.1:1700{NODE_ID}'] -protocol = 'lmtp' -tls.implicit = false - -[server.socket] -reuse-addr = true - -[server.tls] -enable = true -implicit = false -certificate = "default" - -[session.ehlo] -reject-non-fqdn = false - -[session.rcpt] -relay = [ { if = "!is_empty(authenticated_as)", then = true }, - { else = false } ] -directory = "'{STORE}'" - -[session.rcpt.errors] -total = 5 -wait = "1ms" - -[session.auth] -mechanisms = "[plain, login, oauthbearer]" -directory = "'{STORE}'" - -[resolver] -type = "system" - -[queue.strategy] -route = [ { if = "rcpt_domain == 'example.com'", then = "'local'" }, - { else = "'mx'" } ] - -[store."foundationdb"] -type = "foundationdb" - -[store."postgresql"] -type = "postgresql" -host = "localhost" -port = 5432 -database = "stalwart" -user = "postgres" -password = "mysecretpassword" - -[store."mysql"] -type = "mysql" -host = "localhost" -port = 3307 -database = "stalwart" -user = "root" -password = "password" - -[certificate.default] -cert = "%{file:{CERT}}%" -private-key = "%{file:{PK}}%" - -[storage] -data = "{STORE}" -fts = "{STORE}" -blob = "{STORE}" -lookup = "{STORE}" -directory = "{STORE}" - -[directory."{STORE}"] -type = "internal" -store = "{STORE}" - -[imap.auth] -allow-plain-text = true - -[oauth] -key = "parerga_und_paralipomena" - -[spam-filter] -enable = false - -[tracer.console] -type = "console" -level = "{LEVEL}" -multiline = false -ansi = true -disabled-events = ["network.*", "telemetry.webhook-error", "http.request-body", - "eval.result", "store.*", "dkim.*", "queue.*", "delivery.*", - "*.raw-input", "*.raw-output" ] -"#; - -const SERVER_NATS: &str = r#" -[store."nats"] -type = "nats" -urls = "127.0.0.1:4444" -"#; - -const SERVER_REDIS: &str = r#" -[store."redis"] -type = "redis" -urls = "redis://127.0.0.1" -redis-type = "single" - -"#; diff --git a/tests/src/cluster/stress.rs b/tests/src/cluster/stress.rs index b0f8fa8f..22d59638 100644 --- a/tests/src/cluster/stress.rs +++ b/tests/src/cluster/stress.rs @@ -4,8 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::jmap::{assert_is_empty, mail::mailbox::destroy_all_mailboxes_no_wait, wait_for_tasks}; -use common::Server; +use crate::utils::server::{DestroyAllMailboxes, TestServer, TestServerBuilder}; use email::{ cache::{MessageCacheFetch, email::MessageCacheAccess}, message::metadata::MessageData, @@ -16,6 +15,7 @@ use jmap_client::{ core::set::{SetErrorType, SetObject}, mailbox::{self, Mailbox, Role}, }; +use registry::schema::prelude::ObjectType; use std::{str::FromStr, sync::Arc, time::Duration}; use store::{ ValueKey, @@ -28,22 +28,33 @@ use types::{collection::Collection, id::Id}; const TEST_USER_ID: u32 = 1; const NUM_PASSES: usize = 1; -pub async fn test(server: Server, mut client: Client) { - println!("Running cluster concurrency stress tests..."); - server - .core - .storage - .data - .get_or_create_principal_id("john", directory::Type::Individual) +#[tokio::test(flavor = "multi_thread")] +pub async fn stress_tests() { + println!("Running concurrency stress tests..."); + + let mut test = TestServerBuilder::new("stress_tests") .await - .unwrap(); - client.set_default_account_id(Id::from(TEST_USER_ID).to_string()); - let client = Arc::new(client); - email_tests(server.clone(), client.clone()).await; - mailbox_tests(server.clone(), client.clone()).await; + .with_default_listeners() + .await + .build() + .await; + let admin = test.create_admin_account("admin@example.com").await; + admin + .registry_destroy_all(ObjectType::MtaConnectionStrategy) + .await; + admin + .registry_destroy_all(ObjectType::MtaInboundThrottle) + .await; + test.insert_account(admin); + + email_tests(&test).await; + mailbox_tests(&test).await; } -async fn email_tests(server: Server, client: Arc) { +async fn email_tests(test: &TestServer) { + let server = &test.server; + let client = Arc::new(test.account("admin@example.com").jmap_client().await); + for pass in 0..NUM_PASSES { println!( "----------------- EMAIL STRESS TEST {} -----------------", @@ -265,12 +276,14 @@ async fn email_tests(server: Server, client: Arc) { } test.wait_for_tasks().await; - destroy_all_mailboxes_no_wait(&client).await; - assert_is_empty(&server).await; + client.destroy_all_mailboxes().await; + test.assert_is_empty().await; } } -async fn mailbox_tests(server: Server, client: Arc) { +async fn mailbox_tests(test: &TestServer) { + let client = Arc::new(test.account("admin@example.com").jmap_client().await); + let mailboxes = Arc::new(vec![ "test/test1/test2/test3".to_string(), "test1/test2/test3".to_string(), @@ -361,7 +374,7 @@ async fn mailbox_tests(server: Server, client: Arc) { { let _ = client.mailbox_destroy(&mailbox_id, true).await; } - assert_is_empty(&server).await; + test.assert_is_empty().await; } async fn create_mailbox(client: &Client, mailbox: &str) -> Vec { diff --git a/tests/src/directory/integration.rs b/tests/src/directory/integration.rs new file mode 100644 index 00000000..e1dd11d9 --- /dev/null +++ b/tests/src/directory/integration.rs @@ -0,0 +1,197 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{ + directory::ldap::ldap_test_directory, + utils::{server::TestServerBuilder, smtp::SmtpConnection}, +}; +use ahash::AHashMap; +use email::cache::MessageCacheFetch; +use registry::schema::structs::{Account, AccountSettings, Directory}; +use types::id::Id; + +pub async fn test() { + let test = TestServerBuilder::new("directory_integration_test") + .await + .with_default_listeners() + .await + .with_object(Directory::Ldap(ldap_test_directory())) + .await + .build() + .await; + let admin = test.account("admin"); + admin.mta_no_auth().await; + admin.mta_disable_spam_filter().await; + admin.reload_settings().await; + + // Test account creation by login + let account = crate::utils::account::Account::new( + "john.doe@example.org", + "this is John's LDAP password", + &[], + "", + Id::from(u32::MAX), + ); + assert_eq!( + account + .registry_get::(Id::singleton()) + .await + .description + .as_deref(), + Some("John Doe") + ); + + // Test account creation by rcpt + let mut lmtp = SmtpConnection::connect().await; + for rcpt in [ + "corporate@example.org", + "jane.smith@example.org", + "john@example.org", + "bill@example.org", + "sales@example.org", + ] { + lmtp.ingest( + "bill@remote.org", + &[rcpt], + &TEST_EMAIL.replace("$RCPT", rcpt), + ) + .await; + } + + // Fetch all accounts + let mut accounts = admin + .registry_get_all::() + .await + .into_iter() + .map(|(id, account)| { + ( + match &account { + Account::User(user_account) => user_account.name.clone(), + Account::Group(group_account) => group_account.name.clone(), + }, + (account, id), + ) + }) + .collect::>(); + assert_eq!(accounts.len(), 5, "Got: {accounts:#?}"); + + // Validate accounts + for (name, description, secret, groups, aliases) in [ + ( + "john.doe", + "John Doe", + "$app$8958830913002348890$", + &["sales"][..], + &["john"][..], + ), + ( + "jane.smith", + "Jane Smith", + "$app$4096614298472586996$", + &["sales", "corporate"][..], + &[][..], + ), + ( + "bill.foobar", + "Bill Foobar", + "", + &["corporate"][..], + &["bill"][..], + ), + ] { + let (account, id) = accounts + .remove(name) + .map(|(account, id)| (account.into_user().unwrap(), id)) + .unwrap(); + assert_eq!(account.description.as_deref(), Some(description)); + if !secret.is_empty() { + assert_eq!( + test.server + .registry() + .object::(id) + .await + .unwrap() + .unwrap() + .into_user() + .unwrap() + .credentials + .values() + .next() + .and_then(|v| v.as_main_credential()) + .map(|v| v.secret.as_str()), + Some(secret) + ); + } + for group in groups { + let id = accounts.get(*group).unwrap().1; + assert!( + account + .member_group_ids + .iter() + .any(|group_id| group_id == &id), + "Account {name} is not a member of group {group}" + ); + } + for alias in aliases { + assert!( + account + .aliases + .iter() + .any(|account_alias| account_alias.name == *alias), + "Account {name} does not have alias {alias}" + ); + } + assert_eq!( + test.server + .get_cached_messages(id.document_id()) + .await + .unwrap() + .emails + .index + .len(), + 1 + ); + } + + // Validate groups + for (name, description, aliases) in [ + ("sales", "sales", &[][..]), + ("corporate", "corporate", &["everyone"][..]), + ] { + let (account, id) = accounts + .remove(name) + .map(|(account, id)| (account.into_group().unwrap(), id)) + .unwrap(); + assert_eq!(account.description.as_deref(), Some(description)); + for alias in aliases { + assert!( + account + .aliases + .iter() + .any(|account_alias| account_alias.name == *alias), + "Group {name} does not have alias {alias}" + ); + } + assert_eq!( + test.server + .get_cached_messages(id.document_id()) + .await + .unwrap() + .emails + .index + .len(), + 1 + ); + } +} + +const TEST_EMAIL: &str = r#"From: bill@remote.org +To: $RCPT +Subject: TPS Report for $RCPT + +I'm going to need those TPS reports ASAP. So, if you could do that, that'd be great. + +"#; diff --git a/tests/src/directory/ldap.rs b/tests/src/directory/ldap.rs index e72bafcc..64bfe693 100644 --- a/tests/src/directory/ldap.rs +++ b/tests/src/directory/ldap.rs @@ -10,34 +10,8 @@ use registry::{ types::map::Map, }; -#[tokio::test] -async fn ldap_directory() { - let mut config = structs::LdapDirectory { - url: "ldap://localhost".into(), - use_tls: false, - attr_class: Map::new(vec!["objectClass".to_string()]), - attr_description: Map::new(vec!["cn".to_string()]), - attr_email: Map::new(vec!["mail".to_string()]), - attr_email_alias: Map::new(vec!["mailAlias".to_string()]), - attr_member_of: Map::new(vec!["memberOf".to_string()]), - attr_secret: Map::new(vec![]), - attr_secret_changed: Map::new(vec!["shadowLastChange".to_string()]), - base_dn: "dc=stalwart,dc=test".into(), - bind_dn: "cn=admin,dc=stalwart,dc=test".to_string().into(), - bind_secret: SecretKeyOptional::Value(SecretKeyValue { - secret: "admin".into(), - }), - filter_member_of: "(&(objectClass=groupOfNames)(member=?))".to_string().into(), - filter_login: "(&(objectClass=inetOrgPerson)(mail=?))".into(), - filter_mailbox: concat!( - "(|(&(objectClass=inetOrgPerson)(|(mail=?)(mailAlias=?)))", - "(&(objectClass=groupOfNames)(|(mail=?)(mailAlias=?))))" - ) - .into(), - group_class: "groupOfNames".into(), - bind_authentication: true, - ..Default::default() - }; +pub async fn test() { + let mut config = ldap_test_directory(); // Test bind authentication let ldap = LdapDirectory::open(config.clone()).await.unwrap(); @@ -156,3 +130,32 @@ async fn ldap_directory() { Recipient::Invalid ); } + +pub fn ldap_test_directory() -> structs::LdapDirectory { + structs::LdapDirectory { + url: "ldap://localhost".into(), + use_tls: false, + attr_class: Map::new(vec!["objectClass".to_string()]), + attr_description: Map::new(vec!["cn".to_string()]), + attr_email: Map::new(vec!["mail".to_string()]), + attr_email_alias: Map::new(vec!["mailAlias".to_string()]), + attr_member_of: Map::new(vec!["memberOf".to_string()]), + attr_secret: Map::new(vec![]), + attr_secret_changed: Map::new(vec!["shadowLastChange".to_string()]), + base_dn: "dc=stalwart,dc=test".into(), + bind_dn: "cn=admin,dc=stalwart,dc=test".to_string().into(), + bind_secret: SecretKeyOptional::Value(SecretKeyValue { + secret: "admin".into(), + }), + filter_member_of: "(&(objectClass=groupOfNames)(member=?))".to_string().into(), + filter_login: "(&(objectClass=inetOrgPerson)(mail=?))".into(), + filter_mailbox: concat!( + "(|(&(objectClass=inetOrgPerson)(|(mail=?)(mailAlias=?)))", + "(&(objectClass=groupOfNames)(|(mail=?)(mailAlias=?))))" + ) + .into(), + group_class: "groupOfNames".into(), + bind_authentication: true, + ..Default::default() + } +} diff --git a/tests/src/directory/mod.rs b/tests/src/directory/mod.rs index c4b0c26f..7a4b0e37 100644 --- a/tests/src/directory/mod.rs +++ b/tests/src/directory/mod.rs @@ -4,6 +4,19 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +pub mod integration; pub mod ldap; pub mod oidc; -//pub mod sql; +#[cfg(feature = "sqlite")] +pub mod sql; +pub mod synchronization; + +#[tokio::test(flavor = "multi_thread")] +pub async fn directory_tests() { + ldap::test().await; + oidc::test().await; + #[cfg(feature = "sqlite")] + sql::test().await; + synchronization::test().await; + integration::test().await; +} diff --git a/tests/src/directory/oidc.rs b/tests/src/directory/oidc.rs index 016046bd..b498025c 100644 --- a/tests/src/directory/oidc.rs +++ b/tests/src/directory/oidc.rs @@ -11,8 +11,7 @@ use directory::{Account, Credentials, Directory, backend::oidc::OpenIdDirectory}; use registry::{schema::structs, types::map::Map}; -#[tokio::test] -async fn oidc_directory() { +pub async fn test() { let config = structs::OidcDirectory { issuer_url: "http://localhost:9080/realms/stalwart".to_string(), claim_username: "preferred_username".to_string(), diff --git a/tests/src/directory/sql.rs b/tests/src/directory/sql.rs index 7785a3fb..4bb01835 100644 --- a/tests/src/directory/sql.rs +++ b/tests/src/directory/sql.rs @@ -4,573 +4,141 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use mail_send::Credentials; +use directory::{Account, Credentials, Group, Recipient, backend::sql::SqlDirectory}; +use registry::schema::structs::{self, SqlAuthStore}; +use store::{Store, backend::sqlite::SqliteStore}; -#[allow(unused_imports)] -use store::{InMemoryStore, Store}; +pub async fn test() { + let sql_store = Store::SQLite(SqliteStore::open_memory().unwrap().into()); -use crate::{ - directory::{DirectoryTest, IntoTestPrincipal, TestPrincipal, map_account_id, map_account_ids}, - store::cleanup::store_destroy, -}; - -use super::DirectoryStore; - -#[tokio::test] -async fn sql_directory() { - // Enable logging - /*tracing::subscriber::set_global_default( - tracing_subscriber::FmtSubscriber::builder() - .with_max_level(tracing::Level::TRACE) - .finish(), - ) - .unwrap();*/ - - // Obtain directory handle - for directory_id in ["sqlite", "postgresql", "mysql"] { - // Parse config - let mut config = DirectoryTest::new(directory_id.into()).await; - - println!("Testing SQL directory {:?}", directory_id); - let handle = config.directories.directories.remove(directory_id).unwrap(); - let store = DirectoryStore { - store: config.stores.stores.remove(directory_id).unwrap(), - }; - let base_store = &store.store; - let core = config.server; - - // Create tables - store_destroy(base_store).await; - store.create_test_directory().await; - - // Create test users - store - .create_test_user("admin", "very_secret", "Administrator") - .await; - store.create_test_user("john", "12345", "John Doe").await; - store.create_test_user("jane", "abcde", "Jane Doe").await; - store - .create_test_user( - "bill", - "$2y$05$bvIG6Nmid91Mu9RcmmWZfO5HJIMCT8riNW0hEp8f6/FuA2/mHZFpe", - "Bill Foobar", - ) - .await; - store.set_test_quota("bill", 500000).await; - - // Create test groups - store.create_test_group("sales", "Sales Team").await; - store.create_test_group("support", "Support Team").await; - - // Link users to groups - store.add_to_group("john", "sales").await; - store.add_to_group("jane", "sales").await; - store.add_to_group("jane", "support").await; - - // Add email addresses - store - .link_test_address("john", "john@example.org", "primary") - .await; - store - .link_test_address("jane", "jane@example.org", "primary") - .await; - store - .link_test_address("bill", "bill@example.org", "primary") - .await; - - // Add aliases and lists - store - .link_test_address("john", "john.doe@example.org", "alias") - .await; - store - .link_test_address("john", "jdoe@example.org", "alias") - .await; - store - .link_test_address("john", "info@example.org", "list") - .await; - store - .link_test_address("jane", "info@example.org", "list") - .await; - store - .link_test_address("bill", "info@example.org", "list") - .await; - - // Add catch-all user - store - .create_test_user("robert", "abcde", "Robert Foobar") - .await; - store - .link_test_address("robert", "robert@catchall.org", "primary") - .await; - store - .link_test_address("robert", "@catchall.org", "alias") - .await; - - // Test authentication - assert_eq!( - handle - .query( - QueryParams::credentials(&Credentials::Plain { - username: "john".into(), - secret: "12345".into() - }) - .with_return_member_of(true) - ) - .await - .unwrap() - .unwrap() - .into_test(), - TestPrincipal { - id: base_store.get_principal_id("john").await.unwrap().unwrap(), - name: "john".into(), - description: Some("John Doe".into()), - secrets: vec!["12345".into()], - typ: Type::Individual, - member_of: map_account_ids(base_store, vec!["sales"]) - .await - .into_iter() - .map(|v| v.to_string()) - .collect(), - emails: vec![ - "john@example.org".into(), - "jdoe@example.org".into(), - "john.doe@example.org".into() - ], - roles: vec![ROLE_USER.to_string()], - ..Default::default() - } - ); - assert_eq!( - handle - .query( - QueryParams::credentials(&Credentials::Plain { - username: "bill".into(), - secret: "password".into() - }) - .with_return_member_of(true) - ) - .await - .unwrap() - .unwrap() - .into_test(), - TestPrincipal { - id: base_store.get_principal_id("bill").await.unwrap().unwrap(), - name: "bill".into(), - description: Some("Bill Foobar".into()), - secrets: vec![ - "$2y$05$bvIG6Nmid91Mu9RcmmWZfO5HJIMCT8riNW0hEp8f6/FuA2/mHZFpe".into() - ], - typ: Type::Individual, - quota: 500000, - emails: vec!["bill@example.org".into(),], - roles: vec![ROLE_USER.to_string()], - ..Default::default() - } - ); - assert_eq!( - handle - .query( - QueryParams::credentials(&Credentials::Plain { - username: "admin".into(), - secret: "very_secret".into() - }) - .with_return_member_of(true) - ) - .await - .unwrap() - .unwrap() - .into_test(), - TestPrincipal { - id: base_store.get_principal_id("admin").await.unwrap().unwrap(), - name: "admin".into(), - description: Some("Administrator".into()), - secrets: vec!["very_secret".into()], - typ: Type::Individual, - roles: vec![ROLE_ADMIN.to_string()], - ..Default::default() - } - ); - assert!( - handle - .query( - QueryParams::credentials(&Credentials::Plain { - username: "bill".into(), - secret: "invalid".into() - }) - .with_return_member_of(true) - ) - .await - .unwrap() - .is_none() - ); - - // Get user by name - let mut p = handle - .query(QueryParams::name("jane").with_return_member_of(true)) + // Create test directory + for query in [ + concat!( + "CREATE TABLE accounts (name TEXT PRIMARY KEY, secret TEXT, description TEXT,", + " type TEXT NOT NULL, active BOOLEAN DEFAULT TRUE)" + ), + concat!( + "CREATE TABLE group_members (name TEXT NOT NULL, member_of ", + "TEXT NOT NULL, PRIMARY KEY (name, member_of))" + ), + concat!( + "CREATE TABLE emails (name TEXT NOT NULL, address TEXT NOT", + " NULL, PRIMARY KEY (name, address))" + ), + concat!( + "INSERT INTO accounts (name, secret, description, type) ", + "VALUES ('john@example.org', 'john secret', 'John Doe', 'individual')" + ), + concat!( + "INSERT INTO accounts (name, secret, description, type) ", + "VALUES ('jane@example.org', 'jane secret', 'Jane Doe', 'individual')" + ), + concat!( + "INSERT INTO accounts (name, secret, description, type) ", + "VALUES ('sales@example.org', NULL, 'Sales Team', 'group')" + ), + concat!( + "INSERT INTO group_members (name, member_of) VALUES ", + "('john@example.org', 'sales@example.org')" + ), + concat!( + "INSERT INTO group_members (name, member_of) VALUES ", + "('jane@example.org', 'sales@example.org')" + ), + concat!( + "INSERT INTO emails (name, address) VALUES ", + "('john@example.org', 'john.doe@example.org')" + ), + ] { + sql_store + .sql_query::(query, vec![]) .await - .unwrap() - .unwrap() - .into_test(); - p.member_of.sort(); - assert_eq!( - p, - TestPrincipal { - id: base_store.get_principal_id("jane").await.unwrap().unwrap(), - name: "jane".into(), - description: Some("Jane Doe".into()), - typ: Type::Individual, - secrets: vec!["abcde".into()], - member_of: map_account_ids(base_store, vec!["sales", "support"]) - .await - .into_iter() - .map(|v| v.to_string()) - .collect(), - emails: vec!["jane@example.org".into(),], - roles: vec![ROLE_USER.to_string()], - ..Default::default() - } - ); - - // Get group by name - assert_eq!( - handle - .query(QueryParams::name("sales").with_return_member_of(true)) - .await - .unwrap() - .unwrap() - .into_test(), - TestPrincipal { - id: base_store.get_principal_id("sales").await.unwrap().unwrap(), - name: "sales".into(), - description: Some("Sales Team".into()), - typ: Type::Group, - roles: vec![ROLE_USER.to_string()], - ..Default::default() - } - ); - - // Ids by email - assert_eq!( - core.email_to_id(&handle, "jane@example.org", 0) - .await - .unwrap(), - Some(map_account_id(base_store, "jane").await) - ); - assert_eq!( - core.email_to_id(&handle, "jane+alias@example.org", 0) - .await - .unwrap(), - Some(map_account_id(base_store, "jane").await) - ); - assert_eq!( - core.email_to_id(&handle, "unknown@example.org", 0) - .await - .unwrap(), - None - ); - assert_eq!( - core.email_to_id(&handle, "anything@catchall.org", 0) - .await - .unwrap(), - Some(map_account_id(base_store, "robert").await) - ); - - // Domain validation - assert!(handle.is_local_domain("example.org").await.unwrap()); - assert!(!handle.is_local_domain("other.org").await.unwrap()); - - // RCPT TO - assert_eq!( - core.rcpt(&handle, "jane@example.org", 0).await.unwrap(), - RcptType::Mailbox - ); - assert_eq!( - core.rcpt(&handle, "info@example.org", 0).await.unwrap(), - RcptType::Mailbox - ); - assert_eq!( - core.rcpt(&handle, "jane+alias@example.org", 0) - .await - .unwrap(), - RcptType::Mailbox - ); - assert_eq!( - core.rcpt(&handle, "info+alias@example.org", 0) - .await - .unwrap(), - RcptType::Mailbox - ); - assert_eq!( - core.rcpt(&handle, "random_user@catchall.org", 0) - .await - .unwrap(), - RcptType::Mailbox - ); - assert_eq!( - core.rcpt(&handle, "invalid@example.org", 0).await.unwrap(), - RcptType::Invalid - ); - - // VRFY - assert_eq!( - core.vrfy(&handle, "jane", 0).await.unwrap(), - vec!["jane@example.org".to_string()] - ); - assert_eq!( - core.vrfy(&handle, "john", 0).await.unwrap(), - vec![ - "john.doe@example.org".to_string(), - "john@example.org".to_string(), - ] - ); - assert_eq!( - core.vrfy(&handle, "jane+alias@example", 0).await.unwrap(), - vec!["jane@example.org".to_string()] - ); - assert_eq!( - core.vrfy(&handle, "info", 0).await.unwrap(), - Vec::::new() - ); - assert_eq!( - core.vrfy(&handle, "invalid", 0).await.unwrap(), - Vec::::new() - ); - - // EXPN (now handled by the internal store) - /*assert_eq!( - core.expn(&handle, "info@example.org", 0).await.unwrap(), - vec![ - "bill@example.org".into(), - "jane@example.org".into(), - "john@example.org".into() - ] - ); - assert_eq!( - core.expn(&handle, "john@example.org", 0).await.unwrap(), - Vec::::new() - );*/ - } -} - -impl DirectoryStore { - pub async fn create_test_directory(&self) { - // Create tables - for table in ["accounts", "group_members", "emails"] { - self.store - .sql_query::(&format!("DROP TABLE IF EXISTS {table}"), vec![]) - .await - .unwrap(); - } - for query in [ - concat!( - "CREATE TABLE accounts (name TEXT PRIMARY KEY, secret TEXT, description TEXT,", - " type TEXT NOT NULL, quota INTEGER ", - "DEFAULT 0, active BOOLEAN DEFAULT TRUE)" - ), - concat!( - "CREATE TABLE group_members (name TEXT NOT NULL, member_of ", - "TEXT NOT NULL, PRIMARY KEY (name, member_of))" - ), - concat!( - "CREATE TABLE emails (name TEXT NOT NULL, address TEXT NOT", - " NULL, type TEXT, PRIMARY KEY (name, address))" - ), - "INSERT INTO accounts (name, secret, type) VALUES ('admin', 'secret', 'admin')", - ] { - let query = if self.is_mysql() { - query.replace("TEXT", "VARCHAR(255)") - } else { - query.into() - }; - - self.store - .sql_query::(&query, vec![]) - .await - .unwrap_or_else(|_| panic!("failed for {query}")); - } - } - - pub async fn create_test_user(&self, login: &str, secret: &str, name: &str) { - let account_type = if login == "admin" { - "admin" - } else { - "individual" - }; - self.store - .sql_query::( - if self.is_postgresql() { - concat!( - "INSERT INTO accounts (name, secret, description, ", - "type, active) VALUES ($1, $2, $3, $4, true) ", - "ON CONFLICT (name) ", - "DO UPDATE SET secret = $2, description = $3, type = $4, active = true" - ) - } else if self.is_mysql() { - concat!( - "INSERT INTO accounts (name, secret, description, ", - "type, active) VALUES (?, ?, ?, ?, true) ", - "ON DUPLICATE KEY UPDATE ", - "secret = VALUES(secret), description = VALUES(description), ", - "type = VALUES(type), active = true" - ) - } else { - concat!( - "INSERT INTO accounts (name, secret, description, ", - "type, active) VALUES (?, ?, ?, ?, true) ", - "ON CONFLICT(name) DO UPDATE SET ", - "secret = excluded.secret, description = excluded.description, ", - "type = excluded.type, active = true" - ) - }, - vec![ - login.into(), - secret.into(), - name.into(), - account_type.into(), - ], - ) - .await - .unwrap(); - } - - pub async fn create_test_user_with_email(&self, login: &str, secret: &str, name: &str) { - self.create_test_user(login, secret, name).await; - self.link_test_address(login, login, "primary").await; - } - - pub async fn create_test_group(&self, login: &str, name: &str) { - self.store - .sql_query::( - if self.is_postgresql() { - concat!( - "INSERT INTO accounts (name, description, ", - "type, active) VALUES ($1, $2, $3, $4) ON CONFLICT (name) DO NOTHING" - ) - } else if self.is_mysql() { - concat!( - "INSERT IGNORE INTO accounts (name, description, ", - "type, active) VALUES (?, ?, ?, ?)" - ) - } else { - concat!( - "INSERT OR IGNORE INTO accounts (name, description, ", - "type, active) VALUES (?, ?, ?, ?)" - ) - }, - vec![login.into(), name.into(), "group".into(), true.into()], - ) - .await - .unwrap(); - } - - pub async fn create_test_group_with_email(&self, login: &str, name: &str) { - self.create_test_group(login, name).await; - self.link_test_address(login, login, "primary").await; - } - - pub async fn link_test_address(&self, login: &str, address: &str, typ: &str) { - self.store - .sql_query::( - if self.is_postgresql() { - "INSERT INTO emails (name, address, type) VALUES ($1, $2, $3) ON CONFLICT (name, address) DO NOTHING" - } else if self.is_mysql() { - "INSERT IGNORE INTO emails (name, address, type) VALUES (?, ?, ?)" - } else { - "INSERT OR IGNORE INTO emails (name, address, type) VALUES (?, ?, ?)" - }, - vec![login.into(), address.into(), typ.into()], - ) - .await - .unwrap(); - } - - pub async fn set_test_quota(&self, login: &str, quota: u32) { - self.store - .sql_query::( - if self.is_postgresql() { - "UPDATE accounts SET quota = $1 where name = $2" - } else { - "UPDATE accounts SET quota = ? where name = ?" - }, - vec![quota.into(), login.into()], - ) - .await - .unwrap(); - } - - pub async fn add_to_group(&self, login: &str, group: &str) { - self.store - .sql_query::( - if self.is_postgresql() { - "INSERT INTO group_members (name, member_of) VALUES ($1, $2)" - } else { - "INSERT INTO group_members (name, member_of) VALUES (?, ?)" - }, - vec![login.into(), group.into()], - ) - .await - .unwrap(); - } - - pub async fn remove_from_group(&self, login: &str, group: &str) { - self.store - .sql_query::( - if self.is_postgresql() { - "DELETE FROM group_members WHERE name = $1 AND member_of = $2" - } else { - "DELETE FROM group_members WHERE name = ? AND member_of = ?" - }, - vec![login.into(), group.into()], - ) - .await - .unwrap(); - } - - pub async fn remove_test_alias(&self, login: &str, alias: &str) { - self.store - .sql_query::( - if self.is_postgresql() { - "DELETE FROM emails WHERE name = $1 AND address = $2" - } else { - "DELETE FROM emails WHERE name = ? AND address = ?" - }, - vec![login.into(), alias.into()], - ) - .await - .unwrap(); - } - - fn is_mysql(&self) -> bool { - #[cfg(feature = "mysql")] - { - matches!(self.store, Store::MySQL(_)) - } - #[cfg(not(feature = "mysql"))] - { - false - } - } - - fn is_postgresql(&self) -> bool { - #[cfg(feature = "postgres")] - { - matches!(self.store, Store::PostgreSQL(_)) - } - #[cfg(not(feature = "postgres"))] - { - false - } - } - - #[allow(dead_code)] - fn is_sqlite(&self) -> bool { - #[cfg(feature = "sqlite")] - { - matches!(self.store, Store::SQLite(_)) - } - #[cfg(not(feature = "sqlite"))] - { - false - } + .unwrap_or_else(|_| panic!("failed for {query}")); } + + let config = structs::SqlDirectory { + query_login: concat!( + "SELECT name, secret, description, type FROM accounts ", + "WHERE name = $1 AND active = true" + ) + .into(), + query_recipient: concat!( + "SELECT name, secret, description, type FROM accounts ", + "WHERE name = $1 AND active = true" + ) + .into(), + query_email_aliases: concat!("SELECT address FROM emails ", "WHERE name = $1") + .to_string() + .into(), + query_member_of: concat!("SELECT member_of FROM group_members ", "WHERE name = $1") + .to_string() + .into(), + column_class: "type".to_string().into(), + column_description: "description".to_string().into(), + column_email: "name".into(), + column_secret: "secret".into(), + store: SqlAuthStore::Default, + }; + + // Test authentication + let sql = SqlDirectory::open(config, &sql_store).await.unwrap(); + assert_eq!( + sql.authenticate(&Credentials::Basic { + username: "john@example.org".to_string(), + secret: "john secret".to_string(), + mfa_token: None, + }) + .await + .unwrap(), + Account { + email: "john@example.org".to_string(), + email_aliases: vec!["john.doe@example.org".to_string(),], + secret: Some("john secret".to_string()), + groups: vec!["sales@example.org".to_string()], + description: Some("John Doe".to_string()), + } + ); + assert!( + sql.authenticate(&Credentials::Basic { + username: "john@example.org".to_string(), + secret: "wrong secret".to_string(), + mfa_token: None, + }) + .await + .is_err() + ); + + // Test recipient lookup + assert_eq!( + sql.recipient("john@example.org").await.unwrap(), + Recipient::Account(Account { + email: "john@example.org".to_string(), + email_aliases: vec!["john.doe@example.org".to_string()], + secret: Some("john secret".to_string()), + groups: vec!["sales@example.org".to_string()], + description: Some("John Doe".to_string()), + }) + ); + assert_eq!( + sql.recipient("jane@example.org").await.unwrap(), + Recipient::Account(Account { + email: "jane@example.org".to_string(), + email_aliases: vec![], + secret: Some("jane secret".to_string()), + groups: vec!["sales@example.org".to_string()], + description: Some("Jane Doe".to_string()), + }) + ); + assert_eq!( + sql.recipient("sales@example.org").await.unwrap(), + Recipient::Group(Group { + email: "sales@example.org".to_string(), + email_aliases: vec![], + description: Some("Sales Team".to_string()) + }) + ); + assert_eq!( + sql.recipient("unknown@example.org").await.unwrap(), + Recipient::Invalid + ); } diff --git a/tests/src/directory/synchronization.rs b/tests/src/directory/synchronization.rs new file mode 100644 index 00000000..71987375 --- /dev/null +++ b/tests/src/directory/synchronization.rs @@ -0,0 +1,271 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::utils::server::TestServerBuilder; +use registry::schema::{ + prelude::ObjectType, + structs::{Account, Domain, EmailAlias}, +}; +use types::id::Id; + +pub async fn test() { + let test = TestServerBuilder::new("directory_synchronization_test") + .await + .with_default_listeners() + .await + .disable_services() + .build() + .await; + let admin = test.account("admin"); + + // Synchronizing an account with an unknown domain should fail + assert!( + test.server + .synchronize_account(directory::Account { + email: "john@unknown.org".to_string(), + email_aliases: vec![], + secret: "supersecret".to_string().into(), + groups: vec![], + description: "John Doe".to_string().into(), + }) + .await + .is_err() + ); + + // Initial account synchronization + let mut account_in = directory::Account { + email: "john@example.org".to_string(), + email_aliases: vec![ + "john.doe@example.org".to_string(), + "j.doe@example.org".to_string(), + ], + secret: "supersecret".to_string().into(), + groups: vec![ + "corporate@example.org".to_string(), + "sales@example.org".to_string(), + ], + description: "John Doe".to_string().into(), + }; + let result = test + .server + .synchronize_account(account_in.clone()) + .await + .unwrap(); + let account_id = Id::from(result.id); + let account_out = test + .server + .registry() + .object::(account_id) + .await + .unwrap() + .unwrap() + .into_user() + .unwrap(); + let domain_id = account_out.domain_id; + assert_eq!( + admin.registry_get::(domain_id).await.name, + "example.org" + ); + assert_eq!(account_out.name, "john"); + assert_eq!(account_out.description.as_deref(), Some("John Doe")); + assert_eq!( + account_out + .credentials + .values() + .next() + .and_then(|v| v.as_main_credential()) + .map(|c| c.secret.as_str()), + Some("supersecret") + ); + assert_eq!(account_out.aliases.len(), 2); + let aliases = account_out.aliases.iter().collect::>(); + assert_eq!( + aliases[0], + &EmailAlias { + description: None, + domain_id, + enabled: true, + name: "john.doe".to_string(), + } + ); + assert_eq!( + aliases[1], + &EmailAlias { + description: None, + domain_id, + enabled: true, + name: "j.doe".to_string(), + } + ); + assert_eq!(account_out.member_group_ids.len(), 2); + for (idx, group_id) in account_out.member_group_ids.iter().enumerate() { + let group = admin + .registry_get::(*group_id) + .await + .into_group() + .unwrap(); + assert_eq!(group.name, if idx == 0 { "corporate" } else { "sales" }); + assert_eq!(group.domain_id, domain_id); + } + assert_eq!( + test.server + .registry() + .count_object(ObjectType::Account) + .await + .unwrap(), + 3 + ); + assert_eq!( + test.server + .registry() + .count_object(ObjectType::Domain) + .await + .unwrap(), + 1 + ); + + // No changes should not cause any updates + assert_eq!( + test.server + .synchronize_account(account_in.clone()) + .await + .unwrap() + .id, + account_id.document_id() + ); + assert_eq!( + test.server + .registry() + .object::(account_id) + .await + .unwrap() + .unwrap() + .into_user() + .unwrap(), + account_out + ); + assert_eq!( + test.server + .registry() + .count_object(ObjectType::Account) + .await + .unwrap(), + 3 + ); + + // Make some changes and synchronize again + account_in.description = "Johnathan Doe".to_string().into(); + account_in + .email_aliases + .push("johnny@example.org".to_string()); + account_in.groups.pop(); + account_in.groups.push("support@example.org".to_string()); + account_in.secret = "evenmoresecret".to_string().into(); + assert_eq!( + test.server + .synchronize_account(account_in.clone()) + .await + .unwrap() + .id, + account_id.document_id() + ); + let account_out = test + .server + .registry() + .object::(account_id) + .await + .unwrap() + .unwrap() + .into_user() + .unwrap(); + assert_eq!( + account_out + .credentials + .values() + .next() + .and_then(|v| v.as_main_credential()) + .map(|c| c.secret.as_str()), + Some("evenmoresecret") + ); + assert_eq!(account_out.description.as_deref(), Some("Johnathan Doe")); + assert_eq!(account_out.aliases.len(), 3); + let aliases = account_out.aliases.iter().collect::>(); + assert_eq!( + aliases[2], + &EmailAlias { + description: None, + domain_id, + enabled: true, + name: "johnny".to_string(), + } + ); + assert_eq!(account_out.member_group_ids.len(), 2); + let account_groups = account_out + .member_group_ids + .iter() + .copied() + .collect::>(); + for (idx, group_id) in account_groups.iter().enumerate() { + let group = admin + .registry_get::(*group_id) + .await + .into_group() + .unwrap(); + assert_eq!(group.name, if idx == 0 { "corporate" } else { "support" }); + assert_eq!(group.domain_id, domain_id); + } + assert_eq!( + test.server + .registry() + .count_object(ObjectType::Account) + .await + .unwrap(), + 4 + ); + + // Synchronize a group + assert_eq!( + test.server + .synchronize_group(directory::Group { + email: "corporate@example.org".to_string(), + email_aliases: vec!["everyone@example.org".to_string()], + description: "Corporate Group".to_string().into(), + }) + .await + .unwrap(), + account_groups[0].document_id() + ); + let group_out = test + .server + .registry() + .object::(account_groups[0]) + .await + .unwrap() + .unwrap() + .into_group() + .unwrap(); + assert_eq!(group_out.name, "corporate"); + assert_eq!(group_out.description.as_deref(), Some("Corporate Group")); + assert_eq!(group_out.aliases.len(), 1); + let aliases = group_out.aliases.iter().collect::>(); + assert_eq!( + aliases[0], + &EmailAlias { + description: None, + domain_id, + enabled: true, + name: "everyone".to_string(), + } + ); + assert_eq!( + test.server + .registry() + .count_object(ObjectType::Account) + .await + .unwrap(), + 4 + ); +} diff --git a/tests/src/imap/idle.rs b/tests/src/imap/idle.rs index b22a7deb..0aa47f9c 100644 --- a/tests/src/imap/idle.rs +++ b/tests/src/imap/idle.rs @@ -163,7 +163,7 @@ pub async fn test( .assert_contains("* 0 EXISTS"); // Test SMTP delivery notifications - let mut lmtp = SmtpConnection::connect_port(if is_cluster_test { 17000 } else { 11200 }).await; + let mut lmtp = SmtpConnection::connect().await; lmtp.ingest( "bill@example.com", &["jdoe@example.com"], diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index 2c390f1f..404c7607 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -41,7 +41,7 @@ use serde_json::json; use std::{path::PathBuf, time::Instant}; use utils::map::vec_map::VecMap; -#[tokio::test] +#[tokio::test(flavor = "multi_thread")] pub async fn imap_tests() { let mut test = TestServerBuilder::new("imap_tests") .await diff --git a/tests/src/lib.rs b/tests/src/lib.rs index 974d2f4a..2684ad5d 100644 --- a/tests/src/lib.rs +++ b/tests/src/lib.rs @@ -13,10 +13,8 @@ use jemallocator::Jemalloc; #[global_allocator] static GLOBAL: Jemalloc = Jemalloc; -/* #[cfg(test)] pub mod cluster; -*/ #[cfg(test)] pub mod directory; #[cfg(test)] @@ -45,7 +43,6 @@ pub trait AssertConfig { #[cfg(test)] impl AssertConfig for Bootstrap { fn assert_no_errors(self) -> Self { - let todo = "cluster tests"; if !self.errors.is_empty() { panic!("Errors: {:#?}", self.errors); } diff --git a/tests/src/utils/cleanup.rs b/tests/src/utils/cleanup.rs index 05ff9a76..57183e73 100644 --- a/tests/src/utils/cleanup.rs +++ b/tests/src/utils/cleanup.rs @@ -101,10 +101,9 @@ async fn store_destroy_sql_indexes(store: &Store) { #[cfg(feature = "mysql")] let table = index.mysql_table(); - store + let _ = store .sql_query::(&format!("TRUNCATE TABLE {table}"), vec![]) - .await - .unwrap(); + .await; } } } diff --git a/tests/src/utils/jmap.rs b/tests/src/utils/jmap.rs index 57c537bd..aeed683a 100644 --- a/tests/src/utils/jmap.rs +++ b/tests/src/utils/jmap.rs @@ -44,19 +44,35 @@ impl Account { .into_iter() .map(|id| Value::String(id.to_string())) .collect::>(); - self.jmap_method_calls(json!([[ - format!("{object}/get"), - { - "accountId": account.id_string(), - "properties": properties - .into_iter() - .map(|p| Value::String(p.to_string())) - .collect::>(), - "ids": if !ids.is_empty() { Some(ids) } else { None } - }, - "0" - ]])) - .await + + if account.id().document_id() != u32::MAX { + self.jmap_method_calls(json!([[ + format!("{object}/get"), + { + "accountId": account.id_string(), + "properties": properties + .into_iter() + .map(|p| Value::String(p.to_string())) + .collect::>(), + "ids": if !ids.is_empty() { Some(ids) } else { None } + }, + "0" + ]])) + .await + } else { + self.jmap_method_calls(json!([[ + format!("{object}/get"), + { + "properties": properties + .into_iter() + .map(|p| Value::String(p.to_string())) + .collect::>(), + "ids": if !ids.is_empty() { Some(ids) } else { None } + }, + "0" + ]])) + .await + } } pub async fn jmap_query( diff --git a/tests/src/utils/server.rs b/tests/src/utils/server.rs index cad99810..442a308f 100644 --- a/tests/src/utils/server.rs +++ b/tests/src/utils/server.rs @@ -40,10 +40,10 @@ use pop3::Pop3SessionManager; use registry::{ schema::{ enums::{DataStoreType, EventPolicy, NetworkListenerProtocol, TracingLevel}, - prelude::{Object, SocketAddr}, + prelude::{Object, ObjectType, SocketAddr}, structs::{ - Certificate, Expression, Http, NetworkListener, PublicText, SecretKeyFile, SecretText, - Tracer, TracerStdout, + Authentication, Certificate, Expression, Http, NetworkListener, PublicText, + SecretKeyFile, SecretText, Tracer, TracerStdout, }, }, types::{EnumImpl, map::Map}, @@ -61,7 +61,7 @@ use smtp::{ use std::{path::PathBuf, str::FromStr, sync::Arc}; use store::{ RegistryStore, Store, ValueKey, - registry::{bootstrap::Bootstrap, write::RegistryWrite}, + registry::{RegistryQuery, bootstrap::Bootstrap, write::RegistryWrite}, write::{AlignedBytes, Archive}, }; use tokio::sync::{mpsc, watch}; @@ -92,6 +92,16 @@ pub struct TestServerBuilder { impl TestServerBuilder { pub async fn new(test_name: &str) -> Self { let reset = std::env::var("NO_INSERT").is_err(); + + Self::new_with_role(test_name, "mail.example.org".to_string(), None, reset).await + } + + pub async fn new_with_role( + test_name: &str, + hostname: String, + node_role: Option, + reset: bool, + ) -> Self { let temp_dir = TempDir::new(test_name, reset); let path = temp_dir.path.to_string_lossy().to_string(); let data_store = build_data_store( @@ -114,7 +124,7 @@ impl TestServerBuilder { Self { bootstrap: Bootstrap::new( - RegistryStore::new(&path, store, "mail.example.org".to_string(), 1, None).await, + RegistryStore::new(&path, store, hostname, 1, node_role).await, ) .await, http_listener_port: 8899, @@ -149,8 +159,7 @@ impl TestServerBuilder { .await } - pub async fn with_http_listener(mut self, port: u16) -> Self { - self.http_listener_port = port; + pub async fn with_http_listener(self, port: u16) -> Self { self.with_listener(NetworkListenerProtocol::Http, "jmap", port, true) .await .with_object(Http { @@ -169,6 +178,11 @@ impl TestServerBuilder { .await } + pub async fn with_imap_listener(self, port: u16) -> Self { + self.with_listener(NetworkListenerProtocol::Imap, "imap", port, false) + .await + } + pub async fn with_dummy_tls_cert(self) -> Self { let mut cert_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); cert_path.push("resources"); @@ -191,12 +205,15 @@ impl TestServerBuilder { } pub async fn with_listener( - self, + mut self, protocol: NetworkListenerProtocol, name: &str, port: u16, tls_implicit: bool, ) -> Self { + if protocol == NetworkListenerProtocol::Http { + self.http_listener_port = port; + } self.insert_object(NetworkListener { bind: Map::new(vec![ SocketAddr::from_str(&format!("127.0.0.1:{port}")).unwrap(), @@ -245,59 +262,91 @@ impl TestServerBuilder { .unwrap_id(trc::location!()) } - pub async fn build(mut self) -> TestServer { - // Register stores from environment - self.bootstrap.registry.insert_stores_from_env().await; + pub async fn build(self) -> TestServer { + self.build_with_opts(true).await + } - // Enable logging if requested - let level = std::env::var("LOG") - .map(|log| TracingLevel::parse(&log).expect("Invalid log level")) - .ok(); + pub async fn build_with_opts(mut self, init_store: bool) -> TestServer { + if init_store { + // Register stores from environment + self.bootstrap.registry.insert_stores_from_env().await; - self.insert_object(Tracer::Stdout(TracerStdout { - enable: level.is_some() || self.logging_enabled, - level: level.unwrap_or(TracingLevel::Info), - ansi: true, - multiline: false, - events: Map::new( - EventType::variants() - .iter() - .filter(|ev| { - let ev = ev.as_str(); - ev.starts_with("network.") - || ev.starts_with("http.connection-") - || ev == "telemetry.webhook-error" - || ev == "http.request-body" - || ev == "http.request-url" - || ev == "tls.no-certificates-available" - || ev == "store.cache-hit" - }) - .copied() - .collect(), - ), - events_policy: EventPolicy::Exclude, - ..Default::default() - })) - .await; + // Enable logging if requested + let level = std::env::var("LOG") + .map(|log| TracingLevel::parse(&log).expect("Invalid log level")) + .ok(); + + self.insert_object(Tracer::Stdout(TracerStdout { + enable: level.is_some() || self.logging_enabled, + level: level.unwrap_or(TracingLevel::Info), + ansi: true, + multiline: false, + events: Map::new( + EventType::variants() + .iter() + .filter(|ev| { + let ev = ev.as_str(); + ev.starts_with("network.") + || ev.starts_with("http.connection-") + || ev == "telemetry.webhook-error" + || ev == "http.request-body" + || ev == "http.request-url" + || ev == "tls.no-certificates-available" + || ev == "store.cache-hit" + }) + .copied() + .collect(), + ), + events_policy: EventPolicy::Exclude, + ..Default::default() + })) + .await; + } // Start listeners let mut servers = Listeners::parse(&mut self.bootstrap).await; servers.bind_and_drop_priv(&mut self.bootstrap); + if init_store { + // Add safe defaults if missing + self.bootstrap.insert_safe_defaults().await; + + // Add directory + if let Some(directory_id) = self + .bootstrap + .registry + .query::>(RegistryQuery::new(ObjectType::Directory)) + .await + .unwrap() + .first() + { + let mut auth = self + .bootstrap + .registry + .object::(Id::singleton()) + .await + .unwrap() + .unwrap(); + auth.directory_id = Some(*directory_id); + self.bootstrap + .registry + .write(RegistryWrite::insert(&auth.into())) + .await + .unwrap(); + } + } + // Parse storage let storage = Storage::parse(&mut self.bootstrap).await; // Reset search store - if self.reset { + if init_store && self.reset { search_store_destroy(&storage.search).await; } // Parse telemetry let telemetry = Telemetry::parse(&mut self.bootstrap, &storage).await; - // Add safe defaults if missing - self.bootstrap.insert_safe_defaults().await; - // Parse components let core = Box::pin(Core::parse(&mut self.bootstrap, storage)).await; let data = Data::parse(&mut self.bootstrap).await; diff --git a/tests/src/utils/storage.rs b/tests/src/utils/storage.rs index bd00d79b..3f7e483e 100644 --- a/tests/src/utils/storage.rs +++ b/tests/src/utils/storage.rs @@ -73,9 +73,9 @@ pub fn build_data_store(typ: DataStoreType, path: &str) -> DataStore { DataStoreType::PostgreSql => DataStore::PostgreSql(PostgreSqlStore { host: "localhost".into(), port: 5432, - auth_username: "postgres".to_string().into(), + auth_username: "stalwart".to_string().into(), auth_secret: SecretKeyOptional::Value(SecretKeyValue { - secret: "mysecretpassword".into(), + secret: "stalwart".into(), }), database: "stalwart".into(), use_tls: false,