diff --git a/config.example.toml b/config.example.toml index 350ebbc..a644e1f 100644 --- a/config.example.toml +++ b/config.example.toml @@ -70,6 +70,17 @@ username = "user2" password = "password2" # Optional DKIM configuration +# Sign every outgoing message using this domain and key. +# For strict From-domain selection, replace this table with one or more +# [[server.dkim]] entries (see docs/src/pages/dkim.md). Example: +# [[server.dkim]] +# domain = "example.com" +# selector = "default" +# private_key = "/etc/hedwig/dkim/example.com.key" +# [[server.dkim]] +# domain = "another.com" +# selector = "default" +# private_key = "/etc/hedwig/dkim/another.com.key" [server.dkim] domain = "example.com" selector = "default" diff --git a/dev/verify-multi-domain.py b/dev/verify-multi-domain.py new file mode 100644 index 0000000..b14d5dd --- /dev/null +++ b/dev/verify-multi-domain.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +"""Verify multi-domain DKIM against the Docker DNS/SMTP harness, for both queues. + +Start dev DNS and leave 172.30.0.4 free (see verify-config-reload.py). +Run with an image-compatible binary: + uv run --with dkimpy --with pynacl dev/verify-multi-domain.py /path/to/hedwig +Keys and queues are temporary; DKIM verification uses local public keys, no DNS. +""" +import argparse +import base64 +import os +import pathlib +import runpy +import smtplib +import subprocess +import tempfile +import time +import uuid + +import dkim + +helpers = runpy.run_path(str(pathlib.Path(__file__).with_name("verify-config-reload.py"))) +run, wait_for = helpers["run"], helpers["wait_for"] + + +def verify(binary, backend): + suffix = uuid.uuid4().hex[:8] + server, sink = f"hedwig-domains-{suffix}", f"hedwig-domains-sink-{suffix}" + with tempfile.TemporaryDirectory(prefix="hedwig-domains-") as temporary: + scratch = pathlib.Path(temporary) + run("openssl", "genrsa", "-out", str(scratch / "rsa.key"), "2048") + run("openssl", "genpkey", "-algorithm", "ED25519", "-out", str(scratch / "ed.key")) + rsa_public = subprocess.check_output(["openssl", "pkey", "-in", str(scratch / "rsa.key"), "-pubout", "-outform", "DER"]) + ed_public = subprocess.check_output(["openssl", "pkey", "-in", str(scratch / "ed.key"), "-pubout", "-outform", "DER"])[-32:] + keys = { + b"rsa._domainkey.first.test": b"v=DKIM1; k=rsa; p=" + base64.b64encode(rsa_public), + b"ed._domainkey.second.test": b"v=DKIM1; k=ed25519; p=" + base64.b64encode(ed_public), + b"rotated._domainkey.first.test": b"v=DKIM1; k=ed25519; p=" + base64.b64encode(ed_public), + } + def entry(domain, selector, key, kind="rsa", table="[[server.dkim]]"): + return f'{table}\ndomain = "{domain}"\nselector = "{selector}"\nprivate_key = "/scratch/{key}"\nkey_type = "{kind}"\n' + first = entry("first.test", "rsa", "rsa.key") + second = entry("second.test", "ed", "ed.key", "ed25519") + def config(entries): + return f'''[log] +level = "info" +format = "json" +[server] +workers = 1 +max_retries = 1 +outbound_local = true +[[server.listeners]] +addr = "0.0.0.0:2525" +{entries} +[storage] +storage_type = "{backend}" +base_path = "/scratch/spool" +''' + config_path = scratch / "config.toml" + config_path.write_text(config(first + second)) + sink_code = helpers["SINK"].replace(' p = pathlib.Path("/scratch/messages")', ' if pathlib.Path("/scratch/hold").exists():\n pathlib.Path("/scratch/held").touch()\n while pathlib.Path("/scratch/hold").exists():\n await asyncio.sleep(0.05)\n p = pathlib.Path("/scratch/messages")') + (scratch / "sink.py").write_text(sink_code) + try: + run("docker", "run", "-d", "--name", sink, "--network", "smtp_test", "--ip", "172.30.0.4", + "-v", f"{scratch}:/scratch", "python:3.12-slim", "python", "/scratch/sink.py") + wait_for(lambda: "sink ready" in run("docker", "logs", sink), "SMTP sink") + run("docker", "run", "-d", "--name", server, "--network", "smtp_test", "--dns", "172.30.0.2", + "-p", "127.0.0.1::2525", "-v", f"{scratch}:/scratch", "-v", f"{binary}:/hedwig:ro", + "dev-smtp:latest", "/hedwig", "--config", "/scratch/config.toml") + port = int(run("docker", "port", server, "2525/tcp").strip().rsplit(":", 1)[1]) + def connect(): + client = smtplib.SMTP("127.0.0.1", port, timeout=5) + client.ehlo() + return client + client = wait_for(connect, "Hedwig startup") + def send(header, domain, selector): + count = len(list((scratch / "messages").glob("*"))) + # The envelope intentionally differs from the visible From domain. + client.sendmail("bounce@envelope.test", ["recipient@throttle.test"], + f"{header}\r\nTo: recipient@throttle.test\r\nSubject: domains\r\n\r\nhello\r\n") + raw = wait_for(lambda: (scratch / "messages" / str(count)).read_bytes(), "signed delivery") + assert f"d={domain}".encode() in raw and f"s={selector}".encode() in raw, raw + assert dkim.verify(raw, dnsfunc=lambda name, **kw: keys.get(name.rstrip(b"."))), raw + send("From: Alice ", "first.test", "rsa") + send("From: Bob\r\n ", "second.test", "ed") + for header in ["From: a@unknown.test", "From: a@sub.first.test", "Subject: missing From", + "From: a@first.test\r\nFrom: b@second.test", "From: a@first.test, b@first.test", + "From: invalid", "From: ", "From: group: a@first.test;"]: + try: + client.sendmail("bounce@envelope.test", ["recipient@throttle.test"], f"{header}\r\n\r\nhello\r\n") + except smtplib.SMTPDataError as error: + assert error.smtp_code == 550 and error.smtp_error.startswith(b"5.7.1"), error + else: + raise AssertionError(f"accepted {header}") + # A rejected DATA transaction leaves the session usable. + send("From: a@first.test", "first.test", "rsa") + def reload(entries, failure=False): + before = len(run("docker", "logs", server)) + config_path.write_text(config(entries)) + run("docker", "kill", "--signal", "HUP", server) + expected = "configuration reload failed" if failure else "configuration reload completed" + wait_for(lambda: expected in run("docker", "logs", server)[before:].lower(), "reload") + for invalid in [first + entry("FIRST.TEST", "rsa", "rsa.key"), + first + entry("third.test", "rsa", "missing.key")]: + reload(invalid, failure=True) + send("From: a@second.test", "second.test", "ed") + reload(entry("first.test", "rotated", "ed.key", "ed25519") + second) + send("From: a@first.test", "first.test", "rotated") + reload(first) + try: + client.sendmail("bounce@envelope.test", ["recipient@throttle.test"], "From: a@second.test\r\n\r\nhello\r\n") + except smtplib.SMTPDataError as error: + assert error.smtp_code == 550 + else: + raise AssertionError("removed domain accepted") + # Legacy table signs unrelated From domains exactly as before. + reload(entry("first.test", "rsa", "rsa.key", table="[server.dkim]")) + send("From: a@unknown.test", "first.test", "rsa") + # Hold the only worker on an earlier message while a second-domain + # message is queued, then remove that domain before it is attempted. + reload(first + second) + (scratch / "hold").touch() + client.sendmail("bounce@envelope.test", ["recipient@throttle.test"], "From: a@first.test\r\n\r\nblocker\r\n") + wait_for(lambda: (scratch / "held").exists(), "worker blocked at sink") + client.sendmail("bounce@envelope.test", ["recipient@throttle.test"], "From: a@second.test\r\n\r\npending\r\n") + reload(first) + before = len(run("docker", "logs", server)) + (scratch / "hold").unlink() + wait_for(lambda: "Sending domain second.test is not configured" in run("docker", "logs", server)[before:], "queued domain removal defers and exhausts retries") + logs = run("docker", "logs", server)[before:] + assert '"status":"deferred"' in logs, logs + assert '"status":"bounced"' not in logs, logs + # With one allowed attempt the deferred message must eventually + # move to the bounce archive, on both queue implementations. + deadline = time.monotonic() + 110 + while time.monotonic() < deadline: + if any(b"pending" in path.read_bytes() for path in (scratch / "spool" / "bounced").rglob("*") if path.is_file()): + break + time.sleep(0.5) + else: + raise AssertionError("removed-domain message did not bounce after retry exhaustion") + client.quit() + print(f"PASS {backend}: RSA/Ed25519 signatures verified, From routing, rejection, rotation, atomic reload, legacy compatibility, queued domain removal defers and exhausts retries") + except Exception: + subprocess.run(["docker", "logs", server], check=False) + raise + finally: + for name in (server, sink): + subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + run("docker", "run", "--rm", "-v", f"{scratch}:/scratch", "dev-smtp:latest", + "chown", "-R", f"{os.getuid()}:{os.getgid()}", "/scratch") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("binary", type=pathlib.Path) + args = parser.parse_args() + binary = args.binary.resolve() + assert binary.is_file(), binary + for backend in ("log", "fs"): + verify(binary, backend) diff --git a/docs/src/pages/dkim.md b/docs/src/pages/dkim.md index f8c9b29..3177881 100644 --- a/docs/src/pages/dkim.md +++ b/docs/src/pages/dkim.md @@ -8,7 +8,7 @@ description: Configure DKIM signing and generate keys. DKIM (DomainKeys Identified Mail) allows receiving mail servers to verify that emails were sent by an authorized sender. -## Configuration +## Sign all messages with one domain ```toml [server.dkim] @@ -17,8 +17,82 @@ selector = "default" private_key = "/path/to/dkim/private.key" ``` +With `[server.dkim]`, Hedwig signs every outgoing message using the configured +domain and key, regardless of the sender’s `From:` address. In this example, +all messages are signed as `yourdomain.com`. + +If you omit DKIM configuration, Hedwig does not add a signature. Any signatures +already present on the message are kept. + +## Multiple sending domains + +Replace the single table with an array of tables to select a signer using the +message's **`From:` header**, independently of the SMTP `MAIL FROM` address: + +```toml +[[server.dkim]] +domain = "example.com" +selector = "default" +private_key = "/etc/hedwig/dkim/example.com.key" + +[[server.dkim]] +domain = "another.com" +selector = "default" +private_key = "/etc/hedwig/dkim/another.com.key" +key_type = "ed25519" +``` + +`alice@example.com` uses the first key; `bob@another.com` uses the second. +Publish the corresponding DKIM TXT record for each domain. Listeners, the SMTP +hostname, queue, workers, and outbound connections remain shared. + +- Matching is case-insensitive and exact. Configure subdomains separately; + wildcard and parent-domain fallback are not supported. +- Use ASCII DNS names (punycode/A-label form for internationalized domains). +- Exactly one `From:` header with one mailbox is required. Missing, malformed, + multiple-mailbox, and unconfigured senders receive `550 5.7.1` after DATA, + before the message is queued. The connection remains usable. +- Even a one-entry array enables these checks. Do not combine `[server.dkim]` + and `[[server.dkim]]` in the same configuration. +- Empty lists, duplicate domains, invalid domains, and unreadable or invalid + keys prevent startup or reject a reload. Failed reloads retain all previous + signers. These identities do not restrict individual SMTP users to domains. + +SIGHUP reloads all signers atomically. Queued messages use the current key when +attempted, so rotation applies to pending mail too. When a domain is removed +from the array, its queued messages are deferred under the normal bounded retry +policy, eventually bouncing if the domain is not restored. Disabling DKIM +entirely stops Hedwig from adding signatures, including to pending mail. +Already-running signing operations may finish using the previous snapshot. + +When switching from a single signer or unsigned relaying to domain-based +signing, queued messages also undergo the stricter `From:` validation. Messages +with malformed or ambiguous `From:` headers bounce on their next attempt; +messages with a valid but unconfigured domain retry. Drain the queue before +switching if you need pending messages delivered under the previous rules. + +For a multi-domain configuration, select the entry when generating its key: + +```bash +hedwig --config config.toml dkim-generate --domain example.com +hedwig --config config.toml dkim-generate --domain another.com +``` + +The entry supplies its selector, private-key path, and key type; flags can +override them. `--domain` is required in this mode, even with one entry. An +unconfigured domain requires explicit `--selector` and `--private-key` flags. +Key generation writes the configured private-key file; publish its DNS record +before enabling a new identity or rotating the active signer. + ## Generating keys +When overriding a selector or key path, the configured key type remains the +default. Changing it with `--key-type` requires an explicit `--private-key` path. +Update the configured key type, key path, and DNS record before enabling the new +key. With a single signing domain, +overriding `--domain` with a different domain also requires `--private-key` so +the command does not overwrite the configured domain’s key by default. + ```bash ./target/release/hedwig dkim-generate ``` @@ -37,7 +111,7 @@ Available flags: - `--domain`: Domain for DKIM signature - `--selector`: DKIM selector - `--private-key`: Path to save the private key -- `--key-type`: Key type (rsa or ed25519, default: rsa) +- `--key-type`: Key type (rsa or ed25519, default: configured key type, or rsa if none) Add the DNS TXT record output by the command: diff --git a/smtp-server/src/callbacks.rs b/smtp-server/src/callbacks.rs index c5fbd86..b9b3dc9 100644 --- a/smtp-server/src/callbacks.rs +++ b/smtp-server/src/callbacks.rs @@ -609,6 +609,12 @@ impl SmtpCallbacks for Callbacks { // Handles the DATA command. async fn on_data(&self, email: Email) -> Result<(), SmtpError> { + self.runtime + .load() + .signer_for(&email.body) + .map_err(|error| SmtpError::DataRejected { + message: format!("5.7.1 {error}"), + })?; self.process_email(email).await?; Ok(()) } @@ -1656,6 +1662,60 @@ mod tests { } } + #[tokio::test] + async fn rejected_from_never_reaches_storage_or_worker_channel() { + use crate::config::{CfgDKIM, CfgDkim, DkimKeyType}; + use futures::StreamExt; + let mut cfg = create_test_config(); + cfg.server.dkim = Some(CfgDkim::Domains(vec![CfgDKIM { + domain: "example.com".into(), + selector: "default".into(), + private_key: format!( + "{}/../dev/certs/dkim-private.pem", + env!("CARGO_MANIFEST_DIR") + ), + key_type: DkimKeyType::Rsa, + }])); + let directory = tempfile::tempdir().unwrap(); + let storage = Arc::new( + crate::storage::fs_storage::FileSystemStorage::new(directory.path().to_str().unwrap()) + .await + .unwrap(), + ); + let (sender_channel, receiver) = async_channel::unbounded(); + let callbacks = Callbacks { + runtime: Arc::new(RuntimeConfig::new(cfg).unwrap()), + storage: storage.clone(), + sender_channel, + log_queue: None, + }; + for headers in [ + "From: a@unknown.com", + "From: a@example.com\r\nFrom: b@example.com", + ] { + let result = callbacks + .on_data(Email { + from: "bounce@example.com".into(), + to: vec!["r@example.com".into()], + body: format!("{headers}\r\n\r\nbody").into(), + }) + .await; + assert!(matches!(result, Err(SmtpError::DataRejected { .. }))); + assert!(receiver.is_empty()); + assert!(storage.list(Status::Queued).next().await.is_none()); + } + callbacks + .on_data(Email { + from: "bounce@other.com".into(), + to: vec!["r@example.com".into()], + body: "From: a@example.com\r\n\r\nbody".into(), + }) + .await + .unwrap(); + assert_eq!(receiver.len(), 1); + assert!(storage.list(Status::Queued).next().await.unwrap().is_ok()); + } + #[tokio::test] async fn test_on_data() { // Test on_data method (lines 339-341) diff --git a/smtp-server/src/config.rs b/smtp-server/src/config.rs index 1767ff8..b4a0c35 100644 --- a/smtp-server/src/config.rs +++ b/smtp-server/src/config.rs @@ -39,7 +39,7 @@ pub struct CfgServer { pub workers: Option, pub max_retries: Option, pub auth: Option>, - pub dkim: Option, + pub dkim: Option, pub disable_outbound: Option, pub outbound_local: Option, pub helo_hostname: Option, @@ -81,7 +81,7 @@ pub struct CfgStorage { pub cleanup: Option, } -#[derive(Debug, Deserialize, Serialize, Clone, Default)] +#[derive(Debug, Deserialize, Serialize, Clone, Default, PartialEq, Eq)] #[serde(rename_all = "lowercase")] pub enum DkimKeyType { #[default] @@ -104,6 +104,48 @@ impl Default for CfgLog { } } +/// A table preserves global signing; an array opts into strict From-domain routing. +#[derive(Debug, Serialize, Clone)] +#[serde(untagged)] +pub enum CfgDkim { + Legacy(CfgDKIM), + Domains(Vec), +} + +// Dispatch by shape so errors inside an entry retain their field diagnostics. +impl<'de> Deserialize<'de> for CfgDkim { + fn deserialize>(deserializer: D) -> Result { + struct Visitor; + impl<'de> serde::de::Visitor<'de> for Visitor { + type Value = CfgDkim; + fn expecting(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("a DKIM table or an array of DKIM tables") + } + fn visit_map>( + self, + map: M, + ) -> Result { + CfgDKIM::deserialize(serde::de::value::MapAccessDeserializer::new(map)) + .map(CfgDkim::Legacy) + } + fn visit_seq>( + self, + seq: S, + ) -> Result { + Vec::::deserialize(serde::de::value::SeqAccessDeserializer::new(seq)) + .map(CfgDkim::Domains) + } + } + deserializer.deserialize_any(Visitor) + } +} + +impl From for CfgDkim { + fn from(value: CfgDKIM) -> Self { + Self::Legacy(value) + } +} + #[derive(Debug, Deserialize, Serialize, Clone)] pub struct CfgDKIM { pub domain: String, @@ -354,6 +396,50 @@ mod tests { settings.try_deserialize().expect("deserialize config") } + #[test] + fn dkim_file_formats_preserve_field_errors() { + for extension in ["toml", "huml"] { + for array in [false, true] { + let source = if extension == "toml" { + format!("{MINIMAL_CFG}\n{}\ndomain = \"example.com\"\nselector = \"default\"\nprivate_key = \"key.pem\"\nkey_type = \"ed25519\"\n", + if array { "[[server.dkim]]" } else { "[server.dkim]" }) + } else { + let entry = if array { + " - ::\n domain: \"example.com\"\n selector: \"default\"\n private_key: \"key.pem\"\n key_type: \"ed25519\"\n" + } else { + " domain: \"example.com\"\n selector: \"default\"\n private_key: \"key.pem\"\n key_type: \"ed25519\"\n" + }; + format!("server::\n listeners::\n - ::\n addr: \"127.0.0.1:2525\"\n dkim::\n{entry}storage::\n base_path: \"/tmp/hedwig-test\"\n") + }; + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(format!("config.{extension}")); + std::fs::write(&path, &source).unwrap(); + let cfg = Cfg::load(path.to_str().unwrap()).unwrap(); + assert_eq!(matches!(cfg.server.dkim, Some(CfgDkim::Domains(_))), array); + let missing = source + .lines() + .filter(|line| !line.contains("selector")) + .collect::>() + .join("\n") + + "\n"; + for (invalid, expected) in [ + (missing, "missing field `selector`"), + ( + source.replace("ed25519", "not-a-key-type"), + "not-a-key-type", + ), + ] { + std::fs::write(&path, invalid).unwrap(); + let error = format!("{:#}", Cfg::load(path.to_str().unwrap()).unwrap_err()); + assert!( + error.contains(expected), + "{extension} array={array}: {error}" + ); + } + } + } + } + #[test] fn auth_password_deserializes_but_is_never_serialized() { let auth: CfgAuth = diff --git a/smtp-server/src/dkim/mod.rs b/smtp-server/src/dkim/mod.rs index 1f45918..af90818 100644 --- a/smtp-server/src/dkim/mod.rs +++ b/smtp-server/src/dkim/mod.rs @@ -1,4 +1,4 @@ -use crate::config::{Cfg, CfgDKIM, DkimKeyType}; +use crate::config::{Cfg, CfgDKIM, CfgDkim, DkimKeyType}; use base64::Engine; use clap::Parser; use miette::{bail, Context, IntoDiagnostic, Result}; @@ -15,54 +15,156 @@ pub const DEFAULT_DKIM_KEY_BITS: usize = 2048; pub async fn generate_dkim_keys(config_path: &str, args: DkimGenerateArgs) -> Result<()> { let cfg = Cfg::load(config_path).wrap_err("error loading configuration")?; - let dkim_config = - if args.domain.is_some() || args.selector.is_some() || args.private_key.is_some() { - let domain = match args.domain { - Some(d) => d, - None => match &cfg.server.dkim { - Some(config) => config.domain.clone(), - None => bail!("Domain is required when not in config file"), - }, - }; + let dkim_config = key_generation_config(cfg.server.dkim.as_ref(), args)?; - let selector = match args.selector { - Some(s) => s, - None => match &cfg.server.dkim { - Some(config) => config.selector.clone(), - None => bail!("Selector is required when not in config file"), - }, - }; + match dkim_config.key_type { + DkimKeyType::Rsa => generate_rsa_keys(&dkim_config).await, + DkimKeyType::Ed25519 => generate_ed25519_keys(&dkim_config).await, + } +} - let private_key = match args.private_key { - Some(p) => p, - None => match &cfg.server.dkim { - Some(config) => config.private_key.clone(), - None => bail!("Private key path is required when not in config file"), - }, - }; +/// Select config defaults without accidentally generating a key for another domain. +fn key_generation_config(config: Option<&CfgDkim>, args: DkimGenerateArgs) -> Result { + let defaults = match config { + Some(CfgDkim::Legacy(entry)) => { + if args + .domain + .as_ref() + .is_some_and(|domain| !entry.domain.eq_ignore_ascii_case(domain)) + && args.private_key.is_none() + { + bail!("--private-key is required when overriding the configured DKIM domain"); + } + Some(entry) + } + Some(CfgDkim::Domains(entries)) => { + let domain = args.domain.as_deref().ok_or_else(|| { + miette::miette!("--domain is required with multi-domain DKIM configuration") + })?; + entries + .iter() + .find(|entry| entry.domain.eq_ignore_ascii_case(domain)) + } + None => None, + }; + if args.domain.is_none() + && args.selector.is_none() + && args.private_key.is_none() + && args.key_type.is_none() + { + return defaults.cloned().ok_or_else(|| { + miette::miette!("DKIM configuration is missing in config file and no flags provided") + }); + } + let domain = args + .domain + .or_else(|| defaults.map(|v| v.domain.clone())) + .ok_or_else(|| miette::miette!("Domain is required when not in config file"))?; + let selector = args + .selector + .or_else(|| defaults.map(|v| v.selector.clone())) + .ok_or_else(|| miette::miette!("Selector is required when not in config file"))?; + let key_type = match args.key_type.as_deref() { + Some("rsa") => DkimKeyType::Rsa, + Some("ed25519") => DkimKeyType::Ed25519, + Some(_) => bail!("Invalid key type. Use 'rsa' or 'ed25519'"), + None => defaults.map(|v| v.key_type.clone()).unwrap_or_default(), + }; + if defaults.is_some_and(|entry| entry.key_type != key_type) && args.private_key.is_none() { + bail!("--private-key is required when changing the configured DKIM key type; update key_type and the DNS record before enabling the new key"); + } + let private_key = args + .private_key + .or_else(|| defaults.map(|v| v.private_key.clone())) + .ok_or_else(|| miette::miette!("Private key path is required when not in config file"))?; + Ok(CfgDKIM { + domain, + selector, + private_key, + key_type, + }) +} - let key_type = match args.key_type.as_str() { - "rsa" => DkimKeyType::Rsa, - "ed25519" => DkimKeyType::Ed25519, - _ => bail!("Invalid key type. Use 'rsa' or 'ed25519'"), - }; +/// Domains are DNS names in ASCII (use A-label/punycode for IDNs). +/// Exact matching deliberately excludes wildcard and parent-domain fallback. +pub(crate) fn normalize_domain(domain: &str) -> Option { + (domain.len() <= 253 + && domain.split('.').all(|label| { + !label.is_empty() + && label.len() <= 63 + && label.as_bytes()[0].is_ascii_alphanumeric() + && label.as_bytes()[label.len() - 1].is_ascii_alphanumeric() + && label + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-') + })) + .then(|| domain.to_ascii_lowercase()) +} - CfgDKIM { - domain, - selector, - private_key, - key_type, - } - } else { - match &cfg.server.dkim { - Some(config) => config.clone(), - None => bail!("DKIM configuration is missing in config file and no flags provided"), - } - }; +pub(crate) fn from_domain(body: &[u8]) -> Result { + let message = mail_parser::MessageParser::default() + .parse_headers(body) + .ok_or_else(|| miette::miette!("Invalid message headers"))?; + if message + .headers() + .iter() + .filter(|h| h.name == mail_parser::HeaderName::From) + .count() + != 1 + { + bail!("Exactly one From header is required"); + } + let Some(mail_parser::Address::List(addresses)) = message.from() else { + bail!("From must contain one mailbox"); + }; + if addresses.len() != 1 { + bail!("From must contain one mailbox"); + } + let address = addresses[0] + .address() + .ok_or_else(|| miette::miette!("Invalid From mailbox"))?; + let parsed = email_address_parser::EmailAddress::parse(address, None) + .ok_or_else(|| miette::miette!("Invalid From mailbox"))?; + let domain = + normalize_domain(&parsed.domain()).ok_or_else(|| miette::miette!("Invalid From domain"))?; + // mail-parser deliberately recovers malformed addresses (for example an + // unclosed angle bracket). Require the raw field to parse as one mailbox + // too, so recovery cannot turn malformed input into an authorized sender. + let raw = message + .headers_raw() + .find(|(name, _)| name.eq_ignore_ascii_case("From")) + .map(|(_, value)| value) + .ok_or_else(|| miette::miette!("Invalid From header"))?; + let raw_addresses = mailparse::addrparse(raw) + .into_diagnostic() + .wrap_err("Invalid From header")?; + let [mailparse::MailAddr::Single(mailbox)] = raw_addresses.as_slice() else { + bail!("From must contain one mailbox"); + }; + let raw_address = email_address_parser::EmailAddress::parse(&mailbox.addr, None) + .ok_or_else(|| miette::miette!("Invalid From mailbox"))?; + if normalize_domain(&raw_address.domain()).as_ref() != Some(&domain) { + bail!("Ambiguous From domain"); + } + Ok(domain) +} - match dkim_config.key_type { - DkimKeyType::Rsa => generate_rsa_keys(&dkim_config).await, - DkimKeyType::Ed25519 => generate_ed25519_keys(&dkim_config).await, +#[derive(Debug, thiserror::Error, miette::Diagnostic)] +#[error("Sending domain {0} is not configured")] +pub(crate) struct UnconfiguredDomain(pub String); + +impl crate::reload::RuntimeSnapshot { + pub(crate) fn signer_for(&self, body: &[u8]) -> Result> { + match &self.dkim_domains { + Some(domains) => { + let domain = from_domain(body)?; + domains + .get(&domain) + .map(Some) + .ok_or_else(|| UnconfiguredDomain(domain).into()) + } + None => Ok(self.dkim_signer.as_ref()), + } } } @@ -154,6 +256,162 @@ pub struct DkimGenerateArgs { pub private_key: Option, /// Key type (rsa or ed25519) - #[arg(long, default_value = "rsa")] - pub key_type: String, + #[arg(long)] + pub key_type: Option, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn from_domain_requires_one_unambiguous_mailbox() { + for header in [ + "From: Alice ", + "From: Alice\r\n ", + "From: \"a@b\" ", + ] { + assert_eq!( + from_domain(format!("{header}\r\nSubject: test\r\n\r\nbody").as_bytes()).unwrap(), + "example.com" + ); + } + for header in [ + "Subject: no from", + "From: a@example.com\r\nFrom: b@example.com", + "From: a@example.com, b@example.com", + "From: friends: a@example.com;", + "From: invalid", + "From: ", + "From: a@bad_domain.com", + "From: a@[127.0.0.1]", + "From:", + ] { + assert!( + from_domain(format!("{header}\r\n\r\nbody").as_bytes()).is_err(), + "accepted {header}" + ); + } + } + + #[test] + fn key_generation_preserves_algorithm_and_requires_new_domain_path() { + let config = CfgDkim::Legacy(CfgDKIM { + domain: "example.com".into(), + selector: "old".into(), + private_key: "live.pem".into(), + key_type: DkimKeyType::Ed25519, + }); + let args = || DkimGenerateArgs { + domain: None, + selector: Some("new".into()), + private_key: None, + key_type: None, + }; + assert!(matches!( + key_generation_config(Some(&config), args()) + .unwrap() + .key_type, + DkimKeyType::Ed25519 + )); + let mut explicit = args(); + explicit.selector = None; + explicit.key_type = Some("rsa".into()); + explicit.private_key = Some("rsa.pem".into()); + assert!(matches!( + key_generation_config(Some(&config), explicit) + .unwrap() + .key_type, + DkimKeyType::Rsa + )); + let mut other = args(); + other.domain = Some("other.com".into()); + assert!(key_generation_config(Some(&config), other) + .unwrap_err() + .to_string() + .contains("--private-key")); + let mut other = args(); + other.domain = Some("other.com".into()); + other.private_key = Some("other.pem".into()); + let selected = key_generation_config(Some(&config), other).unwrap(); + assert_eq!(selected.private_key, "other.pem"); + let mut same = args(); + same.domain = Some("EXAMPLE.COM".into()); + assert_eq!( + key_generation_config(Some(&config), same) + .unwrap() + .private_key, + "live.pem" + ); + } + + #[test] + fn changing_algorithm_requires_explicit_path_in_both_modes() { + for current in [DkimKeyType::Rsa, DkimKeyType::Ed25519] { + let entry = CfgDKIM { + domain: "example.com".into(), + selector: "default".into(), + private_key: "live.pem".into(), + key_type: current.clone(), + }; + for config in [ + CfgDkim::Legacy(entry.clone()), + CfgDkim::Domains(vec![entry.clone()]), + ] { + for requested in ["rsa", "ed25519"] { + let args = |path| DkimGenerateArgs { + domain: matches!(config, CfgDkim::Domains(_)).then(|| "example.com".into()), + selector: None, + private_key: path, + key_type: Some(requested.into()), + }; + let changed = (current == DkimKeyType::Rsa) != (requested == "rsa"); + let result = key_generation_config(Some(&config), args(None)); + if changed { + assert!(result.unwrap_err().to_string().contains("--private-key")); + } else { + assert_eq!(result.unwrap().private_key, "live.pem"); + } + let selected = + key_generation_config(Some(&config), args(Some("new.pem".into()))).unwrap(); + assert_eq!(selected.private_key, "new.pem"); + assert_eq!(selected.key_type == DkimKeyType::Rsa, requested == "rsa"); + } + } + } + } + + #[test] + fn config_formats_and_key_generation_defaults() { + let entry = CfgDKIM { + domain: "example.com".into(), + selector: "one".into(), + private_key: "one.pem".into(), + key_type: DkimKeyType::Ed25519, + }; + let args = |domain| DkimGenerateArgs { + domain, + selector: None, + private_key: None, + key_type: None, + }; + let legacy: CfgDkim = + serde_json::from_value(serde_json::to_value(&entry).unwrap()).unwrap(); + assert!(matches!(legacy, CfgDkim::Legacy(_))); + assert!(matches!( + key_generation_config(Some(&legacy), args(None)) + .unwrap() + .key_type, + DkimKeyType::Ed25519 + )); + let multi: CfgDkim = serde_json::from_value(serde_json::json!([entry])).unwrap(); + assert!(matches!(multi, CfgDkim::Domains(_))); + assert!(key_generation_config(Some(&multi), args(None)).is_err()); + let selected = + key_generation_config(Some(&multi), args(Some("EXAMPLE.COM".into()))).unwrap(); + assert_eq!(selected.private_key, "one.pem"); + assert!(matches!(selected.key_type, DkimKeyType::Ed25519)); + assert!(key_generation_config(Some(&multi), args(Some("unknown.com".into()))).is_err()); + } } diff --git a/smtp-server/src/main.rs b/smtp-server/src/main.rs index 93075dc..9140012 100644 --- a/smtp-server/src/main.rs +++ b/smtp-server/src/main.rs @@ -687,6 +687,7 @@ async fn wait_for_shutdown_signal( ); } else { error!( + dkim_index = failure.dkim_index, component = failure.component, stage = failure.stage, "configuration reload failed; previous configuration retained" diff --git a/smtp-server/src/reload.rs b/smtp-server/src/reload.rs index 7671a58..2cc566b 100644 --- a/smtp-server/src/reload.rs +++ b/smtp-server/src/reload.rs @@ -15,7 +15,7 @@ use tracing::Level; use crate::{ callbacks::DomainFilters, - config::{Cfg, FilterType}, + config::{Cfg, CfgDkim, FilterType}, worker::{DkimSignerType, Worker}, }; @@ -25,6 +25,7 @@ pub(crate) struct RuntimeSnapshot { pub(crate) to_domain_filters: DomainFilters, pub(crate) auth: HashMap, pub(crate) dkim_signer: Option, + pub(crate) dkim_domains: Option>, tls_acceptors: Vec>, log_level: Level, } @@ -46,6 +47,7 @@ pub struct PreparationError { pub component: &'static str, pub stage: &'static str, pub listener_index: Option, + pub dkim_index: Option, } impl PreparationError { @@ -54,20 +56,34 @@ impl PreparationError { component, stage, listener_index: None, + dkim_index: None, } } + fn at_dkim_entry(mut self, index: usize) -> Self { + self.dkim_index = Some(index); + self + } + fn listener(component: &'static str, stage: &'static str, index: usize) -> Self { Self { component, stage, listener_index: Some(index), + dkim_index: None, } } } impl std::fmt::Display for PreparationError { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if let Some(index) = self.dkim_index { + return write!( + formatter, + "{} {} for DKIM entry {index}", + self.component, self.stage + ); + } match self.listener_index { Some(index) => write!( formatter, @@ -137,16 +153,41 @@ impl RuntimeSnapshot { .flatten() .map(|a| (a.username.clone(), a.password.clone())) .collect(); - let dkim_signer = match &cfg.server.dkim { - Some(dkim) => { - let key = std::fs::read_to_string(&dkim.private_key) - .map_err(|_| PreparationError::new("server.dkim.private_key", "read"))?; - Some( - Worker::create_dkim_signer(dkim, &key) - .map_err(|_| PreparationError::new("server.dkim.private_key", "parse"))?, - ) + let load_signer = |dkim: &crate::config::CfgDKIM| { + let key = std::fs::read_to_string(&dkim.private_key) + .map_err(|_| PreparationError::new("server.dkim.private_key", "read"))?; + Worker::create_dkim_signer(dkim, &key) + .map_err(|_| PreparationError::new("server.dkim.private_key", "parse")) + }; + let (dkim_signer, dkim_domains) = match &cfg.server.dkim { + Some(CfgDkim::Legacy(dkim)) => (Some(load_signer(dkim)?), None), + Some(CfgDkim::Domains(entries)) => { + let mut domains = HashMap::new(); + if entries.is_empty() { + return Err(PreparationError::new("server.dkim", "empty domain list")); + } + for (index, entry) in entries.iter().enumerate() { + let domain = crate::dkim::normalize_domain(&entry.domain).ok_or_else(|| { + PreparationError::new("server.dkim.domain", "invalid domain") + .at_dkim_entry(index) + })?; + if domains.contains_key(&domain) { + return Err(PreparationError::new( + "server.dkim.domain", + "duplicate domain", + ) + .at_dkim_entry(index)); + } + let mut entry = entry.clone(); + entry.domain = domain.clone(); + domains.insert( + domain, + load_signer(&entry).map_err(|error| error.at_dkim_entry(index))?, + ); + } + (None, Some(domains)) } - None => None, + None => (None, None), }; let tls_acceptors = cfg .server @@ -168,6 +209,7 @@ impl RuntimeSnapshot { to_domain_filters: DomainFilters::build(&cfg, |t| matches!(t, FilterType::ToDomain)), auth, dkim_signer, + dkim_domains, tls_acceptors, log_level, cfg, @@ -227,7 +269,10 @@ fn unsupported_changes(old: &Cfg, new: &Cfg) -> Result> { if path == "log.level" || path == "filters" || path.starts_with("filters[") { return false; } - if path == "server.dkim" || path.starts_with("server.dkim.") { + if path == "server.dkim" + || path.starts_with("server.dkim.") + || path.starts_with("server.dkim[") + { return false; } if auth_reloadable && (path == "server.auth" || path.starts_with("server.auth[")) { @@ -441,12 +486,15 @@ mod tests { let before = runtime.current.load_full(); let mut new = cfg(); new.server.workers = Some(99); - new.server.dkim = Some(CfgDKIM { - domain: "example.test".into(), - selector: "new".into(), - private_key: "/definitely/missing/dkim.pem".into(), - key_type: DkimKeyType::Rsa, - }); + new.server.dkim = Some( + CfgDKIM { + domain: "example.test".into(), + selector: "new".into(), + private_key: "/definitely/missing/dkim.pem".into(), + key_type: DkimKeyType::Rsa, + } + .into(), + ); let error = runtime.reload(new, |_| Ok(())).unwrap_err(); assert!(matches!( error, @@ -478,6 +526,7 @@ mod tests { component: "log.level", stage: "parse", listener_index: None, + dkim_index: None, })) )); assert_snapshot_retained(&runtime, &before); @@ -488,18 +537,22 @@ mod tests { let runtime = RuntimeConfig::new(cfg()).unwrap(); let before = runtime.current.load_full(); let mut new = cfg(); - new.server.dkim = Some(CfgDKIM { - domain: "example.test".into(), - selector: "new".into(), - private_key: "/definitely/missing/dkim.pem".into(), - key_type: DkimKeyType::Rsa, - }); + new.server.dkim = Some( + CfgDKIM { + domain: "example.test".into(), + selector: "new".into(), + private_key: "/definitely/missing/dkim.pem".into(), + key_type: DkimKeyType::Rsa, + } + .into(), + ); assert!(matches!( runtime.reload(new, |_| Ok(())), Err(ReloadError::Preparation(PreparationError { component: "server.dkim.private_key", stage: "read", listener_index: None, + dkim_index: None, })) )); assert_snapshot_retained(&runtime, &before); @@ -512,12 +565,15 @@ mod tests { let runtime = RuntimeConfig::new(cfg()).unwrap(); let before = runtime.current.load_full(); let mut new = cfg(); - new.server.dkim = Some(CfgDKIM { - domain: "example.test".into(), - selector: "new".into(), - private_key: key.path().to_string_lossy().into_owned(), - key_type: DkimKeyType::Rsa, - }); + new.server.dkim = Some( + CfgDKIM { + domain: "example.test".into(), + selector: "new".into(), + private_key: key.path().to_string_lossy().into_owned(), + key_type: DkimKeyType::Rsa, + } + .into(), + ); let error = runtime.reload(new, |_| Ok(())).unwrap_err(); assert!(matches!( error, @@ -525,6 +581,7 @@ mod tests { component: "server.dkim.private_key", stage: "parse", listener_index: None, + dkim_index: None, }) )); assert!(!format!("{error:?}").contains("candidate-secret")); @@ -534,18 +591,27 @@ mod tests { #[test] fn dkim_rotation_disable_and_reenable_publish_atomically() { let mut initial = cfg(); - initial.server.dkim = Some(CfgDKIM { - domain: "example.test".into(), - selector: "one".into(), - private_key: fixture("dkim-private.pem"), - key_type: DkimKeyType::Rsa, - }); + initial.server.dkim = Some( + CfgDKIM { + domain: "example.test".into(), + selector: "one".into(), + private_key: fixture("dkim-private.pem"), + key_type: DkimKeyType::Rsa, + } + .into(), + ); let runtime = RuntimeConfig::new(initial.clone()).unwrap(); let mut rotated = initial.clone(); - rotated.server.dkim.as_mut().unwrap().selector = "two".into(); + let Some(CfgDkim::Legacy(entry)) = &mut rotated.server.dkim else { + panic!("legacy config") + }; + entry.selector = "two".into(); runtime.reload(rotated, |_| Ok(())).unwrap(); assert_eq!( - runtime.load().cfg.server.dkim.as_ref().unwrap().selector, + match runtime.load().cfg.server.dkim.as_ref().unwrap() { + CfgDkim::Legacy(entry) => entry.selector.clone(), + _ => panic!("legacy config"), + }, "two" ); let mut disabled = initial.clone(); @@ -556,6 +622,81 @@ mod tests { assert!(runtime.load().dkim_signer.is_some()); } + #[test] + fn multi_domain_selection_reload_and_validation() { + let entry = |domain: &str| CfgDKIM { + domain: domain.into(), + selector: "one".into(), + private_key: fixture("dkim-private.pem"), + key_type: DkimKeyType::Rsa, + }; + let mut initial = cfg(); + initial.server.dkim = Some(CfgDkim::Domains(vec![ + entry("EXAMPLE.com"), + entry("another.com"), + ])); + let runtime = RuntimeConfig::new(initial.clone()).unwrap(); + let body = |domain| format!("From: Sender \r\n\r\nhello").into_bytes(); + for domain in ["example.com", "EXAMPLE.COM", "another.com"] { + assert!(runtime.load().signer_for(&body(domain)).unwrap().is_some()); + } + for domain in ["unknown.com", "news.example.com"] { + assert!(runtime + .load() + .signer_for(&body(domain)) + .err() + .unwrap() + .downcast_ref::() + .is_some()); + } + let before = runtime.current.load_full(); + for (entries, expected_index) in [ + (vec![], None), + (vec![entry("example.com"), entry("EXAMPLE.COM")], Some(1)), + (vec![entry("example.com"), entry("*.example.com")], Some(1)), + ] { + let mut bad = initial.clone(); + bad.server.dkim = Some(CfgDkim::Domains(entries)); + let Err(ReloadError::Preparation(error)) = runtime.reload(bad, |_| Ok(())) else { + panic!("expected domain validation failure") + }; + assert_eq!(error.dkim_index, expected_index); + assert_snapshot_retained(&runtime, &before); + } + let mut bad = initial.clone(); + let mut broken = entry("third.com"); + broken.private_key = "/missing/key".into(); + bad.server.dkim = Some(CfgDkim::Domains(vec![entry("example.com"), broken])); + let Err(ReloadError::Preparation(error)) = runtime.reload(bad, |_| Ok(())) else { + panic!("expected key failure") + }; + assert_eq!(error.dkim_index, Some(1)); + assert!(error.to_string().contains("DKIM entry 1")); + assert!(!error.to_string().contains("/missing/key")); + assert_snapshot_retained(&runtime, &before); + let mut changed = initial.clone(); + let mut rotated = entry("example.com"); + rotated.selector = "two".into(); + changed.server.dkim = Some(CfgDkim::Domains(vec![rotated])); + runtime.reload(changed, |_| Ok(())).unwrap(); + assert!(runtime.load().signer_for(&body("another.com")).is_err()); + // Existing readers keep their complete snapshot through replacement. + assert!(before.signer_for(&body("another.com")).unwrap().is_some()); + let mut legacy = initial; + legacy.server.dkim = Some(entry("example.com").into()); + runtime.reload(legacy, |_| Ok(())).unwrap(); + assert!(runtime + .load() + .signer_for(&body("unknown.com")) + .unwrap() + .is_some()); + assert!(runtime + .load() + .signer_for(b"no From header") + .unwrap() + .is_some()); + } + #[test] fn tls_preparation_failure_retains_snapshot() { let _ = rustls::crypto::ring::default_provider().install_default(); @@ -576,6 +717,7 @@ mod tests { component: "server.listeners.tls.certificate", stage: "read", listener_index: Some(0), + dkim_index: None, })) )); assert_snapshot_retained(&runtime, &before); @@ -603,6 +745,7 @@ mod tests { component: "server.listeners.tls.certificate", stage: "parse", listener_index: Some(0), + dkim_index: None, }) )); assert!(!format!("{error:?}").contains("candidate-secret")); diff --git a/smtp-server/src/worker/mod.rs b/smtp-server/src/worker/mod.rs index 0ae34d0..2060a66 100644 --- a/smtp-server/src/worker/mod.rs +++ b/smtp-server/src/worker/mod.rs @@ -177,6 +177,15 @@ pub struct Worker { mta_sts: Arc, } +/// Bound the exponent before computing it, including for configurations that +/// allow more than 32 retries. Both queue backends use the same ceiling. +fn retry_delay(initial_delay: Duration, max_delay: Duration, attempts: u32) -> Duration { + std::cmp::min( + initial_delay.saturating_mul(2_u32.pow(attempts.min(24))), + max_delay, + ) +} + impl Worker { pub async fn new( channel: Receiver, @@ -384,6 +393,12 @@ impl Worker { // generic display — same as before for non-SMTP failures. let (outcome, smtp_response) = match e.downcast_ref::() { Some(c) => (c.outcome, c.smtp_response.clone()), + None if e + .downcast_ref::() + .is_some() => + { + (SendOutcome::Defer, format!("{e:#}")) + } None => (SendOutcome::Bounce, format!("{:#}", e)), }; @@ -442,8 +457,7 @@ impl Worker { } async fn defer_email(&self, job: &Job, smtp_response: &str) -> Result<()> { - let delay = self.initial_delay * (2_u32.pow(job.attempts)); - let delay = std::cmp::min(delay, self.max_delay); + let delay = retry_delay(self.initial_delay, self.max_delay, job.attempts); info!( msg_id = ?job.job_id, @@ -552,7 +566,7 @@ impl Worker { /// final outbound bytes with a single copy of the message. fn sign_outbound(&self, body: &[u8]) -> Result> { let runtime = self.runtime.load(); - let Some(signer) = &runtime.dkim_signer else { + let Some(signer) = runtime.signer_for(body)? else { let (out, _) = Self::strip_outbound_headers(body, false, 0) .wrap_err("Failed to remove Bcc header")?; return Ok(out); @@ -846,6 +860,23 @@ impl Worker { let raw_email = match self.sign_outbound(body) { Ok(raw) => raw, Err(e) => { + if e.downcast_ref::() + .is_some() + { + let delay = retry_delay(self.initial_delay, self.max_delay, job.attempts); + log_delivery( + "deferred", + &recipients.join(","), + &format!("{e:#}"), + job.attempts + 1, + ); + metrics::email_deferred(); + return JobOutcome::Deferred { + next_attempt_ms: Utc::now().timestamp_millis() + delay.as_millis() as i64, + remaining_recipients: recipients.to_vec(), + error: format!("preparing outbound message: {e:#}"), + }; + } return self .bounce_claim(job, body, format!("preparing outbound message: {e:#}")) .await; @@ -946,10 +977,7 @@ impl Worker { return JobOutcome::RateLimited { retry_after }; } } - let delay = std::cmp::min( - self.initial_delay * 2_u32.pow(job.attempts.min(24)), - self.max_delay, - ); + let delay = retry_delay(self.initial_delay, self.max_delay, job.attempts); metrics::email_deferred(); return JobOutcome::Deferred { next_attempt_ms: Utc::now().timestamp_millis() + delay.as_millis() as i64, @@ -1267,6 +1295,17 @@ mod tests { use crate::config::{CfgDKIM, DkimKeyType}; use std::str; + #[test] + fn retry_backoff_caps_large_attempt_counts_without_overflow() { + let initial = Duration::from_secs(60); + let maximum = Duration::from_secs(86400); + assert_eq!(retry_delay(initial, maximum, 0), initial); + assert_eq!(retry_delay(initial, maximum, 1), Duration::from_secs(120)); + for attempts in [11, 24, 31, 32, 64, u32::MAX] { + assert_eq!(retry_delay(initial, maximum, attempts), maximum); + } + } + #[test] fn test_create_dkim_signer_accepts_pkcs1_rsa_pem() { use rsa::pkcs1::EncodeRsaPrivateKey; diff --git a/smtp/src/lib.rs b/smtp/src/lib.rs index 4754d54..477166f 100644 --- a/smtp/src/lib.rs +++ b/smtp/src/lib.rs @@ -162,6 +162,10 @@ pub enum SmtpError { #[error("Mail rejected: {message}")] RcptToDenied { message: String }, + /// A permanent message policy failure after DATA. + #[error("Message rejected: {message}")] + DataRejected { message: String }, + #[error("Authentication error")] #[diagnostic(code(smtp::auth_error))] AuthError, @@ -690,10 +694,19 @@ impl SmtpServer { } data_buffer.clear(); *data_scanned = 0; - self.callbacks + match self + .callbacks .on_data(std::mem::take(&mut session.email)) - .await?; - stream.write_line(b"250 OK\r\n").await?; + .await + { + Ok(()) => stream.write_line(b"250 OK\r\n").await?, + Err(SmtpError::DataRejected { message }) => { + stream + .write_line(format!("550 {message}\r\n").as_bytes()) + .await?; + } + Err(error) => return Err(error.into()), + } session.state = SessionState::Authenticated; } None => { @@ -1177,6 +1190,7 @@ mod tests { struct RecordingCallbacks { emails: StdMutex>, + reject_first: std::sync::atomic::AtomicBool, } #[async_trait] @@ -1197,6 +1211,14 @@ mod tests { Ok(()) } async fn on_data(&self, email: Email) -> Result<(), SmtpError> { + if self + .reject_first + .swap(false, std::sync::atomic::Ordering::Relaxed) + { + return Err(SmtpError::DataRejected { + message: "5.7.1 Sending domain is not configured".into(), + }); + } self.emails.lock().unwrap().push(email); Ok(()) } @@ -1210,6 +1232,7 @@ mod tests { async fn run_chunked_data_session(body: &str, chunk_size: usize) -> String { let callbacks = Arc::new(RecordingCallbacks { emails: StdMutex::new(Vec::new()), + reject_first: std::sync::atomic::AtomicBool::new(false), }); let server = SmtpServer { callbacks: callbacks.clone(), @@ -1283,6 +1306,7 @@ mod tests { async fn recipient_count_is_capped_per_message() { let callbacks = Arc::new(RecordingCallbacks { emails: StdMutex::new(Vec::new()), + reject_first: std::sync::atomic::AtomicBool::new(false), }); let server = SmtpServer { callbacks: callbacks.clone(), @@ -1355,6 +1379,18 @@ mod tests { assert!(!emails[0].to.iter().any(|r| r.contains("one-too-many"))); } + #[tokio::test] + async fn data_rejection_preserves_pipelined_next_transaction() { + let (reply, emails) = run_session_with_rejection(DEFAULT_MAX_MESSAGE_SIZE, &[ + b"EHLO client.test\r\nMAIL FROM:\r\nRCPT TO:\r\nDATA\r\nFrom: first@example.com\r\n\r\nrejected\r\n.\r\nMAIL FROM:\r\nRCPT TO:\r\nDATA\r\nFrom: second@example.com\r\n\r\naccepted\r\n.\r\nQUIT\r\n" + ], true).await; + assert_eq!(reply.matches("550 5.7.1").count(), 1, "{reply}"); + assert!(reply.contains("221"), "{reply}"); + assert_eq!(emails.len(), 1, "{reply}"); + assert!(emails[0].from.contains("second@example.com")); + assert!(emails[0].body.ends_with(b"accepted")); + } + /// Drives a session over a duplex stream and returns every byte the /// server wrote, given a list of client writes performed after DATA is /// acknowledged. Each write goes out as one segment. @@ -1363,8 +1399,17 @@ mod tests { } async fn run_session_with(max_message_size: usize, writes: &[&[u8]]) -> (String, Vec) { + run_session_with_rejection(max_message_size, writes, false).await + } + + async fn run_session_with_rejection( + max_message_size: usize, + writes: &[&[u8]], + reject_first: bool, + ) -> (String, Vec) { let callbacks = Arc::new(RecordingCallbacks { emails: StdMutex::new(Vec::new()), + reject_first: std::sync::atomic::AtomicBool::new(reject_first), }); let server = SmtpServer { callbacks: callbacks.clone(), @@ -1791,6 +1836,7 @@ mod tests { // RecordingCallbacks keeps the default supports_cram_md5() == false. let callbacks = Arc::new(RecordingCallbacks { emails: StdMutex::new(Vec::new()), + reject_first: std::sync::atomic::AtomicBool::new(false), }); let server = SmtpServer { callbacks,