From 7ea7904ea093b1f213425bff59749d3375542d58 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jagoda=20=C5=9Al=C4=85zak?= Date: Tue, 12 May 2026 07:13:20 +0200 Subject: [PATCH] feat(transport): Destination worker pool MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements a per-destination worker pool, so that connections to the same destination are not parallelized, but instead queued. If a queue is full, new messages are immediately deferred, before mail data is sent from postfix. Closes: #141 Signed-off-by: Jagoda Ślązak --- filtermail/src/http_server.rs | 50 ++- filtermail/src/inbound.rs | 49 +- filtermail/src/main.rs | 3 +- filtermail/src/outbound.rs | 62 ++- filtermail/src/smtp_responses.rs | 7 + filtermail/src/smtp_server.rs | 140 +++--- filtermail/src/transport.rs | 540 ++++++++--------------- filtermail/src/transport/https_client.rs | 57 +++ filtermail/src/transport/worker.rs | 447 +++++++++++++++++++ 9 files changed, 881 insertions(+), 474 deletions(-) create mode 100644 filtermail/src/smtp_responses.rs create mode 100644 filtermail/src/transport/https_client.rs create mode 100644 filtermail/src/transport/worker.rs diff --git a/filtermail/src/http_server.rs b/filtermail/src/http_server.rs index 4335976c..d9115398 100644 --- a/filtermail/src/http_server.rs +++ b/filtermail/src/http_server.rs @@ -1,4 +1,4 @@ -use crate::smtp_server::{Envelope, SmtpHandler}; +use crate::smtp_server::{SmtpHandler, Transaction}; use http_body_util::combinators::BoxBody; use http_body_util::{BodyExt, Full}; use hyper::body::{Bytes, Incoming}; @@ -93,6 +93,8 @@ impl Service> for MxDelivService )?); } + let mut transaction = Transaction::default(); + let mail_from = req .headers() .get(crate::transport::HEADER_MAIL_FROM) @@ -100,16 +102,14 @@ impl Service> for MxDelivService .unwrap_or("") .to_string(); - match handler.handle_mail(&mail_from) { - Ok(_) => {} - Err(e) => { - return Ok(Response::builder() - .status(400) - .body(Full::new(Bytes::from(e)).boxed())?); - } - }; + if let Err(e) = handler.handle_mail_from(&mail_from) { + return Ok(Response::builder() + .status(400) + .body(Full::new(Bytes::from(e)).boxed())?); + } + transaction.envelope.mail_from = mail_from; - let rcpt_to = req + let rcpt_to: Vec = req .headers() .get_all(crate::transport::HEADER_RCPT_TO) .iter() @@ -117,6 +117,21 @@ impl Service> for MxDelivService .map(ToString::to_string) .collect(); + for r in &rcpt_to { + if let Err(e) = handler.handle_rcpt_to(r, &mut transaction) { + return Ok(Response::builder() + .status(400) + .body(Full::new(Bytes::from(e)).boxed())?); + } + } + transaction.envelope.rcpt_to = rcpt_to; + + if let Err(e) = handler.handle_data_start(&transaction) { + return Ok(Response::builder() + .status(400) + .body(Full::new(Bytes::from(e)).boxed())?); + } + let body_limited = http_body_util::Limited::new(req.into_body(), max_size); let body_bytes = match body_limited.collect().await { Ok(body) => body.to_bytes(), @@ -127,24 +142,19 @@ impl Service> for MxDelivService } }; - let mut envelope = Envelope { - origin_ip: "".to_string(), - mail_from, - rcpt_to, - data: body_bytes.to_vec(), - }; + transaction.envelope.data = body_bytes.to_vec(); - log::debug!("(HTTP) MAIL FROM:<{}>", envelope.mail_from); - for rcpt in &envelope.rcpt_to { + log::debug!("(HTTP) MAIL FROM:<{}>", transaction.envelope.mail_from); + for rcpt in &transaction.envelope.rcpt_to { log::debug!("(HTTP) RCPT TO:<{}>", rcpt); } log::trace!( "(HTTP) DATA:\n{:?}", - String::from_utf8_lossy(&envelope.data) + String::from_utf8_lossy(&transaction.envelope.data) ); - match handler.handle_data(&mut envelope).await { + match handler.handle_data_dot(&mut transaction).await { Ok(response) => Ok(Response::builder() .status(200) .body(Full::new(Bytes::from(response)).boxed())?), diff --git a/filtermail/src/inbound.rs b/filtermail/src/inbound.rs index dc9dbf7b..9328e58d 100644 --- a/filtermail/src/inbound.rs +++ b/filtermail/src/inbound.rs @@ -1,12 +1,12 @@ //! Module for handling incoming SMTP messages. -use crate::ENCRYPTION_NEEDED_523; use crate::config::Config; use crate::dkim_verifier::DkimVerifier; use crate::message::{check_encrypted, is_securejoin}; use crate::smtp_client::SmtpConnectionPool; +use crate::smtp_responses::ENCRYPTION_NEEDED_523; pub use crate::smtp_server::Envelope; -use crate::smtp_server::SmtpHandler; +use crate::smtp_server::{SmtpHandler, Transaction}; use crate::utils::{AddressDomain, build_resolver, extract_address, log_eml}; use async_trait::async_trait; use hickory_resolver::TokioResolver; @@ -69,12 +69,10 @@ impl IncomingBeforeQueueHandler { #[async_trait] impl SmtpHandler for IncomingBeforeQueueHandler { - fn handle_mail(&self, _address: &str) -> Result<(), String> { - Ok(()) - } + type State = (); - async fn check_data(&self, envelope: &mut Envelope) -> Result<(), String> { - let message = match parse_mail(&envelope.data) { + async fn check_data(&self, transaction: &mut Transaction) -> Result<(), String> { + let message = match parse_mail(&transaction.envelope.data) { Ok(m) => m, Err(e) => return Err(format!("500 Failed to parse message: {}", e)), }; @@ -92,16 +90,21 @@ impl SmtpHandler for IncomingBeforeQueueHandler { log::debug!("Processing DATA message from {from_addr}"); - if !envelope.mail_from.eq_ignore_ascii_case(&from_addr) { + if !transaction + .envelope + .mail_from + .eq_ignore_ascii_case(&from_addr) + { // If the MAIL FROM doesn't match the From header, we do not reject the mail, // as this can be caused by e.g. SRS forwarding. // Instead, we reset the envelope address, so it is reinjected as // `MAIL FROM:<>` to prevent sending a bounce message. // - envelope.mail_from = String::new(); + transaction.envelope.mail_from = String::new(); } - envelope.rcpt_to = envelope + transaction.envelope.rcpt_to = transaction + .envelope .rcpt_to .iter() .filter(|s| { @@ -121,7 +124,7 @@ impl SmtpHandler for IncomingBeforeQueueHandler { // Allow encrypted or securejoin messages if mail_encrypted || is_securejoin(&message) { log::info!("Incoming: Filtering encrypted mail."); - return self.verify_origin(envelope, &from_addr).await; + return self.verify_origin(&transaction.envelope, &from_addr).await; } log::info!("Incoming: Filtering unencrypted mail."); @@ -132,26 +135,26 @@ impl SmtpHandler for IncomingBeforeQueueHandler { && from_addr.to_lowercase().starts_with("mailer-daemon@") && message.ctype.mimetype == "multipart/report" { - return self.verify_origin(envelope, &from_addr).await; + return self.verify_origin(&transaction.envelope, &from_addr).await; } - for recipient in &envelope.rcpt_to { + for recipient in &transaction.envelope.rcpt_to { if !self.config.is_cleartext_ok(recipient) { log::warn!("Rejected unencrypted mail from: {from_addr}"); return Err(ENCRYPTION_NEEDED_523.to_string()); } } - self.verify_origin(envelope, &from_addr).await + self.verify_origin(&transaction.envelope, &from_addr).await } - async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String> { + async fn reinject_mail(&self, transaction: &Transaction) -> Result<(), String> { log::debug!("Re-injecting the mail that passed checks"); let hostname = format!("[{}]", self.config.filtermail_host); crate::smtp_client::send( &self.config.postfix_host, self.config.postfix_reinject_port_incoming, - envelope, + &transaction.envelope, &hostname, None, self.dns_resolver.clone(), @@ -189,12 +192,14 @@ mod tests { config: Config, ) -> TestResult { let handler = IncomingBeforeQueueHandler::new(config, false)?; - let mut envelope = Envelope { - mail_from: address.to_string(), - origin_ip: "".to_string(), // Currently shouldn't be relevant. - data: eml.to_vec(), - rcpt_to: vec!["does.not.matter@example.org".to_string()], + let mut transaction = Transaction { + envelope: Envelope { + mail_from: address.to_string(), + data: eml.to_vec(), + rcpt_to: vec!["does.not.matter@example.org".to_string()], + }, + ..Default::default() }; - Ok(handler.check_data(&mut envelope).await?) + Ok(handler.check_data(&mut transaction).await?) } } diff --git a/filtermail/src/main.rs b/filtermail/src/main.rs index 165e8802..aa159242 100644 --- a/filtermail/src/main.rs +++ b/filtermail/src/main.rs @@ -33,6 +33,7 @@ pub(crate) mod message; pub(crate) mod openpgp; pub(crate) mod outbound; pub(crate) mod smtp_client; +mod smtp_responses; pub(crate) mod smtp_server; mod tls; mod transport; @@ -50,8 +51,6 @@ use std::process; use std::str::FromStr; use std::sync::Arc; -const ENCRYPTION_NEEDED_523: &str = "523 Encryption Needed: Invalid Unencrypted Mail"; - #[derive(Debug, Copy, Clone, PartialEq, Eq)] enum Mode { Outgoing, diff --git a/filtermail/src/outbound.rs b/filtermail/src/outbound.rs index cacf6e52..5fc96a42 100644 --- a/filtermail/src/outbound.rs +++ b/filtermail/src/outbound.rs @@ -1,11 +1,11 @@ //! Module for handling outgoing SMTP messages. -use crate::ENCRYPTION_NEEDED_523; use crate::config::Config; use crate::message::{check_encrypted, is_securejoin}; use crate::smtp_client::SmtpConnectionPool; -pub use crate::smtp_server::Envelope; -use crate::smtp_server::SmtpHandler; +use crate::smtp_responses::ENCRYPTION_NEEDED_523; +use crate::smtp_responses::OK_250; +use crate::smtp_server::{SmtpHandler, Transaction}; use crate::utils::{build_resolver, extract_address}; use async_trait::async_trait; use governor::clock::MonotonicClock; @@ -53,7 +53,9 @@ impl OutgoingBeforeQueueHandler { #[async_trait] impl SmtpHandler for OutgoingBeforeQueueHandler { - fn handle_mail(&self, address: &str) -> Result<(), String> { + type State = (); + + fn handle_mail_from(&self, address: &str) -> Result<(), String> { log::debug!("handle_MAIL from {address}"); let parts: Vec<&str> = address.split('@').collect(); @@ -79,8 +81,8 @@ impl SmtpHandler for OutgoingBeforeQueueHandler { Ok(()) } - async fn check_data(&self, envelope: &mut Envelope) -> Result<(), String> { - let message = match parse_mail(&envelope.data) { + async fn check_data(&self, transaction: &mut Transaction) -> Result<(), String> { + let message = match parse_mail(&transaction.envelope.data) { Ok(m) => m, Err(e) => return Err(format!("500 Failed to parse message: {}", e)), }; @@ -97,7 +99,8 @@ impl SmtpHandler for OutgoingBeforeQueueHandler { let from_addr = extract_address(&from_header) .ok_or(format!("500 Invalid FROM header: {from_header}"))?; - envelope.rcpt_to = envelope + transaction.envelope.rcpt_to = transaction + .envelope .rcpt_to .iter() .filter(|s| { @@ -113,12 +116,19 @@ impl SmtpHandler for OutgoingBeforeQueueHandler { // MAIL FROM is our source of truth for outbound messages, // as this address is checked by postfix against the username before sending it // to filtermail. - log::debug!("Processing DATA message from {}", envelope.mail_from); + log::debug!( + "Processing DATA message from {}", + transaction.envelope.mail_from + ); - if !envelope.mail_from.eq_ignore_ascii_case(&from_addr) { + if !transaction + .envelope + .mail_from + .eq_ignore_ascii_case(&from_addr) + { return Err(format!( "500 Invalid FROM <{}> for <{}>", - from_addr, envelope.mail_from + from_addr, transaction.envelope.mail_from )); } @@ -131,8 +141,8 @@ impl SmtpHandler for OutgoingBeforeQueueHandler { log::info!("Outgoing: Filtering unencrypted mail."); // Allow self-sent Autocrypt Setup Message - if envelope.rcpt_to.len() == 1 - && let Some(rcpt_to) = envelope.rcpt_to.first() + if transaction.envelope.rcpt_to.len() == 1 + && let Some(rcpt_to) = transaction.envelope.rcpt_to.first() && *rcpt_to == from_addr { let subject = message @@ -148,13 +158,13 @@ impl SmtpHandler for OutgoingBeforeQueueHandler { Err(ENCRYPTION_NEEDED_523.to_string()) } - async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String> { + async fn reinject_mail(&self, transaction: &Transaction) -> Result<(), String> { log::debug!("Re-injecting the mail that passed checks"); let hostname = format!("[{}]", self.config.filtermail_host); crate::smtp_client::send( &self.config.postfix_host, self.config.postfix_reinject_port, - envelope, + &transaction.envelope, &hostname, None, self.dns_resolver.clone(), @@ -169,21 +179,27 @@ impl SmtpHandler for OutgoingBeforeQueueHandler { Ok(()) } - async fn handle_data(&self, envelope: &mut Envelope) -> Result { + async fn handle_data_dot( + &self, + transaction: &mut Transaction, + ) -> Result { log::debug!("handle_DATA before-queue"); - self.check_data(envelope).await?; - if self.config.is_disabled(&envelope.mail_from) { - log::warn!("Dropping mail; Sender {} is disabled.", envelope.mail_from); - return Ok("250 OK".to_string()); + self.check_data(transaction).await?; + if self.config.is_disabled(&transaction.envelope.mail_from) { + log::warn!( + "Dropping mail; Sender {} is disabled.", + transaction.envelope.mail_from + ); + return Ok(OK_250.to_string()); } - if envelope.rcpt_to.is_empty() { + if transaction.envelope.rcpt_to.is_empty() { log::warn!("Dropping mail; All recipients disabled."); - return Ok("250 OK".to_string()); + return Ok(OK_250.to_string()); } - self.reinject_mail(envelope).await.map_err(|e| { + self.reinject_mail(transaction).await.map_err(|e| { log::warn!("Failed to reinject mail: {e}"); e })?; - Ok("250 OK".to_string()) + Ok(OK_250.to_string()) } } diff --git a/filtermail/src/smtp_responses.rs b/filtermail/src/smtp_responses.rs new file mode 100644 index 00000000..e12fba2a --- /dev/null +++ b/filtermail/src/smtp_responses.rs @@ -0,0 +1,7 @@ +pub const OK_250: &str = "250 OK"; +pub const OK_HTTPS_250: &str = "250 OK (HTTPS)"; +pub const OK_SMTP_250: &str = "250 OK (SMTP)"; + +pub const ENCRYPTION_NEEDED_523: &str = "523 Encryption Needed: Invalid Unencrypted Mail"; +pub const LOCAL_ERROR_451: &str = "451 Local error"; +pub const WORKER_BUSY_421: &str = "421 Worker for this destination is busy"; diff --git a/filtermail/src/smtp_server.rs b/filtermail/src/smtp_server.rs index f2ff8109..f1458caf 100644 --- a/filtermail/src/smtp_server.rs +++ b/filtermail/src/smtp_server.rs @@ -1,8 +1,10 @@ //! A simplified SMTP server implementation for internal communication. +use crate::smtp_responses::OK_250; use crate::utils::{extract_address, log_eml}; use async_trait::async_trait; use memchr::{Memchr, memmem}; +use std::fmt::Debug; use std::sync::Arc; use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}; @@ -12,9 +14,7 @@ use tokio::net::{TcpListener, TcpStream}; #[derive(Debug, Default, Clone)] pub struct Envelope { pub mail_from: String, - pub origin_ip: String, pub rcpt_to: Vec, - /// Mail data as transmitted over SMTP/LMTP. /// /// Described in . @@ -25,6 +25,16 @@ pub struct Envelope { pub data: Vec, } +/// Represent an ongoing SMTP transaction. +/// +/// Every new connection starts with an empty envelope and handler state. +/// A RSET command starts a new transaction, which clears the envelope and state. +#[derive(Debug, Default)] +pub struct Transaction { + pub envelope: Envelope, + pub state: S, +} + /// Checks if mail data is valid. fn is_valid_data(data: &[u8]) -> bool { // DATA must end with . @@ -59,19 +69,55 @@ fn is_valid_data(data: &[u8]) -> bool { /// Trait defining the SMTP handler interface. #[async_trait] pub trait SmtpHandler: Send + Sync { - /// Handles the MAIL FROM command. - fn handle_mail(&self, address: &str) -> Result<(), String>; + /// Transaction state type associated with this handler. + type State: Debug + Default + Send; /// Checks the DATA command before reinjection. /// /// Can optionally modify the envelope before reinjection. - async fn check_data(&self, envelope: &mut Envelope) -> Result<(), String>; + /// + /// Default implementation is no-op. + async fn check_data(&self, _transaction: &mut Transaction) -> Result<(), String> { + Ok(()) + } /// Reinjects the mail back to postfix. - async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String>; + /// + /// Default implementation is no-op. + async fn reinject_mail(&self, _transaction: &Transaction) -> Result<(), String> { + Ok(()) + } - /// Handles the DATA command. - async fn handle_data(&self, envelope: &mut Envelope) -> Result { + /// Handles the MAIL FROM command. + /// + /// Default implementation is no-op. + fn handle_mail_from(&self, _address: &str) -> Result<(), String> { + Ok(()) + } + + /// Handles the RCPT TO command. + /// + /// Default implementation is no-op. + fn handle_rcpt_to( + &self, + _address: &str, + _transaction: &mut Transaction, + ) -> Result<(), String> { + Ok(()) + } + + /// Handles the DATA command. Called after receiving DATA, before receiving actual data. + /// + /// Default implementation is no-op. + fn handle_data_start(&self, _transaction: &Transaction) -> Result<(), String> { + Ok(()) + } + + /// Handles the end of DATA command. Called after receiving the final dot. + async fn handle_data_dot( + &self, + transaction: &mut Transaction, + ) -> Result { log::debug!("handle_DATA before-queue"); // Check if the DATA is valid @@ -80,20 +126,20 @@ pub trait SmtpHandler: Send + Sync { // We are not going to normalize newlines // and escape the dots in the mail data. // If mail data turned out to be invalid, reject immediately. - if !is_valid_data(&envelope.data) { + if !is_valid_data(&transaction.envelope.data) { return Err("500 Invalid DATA".to_string()); } - self.check_data(envelope).await?; - if envelope.rcpt_to.is_empty() { + self.check_data(transaction).await?; + if transaction.envelope.rcpt_to.is_empty() { log::warn!("Dropping mail; All recipients disabled."); - return Ok("250 OK".to_string()); + return Ok(OK_250.to_string()); } - self.reinject_mail(envelope).await.map_err(|e| { + self.reinject_mail(transaction).await.map_err(|e| { log::warn!("Failed to reinject mail: {e}"); e })?; - Ok("250 OK".to_string()) + Ok("OK_250".to_string()) } } @@ -150,7 +196,7 @@ where writer.write_all(b"220 filtermail SMTP\r\n").await?; writer.flush().await?; - let mut envelope = Envelope::default(); + let mut transaction = Transaction::default(); 'connection: loop { line.clear(); @@ -181,28 +227,24 @@ where || cmd.to_uppercase().starts_with("LHLO") { writer - .write_all(b"250-filtermail\r\n250-XFORWARD ADDR\r\n250-8BITMIME\r\n250 OK\r\n") + .write_all(b"250-filtermail\r\n250-8BITMIME\r\n250 OK\r\n") .await?; writer.flush().await?; } else if cmd.to_uppercase().starts_with("MAIL FROM:<>") { // bounce message - envelope.mail_from = String::new(); - writer.write_all(b"250 OK\r\n").await?; + transaction.envelope.mail_from = String::new(); + writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?; writer.flush().await?; } else if cmd.to_uppercase().starts_with("MAIL FROM:") { if let Some(from) = extract_address(cmd) { - match handler.handle_mail(&from) { - Ok(_) => { - envelope.mail_from = from; - writer.write_all(b"250 OK\r\n").await?; - writer.flush().await?; - } - Err(e) => { - writer.write_all(format!("{}\r\n", e).as_bytes()).await?; - writer.flush().await?; - break 'connection; - } + if let Err(e) = handler.handle_mail_from(&from) { + writer.write_all(format!("{}\r\n", e).as_bytes()).await?; + writer.flush().await?; + continue 'connection; } + transaction.envelope.mail_from = from; + writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?; + writer.flush().await?; } else { log::warn!("Invalid MAIL FROM command. Can't extract address. Received: {cmd}"); writer @@ -212,11 +254,21 @@ where } } else if cmd.to_uppercase().starts_with("RCPT TO:") { if let Some(to) = extract_address(cmd) { - envelope.rcpt_to.push(to); - writer.write_all(b"250 OK\r\n").await?; + if let Err(e) = handler.handle_rcpt_to(&to, &mut transaction) { + writer.write_all(format!("{}\r\n", e).as_bytes()).await?; + writer.flush().await?; + continue 'connection; + } + transaction.envelope.rcpt_to.push(to); + writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?; writer.flush().await?; } } else if cmd.to_uppercase().starts_with("DATA") { + if let Err(e) = handler.handle_data_start(&transaction) { + writer.write_all(format!("{}\r\n", e).as_bytes()).await?; + writer.flush().await?; + continue 'connection; + } writer .write_all(b"354 End data with .\r\n") .await?; @@ -255,14 +307,14 @@ where .write_all(b"552 Message exceeds maximum size\r\n") .await?; writer.flush().await?; - break 'connection; + continue 'connection; } } - envelope.data = data; + transaction.envelope.data = data; // Process the message - match handler.handle_data(&mut envelope).await { + match handler.handle_data_dot(&mut transaction).await { Ok(response) => { log::debug!("Sent: {response}"); writer @@ -277,29 +329,17 @@ where } } - envelope = Envelope::default(); - } else if cmd.to_uppercase().starts_with("XFORWARD") { - // https://www.postfix.org/XFORWARD_README.html - if let Some(addr_part) = cmd - .split_whitespace() - .find(|part| part.to_uppercase().starts_with("ADDR=")) - && let Some(ip) = addr_part.strip_prefix("ADDR=") - { - let ip = ip.to_lowercase(); - envelope.origin_ip = ip.strip_prefix("ipv6:").unwrap_or(&ip).to_string(); - writer.write_all(b"250 OK\r\n").await?; - writer.flush().await?; - } + transaction = Transaction::default(); } else if cmd.to_uppercase().starts_with("QUIT") { writer.write_all(b"221 OK\r\n").await?; writer.flush().await?; break 'connection; } else if cmd.to_uppercase().starts_with("RSET") { - envelope = Envelope::default(); - writer.write_all(b"250 OK\r\n").await?; + transaction = Transaction::default(); + writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?; writer.flush().await?; } else if cmd.to_uppercase().starts_with("NOOP") { - writer.write_all(b"250 OK\r\n").await?; + writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?; writer.flush().await?; } else { writer.write_all(b"500 Command not recognized\r\n").await?; diff --git a/filtermail/src/transport.rs b/filtermail/src/transport.rs index 06fd4d62..0d02a46b 100644 --- a/filtermail/src/transport.rs +++ b/filtermail/src/transport.rs @@ -1,375 +1,112 @@ +mod https_client; +mod worker; + use crate::config::Config; -use crate::smtp_client::{SmtpConnectionPool, TlsConfig}; -use crate::smtp_server::{Envelope, SmtpHandler}; -use crate::tls; -use crate::utils::{AddressDomain, build_resolver}; +use crate::smtp_responses::{LOCAL_ERROR_451, WORKER_BUSY_421}; +use crate::smtp_server::{SmtpHandler, Transaction}; +use crate::utils::AddressDomain; use async_trait::async_trait; -use hickory_resolver::{TokioResolver, proto::rr::RData}; -use http_body_util::BodyExt; -use hyper::body::Bytes; -use hyper_rustls::HttpsConnector; -use hyper_util::client::legacy::connect::HttpConnector; use std::collections::BTreeMap; use std::str::FromStr; -use std::sync::Arc; -use std::time::Duration; -use tokio::task::{JoinHandle, JoinSet}; -use tokio_rustls::rustls; +use tokio::sync::mpsc::OwnedPermit; +use tokio::task::JoinSet; +use worker::{WorkerMessage, WorkerPool}; pub const HEADER_MAIL_FROM: &str = "X-MAIL-FROM"; pub const HEADER_RCPT_TO: &str = "X-MAIL-TO"; -/// Cheaply clonable HTTPS client. -/// -/// Holds regular secure variant and relaxed - without certificate verification. -/// -/// Connection pool handled internally by [`hyper_util::client::legacy::Client`]. -#[derive(Clone)] -struct HttpsClient { - pub secure: hyper_util::client::legacy::Client< - HttpsConnector, - http_body_util::Full, - >, - pub relaxed: hyper_util::client::legacy::Client< - HttpsConnector, - http_body_util::Full, - >, -} - -impl HttpsClient { - /// Creates a new `[HttpsClient]`. - pub fn new( - tls_resumption_store: Arc, - ) -> Result { - let tls_client_config = tls::configure_rustls(tls_resumption_store.clone(), false)?; - let https_connector = hyper_rustls::HttpsConnectorBuilder::new() - .with_tls_config(tls_client_config) - .https_only() - .enable_http1() - .enable_http2() - .build(); - let https_client = - hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) - .build(https_connector); - - let tls_client_config_relaxed = tls::configure_rustls(tls_resumption_store, true)?; - let https_connector_relaxed = hyper_rustls::HttpsConnectorBuilder::new() - .with_tls_config(tls_client_config_relaxed) - .https_only() - .enable_http1() - .enable_http2() - .build(); - let https_client_relaxed = - hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) - .build(https_connector_relaxed); - - Ok(Self { - secure: https_client, - relaxed: https_client_relaxed, - }) - } -} - pub struct TransportHandler { - config: Config, - dns_resolver: Arc, - tls_resumption_store: Arc, - smtp_connection_pool: Arc, - https_client: HttpsClient, - mxdeliv_unsupported_hosts: Arc>, - monitor_handle: JoinHandle<()>, + workers: WorkerPool, } impl TransportHandler { + /// Creates a new [`TransportHandler`]. pub fn new(config: Config) -> Result { - let dns_resolver = Arc::new(build_resolver()?); - let tls_resumption_store = Arc::new(rustls::client::ClientSessionMemoryCache::new(256)); - let https_client = HttpsClient::new(tls_resumption_store.clone())?; + let workers = WorkerPool::new(config)?; - let mxdeliv_cache = Arc::new(retainer::Cache::new()); - let mxdeliv_cache_clone = mxdeliv_cache.clone(); - - let monitor_handle = tokio::spawn(async move { - mxdeliv_cache_clone - .monitor(4, 0.25, Duration::from_secs(10)) - .await - }); - - Ok(Self { - config, - dns_resolver, - tls_resumption_store, - smtp_connection_pool: SmtpConnectionPool::new(), - https_client, - mxdeliv_unsupported_hosts: mxdeliv_cache, - monitor_handle, - }) + Ok(Self { workers }) } - /// Handles a single email transaction for a single recipient domain. - #[expect(clippy::too_many_arguments)] - async fn handle_single_domain( - tls_resumption_store: Arc, - smtp_connection_pool: Arc, - mxdeliv_unsupported_hosts: Arc>, - https_client: HttpsClient, - dns_resolver: Arc, - domain: AddressDomain, - envelope: Envelope, - client_hostname: String, - ) -> Result { - let mut allow_invalid_cert = false; - let mut skip_tls = false; // only respected by smtp channel - - let mx_hosts = match domain { - // no-DNS setup; assume the ip from email address is the destination. - AddressDomain::Literal(ip) => { - // We allow self-signed certs on IP-based relays. - allow_invalid_cert = true; - vec![(0, ip)] - } - AddressDomain::Name(mx_domain) => { - if mx_domain.eq_ignore_ascii_case("nauta.cu") { - // Special case; We don't want to defederate nauta.cu, - // which doesn't support STARTTLS at all. - skip_tls = true; - } else if mx_domain.starts_with('_') { - // We use domains starting with `_` for test deployments. - // (You can't request a non-wildcard cert for such domain) - allow_invalid_cert = true; - } - let query = format!("{mx_domain}."); - - match dns_resolver.mx_lookup(query).await { - Ok(mx_records) => { - let mut hosts: Vec<(u16, String)> = Vec::new(); - for mx_record in mx_records.answers() { - let mx = match mx_record.data { - RData::MX(ref mx) => mx, - _ => continue, - }; - - // Null MX / RFC7505 - if mx.exchange.is_root() { - // From RFC7505 section 3: - // > A domain that advertises a null MX MUST NOT - // > advertise any other MX RR. - // We assume this is the only record and exit early. - return Err( - "556 5.1.10 Permanent failure: Recipient address has null MX" - .to_string(), - ); - } - - let host = mx.exchange.to_string().trim_end_matches('.').to_string(); - hosts.push((mx.preference, host)) - } - hosts.sort(); - hosts - } - Err(e) => { - if e.is_no_records_found() { - // "implicit MX" as described by section 5.1 of RFC5321 - // https://datatracker.ietf.org/doc/html/rfc5321#section-5.1 - log::debug!("No MX record found, using implicit MX: {mx_domain}"); - vec![(0, mx_domain)] - } else if e.is_nx_domain() { - return Err(format!("512 Domain {mx_domain} does not exist")); - } else { - return Err(format!("421 DNS resolution failed for {mx_domain}")); - } - } - } - } - }; - - let tls_config = match skip_tls { - true => None, - false => Some(TlsConfig { - allow_invalid_cert, - session_cache: tls_resumption_store, - }), - }; - - let mut last_error = None; - - // we try sequentially in order of MX preference, - // but the IPv4 and IPv6 connections (after `smtp_client::send` resolves mx hostname) - // happens in parallel. - 'try_relay: for (_, mx_host) in mx_hosts { - let skip_mxdeliv = mxdeliv_unsupported_hosts - .get(&mx_host) - .await - .map(|guard| *guard.value()) - .unwrap_or(false); - - // HTTPS channel - if skip_mxdeliv { - log::debug!("Skipping HTTP delivery to host that failed recently: {mx_host}"); - } else { - match Self::https_delivery( - https_client.clone(), - mx_host.clone(), - &envelope, - allow_invalid_cert, - ) - .await - { - Ok(_) => { - return Ok("250 Ok (HTTPS)".to_string()); - } - Err(e) => { - log::debug!("HTTPS delivery to {mx_host} failed: {e}"); - } - } - } - - // SMTP channel (fallback) - match crate::smtp_client::send( - &mx_host, - 25, - &envelope, - &client_hostname, - tls_config.clone(), - dns_resolver.clone(), - smtp_connection_pool.clone(), - ) - .await - { - Ok(_) => { - // Switches this host to SMTP for 30 minutes. - // Note: this MUST happen only after a successful SMTP delivery, - // or otherwise any http error will lock us out of any way to - // deliver to a relay with a blocked port 25 for 30 minutes. - mxdeliv_unsupported_hosts - .insert(mx_host.clone(), true, Duration::from_mins(30)) - .await; - return Ok("250 Ok (SMTP)".to_string()); - } - Err(error) => { - match &error { - // We only want to try other MX hosts if we encounter a problem - // related to connection. - // (So we don't spam other servers if the message is actually rejected.) - crate::error::Error::Io(_) - | crate::error::Error::ConnectionFailed(_) - | crate::error::Error::Tls(_) => { - // Make sure we quickly retry HTTP if SMTP failed to connect - mxdeliv_unsupported_hosts.remove(&mx_host).await; - log::warn!( - "Connection error relaying to mail server {mx_host}: {error}" - ); - last_error = Some((error.smtp_response(), mx_host.clone())); - continue 'try_relay; - } - crate::error::Error::MailSend { .. } => { - log::warn!("Message rejected by mail server {mx_host}: {error}"); - return Err(error.smtp_response()); - } - _ => { - log::warn!( - "Unexpected error while delivering to mail server {mx_host}: {error}" - ); - return Err(format!( - "{} (while attempting delivery to {mx_host})", - error.smtp_response() - )); - } - } - } - } - } - - let (error, mx_host) = last_error.unwrap_or(("?".to_string(), "?".to_string())); - Err(format!( - "421 Failed to connect to any mail server; last attempt to {mx_host}: {error}" - )) - } - - /// Performs mail delivery to `mx_host` over HTTPS. + /// Same as [`Self::new`], but lets you set worker queue size. /// - /// Times out after 60s. - async fn https_delivery( - https_client: HttpsClient, - mx_host: String, - envelope: &Envelope, - allow_invalid_cert: bool, - ) -> Result<(), crate::error::Error> { - let request: hyper::Request> = { - let mut builder = hyper::Request::builder() - .method(hyper::Method::POST) - .uri(format!("https://{mx_host}/mxdeliv")); + /// Only used for tests. + #[cfg(test)] + pub fn with_queue_size(config: Config, queue_size: usize) -> Result { + let workers = WorkerPool::with_queue_size(config, queue_size)?; - if !envelope.mail_from.is_empty() { - builder = builder.header(HEADER_MAIL_FROM, &envelope.mail_from); - } - - for rcpt_to in &envelope.rcpt_to { - builder = builder.header(HEADER_RCPT_TO, rcpt_to); - } - - builder.body(http_body_util::Full::from(envelope.data.clone()))? - }; - - let client = if allow_invalid_cert { - https_client.relaxed - } else { - https_client.secure - }; - - let response = tokio::time::timeout(Duration::from_secs(60), client.request(request)) - .await - .map_err(|_| crate::error::Error::MailSend { - context: "HTTPS delivery".to_string(), - raw_smtp_answer: "[timeout]".to_string(), - host: mx_host.clone(), - })??; - if response.status().is_success() { - Ok(()) - } else { - let response_body = response.collect().await?.to_bytes(); - Err(crate::error::Error::MailSend { - context: "HTTPS delivery".to_string(), - raw_smtp_answer: String::from_utf8_lossy(&response_body).into(), - host: mx_host, - }) - } + Ok(Self { workers }) } } -impl Drop for TransportHandler { - fn drop(&mut self) { - self.monitor_handle.abort(); - } +#[derive(Debug, Default)] +pub struct TransactionState { + permits: BTreeMap>, } #[async_trait] impl SmtpHandler for TransportHandler { - /// NO-OP - fn handle_mail(&self, _: &str) -> Result<(), String> { + type State = TransactionState; + + fn handle_rcpt_to( + &self, + address: &str, + transaction: &mut Transaction, + ) -> Result<(), String> { + let domain = AddressDomain::from_str(address).map_err(|e| e.smtp_response())?; + + if transaction.state.permits.contains_key(&domain) { + // We already acquired a permit for this domain + return Ok(()); + } + + log::trace!( + "Trying to acquire a permit for {} worker...", + domain.as_ref() + ); + if let Some(permit) = self.workers.get_permit(&domain) { + transaction.state.permits.insert(domain, permit); + } + Ok(()) } - /// NO-OP - async fn check_data(&self, _: &mut Envelope) -> Result<(), String> { - Ok(()) - } + fn handle_data_start(&self, transaction: &Transaction) -> Result<(), String> { + // We want to prevent needlessly sending data from postfix to filtermail, + // so we fail here if we didn't get any permit. + // + // Examplary scenario: + // Consider destinations A and B, where A is unavailable. + // We are sending a message to a group of 1@A, 2@A, 1@B, 2@B. + // After handle_rcpt_to on every recipient, we end up with a permit for domain B (A fails). + // handle_data_start passes and mail data is transmitted to filtermail. + // Delivery to B is performed; 1@B and 2@B receive message and a message to 1@A and 2@A + // is deferred. + // After some time the message is retried, now we only try to acquire permit for A, + // but fail -> empty `transaction.state.permits` + // handle_data_start fails and mail data is not sent to filtermail. + // This greatly reduces RAM usage, as unavailable destination can cause large numbers of + // deferred mails to be constantly retried. + + if transaction.state.permits.is_empty() { + return Err(WORKER_BUSY_421.to_string()); + } - /// NO-OP - async fn reinject_mail(&self, _: &Envelope) -> Result<(), String> { Ok(()) } /// Handles the DATA command and returns LMTP responses as single string. /// /// Never returns an error, as LMTP response is composite. - async fn handle_data(&self, envelope: &mut Envelope) -> Result { + async fn handle_data_dot( + &self, + transaction: &mut Transaction, + ) -> Result { let mut domain_rcpts_map = BTreeMap::new(); - for rcpt in &envelope.rcpt_to { + for rcpt in &transaction.envelope.rcpt_to { let domain = AddressDomain::from_str(rcpt) // Currently we cancel all transactions if any recipient address is invalid. - .map_err(|e| e.lmtp_response(envelope.rcpt_to.len()))?; + .map_err(|e| e.lmtp_response(transaction.envelope.rcpt_to.len()))?; domain_rcpts_map .entry(domain) .or_insert_with(Vec::new) @@ -382,40 +119,50 @@ impl SmtpHandler for TransportHandler { for (rcpt_domain, rcpts) in &domain_rcpts_map { let domain_envelope = { - let mut envelope = envelope.clone(); + let mut envelope = transaction.envelope.clone(); envelope.rcpt_to = rcpts.clone(); envelope }; - let task_id = transactions - .spawn(Self::handle_single_domain( - self.tls_resumption_store.clone(), - self.smtp_connection_pool.clone(), - self.mxdeliv_unsupported_hosts.clone(), - self.https_client.clone(), - self.dns_resolver.clone(), - rcpt_domain.clone(), - domain_envelope, - self.config.mail_domain.clone(), - )) - .id(); - task_id_domain_map.insert(task_id, rcpt_domain); + let receiver_task_id = + if let Some(permit) = transaction.state.permits.remove(rcpt_domain) { + let (message, receiver) = WorkerMessage::new(domain_envelope); + permit.send(message); + // todo: receiver timeout? + transactions.spawn(receiver).id() + } else { + transactions + .spawn(async move { Ok(Err(WORKER_BUSY_421.to_string())) }) + .id() + }; + task_id_domain_map.insert(receiver_task_id, rcpt_domain); } let mut rcpt_response_map = BTreeMap::new(); while let Some(result) = transactions.join_next_with_id().await { - let domain_response = match result { - Ok((id, Ok(resp))) | Ok((id, Err(resp))) => { - task_id_domain_map.remove(&id).map(|domain| (domain, resp)) + let domain = match &result { + Ok((id, _)) => task_id_domain_map.remove(id), + Err(e) => task_id_domain_map.remove(&e.id()), + }; + + let smtp_response = match result { + Ok((_, Ok(Ok(resp)))) | Ok((_, Ok(Err(resp)))) => resp, + Ok((_, Err(e))) => { + log::error!( + "Worker task failed while delivering to {}: {e}", + domain.map(AsRef::as_ref).unwrap_or("") + ); + LOCAL_ERROR_451.to_string() } Err(e) => { - log::error!("Failed to join task: {e}"); - task_id_domain_map - .remove(&e.id()) - .map(|domain| (domain, "451 Local error".to_string())) + log::error!( + "Failed to join task while delivering to {}: {e}", + domain.map(AsRef::as_ref).unwrap_or("") + ); + LOCAL_ERROR_451.to_string() } }; - if let Some((domain, smtp_response)) = domain_response + if let Some(domain) = domain && let Some(rcpts) = domain_rcpts_map.get(domain) { for rcpt in rcpts { @@ -425,16 +172,95 @@ impl SmtpHandler for TransportHandler { } // compose lmtp response... - let ordered_responses: Vec = envelope + let ordered_responses: Vec = transaction + .envelope .rcpt_to .iter() .map(|rcpt| { rcpt_response_map .remove(rcpt) - .unwrap_or_else(|| "451 Local error".to_string()) + .unwrap_or_else(|| LOCAL_ERROR_451.to_string()) }) .collect(); Ok(ordered_responses.join("\r\n")) } } + +#[cfg(test)] +mod tests { + use super::*; + use rstest::{fixture, rstest}; + use testresult::TestResult; + + #[fixture] + fn addrs1() -> Vec { + let mut vec = Vec::new(); + for idx in 0..5 { + vec.push(format!("{idx}@one.example.org")) + } + vec + } + + #[fixture] + fn addrs2() -> Vec { + let mut vec = Vec::new(); + for idx in 0..5 { + vec.push(format!("{idx}@two.example.org")) + } + vec + } + + #[rstest] + #[tokio::test] + async fn test_rcpt_to_and_start_data(addrs1: Vec, addrs2: Vec) -> TestResult { + let transport_handler = TransportHandler::with_queue_size(Config::default(), 1)?; + let domain1 = AddressDomain::from_str(addrs1.first().unwrap())?; + let domain2 = AddressDomain::from_str(addrs2.first().unwrap())?; + + { + let mut trans_1 = Transaction::default(); + let mut trans_2 = Transaction::default(); + let mut trans_3 = Transaction::default(); + + transport_handler.handle_rcpt_to(addrs1.first().unwrap(), &mut trans_1)?; + assert!(trans_1.state.permits.contains_key(&domain1)); + + // Within one transaction, we only use one worker permit, so queue_size=1 is enough. + transport_handler.handle_rcpt_to(addrs1.get(1).unwrap(), &mut trans_1)?; + assert!(trans_1.state.permits.contains_key(&domain1)); + + // However, a second transaction with the same domain won't get a permit. + transport_handler.handle_rcpt_to(addrs1.get(2).unwrap(), &mut trans_2)?; + assert!(!trans_2.state.permits.contains_key(&domain1)); + + // Different domain will work though, as it uses a separate worker, with its own queue. + transport_handler.handle_rcpt_to(addrs2.first().unwrap(), &mut trans_2)?; + assert!(trans_2.state.permits.contains_key(&domain2)); + + // Third transaction won't get any permits. + transport_handler.handle_rcpt_to(addrs1.get(3).unwrap(), &mut trans_3)?; + transport_handler.handle_rcpt_to(addrs2.get(2).unwrap(), &mut trans_3)?; + assert!(!trans_3.state.permits.contains_key(&domain1)); + assert!(!trans_3.state.permits.contains_key(&domain2)); + + // all permits granted -> accept DATA command + assert_eq!(transport_handler.handle_data_start(&trans_1), Ok(())); + + // some permits granted -> accept DATA command + assert_eq!(transport_handler.handle_data_start(&trans_2), Ok(())); + + // no permits granted -> reject + assert!(transport_handler.handle_data_start(&trans_3).is_err()); + } + + // Transactions (and owned by them permits) going out of scope frees the queues. + let mut trans_4 = Transaction::default(); + transport_handler.handle_rcpt_to(addrs1.first().unwrap(), &mut trans_4)?; + transport_handler.handle_rcpt_to(addrs2.first().unwrap(), &mut trans_4)?; + assert!(trans_4.state.permits.contains_key(&domain1)); + assert!(trans_4.state.permits.contains_key(&domain2)); + + Ok(()) + } +} diff --git a/filtermail/src/transport/https_client.rs b/filtermail/src/transport/https_client.rs new file mode 100644 index 00000000..eb6e10be --- /dev/null +++ b/filtermail/src/transport/https_client.rs @@ -0,0 +1,57 @@ +use crate::tls; +use hyper::body::Bytes; +use hyper_rustls::HttpsConnector; +use hyper_util::client::legacy::connect::HttpConnector; +use std::sync::Arc; +use tokio_rustls::rustls; + +/// Cheaply clonable HTTPS client. +/// +/// Holds regular secure variant and relaxed - without certificate verification. +/// +/// Connection pool handled internally by [`hyper_util::client::legacy::Client`]. +#[derive(Clone)] +pub(crate) struct HttpsClient { + pub secure: hyper_util::client::legacy::Client< + HttpsConnector, + http_body_util::Full, + >, + pub relaxed: hyper_util::client::legacy::Client< + HttpsConnector, + http_body_util::Full, + >, +} + +impl HttpsClient { + /// Creates a new `[HttpsClient]`. + pub fn new( + tls_resumption_store: Arc, + ) -> Result { + let tls_client_config = tls::configure_rustls(tls_resumption_store.clone(), false)?; + let https_connector = hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(tls_client_config) + .https_only() + .enable_http1() + .enable_http2() + .build(); + let https_client = + hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .build(https_connector); + + let tls_client_config_relaxed = tls::configure_rustls(tls_resumption_store, true)?; + let https_connector_relaxed = hyper_rustls::HttpsConnectorBuilder::new() + .with_tls_config(tls_client_config_relaxed) + .https_only() + .enable_http1() + .enable_http2() + .build(); + let https_client_relaxed = + hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .build(https_connector_relaxed); + + Ok(Self { + secure: https_client, + relaxed: https_client_relaxed, + }) + } +} diff --git a/filtermail/src/transport/worker.rs b/filtermail/src/transport/worker.rs new file mode 100644 index 00000000..64a199a3 --- /dev/null +++ b/filtermail/src/transport/worker.rs @@ -0,0 +1,447 @@ +use crate::config::Config; +use crate::smtp_client::{SmtpConnectionPool, TlsConfig}; +use crate::smtp_responses::{OK_HTTPS_250, OK_SMTP_250}; +use crate::smtp_server::Envelope; +use crate::transport::{HEADER_MAIL_FROM, HEADER_RCPT_TO, https_client::HttpsClient}; +use crate::utils::{AddressDomain, build_resolver}; +use hickory_resolver::TokioResolver; +use hickory_resolver::proto::rr::RData; +use http_body_util::BodyExt; +use hyper::body::Bytes; +use parking_lot::RwLock; +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc::OwnedPermit; +use tokio::sync::{mpsc, oneshot}; +use tokio::task; +use tokio::task::JoinHandle; +use tokio_rustls::rustls; + +/// Message queue size per [`Worker`]. +/// +/// If a queue to a single destination reaches this limit, +/// all new messages will be immediately deferred. +const PER_DESTINATION_QUEUE_SIZE: usize = 30; + +type SMTPResponse = Result; + +pub struct WorkerPool { + inner: RwLock>>, + client_hostname: String, + smtp_connection_pool: Arc, + mxdeliv_unsupported_hosts: Arc>, + monitor_handle: JoinHandle<()>, + dns_resolver: Arc, + queue_size: usize, +} + +impl WorkerPool { + pub fn new(config: Config) -> Result { + let dns_resolver = Arc::new(build_resolver()?); + + let mxdeliv_cache = Arc::new(retainer::Cache::new()); + let mxdeliv_cache_clone = mxdeliv_cache.clone(); + + let monitor_handle = tokio::spawn(async move { + mxdeliv_cache_clone + .monitor(4, 0.25, Duration::from_secs(10)) + .await + }); + + Ok(Self { + inner: Default::default(), + client_hostname: config.mail_domain, + dns_resolver, + smtp_connection_pool: SmtpConnectionPool::new(), + mxdeliv_unsupported_hosts: mxdeliv_cache, + monitor_handle, + queue_size: PER_DESTINATION_QUEUE_SIZE, + }) + } + + /// Same as [`Self::new`], but lets you set the size of the queue. + /// + /// Used only for tests. + #[cfg(test)] + pub fn with_queue_size(config: Config, queue_size: usize) -> Result { + let mut this = Self::new(config)?; + this.queue_size = queue_size; + Ok(this) + } + + fn get_or_create_worker(&self, destination: &AddressDomain) -> Arc { + // NOTE: these locks are blocking, but critical section here is quite small and + // shouldn't cause issues in async code. + // NOTE: read() returns a guard that is dropped before the match statement. + // This must be ensured or else, the write() line would cause a deadlock. + let worker = { + let mut worker = { + let map = self.inner.read(); + map.get(destination).cloned() + }; + // Remove (and re-create) worker if it finished/crashed. + // In reality, this should never happen. + if let Some(w) = &worker + && w.handle.is_finished() + { + log::error!( + "Worker for destination {} crashed! Restarting...", + destination.as_ref() + ); + worker = None; + { + let mut map = self.inner.write(); + map.remove(destination); + } + }; + worker + }; + + match worker { + Some(worker) => worker, + None => { + // Worker for this destination wasn't spawned yet. + let (tx, rx) = mpsc::channel(self.queue_size); + let handle = tokio::spawn(Worker::run( + destination.clone(), + rx, + self.client_hostname.clone(), + self.smtp_connection_pool.clone(), + self.mxdeliv_unsupported_hosts.clone(), + self.dns_resolver.clone(), + )); + log::trace!("Worker {} spawned", handle.id()); + let worker = Arc::new(Worker { tx, handle }); + + self.inner + .write() + .insert(destination.clone(), worker.clone()); + worker + } + } + } + + /// Tries to get an [`OwnedPermit`] to the worker for specified destination. + /// + /// Returns [`None`] if the worker's queue is full. + pub fn get_permit(&self, destination: &AddressDomain) -> Option> { + let worker = self.get_or_create_worker(destination); + worker.tx.clone().try_reserve_owned().ok() + } +} + +impl Drop for WorkerPool { + fn drop(&mut self) { + self.monitor_handle.abort(); + } +} + +#[derive(Debug)] +pub struct Worker { + pub tx: mpsc::Sender, + handle: JoinHandle>, +} + +impl Drop for Worker { + fn drop(&mut self) { + self.handle.abort(); + } +} + +impl Worker { + pub async fn run( + destination: AddressDomain, + mut rx: mpsc::Receiver, + client_hostname: String, + smtp_connection_pool: Arc, + mxdeliv_unsupported_hosts: Arc>, + dns_resolver: Arc, + ) -> Result<(), crate::error::Error> { + let worker_id = task::try_id() + .map(|id| id.to_string()) + .unwrap_or("?".to_string()); + + log::info!( + "Starting worker {worker_id} for destination {}", + destination.as_ref() + ); + + let tls_resumption_store = Arc::new(rustls::client::ClientSessionMemoryCache::new(256)); + let https_client = HttpsClient::new(tls_resumption_store.clone())?; + + while let Some(message) = rx.recv().await { + log::trace!( + "Worker {worker_id} received a message from {}", + message.envelope.mail_from + ); + let result = Self::handle_single_domain( + tls_resumption_store.clone(), + smtp_connection_pool.clone(), + mxdeliv_unsupported_hosts.clone(), + https_client.clone(), + dns_resolver.clone(), + destination.clone(), + message.envelope, + client_hostname.clone(), + ) + .await; + if message.response_tx.send(result).is_err() { + log::error!( + "Worker {worker_id} ({}) failed to send response to transport handler.", + destination.as_ref() + ); + }; + } + + Ok(()) + } + + /// Handles a single email transaction for a single recipient domain. + #[expect(clippy::too_many_arguments)] + async fn handle_single_domain( + tls_resumption_store: Arc, + smtp_connection_pool: Arc, + mxdeliv_unsupported_hosts: Arc>, + https_client: HttpsClient, + dns_resolver: Arc, + domain: AddressDomain, + envelope: Envelope, + client_hostname: String, + ) -> Result { + let mut allow_invalid_cert = false; + let mut skip_tls = false; // only respected by smtp channel + + let mx_hosts = match domain { + // no-DNS setup; assume the ip from email address is the destination. + AddressDomain::Literal(ip) => { + // We allow self-signed certs on IP-based relays. + allow_invalid_cert = true; + vec![(0, ip)] + } + AddressDomain::Name(mx_domain) => { + if mx_domain.eq_ignore_ascii_case("nauta.cu") { + // Special case; We don't want to defederate nauta.cu, + // which doesn't support STARTTLS at all. + skip_tls = true; + } else if mx_domain.starts_with('_') { + // We use domains starting with `_` for test deployments. + // (You can't request a non-wildcard cert for such domain) + allow_invalid_cert = true; + } + let query = format!("{mx_domain}."); + + match dns_resolver.mx_lookup(query).await { + Ok(mx_records) => { + let mut hosts: Vec<(u16, String)> = Vec::new(); + for mx_record in mx_records.answers() { + let mx = match mx_record.data { + RData::MX(ref mx) => mx, + _ => continue, + }; + + // Null MX / RFC7505 + if mx.exchange.is_root() { + // From RFC7505 section 3: + // > A domain that advertises a null MX MUST NOT + // > advertise any other MX RR. + // We assume this is the only record and exit early. + return Err( + "556 5.1.10 Permanent failure: Recipient address has null MX" + .to_string(), + ); + } + + let host = mx.exchange.to_string().trim_end_matches('.').to_string(); + hosts.push((mx.preference, host)) + } + hosts.sort(); + hosts + } + Err(e) => { + if e.is_no_records_found() { + // "implicit MX" as described by section 5.1 of RFC5321 + // https://datatracker.ietf.org/doc/html/rfc5321#section-5.1 + log::debug!("No MX record found, using implicit MX: {mx_domain}"); + vec![(0, mx_domain)] + } else if e.is_nx_domain() { + return Err(format!("512 Domain {mx_domain} does not exist")); + } else { + return Err(format!("421 DNS resolution failed for {mx_domain}")); + } + } + } + } + }; + + let tls_config = match skip_tls { + true => None, + false => Some(TlsConfig { + allow_invalid_cert, + session_cache: tls_resumption_store, + }), + }; + + let mut last_error = None; + + // we try sequentially in order of MX preference, + // but the IPv4 and IPv6 connections (after `smtp_client::send` resolves mx hostname) + // happens in parallel. + 'try_relay: for (_, mx_host) in mx_hosts { + let skip_mxdeliv = mxdeliv_unsupported_hosts + .get(&mx_host) + .await + .map(|guard| *guard.value()) + .is_some(); + + // HTTPS channel + if skip_mxdeliv { + log::debug!("Skipping HTTP delivery to host that failed recently: {mx_host}"); + } else { + match Self::https_delivery( + https_client.clone(), + mx_host.clone(), + &envelope, + allow_invalid_cert, + ) + .await + { + Ok(_) => { + return Ok(OK_HTTPS_250.to_string()); + } + Err(e) => { + log::debug!("HTTPS delivery to {mx_host} failed: {e}"); + } + } + } + + // SMTP channel (fallback) + match crate::smtp_client::send( + &mx_host, + 25, + &envelope, + &client_hostname, + tls_config.clone(), + dns_resolver.clone(), + smtp_connection_pool.clone(), + ) + .await + { + Ok(_) => { + // Switches this host to SMTP for 30 minutes. + // Note: this MUST happen only after a successful SMTP delivery, + // or otherwise any http error will lock us out of any way to + // deliver to a relay with a blocked port 25 for 30 minutes. + mxdeliv_unsupported_hosts + .insert(mx_host.clone(), (), Duration::from_mins(30)) + .await; + return Ok(OK_SMTP_250.to_string()); + } + Err(error) => { + match &error { + // We only want to try other MX hosts if we encounter a problem + // related to connection. + // (So we don't spam other servers if the message is actually rejected.) + crate::error::Error::Io(_) + | crate::error::Error::ConnectionFailed(_) + | crate::error::Error::Tls(_) => { + // Make sure we quickly retry HTTP if SMTP failed to connect + mxdeliv_unsupported_hosts.remove(&mx_host).await; + log::warn!( + "Connection error relaying to mail server {mx_host}: {error}" + ); + last_error = Some((error.smtp_response(), mx_host.clone())); + continue 'try_relay; + } + crate::error::Error::MailSend { .. } => { + log::warn!("Message rejected by mail server {mx_host}: {error}"); + return Err(error.smtp_response()); + } + _ => { + log::warn!( + "Unexpected error while delivering to mail server {mx_host}: {error}" + ); + return Err(format!( + "{} (while attempting delivery to {mx_host})", + error.smtp_response() + )); + } + } + } + } + } + + let (error, mx_host) = last_error.unwrap_or(("?".to_string(), "?".to_string())); + Err(format!( + "421 Failed to connect to any mail server; last attempt to {mx_host}: {error}" + )) + } + + /// Performs mail delivery to `mx_host` over HTTPS. + /// + /// Times out after 60s. + async fn https_delivery( + https_client: HttpsClient, + mx_host: String, + envelope: &Envelope, + allow_invalid_cert: bool, + ) -> Result<(), crate::error::Error> { + let request: hyper::Request> = { + let mut builder = hyper::Request::builder() + .method(hyper::Method::POST) + .uri(format!("https://{mx_host}/mxdeliv")); + + if !envelope.mail_from.is_empty() { + builder = builder.header(HEADER_MAIL_FROM, &envelope.mail_from); + } + + for rcpt_to in &envelope.rcpt_to { + builder = builder.header(HEADER_RCPT_TO, rcpt_to); + } + + builder.body(http_body_util::Full::from(envelope.data.clone()))? + }; + + let client = if allow_invalid_cert { + https_client.relaxed + } else { + https_client.secure + }; + + let response = tokio::time::timeout(Duration::from_secs(60), client.request(request)) + .await + .map_err(|_| crate::error::Error::MailSend { + context: "HTTPS delivery".to_string(), + raw_smtp_answer: "[timeout]".to_string(), + host: mx_host.clone(), + })??; + if response.status().is_success() { + Ok(()) + } else { + let response_body = response.collect().await?.to_bytes(); + Err(crate::error::Error::MailSend { + context: "HTTPS delivery".to_string(), + raw_smtp_answer: String::from_utf8_lossy(&response_body).into(), + host: mx_host, + }) + } + } +} + +pub struct WorkerMessage { + pub envelope: Envelope, + pub response_tx: oneshot::Sender, +} + +impl WorkerMessage { + pub fn new(envelope: Envelope) -> (Self, oneshot::Receiver) { + let (response_tx, response_rx) = oneshot::channel(); + ( + Self { + envelope, + response_tx, + }, + response_rx, + ) + } +}