From 08f1a8013a92cd125931b3769f06ce688f2fdbeb Mon Sep 17 00:00:00 2001 From: mdecimus <11444311+mdecimus@users.noreply.github.com> Date: Mon, 2 Mar 2026 19:26:59 +0100 Subject: [PATCH] JMAP Registry API implementation - part 7 --- crates/common/src/enterprise/config.rs | 4 +- crates/http/src/management/telemetry.rs | 4 +- crates/jmap-proto/src/error/set.rs | 4 + crates/jmap/src/registry/get.rs | 4 +- .../src/registry/mapping/archived_item.rs | 273 +++++++++++++++ .../jmap/src/registry/mapping/deleted_item.rs | 80 ----- .../jmap/src/registry/mapping/masked_email.rs | 2 +- crates/jmap/src/registry/mapping/mod.rs | 2 +- crates/jmap/src/registry/mapping/report.rs | 154 ++++++++- .../jmap/src/registry/mapping/spam_sample.rs | 271 +++++++++++++-- crates/jmap/src/registry/set.rs | 83 ++++- crates/registry/src/utils/archived_item.rs | 120 +++++++ crates/registry/src/utils/mod.rs | 1 + crates/registry/src/utils/report.rs | 163 +-------- crates/registry/src/utils/task.rs | 2 + crates/services/src/housekeeper/mod.rs | 2 +- crates/services/src/task_manager/index.rs | 12 +- crates/services/src/task_manager/mod.rs | 6 +- .../services/src/task_manager/restore_item.rs | 105 ++++++ crates/smtp/src/outbound/delivery.rs | 2 +- crates/smtp/src/outbound/local.rs | 2 +- crates/smtp/src/queue/dsn.rs | 2 +- crates/smtp/src/reporting/analysis.rs | 219 +----------- crates/smtp/src/reporting/dkim.rs | 2 +- crates/smtp/src/reporting/dmarc.rs | 166 ++++----- crates/smtp/src/reporting/inbound.rs | 218 ++++++++++++ crates/smtp/src/reporting/index.rs | 317 ++++++++++++++++++ crates/smtp/src/reporting/mod.rs | 258 +------------- crates/smtp/src/reporting/send.rs | 182 ++++++++++ crates/smtp/src/reporting/spf.rs | 2 +- crates/smtp/src/reporting/tls.rs | 98 +++--- crates/store/src/registry/mod.rs | 6 +- crates/store/src/write/key.rs | 4 +- 33 files changed, 1843 insertions(+), 927 deletions(-) create mode 100644 crates/jmap/src/registry/mapping/archived_item.rs delete mode 100644 crates/jmap/src/registry/mapping/deleted_item.rs create mode 100644 crates/registry/src/utils/archived_item.rs create mode 100644 crates/services/src/task_manager/restore_item.rs create mode 100644 crates/smtp/src/reporting/inbound.rs create mode 100644 crates/smtp/src/reporting/index.rs create mode 100644 crates/smtp/src/reporting/send.rs diff --git a/crates/common/src/enterprise/config.rs b/crates/common/src/enterprise/config.rs index 8a8288e9..22434095 100644 --- a/crates/common/src/enterprise/config.rs +++ b/crates/common/src/enterprise/config.rs @@ -170,7 +170,9 @@ impl Enterprise { // Build the enterprise configuration let mut enterprise = Enterprise { license, - undelete_retention: dr.hold_deleted_for.map(|retention| retention.into_inner()), + undelete_retention: dr + .archive_deleted_items_for + .map(|retention| retention.into_inner()), logo_url, metrics_alerts: Default::default(), spam_filter_llm: SpamFilterLlmConfig::parse(bp, &ai_apis_ids).await, diff --git a/crates/http/src/management/telemetry.rs b/crates/http/src/management/telemetry.rs index 54715fd2..52529b91 100644 --- a/crates/http/src/management/telemetry.rs +++ b/crates/http/src/management/telemetry.rs @@ -238,8 +238,8 @@ impl TelemetryApi for Server { if metric_types.contains(&metric_type) { let value = match metric_type { MetricType::QueueCount => self.total_queued_messages().await?, - MetricType::UserCount => self.total_accounts().await?, - MetricType::DomainCount => self.total_domains().await?, + MetricType::UserCount => self.total_accounts().await? as u64, + MetricType::DomainCount => self.total_domains().await? as u64, _ => unreachable!(), }; Collector::update_gauge(metric_type, value); diff --git a/crates/jmap-proto/src/error/set.rs b/crates/jmap-proto/src/error/set.rs index 7976a1f4..b60a9ed3 100644 --- a/crates/jmap-proto/src/error/set.rs +++ b/crates/jmap-proto/src/error/set.rs @@ -220,6 +220,10 @@ impl SetError { Self::new(SetErrorType::InvalidProperties) } + pub fn invalid_patch() -> Self { + Self::new(SetErrorType::InvalidPatch) + } + pub fn forbidden() -> Self { Self::new(SetErrorType::Forbidden) } diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index 0ce6f0b2..8ee64a2e 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -7,7 +7,7 @@ use crate::registry::mapping::{ RegistryGetResponse, account::account_get, - deleted_item::deleted_item_get, + archived_item::archived_item_get, log::log_get, queued_message::queued_message_get, report::report_get, @@ -268,7 +268,7 @@ impl RegistryGet for Server { | ObjectType::DmarcInternalReport | ObjectType::TlsInternalReport => report_get(get).await.map(|get| get.into_response()), - ObjectType::DeletedItem => deleted_item_get(get).await.map(|get| get.into_response()), + ObjectType::ArchivedItem => archived_item_get(get).await.map(|get| get.into_response()), ObjectType::SpamTrainingSample => { spam_sample_get(get).await.map(|get| get.into_response()) } diff --git a/crates/jmap/src/registry/mapping/archived_item.rs b/crates/jmap/src/registry/mapping/archived_item.rs new file mode 100644 index 00000000..991d6db9 --- /dev/null +++ b/crates/jmap/src/registry/mapping/archived_item.rs @@ -0,0 +1,273 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::registry::mapping::{RegistryGetResponse, RegistrySetResponse}; +use jmap_proto::error::set::SetError; +use jmap_tools::{Key, Value}; +use registry::{ + jmap::IntoValue, + pickle::Pickle, + schema::{ + enums::ArchivedItemStatus, + prelude::{Object, ObjectType, Property}, + structs::{ArchivedItem, Task, TaskRestoreArchivedItem, TaskStatus}, + }, + types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, +}; +use std::str::FromStr; +use store::{ + SerializeInfallible, ValueKey, + ahash::AHashSet, + registry::RegistryQuery, + write::{BatchBuilder, BlobLink, BlobOp, RegistryClass, ValueClass, assert::AssertValue}, +}; +use trc::AddContext; +use types::{blob::BlobClass, id::Id}; + +pub(crate) async fn archived_item_set( + mut set: RegistrySetResponse<'_>, +) -> trc::Result> { + // Archived items cannot be created + set.fail_all_create("Archived items cannot be created"); + + let mut batch = BatchBuilder::new(); + let object_id = set.object_type.to_id(); + 'outer: for (id, value) in set.update.drain(..) { + // Extract new deliverAt value + let mut status = ArchivedItemStatus::Archived; + let mut archived_until = None; + let now = UTCDateTime::now(); + for (key, value) in value.into_expanded_object() { + match (key, value) { + (Key::Property(Property::Status), Value::Str(status_)) => { + let Some(status_) = ArchivedItemStatus::parse(&status_) else { + set.response.not_updated.append( + id, + SetError::invalid_patch() + .with_property(Property::Status) + .with_description("Invalid value for property"), + ); + continue 'outer; + }; + status = status_; + } + (Key::Property(Property::ArchivedUntil), Value::Str(archived_until_)) => { + archived_until = UTCDateTime::from_str(archived_until_.as_ref()) + .ok() + .filter(|da| *da > now); + if archived_until.is_none() { + set.response.not_updated.append( + id, + SetError::invalid_patch() + .with_property(Property::ArchivedUntil) + .with_description("Invalid value for property"), + ); + continue 'outer; + } + } + (Key::Property(Property::Id), _) => {} + (key, _) => { + set.response.not_updated.append( + id, + SetError::invalid_properties().with_property(key.into_owned()), + ); + continue 'outer; + } + } + } + + let item_id = id.id(); + if (status == ArchivedItemStatus::RequestRestore || archived_until.is_some()) + && let Some(item) = + set.server + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { object_id, item_id }, + ))) + .await? + .filter(|item| { + !set.is_account_filtered + || item.inner.account_id() == Some(set.account_id.into()) + }) + { + let revision = item.revision; + if let Some(archived_until) = archived_until + && status != ArchivedItemStatus::Archived + { + // Update archivedUntil + let mut item = ArchivedItem::from(item); + let old_archived_until = item.archived_until(); + + if old_archived_until != archived_until { + item.set_archived_until(archived_until); + let blob_hash = item.blob_id().hash.clone(); + + batch + .with_account_id(item.account_id().document_id()) + .assert_value( + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + AssertValue::Hash(revision), + ) + .clear(BlobOp::Link { + hash: blob_hash.clone(), + to: BlobLink::Temporary { + until: old_archived_until.timestamp() as u64, + }, + }) + .set( + BlobOp::Link { + hash: blob_hash, + to: BlobLink::Temporary { + until: archived_until.timestamp() as u64, + }, + }, + ObjectId::new(ObjectType::ArchivedItem, item_id.into()).serialize(), + ) + .set( + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + item.to_pickled_vec(), + ); + } + } else { + // Schedule restore task + let item = ArchivedItem::from(item); + let account_id = item.account_id(); + + batch + .assert_value( + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + AssertValue::Hash(revision), + ) + .clear(ValueClass::Registry(RegistryClass::Index { + index_id: Property::AccountId.to_id(), + object_id, + item_id, + key: account_id.id().serialize(), + })) + .clear(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id, + })) + .schedule_task(Task::RestoreArchivedItem(TaskRestoreArchivedItem { + account_id, + archived_item_type: item.object_type(), + archived_until: item.archived_until(), + blob_id: item.blob_id().clone(), + created_at: item.created_at(), + status: TaskStatus::now(), + })); + } + + batch.commit_point(); + + set.response.updated.append(id, None); + } else { + set.response.not_updated.append(id, SetError::not_found()); + } + } + + // Process items to destroy + for id in set.destroy.drain(..) { + let item_id = id.id(); + + if let Some(item) = + set.server + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { object_id, item_id }, + ))) + .await? + .filter(|item| { + !set.is_account_filtered || item.account_id().document_id() == set.account_id + }) + { + let account_id = item.account_id().id(); + let until = item.archived_until().timestamp() as u64; + let blob_hash = item.into_blob_id().hash; + + batch + .with_account_id(account_id as u32) + .clear(BlobOp::Link { + hash: blob_hash, + to: BlobLink::Temporary { until }, + }) + .clear(ValueClass::Registry(RegistryClass::Index { + index_id: Property::AccountId.to_id(), + object_id, + item_id, + key: account_id.serialize(), + })) + .clear(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id, + })) + .commit_point(); + + set.response.destroyed.push(id); + } else { + set.response.not_destroyed.append(id, SetError::not_found()); + } + } + + if !batch.is_empty() { + set.server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + + Ok(set) +} + +pub(crate) async fn archived_item_get( + mut get: RegistryGetResponse<'_>, +) -> trc::Result> { + let object_id = get.object_type.to_id(); + let ids = if let Some(ids) = get.ids.take() { + ids + } else { + get.server + .registry() + .query::>( + RegistryQuery::new(get.object_type).with_account(get.account_id), + ) + .await? + .into_iter() + .take(get.server.core.jmap.get_max_objects) + .map(Id::from) + .collect() + }; + + for id in ids { + if let Some(mut item) = get + .server + .store() + .get_value::(ValueKey::from(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id: id.id(), + }))) + .await? + .filter(|item| { + !get.is_account_filtered || item.account_id().document_id() == get.account_id + }) + { + if get.is_account_filtered { + let archived_until = item.archived_until(); + item.blob_id_mut().class = BlobClass::Reserved { + account_id: get.account_id, + expires: archived_until.timestamp() as u64, + }; + } + + get.insert(id, item.into_value()); + } else { + get.not_found(id); + } + } + + Ok(get) +} diff --git a/crates/jmap/src/registry/mapping/deleted_item.rs b/crates/jmap/src/registry/mapping/deleted_item.rs deleted file mode 100644 index 2a16d2eb..00000000 --- a/crates/jmap/src/registry/mapping/deleted_item.rs +++ /dev/null @@ -1,80 +0,0 @@ -/* - * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC - * - * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL - */ - -use crate::registry::mapping::RegistryGetResponse; -use registry::{ - jmap::IntoValue, - schema::{ - prelude::{Object, ObjectInner}, - structs::{DeletedEmail, DeletedFileNode, DeletedItem}, - }, - types::EnumImpl, -}; -use store::{ - ValueKey, - ahash::AHashSet, - registry::RegistryQuery, - write::{RegistryClass, ValueClass}, -}; -use types::{blob::BlobClass, id::Id}; - -pub(crate) async fn deleted_item_get( - mut get: RegistryGetResponse<'_>, -) -> trc::Result> { - let object_id = get.object_type.to_id(); - let ids = if let Some(ids) = get.ids.take() { - ids - } else { - get.server - .registry() - .query::>( - RegistryQuery::new(get.object_type).with_account(get.account_id), - ) - .await? - .into_iter() - .take(get.server.core.jmap.get_max_objects) - .map(Id::from) - .collect() - }; - - for id in ids { - if let Some(mut item) = get - .server - .store() - .get_value::(ValueKey::from(ValueClass::Registry(RegistryClass::Item { - object_id, - item_id: id.id(), - }))) - .await? - { - if get.is_account_filtered - && let ObjectInner::DeletedItem( - DeletedItem::Email(DeletedEmail { - blob_id, - cleanup_at, - .. - }) - | DeletedItem::FileNode(DeletedFileNode { - blob_id, - cleanup_at, - .. - }), - ) = &mut item.inner - { - blob_id.class = BlobClass::Reserved { - account_id: get.account_id, - expires: cleanup_at.timestamp() as u64, - }; - } - - get.insert(id, item.into_value()); - } else { - get.not_found(id); - } - } - - Ok(get) -} diff --git a/crates/jmap/src/registry/mapping/masked_email.rs b/crates/jmap/src/registry/mapping/masked_email.rs index 12d122b0..a414676c 100644 --- a/crates/jmap/src/registry/mapping/masked_email.rs +++ b/crates/jmap/src/registry/mapping/masked_email.rs @@ -16,7 +16,7 @@ use registry::{ structs::MaskedEmail, }, }; -use store::{ahash::AHashSet, registry::RegistryQuery, write::now}; +use store::{registry::RegistryQuery, write::now}; use utils::{DomainPart, map::vec_map::VecMap}; pub(crate) async fn validate_masked_email( diff --git a/crates/jmap/src/registry/mapping/mod.rs b/crates/jmap/src/registry/mapping/mod.rs index 9abf9264..25d6141c 100644 --- a/crates/jmap/src/registry/mapping/mod.rs +++ b/crates/jmap/src/registry/mapping/mod.rs @@ -20,7 +20,7 @@ use types::id::Id; use utils::map::vec_map::VecMap; pub mod account; -pub mod deleted_item; +pub mod archived_item; pub mod log; pub mod masked_email; pub mod principal; diff --git a/crates/jmap/src/registry/mapping/report.rs b/crates/jmap/src/registry/mapping/report.rs index 8de85fec..21aca2fc 100644 --- a/crates/jmap/src/registry/mapping/report.rs +++ b/crates/jmap/src/registry/mapping/report.rs @@ -4,22 +4,159 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::RegistryGetResponse; +use crate::registry::mapping::{RegistryGetResponse, RegistrySetResponse}; use common::Server; +use jmap_proto::error::set::SetError; +use jmap_tools::{Key, Value}; use registry::{ jmap::IntoValue, - schema::prelude::{Object, ObjectType, Property}, - types::EnumImpl, + schema::prelude::{Object, ObjectInner, ObjectType, Property}, + types::{EnumImpl, datetime::UTCDateTime}, }; +use smtp::reporting::index::{ExternalReportIndex, InternalReportIndex}; +use std::str::FromStr; use store::{ IterateParams, U16_LEN, ValueKey, ahash::AHashSet, registry::RegistryQuery, - write::{RegistryClass, ValueClass, key::DeserializeBigEndian}, + write::{BatchBuilder, RegistryClass, ValueClass, key::DeserializeBigEndian}, }; use trc::AddContext; use types::id::Id; +pub(crate) async fn report_set( + mut set: RegistrySetResponse<'_>, +) -> trc::Result> { + let object_id = set.object_type.to_id(); + + // Reports cannot be created + set.fail_all_create("Reports cannot be created"); + + let mut batch = BatchBuilder::new(); + if matches!( + set.object_type, + ObjectType::DmarcInternalReport | ObjectType::TlsInternalReport + ) { + let now = UTCDateTime::now(); + 'outer: for (id, value) in set.update.drain(..) { + // Extract new deliverAt value + let mut deliver_at = None; + for (key, value) in value.into_expanded_object() { + match (key, value) { + (Key::Property(Property::DeliverAt), Value::Str(deliver_at_)) => { + deliver_at = UTCDateTime::from_str(deliver_at_.as_ref()) + .ok() + .filter(|da| *da > now); + if deliver_at.is_none() { + set.response.not_updated.append( + id, + SetError::invalid_patch() + .with_property(Property::DeliverAt) + .with_description("Invalid value for property"), + ); + continue 'outer; + } + } + (Key::Property(Property::Id), _) => {} + (key, _) => { + set.response.not_updated.append( + id, + SetError::invalid_properties().with_property(key.into_owned()), + ); + continue 'outer; + } + } + } + let Some(deliver_at) = deliver_at else { + set.response.not_updated.append( + id, + SetError::invalid_patch() + .with_property(Key::Property(Property::DeliverAt)) + .with_description("Missing required property"), + ); + continue; + }; + + let item_id = id.id(); + let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); + if let Some(mut report_obj) = set + .server + .store() + .get_value::(ValueKey::from(key.clone())) + .await? + { + match &mut report_obj.inner { + ObjectInner::DmarcInternalReport(report) => { + report.reschedule_ops(&mut batch, item_id, report_obj.revision, deliver_at); + } + ObjectInner::TlsInternalReport(report) => { + report.reschedule_ops(&mut batch, item_id, report_obj.revision, deliver_at); + } + _ => {} + } + batch.commit_point(); + + set.response.updated.append(id, None); + } else { + set.response.not_updated.append(id, SetError::not_found()); + } + } + } else { + // External reports cannot be updated + set.fail_all_update("External reports cannot be updated"); + } + + // Process reports to destroy + let tenant_id = set.access_token.tenant_id().map(Id::from); + for id in set.destroy.drain(..) { + let item_id = id.id(); + let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); + if let Some(report) = set + .server + .store() + .get_value::(ValueKey::from(key.clone())) + .await? + .filter(|report| { + !set.is_tenant_filtered || report.inner.member_tenant_id() == tenant_id + }) + { + match &report.inner { + ObjectInner::DmarcExternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + ObjectInner::TlsExternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + ObjectInner::ArfExternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + ObjectInner::DmarcInternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + ObjectInner::TlsInternalReport(report) => { + report.write_ops(&mut batch, item_id, false); + } + _ => {} + } + batch.commit_point(); + + set.response.destroyed.push(id); + } else { + set.response.not_destroyed.append(id, SetError::not_found()); + } + } + + if !batch.is_empty() { + set.server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + + Ok(set) +} + pub(crate) async fn report_get( mut get: RegistryGetResponse<'_>, ) -> trc::Result> { @@ -60,12 +197,11 @@ pub(crate) async fn report_get( item_id: id.id(), }))) .await? + .filter(|report| { + !get.is_tenant_filtered || report.inner.member_tenant_id() == tenant_id + }) { - if !get.is_tenant_filtered || report.inner.member_tenant_id() == tenant_id { - get.insert(id, report.into_value()); - } else { - get.not_found(id); - } + get.insert(id, report.into_value()); } else { get.not_found(id); } diff --git a/crates/jmap/src/registry/mapping/spam_sample.rs b/crates/jmap/src/registry/mapping/spam_sample.rs index 7bc8f36f..f9087c1c 100644 --- a/crates/jmap/src/registry/mapping/spam_sample.rs +++ b/crates/jmap/src/registry/mapping/spam_sample.rs @@ -4,20 +4,244 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::registry::mapping::RegistryGetResponse; +use crate::{ + blob::download::BlobDownload, + registry::mapping::{RegistryGetResponse, RegistrySetResponse}, +}; +use jmap_proto::error::set::SetError; +use jmap_tools::{JsonPointer, JsonPointerItem, Key}; +use mail_parser::{MessageParser, parsers::fields::thread::thread_name}; use registry::{ - jmap::IntoValue, - schema::prelude::{Object, ObjectInner, Property}, - types::EnumImpl, + jmap::{IntoValue, JsonPointerPatch, RegistryJsonPatch}, + pickle::Pickle, + schema::{ + prelude::{ObjectType, Property}, + structs::SpamTrainingSample, + }, + types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, }; use store::{ - ValueKey, + SerializeInfallible, ValueKey, ahash::AHashSet, registry::RegistryQuery, - write::{RegistryClass, ValueClass}, + write::{BatchBuilder, BlobLink, BlobOp, RegistryClass, ValueClass, now}, }; +use trc::AddContext; use types::{blob::BlobClass, id::Id}; +pub(crate) async fn spam_sample_set( + mut set: RegistrySetResponse<'_>, +) -> trc::Result> { + // Spam samples cannot be modified + set.fail_all_update("Spam training samples cannot be modified."); + + let mut batch = BatchBuilder::new(); + let object_id = set.object_type.to_id(); + + // Process samples to create + let hold_samples_for = set + .server + .core + .spam + .classifier + .as_ref() + .map(|config| config.hold_samples_for); + let now = now(); + 'outer: for (id, value) in set.create.drain() { + let mut sample = SpamTrainingSample::default(); + let Some(expires_at) = hold_samples_for.map(|d| now + d) else { + set.response.not_created.append( + id, + SetError::forbidden() + .with_description("Spam classifier is not configured on the server"), + ); + continue; + }; + + for (key, value) in value.into_expanded_object() { + let Key::Property(prop) = key else { + set.response.not_created.append( + id, + SetError::invalid_properties().with_property(key.into_owned()), + ); + continue 'outer; + }; + + if let Err(err) = sample.patch( + JsonPointerPatch::new(&JsonPointer::new(vec![JsonPointerItem::Key( + Key::Property(prop), + )])) + .with_create(true), + value, + ) { + set.response.not_created.append(id, err.into()); + continue 'outer; + }; + } + + if sample.blob_id.hash.is_empty() { + set.response.not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::BlobId) + .with_description("blobId is required"), + ); + continue; + } + + let Some(bytes) = set + .server + .blob_download(&sample.blob_id, set.access_token) + .await? + else { + set.response.not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::BlobId) + .with_description("blobId does not exist or is not accessible"), + ); + continue; + }; + + if bytes.len() > set.server.core.email.mail_max_size { + set.response.not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::BlobId) + .with_description(format!( + "blob size exceeds maximum of {} bytes", + set.server.core.email.mail_max_size + )), + ); + continue; + } + + let Some(message) = MessageParser::new().parse(&bytes) else { + set.response.not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::BlobId) + .with_description("Blob content is not a valid email message"), + ); + continue; + }; + + let (Some(subject), Some(from)) = ( + message.subject().map(thread_name), + message + .from() + .and_then(|from| from.first().and_then(|addr| addr.address())), + ) else { + set.response.not_created.append( + id, + SetError::invalid_properties() + .with_property(Property::BlobId) + .with_description("Email message must have a subject and from header"), + ); + continue; + }; + + sample.subject = subject.to_string(); + sample.from = from.to_lowercase(); + sample.expires_at = UTCDateTime::from_timestamp(expires_at as i64); + if set.is_account_filtered { + sample.account_id = Some(set.account_id.into()); + } + + // Write sample to store + let item_id = set.server.registry().assign_id(); + batch + .set( + BlobOp::Link { + hash: sample.blob_id.hash.clone(), + to: BlobLink::Temporary { until: expires_at }, + }, + ObjectId::new(ObjectType::SpamTrainingSample, item_id.into()).serialize(), + ) + .set( + ValueClass::Registry(RegistryClass::Index { + index_id: Property::AccountId.to_id(), + object_id, + item_id, + key: sample + .account_id + .map(|id| id.id()) + .unwrap_or(u32::MAX as u64) + .serialize(), + }), + vec![], + ) + .set( + ValueClass::Registry(RegistryClass::Item { object_id, item_id }), + sample.to_pickled_vec(), + ); + + set.response.created(id, item_id); + } + + // Process samples to destroy + for id in set.destroy.drain(..) { + let item_id = id.id(); + + if let Some(sample) = set + .server + .store() + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { + object_id, + item_id: id.id(), + }, + ))) + .await? + .filter(|sample| { + !set.is_account_filtered + || sample + .account_id + .is_some_and(|account_id| account_id.document_id() == set.account_id) + }) + { + let account_id = sample + .account_id + .map(|id| id.document_id()) + .unwrap_or(u32::MAX); + + batch + .with_account_id(account_id) + .clear(BlobOp::Link { + hash: sample.blob_id.hash, + to: BlobLink::Temporary { + until: sample.expires_at.timestamp() as u64, + }, + }) + .clear(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id, + })) + .clear(ValueClass::Registry(RegistryClass::Index { + index_id: Property::AccountId.to_id(), + object_id, + item_id, + key: (account_id as u64).serialize(), + })) + .commit_point(); + + set.response.destroyed.push(id); + } else { + set.response.not_destroyed.append(id, SetError::not_found()); + } + } + + if !batch.is_empty() { + set.server + .store() + .write(batch.build_all()) + .await + .caused_by(trc::location!())?; + } + + Ok(set) +} + pub(crate) async fn spam_sample_get( mut get: RegistryGetResponse<'_>, ) -> trc::Result> { @@ -42,32 +266,31 @@ pub(crate) async fn spam_sample_get( }; for id in ids { - if let Some(mut item) = get + if let Some(mut sample) = get .server .store() - .get_value::(ValueKey::from(ValueClass::Registry(RegistryClass::Item { - object_id, - item_id: id.id(), - }))) + .get_value::(ValueKey::from(ValueClass::Registry( + RegistryClass::Item { + object_id, + item_id: id.id(), + }, + ))) .await? + .filter(|sample| { + !get.is_account_filtered + || sample + .account_id + .is_some_and(|account_id| account_id.document_id() == get.account_id) + }) { - if get.is_account_filtered - && let ObjectInner::SpamTrainingSample(item) = &mut item.inner - { - if item - .account_id - .is_none_or(|id| id.document_id() != get.account_id) - { - get.not_found(id); - continue; - } - item.blob_id.class = BlobClass::Reserved { + if get.is_account_filtered { + sample.blob_id.class = BlobClass::Reserved { account_id: get.account_id, - expires: item.expires_at.timestamp() as u64, + expires: sample.expires_at.timestamp() as u64, }; } - get.insert(id, item.into_value()); + get.insert(id, sample.into_value()); } else { get.not_found(id); } diff --git a/crates/jmap/src/registry/set.rs b/crates/jmap/src/registry/set.rs index 2e517f00..c017201a 100644 --- a/crates/jmap/src/registry/set.rs +++ b/crates/jmap/src/registry/set.rs @@ -4,11 +4,16 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +use std::borrow::Cow; + use crate::registry::mapping::{ ObjectResponse, RegistrySetResponse, + archived_item::archived_item_set, masked_email::validate_masked_email, principal::{validate_account, validate_role, validate_tenant_quota}, public_key::validate_public_key, + report::report_set, + spam_sample::spam_sample_set, }; use common::{Server, auth::AccessToken, cache::invalidate::CacheInvalidationBuilder}; use jmap_proto::{ @@ -494,21 +499,40 @@ impl RegistrySet for Server { // Finalize cache invalidation self.invalidate_caches(cache_invalidator).await?; + + Ok(set.into_response()) + } + ObjectType::ArfExternalReport + | ObjectType::DmarcExternalReport + | ObjectType::TlsExternalReport + | ObjectType::DmarcInternalReport + | ObjectType::TlsInternalReport => report_set(set).await.map(|set| set.into_response()), + + ObjectType::ArchivedItem => archived_item_set(set).await.map(|set| set.into_response()), + + ObjectType::SpamTrainingSample => { + spam_sample_set(set).await.map(|set| set.into_response()) + } + + ObjectType::AccountSettings => { + todo!() + } + ObjectType::Credential => { + todo!() + } + + ObjectType::QueuedMessage => { + todo!() + } + ObjectType::Task => { + todo!() + } + ObjectType::Log | ObjectType::Metric | ObjectType::Trace => { + 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"); + Ok(set.into_response()) } - ObjectType::QueuedMessage => {} - ObjectType::Task => {} - ObjectType::ArfExternalReport => {} - ObjectType::DmarcExternalReport => {} - ObjectType::TlsExternalReport => {} - ObjectType::DeletedItem => {} - ObjectType::Metric => {} - ObjectType::Trace => {} - ObjectType::SpamTrainingSample => {} - ObjectType::DmarcInternalReport => {} - ObjectType::TlsInternalReport => {} - ObjectType::Log => {} - ObjectType::AccountSettings => {} - ObjectType::Credential => {} } // Schedule account and tenant deletions @@ -519,8 +543,7 @@ impl RegistrySet for Server { // Domain = trigger DNIM stuff // Validate expressions // Fallback admin password from env or files - - Ok(set.into_response()) + // Individual permissions for each object + create/update/destroy } } @@ -570,6 +593,34 @@ impl RegistrySetResponse<'_> { } } + pub fn fail_all_create(&mut self, error: impl Into>) { + let error = error.into(); + for (client_id, _) in self.create.drain() { + self.response.not_created.append( + client_id, + SetError::forbidden().with_description(error.clone()), + ); + } + } + + pub fn fail_all_update(&mut self, error: impl Into>) { + let error = error.into(); + for (id, _) in self.update.drain(..) { + self.response + .not_updated + .append(id, SetError::forbidden().with_description(error.clone())); + } + } + + pub fn fail_all_destroy(&mut self, error: impl Into>) { + let error = error.into(); + for id in self.destroy.drain(..) { + self.response + .not_destroyed + .append(id, SetError::forbidden().with_description(error.clone())); + } + } + fn into_response(self) -> SetResponse { self.response } diff --git a/crates/registry/src/utils/archived_item.rs b/crates/registry/src/utils/archived_item.rs new file mode 100644 index 00000000..597ff86c --- /dev/null +++ b/crates/registry/src/utils/archived_item.rs @@ -0,0 +1,120 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::schema::prelude::{ArchivedItem, UTCDateTime}; +use types::{blob::BlobId, id::Id}; + +impl ArchivedItem { + pub fn account_id(&self) -> Id { + match self { + ArchivedItem::Email(i) => i.account_id, + ArchivedItem::FileNode(i) => i.account_id, + ArchivedItem::CalendarEvent(i) => i.account_id, + ArchivedItem::ContactCard(i) => i.account_id, + ArchivedItem::SieveScript(i) => i.account_id, + } + } + + pub fn blob_id(&self) -> &BlobId { + match self { + ArchivedItem::Email(i) => &i.blob_id, + ArchivedItem::FileNode(i) => &i.blob_id, + ArchivedItem::CalendarEvent(i) => &i.blob_id, + ArchivedItem::ContactCard(i) => &i.blob_id, + ArchivedItem::SieveScript(i) => &i.blob_id, + } + } + + pub fn archived_until(&self) -> UTCDateTime { + match self { + ArchivedItem::Email(i) => i.archived_until, + ArchivedItem::FileNode(i) => i.archived_until, + ArchivedItem::CalendarEvent(i) => i.archived_until, + ArchivedItem::ContactCard(i) => i.archived_until, + ArchivedItem::SieveScript(i) => i.archived_until, + } + } + + pub fn created_at(&self) -> UTCDateTime { + match self { + ArchivedItem::Email(i) => i.received_at, + ArchivedItem::FileNode(i) => i.created_at, + ArchivedItem::CalendarEvent(i) => i.created_at, + ArchivedItem::ContactCard(i) => i.created_at, + ArchivedItem::SieveScript(i) => i.created_at, + } + } + + pub fn set_account_id(&mut self, value: Id) { + match self { + ArchivedItem::Email(i) => i.account_id = value, + ArchivedItem::FileNode(i) => i.account_id = value, + ArchivedItem::CalendarEvent(i) => i.account_id = value, + ArchivedItem::ContactCard(i) => i.account_id = value, + ArchivedItem::SieveScript(i) => i.account_id = value, + } + } + + pub fn set_blob_id(&mut self, value: BlobId) { + match self { + ArchivedItem::Email(i) => i.blob_id = value, + ArchivedItem::FileNode(i) => i.blob_id = value, + ArchivedItem::CalendarEvent(i) => i.blob_id = value, + ArchivedItem::ContactCard(i) => i.blob_id = value, + ArchivedItem::SieveScript(i) => i.blob_id = value, + } + } + + pub fn set_archived_until(&mut self, value: UTCDateTime) { + match self { + ArchivedItem::Email(i) => i.archived_until = value, + ArchivedItem::FileNode(i) => i.archived_until = value, + ArchivedItem::CalendarEvent(i) => i.archived_until = value, + ArchivedItem::ContactCard(i) => i.archived_until = value, + ArchivedItem::SieveScript(i) => i.archived_until = value, + } + } + + pub fn account_id_mut(&mut self) -> &mut Id { + match self { + ArchivedItem::Email(i) => &mut i.account_id, + ArchivedItem::FileNode(i) => &mut i.account_id, + ArchivedItem::CalendarEvent(i) => &mut i.account_id, + ArchivedItem::ContactCard(i) => &mut i.account_id, + ArchivedItem::SieveScript(i) => &mut i.account_id, + } + } + + pub fn blob_id_mut(&mut self) -> &mut BlobId { + match self { + ArchivedItem::Email(i) => &mut i.blob_id, + ArchivedItem::FileNode(i) => &mut i.blob_id, + ArchivedItem::CalendarEvent(i) => &mut i.blob_id, + ArchivedItem::ContactCard(i) => &mut i.blob_id, + ArchivedItem::SieveScript(i) => &mut i.blob_id, + } + } + + pub fn archived_until_mut(&mut self) -> &mut UTCDateTime { + match self { + ArchivedItem::Email(i) => &mut i.archived_until, + ArchivedItem::FileNode(i) => &mut i.archived_until, + ArchivedItem::CalendarEvent(i) => &mut i.archived_until, + ArchivedItem::ContactCard(i) => &mut i.archived_until, + ArchivedItem::SieveScript(i) => &mut i.archived_until, + } + } + + pub fn into_blob_id(self) -> BlobId { + match self { + ArchivedItem::Email(i) => i.blob_id, + ArchivedItem::FileNode(i) => i.blob_id, + ArchivedItem::CalendarEvent(i) => i.blob_id, + ArchivedItem::ContactCard(i) => i.blob_id, + ArchivedItem::SieveScript(i) => i.blob_id, + } + } +} diff --git a/crates/registry/src/utils/mod.rs b/crates/registry/src/utils/mod.rs index 9d9e90c6..3f8ae516 100644 --- a/crates/registry/src/utils/mod.rs +++ b/crates/registry/src/utils/mod.rs @@ -8,6 +8,7 @@ use crate::schema::prelude::Roles; use types::id::Id; pub mod account; +pub mod archived_item; pub mod cron; pub mod http; pub mod report; diff --git a/crates/registry/src/utils/report.rs b/crates/registry/src/utils/report.rs index fc930b0b..93b1718b 100644 --- a/crates/registry/src/utils/report.rs +++ b/crates/registry/src/utils/report.rs @@ -5,160 +5,11 @@ */ use crate::{ - schema::{ - enums, - prelude::{Property, UTCDateTime}, - structs, - }, - types::{index::IndexBuilder, ipaddr::IpAddr}, + schema::{enums, prelude::UTCDateTime, structs}, + types::ipaddr::IpAddr, }; use mail_auth::report::{tlsrpt::*, *}; use std::borrow::Cow; -use types::id::Id; - -pub trait ReportIndex { - fn text(&self) -> impl Iterator; - - fn tenant_id(&self) -> Option; - - fn expires_at(&self) -> u64; - - fn domains(&self) -> impl Iterator; - - fn build_search_index<'x>(&'x self, index: &mut IndexBuilder<'x>) { - for text in self.text() { - index.text(Property::Domain, text); - } - - if let Some(tenant_id) = self.tenant_id() { - index.search(Property::MemberTenantId, tenant_id.id()); - } - - index.search(Property::ExpiresAt, self.expires_at()); - } -} - -impl ReportIndex for structs::ArfExternalReport { - fn domains(&self) -> impl Iterator { - let report = &self.report; - - report - .reported_domains - .iter() - .filter_map(|s| non_empty(s)) - .chain( - [report.dkim_domain.as_deref()] - .into_iter() - .flatten() - .filter_map(non_empty), - ) - } - - fn text(&self) -> impl Iterator { - let report = &self.report; - - report - .reported_domains - .iter() - .filter_map(|s| non_empty(s)) - .chain( - [ - report.dkim_domain.as_deref(), - report.reporting_mta.as_deref(), - report.original_mail_from.as_deref(), - report.original_rcpt_to.as_deref(), - ] - .into_iter() - .flatten() - .filter_map(non_empty), - ) - .chain(non_empty(&self.from)) - } - - fn tenant_id(&self) -> Option { - self.member_tenant_id - } - - fn expires_at(&self) -> u64 { - self.expires_at.timestamp() as u64 - } -} - -impl ReportIndex for structs::DmarcExternalReport { - fn domains(&self) -> impl Iterator { - let report = &self.report; - - non_empty(&report.policy_domain) - .into_iter() - .filter_map(non_empty) - } - - fn text(&self) -> impl Iterator { - let report = &self.report; - - non_empty(&report.email) - .into_iter() - .filter_map(non_empty) - .chain(non_empty(&report.policy_domain)) - .chain(report.records.iter().flat_map(|r| { - r.envelope_to - .as_deref() - .into_iter() - .filter_map(non_empty) - .chain(non_empty(&r.envelope_from)) - .chain(non_empty(&r.header_from)) - .chain(r.dkim_results.iter().filter_map(|d| non_empty(&d.domain))) - .chain(r.spf_results.iter().filter_map(|s| non_empty(&s.domain))) - })) - .chain(non_empty(&self.from)) - } - - fn tenant_id(&self) -> Option { - self.member_tenant_id - } - - fn expires_at(&self) -> u64 { - self.expires_at.timestamp() as u64 - } -} - -impl ReportIndex for structs::TlsExternalReport { - fn domains(&self) -> impl Iterator { - let report = &self.report; - - report - .policies - .iter() - .flat_map(|p| non_empty(&p.policy_domain).into_iter()) - } - - fn text(&self) -> impl Iterator { - let report = &self.report; - - report - .policies - .iter() - .flat_map(|p| { - non_empty(&p.policy_domain) - .into_iter() - .chain(p.mx_hosts.iter().filter_map(|s| non_empty(s))) - .chain(p.failure_details.iter().flat_map(|fd| { - non_empty_opt(&fd.receiving_mx_hostname) - .into_iter() - .chain(non_empty_opt(&fd.receiving_mx_helo)) - })) - }) - .chain(non_empty(&self.from)) - } - - fn tenant_id(&self) -> Option { - self.member_tenant_id - } - - fn expires_at(&self) -> u64 { - self.expires_at.timestamp() as u64 - } -} impl From for Alignment { fn from(value: enums::DmarcAlignment) -> Self { @@ -938,13 +789,3 @@ fn fo_to_failure_reporting_options(fo: &Option) -> Vec Option<&str> { - if s.is_empty() { None } else { Some(s) } -} - -#[inline(always)] -fn non_empty_opt(s: &Option) -> Option<&str> { - s.as_deref().filter(|s| !s.is_empty()) -} diff --git a/crates/registry/src/utils/task.rs b/crates/registry/src/utils/task.rs index 5d675539..d6f809b8 100644 --- a/crates/registry/src/utils/task.rs +++ b/crates/registry/src/utils/task.rs @@ -18,6 +18,7 @@ impl Task { Task::MergeThreads(task) => task.status = status, Task::DmarcReport(task) => task.status = status, Task::TlsReport(task) => task.status = status, + Task::RestoreArchivedItem(task) => task.status = status, } } @@ -32,6 +33,7 @@ impl Task { Task::MergeThreads(task) => &task.status, Task::DmarcReport(task) => &task.status, Task::TlsReport(task) => &task.status, + Task::RestoreArchivedItem(task) => &task.status, } } diff --git a/crates/services/src/housekeeper/mod.rs b/crates/services/src/housekeeper/mod.rs index a8e6ba1b..2748e707 100644 --- a/crates/services/src/housekeeper/mod.rs +++ b/crates/services/src/housekeeper/mod.rs @@ -10,7 +10,7 @@ use common::{ ipc::{BroadcastEvent, HousekeeperEvent, PurgeType}, }; use email::message::delete::EmailDeletion; -use smtp::reporting::SmtpReporting; +use smtp::reporting::send::MtaReportSend; use spam_filter::modules::classifier::SpamClassifier; use std::{ collections::BinaryHeap, diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 5178acf7..e2938c1c 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -621,7 +621,7 @@ async fn delete_email_metadata( use email::message::metadata::MESSAGE_RECEIVED_MASK; use registry::{ pickle::Pickle, - schema::structs::{DeletedEmail, DeletedItem}, + schema::structs::{ArchivedEmail, ArchivedItem}, types::{datetime::UTCDateTime, id::ObjectId}, }; use store::{ @@ -660,11 +660,11 @@ async fn delete_email_metadata( let until = now + undelete_retention.as_secs(); let blob_hash = BlobHash::from(&metadata.blob_hash); - let item = DeletedItem::Email(DeletedEmail { + let item = ArchivedItem::Email(ArchivedEmail { account_id: account_id.into(), blob_id: BlobId::new(blob_hash.clone(), Default::default()), - cleanup_at: UTCDateTime::from_timestamp(until as i64), - deleted_at: UTCDateTime::now(), + archived_until: UTCDateTime::from_timestamp(until as i64), + archived_at: UTCDateTime::now(), from: from.unwrap_or_default(), received_at: UTCDateTime::from_timestamp( (metadata.rcvd_attach.to_native() & MESSAGE_RECEIVED_MASK) as i64, @@ -673,7 +673,7 @@ async fn delete_email_metadata( size: root_part.offset_end.to_native() as u64, }) .to_pickled_vec(); - let object_id = ObjectType::DeletedItem.to_id(); + let object_id = ObjectType::ArchivedItem.to_id(); let item_id = SnowflakeIdGenerator::from_sequence_id( xxhash_rust::xxh3::xxh3_64(item.as_slice()), ) @@ -685,7 +685,7 @@ async fn delete_email_metadata( hash: blob_hash, to: BlobLink::Temporary { until }, }, - ObjectId::new(ObjectType::DeletedItem, item_id.into()).serialize(), + ObjectId::new(ObjectType::ArchivedItem, item_id.into()).serialize(), ) .set( ValueClass::Registry(RegistryClass::Index { diff --git a/crates/services/src/task_manager/mod.rs b/crates/services/src/task_manager/mod.rs index 26c1c9cb..ebecf638 100644 --- a/crates/services/src/task_manager/mod.rs +++ b/crates/services/src/task_manager/mod.rs @@ -9,6 +9,7 @@ use crate::task_manager::index::SearchIndexTask; use crate::task_manager::lock::TaskLockManager; use crate::task_manager::merge_threads::MergeThreadsTask; use crate::task_manager::report::SubmitReportTask; +use crate::task_manager::restore_item::RestoreItemTask; use alarm::SendAlarmTask; use common::config::server::ServerProtocol; use common::network::limiter::ConcurrencyLimiter; @@ -45,6 +46,7 @@ pub mod index; pub mod lock; pub mod merge_threads; pub mod report; +pub mod restore_item; const QUEUE_REFRESH_INTERVAL: u64 = 60 * 5; // 5 minutes const DEFAULT_LOCK_EXPIRY: u64 = 60 * 5; // 5 minutes @@ -219,6 +221,7 @@ pub fn spawn_task_manager(inner: Arc) { .submit_report(report::ReportId::Tls(task.report_id.id())) .await } + Task::RestoreArchivedItem(task) => server.restore_item(task).await, Task::IndexDocument(_) | Task::UnindexDocument(_) | Task::IndexTrace(_) => unreachable!(), @@ -397,7 +400,7 @@ impl TaskQueueManager for Server { TaskType::MergeThreads => roles .merge_threads .is_enabled_for_integer(task_job.id as u32), - TaskType::DmarcReport | TaskType::TlsReport => true, + TaskType::DmarcReport | TaskType::TlsReport | TaskType::RestoreArchivedItem => true, }; if enabled { @@ -584,6 +587,7 @@ impl TaskInfo for Task { Task::MergeThreads(_) => "MergeThreads", Task::DmarcReport(_) => "DmarcReport", Task::TlsReport(_) => "TlsReport", + Task::RestoreArchivedItem(_) => "RestoreArchivedItem", } } } diff --git a/crates/services/src/task_manager/restore_item.rs b/crates/services/src/task_manager/restore_item.rs new file mode 100644 index 00000000..7680324a --- /dev/null +++ b/crates/services/src/task_manager/restore_item.rs @@ -0,0 +1,105 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use common::{Server, auth::BuildAccessToken}; +use email::{ + mailbox::INBOX_ID, + message::ingest::{EmailIngest, IngestEmail, IngestSource}, +}; +use mail_parser::MessageParser; +use registry::schema::{enums::ArchivedItemType, structs::TaskRestoreArchivedItem}; +use store::write::{BatchBuilder, BlobLink, BlobOp}; +use trc::AddContext; + +use crate::task_manager::TaskResult; + +pub(crate) trait RestoreItemTask: Sync + Send { + fn restore_item( + &self, + task: &TaskRestoreArchivedItem, + ) -> impl Future + Send; +} + +impl RestoreItemTask for Server { + async fn restore_item(&self, task: &TaskRestoreArchivedItem) -> TaskResult { + match restore_item(self, task).await { + Ok(result) => result, + Err(err) => { + let result = TaskResult::temporary(err.to_string()); + trc::error!( + err.account_id(task.account_id.document_id()) + .details("Failed to restore item") + ); + result + } + } + } +} + +async fn restore_item(server: &Server, task: &TaskRestoreArchivedItem) -> trc::Result { + match task.archived_item_type { + ArchivedItemType::Email => { + let account_id = task.account_id.document_id(); + let access_token = server + .access_token(account_id) + .await + .caused_by(trc::location!())?; + + let Some(bytes) = server + .blob_store() + .get_blob(task.blob_id.hash.as_slice(), 0..usize::MAX) + .await? + else { + return Ok(TaskResult::permanent("Blob not found")); + }; + + match server + .email_ingest(IngestEmail { + raw_message: &bytes, + message: MessageParser::new().parse(&bytes), + blob_hash: Some(&task.blob_id.hash), + access_token: &access_token.build(), + mailbox_ids: vec![INBOX_ID], + keywords: vec![], + received_at: (task.created_at.timestamp() as u64).into(), + source: IngestSource::Restore, + session_id: 0, + }) + .await + { + Ok(_) => { + let mut batch = BatchBuilder::new(); + batch.with_account_id(account_id).clear(BlobOp::Link { + hash: task.blob_id.hash.clone(), + to: BlobLink::Temporary { + until: task.archived_until.timestamp() as u64, + }, + }); + server.store().write(batch.build_all()).await?; + + Ok(TaskResult::Success) + } + Err(mut err) + if err.matches(trc::EventType::MessageIngest( + trc::MessageIngestEvent::Error, + )) => + { + Ok(TaskResult::permanent( + err.take_value(trc::Key::Reason) + .and_then(|v| v.into_string()) + .unwrap() + .to_string(), + )) + } + Err(err) => Err(err.caused_by(trc::location!())), + } + } + ArchivedItemType::FileNode + | ArchivedItemType::CalendarEvent + | ArchivedItemType::ContactCard + | ArchivedItemType::SieveScript => Ok(TaskResult::permanent("Not implemented")), + } +} diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index aa79c4ab..7128d8b4 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -20,7 +20,7 @@ use crate::queue::throttle::IsAllowed; use crate::queue::{ Error, FROM_REPORT, HostResponse, MessageWrapper, QueueEnvelope, QueuedMessage, Status, }; -use crate::reporting::SmtpReporting; +use crate::reporting::send::MtaReportSend; use crate::{queue::ErrorDetails, reporting::tls::TlsRptOptions}; use ahash::AHashMap; use common::Server; diff --git a/crates/smtp/src/outbound/local.rs b/crates/smtp/src/outbound/local.rs index dfc96ad7..598cf160 100644 --- a/crates/smtp/src/outbound/local.rs +++ b/crates/smtp/src/outbound/local.rs @@ -11,7 +11,7 @@ use crate::{ MessageSource, MessageWrapper, RCPT_SPAM_PAYLOAD, Status, UnexpectedResponse, quota::HasQueueQuota, spool::SmtpSpool, }, - reporting::SmtpReporting, + reporting::send::MtaReportSend, }; use common::Server; use email::message::delivery::{IngestMessage, IngestRecipient, LocalDeliveryStatus, MailDelivery}; diff --git a/crates/smtp/src/queue/dsn.rs b/crates/smtp/src/queue/dsn.rs index 8985e4a0..a2375367 100644 --- a/crates/smtp/src/queue/dsn.rs +++ b/crates/smtp/src/queue/dsn.rs @@ -10,7 +10,7 @@ use super::{ Recipient, Status, }; use crate::queue::{MessageWrapper, UnexpectedResponse}; -use crate::reporting::SmtpReporting; +use crate::reporting::send::MtaReportSend; use common::Server; use mail_builder::MessageBuilder; use mail_builder::headers::HeaderType; diff --git a/crates/smtp/src/reporting/analysis.rs b/crates/smtp/src/reporting/analysis.rs index 0ffd4c33..6428d4ef 100644 --- a/crates/smtp/src/reporting/analysis.rs +++ b/crates/smtp/src/reporting/analysis.rs @@ -4,32 +4,28 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use ahash::{AHashMap, AHashSet}; +use ahash::AHashSet; use common::{Server, psl}; use mail_auth::{ flate2::read::GzDecoder, - report::{ActionDisposition, DmarcResult, Feedback, Report, tlsrpt::TlsReport}, + report::{Feedback, Report, tlsrpt::TlsReport}, zip, }; use mail_parser::{Message, MimeHeaders, PartType}; use registry::{ - pickle::Pickle, - schema::{ - prelude::ObjectType, - structs::{ArfExternalReport, DmarcExternalReport, TlsExternalReport}, - }, - types::{EnumImpl, datetime::UTCDateTime, index::IndexBuilder}, - utils::report::ReportIndex, + schema::structs::{ArfExternalReport, DmarcExternalReport, TlsExternalReport}, + types::datetime::UTCDateTime, }; use std::{ borrow::Cow, - collections::hash_map::Entry, io::{Cursor, Read}, }; -use store::write::{BatchBuilder, RegistryClass, ValueClass, now}; +use store::write::{BatchBuilder, now}; use trc::IncomingReportEvent; use types::id::Id; +use crate::reporting::{inbound::LogReport, index::ExternalReportIndex}; + enum Compression { None, Gzip, @@ -276,7 +272,6 @@ impl AnalyzeReport for Server { match report { Format::Dmarc(report) => { - let object_id = ObjectType::DmarcExternalReport.to_id(); let mut report = DmarcExternalReport { from, to, @@ -294,20 +289,9 @@ impl AnalyzeReport for Server { .collect::>(), ) .await; - let mut index_builder = IndexBuilder::default(); - report.build_search_index(&mut index_builder); - batch - .registry_index(object_id, item_id, index_builder.keys.iter(), true) - .set( - ValueClass::Registry(RegistryClass::Item { - object_id, - item_id, - }), - report.to_pickled_vec(), - ); + report.write_ops(&mut batch, item_id, true); } Format::Tls(report) => { - let object_id = ObjectType::TlsExternalReport.to_id(); let mut report = TlsExternalReport { from, to, @@ -325,20 +309,9 @@ impl AnalyzeReport for Server { .collect::>(), ) .await; - let mut index_builder = IndexBuilder::default(); - report.build_search_index(&mut index_builder); - batch - .registry_index(object_id, item_id, index_builder.keys.iter(), true) - .set( - ValueClass::Registry(RegistryClass::Item { - object_id, - item_id, - }), - report.to_pickled_vec(), - ); + report.write_ops(&mut batch, item_id, true); } Format::Arf(report) => { - let object_id = ObjectType::ArfExternalReport.to_id(); let mut report = ArfExternalReport { from, to, @@ -356,17 +329,7 @@ impl AnalyzeReport for Server { .collect::>(), ) .await; - let mut index_builder = IndexBuilder::default(); - report.build_search_index(&mut index_builder); - batch - .registry_index(object_id, item_id, index_builder.keys.iter(), true) - .set( - ValueClass::Registry(RegistryClass::Item { - object_id, - item_id, - }), - report.to_pickled_vec(), - ); + report.write_ops(&mut batch, item_id, true); } } @@ -411,165 +374,3 @@ async fn tenant_ids(server: &Server, domains: AHashSet<&str>) -> Option { None } } - -trait LogReport { - fn log(&self); -} - -impl LogReport for Report { - fn log(&self) { - let mut dmarc_pass = 0; - let mut dmarc_quarantine = 0; - let mut dmarc_reject = 0; - let mut dmarc_none = 0; - let mut dkim_pass = 0; - let mut dkim_fail = 0; - let mut dkim_none = 0; - let mut spf_pass = 0; - let mut spf_fail = 0; - let mut spf_none = 0; - - for record in self.records() { - let count = std::cmp::min(record.count(), 1); - - match record.action_disposition() { - ActionDisposition::Pass => { - dmarc_pass += count; - } - ActionDisposition::Quarantine => { - dmarc_quarantine += count; - } - ActionDisposition::Reject => { - dmarc_reject += count; - } - ActionDisposition::None | ActionDisposition::Unspecified => { - dmarc_none += count; - } - } - match record.dmarc_dkim_result() { - DmarcResult::Pass => { - dkim_pass += count; - } - DmarcResult::Fail => { - dkim_fail += count; - } - DmarcResult::Unspecified => { - dkim_none += count; - } - } - match record.dmarc_spf_result() { - DmarcResult::Pass => { - spf_pass += count; - } - DmarcResult::Fail => { - spf_fail += count; - } - DmarcResult::Unspecified => { - spf_none += count; - } - } - } - - trc::event!( - IncomingReport( - if (dmarc_reject + dmarc_quarantine + dkim_fail + spf_fail) > 0 { - IncomingReportEvent::DmarcReportWithWarnings - } else { - IncomingReportEvent::DmarcReport - } - ), - RangeFrom = trc::Value::Timestamp(self.date_range_begin()), - RangeTo = trc::Value::Timestamp(self.date_range_end()), - Domain = self.domain().to_string(), - From = self.email().to_string(), - Id = self.report_id().to_string(), - DmarcPass = dmarc_pass, - DmarcQuarantine = dmarc_quarantine, - DmarcReject = dmarc_reject, - DmarcNone = dmarc_none, - DkimPass = dkim_pass, - DkimFail = dkim_fail, - DkimNone = dkim_none, - SpfPass = spf_pass, - SpfFail = spf_fail, - SpfNone = spf_none, - ); - } -} - -impl LogReport for TlsReport { - fn log(&self) { - for policy in self.policies.iter().take(5) { - let mut details = AHashMap::with_capacity(policy.failure_details.len()); - for failure in &policy.failure_details { - let num_failures = std::cmp::min(1, failure.failed_session_count); - match details.entry(failure.result_type) { - Entry::Occupied(mut e) => { - *e.get_mut() += num_failures; - } - Entry::Vacant(e) => { - e.insert(num_failures); - } - } - } - - trc::event!( - IncomingReport(if policy.summary.total_failure > 0 { - IncomingReportEvent::TlsReportWithWarnings - } else { - IncomingReportEvent::TlsReport - }), - RangeFrom = - trc::Value::Timestamp(self.date_range.start_datetime.to_timestamp() as u64), - RangeTo = trc::Value::Timestamp(self.date_range.end_datetime.to_timestamp() as u64), - Domain = policy.policy.policy_domain.clone(), - From = self.contact_info.as_deref().unwrap_or_default().to_string(), - Id = self.report_id.clone(), - Policy = format!("{:?}", policy.policy.policy_type), - TotalSuccesses = policy.summary.total_success, - TotalFailures = policy.summary.total_failure, - Details = format!("{details:?}"), - ); - } - } -} - -impl LogReport for Feedback<'_> { - fn log(&self) { - trc::event!( - IncomingReport(match self.feedback_type() { - mail_auth::report::FeedbackType::Abuse => IncomingReportEvent::AbuseReport, - mail_auth::report::FeedbackType::AuthFailure => - IncomingReportEvent::AuthFailureReport, - mail_auth::report::FeedbackType::Fraud => IncomingReportEvent::FraudReport, - mail_auth::report::FeedbackType::NotSpam => IncomingReportEvent::NotSpamReport, - mail_auth::report::FeedbackType::Other => IncomingReportEvent::OtherReport, - mail_auth::report::FeedbackType::Virus => IncomingReportEvent::VirusReport, - }), - RangeFrom = trc::Value::Timestamp( - self.arrival_date() - .map(|d| d as u64) - .unwrap_or_else(|| { now() }) - ), - Domain = self - .reported_domain() - .iter() - .map(|d| trc::Value::String(d.as_ref().into())) - .collect::>(), - Hostname = self.reporting_mta().map(|d| trc::Value::String(d.into())), - Url = self - .reported_uri() - .iter() - .map(|d| trc::Value::String(d.as_ref().into())) - .collect::>(), - RemoteIp = self.source_ip(), - Total = self.incidents(), - Result = format!("{:?}", self.delivery_result()), - Details = self - .authentication_results() - .iter() - .map(|d| trc::Value::String(d.as_ref().into())) - .collect::>(), - ); - } -} diff --git a/crates/smtp/src/reporting/dkim.rs b/crates/smtp/src/reporting/dkim.rs index 620eb9ec..920cff10 100644 --- a/crates/smtp/src/reporting/dkim.rs +++ b/crates/smtp/src/reporting/dkim.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{core::Session, reporting::SmtpReporting}; +use crate::{core::Session, reporting::send::MtaReportSend}; use common::network::SessionStream; use mail_auth::{ AuthenticatedMessage, AuthenticationResults, DkimOutput, common::verify::VerifySignature, diff --git a/crates/smtp/src/reporting/dmarc.rs b/crates/smtp/src/reporting/dmarc.rs index 312040fe..e0bdc392 100644 --- a/crates/smtp/src/reporting/dmarc.rs +++ b/crates/smtp/src/reporting/dmarc.rs @@ -5,7 +5,11 @@ */ use super::AggregateTimestamp; -use crate::{core::Session, queue::RecipientDomain, reporting::SmtpReporting}; +use crate::{ + core::Session, + queue::RecipientDomain, + reporting::{index::InternalReportIndex, send::MtaReportSend}, +}; use common::{ Server, config::smtp::report::AggregateFrequency, @@ -25,12 +29,9 @@ use registry::{ schema::{ enums::FailureReportingOption, prelude::{ObjectType, Property}, - structs::{ - DmarcInternalReport, DmarcReport, DmarcReportRecord, Rate, Task, TaskDmarcReport, - TaskStatus, - }, + structs::{DmarcInternalReport, DmarcReport, DmarcReportRecord, Rate}, }, - types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, + types::{EnumImpl, datetime::UTCDateTime}, }; use std::future::Future; use store::{ @@ -337,9 +338,7 @@ impl DmarcReporting for Server { .write(report.policy_identifier) .finalize(), }); - self.core - .storage - .data + self.store() .write(batch.build_all()) .await .caused_by(trc::location!())?; @@ -523,95 +522,76 @@ impl DmarcReporting for Server { (object_id_v.object_id.id().id(), report) } else { let item_id = self.inner.data.queue_id_gen.generate(); - let deliver_at = (event.interval.to_timestamp() + event.interval.as_secs()) as i64; - - batch - .assert_value(pk.clone(), ()) - .set( - pk.clone(), - ObjectIdVersioned { - object_id: ObjectId::new( - ObjectType::DmarcInternalReport, - item_id.into(), - ), - version: 0, - } - .serialize(), - ) - .schedule_task_with_id( - item_id, - Task::DmarcReport(TaskDmarcReport { - report_id: item_id.into(), - status: TaskStatus::at(deliver_at), - }), - ); - let created_at = UTCDateTime::now(); - let deliver_at = UTCDateTime::from_timestamp(deliver_at); + let deliver_at = UTCDateTime::from_timestamp( + (event.interval.to_timestamp() + event.interval.as_secs()) as i64, + ); let policy = PolicyPublished::from_record(event.domain.clone(), &event.dmarc_record); - ( - item_id, - DmarcInternalReport { - created_at, - deliver_at, - domain: event.domain.clone(), - report: DmarcReport { - report_id: format!("{}_{policy_hash}", created_at.timestamp()), - date_range_begin: created_at, - date_range_end: deliver_at, - email: self - .eval_if( - &config.address, - &RecipientDomain::new(event.domain.as_str()), - event.span_id, - ) - .await - .unwrap_or_else(|| "MAILER-DAEMON@localhost".to_string()), - extra_contact_info: self - .eval_if::( - &config.contact_info, - &RecipientDomain::new(event.domain.as_str()), - event.span_id, - ) - .await, - org_name: self - .eval_if::( - &config.org_name, - &RecipientDomain::new(event.domain.as_str()), - event.span_id, - ) - .await - .unwrap_or_default(), - policy_adkim: policy.adkim.into(), - policy_aspf: policy.aspf.into(), - policy_disposition: policy.p.into(), - policy_domain: policy.domain, - policy_failure_reporting_options: match event.dmarc_record.fo { - dmarc::Report::All => vec![FailureReportingOption::All], - dmarc::Report::Any => vec![FailureReportingOption::Any], - dmarc::Report::Dkim => vec![FailureReportingOption::DkimFailure], - dmarc::Report::Spf => vec![FailureReportingOption::SpfFailure], - dmarc::Report::DkimSpf => vec![ - FailureReportingOption::DkimFailure, - FailureReportingOption::SpfFailure, - ], - }, - policy_subdomain_disposition: policy.sp.into(), - policy_testing_mode: policy.testing, - policy_version: None, - version: 1.0.into(), - ..Default::default() + + let report = DmarcInternalReport { + created_at, + deliver_at, + domain: event.domain.clone(), + report: DmarcReport { + report_id: format!("{}_{policy_hash}", created_at.timestamp()), + date_range_begin: created_at, + date_range_end: deliver_at, + email: self + .eval_if( + &config.address, + &RecipientDomain::new(event.domain.as_str()), + event.span_id, + ) + .await + .unwrap_or_else(|| "MAILER-DAEMON@localhost".to_string()), + extra_contact_info: self + .eval_if::( + &config.contact_info, + &RecipientDomain::new(event.domain.as_str()), + event.span_id, + ) + .await, + org_name: self + .eval_if::( + &config.org_name, + &RecipientDomain::new(event.domain.as_str()), + event.span_id, + ) + .await + .unwrap_or_default(), + policy_adkim: policy.adkim.into(), + policy_aspf: policy.aspf.into(), + policy_disposition: policy.p.into(), + policy_domain: policy.domain, + policy_failure_reporting_options: match event.dmarc_record.fo { + dmarc::Report::All => vec![FailureReportingOption::All], + dmarc::Report::Any => vec![FailureReportingOption::Any], + dmarc::Report::Dkim => vec![FailureReportingOption::DkimFailure], + dmarc::Report::Spf => vec![FailureReportingOption::SpfFailure], + dmarc::Report::DkimSpf => vec![ + FailureReportingOption::DkimFailure, + FailureReportingOption::SpfFailure, + ], }, - policy_identifier: policy_hash, - rua: event - .dmarc_record - .rua() - .iter() - .map(|u| u.uri.clone()) - .collect(), + policy_subdomain_disposition: policy.sp.into(), + policy_testing_mode: policy.testing, + policy_version: None, + version: 1.0.into(), + ..Default::default() }, - ) + policy_identifier: policy_hash, + rua: event + .dmarc_record + .rua() + .iter() + .map(|u| u.uri.clone()) + .collect(), + }; + + report.write_ops(&mut batch, item_id, true); + + (item_id, report) }; // Add record diff --git a/crates/smtp/src/reporting/inbound.rs b/crates/smtp/src/reporting/inbound.rs new file mode 100644 index 00000000..98af3361 --- /dev/null +++ b/crates/smtp/src/reporting/inbound.rs @@ -0,0 +1,218 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::core::Session; +use ahash::AHashMap; +use common::{USER_AGENT, config::smtp::report::AddressMatch}; +use mail_auth::report::{ + ActionDisposition, AuthFailureType, DeliveryResult, DmarcResult, Feedback, FeedbackType, + Report, tlsrpt::TlsReport, +}; +use std::{collections::hash_map::Entry, time::SystemTime}; +use store::write::now; +use tokio::io::{AsyncRead, AsyncWrite}; +use trc::IncomingReportEvent; + +impl Session { + pub fn new_auth_failure(&self, ft: AuthFailureType, rejected: bool) -> Feedback<'_> { + Feedback::new(FeedbackType::AuthFailure) + .with_auth_failure(ft) + .with_arrival_date( + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()) as i64, + ) + .with_source_ip(self.data.remote_ip) + .with_reporting_mta(&self.hostname) + .with_user_agent(USER_AGENT) + .with_delivery_result(if rejected { + DeliveryResult::Reject + } else { + DeliveryResult::Unspecified + }) + } + + pub fn is_report(&self) -> bool { + for addr_match in &self.server.core.smtp.report.analysis.addresses { + for addr in &self.data.rcpt_to { + match addr_match { + AddressMatch::StartsWith(prefix) if addr.address_lcase.starts_with(prefix) => { + return true; + } + AddressMatch::EndsWith(suffix) if addr.address_lcase.ends_with(suffix) => { + return true; + } + AddressMatch::Equals(value) if addr.address_lcase.eq(value) => return true, + _ => (), + } + } + } + + false + } +} + +pub(crate) trait LogReport { + fn log(&self); +} + +impl LogReport for Report { + fn log(&self) { + let mut dmarc_pass = 0; + let mut dmarc_quarantine = 0; + let mut dmarc_reject = 0; + let mut dmarc_none = 0; + let mut dkim_pass = 0; + let mut dkim_fail = 0; + let mut dkim_none = 0; + let mut spf_pass = 0; + let mut spf_fail = 0; + let mut spf_none = 0; + + for record in self.records() { + let count = std::cmp::min(record.count(), 1); + + match record.action_disposition() { + ActionDisposition::Pass => { + dmarc_pass += count; + } + ActionDisposition::Quarantine => { + dmarc_quarantine += count; + } + ActionDisposition::Reject => { + dmarc_reject += count; + } + ActionDisposition::None | ActionDisposition::Unspecified => { + dmarc_none += count; + } + } + match record.dmarc_dkim_result() { + DmarcResult::Pass => { + dkim_pass += count; + } + DmarcResult::Fail => { + dkim_fail += count; + } + DmarcResult::Unspecified => { + dkim_none += count; + } + } + match record.dmarc_spf_result() { + DmarcResult::Pass => { + spf_pass += count; + } + DmarcResult::Fail => { + spf_fail += count; + } + DmarcResult::Unspecified => { + spf_none += count; + } + } + } + + trc::event!( + IncomingReport( + if (dmarc_reject + dmarc_quarantine + dkim_fail + spf_fail) > 0 { + IncomingReportEvent::DmarcReportWithWarnings + } else { + IncomingReportEvent::DmarcReport + } + ), + RangeFrom = trc::Value::Timestamp(self.date_range_begin()), + RangeTo = trc::Value::Timestamp(self.date_range_end()), + Domain = self.domain().to_string(), + From = self.email().to_string(), + Id = self.report_id().to_string(), + DmarcPass = dmarc_pass, + DmarcQuarantine = dmarc_quarantine, + DmarcReject = dmarc_reject, + DmarcNone = dmarc_none, + DkimPass = dkim_pass, + DkimFail = dkim_fail, + DkimNone = dkim_none, + SpfPass = spf_pass, + SpfFail = spf_fail, + SpfNone = spf_none, + ); + } +} + +impl LogReport for TlsReport { + fn log(&self) { + for policy in self.policies.iter().take(5) { + let mut details = AHashMap::with_capacity(policy.failure_details.len()); + for failure in &policy.failure_details { + let num_failures = std::cmp::min(1, failure.failed_session_count); + match details.entry(failure.result_type) { + Entry::Occupied(mut e) => { + *e.get_mut() += num_failures; + } + Entry::Vacant(e) => { + e.insert(num_failures); + } + } + } + + trc::event!( + IncomingReport(if policy.summary.total_failure > 0 { + IncomingReportEvent::TlsReportWithWarnings + } else { + IncomingReportEvent::TlsReport + }), + RangeFrom = + trc::Value::Timestamp(self.date_range.start_datetime.to_timestamp() as u64), + RangeTo = trc::Value::Timestamp(self.date_range.end_datetime.to_timestamp() as u64), + Domain = policy.policy.policy_domain.clone(), + From = self.contact_info.as_deref().unwrap_or_default().to_string(), + Id = self.report_id.clone(), + Policy = format!("{:?}", policy.policy.policy_type), + TotalSuccesses = policy.summary.total_success, + TotalFailures = policy.summary.total_failure, + Details = format!("{details:?}"), + ); + } + } +} + +impl LogReport for Feedback<'_> { + fn log(&self) { + trc::event!( + IncomingReport(match self.feedback_type() { + mail_auth::report::FeedbackType::Abuse => IncomingReportEvent::AbuseReport, + mail_auth::report::FeedbackType::AuthFailure => + IncomingReportEvent::AuthFailureReport, + mail_auth::report::FeedbackType::Fraud => IncomingReportEvent::FraudReport, + mail_auth::report::FeedbackType::NotSpam => IncomingReportEvent::NotSpamReport, + mail_auth::report::FeedbackType::Other => IncomingReportEvent::OtherReport, + mail_auth::report::FeedbackType::Virus => IncomingReportEvent::VirusReport, + }), + RangeFrom = trc::Value::Timestamp( + self.arrival_date() + .map(|d| d as u64) + .unwrap_or_else(|| { now() }) + ), + Domain = self + .reported_domain() + .iter() + .map(|d| trc::Value::String(d.as_ref().into())) + .collect::>(), + Hostname = self.reporting_mta().map(|d| trc::Value::String(d.into())), + Url = self + .reported_uri() + .iter() + .map(|d| trc::Value::String(d.as_ref().into())) + .collect::>(), + RemoteIp = self.source_ip(), + Total = self.incidents(), + Result = format!("{:?}", self.delivery_result()), + Details = self + .authentication_results() + .iter() + .map(|d| trc::Value::String(d.as_ref().into())) + .collect::>(), + ); + } +} diff --git a/crates/smtp/src/reporting/index.rs b/crates/smtp/src/reporting/index.rs new file mode 100644 index 00000000..f35513b6 --- /dev/null +++ b/crates/smtp/src/reporting/index.rs @@ -0,0 +1,317 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use registry::{ + schema::{ + prelude::{ObjectType, Property}, + structs::{ + ArfExternalReport, DmarcExternalReport, DmarcInternalReport, Task, TaskDmarcReport, + TaskStatus, TaskTlsReport, TlsExternalReport, TlsInternalReport, + }, + }, + types::{EnumImpl, ObjectImpl, datetime::UTCDateTime, id::ObjectId, index::IndexBuilder}, +}; +use store::{ + SerializeInfallible, U64_LEN, + registry::ObjectIdVersioned, + write::{ + BatchBuilder, RegistryClass, TaskQueueClass, ValueClass, assert::AssertValue, + key::KeySerializer, + }, +}; +use types::id::Id; + +pub trait InternalReportIndex: ObjectImpl { + fn deliver_at(&self) -> UTCDateTime; + + fn set_deliver_at(&mut self, at: UTCDateTime); + + fn task(&self, item_id: u64) -> Task; + + fn primary_key(&self) -> ValueClass; + + fn reschedule_ops( + &mut self, + batch: &mut BatchBuilder, + item_id: u64, + revision: u64, + at: UTCDateTime, + ) { + let current_deliver_at = self.deliver_at(); + + if current_deliver_at != at { + let object = Self::OBJECT; + let object_id = object.to_id(); + let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); + + self.set_deliver_at(at); + + batch + .assert_value(key.clone(), AssertValue::Hash(revision)) + .clear(ValueClass::TaskQueue(TaskQueueClass::Due { + id: item_id, + due: current_deliver_at.timestamp() as u64, + })) + .set( + ValueClass::TaskQueue(TaskQueueClass::Due { + id: item_id, + due: at.timestamp() as u64, + }), + object_id.serialize(), + ) + .set(key, self.to_pickled_vec()); + } + } + + fn write_ops(&self, batch: &mut BatchBuilder, item_id: u64, is_set: bool) { + let object = Self::OBJECT; + let object_id = object.to_id(); + let pk = self.primary_key(); + + if is_set { + batch + .assert_value(pk.clone(), ()) + .set( + pk, + ObjectIdVersioned { + object_id: ObjectId::new(object, item_id.into()), + version: 0, + } + .serialize(), + ) + .schedule_task_with_id(item_id, self.task(item_id)); + } else { + batch + .clear(ValueClass::Registry(RegistryClass::Item { + object_id, + item_id, + })) + .clear(pk) + .clear(ValueClass::TaskQueue(TaskQueueClass::Task { id: item_id })) + .clear(ValueClass::TaskQueue(TaskQueueClass::Due { + id: item_id, + due: self.deliver_at().timestamp() as u64, + })); + } + } +} + +pub trait ExternalReportIndex: ObjectImpl { + fn text(&self) -> impl Iterator; + + fn tenant_id(&self) -> Option; + + fn expires_at(&self) -> u64; + + fn domains(&self) -> impl Iterator; + + fn write_ops(&self, batch: &mut BatchBuilder, item_id: u64, is_set: bool) { + let object_id = Self::OBJECT.to_id(); + let mut index_builder = IndexBuilder::default(); + for text in self.text() { + index_builder.text(Property::Domain, text); + } + + if let Some(tenant_id) = self.tenant_id() { + index_builder.search(Property::MemberTenantId, tenant_id.id()); + } + + index_builder.search(Property::ExpiresAt, self.expires_at()); + batch.registry_index(object_id, item_id, index_builder.keys.iter(), is_set); + + let key = ValueClass::Registry(RegistryClass::Item { object_id, item_id }); + if is_set { + batch.set(key, self.to_pickled_vec()); + } else { + batch.clear(key); + } + } +} + +impl InternalReportIndex for DmarcInternalReport { + fn deliver_at(&self) -> UTCDateTime { + self.deliver_at + } + + fn set_deliver_at(&mut self, at: UTCDateTime) { + self.deliver_at = at; + } + + fn task(&self, item_id: u64) -> Task { + Task::DmarcReport(TaskDmarcReport { + report_id: item_id.into(), + status: TaskStatus::at(self.deliver_at.timestamp()), + }) + } + + fn primary_key(&self) -> ValueClass { + ValueClass::Registry(RegistryClass::PrimaryKey { + object_id: ObjectType::DmarcInternalReport.to_id().into(), + index_id: Property::Domain.to_id(), + key: KeySerializer::new(self.domain.len() + U64_LEN) + .write(self.domain.as_str()) + .write(self.policy_identifier) + .finalize(), + }) + } +} + +impl InternalReportIndex for TlsInternalReport { + fn deliver_at(&self) -> UTCDateTime { + self.deliver_at + } + + fn set_deliver_at(&mut self, at: UTCDateTime) { + self.deliver_at = at; + } + + fn task(&self, item_id: u64) -> Task { + Task::TlsReport(TaskTlsReport { + report_id: item_id.into(), + status: TaskStatus::at(self.deliver_at.timestamp()), + }) + } + + fn primary_key(&self) -> ValueClass { + ValueClass::Registry(RegistryClass::PrimaryKey { + object_id: ObjectType::TlsInternalReport.to_id().into(), + index_id: Property::Domain.to_id(), + key: self.domain.as_bytes().to_vec(), + }) + } +} + +impl ExternalReportIndex for ArfExternalReport { + fn domains(&self) -> impl Iterator { + let report = &self.report; + + report + .reported_domains + .iter() + .filter_map(|s| non_empty(s)) + .chain( + [report.dkim_domain.as_deref()] + .into_iter() + .flatten() + .filter_map(non_empty), + ) + } + + fn text(&self) -> impl Iterator { + let report = &self.report; + + report + .reported_domains + .iter() + .filter_map(|s| non_empty(s)) + .chain( + [ + report.dkim_domain.as_deref(), + report.reporting_mta.as_deref(), + report.original_mail_from.as_deref(), + report.original_rcpt_to.as_deref(), + ] + .into_iter() + .flatten() + .filter_map(non_empty), + ) + .chain(non_empty(&self.from)) + } + + fn tenant_id(&self) -> Option { + self.member_tenant_id + } + + fn expires_at(&self) -> u64 { + self.expires_at.timestamp() as u64 + } +} + +impl ExternalReportIndex for DmarcExternalReport { + fn domains(&self) -> impl Iterator { + let report = &self.report; + + non_empty(&report.policy_domain) + .into_iter() + .filter_map(non_empty) + } + + fn text(&self) -> impl Iterator { + let report = &self.report; + + non_empty(&report.email) + .into_iter() + .filter_map(non_empty) + .chain(non_empty(&report.policy_domain)) + .chain(report.records.iter().flat_map(|r| { + r.envelope_to + .as_deref() + .into_iter() + .filter_map(non_empty) + .chain(non_empty(&r.envelope_from)) + .chain(non_empty(&r.header_from)) + .chain(r.dkim_results.iter().filter_map(|d| non_empty(&d.domain))) + .chain(r.spf_results.iter().filter_map(|s| non_empty(&s.domain))) + })) + .chain(non_empty(&self.from)) + } + + fn tenant_id(&self) -> Option { + self.member_tenant_id + } + + fn expires_at(&self) -> u64 { + self.expires_at.timestamp() as u64 + } +} + +impl ExternalReportIndex for TlsExternalReport { + fn domains(&self) -> impl Iterator { + let report = &self.report; + + report + .policies + .iter() + .flat_map(|p| non_empty(&p.policy_domain).into_iter()) + } + + fn text(&self) -> impl Iterator { + let report = &self.report; + + report + .policies + .iter() + .flat_map(|p| { + non_empty(&p.policy_domain) + .into_iter() + .chain(p.mx_hosts.iter().filter_map(|s| non_empty(s))) + .chain(p.failure_details.iter().flat_map(|fd| { + non_empty_opt(&fd.receiving_mx_hostname) + .into_iter() + .chain(non_empty_opt(&fd.receiving_mx_helo)) + })) + }) + .chain(non_empty(&self.from)) + } + + fn tenant_id(&self) -> Option { + self.member_tenant_id + } + + fn expires_at(&self) -> u64 { + self.expires_at.timestamp() as u64 + } +} + +#[inline(always)] +fn non_empty(s: &str) -> Option<&str> { + if s.is_empty() { None } else { Some(s) } +} + +#[inline(always)] +fn non_empty_opt(s: &Option) -> Option<&str> { + s.as_deref().filter(|s| !s.is_empty()) +} diff --git a/crates/smtp/src/reporting/mod.rs b/crates/smtp/src/reporting/mod.rs index 4241719c..add57eb4 100644 --- a/crates/smtp/src/reporting/mod.rs +++ b/crates/smtp/src/reporting/mod.rs @@ -4,241 +4,20 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ - core::Session, - inbound::DkimSign, - queue::{MessageSource, MessageWrapper, spool::SmtpSpool}, -}; -use common::{ - Server, USER_AGENT, - config::smtp::report::{AddressMatch, AggregateFrequency}, - expr::if_block::IfBlock, - ipc::ReportingEvent, -}; -use mail_auth::{ - common::headers::HeaderWriter, - report::{AuthFailureType, DeliveryResult, Feedback, FeedbackType}, -}; +use common::config::smtp::report::AggregateFrequency; use mail_parser::DateTime; -use std::{future::Future, io, time::SystemTime}; -use tokio::io::{AsyncRead, AsyncWrite}; +use std::time::SystemTime; pub mod analysis; pub mod dkim; pub mod dmarc; +pub mod inbound; +pub mod index; pub mod scheduler; +pub mod send; pub mod spf; pub mod tls; -impl Session { - pub fn new_auth_failure(&self, ft: AuthFailureType, rejected: bool) -> Feedback<'_> { - Feedback::new(FeedbackType::AuthFailure) - .with_auth_failure(ft) - .with_arrival_date( - SystemTime::now() - .duration_since(SystemTime::UNIX_EPOCH) - .map_or(0, |d| d.as_secs()) as i64, - ) - .with_source_ip(self.data.remote_ip) - .with_reporting_mta(&self.hostname) - .with_user_agent(USER_AGENT) - .with_delivery_result(if rejected { - DeliveryResult::Reject - } else { - DeliveryResult::Unspecified - }) - } - - pub fn is_report(&self) -> bool { - for addr_match in &self.server.core.smtp.report.analysis.addresses { - for addr in &self.data.rcpt_to { - match addr_match { - AddressMatch::StartsWith(prefix) if addr.address_lcase.starts_with(prefix) => { - return true; - } - AddressMatch::EndsWith(suffix) if addr.address_lcase.ends_with(suffix) => { - return true; - } - AddressMatch::Equals(value) if addr.address_lcase.eq(value) => return true, - _ => (), - } - } - } - - false - } -} - -pub trait SmtpReporting: Sync + Send { - fn send_report( - &self, - from_addr: &str, - rcpts: impl Iterator + Sync + Send> + Sync + Send, - report: Vec, - sign_config: &IfBlock, - deliver_now: bool, - parent_session_id: u64, - ) -> impl Future + Send; - - fn send_autogenerated( - &self, - from_addr: impl AsRef + Sync + Send, - rcpts: impl Iterator + Sync + Send> + Sync + Send, - raw_message: Vec, - sign_config: Option<&IfBlock>, - parent_session_id: u64, - ) -> impl Future + Send; - - fn schedule_report( - &self, - report: impl Into + Sync + Send, - ) -> impl Future + Send; - - fn sign_message( - &self, - message: &mut MessageWrapper, - config: &IfBlock, - bytes: &[u8], - ) -> impl Future>> + Send; -} - -impl SmtpReporting for Server { - async fn send_report( - &self, - from_addr: &str, - rcpts: impl Iterator + Sync + Send> + Sync + Send, - report: Vec, - sign_config: &IfBlock, - deliver_now: bool, - parent_session_id: u64, - ) { - // Build message - let mut message = self.new_message(from_addr, parent_session_id); - for rcpt_ in rcpts { - message.add_recipient(rcpt_.as_ref(), self).await; - } - - // Sign message - let signature = self.sign_message(&mut message, sign_config, &report).await; - - // Schedule delivery at a random time between now and the next 3 hours - if !deliver_now { - #[cfg(not(feature = "test_mode"))] - { - use common::config::smtp::queue::QueueExpiry; - use rand::Rng; - - let delivery_time = rand::rng().random_range(0u64..10800u64); - for rcpt in &mut message.message.recipients { - rcpt.retry.due += delivery_time; - rcpt.notify.due += delivery_time; - if let QueueExpiry::Ttl(expires) = &mut rcpt.expires { - *expires += delivery_time; - } - } - } - } - - // Queue message - message - .queue( - signature.as_deref(), - &report, - parent_session_id, - self, - MessageSource::Report, - ) - .await; - } - - async fn send_autogenerated( - &self, - from_addr: impl AsRef + Sync + Send, - rcpts: impl Iterator + Sync + Send> + Sync + Send, - raw_message: Vec, - sign_config: Option<&IfBlock>, - parent_session_id: u64, - ) { - // Build message - let mut message = self.new_message(from_addr.as_ref(), parent_session_id); - for rcpt in rcpts { - message.add_recipient(rcpt, self).await; - } - - // Sign message - let signature = if let Some(sign_config) = sign_config { - self.sign_message(&mut message, sign_config, &raw_message) - .await - } else { - None - }; - - // Queue message - message - .queue( - signature.as_deref(), - &raw_message, - parent_session_id, - self, - MessageSource::Autogenerated, - ) - .await; - } - - async fn schedule_report(&self, report: impl Into + Sync + Send) { - if self.inner.ipc.report_tx.send(report.into()).await.is_err() { - trc::event!( - Server(trc::ServerEvent::ThreadError), - CausedBy = trc::location!(), - Details = "Failed to send event to ReportScheduler" - ); - } - } - - async fn sign_message( - &self, - message: &mut MessageWrapper, - config: &IfBlock, - bytes: &[u8], - ) -> Option> { - let sign_with_domain = self - .eval_if::(config, &message.message, message.span_id) - .await?; - - match self.dkim_signers(&sign_with_domain).await { - Ok(Some(signers)) => { - let mut headers = Vec::with_capacity(64); - - for signer in signers.as_ref() { - match signer.sign(bytes) { - Ok(signature) => { - signature.write_header(&mut headers); - } - Err(err) => { - trc::error!( - trc::Error::from(err) - .span_id(message.span_id) - .details("Failed to sign message") - .caused_by(trc::location!()) - ); - } - } - } - - Some(headers) - } - Ok(None) => None, - Err(err) => { - trc::error!( - err.span_id(message.span_id) - .details("Failed to retrieve DKIM signers") - ); - None - } - } - } -} - pub trait AggregateTimestamp { fn to_timestamp(&self) -> u64; fn to_timestamp_(&self, dt: DateTime) -> u64; @@ -292,30 +71,3 @@ impl AggregateTimestamp for AggregateFrequency { self.to_timestamp() + self.as_secs() } } - -pub struct SerializedSize { - bytes_left: usize, -} - -impl SerializedSize { - pub fn new(bytes_left: usize) -> Self { - Self { bytes_left } - } -} - -impl io::Write for SerializedSize { - fn write(&mut self, buf: &[u8]) -> io::Result { - //let c = print!(" (left: {}, buf: {})", self.bytes_left, buf.len()); - let buf_len = buf.len(); - if buf_len <= self.bytes_left { - self.bytes_left -= buf_len; - Ok(buf_len) - } else { - Err(io::Error::other("Size exceeded")) - } - } - - fn flush(&mut self) -> io::Result<()> { - Ok(()) - } -} diff --git a/crates/smtp/src/reporting/send.rs b/crates/smtp/src/reporting/send.rs new file mode 100644 index 00000000..75a10066 --- /dev/null +++ b/crates/smtp/src/reporting/send.rs @@ -0,0 +1,182 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{ + inbound::DkimSign, + queue::{MessageSource, MessageWrapper, spool::SmtpSpool}, +}; +use common::{Server, expr::if_block::IfBlock, ipc::ReportingEvent}; +use mail_auth::common::headers::HeaderWriter; + +pub trait MtaReportSend: Sync + Send { + fn send_report( + &self, + from_addr: &str, + rcpts: impl Iterator + Sync + Send> + Sync + Send, + report: Vec, + sign_config: &IfBlock, + deliver_now: bool, + parent_session_id: u64, + ) -> impl Future + Send; + + fn send_autogenerated( + &self, + from_addr: impl AsRef + Sync + Send, + rcpts: impl Iterator + Sync + Send> + Sync + Send, + raw_message: Vec, + sign_config: Option<&IfBlock>, + parent_session_id: u64, + ) -> impl Future + Send; + + fn schedule_report( + &self, + report: impl Into + Sync + Send, + ) -> impl Future + Send; + + fn sign_message( + &self, + message: &mut MessageWrapper, + config: &IfBlock, + bytes: &[u8], + ) -> impl Future>> + Send; +} + +impl MtaReportSend for Server { + async fn send_report( + &self, + from_addr: &str, + rcpts: impl Iterator + Sync + Send> + Sync + Send, + report: Vec, + sign_config: &IfBlock, + deliver_now: bool, + parent_session_id: u64, + ) { + // Build message + let mut message = self.new_message(from_addr, parent_session_id); + for rcpt_ in rcpts { + message.add_recipient(rcpt_.as_ref(), self).await; + } + + // Sign message + let signature = self.sign_message(&mut message, sign_config, &report).await; + + // Schedule delivery at a random time between now and the next 3 hours + if !deliver_now { + #[cfg(not(feature = "test_mode"))] + { + use common::config::smtp::queue::QueueExpiry; + use rand::Rng; + + let delivery_time = rand::rng().random_range(0u64..10800u64); + for rcpt in &mut message.message.recipients { + rcpt.retry.due += delivery_time; + rcpt.notify.due += delivery_time; + if let QueueExpiry::Ttl(expires) = &mut rcpt.expires { + *expires += delivery_time; + } + } + } + } + + // Queue message + message + .queue( + signature.as_deref(), + &report, + parent_session_id, + self, + MessageSource::Report, + ) + .await; + } + + async fn send_autogenerated( + &self, + from_addr: impl AsRef + Sync + Send, + rcpts: impl Iterator + Sync + Send> + Sync + Send, + raw_message: Vec, + sign_config: Option<&IfBlock>, + parent_session_id: u64, + ) { + // Build message + let mut message = self.new_message(from_addr.as_ref(), parent_session_id); + for rcpt in rcpts { + message.add_recipient(rcpt, self).await; + } + + // Sign message + let signature = if let Some(sign_config) = sign_config { + self.sign_message(&mut message, sign_config, &raw_message) + .await + } else { + None + }; + + // Queue message + message + .queue( + signature.as_deref(), + &raw_message, + parent_session_id, + self, + MessageSource::Autogenerated, + ) + .await; + } + + async fn schedule_report(&self, report: impl Into + Sync + Send) { + if self.inner.ipc.report_tx.send(report.into()).await.is_err() { + trc::event!( + Server(trc::ServerEvent::ThreadError), + CausedBy = trc::location!(), + Details = "Failed to send event to ReportScheduler" + ); + } + } + + async fn sign_message( + &self, + message: &mut MessageWrapper, + config: &IfBlock, + bytes: &[u8], + ) -> Option> { + let sign_with_domain = self + .eval_if::(config, &message.message, message.span_id) + .await?; + + match self.dkim_signers(&sign_with_domain).await { + Ok(Some(signers)) => { + let mut headers = Vec::with_capacity(64); + + for signer in signers.as_ref() { + match signer.sign(bytes) { + Ok(signature) => { + signature.write_header(&mut headers); + } + Err(err) => { + trc::error!( + trc::Error::from(err) + .span_id(message.span_id) + .details("Failed to sign message") + .caused_by(trc::location!()) + ); + } + } + } + + Some(headers) + } + Ok(None) => None, + Err(err) => { + trc::error!( + err.span_id(message.span_id) + .details("Failed to retrieve DKIM signers") + ); + None + } + } + } +} diff --git a/crates/smtp/src/reporting/spf.rs b/crates/smtp/src/reporting/spf.rs index e2ec9bf9..234a5c8f 100644 --- a/crates/smtp/src/reporting/spf.rs +++ b/crates/smtp/src/reporting/spf.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{core::Session, reporting::SmtpReporting}; +use crate::{core::Session, reporting::send::MtaReportSend}; use common::network::SessionStream; use mail_auth::{AuthenticationResults, SpfOutput, report::AuthFailureType}; use registry::schema::structs::Rate; diff --git a/crates/smtp/src/reporting/tls.rs b/crates/smtp/src/reporting/tls.rs index c03d0dc6..cd8f95fa 100644 --- a/crates/smtp/src/reporting/tls.rs +++ b/crates/smtp/src/reporting/tls.rs @@ -5,7 +5,10 @@ */ use super::AggregateTimestamp; -use crate::{queue::RecipientDomain, reporting::SmtpReporting}; +use crate::{ + queue::RecipientDomain, + reporting::{index::InternalReportIndex, send::MtaReportSend}, +}; use common::{ Server, USER_AGENT, config::smtp::{ @@ -24,12 +27,9 @@ use registry::{ schema::{ enums::TlsPolicyType, prelude::{ObjectType, Property}, - structs::{ - Task, TaskStatus, TaskTlsReport, TlsFailureDetails, TlsInternalReport, TlsReport, - TlsReportPolicy, - }, + structs::{TlsFailureDetails, TlsInternalReport, TlsReport, TlsReportPolicy}, }, - types::{EnumImpl, datetime::UTCDateTime, id::ObjectId}, + types::{EnumImpl, datetime::UTCDateTime}, }; use reqwest::header::CONTENT_TYPE; use std::fmt::Write; @@ -300,59 +300,43 @@ impl TlsReporting for Server { (object_id_v.object_id.id().id(), report) } else { let item_id = self.inner.data.queue_id_gen.generate(); - let deliver_at = (event.interval.to_timestamp() + event.interval.as_secs()) as i64; - - batch - .assert_value(pk.clone(), ()) - .set( - pk.clone(), - ObjectIdVersioned { - object_id: ObjectId::new(ObjectType::TlsInternalReport, item_id.into()), - version: 0, - } - .serialize(), - ) - .schedule_task_with_id( - item_id, - Task::TlsReport(TaskTlsReport { - report_id: item_id.into(), - status: TaskStatus::at(deliver_at), - }), - ); - let created_at = UTCDateTime::now(); - let deliver_at = UTCDateTime::from_timestamp(deliver_at); - ( - item_id, - TlsInternalReport { - created_at, - deliver_at, - domain: event.domain.clone(), - report: TlsReport { - report_id: format!("{}_{policy_hash}", created_at.timestamp()), - organization_name: self - .eval_if::( - &config.org_name, - &RecipientDomain::new(&event.domain), - event.span_id, - ) - .await - .clone(), - contact_info: self - .eval_if::( - &config.contact_info, - &RecipientDomain::new(&event.domain), - event.span_id, - ) - .await - .clone(), - date_range_end: deliver_at, - date_range_start: created_at, - policies: vec![], - }, - ..Default::default() + let deliver_at = UTCDateTime::from_timestamp( + (event.interval.to_timestamp() + event.interval.as_secs()) as i64, + ); + + let report = TlsInternalReport { + created_at, + deliver_at, + domain: event.domain.clone(), + report: TlsReport { + report_id: format!("{}_{policy_hash}", created_at.timestamp()), + organization_name: self + .eval_if::( + &config.org_name, + &RecipientDomain::new(&event.domain), + event.span_id, + ) + .await + .clone(), + contact_info: self + .eval_if::( + &config.contact_info, + &RecipientDomain::new(&event.domain), + event.span_id, + ) + .await + .clone(), + date_range_end: deliver_at, + date_range_start: created_at, + policies: vec![], }, - ) + ..Default::default() + }; + + report.write_ops(&mut batch, item_id, true); + + (item_id, report) }; let policy = if let Some(policy) = report diff --git a/crates/store/src/registry/mod.rs b/crates/store/src/registry/mod.rs index a2e2ebe1..132a2cf4 100644 --- a/crates/store/src/registry/mod.rs +++ b/crates/store/src/registry/mod.rs @@ -19,7 +19,7 @@ use registry::{ schema::{ prelude::{Object, ObjectInner, ObjectType, Property}, structs::{ - DeletedItem, DmarcInternalReport, SpamTrainingSample, Task, TlsInternalReport, Trace, + ArchivedItem, DmarcInternalReport, SpamTrainingSample, Task, TlsInternalReport, Trace, }, }, types::{EnumImpl, ObjectImpl, id::ObjectId}, @@ -109,10 +109,10 @@ impl Deserialize for SpamTrainingSample { } } -impl Deserialize for DeletedItem { +impl Deserialize for ArchivedItem { fn deserialize(bytes: &[u8]) -> trc::Result { let mut stream = PickledStream::new(bytes); - DeletedItem::unpickle(&mut stream).ok_or_else(|| { + ArchivedItem::unpickle(&mut stream).ok_or_else(|| { trc::EventType::Registry(trc::RegistryEvent::DeserializationError) .into_err() .caused_by(trc::location!()) diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 3773bc65..14ba4cde 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -465,7 +465,7 @@ impl + Sync + Send + Clone> Key for AnyKey { const MAILBOX_COLLECTION: u8 = Collection::Mailbox as u8; const MAILBOX_COUNTER_FIELD: u8 = MailboxField::UidCounter as u8; -const REG_DELETED_ITEM: u16 = ObjectType::DeletedItem as u16; +const REG_ARCHIVED_ITEM: u16 = ObjectType::ArchivedItem as u16; const REG_SPAM_SAMPLE: u16 = ObjectType::SpamTrainingSample as u16; const REG_ACCOUNT: u16 = ObjectType::Account as u16; const REG_DOMAIN: u16 = ObjectType::Domain as u16; @@ -558,7 +558,7 @@ impl ValueClass { RegistryClass::Item { object_id, .. } => match *object_id { REG_ACCOUNT | REG_DOMAIN | REG_TENANT | REG_ROLE | REG_OAUTH_CLIENT | REG_MAILING_LIST | REG_MASKED_EMAIL | REG_PUBLIC_KEY => SUBSPACE_DIRECTORY, - REG_DELETED_ITEM => SUBSPACE_DELETED_ITEMS, + REG_ARCHIVED_ITEM => SUBSPACE_DELETED_ITEMS, REG_SPAM_SAMPLE => SUBSPACE_SPAM_SAMPLES, REG_TRACE => SUBSPACE_TELEMETRY_SPAN, REG_METRIC => SUBSPACE_TELEMETRY_METRIC,