From c5796b4e7a522094bfe54e27bc7d7595c0169412 Mon Sep 17 00:00:00 2001 From: Maurus Decimus <11444311+mdecimus@users.noreply.github.com> Date: Fri, 3 Jul 2026 16:51:57 +0200 Subject: [PATCH] DKIMv2 signing --- CHANGELOG.md | 1 + README.md | 2 +- crates/jmap/src/registry/mapping/action.rs | 7 +- crates/main/src/test_data.rs | 6 +- crates/smtp/src/inbound/data.rs | 69 ++---- crates/smtp/src/inbound/dkim.rs | 253 +++++++++++++++++++++ crates/smtp/src/inbound/mod.rs | 24 +- crates/smtp/src/outbound/client.rs | 29 ++- crates/smtp/src/outbound/delivery.rs | 53 ++++- crates/smtp/src/outbound/local.rs | 36 ++- crates/smtp/src/outbound/session.rs | 10 +- crates/smtp/src/queue/dsn.rs | 22 +- crates/smtp/src/queue/mod.rs | 11 +- crates/smtp/src/queue/quota.rs | 48 ++-- crates/smtp/src/queue/spool.rs | 141 +++++++++--- crates/smtp/src/reporting/send.rs | 91 ++------ crates/smtp/src/scripts/event_loop.rs | 80 +++---- tests/src/smtp/lookup/utils.rs | 2 +- tests/src/smtp/queue/dsn.rs | 2 +- tests/src/smtp/queue/mod.rs | 2 +- 20 files changed, 588 insertions(+), 301 deletions(-) create mode 100644 crates/smtp/src/inbound/dkim.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 694b07d4..b11bd225 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file. This projec If you are upgrading from v0.16.x, replace the binary (or run `docker pull`). If you are upgrading from v0.15.x and below, please read the [upgrading documentation](https://github.com/stalwartlabs/stalwart/blob/main/UPGRADING/v0_16.md) for more information on how to upgrade from previous versions. ## Added +- DKIM2 implementation ([draft-ietf-dkim-dkim2-spec-03](https://datatracker.ietf.org/doc/draft-ietf-dkim-dkim2-spec/)). ## Changed diff --git a/README.md b/README.md index e085ee30..1805ca1c 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ Key features: - [POP3](https://datatracker.ietf.org/doc/html/rfc1939) server. - [STLS](https://datatracker.ietf.org/doc/html/rfc2595) and [SASL](https://datatracker.ietf.org/doc/html/rfc5034) support as well as other [extensions](https://datatracker.ietf.org/doc/html/rfc2449). - SMTP: - * SMTP server with built-in [DMARC](https://datatracker.ietf.org/doc/html/rfc7489), [DKIM](https://datatracker.ietf.org/doc/html/rfc6376), [SPF](https://datatracker.ietf.org/doc/html/rfc7208) and [ARC](https://datatracker.ietf.org/doc/html/rfc8617) support for message authentication. + * SMTP server with built-in [DMARC](https://datatracker.ietf.org/doc/html/rfc7489), [DKIMv2](https://datatracker.ietf.org/doc/draft-ietf-dkim-dkim2-spec/), [DKIMv1](https://datatracker.ietf.org/doc/html/rfc6376), [SPF](https://datatracker.ietf.org/doc/html/rfc7208) and [ARC](https://datatracker.ietf.org/doc/html/rfc8617) support for message authentication. * Strong transport security through [DANE](https://datatracker.ietf.org/doc/html/rfc6698), [MTA-STS](https://datatracker.ietf.org/doc/html/rfc8461) and [SMTP TLS](https://datatracker.ietf.org/doc/html/rfc8460) reporting. * Automated DKIM key rotation and management. * Inbound throttling and filtering with granular configuration rules, sieve scripting, MTA hooks and milter integration. diff --git a/crates/jmap/src/registry/mapping/action.rs b/crates/jmap/src/registry/mapping/action.rs index c865a29c..2c095893 100644 --- a/crates/jmap/src/registry/mapping/action.rs +++ b/crates/jmap/src/registry/mapping/action.rs @@ -238,8 +238,9 @@ pub(crate) async fn action_set( async fn classify_spam(server: &Server, mut request: SpamClassify) -> Option { // Built spam filter input + let raw_message = request.message.as_bytes(); let message = MessageParser::new() - .parse(request.message.as_bytes()) + .parse(raw_message) .filter(|m| m.root_part().headers().iter().any(|h| !h.name.is_other()))?; let remote_ip = request.remote_ip.into_inner(); @@ -303,7 +304,7 @@ async fn classify_spam(server: &Server, mut request: SpamClassify) -> Option) -> Vec { env_id: Some("env-001".into()), priority: 0, size: raw_messages[0].len() as u64, - quota_keys: Box::new([]), + metadata: Box::new([]), }, // Message 2: DSN bounce message with a temporary failure recipient Message { @@ -211,7 +211,7 @@ fn sample_queued_messages(blob_hashes: Vec) -> Vec { env_id: None, priority: -5, size: raw_messages[1].len() as u64, - quota_keys: Box::new([]), + metadata: Box::new([]), }, // Message 3: Report message with a temporary failure recipient Message { @@ -243,7 +243,7 @@ fn sample_queued_messages(blob_hashes: Vec) -> Vec { env_id: Some("env-report-99".into()), priority: 10, size: raw_messages[2].len() as u64, - quota_keys: Box::new([]), + metadata: Box::new([]), }, ] } diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index ed6b075b..0c7a6934 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -4,13 +4,13 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{AuthResult, DkimSign}; +use super::AuthResult; use crate::{ core::{Session, SessionAddress, State}, - inbound::milter::Modification, + inbound::{dkim::DkimSign, milter::Modification}, queue::{ self, Message, MessageSource, MessageWrapper, QueueEnvelope, RCPT_SPAM_PAYLOAD, - quota::HasQueueQuota, + quota::HasQueueQuota, spool::QueueParams, }, reporting::analysis::AnalyzeReport, scripts::ScriptResult, @@ -75,6 +75,7 @@ impl Session { // Authenticate message let mut auth_message = AuthenticatedMessage::from_parsed( &parsed_message, + &raw_message, self.server.core.smtp.mail_auth.dkim.strict, ); let has_date_header = auth_message.has_date_header(); @@ -279,7 +280,8 @@ impl Session { ); self.data.messages_sent += 1; - return (b"250 2.0.0 Message queued for delivery.\r\n"[..]).into(); + return (b"550 5.7.1 DSN rejected due to DKIM2 verification failure.\r\n"[..]) + .into(); } let time = Instant::now(); @@ -770,49 +772,15 @@ impl Session { headers.extend_from_slice(b"\r\n"); } - // DKIM sign - let raw_message = edited_message.as_deref().unwrap_or(raw_message.as_slice()); - if let Some(sign_with_domain) = self - .server - .eval_if::(&ac.dkim.sign, self, self.data.session_id) - .await - { - match self.server.dkim_signers(&sign_with_domain).await { - Ok(Some(signers)) => { - for signer in &signers.dkim1 { - match signer.sign_chained(&[headers.as_ref(), raw_message]) { - Ok(signature) => { - signature.write_header(&mut headers); - } - Err(err) => { - trc::error!( - trc::Error::from(err) - .span_id(self.data.session_id) - .details("Failed to DKIM sign message") - ); - } - } - } - } - Ok(None) => {} - Err(err) => { - trc::error!( - err.span_id(self.data.session_id) - .details("Failed to retrieve DKIM signers") - ); - } - } - } - // Update size + let original_message = raw_message.as_slice(); + let raw_message = edited_message.as_deref().unwrap_or(raw_message.as_slice()); message.message.size = (raw_message.len() + headers.len()) as u64; // Verify queue quota - if self.server.has_quota(&mut message).await { - // Prepare webhook event - let queue_id = message.queue_id; - + if let Some(metadata) = self.server.has_quota(&mut message).await { // Queue message + let queue_id = message.queue_id; let source = if !self.is_authenticated() { let dmarc_pass = dmarc_result.is_some_and(|result| result == DmarcResult::Pass); @@ -834,13 +802,18 @@ impl Session { } else { MessageSource::Authenticated }; + let dkim_signers = self + .server + .eval_signers(&ac.dkim.sign, self, self.data.session_id) + .await; if message .queue( - Some(&headers), - raw_message, - self.data.session_id, - &self.server, - source, + QueueParams::new(raw_message, self.data.session_id, &self.server, source) + .with_raw_headers(&headers) + .with_dkim_signers(dkim_signers) + .with_original_raw_message(original_message) + .with_original_authenticated_message(auth_message) + .with_metadata(metadata), ) .await { @@ -880,7 +853,7 @@ impl Session { size: 0, env_id: mail_from.dsn_info.map(|i| i.into_boxed_str()), blob_hash: Default::default(), - quota_keys: Default::default(), + metadata: Default::default(), received_from_ip: self.data.remote_ip, received_via_port: self.data.local_port, }; diff --git a/crates/smtp/src/inbound/dkim.rs b/crates/smtp/src/inbound/dkim.rs new file mode 100644 index 00000000..1af02a3b --- /dev/null +++ b/crates/smtp/src/inbound/dkim.rs @@ -0,0 +1,253 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::queue::{MessageWrapper, Metadata, spool::QueueParams}; +use common::{ + Server, + config::smtp::auth::{Dkim1Signer, DkimSigners}, + expr::{functions::ResolveVariable, if_block::IfBlock}, +}; +use mail_auth::{ + AuthenticatedMessage, + common::headers::HeaderWriter, + dkim2::{Hop, MessageInstance}, +}; +use mail_parser::{Address, parsers::MessageStream}; +use std::{collections::HashSet, sync::Arc}; +use utils::sanitize_email; + +pub(crate) trait DkimSign: Sync + Send { + fn sign_message( + &self, + message: &mut MessageWrapper, + params: &mut QueueParams<'_, '_>, + ) -> impl Future>> + Send; + + fn eval_signers( + &self, + if_block: &IfBlock, + resolver: &impl ResolveVariable, + session_id: u64, + ) -> impl Future>> + Send; +} + +impl DkimSign for Server { + async fn sign_message( + &self, + message: &mut MessageWrapper, + params: &mut QueueParams<'_, '_>, + ) -> Option> { + let signers = params.dkim_signers.as_ref().unwrap(); + let raw_message = params.raw_message; + + // DKIM1 signing + let mut headers = Vec::with_capacity(64); + for signer in &signers.dkim1 { + let result = match (signer, params.raw_headers) { + (Dkim1Signer::RsaSha256(signer), None) => signer.sign(raw_message), + (Dkim1Signer::Ed25519Sha256(signer), None) => signer.sign(raw_message), + (Dkim1Signer::RsaSha256(signer), Some(headers)) => { + signer.sign_chained([headers, raw_message].iter().copied()) + } + (Dkim1Signer::Ed25519Sha256(signer), Some(headers)) => { + signer.sign_chained([headers, raw_message].iter().copied()) + } + }; + + match result { + Ok(signature) => { + signature.write_header(&mut headers); + } + Err(err) => { + trc::error!( + trc::Error::from(err) + .span_id(params.session_id) + .details("Failed to sign message") + .caused_by(trc::location!()) + ); + } + } + } + + // DKIM2 signing + if let Some(signer) = &signers.dkim2 + && let Some(modified) = + AuthenticatedMessage::parse_with_opts(raw_message, params.raw_headers, true) + { + // Generate message instance + let original = params.original_authenticated_message.take().or_else(|| { + params + .original_raw_message + .and_then(AuthenticatedMessage::parse) + }); + let instance = MessageInstance::from_message(&modified, original.as_ref()); + if let Some(instance) = &instance { + instance.write(&mut headers); + } + + // Obtain disclosed and undisclosed recipients + let envelopes = message.undisclosed_recipients(&modified); + + // Generate DKIM2 signature for disclosed recipients + if !envelopes.disclosed_recipients.is_empty() { + match signer.sign_with_message_instance( + &modified, + instance.as_ref(), + Hop::real( + message.message.return_path.as_ref(), + envelopes.disclosed_recipients, + ), + ) { + Ok(signature) => { + if envelopes.undisclosed_recipients.is_empty() { + // Happy path: no undisclosed recipients, serialize signature straight to blob + signature.write(&mut headers); + } else { + // Undisclosed recipients present, serialize signature to metadata + let mut header = Vec::with_capacity(64); + signature.write(&mut header); + params.metadata.push(Metadata::Headers { + value: header.into_boxed_slice(), + id: u64::MAX, + }); + } + } + Err(err) => { + trc::error!( + trc::Error::from(err) + .span_id(params.session_id) + .details("Failed to DKIM2 sign message") + ); + } + } + } + + // Generate DKIM2 signature for undisclosed recipients + for (pos, rcpt) in envelopes.undisclosed_recipients { + match signer.sign_with_message_instance( + &modified, + instance.as_ref(), + Hop::real(message.message.return_path.as_ref(), [rcpt]), + ) { + Ok(signature) => { + // Serialize signature to metadata + let mut header = Vec::with_capacity(64); + signature.write(&mut header); + params.metadata.push(Metadata::Headers { + value: header.into_boxed_slice(), + id: pos as u64, + }); + } + Err(err) => { + trc::error!( + trc::Error::from(err) + .span_id(params.session_id) + .details("Failed to DKIM2 sign message") + ); + } + } + } + } + + (!headers.is_empty()).then_some(headers) + } + + async fn eval_signers( + &self, + if_block: &IfBlock, + resolver: &impl ResolveVariable, + session_id: u64, + ) -> Option> { + let sign_with_domain = self + .eval_if::(if_block, resolver, session_id) + .await?; + match self.dkim_signers(&sign_with_domain).await { + Ok(signers) => signers, + Err(err) => { + trc::error!( + err.span_id(session_id) + .details("Failed to retrieve DKIM signers") + ); + None + } + } + } +} + +struct Dkim2Envelopes<'x> { + undisclosed_recipients: Vec<(usize, &'x str)>, + disclosed_recipients: Vec<&'x str>, +} + +impl MessageWrapper { + fn undisclosed_recipients<'x>( + &'x self, + message: &AuthenticatedMessage<'_>, + ) -> Dkim2Envelopes<'x> { + if self.message.recipients.len() == 1 { + return Dkim2Envelopes { + undisclosed_recipients: Vec::new(), + disclosed_recipients: vec![self.message.recipients[0].address.as_ref()], + }; + } + + let mut recipients = HashSet::with_capacity(self.message.recipients.len()); + + for addr in message.headers.iter().filter_map(|(name, value)| { + let name = name.trim_ascii(); + if name.len() == 2 + && (name.eq_ignore_ascii_case(b"to") || name.eq_ignore_ascii_case(b"cc")) + { + MessageStream::new(value).parse_address().into_address() + } else { + None + } + }) { + match addr { + Address::List(addrs) => { + recipients.extend( + addrs + .iter() + .filter_map(|a| a.address()) + .map(sanitize_or_lower), + ); + } + Address::Group(groups) => { + for group in groups { + recipients.extend( + group + .addresses + .iter() + .filter_map(|a| a.address()) + .map(sanitize_or_lower), + ); + } + } + } + } + + let mut undisclosed_recipients = Vec::new(); + let mut disclosed_recipients = Vec::new(); + for (i, rcpt) in self.message.recipients.iter().enumerate() { + if !recipients.contains(rcpt.address.as_ref()) + && !recipients.contains(&sanitize_or_lower(&rcpt.address)) + { + undisclosed_recipients.push((i, rcpt.address.as_ref())); + } else { + disclosed_recipients.push(rcpt.address.as_ref()); + } + } + + Dkim2Envelopes { + undisclosed_recipients, + disclosed_recipients, + } + } +} + +fn sanitize_or_lower(rcpt: &str) -> String { + sanitize_email(rcpt).unwrap_or_else(|| rcpt.to_lowercase()) +} diff --git a/crates/smtp/src/inbound/mod.rs b/crates/smtp/src/inbound/mod.rs index ff255228..99585af9 100644 --- a/crates/smtp/src/inbound/mod.rs +++ b/crates/smtp/src/inbound/mod.rs @@ -4,12 +4,12 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use common::config::smtp::auth::Dkim1Signer; -use mail_auth::{DkimResult, DmarcResult, IprevResult, SpfResult, dkim::Signature, dmarc::Policy}; +use mail_auth::{DkimResult, DmarcResult, IprevResult, SpfResult, dmarc::Policy}; use std::borrow::Cow; pub mod auth; pub mod data; +pub mod dkim; pub mod ehlo; pub mod hooks; pub mod mail; @@ -26,26 +26,6 @@ pub struct FilterResponse { pub disconnect: bool, } -pub trait DkimSign { - fn sign(&self, message: &[u8]) -> mail_auth::Result; - fn sign_chained(&self, message: &[&[u8]]) -> mail_auth::Result; -} - -impl DkimSign for Dkim1Signer { - fn sign(&self, message: &[u8]) -> mail_auth::Result { - match self { - Dkim1Signer::RsaSha256(signer) => signer.sign(message), - Dkim1Signer::Ed25519Sha256(signer) => signer.sign(message), - } - } - fn sign_chained(&self, message: &[&[u8]]) -> mail_auth::Result { - match self { - Dkim1Signer::RsaSha256(signer) => signer.sign_chained(message.iter().copied()), - Dkim1Signer::Ed25519Sha256(signer) => signer.sign_chained(message.iter().copied()), - } - } -} - pub trait AuthResult { fn as_str(&self) -> &'static str; } diff --git a/crates/smtp/src/outbound/client.rs b/crates/smtp/src/outbound/client.rs index be81474d..eb0c61e4 100644 --- a/crates/smtp/src/outbound/client.rs +++ b/crates/smtp/src/outbound/client.rs @@ -171,7 +171,8 @@ impl SmtpClient { pub async fn send_message( &mut self, message: &MessageWrapper, - bdat_cmd: &Option, + rcpt_headers: Option<&[u8]>, + bdat_cmd: &mut Option, params: &SessionParams<'_>, ) -> Result<(), Status>, ErrorDetails>> { match params @@ -183,6 +184,11 @@ impl SmtpClient { Ok(Some(raw_message)) => { tokio::time::timeout(params.conn_strategy.timeout_data, async { if let Some(bdat_cmd) = bdat_cmd { + *bdat_cmd = format!( + "BDAT {} LAST\r\n", + raw_message.len() + rcpt_headers.map(|h| h.len()).unwrap_or(0) + ); + trc::event!( Delivery(DeliveryEvent::RawOutput), SpanId = self.session_id, @@ -190,8 +196,13 @@ impl SmtpClient { Size = bdat_cmd.len() ); - self.write_chunks(&[bdat_cmd.as_bytes(), &raw_message]) - .await + let chunks = if let Some(rcpt_headers) = rcpt_headers { + &[bdat_cmd.as_bytes(), rcpt_headers, &raw_message][..] + } else { + &[bdat_cmd.as_bytes(), &raw_message][..] + }; + + self.write_chunks(chunks).await } else { trc::event!( Delivery(DeliveryEvent::RawOutput), @@ -202,9 +213,15 @@ impl SmtpClient { self.write_chunks(&[b"DATA\r\n"]).await?; self.read().await?.assert_code(354)?; - self.write_message(&raw_message) - .await - .map_err(ClientError::from) + if let Some(rcpt_headers) = rcpt_headers + && let Err(err) = self.write_chunks(&[rcpt_headers]).await + { + Err(err) + } else { + self.write_message(&raw_message) + .await + .map_err(ClientError::from) + } } }) .await diff --git a/crates/smtp/src/outbound/delivery.rs b/crates/smtp/src/outbound/delivery.rs index 6a701f08..5d62a2e5 100644 --- a/crates/smtp/src/outbound/delivery.rs +++ b/crates/smtp/src/outbound/delivery.rs @@ -19,7 +19,8 @@ use crate::queue::dsn::SendDsn; use crate::queue::spool::SmtpSpool; use crate::queue::throttle::IsAllowed; use crate::queue::{ - Error, FROM_REPORT, HostResponse, MessageWrapper, QueueEnvelope, QueuedMessage, Status, + Error, FROM_REPORT, HostResponse, MessageWrapper, Metadata, QueueEnvelope, QueuedMessage, + Status, }; use crate::reporting::send::MtaReportSend; use crate::{queue::ErrorDetails, reporting::tls::TlsRptOptions}; @@ -147,6 +148,7 @@ impl QueuedMessage { }); } + #[allow(clippy::type_complexity)] async fn deliver_task(self, server: Server, mut message: MessageWrapper) -> QueueEventStatus { // Check that the message still has recipients to be delivered let has_pending_delivery = message.has_pending_delivery(); @@ -218,7 +220,19 @@ impl QueuedMessage { // Group recipients by route let queue_config = &server.core.smtp.queue; let now_ = now(); - let mut routes: AHashMap<(&str, &RoutingStrategy), Vec> = AHashMap::new(); + let mut routes: AHashMap<(&str, &RoutingStrategy, Option<&[u8]>), Vec> = + AHashMap::new(); + let mut has_rcpt_headers = false; + let mut default_rcpt_header = None; + for metadata in message.message.metadata.iter() { + if let Metadata::Headers { value, id } = metadata { + has_rcpt_headers = true; + if *id == u64::MAX { + default_rcpt_header = Some(value.as_ref()); + break; + } + } + } for (rcpt_idx, rcpt) in message.message.recipients.iter().enumerate() { if matches!( &rcpt.status, @@ -235,8 +249,21 @@ impl QueuedMessage { message.span_id, ); + // Map RCPT headers + let mut rcpt_headers = default_rcpt_header; + if has_rcpt_headers { + for metadata in message.message.metadata.iter() { + if let Metadata::Headers { value, id } = metadata + && *id == rcpt_idx as u64 + { + rcpt_headers = Some(value.as_ref()); + break; + } + } + } + routes - .entry((rcpt.domain_part(), route)) + .entry((rcpt.domain_part(), route, rcpt_headers)) .or_default() .push(rcpt_idx); } @@ -244,7 +271,7 @@ impl QueuedMessage { let no_ip = IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)); let mut delivery_results: Vec = Vec::new(); - 'next_route: for ((domain, route), rcpt_idxs) in routes { + 'next_route: for ((domain, route, rcpt_headers), rcpt_idxs) in routes { trc::event!( Delivery(DeliveryEvent::DomainDeliveryStart), SpanId = message.span_id, @@ -1226,6 +1253,7 @@ impl QueuedMessage { .deliver( smtp_client, rcpt_idxs, + rcpt_headers, &mut delivery_results, params, ) @@ -1285,6 +1313,7 @@ impl QueuedMessage { .deliver( smtp_client, rcpt_idxs, + rcpt_headers, &mut delivery_results, params, ) @@ -1341,7 +1370,13 @@ impl QueuedMessage { ); message - .deliver(smtp_client, rcpt_idxs, &mut delivery_results, params) + .deliver( + smtp_client, + rcpt_idxs, + rcpt_headers, + &mut delivery_results, + params, + ) .await } } else { @@ -1381,7 +1416,13 @@ impl QueuedMessage { // Deliver message message - .deliver(smtp_client, rcpt_idxs, &mut delivery_results, params) + .deliver( + smtp_client, + rcpt_idxs, + rcpt_headers, + &mut delivery_results, + params, + ) .await } diff --git a/crates/smtp/src/outbound/local.rs b/crates/smtp/src/outbound/local.rs index fdc564b8..f22a26b7 100644 --- a/crates/smtp/src/outbound/local.rs +++ b/crates/smtp/src/outbound/local.rs @@ -5,13 +5,14 @@ */ use crate::{ + inbound::dkim::DkimSign, outbound::DeliveryResult, queue::{ Error, ErrorDetails, FROM_AUTHENTICATED, FROM_UNAUTHENTICATED_DMARC, HostResponse, MessageSource, MessageWrapper, RCPT_SPAM_PAYLOAD, Status, UnexpectedResponse, - quota::HasQueueQuota, spool::SmtpSpool, + quota::HasQueueQuota, + spool::{QueueParams, SmtpSpool}, }, - reporting::send::MtaReportSend, }; use common::Server; use email::message::delivery::{IngestMessage, IngestRecipient, LocalDeliveryStatus, MailDelivery}; @@ -103,26 +104,23 @@ impl MessageWrapper { message.expand_and_add_recipient(rcpt, server).await; } - // Sign message - let signature = server - .sign_message( - &mut message, - &server.core.sieve.sign, - &autogenerated.message, - ) - .await; - // Queue Message - message.message.size = - (autogenerated.message.len() + signature.as_ref().map_or(0, |s| s.len())) as u64; - if server.has_quota(&mut message).await { + message.message.size = autogenerated.message.len() as u64; + if let Some(metadata) = server.has_quota(&mut message).await { + let dkim_signers = server + .eval_signers(&server.core.sieve.sign, &message.message, self.span_id) + .await; + message .queue( - signature.as_deref(), - &autogenerated.message, - self.span_id, - server, - MessageSource::Autogenerated, + QueueParams::new( + &autogenerated.message, + self.span_id, + server, + MessageSource::Autogenerated, + ) + .with_dkim_signers(dkim_signers) + .with_metadata(metadata), ) .await; } else { diff --git a/crates/smtp/src/outbound/session.rs b/crates/smtp/src/outbound/session.rs index cbacfa2c..7f0d5c76 100644 --- a/crates/smtp/src/outbound/session.rs +++ b/crates/smtp/src/outbound/session.rs @@ -38,6 +38,7 @@ impl MessageWrapper { &self, mut smtp_client: SmtpClient, rcpt_idxs: Vec, + rcpt_headers: Option<&[u8]>, statuses: &mut Vec, mut params: SessionParams<'_>, ) { @@ -246,11 +247,12 @@ impl MessageWrapper { // Send message if !accepted_rcpts.is_empty() { let time = Instant::now(); - let bdat_cmd = capabilities - .has_capability(EXT_CHUNKING) - .then(|| format!("BDAT {} LAST\r\n", self.message.size)); + let mut bdat_cmd = capabilities.has_capability(EXT_CHUNKING).then(String::new); - if let Err(status) = smtp_client.send_message(self, &bdat_cmd, ¶ms).await { + if let Err(status) = smtp_client + .send_message(self, rcpt_headers, &mut bdat_cmd, ¶ms) + .await + { trc::event!( Delivery(DeliveryEvent::MessageRejected), SpanId = params.session_id, diff --git a/crates/smtp/src/queue/dsn.rs b/crates/smtp/src/queue/dsn.rs index 9624f08c..eaca6bbe 100644 --- a/crates/smtp/src/queue/dsn.rs +++ b/crates/smtp/src/queue/dsn.rs @@ -9,8 +9,9 @@ use super::{ Error, ErrorDetails, HostResponse, Message, MessageSource, QueueEnvelope, RCPT_DSN_SENT, Recipient, Status, }; +use crate::inbound::dkim::DkimSign; +use crate::queue::spool::QueueParams; use crate::queue::{MessageWrapper, UnexpectedResponse}; -use crate::reporting::send::MtaReportSend; use common::Server; use mail_builder::MessageBuilder; use mail_builder::headers::HeaderType; @@ -42,19 +43,18 @@ impl SendDsn for Server { .expand_and_add_recipient(message.message.return_path.as_ref(), self) .await; - // Sign message - let signature = self - .sign_message(message, &self.core.smtp.queue.dsn.sign, &dsn) - .await; - // Queue DSN + let dkim_signers = self + .eval_signers( + &self.core.smtp.queue.dsn.sign, + &message.message, + message.span_id, + ) + .await; dsn_message .queue( - signature.as_deref(), - &dsn, - message.span_id, - self, - MessageSource::Dsn, + QueueParams::new(&dsn, message.span_id, self, MessageSource::Dsn) + .with_dkim_signers(dkim_signers), ) .await; } diff --git a/crates/smtp/src/queue/mod.rs b/crates/smtp/src/queue/mod.rs index 23577916..9cf933a2 100644 --- a/crates/smtp/src/queue/mod.rs +++ b/crates/smtp/src/queue/mod.rs @@ -69,7 +69,7 @@ pub struct Message { pub priority: i16, pub size: u64, - pub quota_keys: Box<[QuotaKey]>, + pub metadata: Box<[Metadata]>, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -91,9 +91,10 @@ pub struct MessageWrapper { Eq, serde::Deserialize, )] -pub enum QuotaKey { - Size { key: Box<[u8]>, id: u64 }, - Count { key: Box<[u8]>, id: u64 }, +pub enum Metadata { + QueueSize { key: Box<[u8]>, id: u64 }, + QueueCount { key: Box<[u8]>, id: u64 }, + Headers { value: Box<[u8]>, id: u64 }, } #[derive( @@ -127,7 +128,7 @@ pub const FROM_REPORT: u64 = 1 << 36; pub const FROM_AUTOGENERATED: u64 = 1 << 37; pub const RCPT_DSN_SENT: u64 = 1 << 32; -//pub const RCPT_STATUS_CHANGED: u64 = 1 << 33; +//pub const RCPT_UNDISCLOSED: u64 = 1 << 33; pub const RCPT_SPAM_PAYLOAD: u64 = 1 << 34; #[derive( diff --git a/crates/smtp/src/queue/quota.rs b/crates/smtp/src/queue/quota.rs index c4212bda..d94878c7 100644 --- a/crates/smtp/src/queue/quota.rs +++ b/crates/smtp/src/queue/quota.rs @@ -4,7 +4,7 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::{QueueEnvelope, QuotaKey, Status}; +use super::{Metadata, QueueEnvelope, Status}; use crate::{core::throttle::NewKey, queue::MessageWrapper}; use ahash::AHashSet; use common::{Server, config::smtp::queue::QueueQuota, expr::functions::ResolveVariable}; @@ -18,20 +18,23 @@ use trc::QueueEvent; use utils::DomainPart; pub trait HasQueueQuota: Sync + Send { - fn has_quota(&self, message: &mut MessageWrapper) -> impl Future + Send; + fn has_quota( + &self, + message: &mut MessageWrapper, + ) -> impl Future>> + Send; fn check_quota<'x>( &'x self, quota: &'x QueueQuota, envelope: &impl ResolveVariable, size: u64, id: u64, - refs: &mut Vec, + refs: &mut Vec, session_id: u64, ) -> impl Future + Send; } impl HasQueueQuota for Server { - async fn has_quota(&self, message: &mut MessageWrapper) -> bool { + async fn has_quota(&self, message: &mut MessageWrapper) -> Option> { let mut quota_keys = Vec::new(); if !self.core.smtp.queue.quota.sender.is_empty() { @@ -54,7 +57,7 @@ impl HasQueueQuota for Server { Type = "Sender" ); - return false; + return None; } } } @@ -82,7 +85,7 @@ impl HasQueueQuota for Server { Type = "Domain" ); - return false; + return None; } } } @@ -108,14 +111,12 @@ impl HasQueueQuota for Server { Type = "Recipient" ); - return false; + return None; } } } - message.message.quota_keys = quota_keys.into_boxed_slice(); - - true + Some(quota_keys) } async fn check_quota<'x>( @@ -124,7 +125,7 @@ impl HasQueueQuota for Server { envelope: &impl ResolveVariable, size: u64, id: u64, - refs: &mut Vec, + refs: &mut Vec, session_id: u64, ) -> bool { if !quota.expr.is_empty() @@ -147,7 +148,7 @@ impl HasQueueQuota for Server { if used_size + size > max_size { return false; } else { - refs.push(QuotaKey::Size { + refs.push(Metadata::QueueSize { key: key.as_ref().into(), id, }); @@ -167,7 +168,7 @@ impl HasQueueQuota for Server { if total_messages + 1 > max_messages { return false; } else { - refs.push(QuotaKey::Count { + refs.push(Metadata::QueueCount { key: key.as_ref().into(), id, }); @@ -180,7 +181,12 @@ impl HasQueueQuota for Server { impl MessageWrapper { pub fn release_quota(&mut self, batch: &mut BatchBuilder) { - if self.message.quota_keys.is_empty() { + if !self.message.metadata.iter().any(|metadata| { + matches!( + metadata, + Metadata::QueueSize { .. } | Metadata::QueueCount { .. } + ) + }) { return; } let mut quota_ids = Vec::with_capacity(self.message.recipients.len()); @@ -199,27 +205,27 @@ impl MessageWrapper { } if !quota_ids.is_empty() { - let mut quota_keys = Vec::new(); - for quota_key in std::mem::take(&mut self.message.quota_keys) { - match quota_key { - QuotaKey::Count { id, key } if quota_ids.contains(&id) => { + let mut metadata = Vec::new(); + for entry in std::mem::take(&mut self.message.metadata) { + match entry { + Metadata::QueueCount { id, key } if quota_ids.contains(&id) => { batch.add( ValueClass::Queue(QueueClass::QuotaCount(key.into_vec())), -1, ); } - QuotaKey::Size { id, key } if quota_ids.contains(&id) => { + Metadata::QueueSize { id, key } if quota_ids.contains(&id) => { batch.add( ValueClass::Queue(QueueClass::QuotaSize(key.into_vec())), -(self.message.size as i64), ); } _ => { - quota_keys.push(quota_key); + metadata.push(entry); } } } - self.message.quota_keys = quota_keys.into_boxed_slice(); + self.message.metadata = metadata.into_boxed_slice(); } } } diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 5285b32a..f6a9f301 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -5,19 +5,22 @@ */ use super::{ - ArchivedMessage, ArchivedStatus, Message, MessageSource, QueueEnvelope, QueueId, QueuedMessage, - QuotaKey, Recipient, Schedule, Status, + ArchivedMessage, ArchivedStatus, Message, MessageSource, Metadata, QueueEnvelope, QueueId, + QueuedMessage, Recipient, Schedule, Status, }; +use crate::inbound::dkim::DkimSign; use crate::queue::manager::{LockedMessage, Queue}; use crate::queue::{ FROM_AUTHENTICATED, FROM_AUTOGENERATED, FROM_DSN, FROM_REPORT, FROM_UNAUTHENTICATED, FROM_UNAUTHENTICATED_DMARC, MessageWrapper, }; use ahash::{AHashMap, AHashSet}; +use common::config::smtp::auth::DkimSigners; use common::config::smtp::queue::{ArchivedQueueExpiry, QueueName}; use common::ipc::QueueEvent; use common::network::RcptResolution; use common::{KV_LOCK_QUEUE_MESSAGE, Server}; +use mail_auth::AuthenticatedMessage; use registry::schema::prelude::{ObjectType, Property}; use registry::schema::structs::SpamTrainingSample; use registry::types::datetime::UTCDateTime; @@ -27,6 +30,7 @@ use std::borrow::Cow; use std::collections::hash_map::Entry; use std::future::Future; use std::net::{IpAddr, Ipv4Addr}; +use std::sync::Arc; use std::time::SystemTime; use store::write::key::DeserializeBigEndian; use store::write::serialize::rkyv_deserialize; @@ -100,7 +104,7 @@ impl SmtpSpool for Server { priority: 0, size: 0, blob_hash: Default::default(), - quota_keys: Default::default(), + metadata: Default::default(), received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), received_via_port: 0, }, @@ -308,15 +312,38 @@ fn lock_id(queue_id: QueueId, queue_name: QueueName) -> [u8; 16] { id } +pub(crate) struct QueueParams<'x, 'y> { + pub raw_message: &'x [u8], + pub raw_headers: Option<&'x [u8]>, + pub metadata: Vec, + pub original_raw_message: Option<&'x [u8]>, + pub original_authenticated_message: Option>, + pub dkim_signers: Option>, + pub session_id: u64, + pub server: &'y Server, + pub source: MessageSource, +} + impl MessageWrapper { - pub async fn queue( - mut self, - raw_headers: Option<&[u8]>, - raw_message: &[u8], - session_id: u64, - server: &Server, - source: MessageSource, - ) -> bool { + pub(crate) async fn queue<'x, 'y>(mut self, mut params: QueueParams<'x, 'y>) -> bool { + // Add DKIM signatures + let dkim_headers = if params.dkim_signers.is_some() { + params.server.sign_message(&mut self, &mut params).await + } else { + None + }; + + // Fetch params + let QueueParams { + raw_message, + raw_headers, + session_id, + server, + source, + metadata, + .. + } = params; + // Set flags let (flags, event, train_spam) = match source { MessageSource::Authenticated => ( @@ -351,8 +378,12 @@ impl MessageWrapper { self.message.flags |= flags; // Write blob - let message = if let Some(raw_headers) = raw_headers { - let mut message = Vec::with_capacity(raw_headers.len() + raw_message.len()); + let raw_headers = raw_headers.unwrap_or_default(); + let dkim_headers = dkim_headers.as_deref().unwrap_or_default(); + let message = if !raw_headers.is_empty() || !dkim_headers.is_empty() { + let mut message = + Vec::with_capacity(raw_headers.len() + dkim_headers.len() + raw_message.len()); + message.extend_from_slice(dkim_headers); message.extend_from_slice(raw_headers); message.extend_from_slice(raw_message); Cow::Owned(message) @@ -361,10 +392,11 @@ impl MessageWrapper { }; self.message.blob_hash = BlobHash::generate(message.as_ref()); - // Generate id + // Update size if self.message.size == 0 { self.message.size = message.len() as u64; } + self.message.metadata = metadata.into_boxed_slice(); // Reserve and write blob let mut batch = BatchBuilder::new(); @@ -434,17 +466,18 @@ impl MessageWrapper { let mut batch = BatchBuilder::new(); // Reserve quotas - for quota_key in &self.message.quota_keys { - match quota_key { - QuotaKey::Count { key, .. } => { + for metadata in &self.message.metadata { + match metadata { + Metadata::QueueCount { key, .. } => { batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), 1); } - QuotaKey::Size { key, .. } => { + Metadata::QueueSize { key, .. } => { batch.add( ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())), self.message.size as i64, ); } + Metadata::Headers { .. } => {} } } @@ -756,17 +789,18 @@ impl MessageWrapper { } // Release all quotas - for quota_key in self.message.quota_keys { - match quota_key { - QuotaKey::Count { key, .. } => { + for metadata in self.message.metadata { + match metadata { + Metadata::QueueCount { key, .. } => { batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), -1); } - QuotaKey::Size { key, .. } => { + Metadata::QueueSize { key, .. } => { batch.add( ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())), -(self.message.size as i64), ); } + Metadata::Headers { .. } => {} } } @@ -918,17 +952,18 @@ impl MessageWrapper { ))); } - for quota_key in self.message.quota_keys { - match quota_key { - QuotaKey::Count { key, .. } => { + for metadata in self.message.metadata { + match metadata { + Metadata::QueueCount { key, .. } => { batch.add(ValueClass::Queue(QueueClass::QuotaCount(key.to_vec())), -1); } - QuotaKey::Size { key, .. } => { + Metadata::QueueSize { key, .. } => { batch.add( ValueClass::Queue(QueueClass::QuotaSize(key.to_vec())), -(self.message.size as i64), ); } + Metadata::Headers { .. } => {} } } @@ -1047,3 +1082,57 @@ impl ArchivedMessage { next_notify } } + +impl<'x, 'y> QueueParams<'x, 'y> { + pub fn new( + raw_message: &'x [u8], + session_id: u64, + server: &'y Server, + source: MessageSource, + ) -> Self { + QueueParams { + raw_message, + dkim_signers: None, + raw_headers: None, + session_id, + server, + source, + original_raw_message: None, + original_authenticated_message: None, + metadata: Vec::new(), + } + } + + pub fn with_original_authenticated_message( + mut self, + authenticated_message: AuthenticatedMessage<'x>, + ) -> Self { + self.original_authenticated_message = Some(authenticated_message); + self + } + + pub fn with_original_raw_message(mut self, raw_message: &'x [u8]) -> Self { + self.original_raw_message = Some(raw_message); + self + } + + pub fn with_dkim_signers(mut self, dkim_signers: Option>) -> Self { + self.dkim_signers = dkim_signers; + self + } + + pub fn with_raw_headers(mut self, raw_headers: &'x [u8]) -> Self { + self.raw_headers = Some(raw_headers); + self + } + + pub fn with_raw_headers_opt(mut self, raw_headers: Option<&'x [u8]>) -> Self { + self.raw_headers = raw_headers; + self + } + + pub fn with_metadata(mut self, metadata: Vec) -> Self { + self.metadata = metadata; + self + } +} diff --git a/crates/smtp/src/reporting/send.rs b/crates/smtp/src/reporting/send.rs index c0d1349e..44302fa2 100644 --- a/crates/smtp/src/reporting/send.rs +++ b/crates/smtp/src/reporting/send.rs @@ -5,11 +5,13 @@ */ use crate::{ - inbound::DkimSign, - queue::{MessageSource, MessageWrapper, spool::SmtpSpool}, + inbound::dkim::DkimSign, + queue::{ + MessageSource, + spool::{QueueParams, SmtpSpool}, + }, }; use common::{Server, expr::if_block::IfBlock, ipc::ReportingEvent}; -use mail_auth::common::headers::HeaderWriter; pub trait MtaReportSend: Sync + Send { fn send_report( @@ -35,13 +37,6 @@ pub trait MtaReportSend: Sync + Send { &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 { @@ -60,9 +55,6 @@ impl MtaReportSend for Server { message.add_expanded_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"))] @@ -82,13 +74,13 @@ impl MtaReportSend for Server { } // Queue message + let dkim_signers = self + .eval_signers(sign_config, &message.message, parent_session_id) + .await; message .queue( - signature.as_deref(), - &report, - parent_session_id, - self, - MessageSource::Report, + QueueParams::new(&report, parent_session_id, self, MessageSource::Report) + .with_dkim_signers(dkim_signers), ) .await; } @@ -107,22 +99,22 @@ impl MtaReportSend for Server { message.add_expanded_recipient(rcpt, self).await; } - // Sign message - let signature = if let Some(sign_config) = sign_config { - self.sign_message(&mut message, sign_config, &raw_message) + // Queue message + let dkim_signers = if let Some(sign_config) = sign_config { + self.eval_signers(sign_config, &message.message, parent_session_id) .await } else { None }; - - // Queue message message .queue( - signature.as_deref(), - &raw_message, - parent_session_id, - self, - MessageSource::Autogenerated, + QueueParams::new( + &raw_message, + parent_session_id, + self, + MessageSource::Autogenerated, + ) + .with_dkim_signers(dkim_signers), ) .await; } @@ -136,47 +128,4 @@ impl MtaReportSend for Server { ); } } - - 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.dkim1 { - 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/scripts/event_loop.rs b/crates/smtp/src/scripts/event_loop.rs index a8cf4c77..004c11b9 100644 --- a/crates/smtp/src/scripts/event_loop.rs +++ b/crates/smtp/src/scripts/event_loop.rs @@ -4,12 +4,12 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use crate::{ - inbound::DkimSign, - queue::{MessageSource, quota::HasQueueQuota, spool::SmtpSpool}, +use crate::queue::{ + MessageSource, + quota::HasQueueQuota, + spool::{QueueParams, SmtpSpool}, }; use common::{Server, config::smtp::queue::QueueExpiry, scripts::plugins::PluginContext}; -use mail_auth::common::headers::HeaderWriter; use mail_parser::{Encoding, Message, MessagePart, PartType}; use sieve::{ Event, Input, MatchAs, Recipient, Sieve, @@ -19,7 +19,7 @@ use smtp_proto::{ MAIL_BY_TRACE, MAIL_RET_FULL, MAIL_RET_HDRS, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, }; -use std::{borrow::Cow, future::Future, sync::Arc, time::Instant}; +use std::{future::Future, sync::Arc, time::Instant}; use trc::SieveEvent; use super::{ScriptModification, ScriptParameters, ScriptResult}; @@ -277,59 +277,35 @@ impl RunScript for Server { instance.message().raw_message().into() }; if let Some(raw_message) = raw_message.filter(|m| !m.is_empty()) { - let headers = if let Some(sign_domain) = ¶ms.sign_domain { - match self.dkim_signers(sign_domain).await { - Ok(Some(signers)) => { - let mut headers = Vec::new(); - - for signer in &signers.dkim1 { - match signer.sign(raw_message) { - Ok(signature) => { - signature.write_header(&mut headers); - } - Err(err) => { - trc::error!( - trc::Error::from(err) - .span_id(session_id) - .caused_by(trc::location!()) - .details("DKIM sign failed") - ); - } - } - } - - if is_forward { - headers.extend_from_slice( - params.headers.unwrap_or_default(), + if let Some(metadata) = self.has_quota(&mut message).await { + let dkim_signers = if let Some(sign_domain) = ¶ms.sign_domain { + match self.dkim_signers(sign_domain).await { + Ok(signers) => signers, + Err(err) => { + trc::error!( + err.details("Failed to obtain DKIM signers") + .caused_by(trc::location!()) ); + + None } - - Some(Cow::Owned(headers)) } - Ok(None) => None, - Err(err) => { - trc::error!( - err.details("Failed to obtain DKIM signers") - .caused_by(trc::location!()) - ); + } else { + None + }; - None - } - } - } else if is_forward { - params.headers.map(Cow::Borrowed) - } else { - None - }; - - if self.has_quota(&mut message).await { message .queue( - headers.as_deref(), - raw_message, - session_id, - self, - MessageSource::Autogenerated, + QueueParams::new( + raw_message, + session_id, + self, + MessageSource::Autogenerated, + ) + .with_dkim_signers(dkim_signers) + .with_raw_headers_opt(params.headers.filter(|_| is_forward)) + .with_original_raw_message(instance.message().raw_message()) + .with_metadata(metadata), ) .await; } else { diff --git a/tests/src/smtp/lookup/utils.rs b/tests/src/smtp/lookup/utils.rs index f7b05a55..51a1967a 100644 --- a/tests/src/smtp/lookup/utils.rs +++ b/tests/src/smtp/lookup/utils.rs @@ -207,7 +207,7 @@ async fn strategies() { env_id: None, priority: 0, size: 978, - quota_keys: Default::default(), + metadata: Default::default(), }; assert_eq!( diff --git a/tests/src/smtp/queue/dsn.rs b/tests/src/smtp/queue/dsn.rs index dcc45b21..8d9bc72d 100644 --- a/tests/src/smtp/queue/dsn.rs +++ b/tests/src/smtp/queue/dsn.rs @@ -70,7 +70,7 @@ async fn generate_dsn() { env_id: None, priority: 0, blob_hash: BlobHash::generate(dsn_original.as_bytes()), - quota_keys: Default::default(), + metadata: Default::default(), received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), received_via_port: 0, }, diff --git a/tests/src/smtp/queue/mod.rs b/tests/src/smtp/queue/mod.rs index d399181c..e734d73c 100644 --- a/tests/src/smtp/queue/mod.rs +++ b/tests/src/smtp/queue/mod.rs @@ -42,7 +42,7 @@ pub fn new_message(queue_id: u64) -> MessageWrapper { flags: 0, env_id: None, priority: 0, - quota_keys: Default::default(), + metadata: Default::default(), blob_hash: Default::default(), received_from_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), received_via_port: 0,