cqrenet/relay
5
0
Fork 0
mirror of https://github.com/chatmail/relay.git synced 2026-08-10 18:40:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(transport): Destination worker pool

Browse Source 
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 <jslazak@jslazak.com>
This commit is contained in:
Jagoda Ślązak
2026-05-12 07:13:20 +02:00
parent 7d0db2c46e
commit 7ea7904ea0
9 changed files with 881 additions and 474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
filtermail/src/http_server.rs
+30 -20
View File
@@ -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<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
)?);
}
let mut transaction = Transaction::default();
let mail_from = req
.headers()
.get(crate::transport::HEADER_MAIL_FROM)
@@ -100,16 +102,14 @@ impl<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
.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<String> = req
.headers()
.get_all(crate::transport::HEADER_RCPT_TO)
.iter()
@@ -117,6 +117,21 @@ impl<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
.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<H: SmtpHandler + 'static> Service<Request<Incoming>> for MxDelivService<H>
}
};
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())?),
filtermail/src/inbound.rs
+27 -22
View File
@@ -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<Self::State>) -> 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.
// <https://github.com/chatmail/filtermail/issues/67>
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<Self::State>) -> 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?)
}
}
filtermail/src/main.rs
+1 -2
View File
@@ -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,
filtermail/src/outbound.rs
+39 -23
View File
@@ -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<Self::State>) -> 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<Self::State>) -> 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<String, String> {
async fn handle_data_dot(
&self,
transaction: &mut Transaction<Self::State>,
) -> Result<String, String> {
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())
}
}
filtermail/src/smtp_responses.rs
+7
View File
@@ -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";
filtermail/src/smtp_server.rs
+90 -50
View File
@@ -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<String>,
/// Mail data as transmitted over SMTP/LMTP.
///
/// Described in <https://www.rfc-editor.org/rfc/rfc5321.html#section-2.3.9>.
@@ -25,6 +25,16 @@ pub struct Envelope {
pub data: Vec<u8>,
}
/// 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<S: Debug + Default> {
pub envelope: Envelope,
pub state: S,
}
/// Checks if mail data is valid.
fn is_valid_data(data: &[u8]) -> bool {
// DATA must end with <CRLF>.
@@ -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<Self::State>) -> 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<Self::State>) -> Result<(), String> {
Ok(())
}
/// Handles the DATA command.
async fn handle_data(&self, envelope: &mut Envelope) -> Result<String, String> {
/// 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<Self::State>,
) -> 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<Self::State>) -> Result<(), String> {
Ok(())
}
/// Handles the end of DATA command. Called after receiving the final dot.
async fn handle_data_dot(
&self,
transaction: &mut Transaction<Self::State>,
) -> Result<String, String> {
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 <CR><LF>.<CR><LF>\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?;
filtermail/src/transport.rs
+183 -357
View File
@@ -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<HttpConnector>,
http_body_util::Full<Bytes>,
>,
pub relaxed: hyper_util::client::legacy::Client<
HttpsConnector<HttpConnector>,
http_body_util::Full<Bytes>,
>,
}
impl HttpsClient {
/// Creates a new `[HttpsClient]`.
pub fn new(
tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
) -> Result<Self, crate::error::Error> {
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<TokioResolver>,
tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
smtp_connection_pool: Arc<SmtpConnectionPool>,
https_client: HttpsClient,
mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, bool>>,
monitor_handle: JoinHandle<()>,
workers: WorkerPool,
}
impl TransportHandler {
/// Creates a new [`TransportHandler`].
pub fn new(config: Config) -> Result<Self, crate::error::Error> {
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<rustls::client::ClientSessionMemoryCache>,
smtp_connection_pool: Arc<SmtpConnectionPool>,
mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, bool>>,
https_client: HttpsClient,
dns_resolver: Arc<TokioResolver>,
domain: AddressDomain,
envelope: Envelope,
client_hostname: String,
) -> Result<String, String> {
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<http_body_util::Full<Bytes>> = {
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<Self, crate::error::Error> {
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<AddressDomain, OwnedPermit<WorkerMessage>>,
}
#[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<Self::State>,
) -> 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<Self::State>) -> 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<String, String> {
async fn handle_data_dot(
&self,
transaction: &mut Transaction<Self::State>,
) -> Result<String, String> {
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("<unknown>")
);
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("<unknown>")
);
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<String> = envelope
let ordered_responses: Vec<String> = 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<String> {
let mut vec = Vec::new();
for idx in 0..5 {
vec.push(format!("{idx}@one.example.org"))
}
vec
}
#[fixture]
fn addrs2() -> Vec<String> {
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<String>, addrs2: Vec<String>) -> 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(())
}
}
filtermail/src/transport/https_client.rs
+57
View File
@@ -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<HttpConnector>,
http_body_util::Full<Bytes>,
>,
pub relaxed: hyper_util::client::legacy::Client<
HttpsConnector<HttpConnector>,
http_body_util::Full<Bytes>,
>,
}
impl HttpsClient {
/// Creates a new `[HttpsClient]`.
pub fn new(
tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
) -> Result<Self, crate::error::Error> {
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,
})
}
}
filtermail/src/transport/worker.rs
+447
View File
@@ -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<String, String>;
pub struct WorkerPool {
inner: RwLock<BTreeMap<AddressDomain, Arc<Worker>>>,
client_hostname: String,
smtp_connection_pool: Arc<SmtpConnectionPool>,
mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
monitor_handle: JoinHandle<()>,
dns_resolver: Arc<TokioResolver>,
queue_size: usize,
}
impl WorkerPool {
pub fn new(config: Config) -> Result<Self, crate::error::Error> {
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<Self, crate::error::Error> {
let mut this = Self::new(config)?;
this.queue_size = queue_size;
Ok(this)
}
fn get_or_create_worker(&self, destination: &AddressDomain) -> Arc<Worker> {
// 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<OwnedPermit<WorkerMessage>> {
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<WorkerMessage>,
handle: JoinHandle<Result<(), crate::error::Error>>,
}
impl Drop for Worker {
fn drop(&mut self) {
self.handle.abort();
}
}
impl Worker {
pub async fn run(
destination: AddressDomain,
mut rx: mpsc::Receiver<WorkerMessage>,
client_hostname: String,
smtp_connection_pool: Arc<SmtpConnectionPool>,
mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
dns_resolver: Arc<TokioResolver>,
) -> 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<rustls::client::ClientSessionMemoryCache>,
smtp_connection_pool: Arc<SmtpConnectionPool>,
mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
https_client: HttpsClient,
dns_resolver: Arc<TokioResolver>,
domain: AddressDomain,
envelope: Envelope,
client_hostname: String,
) -> Result<String, String> {
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<http_body_util::Full<Bytes>> = {
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<SMTPResponse>,
}
impl WorkerMessage {
pub fn new(envelope: Envelope) -> (Self, oneshot::Receiver<SMTPResponse>) {
let (response_tx, response_rx) = oneshot::channel();
(
Self {
envelope,
response_tx,
},
response_rx,
)
}
}