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

feat(transport): Remote delivery over SMTP (#104)

Browse Source 
Implements a new mode "transport",
that allows filtermail to be used for
remote delivery.

In transport mode, filtermail listens for LMTP
connections, splits messages by domain
and performs delivery to remote MTAs over SMTP.

SMTP client tries to open socket on all resolved
addresses in parallel and uses one that succeeds
the first, fixing the issue described in:
chatmail/relay#900

Extends the built-in SMTP client
with STARTTLS support.

Extends the built-in SMTP server
with LMTP greeting support.

Groundwork required for mxdeliv endpoint
(HTTP channel for MTA-to-MTA communication):
chatmail/relay#900

Signed-off-by: Jagoda Ślązak <jslazak@jslazak.com>
This commit is contained in:
Jagoda Estera Ślązak
2026-04-15 22:12:09 +02:00
committed by GitHub
parent ec0413bd5f
commit cf64778f71
15 changed files with 816 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
filtermail/Cargo.lock
Generated
+64 -6
View File
@@ -98,9 +98,9 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
[[package]]
name = "bitflags"
version = "2.10.0"
version = "2.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3"
checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af"
[[package]]
name = "block-buffer"
@@ -408,7 +408,10 @@ dependencies = [
"testresult",
"thiserror",
"tokio",
"tokio-io-timeout",
"tokio-rustls",
"viadkim",
"webpki-roots",
]
[[package]]
@@ -933,9 +936,9 @@ dependencies = [
[[package]]
name = "num-conv"
version = "0.2.0"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf97ec579c3c42f953ef76dbf8d55ac91fb219dde70e49aa4a6b7d74e9919050"
checksum = "c6673768db2d862beb9b39a78fdcb1a69439615d5794a1be50caa9bc92c81967"
[[package]]
name = "num-integer"
@@ -1336,6 +1339,21 @@ dependencies = [
"semver",
]
[[package]]
name = "rustls"
version = "0.23.38"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69f9466fb2c14ea04357e91413efb882e2a6d4a406e625449bc0a5d360d53a21"
dependencies = [
"log",
"once_cell",
"ring",
"rustls-pki-types",
"rustls-webpki",
"subtle",
"zeroize",
]
[[package]]
name = "rustls-pki-types"
version = "1.14.0"
@@ -1345,6 +1363,17 @@ dependencies = [
"zeroize",
]
[[package]]
name = "rustls-webpki"
version = "0.103.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "20a6af516fea4b20eccceaf166e8aa666ac996208e8a644ce3ef5aa783bc7cd4"
dependencies = [
"ring",
"rustls-pki-types",
"untrusted",
]
[[package]]
name = "rustversion"
version = "1.0.22"
@@ -1609,9 +1638,9 @@ checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"
[[package]]
name = "tokio"
version = "1.51.0"
version = "1.51.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd"
checksum = "f66bf9585cda4b724d3e78ab34b73fb2bbaba9011b9bfdf69dc836382ea13b8c"
dependencies = [
"bytes",
"libc",
@@ -1624,6 +1653,16 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "tokio-io-timeout"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0bd86198d9ee903fedd2f9a2e72014287c0d9167e4ae43b5853007205dda1b76"
dependencies = [
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-macros"
version = "2.7.0"
@@ -1635,6 +1674,16 @@ dependencies = [
"syn",
]
[[package]]
name = "tokio-rustls"
version = "0.26.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61"
dependencies = [
"rustls",
"tokio",
]
[[package]]
name = "toml_datetime"
version = "0.7.5+spec-1.1.0"
@@ -1867,6 +1916,15 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "webpki-roots"
version = "1.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22cfaf3c063993ff62e73cb4311efde4db1efb31ab78a3e5c457939ad5cc0bed"
dependencies = [
"rustls-pki-types",
]
[[package]]
name = "widestring"
version = "1.2.1"
filtermail/Cargo.toml
+7
View File
@@ -23,6 +23,13 @@ viadkim = { version = "0.2.0" }
hickory-resolver = { version = "0.25.2", features = ["dnssec-ring"] }
lru = "0.16.3"
parking_lot = "0.12.5"
tokio-rustls = { version = "0.26.4", default-features = false, features = [
"ring",
"logging",
"tls12",
] }
webpki-roots = "1.0.6"
tokio-io-timeout = "1.2.1"
[dev-dependencies]
rstest = "0.26.1"
filtermail/README.md
+18 -6
View File
@@ -11,16 +11,16 @@ and handles per-sender rate limiting.
## Usage
```plain
filtermail <config> (incoming|outgoing)
filtermail <config> (incoming|outgoing|transport)
```
where `<config>` is a path to `chatmail.ini` configuration file.
Filtermail can be used in `incoming` or `outgoing` mode that apply different settings
to filter either incoming or outgoing emails.
Filtermail can be used in `incoming`, `outgoing` or `transport` mode.
### Incoming mode
Filtermail in incoming mode performs following steps:
Filtermail in incoming mode acts as a proxy filter
for messages received from remote MTAs and performs following steps:
1. Rejects messages if `DATA` exceeds configured message size limit.
2. Rejects messages that do not meet at least one of the following criteria:
@@ -43,7 +43,8 @@ Filtermail in incoming mode performs following steps:
### Outgoing mode
Filtermail in outgoing mode performs following steps:
Filtermail in outgoing mode acts as a proxy filter
for messages received from clients and performs following steps:
1. Rejects messages at `MAIL FROM` stage if the address exceeded rate limit.
2. Rejects messages if `DATA` exceeds configured message size limit.
@@ -55,6 +56,15 @@ Filtermail in outgoing mode performs following steps:
- self-sent Autocrypt Setup Message,
- all recipients match `passthrough_recipients`.
### Transport mode
Filtermail in transport mode is used for final delivery to remote MTAs.
As opposed to incoming/outgoing, it accepts connections from postfix over LMTP instead of SMTP,
to allow returning per-recipient status back to postfix.
Received message is split per-domain and sent to recipients' MX servers over SMTP,
enforcing TLS.
As opposed to postfix, IPv4 and IPv6 connections are tried in parallel and first successful connection is used.
## Configuration
### chatmail.ini
@@ -66,6 +76,8 @@ but implements a custom parser that only requires a small subset of configuratio
defaults to `10080`.
- `filtermail_smtp_port_incoming` - port to listen on in incoming mode,
defaults to `10081`.
- `filtermail_lmtp_port_transport` - port to listen on in transport mode,
defaults to `10083`.
- `postfix_reinject_port` - port to reinject messages to postfix in outgoing mode,
defaults to `10025`.
- `postfix_reinject_port_incoming` - port to reinject messages to postfix in incoming mode,
@@ -113,7 +125,7 @@ Although unsupported, it may still work outside of this context or even without
with few considerations:
- Filtermail expects to receive messages from a trusted server,
and thus should not be exposed directly to the internet.
and thus should not listen on ports exposed directly to the internet.
- Issues outside of chatmail relay context are not necessarily considered bugs;
PRs fixing them are not guaranteed to be accepted.
(Trivial changes may still be considered,
filtermail/src/config.rs
+6
View File
@@ -14,6 +14,8 @@ pub struct Config {
pub filtermail_smtp_port: u16,
#[serde(default = "Config::default_filtermail_smtp_port_incoming")]
pub filtermail_smtp_port_incoming: u16,
#[serde(default = "Config::default_filtermail_lmtp_port_transport")]
pub filtermail_lmtp_port_transport: u16,
#[serde(default = "Config::default_postfix_host")]
pub postfix_host: String,
#[serde(default = "Config::default_postfix_reinject_port")]
@@ -108,6 +110,9 @@ impl Config {
const fn default_filtermail_smtp_port_incoming() -> u16 {
10081
}
const fn default_filtermail_lmtp_port_transport() -> u16 {
10083
}
fn default_postfix_host() -> String {
"127.0.0.1".to_owned()
}
@@ -138,6 +143,7 @@ impl Default for Config {
filtermail_host: Self::default_filtermail_host(),
filtermail_smtp_port: Self::default_filtermail_smtp_port(),
filtermail_smtp_port_incoming: Self::default_filtermail_smtp_port_incoming(),
filtermail_lmtp_port_transport: Self::default_filtermail_lmtp_port_transport(),
postfix_host: Self::default_postfix_host(),
postfix_reinject_port: Self::default_postfix_reinject_port(),
postfix_reinject_port_incoming: Self::default_postfix_reinject_port_incoming(),
filtermail/src/dkim_verifier.rs
+7 -22
View File
@@ -1,4 +1,3 @@
use hickory_resolver::name_server::TokioConnectionProvider;
use hickory_resolver::{Name, TokioResolver};
use lru::LruCache;
use std::io;
@@ -29,7 +28,7 @@ fn normalize_rdata(txt_data: &str) -> String {
/// DNS resolver for DKIM TXT records, that caches RDATA in memory.
#[derive(Clone)]
struct CachedResolver {
dns_resolver: TokioResolver,
dns_resolver: Arc<TokioResolver>,
// Note: Arc is required despite we are holding the whole handler in an Arc,
// because viadkim will internally clone the resolver (LookupTxt + Clone + 'static)
// to parallelize lookups in case of multiple signatures...
@@ -38,26 +37,13 @@ struct CachedResolver {
impl CachedResolver {
/// Creates a new [`CachedResolver`].
pub fn new() -> Result<Self, crate::error::Error> {
// Use resolv.conf
let dns_resolver = {
let mut builder = TokioResolver::builder(TokioConnectionProvider::default())?;
// https://github.com/hickory-dns/hickory-dns/issues/3519
builder.options_mut().validate = true;
builder.build()
};
assert!(
dns_resolver.options().validate,
"incorrect resolver config: DNSSEC disabled; exiting"
);
pub fn new(dns_resolver: Arc<TokioResolver>) -> Self {
let cache = Arc::new(parking_lot::Mutex::new(LruCache::new(LRU_CACHE_CAPACITY)));
Ok(Self {
Self {
dns_resolver,
cache,
})
}
}
/// Invalidates the cached RDATA for a given selector and domain.
@@ -148,7 +134,6 @@ impl LookupTxt for MockResolver {
}
/// Either a real resolver or a mock.
#[allow(clippy::large_enum_variant)]
#[derive(Clone)]
enum Resolver {
/// A [`CachedResolver`]
@@ -190,13 +175,13 @@ pub struct DkimVerifier {
impl DkimVerifier {
/// Creates a new [`DkimVerifier`] with the provided resolver.
pub fn new() -> Result<Self, crate::error::Error> {
let resolver = CachedResolver::new()?.into();
pub fn new(dns_resolver: Arc<TokioResolver>) -> Self {
let resolver = CachedResolver::new(dns_resolver).into();
let config = viadkim::Config {
lookup_timeout: Duration::from_secs(60),
..Default::default()
};
Ok(Self { resolver, config })
Self { resolver, config }
}
/// Creates a new [`DkimVerifier`] with a mock resolver that always returns the provided TXT record.
filtermail/src/error.rs
+17
View File
@@ -1,5 +1,7 @@
//! Error types.
use tokio_rustls::rustls;
/// Error type for filtermail.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
@@ -19,6 +21,12 @@ pub enum Error {
},
#[error("Invalid email address: {0}")]
InvalidEmailAddress(String),
#[error("Failed to connect to any of the following addresses: {0:?}")]
ConnectionFailed(Vec<String>),
#[error(transparent)]
Tls(#[from] rustls::Error),
#[error(transparent)]
InvalidDnsName(#[from] rustls::pki_types::InvalidDnsNameError),
}
impl Error {
@@ -34,4 +42,13 @@ impl Error {
_ => "451 Local error".to_string(),
}
}
/// Same as [`smtp_response`](Self::smtp_response) but formats the same response
/// for each recipient, as expected by LMTP.
pub fn lmtp_response(&self, recipient_count: usize) -> String {
let response = self.smtp_response();
std::iter::repeat_n(response, recipient_count)
.collect::<Vec<_>>()
.join("\r\n")
}
}
filtermail/src/inbound.rs
+21 -14
View File
@@ -6,29 +6,29 @@ use crate::dkim_verifier::DkimVerifier;
use crate::message::{check_encrypted, is_securejoin};
pub use crate::smtp_server::Envelope;
use crate::smtp_server::SmtpHandler;
use crate::utils::{AddressDomain, extract_address, log_eml};
use crate::utils::{AddressDomain, build_resolver, extract_address, log_eml};
use async_trait::async_trait;
use hickory_resolver::TokioResolver;
use mailparse::{MailHeaderMap, parse_mail};
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
/// Handler for incoming SMTP messages.
pub struct IncomingBeforeQueueHandler {
config: Config,
dns_resolver: Arc<TokioResolver>,
dkim_verifier: DkimVerifier,
skip_dkim: bool,
reinject_addr: SocketAddr,
}
impl IncomingBeforeQueueHandler {
pub fn new(config: Config, skip_dkim: bool) -> Result<Self, crate::error::Error> {
let reinject_addr =
crate::resolve_addr(&config.postfix_host, config.postfix_reinject_port_incoming)?;
let dns_resolver = Arc::new(build_resolver()?);
Ok(Self {
config,
dkim_verifier: DkimVerifier::new()?,
dns_resolver: dns_resolver.clone(),
dkim_verifier: DkimVerifier::new(dns_resolver),
skip_dkim,
reinject_addr,
})
}
@@ -144,13 +144,20 @@ impl SmtpHandler for IncomingBeforeQueueHandler {
async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String> {
log::debug!("Re-injecting the mail that passed checks");
crate::smtp_client::send(self.reinject_addr, envelope)
.await
.map_err(|e| {
log::warn!("Failed to re-inject mail: {}", e);
e.smtp_response()
})?;
let hostname = format!("[{}]", self.config.filtermail_host);
crate::smtp_client::send(
&self.config.postfix_host,
self.config.postfix_reinject_port_incoming,
envelope,
&hostname,
None,
self.dns_resolver.clone(),
)
.await
.map_err(|e| {
log::warn!("Failed to re-inject mail: {}", e);
e.smtp_response()
})?;
Ok(())
}
filtermail/src/main.rs
+73 -44
View File
@@ -33,40 +33,45 @@ pub(crate) mod openpgp;
pub(crate) mod outbound;
pub(crate) mod smtp_client;
pub(crate) mod smtp_server;
mod tls;
mod transport;
pub(crate) mod utils;
use crate::transport::TransportHandler;
use config::Config;
use env_logger::Env;
use inbound::IncomingBeforeQueueHandler;
use outbound::OutgoingBeforeQueueHandler;
use smtp_server::run_smtp_server;
use std::env;
use std::net::{SocketAddr, ToSocketAddrs};
use std::process;
use std::str::FromStr;
use std::sync::Arc;
const ENCRYPTION_NEEDED_523: &str = "523 Encryption Needed: Invalid Unencrypted Mail";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
enum Mode {
Outgoing,
Incoming,
Transport,
}
impl std::str::FromStr for Mode {
impl FromStr for Mode {
type Err = &'static str;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
fn from_str(mode: &str) -> Result<Self, Self::Err> {
match mode {
"outgoing" => Ok(Mode::Outgoing),
"incoming" => Ok(Mode::Incoming),
_ => Err("Error: mode must be 'incoming' or 'outgoing'"),
"transport" => Ok(Mode::Transport),
_ => Err("Error: mode must be 'incoming', 'outgoing' or 'transport'"),
}
}
}
#[tokio::main]
async fn main() {
async fn main() -> Result<(), error::Error> {
// default to info level
let env = Env::new().filter_or("RUST_LOG", "info");
env_logger::Builder::from_env(env)
@@ -74,13 +79,17 @@ async fn main() {
.format_timestamp(None)
.init();
tokio_rustls::rustls::crypto::ring::default_provider()
.install_default()
.expect("Failed to set up rustls crypto provider.");
let args: Vec<String> = env::args().collect();
if args.len() != 3 {
eprintln!(
"Usage: {} <config_file> <mode>",
args.first().unwrap_or(&"filtermail".to_string())
);
eprintln!(" mode: incoming or outgoing");
eprintln!(" mode: incoming, outgoing or transport");
process::exit(1);
}
@@ -91,7 +100,7 @@ async fn main() {
unreachable!("args length checked above")
};
let mode = match mode.parse::<Mode>() {
let mode = match Mode::from_str(mode) {
Ok(mode) => mode,
Err(e) => {
eprintln!("{e}");
@@ -107,45 +116,65 @@ async fn main() {
}
};
if mode == Mode::Outgoing {
let addr = (config.filtermail_host, config.filtermail_smtp_port);
let handler = Arc::new(OutgoingBeforeQueueHandler::new(config.clone()).unwrap());
let max_size = config.max_message_size;
log::debug!("Outgoing SMTP server listening on {}:{}", addr.0, addr.1);
match mode {
Mode::Outgoing => {
let addr = (config.filtermail_host, config.filtermail_smtp_port);
let handler = Arc::new(OutgoingBeforeQueueHandler::new(config.clone())?);
let max_size = config.max_message_size;
log::debug!("Outgoing SMTP server listening on {}:{}", addr.0, addr.1);
if let Err(e) = run_smtp_server(&addr, handler, max_size).await {
eprintln!("Server error: {}", e);
process::exit(1);
if let Err(e) = run_smtp_server(&addr, handler, max_size).await {
eprintln!("Server error: {}", e);
process::exit(1);
}
}
} else {
// Skip DKIM verification (used for tests).
let skip_dkim = env::var("FILTERMAIL_SKIP_DKIM")
.map(|val| val == "1" || val.eq_ignore_ascii_case("true"))
.unwrap_or(false);
Mode::Incoming => {
// Skip DKIM verification (used for tests).
let skip_dkim = env::var("FILTERMAIL_SKIP_DKIM")
.map(|val| val == "1" || val.eq_ignore_ascii_case("true"))
.unwrap_or(false);
if skip_dkim {
log::warn!("DKIM verification DISABLED! This should not be used in production.");
if skip_dkim {
log::warn!("DKIM verification DISABLED! This should not be used in production.");
}
let handler = Arc::new(IncomingBeforeQueueHandler::new(config.clone(), skip_dkim)?);
let max_size = config.max_message_size;
let mut server_set = tokio::task::JoinSet::new();
let addr_smtp = (config.filtermail_host, config.filtermail_smtp_port_incoming);
let handler_smtp = handler.clone();
server_set
.spawn(async move { run_smtp_server(&addr_smtp, handler_smtp, max_size).await });
log::debug!(
"Incoming SMTP server listening on {}:{}",
addr_smtp.0,
addr_smtp.1
);
while let Some(result) = server_set.join_next().await {
if let Err(e) = result {
eprintln!("Server error: {}", e);
process::exit(1);
}
}
}
Mode::Transport => {
let addr = (
config.filtermail_host,
config.filtermail_lmtp_port_transport,
);
let handler = Arc::new(TransportHandler::new(config.clone())?);
let max_size = config.max_message_size;
log::debug!("Transport SMTP server listening on {}:{}", addr.0, addr.1);
let addr = (config.filtermail_host, config.filtermail_smtp_port_incoming);
let handler = Arc::new(
// We want to panic here if the handler cannot be created.
IncomingBeforeQueueHandler::new(config.clone(), skip_dkim).unwrap(),
);
let max_size = config.max_message_size;
log::debug!("Incoming SMTP server listening on {}:{}", addr.0, addr.1);
if let Err(e) = run_smtp_server(&addr, handler, max_size).await {
eprintln!("Server error: {}", e);
process::exit(1);
if let Err(e) = run_smtp_server(&addr, handler, max_size).await {
eprintln!("Server error: {}", e);
process::exit(1);
}
}
}
}
fn resolve_addr(host: &str, port: u16) -> Result<SocketAddr, error::Error> {
log::debug!("Resolving {host}");
Ok((host, port)
.to_socket_addrs()?
.next()
.ok_or(std::io::Error::other("Cannot resolve host"))?)
};
Ok(())
}
filtermail/src/outbound.rs
+20 -13
View File
@@ -5,28 +5,28 @@ use crate::config::Config;
use crate::message::{check_encrypted, is_securejoin, recipient_matches_passthrough};
pub use crate::smtp_server::Envelope;
use crate::smtp_server::SmtpHandler;
use crate::utils::extract_address;
use crate::utils::{build_resolver, extract_address};
use async_trait::async_trait;
use governor::{DefaultKeyedRateLimiter, Quota, RateLimiter};
use hickory_resolver::TokioResolver;
use mailparse::{MailHeaderMap, parse_mail};
use std::net::SocketAddr;
use std::sync::Arc;
/// Handler for outgoing SMTP messages.
pub struct OutgoingBeforeQueueHandler {
config: Config,
reinject_addr: SocketAddr,
dns_resolver: Arc<TokioResolver>,
send_rate_limiter: DefaultKeyedRateLimiter<String>,
}
impl OutgoingBeforeQueueHandler {
pub fn new(config: Config) -> Result<Self, crate::error::Error> {
let reinject_addr =
crate::resolve_addr(&config.postfix_host, config.postfix_reinject_port)?;
let quota = Quota::per_minute(config.max_user_send_per_minute)
.allow_burst(config.max_user_send_burst_size);
let dns_resolver = Arc::new(build_resolver()?);
Ok(Self {
config,
reinject_addr,
dns_resolver,
send_rate_limiter: RateLimiter::keyed(quota),
})
}
@@ -142,13 +142,20 @@ impl SmtpHandler for OutgoingBeforeQueueHandler {
async fn reinject_mail(&self, envelope: &Envelope) -> Result<(), String> {
log::debug!("Re-injecting the mail that passed checks");
crate::smtp_client::send(self.reinject_addr, envelope)
.await
.map_err(|e| {
log::warn!("Failed to re-inject mail: {}", e);
e.smtp_response()
})?;
let hostname = format!("[{}]", self.config.filtermail_host);
crate::smtp_client::send(
&self.config.postfix_host,
self.config.postfix_reinject_port,
envelope,
&hostname,
None,
self.dns_resolver.clone(),
)
.await
.map_err(|e| {
log::warn!("Failed to re-inject mail: {}", e);
e.smtp_response()
})?;
Ok(())
}
filtermail/src/smtp_client.rs
+225 -13
View File
@@ -1,22 +1,173 @@
use crate::smtp_server::Envelope;
use std::net::SocketAddr;
use hickory_resolver::TokioResolver;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufStream};
use tokio::net::TcpSocket;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::net::TcpStream;
use tokio::task::JoinSet;
use tokio_io_timeout::TimeoutStream;
use tokio_rustls::rustls::client::ClientSessionMemoryCache;
/// A [`TcpStream`] used for SMTP communication.
#[expect(clippy::large_enum_variant)]
enum SmtpStream {
/// A plain TCP stream.
Plain(Pin<Box<TimeoutStream<TcpStream>>>),
/// A TLS-encrypted stream.
Tls(tokio_rustls::TlsStream<Pin<Box<TimeoutStream<TcpStream>>>>),
}
impl SmtpStream {
/// 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 {
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)));
Self::Plain(Box::pin(timeout_stream))
}
}
#[derive(Debug, Clone)]
pub struct TlsConfig {
pub(crate) allow_invalid_cert: bool,
pub(crate) session_cache: Arc<ClientSessionMemoryCache>,
}
impl AsyncWrite for SmtpStream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
match self.get_mut() {
SmtpStream::Plain(stream) => Pin::new(stream).poll_write(cx, buf),
SmtpStream::Tls(stream) => Pin::new(stream).poll_write(cx, buf),
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
match self.get_mut() {
SmtpStream::Plain(stream) => Pin::new(stream).poll_flush(cx),
SmtpStream::Tls(stream) => Pin::new(stream).poll_flush(cx),
}
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
match self.get_mut() {
SmtpStream::Plain(stream) => Pin::new(stream).poll_shutdown(cx),
SmtpStream::Tls(stream) => Pin::new(stream).poll_shutdown(cx),
}
}
}
impl AsyncRead for SmtpStream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
match self.get_mut() {
SmtpStream::Plain(stream) => Pin::new(stream).poll_read(cx, buf),
SmtpStream::Tls(stream) => Pin::new(stream).poll_read(cx, buf),
}
}
}
/// Converts address and port to a list of socket addresses.
///
/// Performs non-blocking DNS resolution if address is a domain name,
/// or returns a single socket address if address is an IP.
async fn to_socket_addrs(
address: &str,
port: u16,
dns_resolver: Arc<TokioResolver>,
) -> Result<Vec<std::net::SocketAddr>, crate::error::Error> {
log::trace!("Resolving {address}...");
if let Ok(ip) = address.parse() {
Ok(vec![std::net::SocketAddr::new(ip, port)])
} else {
let lookup = dns_resolver.lookup_ip(address).await?;
Ok(lookup
.iter()
.map(|ip| std::net::SocketAddr::new(ip, port))
.collect())
}
}
/// Establishes a TCP connection to the given address and port, trying all resolved IPs in parallel.
async fn establish_tcp_connection(
address: &str,
port: u16,
dns_resolver: Arc<TokioResolver>,
) -> Result<TcpStream, crate::error::Error> {
let mut set: JoinSet<tokio::io::Result<TcpStream>> = JoinSet::new();
let socket_addrs = to_socket_addrs(address, port, dns_resolver).await?;
for addr in socket_addrs.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??;
stream.set_nodelay(true)?;
Ok(stream)
});
}
let mut stream: Option<TcpStream> = None;
while let Some(result) = set.join_next().await {
match result {
Ok(Ok(s)) => {
stream = Some(s);
break;
}
Ok(Err(e)) => log::debug!("Failed to connect to socket: {e}"),
Err(e) => log::debug!("Failed to join task: {e}"),
}
}
match stream {
Some(s) => Ok(s),
None => Err(crate::error::Error::ConnectionFailed(
socket_addrs.into_iter().map(|a| a.to_string()).collect(),
)),
}
}
/// Sends an email using an SMTP server at `smtp_addr`.
pub async fn send(smtp_addr: SocketAddr, envelope: &Envelope) -> Result<(), crate::error::Error> {
let socket = TcpSocket::new_v4()?;
///
/// 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.
pub async fn send(
address: &str,
port: u16,
envelope: &Envelope,
client_hostname: &str,
tls_config: Option<TlsConfig>,
dns_resolver: Arc<TokioResolver>,
) -> Result<(), crate::error::Error> {
let stream = establish_tcp_connection(address, port, dns_resolver).await?;
// Disable Nagle's algorithm.
socket.set_nodelay(true)?;
log::debug!(
"SMTP client: successfully connected to {}",
stream.peer_addr()?
);
let stream = socket.connect(smtp_addr).await?;
let mut buf_stream = BufStream::new(stream);
let mut buf_stream = BufStream::new(SmtpStream::plain(stream));
let mut response = String::new();
macro_rules! smtp_write {
($command: expr) => {
log::trace!(
"SMTP client: sending: {}",
String::from_utf8_lossy($command)
);
buf_stream.write_all($command).await?;
buf_stream.flush().await?;
};
@@ -24,14 +175,24 @@ pub async fn send(smtp_addr: SocketAddr, envelope: &Envelope) -> Result<(), crat
macro_rules! smtp_read {
($context:expr, $expected_code:expr) => {
buf_stream.read_line(&mut response).await?;
response.clear();
let mut next_line = String::new();
buf_stream.read_line(&mut next_line).await?;
response.push_str(&next_line);
while let Some(c) = next_line.as_bytes().get(3)
&& *c == b'-'
{
next_line.clear();
buf_stream.read_line(&mut next_line).await?;
response.push_str(&next_line);
}
log::trace!("SMTP response for {}:\n{}", $context, response);
if !response.starts_with($expected_code) {
return Err(crate::error::Error::MailSend {
context: $context.to_string(),
raw_smtp_answer: response.clone(),
});
}
response.clear();
};
}
@@ -45,8 +206,59 @@ pub async fn send(smtp_addr: SocketAddr, envelope: &Envelope) -> Result<(), crat
// Read initial greeting
smtp_read!("initial greeting", "220");
// Greet (Using HELO as we don't want to deal with extended SMTP anyway.)
smtp_cmd!(b"HELO localhost\r\n", "HELO", "250");
if tls_config.is_some() {
smtp_cmd!(
format!("EHLO {client_hostname}\r\n").as_bytes(),
"EHLO",
"250"
);
} else {
smtp_cmd!(
format!("HELO {client_hostname}\r\n").as_bytes(),
"HELO",
"250"
);
};
// STARTTLS
if let Some(tls_config) = tls_config {
if !response.to_uppercase().contains("STARTTLS") {
// TLS was requested, but server doesn't support STARTTLS.
return Err(crate::error::Error::MailSend {
context: "STARTTLS".to_string(),
raw_smtp_answer: response.clone(),
});
}
log::trace!("Initiating STARTTLS...");
smtp_cmd!(b"STARTTLS\r\n", "STARTTLS", "220");
let stream = buf_stream.into_inner();
let raw_tcp = match stream {
SmtpStream::Plain(s) => s,
SmtpStream::Tls(_) => {
unreachable!("This is the first and only place we upgrade to TLS.")
}
};
let tls_stream = crate::tls::wrap_rustls(
address,
raw_tcp,
tls_config.session_cache,
tls_config.allow_invalid_cert,
)
.await?;
let smtp_stream = SmtpStream::Tls(tls_stream);
buf_stream = BufStream::new(smtp_stream);
smtp_cmd!(
format!("EHLO {client_hostname}\r\n").as_bytes(),
"EHLO after STARTTLS",
"250"
);
}
// MAIL FROM
smtp_cmd!(
filtermail/src/smtp_server.rs
+7 -2
View File
@@ -50,7 +50,7 @@ pub async fn run_smtp_server<H>(
addr: &impl tokio::net::ToSocketAddrs,
handler: Arc<H>,
max_size: usize,
) -> Result<(), Box<dyn std::error::Error>>
) -> Result<(), crate::error::Error>
where
H: SmtpHandler + 'static,
{
@@ -112,7 +112,12 @@ where
if cmd.to_uppercase().starts_with("HELO") {
writer.write_all(b"250-filtermail\r\n250 OK\r\n").await?;
writer.flush().await?;
} else if cmd.to_uppercase().starts_with("EHLO") {
} 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-XFORWARD ADDR\r\n250 OK\r\n")
.await?;
filtermail/src/tls.rs
+46
View File
@@ -0,0 +1,46 @@
//! TLS support.
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio_rustls::rustls::client::ClientSessionMemoryCache;
use tokio_rustls::{TlsStream, rustls};
mod danger;
use danger::NoCertificateVerification;
pub async fn wrap_rustls<IO>(
hostname: &str,
stream: IO,
resumption_store: Arc<ClientSessionMemoryCache>,
dangerous_no_cert_verification: bool,
) -> Result<TlsStream<IO>, crate::error::Error>
where
IO: AsyncRead + AsyncWrite + Unpin,
{
let mut root_cert_store = rustls::RootCertStore::empty();
root_cert_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
let mut config = rustls::ClientConfig::builder()
.with_root_certificates(root_cert_store)
.with_no_client_auth();
// Enable TLS 1.3 session resumption
// as defined in <https://www.rfc-editor.org/rfc/rfc8446#section-2.2>.
//
// Obsolete TLS 1.2 mechanisms defined in RFC 5246
// and RFC 5077 have worse security
// and are not worth increasing
// attack surface: <https://words.filippo.io/we-need-to-talk-about-session-tickets/>.
config.resumption = rustls::client::Resumption::store(resumption_store)
.tls12_resumption(rustls::client::Tls12Resumption::Disabled);
if dangerous_no_cert_verification {
config
.dangerous()
.set_certificate_verifier(Arc::new(NoCertificateVerification::default()));
}
let tls = tokio_rustls::TlsConnector::from(Arc::new(config));
let name = rustls::pki_types::ServerName::try_from(hostname)?.to_owned();
let tls_stream = tls.connect(name, stream).await?;
Ok(tls_stream.into())
}
filtermail/src/tls/danger.rs
+49
View File
@@ -0,0 +1,49 @@
//! Dangerous TLS implementation of accepting invalid certificates for Rustls.
use rustls::pki_types::{CertificateDer, ServerName, UnixTime};
use tokio_rustls::rustls;
#[derive(Debug, Default)]
pub(super) struct NoCertificateVerification();
impl rustls::client::danger::ServerCertVerifier for NoCertificateVerification {
fn verify_server_cert(
&self,
_end_entity: &CertificateDer<'_>,
_intermediates: &[CertificateDer<'_>],
_server_name: &ServerName<'_>,
_ocsp_response: &[u8],
_now: UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
message: &[u8],
cert: &CertificateDer<'_>,
dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
let provider = rustls::crypto::ring::default_provider();
let supported_schemes = &provider.signature_verification_algorithms;
rustls::crypto::verify_tls12_signature(message, cert, dss, supported_schemes)
}
fn verify_tls13_signature(
&self,
message: &[u8],
cert: &CertificateDer<'_>,
dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
let provider = rustls::crypto::ring::default_provider();
let supported_schemes = &provider.signature_verification_algorithms;
rustls::crypto::verify_tls13_signature(message, cert, dss, supported_schemes)
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
let provider = rustls::crypto::ring::default_provider();
provider
.signature_verification_algorithms
.supported_schemes()
}
}
filtermail/src/transport.rs
+227
View File
@@ -0,0 +1,227 @@
use crate::config::Config;
use crate::smtp_client::TlsConfig;
use crate::smtp_server::{Envelope, SmtpHandler};
use crate::utils::{AddressDomain, build_resolver};
use async_trait::async_trait;
use hickory_resolver::TokioResolver;
use std::collections::BTreeMap;
use std::str::FromStr;
use std::sync::Arc;
use tokio::task::JoinSet;
use tokio_rustls::rustls;
pub struct TransportHandler {
config: Config,
dns_resolver: Arc<TokioResolver>,
tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
}
impl TransportHandler {
pub fn new(config: Config) -> Result<Self, crate::error::Error> {
let dns_resolver = Arc::new(build_resolver()?);
let tls_resumption_store = Arc::new(rustls::client::ClientSessionMemoryCache::new(256));
Ok(Self {
config,
dns_resolver,
tls_resumption_store,
})
}
/// Handles a single email transaction for a single recipient domain.
async fn handle_single_domain(
tls_resumption_store: Arc<rustls::client::ClientSessionMemoryCache>,
dns_resolver: Arc<TokioResolver>,
domain: AddressDomain,
envelope: Envelope,
client_hostname: String,
) -> Result<String, String> {
let mut allow_invalid_cert = false;
let mut skip_tls = false;
let mx_hosts = match domain {
// no-DNS setup; assume the ip from email address is the destination.
AddressDomain::Literal(ip) => {
// We allow self-signed certs on IP-based relays.
allow_invalid_cert = true;
vec![(0, ip)]
}
AddressDomain::Name(mx_domain) => {
if mx_domain.eq_ignore_ascii_case("nauta.cu") {
// Special case; We don't want to defederate nauta.cu,
// which doesn't support STARTTLS at all.
skip_tls = true;
} else if mx_domain.starts_with('_') {
// We use domains starting with `_` for test deployments.
// (You can't request a non-wildcard cert for such domain)
allow_invalid_cert = true;
}
let query = format!("{mx_domain}.");
let mx_records = dns_resolver.mx_lookup(query).await.map_err(|e| {
if e.is_no_records_found() {
format!("512 No MX records for {mx_domain}")
} else if e.is_nx_domain() {
format!("512 Domain {mx_domain} does not exist")
} else {
format!("421 DNS resolution failed for {mx_domain}")
}
})?;
let mut hosts: Vec<(u16, String)> = mx_records
.iter()
.map(|mx| {
let host = mx.exchange().to_string().trim_end_matches('.').to_string();
(mx.preference(), host)
})
.collect();
hosts.sort();
hosts
}
};
let tls_config = match skip_tls {
true => None,
false => Some(TlsConfig {
allow_invalid_cert,
session_cache: tls_resumption_store,
}),
};
// we try sequentially in order of MX preference,
// but the IPv4 and IPv6 connections (after `smtp_client::send` resolves mx hostname)
// happens in parallel.
'try_relay: for (_, mx_host) in mx_hosts {
match crate::smtp_client::send(
&mx_host,
25,
&envelope,
&client_hostname,
tls_config.clone(),
dns_resolver.clone(),
)
.await
{
Ok(_) => {
return Ok("250 Ok".to_string());
}
Err(error) => {
match error {
// We only want to try other MX hosts if we encounter a problem
// related to connection.
// (So we don't spam other servers if the message is actually rejected.)
crate::error::Error::Io(io_err) => {
log::warn!("I/O error relaying to mail server {mx_host}: {io_err}");
continue 'try_relay;
}
crate::error::Error::ConnectionFailed(_) => {
log::warn!("Failed to connect to mail server {mx_host}: {error}");
continue 'try_relay;
}
crate::error::Error::Tls(tls_err) => {
log::warn!("TLS error relaying to mail server {mx_host}: {tls_err}");
continue 'try_relay;
}
_ => {
log::warn!("Message rejected by mail server {mx_host}: {error}");
return Err(error.smtp_response());
}
}
}
}
}
Err("421 Failed to connect to any mail server".to_string())
}
}
#[async_trait]
impl SmtpHandler for TransportHandler {
/// NO-OP
fn handle_mail(&self, _: &str) -> Result<(), String> {
Ok(())
}
/// NO-OP
async fn check_data(&self, _: &mut Envelope) -> Result<(), String> {
Ok(())
}
/// NO-OP
async fn reinject_mail(&self, _: &Envelope) -> Result<(), String> {
Ok(())
}
/// Handles the DATA command and returns LMTP responses as single string.
///
/// Never returns an error, as LMTP response is composite.
async fn handle_data(&self, envelope: &mut Envelope) -> Result<String, String> {
let mut domain_rcpts_map = BTreeMap::new();
for rcpt in &envelope.rcpt_to {
let domain = AddressDomain::from_str(rcpt)
// Currently we cancel all transactions if any recipient address is invalid.
.map_err(|e| e.lmtp_response(envelope.rcpt_to.len()))?;
domain_rcpts_map
.entry(domain)
.or_insert_with(Vec::new)
.push(rcpt.to_string());
}
// one transaction per domain
let mut transactions = JoinSet::new();
let mut task_id_domain_map = BTreeMap::new();
for (rcpt_domain, rcpts) in &domain_rcpts_map {
let domain_envelope = {
let mut envelope = envelope.clone();
envelope.rcpt_to = rcpts.clone();
envelope
};
let task_id = transactions
.spawn(Self::handle_single_domain(
self.tls_resumption_store.clone(),
self.dns_resolver.clone(),
rcpt_domain.clone(),
domain_envelope,
self.config.mail_domain.clone(),
))
.id();
task_id_domain_map.insert(task_id, rcpt_domain);
}
let mut rcpt_response_map = BTreeMap::new();
while let Some(result) = transactions.join_next_with_id().await {
let domain_response = match result {
Ok((id, Ok(resp))) | Ok((id, Err(resp))) => {
task_id_domain_map.remove(&id).map(|domain| (domain, resp))
}
Err(e) => {
log::error!("Failed to join task: {e}");
task_id_domain_map
.remove(&e.id())
.map(|domain| (domain, "451 Local error".to_string()))
}
};
if let Some((domain, smtp_response)) = domain_response
&& let Some(rcpts) = domain_rcpts_map.get(domain)
{
for rcpt in rcpts {
rcpt_response_map.insert(rcpt, smtp_response.clone());
}
}
}
// compose lmtp response...
let ordered_responses: Vec<String> = envelope
.rcpt_to
.iter()
.map(|rcpt| {
rcpt_response_map
.remove(rcpt)
.unwrap_or_else(|| "451 Local error".to_string())
})
.collect();
Ok(ordered_responses.join("\r\n"))
}
}
filtermail/src/utils.rs
+29 -1
View File
@@ -1,3 +1,5 @@
use hickory_resolver::TokioResolver;
use hickory_resolver::name_server::TokioConnectionProvider;
use mailparse::MailAddr;
use std::path::PathBuf;
use std::str::FromStr;
@@ -30,7 +32,7 @@ pub fn extract_address(input: &str) -> Option<String> {
/// Domain part of an email address, either a domain-literal (IP address in square brackets with
/// optional protocol prefix) or a regular domain name.
#[derive(Debug, PartialEq, Eq)]
#[derive(Debug, PartialEq, Eq, Hash, Clone, PartialOrd, Ord)]
pub enum AddressDomain {
/// Domain literal, e.g.
/// - `192.0.2.0` in `test@[192.0.2.0]`,
@@ -76,6 +78,15 @@ impl FromStr for AddressDomain {
}
}
impl AsRef<str> for AddressDomain {
fn as_ref(&self) -> &str {
match self {
AddressDomain::Literal(literal) => literal.as_ref(),
AddressDomain::Name(name) => name.as_ref(),
}
}
}
/// Logs email to `/tmp/filtermail-rejected/<reason>/<timestamp>.eml`
/// and returns the file path.
///
@@ -93,6 +104,23 @@ pub async fn log_eml(reason: &str, data: &[u8]) -> Result<PathBuf, crate::error:
Ok(path)
}
/// Creates a DNS resolver with DNSSEC enabled and system configuration (resolv.conf).
pub fn build_resolver() -> Result<TokioResolver, crate::error::Error> {
let dns_resolver = {
let mut builder = TokioResolver::builder(TokioConnectionProvider::default())?;
// https://github.com/hickory-dns/hickory-dns/issues/3519
builder.options_mut().validate = true;
builder.build()
};
assert!(
dns_resolver.options().validate,
"incorrect resolver config: DNSSEC disabled; exiting"
);
Ok(dns_resolver)
}
#[cfg(test)]
mod tests {
use super::*;