diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index 1431d21d..d200b54a 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -72,6 +72,7 @@ impl Data { jmap_id_gen: id_generator.clone(), queue_id_gen: id_generator.clone(), span_id_gen: id_generator, + queue_status: true.into(), webadmin: config .value("webadmin.path") .map(|path| WebAdminManager::new(path.into())) @@ -245,6 +246,7 @@ impl Default for Data { jmap_id_gen: Default::default(), queue_id_gen: Default::default(), span_id_gen: Default::default(), + queue_status: true.into(), webadmin: Default::default(), config_version: Default::default(), jmap_limiter: Default::default(), diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 9411e696..7b10353c 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -8,7 +8,10 @@ use std::{ collections::BTreeMap, hash::{BuildHasher, Hasher}, net::{IpAddr, Ipv4Addr, Ipv6Addr}, - sync::{atomic::AtomicU8, Arc}, + sync::{ + atomic::{AtomicBool, AtomicU8}, + Arc, + }, }; use ahash::{AHashMap, AHashSet, RandomState}; @@ -118,6 +121,7 @@ pub struct Data { pub jmap_id_gen: SnowflakeIdGenerator, pub queue_id_gen: SnowflakeIdGenerator, pub span_id_gen: SnowflakeIdGenerator, + pub queue_status: AtomicBool, pub webadmin: WebAdminManager, pub config_version: AtomicU8, diff --git a/crates/jmap/src/api/management/enterprise/undelete.rs b/crates/jmap/src/api/management/enterprise/undelete.rs index c873d541..c5cef6c6 100644 --- a/crates/jmap/src/api/management/enterprise/undelete.rs +++ b/crates/jmap/src/api/management/enterprise/undelete.rs @@ -134,13 +134,12 @@ impl UndeleteApi for Server { .await? .ok_or_else(|| trc::ResourceEvent::NotFound.into_err())?; - let requests = - serde_json::from_slice::>>( - body.as_deref().unwrap_or_default(), - ) - .ok() - .and_then(|request| { - request + let requests: Vec> = + match serde_json::from_slice::< + Option>>, + >(body.as_deref().unwrap_or_default()) + { + Ok(Some(requests)) => requests .into_iter() .map(|request| { UndeleteRequest { @@ -154,12 +153,14 @@ impl UndeleteApi for Server { collection: Collection::from_str(request.collection.as_str()) .ok()?, time: DateTime::parse_rfc3339(request.time.as_str())? - .to_timestamp(), + .to_timestamp() + as u64, cancel_deletion: if let Some(cancel_deletion) = request.cancel_deletion { - DateTime::parse_rfc3339(cancel_deletion.as_str())? + (DateTime::parse_rfc3339(cancel_deletion.as_str())? .to_timestamp() + as u64) .into() } else { None @@ -168,8 +169,24 @@ impl UndeleteApi for Server { .into() }) .collect::>>() - }) - .ok_or_else(|| trc::ResourceEvent::BadParameters.into_err())?; + .ok_or_else(|| trc::ResourceEvent::BadParameters.into_err())?, + Ok(None) => { + let deleted = self.core.list_deleted(account_id).await?; + let mut results = Vec::with_capacity(deleted.len()); + for blob in deleted { + results.push(UndeleteRequest { + hash: blob.hash, + collection: Collection::from(blob.collection), + time: blob.deleted_at, + cancel_deletion: blob.expires_at.into(), + }); + } + results + } + Err(_) => { + return Err(trc::ResourceEvent::BadParameters.into_err()); + } + }; let mut results = Vec::with_capacity(requests.len()); let mut batch = BatchBuilder::new(); @@ -192,7 +209,7 @@ impl UndeleteApi for Server { .caused_by(trc::location!())?, mailbox_ids: vec![INBOX_ID], keywords: vec![], - received_at: (request.time as u64).into(), + received_at: request.time.into(), source: IngestSource::Restore, spam_classify: false, spam_train: false, @@ -205,7 +222,7 @@ impl UndeleteApi for Server { if let Some(cancel_deletion) = request.cancel_deletion { batch.clear(ValueClass::Blob(BlobOp::Reserve { hash: request.hash, - until: cancel_deletion as u64, + until: cancel_deletion, })); } } diff --git a/crates/jmap/src/api/management/principal.rs b/crates/jmap/src/api/management/principal.rs index 9897d93b..3fb9f649 100644 --- a/crates/jmap/src/api/management/principal.rs +++ b/crates/jmap/src/api/management/principal.rs @@ -34,7 +34,7 @@ pub enum AccountAuthRequest { EnableOtpAuth { url: String }, DisableOtpAuth { url: Option }, AddAppPassword { name: String, password: String }, - RemoveAppPassword { name: String }, + RemoveAppPassword { name: Option }, } #[derive(Debug, serde::Serialize, serde::Deserialize)] @@ -265,6 +265,126 @@ impl PrincipalManager for Server { })) .into_http_response()) } + (None, &Method::DELETE) => { + // List principal ids + let params = UrlParams::new(req.uri().query()); + let filter = params.get("filter"); + let typ = params.parse::("type").ok_or_else(|| { + trc::EventType::Resource(trc::ResourceEvent::BadParameters) + .into_err() + .details("Invalid type") + })?; + if params.get("confirm").map_or(true, |c| c != "true") { + return Err(trc::EventType::Resource(trc::ResourceEvent::BadParameters) + .into_err() + .details("Missing confirmation parameter")); + } + + // Validate the access token + access_token.assert_has_permission(match typ { + Type::Individual => Permission::IndividualDelete, + Type::Group => Permission::GroupDelete, + Type::List => Permission::MailingListDelete, + Type::Domain => Permission::DomainDelete, + Type::Tenant => Permission::TenantDelete, + Type::Role => Permission::RoleDelete, + Type::ApiKey => Permission::ApiKeyDelete, + Type::OauthClient => Permission::OauthClientDelete, + Type::Resource | Type::Location | Type::Other => Permission::PrincipalDelete, + })?; + + let mut tenant = access_token.tenant.map(|t| t.id); + + #[cfg(feature = "enterprise")] + if self.core.is_enterprise_edition() { + if tenant.is_none() { + // Limit search to current tenant + if let Some(tenant_name) = params.get("tenant") { + tenant = self + .core + .storage + .data + .get_principal_info(tenant_name) + .await? + .filter(|p| p.typ == Type::Tenant) + .map(|p| p.id); + } + } + } else if typ == Type::Tenant { + return Err(manage::enterprise()); + } + + let principals = self + .core + .storage + .data + .list_principals(filter, tenant, &[typ], &[PrincipalField::Name], 0, 0) + .await?; + + let found = !principals.items.is_empty(); + if found { + let server = self.clone(); + tokio::spawn(async move { + let has_bayes = server + .core + .spam + .bayes + .as_ref() + .map_or(false, |c| c.account_classify); + for principal in principals.items { + // Delete account + if let Err(err) = server + .store() + .delete_principal(QueryBy::Id(principal.id())) + .await + { + trc::error!(err.details("Failed to delete principal")); + continue; + } + + if matches!(typ, Type::Individual | Type::Group) { + // Remove FTS index + if let Err(err) = + server.core.storage.fts.remove_all(principal.id()).await + { + trc::error!(err.details("Failed to delete FTS index")); + } + + // Delete bayes model + if has_bayes { + let mut key = + Vec::with_capacity(std::mem::size_of::() + 1); + key.push(KV_BAYES_MODEL_USER); + key.extend_from_slice(&principal.id().to_be_bytes()); + + if let Err(err) = + server.in_memory_store().key_delete_prefix(&key).await + { + trc::error!( + err.details("Failed to delete user bayes model") + ); + } + } + } + } + + if matches!(typ, Type::Role | Type::Tenant) { + // Update permissions cache + server.inner.cache.permissions.clear(); + server + .inner + .cache + .permissions_version + .fetch_add(1, Ordering::Relaxed); + } + }); + } + + Ok(JsonResponse::new(json!({ + "data": found, + })) + .into_http_response()) + } (Some(name), method) => { // Fetch, update or delete principal let name = decode_path_element(name); @@ -344,31 +464,31 @@ impl PrincipalManager for Server { })?; // Delete account - self.core - .storage - .data + self.store() .delete_principal(QueryBy::Id(account_id)) .await?; - // Remove FTS index if matches!(typ, Type::Individual | Type::Group) { + // Remove FTS index self.core.storage.fts.remove_all(account_id).await?; - } - // Delete bayes model - if self - .core - .spam - .bayes - .as_ref() - .map_or(false, |c| c.account_classify) - { - let mut key = Vec::with_capacity(std::mem::size_of::() + 1); - key.push(KV_BAYES_MODEL_USER); - key.extend_from_slice(&account_id.to_be_bytes()); + // Delete bayes model + if self + .core + .spam + .bayes + .as_ref() + .map_or(false, |c| c.account_classify) + { + let mut key = Vec::with_capacity(std::mem::size_of::() + 1); + key.push(KV_BAYES_MODEL_USER); + key.extend_from_slice(&account_id.to_be_bytes()); - if let Err(err) = self.in_memory_store().key_delete_prefix(&key).await { - trc::error!(err.details("Failed to delete user bayes model")); + if let Err(err) = + self.in_memory_store().key_delete_prefix(&key).await + { + trc::error!(err.details("Failed to delete user bayes model")); + } } } @@ -668,9 +788,10 @@ impl PrincipalManager for Server { AccountAuthRequest::AddAppPassword { name, password } => { (PrincipalAction::AddItem, format!("$app${name}${password}")) } - AccountAuthRequest::RemoveAppPassword { name } => { - (PrincipalAction::RemoveItem, format!("$app${name}")) - } + AccountAuthRequest::RemoveAppPassword { name } => ( + PrincipalAction::RemoveItem, + format!("$app${}", name.unwrap_or_default()), + ), }; actions.push(PrincipalUpdate { diff --git a/crates/jmap/src/api/management/queue.rs b/crates/jmap/src/api/management/queue.rs index 56ba0c58..3ddf4491 100644 --- a/crates/jmap/src/api/management/queue.rs +++ b/crates/jmap/src/api/management/queue.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use std::future::Future; +use std::{future::Future, sync::atomic::Ordering}; use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; use common::{auth::AccessToken, ipc::QueueEvent, Server}; @@ -81,6 +81,7 @@ pub struct Recipient { #[derive(Debug, serde::Serialize, serde::Deserialize)] #[serde(tag = "type")] +#[serde(rename_all = "camelCase")] pub enum Report { Tls { id: String, @@ -171,109 +172,24 @@ impl QueueManagement for Server { // Validate the access token access_token.assert_has_permission(Permission::MessageQueueList)?; - let text = params.get("text"); - let from = params.get("from"); - let to = params.get("to"); - let before = params - .parse::("before") - .map(|t| t.into_inner()); - let after = params - .parse::("after") - .map(|t| t.into_inner()); - let page = params.parse::("page").unwrap_or_default(); - let limit = params.parse::("limit").unwrap_or_default(); - let values = params.has_key("values"); + let result = fetch_queued_messages(self, ¶ms, &tenant_domains).await?; - let range_start = params.parse::("range-start").unwrap_or_default(); - let range_end = params.parse::("range-end").unwrap_or(u64::MAX); - let max_total = params.parse::("max-total").unwrap_or_default(); + let queue_status = self.inner.data.queue_status.load(Ordering::Relaxed); - let mut result_ids = Vec::new(); - let mut result_values = Vec::new(); - let from_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(range_start))); - let to_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(range_end))); - let has_filters = text.is_some() - || from.is_some() - || to.is_some() - || before.is_some() - || after.is_some(); - let mut offset = page.saturating_sub(1) * limit; - let mut total = 0; - let mut total_returned = 0; - self.core - .storage - .data - .iterate( - IterateParams::new(from_key, to_key).ascending(), - |key, value| { - let message = Bincode::::deserialize(value) - .add_context(|ctx| ctx.ctx(trc::Key::Key, key))? - .inner; - let matches = tenant_domains - .as_ref() - .map_or(true, |domains| message.has_domain(domains)) - && (!has_filters - || (text - .as_ref() - .map(|text| { - message.return_path.contains(text) - || message - .recipients - .iter() - .any(|r| r.address_lcase.contains(text)) - }) - .unwrap_or_else(|| { - from.as_ref().map_or(true, |from| { - message.return_path.contains(from) - }) && to.as_ref().map_or(true, |to| { - message - .recipients - .iter() - .any(|r| r.address_lcase.contains(to)) - }) - }) - && before.as_ref().map_or(true, |before| { - message.next_delivery_event() < *before - }) - && after.as_ref().map_or(true, |after| { - message.next_delivery_event() > *after - }))); - - if matches { - if offset == 0 { - if limit == 0 || total_returned < limit { - if values { - result_values.push(Message::from(&message)); - } else { - result_ids.push(key.deserialize_be_u64(0)?); - } - total_returned += 1; - } - } else { - offset -= 1; - } - - total += 1; - } - - Ok(max_total == 0 || total < max_total) - }, - ) - .await - .caused_by(trc::location!())?; - - Ok(if values { + Ok(if !result.values.is_empty() { JsonResponse::new(json!({ "data":{ - "items": result_values, - "total": total, + "items": result.values, + "total": result.total, + "status": queue_status, }, })) } else { JsonResponse::new(json!({ "data": { - "items": result_ids, - "total": total, + "items": result.ids, + "total": result.total, + "status": queue_status, }, })) } @@ -300,6 +216,61 @@ impl QueueManagement for Server { Err(trc::ResourceEvent::NotFound.into_err()) } } + ("messages", None, &Method::PATCH) => { + // Validate the access token + access_token.assert_has_permission(Permission::MessageQueueUpdate)?; + + let time = params + .parse::("at") + .map(|t| t.into_inner()) + .unwrap_or_else(now); + let result = fetch_queued_messages(self, ¶ms, &tenant_domains).await?; + + let found = !result.ids.is_empty(); + if found { + let server = self.clone(); + tokio::spawn(async move { + for id in result.ids { + if let Some(mut message) = server.read_message(id).await { + let prev_event = message.next_event().unwrap_or_default(); + let mut has_changes = false; + + for domain in &mut message.domains { + if matches!( + domain.status, + Status::Scheduled | Status::TemporaryFailure(_) + ) { + domain.retry.due = time; + if domain.expires > time { + domain.expires = time + 10; + } + has_changes = true; + } + } + + if has_changes { + let next_event = message.next_event().unwrap_or_default(); + message + .save_changes(&server, prev_event.into(), next_event.into()) + .await; + } + } + } + + let _ = server + .inner + .ipc + .queue_tx + .send(QueueEvent::Refresh(None)) + .await; + }); + } + + Ok(JsonResponse::new(json!({ + "data": found, + })) + .into_http_response()) + } ("messages", Some(queue_id), &Method::PATCH) => { // Validate the access token access_token.assert_has_permission(Permission::MessageQueueUpdate)?; @@ -359,6 +330,50 @@ impl QueueManagement for Server { Err(trc::ResourceEvent::NotFound.into_err()) } } + ("messages", None, &Method::DELETE) => { + // Validate the access token + access_token.assert_has_permission(Permission::MessageQueueDelete)?; + + let result = fetch_queued_messages(self, ¶ms, &tenant_domains).await?; + + let found = !result.ids.is_empty(); + if found { + let server = self.clone(); + tokio::spawn(async move { + let is_active = server.inner.data.queue_status.load(Ordering::Relaxed); + + if is_active { + let _ = server + .inner + .ipc + .queue_tx + .send(QueueEvent::Paused(true)) + .await; + } + + for id in result.ids { + if let Some(message) = server.read_message(id).await { + let prev_event = message.next_event().unwrap_or_default(); + message.remove(&server, prev_event).await; + } + } + + if is_active { + let _ = server + .inner + .ipc + .queue_tx + .send(QueueEvent::Paused(false)) + .await; + } + }); + } + + Ok(JsonResponse::new(json!({ + "data": found, + })) + .into_http_response()) + } ("messages", Some(queue_id), &Method::DELETE) => { // Validate the access token access_token.assert_has_permission(Permission::MessageQueueDelete)?; @@ -450,83 +465,12 @@ impl QueueManagement for Server { // Validate the access token access_token.assert_has_permission(Permission::OutgoingReportList)?; - let domain = params.get("domain").map(|d| d.to_lowercase()); - let type_ = params.get("type").and_then(|t| match t { - "dmarc" => 0u8.into(), - "tls" => 1u8.into(), - _ => None, - }); - let page: usize = params.parse("page").unwrap_or_default(); - let limit: usize = params.parse("limit").unwrap_or_default(); - - let range_start = params.parse::("range-start").unwrap_or_default(); - let range_end = params.parse::("range-end").unwrap_or(u64::MAX); - let max_total = params.parse::("max-total").unwrap_or_default(); - - let mut result = Vec::new(); - let from_key = ValueKey::from(ValueClass::Queue(QueueClass::DmarcReportHeader( - ReportEvent { - due: range_start, - policy_hash: 0, - seq_id: 0, - domain: String::new(), - }, - ))); - let to_key = ValueKey::from(ValueClass::Queue(QueueClass::TlsReportHeader( - ReportEvent { - due: range_end, - policy_hash: 0, - seq_id: 0, - domain: String::new(), - }, - ))); - let mut offset = page.saturating_sub(1) * limit; - let mut total = 0; - let mut total_returned = 0; - self.core - .storage - .data - .iterate( - IterateParams::new(from_key, to_key).ascending().no_values(), - |key, _| { - if type_.map_or(true, |t| t == *key.last().unwrap()) { - let event = ReportEvent::deserialize(key)?; - if tenant_domains - .as_ref() - .map_or(true, |domains| domains.contains(&event.domain)) - && event.seq_id != 0 - && domain.as_ref().map_or(true, |d| event.domain.contains(d)) - { - if offset == 0 { - if limit == 0 || total_returned < limit { - result.push( - if *key.last().unwrap() == 0 { - QueueClass::DmarcReportHeader(event) - } else { - QueueClass::TlsReportHeader(event) - } - .queue_id(), - ); - total_returned += 1; - } - } else { - offset -= 1; - } - - total += 1; - } - } - - Ok(max_total == 0 || total < max_total) - }, - ) - .await - .caused_by(trc::location!())?; + let result = fetch_queued_reports(self, ¶ms, &tenant_domains).await?; Ok(JsonResponse::new(json!({ "data": { - "items": result, - "total": total, + "items": result.ids.into_iter().map(|id| id.queue_id()).collect::>(), + "total": result.total, }, })) .into_http_response()) @@ -577,6 +521,34 @@ impl QueueManagement for Server { Err(trc::ResourceEvent::NotFound.into_err()) } } + ("reports", None, &Method::DELETE) => { + // Validate the access token + access_token.assert_has_permission(Permission::OutgoingReportDelete)?; + + let result = fetch_queued_reports(self, ¶ms, &tenant_domains).await?; + let found = !result.ids.is_empty(); + if found { + let server = self.clone(); + tokio::spawn(async move { + for id in result.ids { + match id { + QueueClass::DmarcReportHeader(event) => { + server.delete_dmarc_report(event).await; + } + QueueClass::TlsReportHeader(event) => { + server.delete_tls_report(vec![event]).await; + } + _ => (), + } + } + }); + } + + Ok(JsonResponse::new(json!({ + "data": found, + })) + .into_http_response()) + } ("reports", Some(report_id), &Method::DELETE) => { // Validate the access token access_token.assert_has_permission(Permission::OutgoingReportDelete)?; @@ -610,6 +582,33 @@ impl QueueManagement for Server { Err(trc::ResourceEvent::NotFound.into_err()) } } + ("status", None, &Method::GET) => { + // Validate the access token + access_token.assert_has_permission(Permission::MessageQueueGet)?; + + Ok(JsonResponse::new(json!({ + "data": self.inner.data.queue_status.load(Ordering::Relaxed), + })) + .into_http_response()) + } + ("status", Some(action), &Method::PATCH) => { + // Validate the access token + access_token.assert_has_permission(Permission::MessageQueueUpdate)?; + + let prev_status = self.inner.data.queue_status.load(Ordering::Relaxed); + + let _ = self + .inner + .ipc + .queue_tx + .send(QueueEvent::Paused(action == "stop")) + .await; + + Ok(JsonResponse::new(json!({ + "data": prev_status, + })) + .into_http_response()) + } _ => Err(trc::ResourceEvent::NotFound.into_err()), } } @@ -678,6 +677,197 @@ impl From<&queue::Message> for Message { } } +struct QueuedMessages { + ids: Vec, + values: Vec, + total: usize, +} + +async fn fetch_queued_messages( + server: &Server, + params: &UrlParams<'_>, + tenant_domains: &Option>, +) -> trc::Result { + let text = params.get("text"); + let from = params.get("from"); + let to = params.get("to"); + let before = params + .parse::("before") + .map(|t| t.into_inner()); + let after = params + .parse::("after") + .map(|t| t.into_inner()); + let page = params.parse::("page").unwrap_or_default(); + let limit = params.parse::("limit").unwrap_or_default(); + let values = params.has_key("values"); + + let range_start = params.parse::("range-start").unwrap_or_default(); + let range_end = params.parse::("range-end").unwrap_or(u64::MAX); + let max_total = params.parse::("max-total").unwrap_or_default(); + + let mut result = QueuedMessages { + ids: Vec::new(), + values: Vec::new(), + total: 0, + }; + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(range_start))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::Message(range_end))); + let has_filters = + text.is_some() || from.is_some() || to.is_some() || before.is_some() || after.is_some(); + let mut offset = page.saturating_sub(1) * limit; + let mut total_returned = 0; + + server + .core + .storage + .data + .iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + let message = Bincode::::deserialize(value) + .add_context(|ctx| ctx.ctx(trc::Key::Key, key))? + .inner; + let matches = tenant_domains + .as_ref() + .map_or(true, |domains| message.has_domain(domains)) + && (!has_filters + || (text + .as_ref() + .map(|text| { + message.return_path.contains(text) + || message + .recipients + .iter() + .any(|r| r.address_lcase.contains(text)) + }) + .unwrap_or_else(|| { + from.as_ref() + .map_or(true, |from| message.return_path.contains(from)) + && to.as_ref().map_or(true, |to| { + message + .recipients + .iter() + .any(|r| r.address_lcase.contains(to)) + }) + }) + && before + .as_ref() + .map_or(true, |before| message.next_delivery_event() < *before) + && after + .as_ref() + .map_or(true, |after| message.next_delivery_event() > *after))); + + if matches { + if offset == 0 { + if limit == 0 || total_returned < limit { + if values { + result.values.push(Message::from(&message)); + } else { + result.ids.push(key.deserialize_be_u64(0)?); + } + total_returned += 1; + } + } else { + offset -= 1; + } + + result.total += 1; + } + + Ok(max_total == 0 || result.total < max_total) + }, + ) + .await + .caused_by(trc::location!()) + .map(|_| result) +} + +struct QueuedReports { + ids: Vec, + total: usize, +} + +async fn fetch_queued_reports( + server: &Server, + params: &UrlParams<'_>, + tenant_domains: &Option>, +) -> trc::Result { + let domain = params.get("domain").map(|d| d.to_lowercase()); + let type_ = params.get("type").and_then(|t| match t { + "dmarc" => 0u8.into(), + "tls" => 1u8.into(), + _ => None, + }); + let page: usize = params.parse("page").unwrap_or_default(); + let limit: usize = params.parse("limit").unwrap_or_default(); + + let range_start = params.parse::("range-start").unwrap_or_default(); + let range_end = params.parse::("range-end").unwrap_or(u64::MAX); + let max_total = params.parse::("max-total").unwrap_or_default(); + + let mut result = QueuedReports { + ids: Vec::new(), + total: 0, + }; + let from_key = ValueKey::from(ValueClass::Queue(QueueClass::DmarcReportHeader( + ReportEvent { + due: range_start, + policy_hash: 0, + seq_id: 0, + domain: String::new(), + }, + ))); + let to_key = ValueKey::from(ValueClass::Queue(QueueClass::TlsReportHeader( + ReportEvent { + due: range_end, + policy_hash: 0, + seq_id: 0, + domain: String::new(), + }, + ))); + let mut offset = page.saturating_sub(1) * limit; + let mut total_returned = 0; + + server + .core + .storage + .data + .iterate( + IterateParams::new(from_key, to_key).ascending().no_values(), + |key, _| { + if type_.map_or(true, |t| t == *key.last().unwrap()) { + let event = ReportEvent::deserialize(key)?; + if tenant_domains + .as_ref() + .map_or(true, |domains| domains.contains(&event.domain)) + && event.seq_id != 0 + && domain.as_ref().map_or(true, |d| event.domain.contains(d)) + { + if offset == 0 { + if limit == 0 || total_returned < limit { + result.ids.push(if *key.last().unwrap() == 0 { + QueueClass::DmarcReportHeader(event) + } else { + QueueClass::TlsReportHeader(event) + }); + total_returned += 1; + } + } else { + offset -= 1; + } + + result.total += 1; + } + } + + Ok(max_total == 0 || result.total < max_total) + }, + ) + .await + .caused_by(trc::location!()) + .map(|_| result) +} + impl Report { fn dmarc(event: ReportEvent, report: report::Report, rua: Vec) -> Self { Self::Dmarc { diff --git a/crates/jmap/src/api/management/report.rs b/crates/jmap/src/api/management/report.rs index f80a745d..0164a924 100644 --- a/crates/jmap/src/api/management/report.rs +++ b/crates/jmap/src/api/management/report.rs @@ -97,142 +97,15 @@ impl ManageReports for Server { access_token.assert_has_permission(Permission::IncomingReportList)?; let params = UrlParams::new(req.uri().query()); - let filter = params.get("text"); - let page: usize = params.parse::("page").unwrap_or_default(); - let limit: usize = params.parse::("limit").unwrap_or_default(); - let range_start = params.parse::("range-start").unwrap_or_default(); - let range_end = params.parse::("range-end").unwrap_or(u64::MAX); - let max_total = params.parse::("max-total").unwrap_or_default(); - - let (from_key, to_key, typ) = match class { - "dmarc" => ( - ValueKey::from(ValueClass::Report(ReportClass::Dmarc { - id: range_start, - expires: 0, - })), - ValueKey::from(ValueClass::Report(ReportClass::Dmarc { - id: range_end, - expires: u64::MAX, - })), - ReportType::Dmarc, - ), - "tls" => ( - ValueKey::from(ValueClass::Report(ReportClass::Tls { - id: range_start, - expires: 0, - })), - ValueKey::from(ValueClass::Report(ReportClass::Tls { - id: range_end, - expires: u64::MAX, - })), - ReportType::Tls, - ), - "arf" => ( - ValueKey::from(ValueClass::Report(ReportClass::Arf { - id: range_start, - expires: 0, - })), - ValueKey::from(ValueClass::Report(ReportClass::Arf { - id: range_end, - expires: u64::MAX, - })), - ReportType::Arf, - ), - _ => unreachable!(), - }; - - let mut results = Vec::new(); - let mut offset = page.saturating_sub(1) * limit; - let mut total = 0; - let mut last_id = 0; - let has_filters = filter.is_some() || tenant_domains.is_some(); - self.core - .storage - .data - .iterate( - IterateParams::new(from_key, to_key) - .set_values(has_filters) - .descending(), - |key, value| { - // Skip chunked records - let id = key.deserialize_be_u64(U64_LEN + 1)?; - if id == last_id { - return Ok(true); - } - last_id = id; - - // TODO: Support filtering chunked records (over 10MB) on FDB - let matches = if has_filters { - match typ { - ReportType::Dmarc => { - let report = Bincode::< - IncomingReport, - >::deserialize( - value - ) - .caused_by(trc::location!())? - .inner; - - filter.map_or(true, |f| report.contains(f)) - && tenant_domains - .as_ref() - .map_or(true, |domains| report.has_domain(domains)) - } - ReportType::Tls => { - let report = - Bincode::>::deserialize( - value, - ) - .caused_by(trc::location!())? - .inner; - - filter.map_or(true, |f| report.contains(f)) - && tenant_domains - .as_ref() - .map_or(true, |domains| report.has_domain(domains)) - } - ReportType::Arf => { - let report = - Bincode::>::deserialize(value) - .caused_by(trc::location!())? - .inner; - - filter.map_or(true, |f| report.contains(f)) - && tenant_domains - .as_ref() - .map_or(true, |domains| report.has_domain(domains)) - } - } - } else { - true - }; - - if matches { - if offset == 0 { - if limit == 0 || results.len() < limit { - results.push(format!( - "{}_{}", - id, - key.deserialize_be_u64(1)? - )); - } - } else { - offset -= 1; - } - - total += 1; - } - - Ok(max_total == 0 || total < max_total) - }, - ) - .await - .caused_by(trc::location!())?; + let IncomingReports { ids, total } = + fetch_incoming_reports(self, class, ¶ms, &tenant_domains).await?; Ok(JsonResponse::new(json!({ "data": { - "items": results, + "items": ids.into_iter().map(|(id, expires)| { + format!("{id}_{expires}") + }).collect::>(), "total": total, }, })) @@ -312,6 +185,59 @@ impl ManageReports for Server { Err(trc::ResourceEvent::NotFound.into_err()) } } + (class @ ("dmarc" | "tls" | "arf"), None, &Method::DELETE) => { + // Validate the access token + access_token.assert_has_permission(Permission::IncomingReportDelete)?; + + let params = UrlParams::new(req.uri().query()); + + let IncomingReports { ids, .. } = + fetch_incoming_reports(self, class, ¶ms, &tenant_domains).await?; + + let found = !ids.is_empty(); + if found { + let class = match class { + "dmarc" => ReportClass::Dmarc { id: 0, expires: 0 }, + "tls" => ReportClass::Tls { id: 0, expires: 0 }, + "arf" => ReportClass::Arf { id: 0, expires: 0 }, + _ => unreachable!(), + }; + let server = self.clone(); + tokio::spawn(async move { + let mut batch = BatchBuilder::new(); + + for (id, expires) in ids { + let report_id = match &class { + ReportClass::Dmarc { .. } => ReportClass::Dmarc { id, expires }, + ReportClass::Tls { .. } => ReportClass::Tls { id, expires }, + ReportClass::Arf { .. } => ReportClass::Arf { id, expires }, + }; + + batch.clear(ValueClass::Report(report_id)); + + if batch.ops.len() > 1000 { + if let Err(err) = + server.core.storage.data.write(batch.build()).await + { + trc::error!(err.caused_by(trc::location!())); + } + batch = BatchBuilder::new(); + } + } + + if !batch.ops.is_empty() { + if let Err(err) = server.core.storage.data.write(batch.build()).await { + trc::error!(err.caused_by(trc::location!())); + } + } + }); + } + + Ok(JsonResponse::new(json!({ + "data": found, + })) + .into_http_response()) + } (class @ ("dmarc" | "tls" | "arf"), Some(report_id), &Method::DELETE) => { // Validate the access token access_token.assert_has_permission(Permission::IncomingReportDelete)?; @@ -371,6 +297,147 @@ impl ManageReports for Server { } } +struct IncomingReports { + ids: Vec<(u64, u64)>, + total: usize, +} + +async fn fetch_incoming_reports( + server: &Server, + class: &str, + params: &UrlParams<'_>, + tenant_domains: &Option>, +) -> trc::Result { + let filter = params.get("text"); + let page: usize = params.parse::("page").unwrap_or_default(); + let limit: usize = params.parse::("limit").unwrap_or_default(); + + let range_start = params.parse::("range-start").unwrap_or_default(); + let range_end = params.parse::("range-end").unwrap_or(u64::MAX); + let max_total = params.parse::("max-total").unwrap_or_default(); + + let (from_key, to_key, typ) = match class { + "dmarc" => ( + ValueKey::from(ValueClass::Report(ReportClass::Dmarc { + id: range_start, + expires: 0, + })), + ValueKey::from(ValueClass::Report(ReportClass::Dmarc { + id: range_end, + expires: u64::MAX, + })), + ReportType::Dmarc, + ), + "tls" => ( + ValueKey::from(ValueClass::Report(ReportClass::Tls { + id: range_start, + expires: 0, + })), + ValueKey::from(ValueClass::Report(ReportClass::Tls { + id: range_end, + expires: u64::MAX, + })), + ReportType::Tls, + ), + "arf" => ( + ValueKey::from(ValueClass::Report(ReportClass::Arf { + id: range_start, + expires: 0, + })), + ValueKey::from(ValueClass::Report(ReportClass::Arf { + id: range_end, + expires: u64::MAX, + })), + ReportType::Arf, + ), + _ => unreachable!(), + }; + + let mut results = IncomingReports { + ids: Vec::new(), + total: 0, + }; + let mut offset = page.saturating_sub(1) * limit; + let mut last_id = 0; + let has_filters = filter.is_some() || tenant_domains.is_some(); + + server + .core + .storage + .data + .iterate( + IterateParams::new(from_key, to_key) + .set_values(has_filters) + .descending(), + |key, value| { + // Skip chunked records + let id = key.deserialize_be_u64(U64_LEN + 1)?; + if id == last_id { + return Ok(true); + } + last_id = id; + + // TODO: Support filtering chunked records (over 10MB) on FDB + let matches = if has_filters { + match typ { + ReportType::Dmarc => { + let report = + Bincode::>::deserialize( + value, + ) + .caused_by(trc::location!())? + .inner; + + filter.map_or(true, |f| report.contains(f)) + && tenant_domains + .as_ref() + .map_or(true, |domains| report.has_domain(domains)) + } + ReportType::Tls => { + let report = Bincode::>::deserialize(value) + .caused_by(trc::location!())? + .inner; + + filter.map_or(true, |f| report.contains(f)) + && tenant_domains + .as_ref() + .map_or(true, |domains| report.has_domain(domains)) + } + ReportType::Arf => { + let report = Bincode::>::deserialize(value) + .caused_by(trc::location!())? + .inner; + + filter.map_or(true, |f| report.contains(f)) + && tenant_domains + .as_ref() + .map_or(true, |domains| report.has_domain(domains)) + } + } + } else { + true + }; + + if matches { + if offset == 0 { + if limit == 0 || results.ids.len() < limit { + results.ids.push((id, key.deserialize_be_u64(1)?)); + } + } else { + offset -= 1; + } + + results.total += 1; + } + + Ok(max_total == 0 || results.total < max_total) + }, + ) + .await + .caused_by(trc::location!()) + .map(|_| results) +} + fn parse_incoming_report_id(class: &str, id: &str) -> Option { let mut parts = id.split('_'); let id = parts.next()?.parse().ok()?; diff --git a/crates/jmap/src/api/management/settings.rs b/crates/jmap/src/api/management/settings.rs index fcec4d13..ff41763d 100644 --- a/crates/jmap/src/api/management/settings.rs +++ b/crates/jmap/src/api/management/settings.rs @@ -18,12 +18,15 @@ use std::future::Future; #[derive(Debug, serde::Serialize, serde::Deserialize)] #[serde(tag = "type")] +#[serde(rename_all = "camelCase")] pub enum UpdateSettings { Delete { keys: Vec, }, Clear { prefix: String, + #[serde(default)] + filter: Option, }, Insert { prefix: Option, @@ -151,26 +154,29 @@ impl ManageSettings for Server { })) .into_http_response()) } else { - let total = settings.len(); - let items = settings - .into_iter() - .filter_map(|(k, v)| { - if filter.is_empty() - || k.to_lowercase().contains(&filter) - || v.to_lowercase().contains(&filter) - { - let k = k.strip_prefix(&prefix).map(|k| k.to_string()).unwrap_or(k); - Some(json!({ - "_id": k, - "_value": v, - })) + let mut total = 0; + let mut items = Vec::new(); + + for (k, v) in settings { + if filter.is_empty() + || k.to_lowercase().contains(&filter) + || v.to_lowercase().contains(&filter) + { + if offset == 0 { + if limit == 0 || items.len() < limit { + let k = + k.strip_prefix(&prefix).map(|k| k.to_string()).unwrap_or(k); + items.push(json!({ + "_id": k, + "_value": v, + })); + } } else { - None + offset -= 1; } - }) - .skip(offset) - .take(if limit == 0 { total } else { limit }) - .collect::>(); + total += 1; + } + } Ok(JsonResponse::new(json!({ "data": { @@ -282,8 +288,20 @@ impl ManageSettings for Server { self.core.storage.config.clear(key).await?; } } - UpdateSettings::Clear { prefix } => { - self.core.storage.config.clear_prefix(&prefix).await?; + UpdateSettings::Clear { prefix, filter } => { + if let Some(filter) = filter { + for (key, value) in + self.core.storage.config.list(&prefix, false).await? + { + if value.to_lowercase().contains(&filter) + || key.to_lowercase().contains(&filter) + { + self.core.storage.config.clear(key).await?; + } + } + } else { + self.core.storage.config.clear_prefix(&prefix).await?; + } } UpdateSettings::Insert { prefix, diff --git a/crates/jmap/src/auth/oauth/mod.rs b/crates/jmap/src/auth/oauth/mod.rs index 7fddc8a1..c9fe1af8 100644 --- a/crates/jmap/src/auth/oauth/mod.rs +++ b/crates/jmap/src/auth/oauth/mod.rs @@ -143,6 +143,7 @@ pub enum ErrorType { #[derive(Debug, Serialize, Deserialize)] #[serde(tag = "type")] +#[serde(rename_all = "camelCase")] pub enum OAuthCodeRequest { Code { client_id: String, diff --git a/crates/jmap/src/email/crypto.rs b/crates/jmap/src/email/crypto.rs index 758177a5..b4fae099 100644 --- a/crates/jmap/src/email/crypto.rs +++ b/crates/jmap/src/email/crypto.rs @@ -69,6 +69,7 @@ pub struct EncryptionParams { #[derive(Debug, serde::Serialize, serde::Deserialize, Default)] #[serde(tag = "type")] +#[serde(rename_all = "camelCase")] pub enum EncryptionType { PGP { algo: Algorithm, diff --git a/crates/smtp/src/queue/manager.rs b/crates/smtp/src/queue/manager.rs index ff7e570d..8102a701 100644 --- a/crates/smtp/src/queue/manager.rs +++ b/crates/smtp/src/queue/manager.rs @@ -78,6 +78,10 @@ impl Queue { (on_hold.into(), false) } Ok(Some(QueueEvent::Paused(paused))) => { + self.core + .data + .queue_status + .store(!paused, Ordering::Relaxed); is_paused = paused; (None, false) } diff --git a/crates/utils/src/config/mod.rs b/crates/utils/src/config/mod.rs index cf7fbcb8..649bc89a 100644 --- a/crates/utils/src/config/mod.rs +++ b/crates/utils/src/config/mod.rs @@ -27,6 +27,7 @@ pub struct Config { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(tag = "type")] +#[serde(rename_all = "camelCase")] pub enum ConfigWarning { Missing, AppliedDefault { default: String }, @@ -37,6 +38,7 @@ pub enum ConfigWarning { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(tag = "type")] +#[serde(rename_all = "camelCase")] pub enum ConfigError { Parse { error: String }, Build { error: String }, diff --git a/tests/src/smtp/management/queue.rs b/tests/src/smtp/management/queue.rs index bb4021d0..0a553b74 100644 --- a/tests/src/smtp/management/queue.rs +++ b/tests/src/smtp/management/queue.rs @@ -446,6 +446,47 @@ async fn manage_queue() { } } + // Bulk cancel + assert_eq!( + api.request::>(Method::GET, "/api/queue/messages?values=1") + .await + .unwrap() + .unwrap_data() + .items + .len(), + 3 + ); + assert!(api + .request::(Method::DELETE, "/api/queue/messages?text=example2.com") + .await + .unwrap() + .unwrap_data()); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + api.request::>(Method::GET, "/api/queue/messages") + .await + .unwrap() + .unwrap_data() + .items + .len(), + 2 + ); + assert!(api + .request::(Method::DELETE, "/api/queue/messages") + .await + .unwrap() + .unwrap_data()); + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + api.request::>(Method::GET, "/api/queue/messages") + .await + .unwrap() + .unwrap_data() + .items + .len(), + 0 + ); + // Test authentication error assert_eq!( reqwest::Client::builder() diff --git a/tests/src/smtp/management/report.rs b/tests/src/smtp/management/report.rs index 28754963..6937e2f0 100644 --- a/tests/src/smtp/management/report.rs +++ b/tests/src/smtp/management/report.rs @@ -223,6 +223,22 @@ async fn manage_reports() { assert!(ids.next().unwrap().is_none()); assert!(ids.next().unwrap().is_some()); assert!(ids.next().unwrap().is_some()); + + // Cancel all reports + assert!(api + .request::(Method::DELETE, "/api/queue/reports") + .await + .unwrap() + .unwrap_data()); + assert_eq!( + api.request::>(Method::GET, "/api/queue/reports") + .await + .unwrap() + .unwrap_data() + .items + .len(), + 0 + ); } impl ManagementApi {