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

test: Test filtermail-transport

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

Signed-off-by: Jagoda Ślązak <jslazak@jslazak.com>
This commit is contained in:
Jagoda Ślązak
2026-06-22 18:27:30 +02:00
parent 7ea7904ea0
commit c7452526fa
13 changed files with 834 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
filtermail/Cargo.lock
Generated
+86
View File
@@ -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
+2
View File
@@ -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
+22 -8
View File
@@ -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,
None,
client_config,
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
+10 -3
View File
@@ -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
+19 -7
View File
@@ -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,
None,
client_config,
self.dns_resolver.clone(),
self.smtp_connection_pool.clone(),
)
filtermail/src/smtp_client.rs
+98 -61
View File
@@ -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 stream: BufStream<SmtpStream>,
pub struct SmtpConnection<S> {
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 {
pool: Arc<retainer::Cache<(String, u16), SmtpConnection>>,
pub struct SmtpConnectionPool<S>
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> {
let mut set: JoinSet<tokio::io::Result<TcpStream>> = JoinSet::new();
context: S::ConnectionContext,
) -> 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 =
tokio::time::timeout(Duration::from_secs(60), TcpStream::connect(addr)).await??;
let stream: S =
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,
tls_config: Option<TlsConfig>,
config: ClientConfig<'_>,
dns_resolver: Arc<TokioResolver>,
pool: Arc<SmtpConnectionPool>,
) -> Result<(), crate::error::Error> {
let (mut buf_stream, reused, mut pipelining) = if let Some(connection) =
pool.take(address, port).await
{
log::debug!(
"Reusing existing connection to {}",
connection.stream.get_ref().format_host(address)
);
if tls_config.is_some() {
// This should never happen,
// assert to make sure we never accidentally use a plain connection while expecting TLS.
assert!(
matches!(connection.stream.get_ref(), SmtpStream::Tls(_)),
"Expected TLS stream from pool, but got plain stream."
pool: Arc<SmtpConnectionPool<S>>,
) -> Result<(), crate::error::Error>
where
S: TcpStreamTrait + TcpConnect,
{
let greeting = if config.lmtp { "LHLO" } else { "EHLO" };
let (mut buf_stream, reused, mut pipelining) =
if let Some(connection) = pool.take(address, port).await {
log::debug!(
"Reusing existing connection to {}",
connection.stream.get_ref().format_host(address)
);
}
(connection.stream, true, connection.pipelining)
} else {
let stream =
SmtpStream::plain(establish_tcp_connection(address, port, dns_resolver.clone()).await?);
log::debug!("Successfully connected to {}", stream.format_host(address));
(BufStream::new(stream), false, false)
};
if config.tls_config.is_some() {
// This should never happen,
// assert to make sure we never accidentally use a plain connection while expecting TLS.
assert!(
matches!(connection.stream.get_ref(), SmtpStream::Tls(_)),
"Expected TLS stream from pool, but got plain stream."
);
}
(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(),
"EHLO",
format!("{greeting} {}\r\n", { config.client_hostname }).as_bytes(),
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
+51 -58
View File
@@ -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::net::{TcpListener, TcpStream};
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufStream};
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 reader = BufReader::new(reader);
let mut writer = BufWriter::new(writer);
let mut bufstream = BufStream::new(stream);
let mut line = String::new();
writer.write_all(b"220 filtermail SMTP\r\n").await?;
writer.flush().await?;
macro_rules! smtp_write {
($($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?;
writer.flush().await?;
smtp_write!("250-filtermail\r\n250 OK\r\n");
} 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
.write_all(b"250-filtermail\r\n250-8BITMIME\r\n250 OK\r\n")
.await?;
writer.flush().await?;
smtp_write!("250-filtermail\r\n250-8BITMIME\r\n250 OK\r\n");
} 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?;
writer.flush().await?;
smtp_write!("{OK_250}\r\n");
} 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?;
writer.flush().await?;
smtp_write!("{}\r\n", e);
continue 'connection;
}
transaction.envelope.mail_from = from;
writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
writer.flush().await?;
smtp_write!("{OK_250}\r\n");
} else {
log::warn!("Invalid MAIL FROM command. Can't extract address. Received: {cmd}");
writer
.write_all(b"500 Invalid address in MAIL FROM\r\n")
.await?;
writer.flush().await?;
smtp_write!("500 Invalid address in MAIL FROM\r\n");
}
} 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?;
writer.flush().await?;
smtp_write!("{}\r\n", e);
continue 'connection;
}
transaction.envelope.rcpt_to.push(to);
writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
writer.flush().await?;
smtp_write!("{OK_250}\r\n");
}
} 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?;
smtp_write!("{}\r\n", e);
continue 'connection;
}
writer
.write_all(b"354 End data with <CR><LF>.<CR><LF>\r\n")
.await?;
writer.flush().await?;
smtp_write!("354 End data with <CR><LF>.<CR><LF>\r\n");
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
.write_all(b"552 Message exceeds maximum size\r\n")
.await?;
writer.flush().await?;
smtp_write!("552 Message exceeds maximum size\r\n");
continue 'connection;
}
}
@@ -317,33 +316,27 @@ where
match handler.handle_data_dot(&mut transaction).await {
Ok(response) => {
log::debug!("Sent: {response}");
writer
.write_all(format!("{}\r\n", response).as_bytes())
.await?;
writer.flush().await?;
smtp_write!("{}\r\n", response);
}
Err(e) => {
log::debug!("Sent: {e}");
writer.write_all(format!("{}\r\n", e).as_bytes()).await?;
writer.flush().await?;
smtp_write!("{}\r\n", e);
}
}
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?;
writer.flush().await?;
smtp_write!("221 OK\r\n");
break 'connection;
} else if cmd.to_uppercase().starts_with("RSET") {
transaction = Transaction::default();
writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
writer.flush().await?;
smtp_write!("{OK_250}\r\n");
} else if cmd.to_uppercase().starts_with("NOOP") {
writer.write_all(format!("{OK_250}\r\n").as_bytes()).await?;
writer.flush().await?;
smtp_write!("{OK_250}\r\n");
} else {
writer.write_all(b"500 Command not recognized\r\n").await?;
writer.flush().await?;
smtp_write!("500 Command not recognized\r\n");
}
}
filtermail/src/snapshots/filtermail__transport__tests__smtp_send_mail.snap
+64
View File
@@ -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
+73
View File
@@ -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
+59
View File
@@ -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
+127
View File
@@ -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
+183 -5
View File
@@ -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 {
workers: WorkerPool,
pub struct TransportHandler<S: TcpConnect> {
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
+40 -15
View File
@@ -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,
tls_config.clone(),
client_config,
dns_resolver.clone(),
smtp_connection_pool.clone(),
)