Merge upstream v0.16.23

Five conflicts, resolved:

- crates/common/src/auth/authentication.rs: upstream's get_directory_for_token
  and JwtClaims replace extract_jwt_domain; the per-domain directory code
  (DIR-1, DIR-5 to DIR-7) is kept, and the token lookup routes through it.
  The release's one new Enterprise snippet was the body of
  get_directory_for_issuer, which stays returning None: a token naming no
  address gets the server default, as DIR-2 specifies and as v0.16.22 did.
- crates/common/src/manager/application.rs: upstream's rewrite of the tests,
  with the temp directory names renamed again, and the 5(a) notice the
  name-purge change should have added.
- crates/common/src/network/mta.rs: both sides' imports.
- crates/main/Cargo.toml: the AGPL-only license kept, version 0.16.23.
- Cargo.lock: upstream's, with the fork's crates added by Cargo.

crates/smtp/Cargo.toml

@@ -7,7 +7,7 @@ homepage = "https://stalw.art/smtp"
 keywords = ["smtp", "email", "mail", "server"]
 categories = ["email"]
 license = "AGPL-3.0-only OR LicenseRef-SEL"
-version = "0.16.22"
+version = "0.16.23"
 edition = "2024"
 
 [dependencies]

crates/smtp/src/core/mod.rs

@@ -11,6 +11,7 @@ use common::{
     config::smtp::auth::VerifyStrategy,
     network::{ServerInstance, asn::AsnGeoLookupResult},
 };
+use email::message::delivery::ORCPT_ADDR_TYPE;
 use mail_auth::{IprevOutput, SpfOutput};
 use smtp_proto::request::receiver::{
     BdatReceiver, DataReceiver, DummyDataReceiver, DummyLineReceiver, LineReceiver, RequestReceiver,
@@ -306,10 +307,13 @@ impl SessionAddress {
         }
     }
 
-    pub fn report_address(&self) -> &str {
+    pub fn orig_address(&self) -> &str {
+        self.dsn_info.as_deref().unwrap_or(&self.address_lcase)
+    }
+
+    pub fn orcpt_parameter(&self) -> Option<String> {
         self.dsn_info
-            .as_ref()
-            .and_then(|v| v.strip_prefix("rfc822;"))
-            .unwrap_or(&self.address_lcase)
+            .as_deref()
+            .map(|orcpt| format!("{ORCPT_ADDR_TYPE}{}", orcpt.to_lowercase()))
     }
 }

crates/smtp/src/inbound/data.rs

@@ -443,7 +443,7 @@ impl<T: SessionStream> Session<T> {
             if !rc.analysis.forward {
                 self.data
                     .rcpt_to
-                    .retain(|rcpt| !rc.analysis.is_report_address(rcpt.report_address()));
+                    .retain(|rcpt| !rc.analysis.is_report_address(rcpt.orig_address()));
             }
 
             if self.data.rcpt_to.is_empty() {

crates/smtp/src/inbound/rcpt.rs

@@ -202,8 +202,8 @@ impl<T: SessionStream> Session<T> {
                 let mut new_addr = SessionAddress::new(address);
 
                 if !self.data.rcpt_to.contains(&new_addr) {
-                    new_addr.dsn_info = format!("rfc822;{}", orig_addr.address_lcase).into();
                     new_addr.flags = orig_addr.flags;
+                    new_addr.dsn_info = orig_addr.address_lcase.into();
                     self.data.rcpt_to.push(new_addr);
                 } else {
                     trc::event!(
@@ -353,7 +353,6 @@ impl<T: SessionStream> Session<T> {
         // Expand list
         if let Some(members) = rcpt_members {
             let list_addr = self.data.rcpt_to.pop().unwrap();
-            let orcpt = format!("rfc822;{}", list_addr.address_lcase);
             for member in members.as_ref() {
                 let member_lcase = member.to_lowercase();
                 let is_local = match self
@@ -399,7 +398,7 @@ impl<T: SessionStream> Session<T> {
                 if !self.data.rcpt_to.contains(&member_addr)
                     && member_addr.address_lcase != list_addr.address_lcase
                 {
-                    member_addr.dsn_info = orcpt.clone().into();
+                    member_addr.dsn_info = list_addr.address_lcase.clone().into();
                     member_addr.flags = list_addr.flags;
                     self.data.rcpt_to.push(member_addr);
                 }

crates/smtp/src/inbound/spam.rs

@@ -89,17 +89,7 @@ impl<T: SessionStream> Session<T> {
                 .iter()
                 .map(|r| r.address_lcase.as_str())
                 .collect(),
-            env_rcpt_orig_to: self
-                .data
-                .rcpt_to
-                .iter()
-                .map(|r| {
-                    r.dsn_info
-                        .as_deref()
-                        .and_then(|info| info.strip_prefix("rfc822;"))
-                        .unwrap_or(r.address_lcase.as_str())
-                })
-                .collect(),
+            env_rcpt_orig_to: self.data.rcpt_to.iter().map(|r| r.orig_address()).collect(),
             is_test: false,
             is_train: false,
         }

crates/smtp/src/outbound/delivery.rs

@@ -15,8 +15,8 @@ use crate::outbound::lookup::{DnsLookup, SourceIp};
 use crate::outbound::mta_sts::lookup::MtaStsLookup;
 use crate::outbound::mta_sts::verify::VerifyPolicy;
 use crate::outbound::{client::StartTlsResult, dane::verify::TlsaVerify};
-use crate::queue::dsn::SendDsn;
-use crate::queue::spool::SmtpSpool;
+use crate::queue::dsn::{DsnStatus, SendDsn};
+use crate::queue::spool::{DSN_RETRY, SmtpSpool};
 use crate::queue::throttle::IsAllowed;
 use crate::queue::{
     Error, FROM_REPORT, HostResponse, MessageWrapper, Metadata, QueueEnvelope, QueuedMessage,
@@ -155,7 +155,7 @@ impl QueuedMessage {
         let span_id = message.span_id;
 
         // Send any due Delivery Status Notifications
-        server.send_dsn(&mut message).await;
+        let dsn_status = server.send_dsn(&mut message).await;
 
         match has_pending_delivery {
             PendingDelivery::Yes(true)
@@ -163,21 +163,27 @@ impl QueuedMessage {
                     .message
                     .next_delivery_event(self.queue_name.into())
                     .is_some_and(|due| due <= now()) => {}
-            PendingDelivery::No => {
+            PendingDelivery::No if dsn_status == DsnStatus::Completed => {
                 trc::event!(
                     Delivery(DeliveryEvent::Completed),
                     SpanId = span_id,
                     Elapsed = trc::Value::Duration((now() - message.message.created) * 1000)
                 );
 
-                // All message recipients expired, do not re-queue. (DSN has been already sent)
+                // All message recipients expired, do not re-queue.
                 message.remove(&server, self.due.into()).await;
 
                 return QueueEventStatus::Completed;
             }
+            PendingDelivery::No => {
+                message
+                    .save_changes(&server, self.due.into(), Some(now() + DSN_RETRY))
+                    .await;
+                return QueueEventStatus::Deferred;
+            }
             _ => {
                 // Re-queue the message if its not yet due for delivery
-                message.save_changes(&server, self.due.into()).await;
+                message.save_changes(&server, self.due.into(), None).await;
                 return QueueEventStatus::Deferred;
             }
         }
@@ -208,7 +214,7 @@ impl QueuedMessage {
                     }
                 }
 
-                message.save_changes(&server, self.due.into()).await;
+                message.save_changes(&server, self.due.into(), None).await;
 
                 return QueueEventStatus::Deferred;
             }
@@ -1485,7 +1491,7 @@ impl QueuedMessage {
         }
 
         // Send Delivery Status Notifications
-        server.send_dsn(&mut message).await;
+        let dsn_status = server.send_dsn(&mut message).await;
 
         // Notify queue manager
         if message.message.next_event(None).is_some() {
@@ -1501,7 +1507,13 @@ impl QueuedMessage {
             );
 
             // Save changes to disk
-            message.save_changes(&server, self.due.into()).await;
+            message.save_changes(&server, self.due.into(), None).await;
+
+            QueueEventStatus::Deferred
+        } else if dsn_status == DsnStatus::Deferred {
+            message
+                .save_changes(&server, self.due.into(), Some(now() + DSN_RETRY))
+                .await;
 
             QueueEventStatus::Deferred
         } else {

crates/smtp/src/outbound/local.rs

@@ -120,7 +120,7 @@ impl MessageWrapper {
                     )
                     .await;
 
-                message
+                let _ = message
                     .queue(
                         QueueParams::new(&autogenerated.message, self.span_id, server)
                             .with_dkim_signers(dkim_signers)

crates/smtp/src/queue/dsn.rs

@@ -10,9 +10,10 @@ use super::{
     Recipient, Status,
 };
 use crate::inbound::dkim::DkimSign;
-use crate::queue::spool::QueueParams;
+use crate::queue::spool::{DSN_RETRY, QueueParams};
 use crate::queue::{MessageWrapper, UnexpectedResponse};
 use common::Server;
+use email::message::delivery::ORCPT_ADDR_TYPE;
 use mail_builder::MessageBuilder;
 use mail_builder::headers::HeaderType;
 use mail_builder::headers::content_type::ContentType;
@@ -25,16 +26,24 @@ use std::fmt::Write;
 use std::future::Future;
 use store::write::now;
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DsnStatus {
+    Completed,
+    Deferred,
+}
+
 pub trait SendDsn: Sync + Send {
-    fn send_dsn(&self, message: &mut MessageWrapper) -> impl Future<Output = ()> + Send;
+    fn send_dsn(&self, message: &mut MessageWrapper) -> impl Future<Output = DsnStatus> + Send;
     fn log_dsn(&self, message: &MessageWrapper) -> impl Future<Output = ()> + Send;
 }
 
 impl SendDsn for Server {
-    async fn send_dsn(&self, message: &mut MessageWrapper) {
+    async fn send_dsn(&self, message: &mut MessageWrapper) -> DsnStatus {
         // Send DSN events
         self.log_dsn(message).await;
 
+        let mut status = DsnStatus::Completed;
+
         if !message.message.return_path.is_empty() {
             // Build DSN
             if let Some(dsn) = message.build_dsn(self).await {
@@ -51,12 +60,19 @@ impl SendDsn for Server {
                         message.span_id,
                     )
                     .await;
-                dsn_message
+                if dsn_message
                     .queue(
                         QueueParams::new(&dsn, message.span_id, self)
                             .with_dkim_signers(dkim_signers),
                     )
-                    .await;
+                    .await
+                {
+                    message.mark_dsn_sent();
+                } else {
+                    status = DsnStatus::Deferred;
+                }
+            } else {
+                message.mark_dsn_sent();
             }
         } else {
             // Handle double bounce
@@ -64,7 +80,9 @@ impl SendDsn for Server {
         }
 
         // Update next DSN notify times
-        message.update_next_dsn(self).await;
+        message.update_next_dsn(self, status).await;
+
+        status
     }
 
     async fn log_dsn(&self, message: &MessageWrapper) {
@@ -132,7 +150,7 @@ impl SendDsn for Server {
 const MAX_HEADER_SIZE: usize = 4096;
 
 impl MessageWrapper {
-    pub async fn build_dsn(&mut self, server: &Server) -> Option<Vec<u8>> {
+    pub async fn build_dsn(&self, server: &Server) -> Option<Vec<u8>> {
         let config = &server.core.smtp.queue;
         let now = now();
 
@@ -141,13 +159,12 @@ impl MessageWrapper {
         let mut txt_failed = String::new();
         let mut dsn = String::new();
 
-        for rcpt in &mut self.message.recipients {
+        for rcpt in &self.message.recipients {
             if rcpt.has_flag(RCPT_DSN_SENT | RCPT_NOTIFY_NEVER) {
                 continue;
             }
             match &rcpt.status {
                 Status::Completed(response) => {
-                    rcpt.flags |= RCPT_DSN_SENT;
                     if !rcpt.has_flag(RCPT_NOTIFY_SUCCESS) {
                         continue;
                     }
@@ -164,7 +181,6 @@ impl MessageWrapper {
                     response.write_dsn_text(&rcpt.address, &mut txt_delay);
                 }
                 Status::PermanentFailure(response) => {
-                    rcpt.flags |= RCPT_DSN_SENT;
                     if !rcpt.has_flag(RCPT_NOTIFY_FAILURE) {
                         continue;
                     }
@@ -357,7 +373,7 @@ impl MessageWrapper {
             .into()
     }
 
-    pub async fn update_next_dsn(&mut self, server: &Server) {
+    pub async fn update_next_dsn(&mut self, server: &Server, status: DsnStatus) {
         let now = now();
         let mut notify_changes = Vec::new();
         for (rcpt_idx, rcpt) in self.message.recipients.iter().enumerate() {
@@ -366,6 +382,11 @@ impl MessageWrapper {
                 Status::TemporaryFailure(_) | Status::Scheduled
             ) && rcpt.notify.due <= now
             {
+                if status == DsnStatus::Deferred {
+                    notify_changes.push((rcpt_idx, 0, now + DSN_RETRY));
+                    continue;
+                }
+
                 let envelope = QueueEnvelope::new(&self.message, rcpt);
 
                 let queue_id = server
@@ -391,6 +412,19 @@ impl MessageWrapper {
         }
     }
 
+    fn mark_dsn_sent(&mut self) {
+        for rcpt in &mut self.message.recipients {
+            if !rcpt.has_flag(RCPT_DSN_SENT | RCPT_NOTIFY_NEVER)
+                && matches!(
+                    rcpt.status,
+                    Status::Completed(_) | Status::PermanentFailure(_)
+                )
+            {
+                rcpt.flags |= RCPT_DSN_SENT;
+            }
+        }
+    }
+
     fn handle_double_bounce(&mut self) {
         let mut is_double_bounce = Vec::with_capacity(0);
         let now = now();
@@ -523,7 +557,7 @@ impl Message {
 impl Recipient {
     fn write_dsn(&self, dsn: &mut String) {
         if let Some(orcpt) = &self.orcpt {
-            let _ = write!(dsn, "Original-Recipient: rfc822;{orcpt}\r\n");
+            let _ = write!(dsn, "Original-Recipient: {ORCPT_ADDR_TYPE}{orcpt}\r\n");
         }
         let _ = write!(dsn, "Final-Recipient: rfc822;{}\r\n", self.address);
     }

crates/smtp/src/queue/spool.rs

@@ -45,6 +45,7 @@ use utils::DomainPart;
 
 pub const LOCK_EXPIRY: u64 = 10 * 60; // 10 minutes
 pub const QUEUE_REFRESH: u64 = 5 * 60; // 5 minutes
+pub const DSN_RETRY: u64 = 5 * 60; // 5 minutes
 pub(crate) const INFINITE_LOCK: u64 = 60 * 60 * 24 * 365; // 1 year
 const CANDIDATE_OVERSCAN: usize = 4;
 const MAX_PREALLOCATED_CANDIDATES: usize = 1024;
@@ -370,6 +371,7 @@ pub(crate) struct QueueParams<'x, 'y> {
 }
 
 impl MessageWrapper {
+    #[must_use]
     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() {
@@ -669,7 +671,12 @@ impl MessageWrapper {
         recipient.queue = queue.virtual_queue;
     }
 
-    pub async fn save_changes(mut self, server: &Server, prev_event: Option<u64>) -> bool {
+    pub async fn save_changes(
+        mut self,
+        server: &Server,
+        prev_event: Option<u64>,
+        retry_at: Option<u64>,
+    ) -> bool {
         // Release quota for completed deliveries
         let mut batch = BatchBuilder::new();
         self.release_quota(&mut batch);
@@ -684,7 +691,12 @@ impl MessageWrapper {
                 },
             )));
         }
-        for (queue_name, due) in self.message.next_events() {
+        let mut next_events = self.message.next_events();
+        if let Some(retry_at) = retry_at {
+            let due = next_events.entry(self.queue_name).or_insert(retry_at);
+            *due = std::cmp::min(*due, retry_at);
+        }
+        for (queue_name, due) in next_events {
             batch.set(
                 ValueClass::Queue(QueueClass::MessageEvent(store::write::QueueEvent {
                     due,

crates/smtp/src/reporting/dmarc.rs

@@ -316,9 +316,6 @@ impl<T: SessionStream> Session<T> {
         if let Some(dkim2_output) = dkim2_output {
             report_record = report_record.with_dkim2_output(dkim2_output);
         }
-        if let Some(spf_ehlo) = &self.data.spf_ehlo {
-            report_record = report_record.with_spf_output(spf_ehlo, SPFDomainScope::Helo);
-        }
         if let Some(spf_mail_from) = &self.data.spf_mail_from {
             report_record = report_record.with_spf_output(spf_mail_from, SPFDomainScope::MailFrom);
         }

crates/smtp/src/reporting/inbound.rs

@@ -41,7 +41,7 @@ impl<T: AsyncWrite + AsyncRead + Unpin> Session<T> {
         self.data
             .rcpt_to
             .iter()
-            .any(|addr| analysis.is_report_address(addr.report_address()))
+            .any(|addr| analysis.is_report_address(addr.orig_address()))
     }
 }
 

crates/smtp/src/reporting/send.rs

@@ -98,7 +98,7 @@ impl MtaReportSend for Server {
         let dkim_signers = self
             .eval_signers(sign_config, &message.message, parent_session_id)
             .await;
-        message
+        let _ = message
             .queue(
                 QueueParams::new(&report, parent_session_id, self).with_dkim_signers(dkim_signers),
             )
@@ -130,7 +130,7 @@ impl MtaReportSend for Server {
         } else {
             None
         };
-        message
+        let _ = message
             .queue(
                 QueueParams::new(&raw_message, parent_session_id, self)
                     .with_dkim_signers(dkim_signers),

crates/smtp/src/scripts/envelope.rs

@@ -12,6 +12,7 @@ use smtp_proto::{
 use utils::DomainPart;
 
 use crate::core::{SessionAddress, SessionData};
+use email::message::delivery::ORCPT_ADDR_TYPE;
 
 impl SessionData {
     pub fn apply_envelope_modification(&mut self, envelope: Envelope, value: String) {
@@ -111,7 +112,11 @@ impl SessionData {
             }
             Envelope::Orcpt => {
                 if let Some(rcpt_to) = self.rcpt_to.last_mut() {
-                    rcpt_to.dsn_info = value.into();
+                    rcpt_to.dsn_info = value
+                        .strip_prefix(ORCPT_ADDR_TYPE)
+                        .map(str::to_string)
+                        .unwrap_or(value)
+                        .into();
                 }
             }
             Envelope::Envid => {

crates/smtp/src/scripts/event_loop.rs

@@ -297,7 +297,7 @@ impl RunScript for Server {
                                     None
                                 };
 
-                                message
+                                let _ = message
                                     .queue(
                                         QueueParams::new(raw_message, session_id, self)
                                             .with_dkim_signers(dkim_signers)

crates/smtp/src/scripts/exec.rs

@@ -95,10 +95,8 @@ impl<T: SessionStream> Session<T> {
                     params
                         .envelope
                         .push((Envelope::To, rcpt.address_lcase.to_string().into()));
-                    if let Some(orcpt) = &rcpt.dsn_info {
-                        params
-                            .envelope
-                            .push((Envelope::Orcpt, orcpt.as_str().to_lowercase().into()));
+                    if let Some(orcpt) = rcpt.orcpt_parameter() {
+                        params.envelope.push((Envelope::Orcpt, orcpt.into()));
                     }
                 }
             } else {
@@ -109,10 +107,10 @@ impl<T: SessionStream> Session<T> {
 
                 for rcpt in &self.data.rcpt_to {
                     recipients.push(Variable::from(rcpt.address_lcase.to_string()));
-                    orcpts.push(match &rcpt.dsn_info {
+                    orcpts.push(match rcpt.orcpt_parameter() {
                         Some(orcpt) => {
                             has_orcpts = true;
-                            Variable::from(orcpt.as_str().to_lowercase())
+                            Variable::from(orcpt)
                         }
                         None => Variable::default(),
                     });