test: Test filtermail-transport

Adds filtermail-transport tests and
a test setup for recording SMTP transactions.

Signed-off-by: Jagoda Ślązak <jslazak@jslazak.com>

filtermail/Cargo.lock

@@ -260,6 +260,17 @@ dependencies = [
  "crossbeam-utils",
 ]
 
+[[package]]
+name = "console"
+version = "0.16.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d64e8af5551369d19cf50138de61f1c42074ab970f74e99be916646777f8fc87"
+dependencies = [
+ "encode_unicode",
+ "libc",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "const-oid"
 version = "0.9.6"
@@ -465,6 +476,12 @@ version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
 
+[[package]]
+name = "encode_unicode"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
+
 [[package]]
 name = "encoding_rs"
 version = "0.8.35"
@@ -559,6 +576,7 @@ dependencies = [
  "hyper",
  "hyper-rustls",
  "hyper-util",
+ "insta",
  "log",
  "lru",
  "mailparse",
@@ -567,6 +585,7 @@ dependencies = [
  "retainer",
  "rstest",
  "serde",
+ "serial_test",
  "serini",
  "testresult",
  "thiserror",
@@ -631,6 +650,17 @@ version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
 
+[[package]]
+name = "futures-executor"
+version = "0.3.31"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
+dependencies = [
+ "futures-core",
+ "futures-task",
+ "futures-util",
+]
+
 [[package]]
 name = "futures-io"
 version = "0.3.31"
@@ -1148,6 +1178,18 @@ dependencies = [
  "serde_core",
 ]
 
+[[package]]
+name = "insta"
+version = "1.48.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86f0f8fee8c926415c58d6ae43a08523a26faccb2323f5e6b644fe7dd4ef6b82"
+dependencies = [
+ "console",
+ "once_cell",
+ "similar",
+ "tempfile",
+]
+
 [[package]]
 name = "ipconfig"
 version = "0.3.2"
@@ -2002,6 +2044,31 @@ dependencies = [
  "zmij",
 ]
 
+[[package]]
+name = "serial_test"
+version = "3.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "699f4197115b8a7e7ff19c9a315a4bd6fffec26cc4626ef45ecaea389e081c6d"
+dependencies = [
+ "futures-executor",
+ "futures-util",
+ "log",
+ "once_cell",
+ "parking_lot",
+ "serial_test_derive",
+]
+
+[[package]]
+name = "serial_test_derive"
+version = "3.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "94e153fc76e1c6a068703d6d29c508a0b15c061c4b7e43da59cc097bc342673c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "serini"
 version = "0.2.2"
@@ -2065,6 +2132,12 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e"
 
+[[package]]
+name = "similar"
+version = "2.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbbb5d9659141646ae647b42fe094daf6c6192d1620870b449d9557f748b2daa"
+
 [[package]]
 name = "slab"
 version = "0.4.11"
@@ -2183,6 +2256,19 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417"
 
+[[package]]
+name = "tempfile"
+version = "3.27.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
+dependencies = [
+ "fastrand",
+ "getrandom 0.4.2",
+ "once_cell",
+ "rustix",
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "testresult"
 version = "0.4.1"

filtermail/Cargo.toml

@@ -54,6 +54,8 @@ hyper-rustls = { version = "0.27.9", default-features = false, features = [
 [dev-dependencies]
 rstest = "0.26.1"
 testresult = "0.4.1"
+insta = "1.48.0"
+serial_test = "3.5.0"
 
 [profile.release]
 lto = "thin"

filtermail/src/inbound.rs

@@ -7,6 +7,7 @@ use crate::smtp_client::SmtpConnectionPool;
 use crate::smtp_responses::ENCRYPTION_NEEDED_523;
 pub use crate::smtp_server::Envelope;
 use crate::smtp_server::{SmtpHandler, Transaction};
+use crate::tcp::{TcpConnect, TcpStreamTrait};
 use crate::utils::{AddressDomain, build_resolver, extract_address, log_eml};
 use async_trait::async_trait;
 use hickory_resolver::TokioResolver;
@@ -15,15 +16,19 @@ use std::str::FromStr;
 use std::sync::Arc;
 
 /// Handler for incoming SMTP messages.
-pub struct IncomingBeforeQueueHandler {
+pub struct IncomingBeforeQueueHandler<S: TcpConnect> {
     config: Config,
     dns_resolver: Arc<TokioResolver>,
     dkim_verifier: DkimVerifier,
     skip_dkim: bool,
-    smtp_connection_pool: Arc<SmtpConnectionPool>,
+    smtp_connection_pool: Arc<SmtpConnectionPool<S>>,
 }
 
-impl IncomingBeforeQueueHandler {
+impl<S> IncomingBeforeQueueHandler<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+    S::ConnectionContext: Default,
+{
     pub fn new(config: Config, skip_dkim: bool) -> Result<Self, crate::error::Error> {
         let dns_resolver = Arc::new(build_resolver()?);
         Ok(Self {
@@ -31,7 +36,7 @@ impl IncomingBeforeQueueHandler {
             dns_resolver: dns_resolver.clone(),
             dkim_verifier: DkimVerifier::new(dns_resolver),
             skip_dkim,
-            smtp_connection_pool: SmtpConnectionPool::new(),
+            smtp_connection_pool: SmtpConnectionPool::new(Default::default()),
         })
     }
 
@@ -68,7 +73,11 @@ impl IncomingBeforeQueueHandler {
 }
 
 #[async_trait]
-impl SmtpHandler for IncomingBeforeQueueHandler {
+impl<S> SmtpHandler for IncomingBeforeQueueHandler<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+    S::ConnectionContext: Default,
+{
     type State = ();
 
     async fn check_data(&self, transaction: &mut Transaction<Self::State>) -> Result<(), String> {
@@ -151,12 +160,16 @@ impl SmtpHandler for IncomingBeforeQueueHandler {
     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);
+        let client_config = crate::smtp_client::ClientConfig {
+            client_hostname: &hostname,
+            tls_config: None,
+            lmtp: false,
+        };
         crate::smtp_client::send(
             &self.config.postfix_host,
             self.config.postfix_reinject_port_incoming,
             &transaction.envelope,
-            &hostname,
+            client_config,
-            None,
             self.dns_resolver.clone(),
             self.smtp_connection_pool.clone(),
         )
@@ -175,6 +188,7 @@ mod tests {
     use super::*;
     use rstest::{fixture, rstest};
     use testresult::TestResult;
+    use tokio::net::TcpStream;
 
     #[fixture]
     fn config() -> Config {
@@ -191,7 +205,7 @@ mod tests {
         #[case] address: &str,
         config: Config,
     ) -> TestResult {
-        let handler = IncomingBeforeQueueHandler::new(config, false)?;
+        let handler = IncomingBeforeQueueHandler::<TcpStream>::new(config, false)?;
         let mut transaction = Transaction {
             envelope: Envelope {
                 mail_from: address.to_string(),

filtermail/src/main.rs

@@ -35,6 +35,7 @@ pub(crate) mod outbound;
 pub(crate) mod smtp_client;
 mod smtp_responses;
 pub(crate) mod smtp_server;
+mod tcp;
 mod tls;
 mod transport;
 pub(crate) mod utils;
@@ -50,6 +51,7 @@ use std::env;
 use std::process;
 use std::str::FromStr;
 use std::sync::Arc;
+use tokio::net::TcpStream;
 
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 enum Mode {
@@ -120,7 +122,9 @@ async fn main() -> Result<(), error::Error> {
     match mode {
         Mode::Outgoing => {
             let addr = (config.filtermail_host, config.filtermail_smtp_port);
-            let handler = Arc::new(OutgoingBeforeQueueHandler::new(config.clone())?);
+            let handler = Arc::new(OutgoingBeforeQueueHandler::<TcpStream>::new(
+                config.clone(),
+            )?);
             let max_size = config.max_message_size;
             log::debug!("Outgoing SMTP server listening on {}:{}", addr.0, addr.1);
 
@@ -139,7 +143,10 @@ async fn main() -> Result<(), error::Error> {
                 log::warn!("DKIM verification DISABLED! This should not be used in production.");
             }
 
-            let handler = Arc::new(IncomingBeforeQueueHandler::new(config.clone(), skip_dkim)?);
+            let handler = Arc::new(IncomingBeforeQueueHandler::<TcpStream>::new(
+                config.clone(),
+                skip_dkim,
+            )?);
             let max_size = config.max_message_size;
 
             let mut server_set = tokio::task::JoinSet::new();
@@ -177,7 +184,7 @@ async fn main() -> Result<(), error::Error> {
                 config.filtermail_host,
                 config.filtermail_lmtp_port_transport,
             );
-            let handler = Arc::new(TransportHandler::new(config.clone())?);
+            let handler = Arc::new(TransportHandler::<TcpStream>::new(config.clone())?);
             let max_size = config.max_message_size;
             log::debug!("Transport SMTP server listening on {}:{}", addr.0, addr.1);
 

filtermail/src/outbound.rs

@@ -6,6 +6,7 @@ use crate::smtp_client::SmtpConnectionPool;
 use crate::smtp_responses::ENCRYPTION_NEEDED_523;
 use crate::smtp_responses::OK_250;
 use crate::smtp_server::{SmtpHandler, Transaction};
+use crate::tcp::{TcpConnect, TcpStreamTrait};
 use crate::utils::{build_resolver, extract_address};
 use async_trait::async_trait;
 use governor::clock::MonotonicClock;
@@ -16,7 +17,7 @@ use mailparse::{MailHeaderMap, parse_mail};
 use std::sync::Arc;
 
 /// Handler for outgoing SMTP messages.
-pub struct OutgoingBeforeQueueHandler {
+pub struct OutgoingBeforeQueueHandler<S: TcpConnect> {
     config: Config,
     dns_resolver: Arc<TokioResolver>,
 
@@ -33,10 +34,14 @@ pub struct OutgoingBeforeQueueHandler {
         MonotonicClock,
         NoOpMiddleware<std::time::Instant>,
     >,
-    smtp_connection_pool: Arc<SmtpConnectionPool>,
+    smtp_connection_pool: Arc<SmtpConnectionPool<S>>,
 }
 
-impl OutgoingBeforeQueueHandler {
+impl<S> OutgoingBeforeQueueHandler<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+    S::ConnectionContext: Default,
+{
     pub fn new(config: Config) -> Result<Self, crate::error::Error> {
         let quota = Quota::per_minute(config.max_user_send_per_minute)
             .allow_burst(config.max_user_send_burst_size);
@@ -46,13 +51,16 @@ impl OutgoingBeforeQueueHandler {
             config,
             dns_resolver,
             send_rate_limiter,
-            smtp_connection_pool: SmtpConnectionPool::new(),
+            smtp_connection_pool: SmtpConnectionPool::new(Default::default()),
         })
     }
 }
 
 #[async_trait]
-impl SmtpHandler for OutgoingBeforeQueueHandler {
+impl<S> SmtpHandler for OutgoingBeforeQueueHandler<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+{
     type State = ();
 
     fn handle_mail_from(&self, address: &str) -> Result<(), String> {
@@ -161,12 +169,16 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
     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);
+        let client_config = crate::smtp_client::ClientConfig {
+            client_hostname: &hostname,
+            tls_config: None,
+            lmtp: false,
+        };
         crate::smtp_client::send(
             &self.config.postfix_host,
             self.config.postfix_reinject_port,
             &transaction.envelope,
-            &hostname,
+            client_config,
-            None,
             self.dns_resolver.clone(),
             self.smtp_connection_pool.clone(),
         )

filtermail/src/smtp_client.rs

@@ -1,4 +1,5 @@
 use crate::smtp_server::Envelope;
+use crate::tcp::{TcpConnect, TcpStreamTrait};
 use hickory_resolver::TokioResolver;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -6,14 +7,13 @@ use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufStream};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
-use tokio::net::TcpStream;
 use tokio::task::{JoinHandle, JoinSet};
 use tokio_io_timeout::TimeoutStream;
 use tokio_rustls::rustls::client::ClientSessionMemoryCache;
 
 /// Wraps SMTP connection, contains stream and ESMTP support information.
-pub struct SmtpConnection {
+pub struct SmtpConnection<S> {
-    pub stream: BufStream<SmtpStream>,
+    pub stream: BufStream<SmtpStream<S>>,
     pub pipelining: bool,
 }
 
@@ -22,14 +22,21 @@ pub struct SmtpConnection {
 /// Connections are cached for up to 100 seconds of idle time.
 ///
 /// Only a single connection is cached per address/port pair.
-pub struct SmtpConnectionPool {
+pub struct SmtpConnectionPool<S>
-    pool: Arc<retainer::Cache<(String, u16), SmtpConnection>>,
+where
+    S: TcpStreamTrait + TcpConnect,
+{
+    pool: Arc<retainer::Cache<(String, u16), SmtpConnection<S>>>,
     monitor_handle: JoinHandle<()>,
+    context: S::ConnectionContext,
 }
 
-impl SmtpConnectionPool {
+impl<S> SmtpConnectionPool<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+{
     /// Creates a new connection pool and starts the cache monitoring task.
-    pub fn new() -> Arc<Self> {
+    pub fn new(context: S::ConnectionContext) -> Arc<Self> {
         let pool = Arc::new(retainer::Cache::new());
         let pool_clone = pool.clone();
 
@@ -39,16 +46,17 @@ impl SmtpConnectionPool {
         Arc::new(Self {
             pool,
             monitor_handle,
+            context,
         })
     }
 
     /// Takes a connection from the pool for the given address and port, if available.
-    pub async fn take(&self, address: &str, port: u16) -> Option<SmtpConnection> {
+    pub async fn take(&self, address: &str, port: u16) -> Option<SmtpConnection<S>> {
         self.pool.remove(&(address.to_string(), port)).await
     }
 
     /// Puts a connection into the pool for the given address and port, with a 100s timeout.
-    pub async fn put(&self, address: &str, port: u16, connection: SmtpConnection) {
+    pub async fn put(&self, address: &str, port: u16, connection: SmtpConnection<S>) {
         // similarly to postfix default -> 100s max idle time.
         self.pool
             .insert(
@@ -60,25 +68,25 @@ impl SmtpConnectionPool {
     }
 }
 
-impl Drop for SmtpConnectionPool {
+impl<S: TcpConnect> Drop for SmtpConnectionPool<S> {
     fn drop(&mut self) {
         self.monitor_handle.abort();
     }
 }
 
-/// A [`TcpStream`] used for SMTP communication.
+/// A [`TcpStream`] wrapper used for SMTP communication.
 #[expect(clippy::large_enum_variant)]
-pub enum SmtpStream {
+pub enum SmtpStream<S> {
     /// A plain TCP stream.
-    Plain(Pin<Box<TimeoutStream<TcpStream>>>),
+    Plain(Pin<Box<TimeoutStream<S>>>),
     /// A TLS-encrypted stream.
-    Tls(tokio_rustls::TlsStream<Pin<Box<TimeoutStream<TcpStream>>>>),
+    Tls(tokio_rustls::TlsStream<Pin<Box<TimeoutStream<S>>>>),
 }
 
-impl SmtpStream {
+impl<S: TcpStreamTrait> SmtpStream<S> {
     /// Creates a new plain SMTP stream from a raw TCP stream,
     /// with read and write timeouts set to 60 seconds.
-    fn plain(stream: TcpStream) -> Self {
+    fn plain(stream: S) -> Self {
         let mut timeout_stream = TimeoutStream::new(stream);
         timeout_stream.set_write_timeout(Some(Duration::from_secs(60)));
         timeout_stream.set_read_timeout(Some(Duration::from_secs(60)));
@@ -127,7 +135,7 @@ pub struct TlsConfig {
     pub(crate) session_cache: Arc<ClientSessionMemoryCache>,
 }
 
-impl AsyncWrite for SmtpStream {
+impl<S: TcpStreamTrait> AsyncWrite for SmtpStream<S> {
     fn poll_write(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -154,7 +162,7 @@ impl AsyncWrite for SmtpStream {
     }
 }
 
-impl AsyncRead for SmtpStream {
+impl<S: TcpStreamTrait> AsyncRead for SmtpStream<S> {
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
@@ -189,25 +197,31 @@ async fn to_socket_addrs(
 }
 
 /// Establishes a TCP connection to the given address and port, trying all resolved IPs in parallel.
-async fn establish_tcp_connection(
+async fn establish_tcp_connection<S>(
     address: &str,
     port: u16,
     dns_resolver: Arc<TokioResolver>,
-) -> Result<TcpStream, crate::error::Error> {
+    context: S::ConnectionContext,
-    let mut set: JoinSet<tokio::io::Result<TcpStream>> = JoinSet::new();
+) -> Result<S, crate::error::Error>
+where
+    S: TcpStreamTrait + TcpConnect,
+{
+    let mut set: JoinSet<tokio::io::Result<S>> = JoinSet::new();
 
     let socket_addrs = to_socket_addrs(address, port, dns_resolver).await?;
 
     for addr in socket_addrs.clone() {
+        let context_clone = context.clone();
         set.spawn(async move {
             log::trace!("SMTP client: connecting to {addr}...");
-            let stream =
+            let stream: S =
-                tokio::time::timeout(Duration::from_secs(60), TcpStream::connect(addr)).await??;
+                tokio::time::timeout(Duration::from_secs(60), S::connect(addr, context_clone))
+                    .await??;
             stream.set_nodelay(true)?;
             Ok(stream)
         });
     }
-    let mut stream: Option<TcpStream> = None;
+    let mut stream: Option<S> = None;
     while let Some(result) = set.join_next().await {
         match result {
             Ok(Ok(s)) => {
@@ -227,46 +241,60 @@ async fn establish_tcp_connection(
     }
 }
 
+/// SMTP/LMTP client configuration options.
+pub struct ClientConfig<'a> {
+    /// Client hostname used for greeting
+    pub client_hostname: &'a str,
+
+    /// If [`Some`], the connection will be upgraded to TLS.
+    /// The client will fail early if the server does not support STARTTLS.
+    pub tls_config: Option<TlsConfig>,
+
+    /// If `true`, switches to `LHLO` greeting and returns per-recipient composite response.
+    pub lmtp: bool,
+}
+
 /// Sends an email using an SMTP server at `smtp_addr`.
-///
-/// If `tls_config` is provided, the connection will be upgraded to TLS.
-/// The client will fail early if the server does not support STARTTLS.
-///
 /// If `address` is a domain that resolves to multiple IP addresses,
 /// all will be tried in parallel and the first successful connection will be used.
 ///
 /// `pool` is used to reuse existing connections to the same address and port, if available.
-pub async fn send(
+pub async fn send<S>(
     address: &str,
     port: u16,
     envelope: &Envelope,
-    client_hostname: &str,
+    config: ClientConfig<'_>,
-    tls_config: Option<TlsConfig>,
     dns_resolver: Arc<TokioResolver>,
-    pool: Arc<SmtpConnectionPool>,
+    pool: Arc<SmtpConnectionPool<S>>,
-) -> Result<(), crate::error::Error> {
+) -> Result<(), crate::error::Error>
-    let (mut buf_stream, reused, mut pipelining) = if let Some(connection) =
+where
-        pool.take(address, port).await
+    S: TcpStreamTrait + TcpConnect,
-    {
+{
-        log::debug!(
+    let greeting = if config.lmtp { "LHLO" } else { "EHLO" };
-            "Reusing existing connection to {}",
+
-            connection.stream.get_ref().format_host(address)
+    let (mut buf_stream, reused, mut pipelining) =
-        );
+        if let Some(connection) = pool.take(address, port).await {
-        if tls_config.is_some() {
+            log::debug!(
-            // This should never happen,
+                "Reusing existing connection to {}",
-            // assert to make sure we never accidentally use a plain connection while expecting TLS.
+                connection.stream.get_ref().format_host(address)
-            assert!(
-                matches!(connection.stream.get_ref(), SmtpStream::Tls(_)),
-                "Expected TLS stream from pool, but got plain stream."
             );
-        }
+            if config.tls_config.is_some() {
-        (connection.stream, true, connection.pipelining)
+                // This should never happen,
-    } else {
+                // assert to make sure we never accidentally use a plain connection while expecting TLS.
-        let stream =
+                assert!(
-            SmtpStream::plain(establish_tcp_connection(address, port, dns_resolver.clone()).await?);
+                    matches!(connection.stream.get_ref(), SmtpStream::Tls(_)),
-        log::debug!("Successfully connected to {}", stream.format_host(address));
+                    "Expected TLS stream from pool, but got plain stream."
-        (BufStream::new(stream), false, false)
+                );
-    };
+            }
+            (connection.stream, true, connection.pipelining)
+        } else {
+            let stream = SmtpStream::plain(
+                establish_tcp_connection(address, port, dns_resolver.clone(), pool.context.clone())
+                    .await?,
+            );
+            log::debug!("Successfully connected to {}", stream.format_host(address));
+            (BufStream::new(stream), false, false)
+        };
 
     let mut response = String::new();
 
@@ -329,7 +357,8 @@ pub async fn send(
         // e.g.: 421 example.org Service closing transmission channel - command timeout
         if response.starts_with("421") {
             log::debug!("Reused connection is dead; establishing new connection...");
-            let stream = establish_tcp_connection(address, port, dns_resolver).await?;
+            let stream: S =
+                establish_tcp_connection(address, port, dns_resolver, pool.context.clone()).await?;
             log::debug!("Successfully connected to {}", stream.peer_addr()?);
             buf_stream = BufStream::new(SmtpStream::plain(stream));
             false
@@ -346,8 +375,8 @@ pub async fn send(
         smtp_read!("initial greeting", "220")?;
 
         smtp_cmd!(
-            format!("EHLO {client_hostname}\r\n").as_bytes(),
+            format!("{greeting} {}\r\n", { config.client_hostname }).as_bytes(),
-            "EHLO",
+            greeting,
             "250"
         )?;
 
@@ -358,7 +387,7 @@ pub async fn send(
         }
 
         // ESMTP: STARTTLS
-        if let Some(tls_config) = tls_config {
+        if let Some(tls_config) = config.tls_config {
             if !response.to_uppercase().contains("STARTTLS") {
                 // TLS was requested, but server doesn't support STARTTLS.
                 return Err(crate::error::Error::MailSend {
@@ -392,7 +421,7 @@ pub async fn send(
             buf_stream = BufStream::new(smtp_stream);
 
             smtp_cmd!(
-                format!("EHLO {client_hostname}\r\n").as_bytes(),
+                format!("EHLO {}\r\n", config.client_hostname).as_bytes(),
                 "EHLO after STARTTLS",
                 "250"
             )?;
@@ -457,7 +486,14 @@ pub async fn send(
     }
 
     smtp_write!(&envelope.data);
-    smtp_cmd!(b".\r\n", "end of DATA", "250")?;
+    smtp_write!(b".\r\n");
+    if config.lmtp {
+        for _ in 0..envelope.rcpt_to.len() {
+            smtp_read!("end of DATA", "250")?;
+        }
+    } else {
+        smtp_read!("end of DATA", "250")?;
+    }
 
     pool.put(
         address,
@@ -477,6 +513,7 @@ mod tests {
     use super::*;
     use rstest::rstest;
     use std::net::SocketAddr;
+    use tokio::net::TcpStream;
 
     #[rstest]
     #[case::ipv4("192.0.2.0:25".parse().ok(), "192.0.2.0", "192.0.2.0:25")]
@@ -489,7 +526,7 @@ mod tests {
         #[case] host: &str,
         #[case] expected: &str,
     ) {
-        let result = SmtpStream::format_host_inner(host, socket_addr);
+        let result = SmtpStream::<TcpStream>::format_host_inner(host, socket_addr);
         assert_eq!(result, expected);
     }
 }

filtermail/src/smtp_server.rs

@@ -7,8 +7,8 @@ use memchr::{Memchr, memmem};
 use std::fmt::Debug;
 use std::sync::Arc;
 use std::time::Duration;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
+use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufStream};
-use tokio::net::{TcpListener, TcpStream};
+use tokio::net::TcpListener;
 
 /// Represents an SMTP envelope with sender, recipients, and raw message data.
 #[derive(Debug, Default, Clone)]
@@ -139,10 +139,19 @@ pub trait SmtpHandler: Send + Sync {
             log::warn!("Failed to reinject mail: {e}");
             e
         })?;
-        Ok("OK_250".to_string())
+        Ok(OK_250.to_string())
     }
 }
 
+/// A mockup handler that does nothing.
+#[cfg(test)]
+pub struct MockHandler;
+
+#[cfg(test)]
+impl SmtpHandler for MockHandler {
+    type State = ();
+}
+
 /// Runs the SMTP server on the specified address with the given handler and maximum message size.
 pub async fn run_smtp_server<H>(
     addr: &impl tokio::net::ToSocketAddrs,
@@ -164,7 +173,7 @@ where
 
                 let handler = handler.clone();
                 tokio::spawn(async move {
-                    if let Err(e) = handle_connection(socket, handler, max_size).await {
+                    if let Err(e) = handle_connection(socket, handler, max_size, false).await {
                         log::error!("Error handling connection: {e}");
                     }
                 });
@@ -180,27 +189,36 @@ where
 }
 
 /// Handles an individual SMTP connection.
-async fn handle_connection<H>(
+///
-    socket: TcpStream,
+/// Setting `auto_quit` to `true` will automatically close connection after receiving the first
+/// message. Should be used only for tests, it's not a behavior described by SMTP spec.
+pub(crate) async fn handle_connection<S, H>(
+    stream: S,
     handler: Arc<H>,
     max_size: usize,
+    auto_quit: bool,
 ) -> Result<(), Box<dyn std::error::Error>>
 where
+    S: AsyncWrite + AsyncRead + Unpin,
     H: SmtpHandler,
 {
-    let (reader, writer) = socket.into_split();
+    let mut bufstream = BufStream::new(stream);
-    let mut reader = BufReader::new(reader);
-    let mut writer = BufWriter::new(writer);
     let mut line = String::new();
 
-    writer.write_all(b"220 filtermail SMTP\r\n").await?;
+    macro_rules! smtp_write {
-    writer.flush().await?;
+        ($($arg:tt)*) => {
+            bufstream.write_all(format!($($arg)*).as_bytes()).await?;
+            bufstream.flush().await?;
+        };
+    }
+
+    smtp_write!("220 filtermail SMTP\r\n");
 
     let mut transaction = Transaction::default();
 
     'connection: loop {
         line.clear();
-        let n = reader.read_line(&mut line).await?;
+        let n = bufstream.read_line(&mut line).await?;
         if n == 0 {
             break 'connection;
         }
@@ -218,66 +236,50 @@ where
         log::debug!("Received: {cmd}");
 
         if cmd.to_uppercase().starts_with("HELO") {
-            writer.write_all(b"250-filtermail\r\n250 OK\r\n").await?;
+            smtp_write!("250-filtermail\r\n250 OK\r\n");
-            writer.flush().await?;
         } else if cmd.to_uppercase().starts_with("EHLO")
             // We support LMTP, but it's not validated;
             // service that expects LMTP will send LMTP responses no matter the greeting.
             // Sufficient for our internal use case.
             || cmd.to_uppercase().starts_with("LHLO")
         {
-            writer
+            smtp_write!("250-filtermail\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
             transaction.envelope.mail_from = String::new();
-            writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
+            smtp_write!("{OK_250}\r\n");
-            writer.flush().await?;
         } else if cmd.to_uppercase().starts_with("MAIL FROM:") {
             if let Some(from) = extract_address(cmd) {
                 if let Err(e) = handler.handle_mail_from(&from) {
-                    writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
+                    smtp_write!("{}\r\n", e);
-                    writer.flush().await?;
                     continue 'connection;
                 }
                 transaction.envelope.mail_from = from;
-                writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
+                smtp_write!("{OK_250}\r\n");
-                writer.flush().await?;
             } else {
                 log::warn!("Invalid MAIL FROM command. Can't extract address. Received: {cmd}");
-                writer
+                smtp_write!("500 Invalid address in MAIL FROM\r\n");
-                    .write_all(b"500 Invalid address in MAIL FROM\r\n")
-                    .await?;
-                writer.flush().await?;
             }
         } else if cmd.to_uppercase().starts_with("RCPT TO:") {
             if let Some(to) = extract_address(cmd) {
                 if let Err(e) = handler.handle_rcpt_to(&to, &mut transaction) {
-                    writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
+                    smtp_write!("{}\r\n", e);
-                    writer.flush().await?;
                     continue 'connection;
                 }
                 transaction.envelope.rcpt_to.push(to);
-                writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
+                smtp_write!("{OK_250}\r\n");
-                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?;
+                smtp_write!("{}\r\n", e);
-                writer.flush().await?;
                 continue 'connection;
             }
-            writer
+            smtp_write!("354 End data with <CR><LF>.<CR><LF>\r\n");
-                .write_all(b"354 End data with <CR><LF>.<CR><LF>\r\n")
-                .await?;
-            writer.flush().await?;
             let mut data = Vec::new();
             let mut data_line = String::new();
             'data_read: loop {
                 data_line.clear();
-                if reader.read_line(&mut data_line).await? == 0 {
+                if bufstream.read_line(&mut data_line).await? == 0 {
                     log::warn!("Unexpected EoF while receiving DATA! Closing connection.");
                     break 'connection;
                 }
@@ -303,10 +305,7 @@ where
                 data.extend_from_slice(data_line.as_bytes());
 
                 if data.len() > max_size {
-                    writer
+                    smtp_write!("552 Message exceeds maximum size\r\n");
-                        .write_all(b"552 Message exceeds maximum size\r\n")
-                        .await?;
-                    writer.flush().await?;
                     continue 'connection;
                 }
             }
@@ -317,33 +316,27 @@ where
             match handler.handle_data_dot(&mut transaction).await {
                 Ok(response) => {
                     log::debug!("Sent: {response}");
-                    writer
+                    smtp_write!("{}\r\n", response);
-                        .write_all(format!("{}\r\n", response).as_bytes())
-                        .await?;
-                    writer.flush().await?;
                 }
                 Err(e) => {
                     log::debug!("Sent: {e}");
-                    writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
+                    smtp_write!("{}\r\n", e);
-                    writer.flush().await?;
                 }
             }
-
+            if auto_quit {
+                break 'connection;
+            }
             transaction = Transaction::default();
         } else if cmd.to_uppercase().starts_with("QUIT") {
-            writer.write_all(b"221 OK\r\n").await?;
+            smtp_write!("221 OK\r\n");
-            writer.flush().await?;
             break 'connection;
         } else if cmd.to_uppercase().starts_with("RSET") {
             transaction = Transaction::default();
-            writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
+            smtp_write!("{OK_250}\r\n");
-            writer.flush().await?;
         } else if cmd.to_uppercase().starts_with("NOOP") {
-            writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
+            smtp_write!("{OK_250}\r\n");
-            writer.flush().await?;
         } else {
-            writer.write_all(b"500 Command not recognized\r\n").await?;
+            smtp_write!("500 Command not recognized\r\n");
-            writer.flush().await?;
         }
     }
 

filtermail/src/snapshots/filtermail__transport__tests__smtp_send_mail.snap

@@ -0,0 +1,64 @@
+---
+source: src/transport.rs
+expression: "format!(\"[postfix -> filtermail-transport]\\r\\n{record}\\r\\n\\\n            [filtermail-transport -> destination A]\\r\\n{}\\r\\n\\\n            [filtermail-transport -> destination B]\\r\\n{}\",\nremote_records[0], remote_records[1])"
+---
+[postfix -> filtermail-transport]
+< 220 filtermail SMTP
+> LHLO postfix
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<a1@localhost>
+< 250 OK
+> RCPT TO:<a2@localhost>
+< 250 OK
+> RCPT TO:<b1@[127.0.0.1]>
+< 250 OK
+> RCPT TO:<b2@[127.0.0.1]>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+> .
+< 250 OK (SMTP)
+250 OK (SMTP)
+250 OK (SMTP)
+250 OK (SMTP)
+
+[filtermail-transport -> destination A]
+< 220 filtermail SMTP
+> EHLO example.org
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<a1@localhost>
+< 250 OK
+> RCPT TO:<a2@localhost>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+.
+< 250 OK
+
+[filtermail-transport -> destination B]
+< 220 filtermail SMTP
+> EHLO example.org
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<b1@[127.0.0.1]>
+< 250 OK
+> RCPT TO:<b2@[127.0.0.1]>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+.
+< 250 OK

filtermail/src/snapshots/filtermail__transport__tests__smtp_send_mail_defer.snap

@@ -0,0 +1,73 @@
+---
+source: src/transport.rs
+expression: "format!(\"TRANSACTION 1\\r\\n\\\n            [postfix -> filtermail-transport]\\r\\n{record_postfix_1}\\r\\n\\\n            [filtermail-transport -> destination A]\\r\\n{record_filtermail_1}\\r\\n\\r\\n\\\n            TRANSACTION 2\\r\\n\\\n            [postfix -> filtermail-transport]\\r\\n{record_postfix_2}\\r\\n\\\n            [filtermail-transport -> destination B]\\r\\n{record_filtermail_2}\",)"
+---
+TRANSACTION 1
+[postfix -> filtermail-transport]
+< 220 filtermail SMTP
+> LHLO postfix
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<a1@localhost>
+< 250 OK
+> RCPT TO:<b1@[127.0.0.1]>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+> .
+< 250 OK (SMTP)
+421 Worker for this destination is busy
+
+[filtermail-transport -> destination A]
+< 220 filtermail SMTP
+> EHLO example.org
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<a1@localhost>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+.
+< 250 OK
+
+
+TRANSACTION 2
+[postfix -> filtermail-transport]
+< 220 filtermail SMTP
+> LHLO postfix
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<b1@[127.0.0.1]>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+> .
+< 250 OK (SMTP)
+
+[filtermail-transport -> destination B]
+< 220 filtermail SMTP
+> EHLO example.org
+< 250-filtermail
+250-8BITMIME
+250 OK
+> MAIL FROM:<sender@here>
+< 250 OK
+> RCPT TO:<b1@[127.0.0.1]>
+< 250 OK
+> DATA
+< 354 End data with <CR><LF>.<CR><LF>
+> message
+.
+< 250 OK

filtermail/src/tcp.rs

@@ -0,0 +1,59 @@
+//! TCP related code.
+
+use async_trait::async_trait;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::{TcpStream, ToSocketAddrs};
+
+#[cfg(test)]
+pub mod rec_stream;
+
+/// Abstraction over [`TcpStream`] allowing e.g. mocking it.
+pub trait TcpStreamTrait: AsyncRead + AsyncWrite + Unpin + Send + Sync + Sized + 'static {
+    /// Returns the remote address that this stream is connected to.
+    fn peer_addr(&self) -> std::io::Result<std::net::SocketAddr>;
+
+    /// Sets the value of the `TCP_NODELAY` option on this socket.
+    fn set_nodelay(&self, nodelay: bool) -> std::io::Result<()>;
+}
+
+/// Trait adding a `connect` method similar to [`TcpStream::connect`],
+/// that allows passing additional context when creating the stream.
+#[async_trait]
+pub trait TcpConnect: TcpStreamTrait {
+    /// Type of additional context passed to [`TcpConnect::connect`].
+    type ConnectionContext: Send + Sync + Clone + 'static;
+
+    /// Opens a TCP connection to a remote host.
+    async fn connect<A: ToSocketAddrs + Send>(
+        addr: A,
+        context: Self::ConnectionContext,
+    ) -> std::io::Result<Self>;
+}
+
+impl TcpStreamTrait for TcpStream {
+    /// Returns the remote address that this stream is connected to.
+    ///
+    /// Delegates to [`TcpStream::peer_addr`].
+    fn peer_addr(&self) -> std::io::Result<std::net::SocketAddr> {
+        TcpStream::peer_addr(self)
+    }
+
+    /// Sets the value of the `TCP_NODELAY` option on this socket.
+    ///
+    /// Delegates to [`TcpStream::set_nodelay`].
+    fn set_nodelay(&self, nodelay: bool) -> std::io::Result<()> {
+        TcpStream::set_nodelay(self, nodelay)
+    }
+}
+
+#[async_trait]
+impl TcpConnect for TcpStream {
+    type ConnectionContext = ();
+
+    /// Opens a TCP connection to a remote host.
+    ///
+    /// Delegates to [`TcpStream::connect`].
+    async fn connect<A: ToSocketAddrs + Send>(addr: A, _: ()) -> std::io::Result<Self> {
+        TcpStream::connect(addr).await
+    }
+}

filtermail/src/tcp/rec_stream.rs

@@ -0,0 +1,127 @@
+//! [`TcpStreamTrait`] implementation that records communication.
+//!
+//! Used for snapshot testing.
+
+use super::{TcpConnect, TcpStreamTrait};
+use async_trait::async_trait;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::net::{TcpStream, ToSocketAddrs};
+use tokio::sync::mpsc::Sender;
+
+/// A stream that behaves similarly to [`TcpStream`],
+/// but additionally records the whole conversation to internal buffer,
+/// and sends it over [`Sender`] when dropped.
+pub struct RecTcpStream {
+    tx: Sender<String>,
+    inner: TcpStream,
+    rec_buffer: String,
+    // read/write arrows
+    arrows: (char, char),
+}
+
+impl RecTcpStream {
+    /// Creates a new [`RecTcpStream`].
+    ///
+    /// Recorded conversation will be sent over `tx`.
+    ///
+    /// Setting `is_server` to `true`, will invert read/write arrow characters,
+    /// so that they correctly used for `> client command` and `< server response`.
+    pub fn new(stream: TcpStream, tx: Sender<String>, is_server: bool) -> Self {
+        Self {
+            tx,
+            inner: stream,
+            rec_buffer: String::new(),
+            arrows: match is_server {
+                true => ('>', '<'),
+                false => ('<', '>'),
+            },
+        }
+    }
+}
+
+impl AsyncWrite for RecTcpStream {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<std::io::Result<usize>> {
+        let mut_self = self.get_mut();
+        let inner_result = Pin::new(&mut mut_self.inner).poll_write(cx, buf);
+        if inner_result.is_pending() {
+            return inner_result;
+        }
+
+        if !buf.is_empty() {
+            let mut data = String::from_utf8_lossy(buf).to_string();
+            data = format!("{} {data}", mut_self.arrows.1);
+            mut_self.rec_buffer.push_str(&data);
+        }
+
+        inner_result
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
+    }
+}
+
+impl AsyncRead for RecTcpStream {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
+        let mut_self = self.get_mut();
+        let inner_result = Pin::new(&mut mut_self.inner).poll_read(cx, buf);
+        if inner_result.is_pending() {
+            return inner_result;
+        }
+
+        let filled = buf.filled();
+        if !filled.is_empty() {
+            let mut data = String::from_utf8_lossy(filled).to_string();
+            data = format!("{} {data}", mut_self.arrows.0);
+            mut_self.rec_buffer.push_str(&data);
+        }
+
+        inner_result
+    }
+}
+
+impl Drop for RecTcpStream {
+    fn drop(&mut self) {
+        let tx = self.tx.clone();
+        let rec_buffer = self.rec_buffer.clone();
+        tokio::spawn(async move { tx.send(rec_buffer).await.unwrap() });
+    }
+}
+
+impl TcpStreamTrait for RecTcpStream {
+    fn peer_addr(&self) -> std::io::Result<SocketAddr> {
+        self.inner.peer_addr()
+    }
+
+    fn set_nodelay(&self, nodelay: bool) -> std::io::Result<()> {
+        self.inner.set_nodelay(nodelay)
+    }
+}
+
+#[async_trait]
+impl TcpConnect for RecTcpStream {
+    type ConnectionContext = Sender<String>;
+
+    async fn connect<A: ToSocketAddrs + Send>(
+        addr: A,
+        tx: Sender<String>,
+    ) -> std::io::Result<Self> {
+        let inner = TcpStream::connect(addr).await?;
+        Ok(Self::new(inner, tx, false))
+    }
+}

filtermail/src/transport.rs

@@ -4,6 +4,7 @@ mod worker;
 use crate::config::Config;
 use crate::smtp_responses::{LOCAL_ERROR_451, WORKER_BUSY_421};
 use crate::smtp_server::{SmtpHandler, Transaction};
+use crate::tcp::{TcpConnect, TcpStreamTrait};
 use crate::utils::AddressDomain;
 use async_trait::async_trait;
 use std::collections::BTreeMap;
@@ -15,11 +16,15 @@ use worker::{WorkerMessage, WorkerPool};
 pub const HEADER_MAIL_FROM: &str = "X-MAIL-FROM";
 pub const HEADER_RCPT_TO: &str = "X-MAIL-TO";
 
-pub struct TransportHandler {
+pub struct TransportHandler<S: TcpConnect> {
-    workers: WorkerPool,
+    workers: WorkerPool<S>,
 }
 
-impl TransportHandler {
+impl<S> TransportHandler<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+    S::ConnectionContext: Default,
+{
     /// Creates a new [`TransportHandler`].
     pub fn new(config: Config) -> Result<Self, crate::error::Error> {
         let workers = WorkerPool::new(config)?;
@@ -44,7 +49,11 @@ pub struct TransactionState {
 }
 
 #[async_trait]
-impl SmtpHandler for TransportHandler {
+impl<S> SmtpHandler for TransportHandler<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+    S::ConnectionContext: Default,
+{
     type State = TransactionState;
 
     fn handle_rcpt_to(
@@ -190,8 +199,90 @@ impl SmtpHandler for TransportHandler {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::smtp_client::SmtpConnectionPool;
+    use crate::smtp_server::{Envelope, MockHandler, run_smtp_server};
+    use crate::tcp::rec_stream::RecTcpStream;
     use rstest::{fixture, rstest};
+    use serial_test::serial;
+    use std::sync::Arc;
+    use std::time::Duration;
     use testresult::TestResult;
+    use tokio::net::{TcpSocket, TcpStream};
+    use tokio::sync::mpsc::Receiver;
+
+    const FILTERMAIL_IP: &str = "127.0.0.1";
+    const FILTERMAIL_PORT: u16 = 10083;
+    const FILTERMAIL_ADDR: (&str, u16) = (FILTERMAIL_IP, FILTERMAIL_PORT);
+
+    /// Spawns a mockup SMTP server that accepts anything on `localhost:10025`.
+    ///
+    /// Returns a receiver that receives records of SMTP conversations.
+    fn spawn_mock_mta() -> TestResult<Receiver<String>> {
+        let socket = TcpSocket::new_v4()?;
+        socket.set_nodelay(true)?;
+        socket.set_reuseport(true)?;
+        socket.bind("127.0.0.1:10025".parse()?)?;
+        let remote_listener = socket.listen(8)?;
+        let (tx, rx) = tokio::sync::mpsc::channel(128);
+        tokio::spawn(async move {
+            while let Ok((stream, _)) = remote_listener.accept().await {
+                let tx_clone = tx.clone();
+                let rec_stream = RecTcpStream::new(stream, tx_clone, true);
+                tokio::spawn(async move {
+                    crate::smtp_server::handle_connection(
+                        rec_stream,
+                        Arc::new(MockHandler),
+                        9999, // arbitrary
+                        true,
+                    )
+                    .await
+                    .unwrap();
+                });
+            }
+        });
+        Ok(rx)
+    }
+
+    /// Spawns filtermail-transport.
+    ///
+    /// Returns a pointer to the underlying handler.
+    fn spawn_filtermail_transport() -> TestResult<Arc<TransportHandler<TcpStream>>> {
+        let config = Config::default();
+        let transport = Arc::new(TransportHandler::with_queue_size(config.clone(), 1)?);
+        tokio::spawn(run_smtp_server(
+            &FILTERMAIL_ADDR,
+            transport.clone(),
+            config.max_message_size,
+        ));
+        Ok(transport)
+    }
+
+    /// Sends envelope over LMTP.
+    ///
+    /// Returns a recorded LMTP conversation.
+    ///
+    /// Does not fail on negative response.
+    async fn lmtp_send(envelope: &Envelope) -> TestResult<String> {
+        let (tx, mut rx) = tokio::sync::mpsc::channel(128);
+        let client_config = crate::smtp_client::ClientConfig {
+            client_hostname: "postfix",
+            tls_config: None,
+            lmtp: true,
+        };
+        let _ = crate::smtp_client::send(
+            FILTERMAIL_IP,
+            FILTERMAIL_PORT,
+            envelope,
+            client_config,
+            Arc::new(crate::utils::build_resolver()?),
+            SmtpConnectionPool::<RecTcpStream>::new(tx),
+        )
+        .await;
+
+        let record = rx.recv().await.unwrap();
+
+        Ok(record)
+    }
 
     #[fixture]
     fn addrs1() -> Vec<String> {
@@ -214,7 +305,8 @@ mod tests {
     #[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 transport_handler =
+            TransportHandler::<TcpStream>::with_queue_size(Config::default(), 1)?;
         let domain1 = AddressDomain::from_str(addrs1.first().unwrap())?;
         let domain2 = AddressDomain::from_str(addrs2.first().unwrap())?;
 
@@ -263,4 +355,90 @@ mod tests {
 
         Ok(())
     }
+
+    #[rstest]
+    #[serial]
+    #[tokio::test]
+    async fn test_smtp_send_mail() -> TestResult {
+        let mut remote_mta = spawn_mock_mta()?;
+        spawn_filtermail_transport()?;
+
+        let envelope = Envelope {
+            mail_from: "sender@here".to_string(),
+            rcpt_to: vec![
+                // Taking advantage of the fact that localhost and [127.0.0.1] are recognized as
+                // different destinations.
+                "a1@localhost".to_string(),
+                "a2@localhost".to_string(),
+                "b1@[127.0.0.1]".to_string(),
+                "b2@[127.0.0.1]".to_string(),
+            ],
+            data: "message\r\n".as_bytes().to_vec(),
+        };
+
+        let record = lmtp_send(&envelope).await?;
+
+        let mut remote_records = [
+            remote_mta.recv().await.unwrap(),
+            remote_mta.recv().await.unwrap(),
+        ];
+        remote_records.sort_by_key(|s| s.contains("b1@[127.0.0.1]"));
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        assert!(remote_mta.is_empty());
+
+        insta::assert_snapshot!(format!(
+            "[postfix -> filtermail-transport]\r\n{record}\r\n\
+            [filtermail-transport -> destination A]\r\n{}\r\n\
+            [filtermail-transport -> destination B]\r\n{}",
+            remote_records[0], remote_records[1]
+        ));
+
+        Ok(())
+    }
+
+    #[rstest]
+    #[serial]
+    #[tokio::test]
+    async fn test_smtp_send_mail_defer() -> TestResult {
+        let mut remote_mta = spawn_mock_mta()?;
+        let transport = spawn_filtermail_transport()?;
+
+        let mut envelope = Envelope {
+            mail_from: "sender@here".to_string(),
+            rcpt_to: vec!["a1@localhost".to_string(), "b1@[127.0.0.1]".to_string()],
+            data: "message\r\n".as_bytes().to_vec(),
+        };
+
+        let (record_postfix_1, record_filtermail_1) = {
+            // simulate full queue on [127.0.0.1] worker
+            let _permit = transport
+                .workers
+                .get_permit(&AddressDomain::Literal("127.0.0.1".to_string()));
+
+            let record_postfix = lmtp_send(&envelope).await?;
+
+            let record_filtermail = remote_mta.recv().await.unwrap();
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            assert!(remote_mta.is_empty());
+            (record_postfix, record_filtermail)
+        };
+
+        // retry deferred
+        envelope.rcpt_to.remove(0);
+        let record_postfix_2 = lmtp_send(&envelope).await?;
+        let record_filtermail_2 = remote_mta.recv().await.unwrap();
+        tokio::time::sleep(Duration::from_secs(1)).await;
+        assert!(remote_mta.is_empty());
+
+        insta::assert_snapshot!(format!(
+            "TRANSACTION 1\r\n\
+            [postfix -> filtermail-transport]\r\n{record_postfix_1}\r\n\
+            [filtermail-transport -> destination A]\r\n{record_filtermail_1}\r\n\r\n\
+            TRANSACTION 2\r\n\
+            [postfix -> filtermail-transport]\r\n{record_postfix_2}\r\n\
+            [filtermail-transport -> destination B]\r\n{record_filtermail_2}",
+        ));
+
+        Ok(())
+    }
 }

filtermail/src/transport/worker.rs

@@ -2,6 +2,7 @@ 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::tcp::{TcpConnect, TcpStreamTrait};
 use crate::transport::{HEADER_MAIL_FROM, HEADER_RCPT_TO, https_client::HttpsClient};
 use crate::utils::{AddressDomain, build_resolver};
 use hickory_resolver::TokioResolver;
@@ -18,6 +19,16 @@ use tokio::task;
 use tokio::task::JoinHandle;
 use tokio_rustls::rustls;
 
+#[cfg(not(test))]
+const SMTP_PORT: u16 = 25;
+#[cfg(not(test))]
+const SMTP_SKIP_TLS: bool = false;
+
+#[cfg(test)]
+const SMTP_PORT: u16 = 10025;
+#[cfg(test)]
+const SMTP_SKIP_TLS: bool = true;
+
 /// Message queue size per [`Worker`].
 ///
 /// If a queue to a single destination reaches this limit,
@@ -26,17 +37,21 @@ const PER_DESTINATION_QUEUE_SIZE: usize = 30;
 
 type SMTPResponse = Result<String, String>;
 
-pub struct WorkerPool {
+pub struct WorkerPool<S: TcpConnect> {
     inner: RwLock<BTreeMap<AddressDomain, Arc<Worker>>>,
     client_hostname: String,
-    smtp_connection_pool: Arc<SmtpConnectionPool>,
+    smtp_connection_pool: Arc<SmtpConnectionPool<S>>,
     mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
     monitor_handle: JoinHandle<()>,
     dns_resolver: Arc<TokioResolver>,
     queue_size: usize,
 }
 
-impl WorkerPool {
+impl<S> WorkerPool<S>
+where
+    S: TcpStreamTrait + TcpConnect,
+    S::ConnectionContext: Default,
+{
     pub fn new(config: Config) -> Result<Self, crate::error::Error> {
         let dns_resolver = Arc::new(build_resolver()?);
 
@@ -53,7 +68,7 @@ impl WorkerPool {
             inner: Default::default(),
             client_hostname: config.mail_domain,
             dns_resolver,
-            smtp_connection_pool: SmtpConnectionPool::new(),
+            smtp_connection_pool: SmtpConnectionPool::<S>::new(Default::default()),
             mxdeliv_unsupported_hosts: mxdeliv_cache,
             monitor_handle,
             queue_size: PER_DESTINATION_QUEUE_SIZE,
@@ -131,7 +146,7 @@ impl WorkerPool {
     }
 }
 
-impl Drop for WorkerPool {
+impl<S: TcpConnect> Drop for WorkerPool<S> {
     fn drop(&mut self) {
         self.monitor_handle.abort();
     }
@@ -150,14 +165,17 @@ impl Drop for Worker {
 }
 
 impl Worker {
-    pub async fn run(
+    pub async fn run<S>(
         destination: AddressDomain,
         mut rx: mpsc::Receiver<WorkerMessage>,
         client_hostname: String,
-        smtp_connection_pool: Arc<SmtpConnectionPool>,
+        smtp_connection_pool: Arc<SmtpConnectionPool<S>>,
         mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
         dns_resolver: Arc<TokioResolver>,
-    ) -> Result<(), crate::error::Error> {
+    ) -> Result<(), crate::error::Error>
+    where
+        S: TcpStreamTrait + TcpConnect,
+    {
         let worker_id = task::try_id()
             .map(|id| id.to_string())
             .unwrap_or("?".to_string());
@@ -199,18 +217,21 @@ impl Worker {
 
     /// Handles a single email transaction for a single recipient domain.
     #[expect(clippy::too_many_arguments)]
-    async fn handle_single_domain(
+    async fn handle_single_domain<S>(
         tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
-        smtp_connection_pool: Arc<SmtpConnectionPool>,
+        smtp_connection_pool: Arc<SmtpConnectionPool<S>>,
         mxdeliv_unsupported_hosts: Arc<retainer::Cache<String, ()>>,
         https_client: HttpsClient,
         dns_resolver: Arc<TokioResolver>,
         domain: AddressDomain,
         envelope: Envelope,
         client_hostname: String,
-    ) -> Result<String, String> {
+    ) -> Result<String, String>
+    where
+        S: TcpStreamTrait + TcpConnect,
+    {
         let mut allow_invalid_cert = false;
-        let mut skip_tls = false; // only respected by smtp channel
+        let mut skip_tls = SMTP_SKIP_TLS; // only respected by smtp channel
 
         let mx_hosts = match domain {
             // no-DNS setup; assume the ip from email address is the destination.
@@ -316,12 +337,16 @@ impl Worker {
             }
 
             // SMTP channel (fallback)
+            let client_config = crate::smtp_client::ClientConfig {
+                client_hostname: &client_hostname,
+                tls_config: tls_config.clone(),
+                lmtp: false,
+            };
             match crate::smtp_client::send(
                 &mx_host,
-                25,
+                SMTP_PORT,
                 &envelope,
-                &client_hostname,
+                client_config,
-                tls_config.clone(),
                 dns_resolver.clone(),
                 smtp_connection_pool.clone(),
             )