From f6c91041d948dfc631210844330b0bbeca092404 Mon Sep 17 00:00:00 2001 From: Maurus Decimus <11444311+mdecimus@users.noreply.github.com> Date: Tue, 14 Apr 2026 20:28:47 +0200 Subject: [PATCH] Applications, Cluster node management and minor fixes --- Cargo.lock | 1 + crates/common/src/auth/permissions.rs | 1 + crates/common/src/cache/reload.rs | 6 + crates/common/src/config/inner.rs | 9 +- crates/common/src/config/mailstore/email.rs | 19 +- crates/common/src/enterprise/mod.rs | 77 +++-- crates/common/src/lib.rs | 5 +- crates/common/src/manager/application.rs | 282 +++++++++++++----- crates/common/src/manager/defaults.rs | 54 +++- crates/common/src/manager/mod.rs | 1 - crates/common/src/telemetry/metrics/store.rs | 2 +- crates/common/src/telemetry/tracers/store.rs | 2 +- crates/email/src/message/ingest.rs | 3 +- crates/groupware/src/calendar/storage.rs | 3 +- crates/http-proto/src/response.rs | 16 + crates/http/src/auth/permissions.rs | 3 + crates/http/src/request.rs | 27 +- crates/jmap/src/registry/get.rs | 4 +- crates/jmap/src/registry/mapping/action.rs | 24 +- .../src/registry/mapping/archived_item.rs | 3 +- crates/jmap/src/registry/mapping/cluster.rs | 78 +++++ crates/jmap/src/registry/mapping/mod.rs | 1 + .../jmap/src/registry/mapping/spam_sample.rs | 3 +- crates/jmap/src/registry/mapping/task.rs | 1 - crates/jmap/src/registry/query.rs | 13 +- crates/jmap/src/registry/set.rs | 10 +- crates/main/src/main.rs | 2 +- crates/main/src/test_data.rs | 6 +- crates/registry/Cargo.toml | 1 + crates/registry/src/pickle.rs | 64 +++- crates/registry/src/schema/mod.rs | 1 + crates/registry/src/types/mod.rs | 9 +- crates/registry/src/utils/task.rs | 1 + crates/services/src/lib.rs | 16 +- crates/services/src/task_manager/index.rs | 3 +- crates/services/src/task_manager/manager.rs | 3 +- crates/smtp/src/queue/spool.rs | 3 +- crates/smtp/src/reporting/dmarc.rs | 3 +- crates/smtp/src/reporting/tls.rs | 3 +- crates/store/src/build/registry.rs | 47 +++ crates/store/src/registry/get.rs | 19 +- crates/store/src/registry/mod.rs | 107 +++---- crates/store/src/registry/write.rs | 3 +- crates/store/src/write/batch.rs | 5 +- crates/trc/src/event/enums.rs | 4 +- crates/trc/src/event/enums_impl.rs | 21 ++ tests/src/store/registry.rs | 30 +- 47 files changed, 745 insertions(+), 254 deletions(-) create mode 100644 crates/jmap/src/registry/mapping/cluster.rs diff --git a/Cargo.lock b/Cargo.lock index 41e90fdc..22c146b3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6022,6 +6022,7 @@ dependencies = [ "ahash", "hashify", "jmap-tools", + "lz4_flex 0.13.0", "mail-auth", "serde", "serde_json", diff --git a/crates/common/src/auth/permissions.rs b/crates/common/src/auth/permissions.rs index f6a623a7..b02f11a6 100644 --- a/crates/common/src/auth/permissions.rs +++ b/crates/common/src/auth/permissions.rs @@ -289,6 +289,7 @@ impl Default for DefaultPermissions { default.superuser.push(permission); } else if name.starts_with("sysDomain") || name.starts_with("sysDkimSignature") + || name.starts_with("sysAcmeProvider") || name.starts_with("sysAccount") || name.starts_with("sysRole") || name.starts_with("sysOAuthClient") diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index 72405218..a4f61511 100644 --- a/crates/common/src/cache/reload.rs +++ b/crates/common/src/cache/reload.rs @@ -89,6 +89,12 @@ impl Server { *self.inner.data.blocked_ips.write() = blocked_ips; } } + ObjectType::Application => { + self.inner.data.applications.reload(&mut bootstrap).await; + if bootstrap.errors.is_empty() { + self.inner.data.applications.unpack_all(self, false).await; + } + } _ => { // Load stores let directory = Directories::build(&mut bootstrap).await; diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index 814e40c1..f967ef63 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -17,6 +17,7 @@ use crate::{ resolver::{Policy, Tlsa}, }, }, + manager::application::WebApplications, network::security::BlockedIps, }; use ahash::{AHashMap, AHashSet}; @@ -54,7 +55,9 @@ impl Data { panic!("Invalid system time, panicking to avoid data corruption"); } - let todo = "TODO: WebApplicationManager initialization"; + // Initialize apps + let applications = WebApplications::new(); + applications.reload(bp).await; let blocked_ips = BlockedIps::parse(bp).await; let lookup_stores = LookupStores::build(bp).await; @@ -84,7 +87,7 @@ impl Data { registry_id_gen: id_generator.clone(), span_id_gen: id_generator, queue_status: true.into(), - applications: Default::default(), + applications, logos: Default::default(), smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"), asn_geo_data: Default::default(), @@ -224,7 +227,7 @@ impl Default for Data { span_id_gen: Default::default(), registry_id_gen: Default::default(), queue_status: true.into(), - applications: Default::default(), + applications: WebApplications::new(), logos: Default::default(), smtp_connectors: TlsConnectors::try_new().unwrap(), asn_geo_data: Default::default(), diff --git a/crates/common/src/config/mailstore/email.rs b/crates/common/src/config/mailstore/email.rs index 5f3dae44..20b38831 100644 --- a/crates/common/src/config/mailstore/email.rs +++ b/crates/common/src/config/mailstore/email.rs @@ -14,8 +14,8 @@ use registry::{ }, prelude::ObjectType, structs::{ - AddressBook, Authentication, Calendar, DataRetention, Domain, Email, Jmap, Search, - SieveUserInterpreter, SystemSettings, + AddressBook, Authentication, Calendar, DataRetention, Domain, Email, FileStorage, Jmap, + Search, SieveUserInterpreter, SystemSettings, }, }, types::EnumImpl, @@ -83,6 +83,7 @@ impl EmailConfig { let sieve = bp.setting_infallible::().await; let search = bp.setting_infallible::().await; let jmap = bp.setting_infallible::().await; + let file = bp.setting_infallible::().await; let calendar = bp.setting_infallible::().await; let address_book = bp.setting_infallible::().await; let system = bp.setting_infallible::().await; @@ -108,23 +109,35 @@ impl EmailConfig { }; // Parse default object quotas - let todo = "make sure all are configurable"; let mut max_objects = ObjectQuota::default(); for (item, max) in [ + (StorageQuota::MaxEmails, email.max_messages), (StorageQuota::MaxMailboxes, email.max_mailboxes), (StorageQuota::MaxSieveScripts, sieve.max_scripts), (StorageQuota::MaxEmailIdentities, email.max_identities), (StorageQuota::MaxEmailSubmissions, email.max_submissions), (StorageQuota::MaxMaskedAddresses, email.max_masked_addresses), (StorageQuota::MaxAppPasswords, auth.max_app_passwords), + (StorageQuota::MaxApiKeys, auth.max_api_keys), + (StorageQuota::MaxPublicKeys, email.max_public_keys), (StorageQuota::MaxPushSubscriptions, jmap.max_subscriptions), (StorageQuota::MaxCalendars, calendar.max_calendars), (StorageQuota::MaxCalendarEvents, calendar.max_events), + ( + StorageQuota::MaxParticipantIdentities, + calendar.max_participant_identities, + ), + ( + StorageQuota::MaxCalendarEventNotifications, + calendar.max_event_notifications, + ), ( StorageQuota::MaxAddressBooks, address_book.max_address_books, ), (StorageQuota::MaxContactCards, address_book.max_contacts), + (StorageQuota::MaxFiles, file.max_files), + (StorageQuota::MaxFolders, file.max_folders), ] { if let Some(max) = max { max_objects.set(item, max as u32); diff --git a/crates/common/src/enterprise/mod.rs b/crates/common/src/enterprise/mod.rs index b68cde9d..60b82b2b 100644 --- a/crates/common/src/enterprise/mod.rs +++ b/crates/common/src/enterprise/mod.rs @@ -15,8 +15,8 @@ pub mod llm; pub mod masked; use crate::{ - Core, LogoCache, Server, config::groupware::CalendarTemplateVariable, expr::Expression, - manager::application::Resource, + Core, LogoCache, Server, USER_AGENT, config::groupware::CalendarTemplateVariable, + expr::Expression, manager::application::Resource, }; use ahash::{AHashMap, AHashSet}; use license::LicenseKey; @@ -161,42 +161,61 @@ impl Server { return Ok(None); } - let domain = psl::domain_str(domain).unwrap_or(domain); - let logo = { self.inner.data.logos.lock().get(domain).cloned() }; - if let Some(logo) = logo { + let mut domain = psl::domain_str(domain).unwrap_or(domain); + let logo_cache = { self.inner.data.logos.lock().get(domain).cloned() }; + if let Some(logo) = logo_cache { return Ok(logo.data); } - let Some((domain_id, tenant_id)) = self.domain(domain).await?.map(|d| (d.id, d.id_tenant)) - else { - return Ok(None); - }; - - let Some(domain_record) = self.registry().object::(domain_id.into()).await? else { - return Ok(None); - }; - let mut logo = domain_record.logo; - - if logo.is_none() - && let Some(tenant_id) = tenant_id + let mut logo_url = None; + let mut domain_id = u32::MAX; + let mut tenant_id = None; + if let Some((d_id, t_id)) = self.domain(domain).await?.map(|d| (d.id, d.id_tenant)) + && let Some(domain_record) = self.registry().object::(domain_id.into()).await? { - logo = self - .registry() - .object::(tenant_id.into()) - .await? - .and_then(|t| t.logo); + logo_url = domain_record.logo; + domain_id = d_id; + tenant_id = t_id; + + if logo_url.is_none() + && let Some(tenant_id) = tenant_id + { + logo_url = self + .registry() + .object::(tenant_id.into()) + .await? + .and_then(|t| t.logo); + } + } else { + domain = "*"; } - let logo_url = logo.or_else(|| self.default_logo_url()); + // Try fetching the default logo + if logo_url.is_none() + && let Some(default_logo_url) = self.default_logo_url() + { + let logo = { self.inner.data.logos.lock().get("*").cloned() }; + if let Some(logo) = logo { + return Ok(logo.data); + } + logo_url = Some(default_logo_url); + } let mut logo = None; if let Some(logo_url) = logo_url { - let response = reqwest::get(logo_url.as_str()).await.map_err(|err| { - trc::ResourceEvent::DownloadExternal - .into_err() - .details("Failed to download logo") - .reason(err) - })?; + let response = reqwest::Client::builder() + .user_agent(USER_AGENT) + .build() + .unwrap() + .get(logo_url.as_str()) + .send() + .await + .map_err(|err| { + trc::ResourceEvent::DownloadExternal + .into_err() + .details("Failed to download logo") + .reason(err) + })?; let content_type = response .headers() diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 786bbd65..d823774e 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -7,6 +7,7 @@ #![warn(clippy::large_futures)] use crate::auth::{AccessTokenInner, EmailAddress}; +use crate::manager::application::WebApplications; use crate::network::asn::AsnGeoLookupData; use crate::{ auth::{AccountCache, DomainCache, EmailCache, MailingListCache, RoleCache, TenantCache}, @@ -39,7 +40,7 @@ use config::{ }; use ipc::{BroadcastEvent, PushEvent, QueueEvent, ReportingEvent}; use mail_auth::{MX, Txt}; -use manager::application::{Resource, WebApplicationManager}; +use manager::application::Resource; use parking_lot::{Mutex, RwLock}; use rustls::sign::CertifiedKey; use std::sync::atomic::AtomicU64; @@ -157,7 +158,7 @@ pub struct Data { pub registry_id_gen: SnowflakeIdGenerator, pub queue_status: AtomicBool, - pub applications: WebApplicationManager, + pub applications: WebApplications, pub logos: Mutex, LogoCache>>, pub smtp_connectors: TlsConnectors, diff --git a/crates/common/src/manager/application.rs b/crates/common/src/manager/application.rs index 312ad937..28d45604 100644 --- a/crates/common/src/manager/application.rs +++ b/crates/common/src/manager/application.rs @@ -4,20 +4,41 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::WEBADMIN_KEY; -use crate::Core; +use crate::{Server, manager::fetch_resource}; use ahash::AHashMap; use arc_swap::ArcSwap; +use registry::schema::{enums::CompressionAlgo, structs::Application}; use std::{ borrow::Cow, io::{self, Cursor, Read}, path::PathBuf, + sync::Arc, + time::Duration, }; -use store::BlobStore; +use store::{ + registry::{RegistryObject, bootstrap::Bootstrap}, + write::{BatchBuilder, BlobLink, BlobOp, now}, +}; +use trc::{AddContext, Key}; +use types::blob_hash::BlobHash; +const APP_BLOB_PREFIX: &str = "STALWART_APP_"; +const MAX_APP_SIZE: usize = 100 * 1024 * 1024; + +#[allow(clippy::type_complexity)] +pub struct WebApplications { + applications: ArcSwap>, + routes: ArcSwap>>>>, +} + +#[derive(Clone)] pub struct WebApplicationManager { bundle_path: TempDir, - routes: ArcSwap>>, + prefixes: Vec, + description: String, + url: String, + expiry: u64, + blob_key: BlobHash, } #[derive(Default, Clone)] @@ -35,22 +56,42 @@ impl Resource { } } -impl WebApplicationManager { - pub fn new(base_path: PathBuf) -> Self { +pub struct AppResource { + pub resource: Resource>, + pub no_cache: bool, +} + +impl WebApplications { + pub fn new() -> Self { Self { - bundle_path: TempDir::new(base_path), - routes: ArcSwap::from_pointee(Default::default()), + applications: ArcSwap::new(Arc::new(Vec::new())), + routes: ArcSwap::new(Arc::new(AHashMap::new())), } } - pub async fn get(&self, path: &str) -> trc::Result>> { - let routes = self.routes.load(); - if let Some(resource) = routes.get(path).or_else(|| routes.get("index.html")) { + pub async fn serve(&self, prefix: &str, path: &str) -> trc::Result> { + if let Some(routes) = self.routes.load().get(prefix) + && let Some((is_index, resource)) = routes + .get(path) + .map(|res| (path == "index.html", res)) + .or_else(|| routes.get("index.html").map(|res| (true, res))) + { tokio::fs::read(&resource.contents) .await - .map(|contents| Resource { - content_type: resource.content_type.clone(), - contents, + .map(|mut contents| { + if is_index && let Ok(html) = std::str::from_utf8(&contents) { + contents = html + .replace(" trc::Result<()> { + pub async fn reload(&self, bp: &mut Bootstrap) { + let mut apps = Vec::new(); + for app in bp.list_infallible::().await { + if app.object.enabled { + apps.push(WebApplicationManager::new(app)); + } + } + self.applications.store(Arc::new(apps)); + } + + pub async fn unpack_all(&self, server: &Server, update: bool) { + let mut routes = AHashMap::new(); + for app in self.applications.load().as_ref() { + if update && let Err(err) = app.delete(server).await { + trc::event!( + Resource(trc::ResourceEvent::Error), + Reason = err, + Url = app.url.clone(), + Details = format!( + "Failed to delete application bundle for prefixes: {}", + app.prefixes.join(", ") + ) + ); + } + match app.unpack(server).await { + Ok(app_routes) => { + let app_routes = Arc::new(app_routes); + + for prefix in &app.prefixes { + routes.insert(prefix.clone(), app_routes.clone()); + } + } + Err(err) => { + trc::event!( + Resource(trc::ResourceEvent::Error), + Reason = err, + Url = app.url.clone(), + Details = format!( + "Failed to unpack application for prefixes: {}", + app.prefixes.join(", ") + ) + ); + } + } + } + self.routes.store(Arc::new(routes)); + } +} + +impl WebApplicationManager { + pub fn new(app: RegistryObject) -> Self { + let base_path = app + .object + .unpack_directory + .map(PathBuf::from) + .unwrap_or_else(std::env::temp_dir) + .join(app.id.id().to_string()); + + Self { + bundle_path: TempDir::new(base_path), + blob_key: BlobHash::generate(format!("{}{}", APP_BLOB_PREFIX, app.id.id()).as_bytes()), + url: app.object.resource_url, + description: app.object.description, + expiry: app.object.auto_update_frequency.as_secs(), + prefixes: app + .object + .url_prefix + .iter() + .map(|prefix| { + prefix + .trim_end_matches('/') + .trim_start_matches('/') + .to_string() + }) + .collect(), + } + } + + async fn unpack(&self, server: &Server) -> trc::Result>> { // Delete any existing bundles self.bundle_path.clean().await.map_err(unpack_error)?; - // Obtain webadmin bundle - let bundle = blob_store - .get_blob(WEBADMIN_KEY, 0..usize::MAX) + // Obtain application bundle + let bundle = if let Some(bundle) = server + .blob_store() + .get_blob(self.blob_key.as_slice(), 0..usize::MAX) .await? - .ok_or_else(|| { - trc::ResourceEvent::NotFound - .caused_by(trc::location!()) - .details("Webadmin bundle not found") - })?; + { + bundle + } else { + // Fetch app bundle + let resource = fetch_resource(&self.url, None, Duration::from_secs(60), MAX_APP_SIZE) + .await + .map_err(|err| { + trc::ResourceEvent::Error + .caused_by(trc::location!()) + .ctx(Key::Url, self.url.clone()) + .reason(err) + .details("Failed to fetch application bundle") + })?; + + // Store in blob store for future use + server + .blob_store() + .put_blob(self.blob_key.as_slice(), &resource, CompressionAlgo::None) + .await + .caused_by(trc::location!())?; + + // Schedule expiration + let mut batch = BatchBuilder::new(); + batch + .set( + BlobOp::Link { + hash: self.blob_key.clone(), + to: BlobLink::Temporary { + until: now() + self.expiry, + }, + }, + vec![], + ) + .set( + BlobOp::Commit { + hash: self.blob_key.clone(), + }, + Vec::new(), + ); + server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + + trc::event!( + Resource(trc::ResourceEvent::ApplicationUpdated), + Url = self.url.clone(), + Details = self.description.clone(), + ); + + resource + }; // Uncompress let mut bundle = zip::ZipArchive::new(Cursor::new(bundle)).map_err(|err| { trc::ResourceEvent::Error .caused_by(trc::location!()) .reason(err) - .details("Failed to decompress webadmin bundle") + .ctx(Key::Url, self.url.clone()) + .details("Failed to decompress application bundle") })?; let mut routes = AHashMap::new(); for i in 0..bundle.len() { @@ -91,7 +260,7 @@ impl WebApplicationManager { trc::ResourceEvent::Error .caused_by(trc::location!()) .reason(err) - .details("Failed to read file from webadmin bundle") + .details("Failed to read file from application bundle") })?; if file.is_dir() { continue; @@ -129,37 +298,21 @@ impl WebApplicationManager { routes.insert(file_name, resource); } - // Update routes - self.routes.store(routes.into()); - - let todo = "use new event"; - /*trc::event!( - Resource(trc::ResourceEvent::WebadminUnpacked), + trc::event!( + Resource(trc::ResourceEvent::ApplicationUnpacked), + Url = self.url.clone(), Path = self.bundle_path.path.to_string_lossy().into_owned(), - );*/ + ); - Ok(()) + Ok(routes) } - pub async fn update(&self, core: &Core) -> trc::Result<()> { - todo!() - /* let bytes = core - .storage - .config - .fetch_resource("webadmin") + async fn delete(&self, server: &Server) -> trc::Result<()> { + server + .blob_store() + .delete_blob(self.blob_key.as_slice()) .await - .map_err(|err| { - trc::ResourceEvent::Error - .caused_by(trc::location!()) - .reason(err) - .details("Failed to download webadmin") - })?; - core.storage.blob.put_blob(WEBADMIN_KEY, &bytes).await*/ - } - - pub async fn update_and_unpack(&self, core: &Core) -> trc::Result<()> { - self.update(core).await?; - self.unpack(&core.storage.blob).await + .map(|_| ()) } } @@ -169,15 +322,14 @@ impl Resource> { } } +#[derive(Clone)] pub struct TempDir { pub path: PathBuf, } impl TempDir { pub fn new(path: PathBuf) -> TempDir { - TempDir { - path: path.join(std::str::from_utf8(WEBADMIN_KEY).unwrap()), - } + TempDir { path } } pub async fn clean(&self) -> io::Result<()> { @@ -191,19 +343,7 @@ impl TempDir { fn unpack_error(err: std::io::Error) -> trc::Error { trc::ResourceEvent::Error .reason(err) - .details("Failed to unpack webadmin bundle") -} - -impl Default for WebApplicationManager { - fn default() -> Self { - Self::new(std::env::temp_dir()) - } -} - -impl Default for TempDir { - fn default() -> Self { - Self::new(std::env::temp_dir()) - } + .details("Failed to unpack application bundle") } impl Drop for TempDir { @@ -211,3 +351,9 @@ impl Drop for TempDir { let _ = std::fs::remove_dir_all(&self.path); } } + +impl Default for WebApplications { + fn default() -> Self { + Self::new() + } +} diff --git a/crates/common/src/manager/defaults.rs b/crates/common/src/manager/defaults.rs index 5f8336c5..1c7d8630 100644 --- a/crates/common/src/manager/defaults.rs +++ b/crates/common/src/manager/defaults.rs @@ -24,9 +24,7 @@ use store::{ bootstrap::Bootstrap, write::{RegistryWrite, RegistryWriteResult}, }, - write::BatchBuilder, }; -use types::id::Id; pub const ASN_IPV4: &str = "https://cdn.jsdelivr.net/npm/@ip-location-db/asn/asn-ipv4.csv"; pub const ASN_IPV6: &str = "https://cdn.jsdelivr.net/npm/@ip-location-db/asn/asn-ipv6.csv"; @@ -473,19 +471,47 @@ async fn insert_safe_defaults(bp: &mut Bootstrap) -> trc::Result<()> { .await?; } - if bp.registry.count_object(ObjectType::SpamRule).await? == 0 - && bp - .registry - .object::(Id::singleton()) - .await? - .is_none_or(|spam| spam.spam_filter_rules_url.is_some()) + #[cfg(not(feature = "test_mode"))] + if bp.registry.count_object(ObjectType::Application).await? == 0 { + bp.registry + .write(RegistryWrite::insert( + &Application { + auto_update_frequency: Duration::from_millis(30 * 24 * 60 * 60 * 1000), + description: "Stalwart Web Interface".to_string(), + enabled: true, + #[cfg(not(feature = "dev_mode"))] + resource_url: + "https://github.com/stalwartlabs/webui/releases/latest/download/webui.zip" + .into(), + #[cfg(feature = "dev_mode")] + resource_url: "file:///Users/me/code/webui/.ignore/webui.zip".into(), + unpack_directory: None, + url_prefix: Map::new(vec!["/admin".into(), "/account".into()]), + } + .into(), + )) + .await?; + } + + #[cfg(not(feature = "test_mode"))] { - let mut batch = BatchBuilder::new(); - batch.schedule_task(Task::SpamFilterMaintenance(TaskSpamFilterMaintenance { - maintenance_type: TaskSpamFilterMaintenanceType::UpdateRules, - status: TaskStatus::now(), - })); - bp.data_store.write(batch.build_all()).await?; + use store::write::BatchBuilder; + use types::id::Id; + + if bp.registry.count_object(ObjectType::SpamRule).await? == 0 + && bp + .registry + .object::(Id::singleton()) + .await? + .is_none_or(|spam| spam.spam_filter_rules_url.is_some()) + { + let mut batch = BatchBuilder::new(); + batch.schedule_task(Task::SpamFilterMaintenance(TaskSpamFilterMaintenance { + maintenance_type: TaskSpamFilterMaintenanceType::UpdateRules, + status: TaskStatus::now(), + })); + bp.data_store.write(batch.build_all()).await?; + } } Ok(()) diff --git a/crates/common/src/manager/mod.rs b/crates/common/src/manager/mod.rs index e77b7a8f..09e18c8e 100644 --- a/crates/common/src/manager/mod.rs +++ b/crates/common/src/manager/mod.rs @@ -20,7 +20,6 @@ pub mod console; pub mod defaults; pub mod restore; -pub const WEBADMIN_KEY: &[u8] = "STALWART_WEBADMIN".as_bytes(); pub const SPAM_TRAINER_KEY: &[u8] = "STALWART_SPAM_TRAIN_DATA.lz4".as_bytes(); pub const SPAM_CLASSIFIER_KEY: &[u8] = "STALWART_SPAM_CLASSIFIER_MODEL.lz4".as_bytes(); diff --git a/crates/common/src/telemetry/metrics/store.rs b/crates/common/src/telemetry/metrics/store.rs index 70055241..6b53e3eb 100644 --- a/crates/common/src/telemetry/metrics/store.rs +++ b/crates/common/src/telemetry/metrics/store.rs @@ -11,8 +11,8 @@ use ahash::AHashMap; use parking_lot::Mutex; use registry::{ - pickle::Pickle, schema::structs::{Metric, MetricCount, MetricSum}, + types::ObjectImpl, }; use std::{future::Future, sync::Arc, time::Duration}; use store::{ diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index 01f37e4c..c2cccb39 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -11,11 +11,11 @@ use crate::{config::telemetry::StoreTracer, telemetry::tracers::TraceEvents}; use ahash::AHashMap; use registry::{ - pickle::Pickle, schema::structs::{ Task, TaskIndexTrace, TaskStatus, Trace, TraceKeyValue, TraceValue, TraceValueIpAddr, TraceValueList, TraceValueString, TraceValueUnsignedInt, }, + types::ObjectImpl, }; use std::{collections::HashSet, future::Future, time::Duration}; use store::{ diff --git a/crates/email/src/message/ingest.rs b/crates/email/src/message/ingest.rs index bf07905b..ee4b549a 100644 --- a/crates/email/src/message/ingest.rs +++ b/crates/email/src/message/ingest.rs @@ -24,13 +24,12 @@ use mail_parser::{ parsers::fields::thread::thread_name, }; use registry::{ - pickle::Pickle, schema::{ enums::IndexDocumentType, prelude::{ObjectType, Permission, Property}, structs::{SpamTrainingSample, Task, TaskIndexDocument, TaskMergeThreads, TaskStatus}, }, - types::{EnumImpl, datetime::UTCDateTime, id::ObjectId, map::Map}, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime, id::ObjectId, map::Map}, }; use std::future::Future; use std::{borrow::Cow, cmp::Ordering, fmt::Write, time::Instant}; diff --git a/crates/groupware/src/calendar/storage.rs b/crates/groupware/src/calendar/storage.rs index 1620a470..e566bc2b 100644 --- a/crates/groupware/src/calendar/storage.rs +++ b/crates/groupware/src/calendar/storage.rs @@ -22,9 +22,8 @@ use common::{ storage::index::ObjectIndexBuilder, }; use registry::{ - pickle::Pickle, schema::structs::{Task, TaskCalendarAlarmEmail, TaskCalendarAlarmNotification, TaskStatus}, - types::{EnumImpl, datetime::UTCDateTime}, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime}, }; use store::{ IterateParams, SerializeInfallible, U32_LEN, ValueKey, diff --git a/crates/http-proto/src/response.rs b/crates/http-proto/src/response.rs index e6221fb5..1707dc78 100644 --- a/crates/http-proto/src/response.rs +++ b/crates/http-proto/src/response.rs @@ -27,6 +27,15 @@ impl HttpResponse { } } + pub fn redirect(location: String) -> Self { + let mut response = HttpResponse::new(StatusCode::FOUND); + response.builder = response + .builder + .status(StatusCode::FOUND) + .header(header::LOCATION, location); + response + } + pub fn with_content_type(mut self, content_type: V) -> Self where V: TryInto, @@ -154,6 +163,13 @@ impl HttpResponse { self } + pub fn with_immutable_cache(mut self) -> Self { + self.builder = self + .builder + .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable"); + self + } + pub fn with_location(mut self, location: V) -> Self where V: TryInto, diff --git a/crates/http/src/auth/permissions.rs b/crates/http/src/auth/permissions.rs index 80602c74..bfc065f9 100644 --- a/crates/http/src/auth/permissions.rs +++ b/crates/http/src/auth/permissions.rs @@ -76,6 +76,9 @@ impl AccountApiHandler for Server { Permission::SysLogCreate, Permission::SysLogDestroy, Permission::SysLogUpdate, + Permission::SysClusterNodeCreate, + Permission::SysClusterNodeUpdate, + Permission::SysClusterNodeDestroy, ] { permissions.clear(p.to_id() as usize); } diff --git a/crates/http/src/request.rs b/crates/http/src/request.rs index aacb044b..4215eb9a 100644 --- a/crates/http/src/request.rs +++ b/crates/http/src/request.rs @@ -596,17 +596,28 @@ impl ParseHttp for Server { return Ok(HtmlResponse::new(page.to_string()).into_http_response()); } - _ => { - let path = req.uri().path(); - let resource = self + external => { + if path.next().is_none() { + return Ok(HttpResponse::redirect(format!("/{external}/"))); + } else if let Some(resource) = self .inner .data .applications - .get(path.strip_prefix('/').unwrap_or(path)) - .await?; - - if !resource.is_empty() { - return Ok(resource.into_http_response()); + .serve( + external, + req.uri() + .path() + .get(external.len() + 2..) + .unwrap_or_default(), + ) + .await? + { + let response = resource.resource.into_http_response(); + return Ok(if !resource.no_cache { + response.with_immutable_cache() + } else { + response.with_no_cache() + }); } } } diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index f9952981..b9b0aac3 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -7,7 +7,7 @@ use crate::registry::{ EnterpriseRegistry, mapping::{ - RegistryGetResponse, account::account_get, log::log_get, + RegistryGetResponse, account::account_get, cluster::cluster_node_get, log::log_get, queued_message::queued_message_get, report::report_get, spam_sample::spam_sample_get, task::task_get, }, @@ -282,7 +282,7 @@ impl RegistryGet for Server { queued_message_get(get).await.map(|get| get.into_response()) } ObjectType::Task => task_get(get).await.map(|get| get.into_response()), - + ObjectType::ClusterNode => cluster_node_get(get).await.map(|get| get.into_response()), ObjectType::ArfExternalReport | ObjectType::DmarcExternalReport | ObjectType::TlsExternalReport diff --git a/crates/jmap/src/registry/mapping/action.rs b/crates/jmap/src/registry/mapping/action.rs index a3cb4e32..ee350f6b 100644 --- a/crates/jmap/src/registry/mapping/action.rs +++ b/crates/jmap/src/registry/mapping/action.rs @@ -33,7 +33,7 @@ use spam_filter::{ analysis::{init::SpamFilterInit, score::SpamFilterAnalyzeScore}, }; use std::time::Instant; -use store::write::now; +use store::{registry::bootstrap::Bootstrap, write::now}; use utils::map::vec_map::VecMap; pub(crate) async fn action_set( @@ -207,6 +207,28 @@ pub(crate) async fn action_set( ); } } + Action::UpdateApps => { + let mut bp = Bootstrap::new_uninitialized(set.server.registry().clone()); + set.server.inner.data.applications.reload(&mut bp).await; + if bp.errors.is_empty() { + set.server + .inner + .data + .applications + .unpack_all(set.server, true) + .await; + set.server + .cluster_broadcast(BroadcastEvent::RegistryChange(RegistryChange::Reload( + ObjectType::Application, + ))) + .await; + set.response.created(id, now()); + } else { + set.response + .not_created + .append(id, map_bootstrap_error(bp.errors)); + } + } } } diff --git a/crates/jmap/src/registry/mapping/archived_item.rs b/crates/jmap/src/registry/mapping/archived_item.rs index 343c84be..0e86e020 100644 --- a/crates/jmap/src/registry/mapping/archived_item.rs +++ b/crates/jmap/src/registry/mapping/archived_item.rs @@ -19,13 +19,12 @@ use jmap_proto::{error::set::SetError, types::state::State}; use jmap_tools::{Key, Value}; use registry::{ jmap::IntoValue, - pickle::Pickle, schema::{ enums::{ArchivedItemStatus, Permission}, prelude::{Object, ObjectType, Property}, structs::{ArchivedItem, Task, TaskRestoreArchivedItem, TaskStatus}, }, - types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime, id::ObjectId}, }; use std::str::FromStr; use store::{ diff --git a/crates/jmap/src/registry/mapping/cluster.rs b/crates/jmap/src/registry/mapping/cluster.rs new file mode 100644 index 00000000..bfe05edf --- /dev/null +++ b/crates/jmap/src/registry/mapping/cluster.rs @@ -0,0 +1,78 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use jmap_proto::{object::registry::RegistryComparator, types::state::State}; +use registry::{jmap::IntoValue, schema::prelude::Property}; +use store::ahash::AHashSet; + +use crate::{ + api::query::QueryResponseBuilder, + registry::mapping::{RegistryGetResponse, RegistryQueryResponse}, +}; + +pub(crate) async fn cluster_node_get( + mut get: RegistryGetResponse<'_>, +) -> trc::Result> { + let nodes = get.server.registry().cluster_node_list().await?; + let mut ids = get + .ids + .take() + .unwrap_or_default() + .into_iter() + .map(|id| id.id()) + .collect::>(); + + for node in nodes { + if ids.is_empty() || ids.remove(&node.node_id) { + get.insert(node.node_id.into(), node.into_value()); + } + } + + for id in ids { + get.not_found(id.into()); + } + + Ok(get) +} + +pub(crate) async fn cluster_node_query( + req: RegistryQueryResponse<'_>, +) -> trc::Result { + if req + .request + .sort + .as_ref() + .and_then(|sort| sort.first()) + .is_some_and(|comp| { + !matches!( + comp.property, + RegistryComparator::Property(Property::NodeId) + ) + }) + { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details("Only sorting by 'nodeId' is supported for cluster nodes".to_string())); + } + + let nodes = req.server.registry().cluster_node_list().await?; + + // Build response + let mut response = QueryResponseBuilder::new( + nodes.len(), + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + + for node in nodes { + if !response.add_id(node.node_id.into()) { + break; + } + } + + Ok(response) +} diff --git a/crates/jmap/src/registry/mapping/mod.rs b/crates/jmap/src/registry/mapping/mod.rs index 25a81526..22bca569 100644 --- a/crates/jmap/src/registry/mapping/mod.rs +++ b/crates/jmap/src/registry/mapping/mod.rs @@ -23,6 +23,7 @@ use utils::map::vec_map::VecMap; pub mod account; pub mod action; +pub mod cluster; pub mod dkim; pub mod domain; pub mod log; diff --git a/crates/jmap/src/registry/mapping/spam_sample.rs b/crates/jmap/src/registry/mapping/spam_sample.rs index 70bb2c89..f78df665 100644 --- a/crates/jmap/src/registry/mapping/spam_sample.rs +++ b/crates/jmap/src/registry/mapping/spam_sample.rs @@ -17,13 +17,12 @@ use jmap_tools::JsonPointer; use mail_parser::{MessageParser, parsers::fields::thread::thread_name}; use registry::{ jmap::{IntoValue, JsonPointerPatch, RegistryJsonPatch}, - pickle::Pickle, schema::{ enums::Permission, prelude::{ObjectType, Property}, structs::SpamTrainingSample, }, - types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime, id::ObjectId}, }; use std::str::FromStr; use store::{ diff --git a/crates/jmap/src/registry/mapping/task.rs b/crates/jmap/src/registry/mapping/task.rs index 3db271d9..2e345734 100644 --- a/crates/jmap/src/registry/mapping/task.rs +++ b/crates/jmap/src/registry/mapping/task.rs @@ -20,7 +20,6 @@ use jmap_proto::{ use jmap_tools::{JsonPointer, JsonPointerItem, Key}; use registry::{ jmap::{IntoValue, JsonPointerPatch, RegistryJsonPatch}, - pickle::Pickle, schema::{ enums::{TaskStatusType, TaskType}, prelude::Property, diff --git a/crates/jmap/src/registry/query.rs b/crates/jmap/src/registry/query.rs index 9bec0af8..5949e1d8 100644 --- a/crates/jmap/src/registry/query.rs +++ b/crates/jmap/src/registry/query.rs @@ -9,8 +9,8 @@ use crate::{ registry::{ EnterpriseRegistry, mapping::{ - RegistryQueryResponse, account::credential_query, log::log_query, - queued_message::queued_message_query, report::report_query, + RegistryQueryResponse, account::credential_query, cluster::cluster_node_query, + log::log_query, queued_message::queued_message_query, report::report_query, spam_sample::spam_sample_query, task::task_query, }, }, @@ -121,6 +121,15 @@ impl RegistryQuery for Server { .await .and_then(|response| response.build()), + ObjectType::ClusterNode => cluster_node_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), + ObjectType::ApiKey | ObjectType::AppPassword => { credential_query(RegistryQueryResponse { server: self, diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index f24ff7e0..566d4250 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -323,7 +323,13 @@ impl RegistrySet for Server { let is_create = matches!(modification, Modification::Create { .. }); let mut unpatched_properties = VecMap::new(); - if is_create { + if is_create + || (is_singleton + && value + .as_object() + .unwrap() + .contains_key(&Key::Property(Property::Type))) + { // Patch object match new_object.patch( @@ -656,7 +662,7 @@ impl RegistrySet for Server { ObjectType::Action => action_set(set).await.map(|set| set.into_response()), - ObjectType::Log | ObjectType::Metric | ObjectType::Trace => { + ObjectType::Log | ObjectType::Metric | ObjectType::Trace | ObjectType::ClusterNode => { set.fail_all_create("Telemetry objects cannot be created"); set.fail_all_update("Telemetry objects cannot be modified"); set.fail_all_destroy("Telemetry objects cannot be deleted"); diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index c9cce9c1..7decd5ad 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -65,7 +65,7 @@ async fn main() -> std::io::Result<()> { #[cfg(feature = "dev_mode")] if std::env::var("INSERT_TEST_DATA").is_ok() { let server = init.inner.build_server(); - //test_data::insert_test_data(&server).await; + test_data::insert_test_data(&server).await; server.insert_test_metrics().await; } diff --git a/crates/main/src/test_data.rs b/crates/main/src/test_data.rs index 0f397599..9d2adb5f 100644 --- a/crates/main/src/test_data.rs +++ b/crates/main/src/test_data.rs @@ -9,7 +9,6 @@ use common::{ config::smtp::queue::{QueueExpiry, QueueName}, }; use registry::{ - pickle::Pickle, schema::{ enums::{ ArfAuthFailureType, ArfDeliveryResult, ArfFeedbackType, ArfIdentityAlignment, @@ -23,7 +22,10 @@ use registry::{ TlsFailureDetails, TlsInternalReport, TlsReport, TlsReportPolicy, }, }, - types::{EnumImpl, datetime::UTCDateTime, float::Float, ipaddr::IpAddr, list::List, map::Map}, + types::{ + EnumImpl, ObjectImpl, datetime::UTCDateTime, float::Float, ipaddr::IpAddr, list::List, + map::Map, + }, }; use smtp::{ queue::{ diff --git a/crates/registry/Cargo.toml b/crates/registry/Cargo.toml index 0de3a9f4..fab59947 100644 --- a/crates/registry/Cargo.toml +++ b/crates/registry/Cargo.toml @@ -15,6 +15,7 @@ jmap-tools = { version = "0.1" } xxhash-rust = { version = "0.8.5", features = ["xxh3"] } mail-auth = { version = "0.8" } tokio = { version = "1.47", features = ["fs"] } +lz4_flex = { version = "0.13", default-features = false } [features] test_mode = [] diff --git a/crates/registry/src/pickle.rs b/crates/registry/src/pickle.rs index e002fb30..4ac749c0 100644 --- a/crates/registry/src/pickle.rs +++ b/crates/registry/src/pickle.rs @@ -4,32 +4,70 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::types::EnumImpl; +use std::{borrow::Cow, collections::HashMap}; use utils::{ codec::leb128::{Leb128_, Leb128Reader, Leb128Writer}, map::vec_map::VecMap, }; -use crate::types::EnumImpl; -use std::collections::HashMap; +const COMPRESS_MARKER: u8 = 1 << 7; +const COMPRESS_WATERMARK: usize = 8192; pub trait Pickle: Sized { fn pickle(&self, out: &mut Vec); fn unpickle(stream: &mut PickledStream<'_>) -> Option; - fn to_pickled_vec(&self) -> Vec { - let mut out = Vec::with_capacity(256); - self.pickle(&mut out); - out - } } pub struct PickledStream<'x> { - data: &'x [u8], + data: Cow<'x, [u8]>, pos: usize, } +pub(crate) fn maybe_compress_pickle(input: Vec) -> Vec { + let input_len = input.len() - 1; // Exclude the version byte + if input_len > COMPRESS_WATERMARK { + let (version, input) = input.split_first().unwrap(); + let mut bytes: Vec = vec![ + version | COMPRESS_MARKER; + lz4_flex::block::get_maximum_output_size(input_len) + + 1 + + std::mem::size_of::() + ]; + + // Compress the data + let compressed_len = + lz4_flex::compress_into(input, &mut bytes[std::mem::size_of::() + 1..]).unwrap(); + if compressed_len < input_len { + // Prepend the length of the uncompressed data + bytes[1..(std::mem::size_of::() + 1)] + .copy_from_slice(&(input_len as u32).to_le_bytes()); + + // Truncate to the actual size + bytes.truncate(compressed_len + std::mem::size_of::() + 1); + return bytes; + } + } + input +} + impl<'x> PickledStream<'x> { - pub fn new(data: &'x [u8]) -> Self { - PickledStream { data, pos: 0 } + pub fn new(data: &'x [u8]) -> Option { + let (marker, data) = data.split_first()?; + if marker & COMPRESS_MARKER != 0 { + lz4_flex::block::decompress_size_prepended(data) + .ok() + .map(|data| PickledStream { + data: Cow::Owned(data), + pos: 0, + }) + } else { + PickledStream { + data: Cow::Borrowed(data), + pos: 0, + } + .into() + } } pub fn read(&mut self) -> Option { @@ -46,7 +84,7 @@ impl<'x> PickledStream<'x> { }) } - pub fn read_bytes(&mut self, len: usize) -> Option<&'x [u8]> { + pub fn read_bytes(&mut self, len: usize) -> Option<&'_ [u8]> { self.data.get(self.pos..self.pos + len).inspect(|_| { self.pos += len; }) @@ -56,8 +94,8 @@ impl<'x> PickledStream<'x> { self.pos >= self.data.len() } - pub fn bytes(&self) -> &'x [u8] { - self.data + pub fn bytes(&self) -> &'_ [u8] { + self.data.as_ref() } pub fn assert_version(&mut self, expected: u8) -> Option { diff --git a/crates/registry/src/schema/mod.rs b/crates/registry/src/schema/mod.rs index a902fd96..61e5c67a 100644 --- a/crates/registry/src/schema/mod.rs +++ b/crates/registry/src/schema/mod.rs @@ -26,6 +26,7 @@ pub mod structs; #[allow(clippy::len_zero)] #[allow(clippy::collapsible_if)] #[allow(clippy::derivable_impls)] +#[allow(clippy::field_reassign_with_default)] pub mod structs_impl; impl Display for Property { diff --git a/crates/registry/src/types/mod.rs b/crates/registry/src/types/mod.rs index c5f04710..ddd8528b 100644 --- a/crates/registry/src/types/mod.rs +++ b/crates/registry/src/types/mod.rs @@ -5,7 +5,7 @@ */ use crate::{ - pickle::Pickle, + pickle::{Pickle, maybe_compress_pickle}, schema::prelude::ObjectType, types::{error::ValidationError, index::IndexBuilder}, }; @@ -39,7 +39,14 @@ pub trait ObjectImpl: { const FLAGS: u64; const OBJECT: ObjectType; + const VERSION: u8; fn validate(&self, errors: &mut Vec) -> bool; fn index<'x>(&'x self, builder: &mut IndexBuilder<'x>); + fn to_pickled_vec(&self) -> Vec { + let mut out = Vec::with_capacity(256); + out.push(Self::VERSION); + self.pickle(&mut out); + maybe_compress_pickle(out) + } } diff --git a/crates/registry/src/utils/task.rs b/crates/registry/src/utils/task.rs index 6b785e2c..ca527aa5 100644 --- a/crates/registry/src/utils/task.rs +++ b/crates/registry/src/utils/task.rs @@ -106,6 +106,7 @@ impl Action { Action::InvalidateNegativeCaches => Permission::ActionInvalidateNegativeCaches, Action::PauseMtaQueue => Permission::ActionPauseMtaQueue, Action::ResumeMtaQueue => Permission::ActionResumeMtaQueue, + Action::UpdateApps => Permission::ActionUpdateApps, } } } diff --git a/crates/services/src/lib.rs b/crates/services/src/lib.rs index 683bc946..7d507662 100644 --- a/crates/services/src/lib.rs +++ b/crates/services/src/lib.rs @@ -6,7 +6,7 @@ use broadcast::publisher::spawn_broadcast_publisher; use common::{ - Inner, + BuildServer, Inner, manager::boot::{BootManager, IpcReceivers}, }; use state_manager::manager::spawn_push_router; @@ -29,19 +29,11 @@ pub trait SpawnServices { impl StartServices for BootManager { async fn start_services(&mut self) { // Unpack webadmin - if let Err(err) = self - .inner + self.inner .data .applications - .unpack(&self.inner.shared_core.load().storage.blob) - .await - { - trc::event!( - Resource(trc::ResourceEvent::Error), - Reason = err, - Details = "Failed to unpack application bundle" - ); - } + .unpack_all(&self.inner.build_server(), false) + .await; self.ipc_rxs.spawn_services(self.inner.clone()); } diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 16f7d27a..66c60cf0 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -578,9 +578,8 @@ async fn delete_email_metadata( { use email::message::metadata::MESSAGE_RECEIVED_MASK; use registry::{ - pickle::Pickle, schema::structs::{ArchivedEmail, ArchivedItem}, - types::{datetime::UTCDateTime, id::ObjectId}, + types::{ObjectImpl, datetime::UTCDateTime, id::ObjectId}, }; use store::{ SerializeInfallible, diff --git a/crates/services/src/task_manager/manager.rs b/crates/services/src/task_manager/manager.rs index d62d810a..4d8fb6ff 100644 --- a/crates/services/src/task_manager/manager.rs +++ b/crates/services/src/task_manager/manager.rs @@ -26,13 +26,12 @@ use common::config::server::ServerProtocol; use common::network::limiter::ConcurrencyLimiter; use common::network::{ServerInstance, TcpAcceptor}; use common::{Inner, Server}; -use registry::pickle::Pickle; use registry::schema::enums::TaskType; use registry::schema::structs::{ Task, TaskManager, TaskRetryStrategy, TaskStatus, TaskStatusFailed, TaskStatusRetry, }; -use registry::types::EnumImpl; use registry::types::datetime::UTCDateTime; +use registry::types::{EnumImpl, ObjectImpl}; use std::collections::hash_map::Entry; use std::future::Future; use std::time::Duration; diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index a4cadc33..6bee013a 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -17,12 +17,11 @@ use ahash::AHashSet; use common::config::smtp::queue::QueueName; use common::ipc::QueueEvent; use common::{KV_LOCK_QUEUE_MESSAGE, Server}; -use registry::pickle::Pickle; use registry::schema::prelude::{ObjectType, Property}; use registry::schema::structs::SpamTrainingSample; -use registry::types::EnumImpl; use registry::types::datetime::UTCDateTime; use registry::types::id::ObjectId; +use registry::types::{EnumImpl, ObjectImpl}; use std::borrow::Cow; use std::collections::hash_map::Entry; use std::future::Future; diff --git a/crates/smtp/src/reporting/dmarc.rs b/crates/smtp/src/reporting/dmarc.rs index 2f65a6f6..c86aa6cd 100644 --- a/crates/smtp/src/reporting/dmarc.rs +++ b/crates/smtp/src/reporting/dmarc.rs @@ -25,13 +25,12 @@ use mail_auth::{ report::{AuthFailureType, IdentityAlignment, PolicyPublished, Record, SPFDomainScope}, }; use registry::{ - pickle::Pickle, schema::{ enums::FailureReportingOption, prelude::{ObjectType, Property}, structs::{DmarcInternalReport, DmarcReport, DmarcReportRecord, Rate}, }, - types::{EnumImpl, datetime::UTCDateTime, map::Map}, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime, map::Map}, }; use std::future::Future; use store::{ diff --git a/crates/smtp/src/reporting/tls.rs b/crates/smtp/src/reporting/tls.rs index c31a0832..e283fe30 100644 --- a/crates/smtp/src/reporting/tls.rs +++ b/crates/smtp/src/reporting/tls.rs @@ -23,13 +23,12 @@ use mail_auth::{ report::tlsrpt::{FailureDetails, PolicyDetails}, }; use registry::{ - pickle::Pickle, schema::{ enums::TlsPolicyType, prelude::{ObjectType, Property}, structs::{TlsFailureDetails, TlsInternalReport, TlsReport, TlsReportPolicy}, }, - types::{EnumImpl, datetime::UTCDateTime}, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime}, }; use reqwest::header::CONTENT_TYPE; use std::fmt::Write; diff --git a/crates/store/src/build/registry.rs b/crates/store/src/build/registry.rs index 46455f1f..9581c062 100644 --- a/crates/store/src/build/registry.rs +++ b/crates/store/src/build/registry.rs @@ -13,6 +13,10 @@ use crate::{ now, }, }; +use registry::{ + schema::{enums::ClusterNodeStatus, structs::ClusterNode}, + types::datetime::UTCDateTime, +}; use std::{path::PathBuf, time::Duration}; use trc::AddContext; @@ -148,6 +152,49 @@ impl RegistryStore { Duration::from_secs(STALE_NODE_TIMEOUT / 2) } + pub async fn cluster_node_list(&self) -> trc::Result> { + let mut results = Vec::new(); + let now = now(); + + self.0 + .store + .iterate( + IterateParams::new( + ValueKey::from(ValueClass::NodeId(0)), + ValueKey::from(ValueClass::NodeId(u16::MAX)), + ) + .ascending(), + |key, value| { + if key.len() == U16_LEN * 3 { + let node_id = key.deserialize_be_u16(U32_LEN)?; + let last_renewal = value.deserialize_be_u64(0)?; + let last_renewal_since_now = now.saturating_sub(last_renewal); + let hostname = value + .get(U64_LEN..) + .and_then(|bytes| std::str::from_utf8(bytes).ok()) + .filter(|text| !text.is_empty()) + .ok_or_else(|| trc::StoreEvent::DataCorruption.into_err())?; + + results.push(ClusterNode { + hostname: hostname.to_string(), + last_renewal: UTCDateTime::from_timestamp(last_renewal.cast_signed()), + node_id: node_id as u64, + status: if last_renewal_since_now > DEAD_NODE_TIMEOUT { + ClusterNodeStatus::Inactive + } else if last_renewal_since_now > STALE_NODE_TIMEOUT { + ClusterNodeStatus::Stale + } else { + ClusterNodeStatus::Active + }, + }); + } + Ok(true) + }, + ) + .await + .map(|_| results) + } + pub async fn refresh_node_id_lease(&self) -> trc::Result<()> { let mut batch = BatchBuilder::new(); batch diff --git a/crates/store/src/registry/get.rs b/crates/store/src/registry/get.rs index fe3807e5..a5d6ed10 100644 --- a/crates/store/src/registry/get.rs +++ b/crates/store/src/registry/get.rs @@ -79,15 +79,16 @@ impl RegistryStore { ), |key, value| { let id = key.deserialize_be_u64(U16_LEN)?; - let mut stream = PickledStream::new(value); - let object = T::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .id(id) - .details(object_type.as_str()) - .ctx(trc::Key::Value, value) - })?; + let object = PickledStream::new(value) + .and_then(|mut stream| T::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .id(id) + .details(object_type.as_str()) + .ctx(trc::Key::Value, value) + })?; results.push(RegistryObject { id: ObjectId::new(object_type, Id::new(id)), diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index e17126c6..3550f6f5 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -88,7 +88,7 @@ impl Deserialize for Object { fn deserialize_with_key(key: &[u8], bytes: &[u8]) -> trc::Result { let revision = xxhash_rust::xxh3::xxh3_64(bytes); ObjectType::from_id(key.deserialize_be_u16(0)?) - .and_then(|object_id| ObjectInner::unpickle(object_id, &mut PickledStream::new(bytes))) + .and_then(|object_id| ObjectInner::unpickle(object_id, &mut PickledStream::new(bytes)?)) .map(|inner| Object { revision, inner }) .ok_or_else(|| { trc::EventType::Registry(trc::RegistryEvent::DeserializationError) @@ -105,61 +105,66 @@ impl Deserialize for Object { impl Deserialize for Task { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - Task::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } impl Deserialize for SpamTrainingSample { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - SpamTrainingSample::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } impl Deserialize for ArchivedItem { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - ArchivedItem::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } impl Deserialize for TlsInternalReport { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - TlsInternalReport::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } impl Deserialize for DmarcInternalReport { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - DmarcInternalReport::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } @@ -208,24 +213,26 @@ impl Deserialize for ObjectIdVersioned { impl Deserialize for Trace { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - Trace::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } impl Deserialize for Metric { fn deserialize(bytes: &[u8]) -> trc::Result { - let mut stream = PickledStream::new(bytes); - Metric::unpickle(&mut stream).ok_or_else(|| { - trc::EventType::Registry(trc::RegistryEvent::DeserializationError) - .into_err() - .caused_by(trc::location!()) - .ctx(trc::Key::Value, bytes) - }) + PickledStream::new(bytes) + .and_then(|mut stream| Self::unpickle(&mut stream)) + .ok_or_else(|| { + trc::EventType::Registry(trc::RegistryEvent::DeserializationError) + .into_err() + .caused_by(trc::location!()) + .ctx(trc::Key::Value, bytes) + }) } } diff --git a/crates/store/src/registry/write.rs b/crates/store/src/registry/write.rs index fc23147e..90a530a0 100644 --- a/crates/store/src/registry/write.rs +++ b/crates/store/src/registry/write.rs @@ -306,8 +306,7 @@ impl RegistryStore { } // It's pickle time! - let mut out = Vec::with_capacity(256); - object.inner.pickle(&mut out); + let out = object.inner.to_pickled_vec(); // Build batch if write_id { diff --git a/crates/store/src/write/batch.rs b/crates/store/src/write/batch.rs index eaadf12a..74fa986f 100644 --- a/crates/store/src/write/batch.rs +++ b/crates/store/src/write/batch.rs @@ -14,7 +14,10 @@ use crate::{ LogCollection, MergeFnc, MergeOperation, Params, SetFnc, SetOperation, TaskQueueClass, }, }; -use registry::{pickle::Pickle, schema::structs::Task, types::EnumImpl}; +use registry::{ + schema::structs::Task, + types::{EnumImpl, ObjectImpl}, +}; use types::{ collection::{Collection, SyncCollection, VanishedCollection}, field::FieldType, diff --git a/crates/trc/src/event/enums.rs b/crates/trc/src/event/enums.rs index d09401e2..0a3ea678 100644 --- a/crates/trc/src/event/enums.rs +++ b/crates/trc/src/event/enums.rs @@ -6,7 +6,7 @@ // This file is auto-generated. Do not edit directly. -pub const TOTAL_EVENT_COUNT: usize = 601; +pub const TOTAL_EVENT_COUNT: usize = 603; pub const TOTAL_METRIC_COUNT: usize = 339; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -592,6 +592,8 @@ pub enum ResourceEvent { BadParameters = 386, Error = 388, DownloadExternal = 387, + ApplicationUpdated = 601, + ApplicationUnpacked = 602, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] diff --git a/crates/trc/src/event/enums_impl.rs b/crates/trc/src/event/enums_impl.rs index e090f9bf..351cb78f 100644 --- a/crates/trc/src/event/enums_impl.rs +++ b/crates/trc/src/event/enums_impl.rs @@ -397,6 +397,8 @@ impl EventType { b"resource.bad-parameters" => EventType::Resource(ResourceEvent::BadParameters), b"resource.error" => EventType::Resource(ResourceEvent::Error), b"resource.download-external" => EventType::Resource(ResourceEvent::DownloadExternal), + b"resource.application-updated" => EventType::Resource(ResourceEvent::ApplicationUpdated), + b"resource.application-unpacked" => EventType::Resource(ResourceEvent::ApplicationUnpacked), b"security.authentication-ban" => EventType::Security(SecurityEvent::AuthenticationBan), b"security.abuse-ban" => EventType::Security(SecurityEvent::AbuseBan), b"security.scan-ban" => EventType::Security(SecurityEvent::ScanBan), @@ -1112,6 +1114,12 @@ impl EventType { EventType::Resource(ResourceEvent::BadParameters) => "resource.bad-parameters", EventType::Resource(ResourceEvent::Error) => "resource.error", EventType::Resource(ResourceEvent::DownloadExternal) => "resource.download-external", + EventType::Resource(ResourceEvent::ApplicationUpdated) => { + "resource.application-updated" + } + EventType::Resource(ResourceEvent::ApplicationUnpacked) => { + "resource.application-unpacked" + } EventType::Security(SecurityEvent::AuthenticationBan) => "security.authentication-ban", EventType::Security(SecurityEvent::AbuseBan) => "security.abuse-ban", EventType::Security(SecurityEvent::ScanBan) => "security.scan-ban", @@ -1740,6 +1748,8 @@ impl EventType { EventType::Resource(ResourceEvent::BadParameters) => 386, EventType::Resource(ResourceEvent::Error) => 388, EventType::Resource(ResourceEvent::DownloadExternal) => 387, + EventType::Resource(ResourceEvent::ApplicationUpdated) => 601, + EventType::Resource(ResourceEvent::ApplicationUnpacked) => 602, EventType::Security(SecurityEvent::AuthenticationBan) => 33, EventType::Security(SecurityEvent::AbuseBan) => 549, EventType::Security(SecurityEvent::ScanBan) => 558, @@ -2384,6 +2394,8 @@ impl EventType { 386 => Some(EventType::Resource(ResourceEvent::BadParameters)), 388 => Some(EventType::Resource(ResourceEvent::Error)), 387 => Some(EventType::Resource(ResourceEvent::DownloadExternal)), + 601 => Some(EventType::Resource(ResourceEvent::ApplicationUpdated)), + 602 => Some(EventType::Resource(ResourceEvent::ApplicationUnpacked)), 33 => Some(EventType::Security(SecurityEvent::AuthenticationBan)), 549 => Some(EventType::Security(SecurityEvent::AbuseBan)), 558 => Some(EventType::Security(SecurityEvent::ScanBan)), @@ -2793,6 +2805,7 @@ impl EventType { EventType::Queue(QueueEvent::ConcurrencyLimitExceeded) => Level::Info, EventType::Queue(QueueEvent::QuotaExceeded) => Level::Info, EventType::Resource(ResourceEvent::DownloadExternal) => Level::Info, + EventType::Resource(ResourceEvent::ApplicationUpdated) => Level::Info, EventType::Security(SecurityEvent::AuthenticationBan) => Level::Info, EventType::Security(SecurityEvent::AbuseBan) => Level::Info, EventType::Security(SecurityEvent::ScanBan) => Level::Info, @@ -3454,6 +3467,12 @@ impl EventType { EventType::Resource(ResourceEvent::BadParameters) => "Bad resource parameters", EventType::Resource(ResourceEvent::Error) => "Resource error", EventType::Resource(ResourceEvent::DownloadExternal) => "Downloading external resource", + EventType::Resource(ResourceEvent::ApplicationUpdated) => { + "Application resource updated" + } + EventType::Resource(ResourceEvent::ApplicationUnpacked) => { + "Application resource unpacked" + } EventType::Security(SecurityEvent::AuthenticationBan) => { "Banned due to authentication errors" } @@ -4351,6 +4370,8 @@ impl EventType { EventType::Resource(ResourceEvent::BadParameters), EventType::Resource(ResourceEvent::Error), EventType::Resource(ResourceEvent::DownloadExternal), + EventType::Resource(ResourceEvent::ApplicationUpdated), + EventType::Resource(ResourceEvent::ApplicationUnpacked), EventType::Security(SecurityEvent::AuthenticationBan), EventType::Security(SecurityEvent::AbuseBan), EventType::Security(SecurityEvent::ScanBan), diff --git a/tests/src/store/registry.rs b/tests/src/store/registry.rs index 983e77f8..02bdabc8 100644 --- a/tests/src/store/registry.rs +++ b/tests/src/store/registry.rs @@ -4,6 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use crate::utils::{registry::UnwrapRegistryId, server::TestServer}; use jmap_tools::JsonPointer; use registry::{ jmap::{IntoValue, JmapValue, JsonPointerPatch, MaybeUnpatched, RegistryJsonPatch}, @@ -16,11 +17,12 @@ use registry::{ CredentialPermissionsList, CustomRoles, DkimManagement, DnsManagement, Domain, EmailAlias, EncryptionAtRest, EncryptionSettings, GroupAccount, MailingList, PasswordCredential, Permissions, PermissionsList, PublicKey, SecondaryCredential, - UserAccount, UserRoles, + SieveUserScript, UserAccount, UserRoles, }, }, types::{ - EnumImpl, datetime::UTCDateTime, id::ObjectId, ipmask::IpAddrOrMask, list::List, map::Map, + EnumImpl, ObjectImpl, datetime::UTCDateTime, id::ObjectId, ipmask::IpAddrOrMask, + list::List, map::Map, }, }; use std::str::FromStr; @@ -34,8 +36,6 @@ use store::{ use types::id::Id; use utils::map::vec_map::VecMap; -use crate::utils::{registry::UnwrapRegistryId, server::TestServer}; - pub async fn test(test: &TestServer) { let r = test.server.registry(); @@ -105,10 +105,28 @@ pub async fn test(test: &TestServer) { }), time_zone: None, }); - let account_picke = account.to_pickled_vec(); + let account_pickle = account.to_pickled_vec(); assert_eq!( account, - Account::unpickle(&mut PickledStream::new(&account_picke)).unwrap() + Account::unpickle(&mut PickledStream::new(&account_pickle).unwrap()).unwrap() + ); + + // Pickle compression test + let script = SieveUserScript { + contents: "A".repeat(100_000), + description: "B".repeat(100_000).into(), + is_active: true, + name: "C".repeat(100_000), + }; + let script_pickle = script.to_pickled_vec(); + assert!( + script_pickle.len() < 8_192, + "Pickle was not compressed: {} bytes", + script_pickle.len() + ); + assert_eq!( + script, + SieveUserScript::unpickle(&mut PickledStream::new(&script_pickle).unwrap()).unwrap() ); // Create a domain and a group