From 56aec86a860e09bb2adcaa01c955daa185beb0a2 Mon Sep 17 00:00:00 2001 From: mdecimus Date: Fri, 21 Jul 2023 20:21:51 +0200 Subject: [PATCH] Milter implementation. --- Cargo.lock | 176 +++---- crates/jmap/Cargo.toml | 2 +- crates/jmap/src/api/http.rs | 1 + crates/smtp/src/config/mod.rs | 20 +- crates/smtp/src/config/session.rs | 61 ++- crates/smtp/src/core/mod.rs | 9 +- crates/smtp/src/inbound/data.rs | 14 +- crates/smtp/src/inbound/milter/client.rs | 366 +++++++++++++ crates/smtp/src/inbound/milter/macros.rs | 201 +++++++ crates/smtp/src/inbound/milter/message.rs | 496 ++++++++++++++++++ crates/smtp/src/inbound/milter/mod.rs | 508 ++++++++++++++++++ crates/smtp/src/inbound/milter/protocol.rs | 549 ++++++++++++++++++++ crates/smtp/src/inbound/milter/receiver.rs | 127 +++++ crates/smtp/src/inbound/mod.rs | 32 +- crates/smtp/src/inbound/session.rs | 32 +- crates/smtp/src/inbound/spawn.rs | 2 +- crates/smtp/src/lib.rs | 1 + crates/smtp/src/outbound/dane/verify.rs | 147 ------ crates/utils/src/listener/listen.rs | 1 + crates/utils/src/listener/mod.rs | 1 + tests/Cargo.toml | 2 +- tests/resources/smtp/config/rules-eval.toml | 5 +- tests/resources/smtp/milter/message.eml | 11 + tests/resources/smtp/milter/message.json | 149 ++++++ tests/src/smtp/config.rs | 135 ++++- tests/src/smtp/inbound/milter.rs | 533 +++++++++++++++++++ tests/src/smtp/inbound/mod.rs | 1 + tests/src/smtp/mod.rs | 36 +- tests/src/smtp/outbound/dane.rs | 142 ++++- tests/src/smtp/session.rs | 10 +- 30 files changed, 3466 insertions(+), 304 deletions(-) create mode 100644 crates/smtp/src/inbound/milter/client.rs create mode 100644 crates/smtp/src/inbound/milter/macros.rs create mode 100644 crates/smtp/src/inbound/milter/message.rs create mode 100644 crates/smtp/src/inbound/milter/mod.rs create mode 100644 crates/smtp/src/inbound/milter/protocol.rs create mode 100644 crates/smtp/src/inbound/milter/receiver.rs create mode 100644 tests/resources/smtp/milter/message.eml create mode 100644 tests/resources/smtp/milter/message.json create mode 100644 tests/src/smtp/inbound/milter.rs diff --git a/Cargo.lock b/Cargo.lock index 814d556c..93eaf633 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -247,7 +247,7 @@ checksum = "0e97ce7de6cf12de5d7226c73f5ba9811622f4db3a5b91b55c53e987e5f91cba" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -269,18 +269,18 @@ checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] name = "async-trait" -version = "0.1.71" +version = "0.1.72" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a564d521dd56509c4c47480d00b80ee55f7e385ae48db5744c67ad50c92d2ebf" +checksum = "cc6dde6e4ed435a4c1ee4e73592f5ba9da2151af10076cc04858746af9352d09" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -342,9 +342,9 @@ dependencies = [ [[package]] name = "axum" -version = "0.6.18" +version = "0.6.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8175979259124331c1d7bf6586ee7e0da434155e4b2d48ec2c8386281d8df39" +checksum = "a6a1de45611fdb535bfde7b7de4fd54f4fd2b17b1737c0a59b69bf9b92074b8c" dependencies = [ "async-trait", "axum-core", @@ -485,13 +485,13 @@ dependencies = [ "lazycell", "log", "peeking_take_while", - "prettyplease 0.2.10", + "prettyplease 0.2.12", "proc-macro2", "quote", "regex", "rustc-hash", "shlex", - "syn 2.0.26", + "syn 2.0.27", "which", ] @@ -697,9 +697,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.3.12" +version = "4.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3eab9e8ceb9afdade1ab3f0fd8dbce5b1b2f468ad653baf10e771781b2b67b73" +checksum = "5b0827b011f6f8ab38590295339817b0d26f344aa4932c3ced71b45b0c54b4a9" dependencies = [ "clap_builder", "clap_derive", @@ -708,9 +708,9 @@ dependencies = [ [[package]] name = "clap_builder" -version = "4.3.12" +version = "4.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f2763db829349bf00cfc06251268865ed4363b93a943174f638daf3ecdba2cd" +checksum = "9441b403be87be858db6a23edb493e7f694761acdc3343d5a0fcaafd304cbc9e" dependencies = [ "anstream", "anstyle", @@ -727,7 +727,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -1100,7 +1100,7 @@ checksum = "487585f4d0c6655fe74905e2504d8ad6908e4db67f744eb140876906c2f3175d" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -1117,9 +1117,9 @@ checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" [[package]] name = "ecdsa" -version = "0.16.7" +version = "0.16.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0997c976637b606099b9985693efa3581e84e41f5c11ba5255f88711058ad428" +checksum = "a4b1e0c257a9e9f25f90ff76d7a68360ed497ee519c8e428d1825ef0000799d4" dependencies = [ "der", "digest 0.10.7", @@ -1280,12 +1280,9 @@ checksum = "95765f67b4b18863968b4a1bd5bb576f732b29a4a28c7cd84c09fa3e2875f33c" [[package]] name = "fastrand" -version = "1.9.0" +version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be" -dependencies = [ - "instant", -] +checksum = "6999dc1837253364c2ebb0704ba97994bd874e8f195d665c50b7548f6ea92764" [[package]] name = "ff" @@ -1422,7 +1419,7 @@ checksum = "83c8d52fe8b46ab822b4decdcc0d6d85aeedfc98f0d52ba2bd4aec4a97807516" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", "try_map", ] @@ -1502,7 +1499,7 @@ checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -2006,17 +2003,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "io-lifetimes" -version = "1.0.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eae7b9aee968036d54dce06cebaefd919e4472e753296daccd6d344e3e2df0c2" -dependencies = [ - "hermit-abi", - "libc", - "windows-sys 0.48.0", -] - [[package]] name = "ipconfig" version = "0.3.2" @@ -2042,7 +2028,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cb0889898416213fab133e1d33a0e5858a48177452750691bde3666d0fdbaf8b" dependencies = [ "hermit-abi", - "rustix 0.38.4", + "rustix", "windows-sys 0.48.0", ] @@ -2083,13 +2069,13 @@ dependencies = [ [[package]] name = "jieba-rs" -version = "0.6.7" +version = "0.6.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "37228e06c75842d1097432d94d02f37fe3ebfca9791c2e8fef6e9db17ed128c1" +checksum = "93f0c1347cd3ac8d7c6e3a2dc33ac496d365cf09fc0831aa61111e1a6738983e" dependencies = [ "cedarwood", "fxhash", - "hashbrown 0.12.3", + "hashbrown 0.14.0", "lazy_static", "phf", "phf_codegen", @@ -2309,12 +2295,6 @@ version = "0.5.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0717cef1bc8b636c6e1c1bbdefc09e6322da8a9321966e8928ef80d20f7f770f" -[[package]] -name = "linux-raw-sys" -version = "0.3.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519" - [[package]] name = "linux-raw-sys" version = "0.4.3" @@ -2359,7 +2339,7 @@ dependencies = [ [[package]] name = "mail-auth" version = "0.3.2" -source = "git+https://github.com/stalwartlabs/mail-auth#0c93f7b18a6370bd3c86671e4d121c0f0b2f0b23" +source = "git+https://github.com/stalwartlabs/mail-auth#a6cd1d6cc0a79943903e8154eecc29f2de003e2a" dependencies = [ "ahash 0.8.3", "flate2", @@ -2400,10 +2380,7 @@ source = "git+https://github.com/stalwartlabs/mail-send#d5ac9b328308fd95709cb8ee dependencies = [ "base64 0.20.0", "gethostname", - "mail-auth", - "mail-builder", "md5", - "rand", "rustls 0.21.5", "smtp-proto", "tokio", @@ -2679,9 +2656,9 @@ dependencies = [ [[package]] name = "num-traits" -version = "0.2.15" +version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "578ede34cf02f8924ab9447f50c28075b4d3e5b269972345e7e0372b38c6cdcd" +checksum = "f30b0abd723be7e2ffca1272140fac1a2f084c77ec3e123c192b66af1ee9e6c2" dependencies = [ "autocfg", "libm", @@ -2756,7 +2733,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -3065,7 +3042,7 @@ dependencies = [ "phf_shared", "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -3094,7 +3071,7 @@ checksum = "ec2e072ecce94ec471b13398d5402c188e76ac03cf74dd1a975161b23a3f6d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -3172,12 +3149,12 @@ dependencies = [ [[package]] name = "prettyplease" -version = "0.2.10" +version = "0.2.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92139198957b410250d43fad93e630d956499a625c527eda65175c8680f83387" +checksum = "6c64d9ba0963cdcea2e1b2230fbae2bab30eb25a174be395c41e764bfb65dd62" dependencies = [ "proc-macro2", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -3215,9 +3192,9 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.65" +version = "1.0.66" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92de25114670a878b1261c79c9f8f729fb97e95bac93f6312f583c60dd6a1dfe" +checksum = "18fb31db3f9bddb2ea821cde30a9f70117e3f119938b5ee630b7403aa6e2ead9" dependencies = [ "unicode-ident", ] @@ -3318,9 +3295,9 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.30" +version = "1.0.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5907a1b7c277254a8b15170f6e7c97cfa60ee7872a3217663bb81151e48184bb" +checksum = "5fe8a65d69dd0808184ebb5f836ab526bb259db23c657efa38711b1072ee47f0" dependencies = [ "proc-macro2", ] @@ -3694,20 +3671,6 @@ dependencies = [ "nom", ] -[[package]] -name = "rustix" -version = "0.37.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d69718bf81c6127a49dc64e44a742e8bb9213c0ff8869a22c308f84c1d4ab06" -dependencies = [ - "bitflags 1.3.2", - "errno", - "io-lifetimes", - "libc", - "linux-raw-sys 0.3.8", - "windows-sys 0.48.0", -] - [[package]] name = "rustix" version = "0.38.4" @@ -3717,7 +3680,7 @@ dependencies = [ "bitflags 2.3.3", "errno", "libc", - "linux-raw-sys 0.4.3", + "linux-raw-sys", "windows-sys 0.48.0", ] @@ -3827,9 +3790,9 @@ dependencies = [ [[package]] name = "scopeguard" -version = "1.1.0" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] name = "scrypt" @@ -3855,9 +3818,9 @@ dependencies = [ [[package]] name = "sec1" -version = "0.7.2" +version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0aec48e813d6b90b15f0b8948af3c63483992dee44c03e9930b3eebdabe046e" +checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" dependencies = [ "base16ct", "der", @@ -3892,9 +3855,9 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.171" +version = "1.0.174" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30e27d1e4fd7659406c492fd6cfaf2066ba8773de45ca75e855590f856dc34a9" +checksum = "3b88756493a5bd5e5395d53baa70b194b05764ab85b59e43e4b8f4e1192fa9b1" dependencies = [ "serde_derive", ] @@ -3910,13 +3873,13 @@ dependencies = [ [[package]] name = "serde_derive" -version = "1.0.171" +version = "1.0.174" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "389894603bd18c46fa56231694f8d827779c0951a667087194cf9de94ed24682" +checksum = "6e5c3a298c7f978e53536f95a63bdc4c4a64550582f31a0359a9afda6aede62e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -3964,7 +3927,7 @@ checksum = "91d129178576168c589c9ec973feedf7d3126c01ac2bf08795109aa35b69fb8f" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -4474,9 +4437,9 @@ dependencies = [ [[package]] name = "stringprep" -version = "0.1.2" +version = "0.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ee348cb74b87454fff4b551cbf727025810a004f88aeacae7f85b87f4e9a1c1" +checksum = "db3737bde7edce97102e0e2b15365bf7a20bfdb5f60f4f9e8d7004258a51a8da" dependencies = [ "unicode-bidi", "unicode-normalization", @@ -4507,9 +4470,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.26" +version = "2.0.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "45c3457aacde3c65315de5031ec191ce46604304d2446e803d71ade03308d970" +checksum = "b60f673f44a8255b9c8c657daf66a596d435f2da81a555b06dc644d080ba45e0" dependencies = [ "proc-macro2", "quote", @@ -4547,15 +4510,14 @@ dependencies = [ [[package]] name = "tempfile" -version = "3.6.0" +version = "3.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "31c0432476357e58790aaa47a8efb0c5138f137343f3b5f23bd36a27e3b0a6d6" +checksum = "5486094ee78b2e5038a6382ed7645bc084dc2ec433426ca4c3cb61e2007b8998" dependencies = [ - "autocfg", "cfg-if", "fastrand", "redox_syscall 0.3.5", - "rustix 0.37.23", + "rustix", "windows-sys 0.48.0", ] @@ -4620,22 +4582,22 @@ dependencies = [ [[package]] name = "thiserror" -version = "1.0.43" +version = "1.0.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a35fc5b8971143ca348fa6df4f024d4d55264f3468c71ad1c2f365b0a4d58c42" +checksum = "611040a08a0439f8248d1990b111c95baa9c704c805fa1f62104b39655fd7f90" dependencies = [ "thiserror-impl", ] [[package]] name = "thiserror-impl" -version = "1.0.43" +version = "1.0.44" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "463fe12d7993d3b327787537ce8dd4dfa058de32fc2b195ef3cde03dc4771e8f" +checksum = "090198534930841fab3a5d1bb637cde49e339654e606195f8d9c76eeb081dc96" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -4749,7 +4711,7 @@ checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -4922,7 +4884,7 @@ checksum = "5f4f31f56159e98206da9efd823404b79b6ef3143b4a7ab76e67b1751b25a4ab" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", ] [[package]] @@ -5190,9 +5152,9 @@ dependencies = [ [[package]] name = "uuid" -version = "1.4.0" +version = "1.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d023da39d1fde5a8a3fe1f3e01ca9632ada0a63e9797de55a879d6e2236277be" +checksum = "79daa5ed5740825c40b389c5e50312b9c86df53fccd33f281df655642b43869d" [[package]] name = "valuable" @@ -5254,7 +5216,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", "wasm-bindgen-shared", ] @@ -5288,7 +5250,7 @@ checksum = "54681b18a46765f095758388f2d0cf16eb8d4169b639ab575a8f5693af210c7b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.26", + "syn 2.0.27", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -5602,9 +5564,9 @@ dependencies = [ [[package]] name = "xml-rs" -version = "0.8.15" +version = "0.8.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a56c84a8ccd4258aed21c92f70c0f6dea75356b6892ae27c24139da456f9336" +checksum = "47430998a7b5d499ccee752b41567bc3afc57e1327dc855b1a2aa44ce29b5fa1" [[package]] name = "xxhash-rust" diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index 0e750730..6d75b4d0 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -13,7 +13,7 @@ directory = { path = "../directory" } smtp-proto = { git = "https://github.com/stalwartlabs/smtp-proto" } mail-parser = { git = "https://github.com/stalwartlabs/mail-parser", features = ["full_encoding", "serde_support", "ludicrous_mode"] } mail-builder = { git = "https://github.com/stalwartlabs/mail-builder", features = ["ludicrous_mode"] } -mail-send = { git = "https://github.com/stalwartlabs/mail-send" } +mail-send = { git = "https://github.com/stalwartlabs/mail-send", default-features = false, features = ["cram-md5", "skip-ehlo"] } sieve-rs = { git = "https://github.com/stalwartlabs/sieve" } serde = { version = "1.0", features = ["derive"]} serde_json = "1.0" diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index 4490519b..947983ff 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -369,6 +369,7 @@ impl SessionManager for JmapSessionManager { stream, local_ip: session.local_ip, remote_ip: session.remote_ip, + remote_port: session.remote_port, span, in_flight: session.in_flight, instance: session.instance, diff --git a/crates/smtp/src/config/mod.rs b/crates/smtp/src/config/mod.rs index 95b65ceb..f9746d4f 100644 --- a/crates/smtp/src/config/mod.rs +++ b/crates/smtp/src/config/mod.rs @@ -33,7 +33,7 @@ pub mod session; pub mod throttle; use std::{ - net::{Ipv4Addr, Ipv6Addr}, + net::{Ipv4Addr, Ipv6Addr, SocketAddr}, path::PathBuf, sync::{atomic::AtomicU64, Arc}, time::Duration, @@ -52,6 +52,8 @@ use sieve::Sieve; use smtp_proto::MtPriority; use utils::config::{Rate, Server, ServerProtocol}; +use crate::inbound::milter; + #[derive(Debug)] pub struct Host { pub address: String, @@ -247,6 +249,7 @@ pub struct Rcpt { pub struct Data { pub script: IfBlock>>, pub pipe_commands: Vec, + pub milters: Vec, // Limits pub max_messages: IfBlock, @@ -268,6 +271,21 @@ pub struct Pipe { pub timeout: IfBlock, } +pub struct Milter { + pub enable: IfBlock, + pub addrs: Vec, + pub hostname: String, + pub port: u16, + pub timeout_connect: Duration, + pub timeout_command: Duration, + pub timeout_data: Duration, + pub tls: bool, + pub tls_allow_invalid_certs: bool, + pub tempfail_on_error: bool, + pub max_frame_len: usize, + pub protocol_version: milter::Version, +} + pub struct SessionConfig { pub timeout: IfBlock, pub duration: IfBlock, diff --git a/crates/smtp/src/config/session.rs b/crates/smtp/src/config/session.rs index 1fab268e..a50c0467 100644 --- a/crates/smtp/src/config/session.rs +++ b/crates/smtp/src/config/session.rs @@ -21,7 +21,7 @@ * for more details. */ -use std::time::Duration; +use std::{net::ToSocketAddrs, time::Duration}; use smtp_proto::*; @@ -46,6 +46,11 @@ pub trait ConfigSession { ctx: &ConfigContext, available_keys: &[EnvelopeKey], ) -> super::Result>; + fn parse_milters( + &self, + ctx: &ConfigContext, + available_keys: &[EnvelopeKey], + ) -> super::Result>; } impl ConfigSession for Config { @@ -384,6 +389,7 @@ impl ConfigSession for Config { .parse_if_block("session.data.add-headers.date", ctx, &available_keys)? .unwrap_or_else(|| IfBlock::new(true)), pipe_commands: self.parse_pipes(ctx, &available_keys)?, + milters: self.parse_milters(ctx, &available_keys)?, }) } @@ -408,6 +414,59 @@ impl ConfigSession for Config { } Ok(pipes) } + + fn parse_milters( + &self, + ctx: &ConfigContext, + available_keys: &[EnvelopeKey], + ) -> super::Result> { + let mut milters = Vec::new(); + for id in self.sub_keys("session.data.milter") { + let hostname = self + .value_require(("session.data.milter", id, "hostname"))? + .to_string(); + let port = self.property_require(("session.data.milter", id, "port"))?; + milters.push(Milter { + enable: self + .parse_if_block(("session.data.milter", id, "enable"), ctx, available_keys)? + .unwrap_or_default(), + addrs: format!("{}:{}", hostname, port) + .to_socket_addrs() + .map_err(|err| format!("Unable to resolve milter hostname {hostname}: {err}"))? + .collect(), + hostname, + port, + timeout_connect: self + .property_or_static(("session.data.milter", id, "timeout.connect"), "30s")?, + timeout_command: self + .property_or_static(("session.data.milter", id, "timeout.command"), "30s")?, + timeout_data: self + .property_or_static(("session.data.milter", id, "timeout.data"), "60s")?, + tls: self.property_or_static(("session.data.milter", id, "tls"), "false")?, + tls_allow_invalid_certs: self.property_or_static( + ("session.data.milter", id, "allow-invalid-certs"), + "false", + )?, + tempfail_on_error: self.property_or_static( + ("session.data.milter", id, "options.tempfail-on-error"), + "true", + )?, + max_frame_len: self.property_or_static( + ("session.data.milter", id, "options.max-response-size"), + "52428800", + )?, + protocol_version: match self.property_or_static::( + ("session.data.milter", id, "options.version"), + "2", + )? { + 6 => milter::Version::V6, + 2 => milter::Version::V2, + v => return Err(format!("Unsupported milter protocol version: {}", v)), + }, + }) + } + Ok(milters) + } } struct Mechanism { diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index 7d15e6ec..4d4d1fe5 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -180,6 +180,7 @@ pub struct Session { pub struct SessionData { pub local_ip: IpAddr, pub remote_ip: IpAddr, + pub remote_port: u16, pub helo_domain: String, pub mail_from: Option, @@ -248,10 +249,11 @@ pub struct SessionParameters { } impl SessionData { - pub fn new(local_ip: IpAddr, remote_ip: IpAddr) -> Self { + pub fn new(local_ip: IpAddr, remote_ip: IpAddr, remote_port: u16) -> Self { SessionData { local_ip, remote_ip, + remote_port, helo_domain: String::new(), mail_from: None, rcpt_to: Vec::new(), @@ -404,6 +406,10 @@ impl crate::inbound::IsTls for NullIo { } fn write_tls_header(&self, _headers: &mut Vec) {} + + fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { + ("", "") + } } #[cfg(feature = "local_delivery")] @@ -514,6 +520,7 @@ impl SessionData { SessionData { local_ip: IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1)), remote_ip: IpAddr::V4(std::net::Ipv4Addr::new(127, 0, 0, 1)), + remote_port: 0, helo_domain: "localhost".into(), mail_from, rcpt_to, diff --git a/crates/smtp/src/inbound/data.rs b/crates/smtp/src/inbound/data.rs index a0a39e79..f02ef917 100644 --- a/crates/smtp/src/inbound/data.rs +++ b/crates/smtp/src/inbound/data.rs @@ -284,8 +284,16 @@ impl Session { } } + // Run Milter filters + let mut edited_message = match self.run_milters(&auth_message).await { + Ok(modifications) => self + .data + .apply_modifications(modifications, &auth_message) + .map(Arc::new), + Err(response) => return response, + }; + // Pipe message - let mut edited_message = None; for pipe in &dc.pipe_commands { if let Some(command_) = pipe.command.eval(self).await { let piped_message = edited_message.as_ref().unwrap_or(&raw_message).clone(); @@ -387,8 +395,8 @@ impl Session { } ScriptResult::Reject(message) => { tracing::debug!(parent: &self.span, - context = "data", - event = "sieve-reject", + context = "sieve", + event = "reject", reason = message); return message.into_bytes().into(); diff --git a/crates/smtp/src/inbound/milter/client.rs b/crates/smtp/src/inbound/milter/client.rs new file mode 100644 index 00000000..ba18d4cd --- /dev/null +++ b/crates/smtp/src/inbound/milter/client.rs @@ -0,0 +1,366 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use rustls::ServerName; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::TcpStream, +}; +use tokio_rustls::{client::TlsStream, TlsConnector}; + +use crate::config::Milter; + +use super::{ + protocol::{SMFIC_CONNECT, SMFIC_HELO, SMFIC_MAIL, SMFIC_RCPT}, + receiver::{FrameResult, Receiver}, + *, +}; + +const MILTER_CHUNK_SIZE: usize = 65535; + +impl MilterClient { + pub async fn connect(config: &Milter, span: tracing::Span) -> Result { + tokio::time::timeout(config.timeout_command, async { + let mut last_err = Error::Disconnected; + for addr in &config.addrs { + match TcpStream::connect(addr).await { + Ok(stream) => { + return Ok(MilterClient { + stream, + timeout_cmd: config.timeout_command, + timeout_data: config.timeout_data, + buf: vec![0u8; 8192], + bytes_read: 0, + receiver: Receiver::with_max_frame_len(config.max_frame_len), + options: 0, + version: config.protocol_version, + span, + }); + } + Err(err) => { + last_err = Error::Io(err); + } + } + } + Err(last_err) + }) + .await + .map_err(|_| Error::Timeout)? + } + + pub async fn into_tls( + self, + tls_connector: &TlsConnector, + tls_hostname: &str, + ) -> Result>> { + tokio::time::timeout(self.timeout_cmd, async { + Ok(MilterClient { + stream: tls_connector + .connect( + ServerName::try_from(tls_hostname).map_err(|_| Error::TLSInvalidName)?, + self.stream, + ) + .await?, + buf: self.buf, + timeout_cmd: self.timeout_cmd, + timeout_data: self.timeout_data, + receiver: self.receiver, + bytes_read: self.bytes_read, + options: self.options, + version: self.version, + span: self.span, + }) + }) + .await + .map_err(|_| Error::Timeout)? + } +} + +impl MilterClient { + pub async fn init(&mut self) -> super::Result { + self.write(Command::OptionNegotiation(Options { + version: match self.version { + Version::V2 => 2, + Version::V6 => 6, + }, + actions: SMFIF_ADDHDRS + | SMFIF_CHGBODY + | SMFIF_ADDRCPT + | SMFIF_DELRCPT + | SMFIF_CHGHDRS + | SMFIF_QUARANTINE + | SMFIF_CHGFROM + | SMFIF_ADDRCPT_PAR, + protocol: SMFIP_SKIP, + })) + .await?; + match self.read().await? { + Response::OptionNegotiation(options) => { + self.options = options.protocol; + Ok(options) + } + response => Err(Error::Unexpected(response)), + } + } + + pub async fn connection( + &mut self, + hostname: impl AsRef<[u8]>, + remote_ip: IpAddr, + remote_port: u16, + macros: Macros<'_>, + ) -> super::Result { + if !self.has_option(SMFIP_NOCONNECT) { + self.write(Command::Macro { + macros: macros.with_cmd_code(SMFIC_CONNECT), + }) + .await?; + self.write(Command::Connect { + hostname: hostname.as_ref(), + port: remote_port, + address: remote_ip, + }) + .await?; + if !self.has_option(SMFIP_NR_CONN) { + return self.read().await?.into_action(); + } + } + + Ok(Action::Accept) + } + + pub async fn helo( + &mut self, + hostname: impl AsRef<[u8]>, + macros: Macros<'_>, + ) -> super::Result { + if !self.has_option(SMFIP_NOHELO) { + self.write(Command::Macro { + macros: macros.with_cmd_code(SMFIC_HELO), + }) + .await?; + self.write(Command::Helo { + hostname: hostname.as_ref(), + }) + .await?; + if !self.has_option(SMFIP_NR_HELO) { + return self.read().await?.into_action(); + } + } + Ok(Action::Accept) + } + + pub async fn mail_from( + &mut self, + addr: A, + params: Option<&[V]>, + macros: Macros<'_>, + ) -> super::Result + where + A: AsRef<[u8]>, + V: AsRef<[u8]>, + { + if !self.has_option(SMFIP_NOMAIL) { + self.write(Command::Macro { + macros: macros.with_cmd_code(SMFIC_MAIL), + }) + .await?; + self.write(Command::MailFrom { + sender: addr.as_ref(), + args: params.map(|params| params.iter().map(|value| value.as_ref()).collect()), + }) + .await?; + if !self.has_option(SMFIP_NR_MAIL) { + return self.read().await?.into_action(); + } + } + Ok(Action::Accept) + } + + pub async fn rcpt_to( + &mut self, + addr: A, + params: Option<&[V]>, + macros: Macros<'_>, + ) -> super::Result + where + A: AsRef<[u8]>, + V: AsRef<[u8]>, + { + if !self.has_option(SMFIP_NORCPT) { + self.write(Command::Macro { + macros: macros.with_cmd_code(SMFIC_RCPT), + }) + .await?; + self.write(Command::Rcpt { + recipient: addr.as_ref(), + args: params.map(|params| params.iter().map(|value| value.as_ref()).collect()), + }) + .await?; + if !self.has_option(SMFIP_NR_RCPT) { + return self.read().await?.into_action(); + } + } + Ok(Action::Accept) + } + + pub async fn headers(&mut self, headers: I) -> super::Result + where + I: Iterator, + H: AsRef<[u8]>, + V: AsRef<[u8]>, + { + if !self.has_option(SMFIP_NOHDRS) { + for (name, value) in headers { + self.write(Command::Header { + name: name.as_ref(), + value: value.as_ref(), + }) + .await?; + if !self.has_option(SMFIP_NR_HDR) { + match self.read().await? { + Response::Action(Action::Accept | Action::Continue) => (), + Response::Action(action) => return Ok(action), + response => return Err(Error::Unexpected(response)), + } + } + } + + // Write EndOfHeaders + self.write(Command::EndOfHeader).await?; + if !self.has_option(SMFIP_NR_EOH) { + return self.read().await?.into_action(); + } + } + Ok(Action::Accept) + } + + pub async fn data(&mut self) -> super::Result { + if matches!(self.version, Version::V6) && !self.has_option(SMFIP_NODATA) { + self.write(Command::Data).await?; + if !self.has_option(SMFIP_NR_DATA) { + return self.read().await?.into_action(); + } + } + Ok(Action::Accept) + } + + pub async fn body(&mut self, body: &[u8]) -> super::Result<(Action, Vec)> { + if !self.has_option(SMFIP_NOBODY) { + // Write body chunks + for value in body.chunks(MILTER_CHUNK_SIZE) { + self.write(Command::Body { value }).await?; + if !self.has_option(SMFIP_NR_BODY) { + match self.read().await? { + Response::Action(Action::Accept | Action::Continue) + | Response::Progress => (), + Response::Skip => break, + Response::Action(reject) => { + return Ok((reject, Vec::new())); + } + response => return Err(Error::Unexpected(response)), + } + } + } + + // Write EndOfBody + self.write(Command::EndOfBody).await?; + + // Collect responses + let mut modifications = Vec::new(); + loop { + match self.read().await? { + Response::Action(action) => { + return Ok((action, modifications)); + } + Response::Modification(modification) => { + modifications.push(modification); + } + Response::Progress => (), + unexpected => { + return Err(Error::Unexpected(unexpected)); + } + } + } + } else { + Ok((Action::Accept, vec![])) + } + } + + pub async fn abort(&mut self) -> super::Result<()> { + self.write(Command::Abort).await + } + + pub async fn quit(&mut self) -> super::Result<()> { + self.write(Command::Quit).await + } + + async fn write(&mut self, action: Command<'_>) -> super::Result<()> { + //let p = println!("Action: {}", action); + tracing::trace!(parent: &self.span, context = "milter", event = "write", "action" = action.to_string()); + + tokio::time::timeout(self.timeout_cmd, async { + self.stream.write_all(action.serialize().as_ref()).await?; + self.stream.flush().await.map_err(Error::Io) + }) + .await + .map_err(|_| Error::Timeout)? + } + + async fn read(&mut self) -> super::Result { + loop { + match self.receiver.read_frame(&self.buf[..self.bytes_read]) { + FrameResult::Frame(frame) => { + if let Some(response) = Response::deserialize(&frame) { + tracing::trace!(parent: &self.span, context = "milter", event = "read", "action" = response.to_string()); + //let p = println!("Response: {}", response); + return Ok(response); + } else { + return Err(Error::FrameInvalid(frame.into_owned())); + } + } + FrameResult::Incomplete => { + self.bytes_read = tokio::time::timeout(self.timeout_data, async { + self.stream.read(&mut self.buf).await.map_err(Error::Io) + }) + .await + .map_err(|_| Error::Timeout)??; + if self.bytes_read == 0 { + return Err(Error::Disconnected); + } + } + FrameResult::TooLarge(size) => return Err(Error::FrameTooLarge(size)), + } + } + } + + #[inline(always)] + fn has_option(&self, opt: u32) -> bool { + self.options & opt == opt + } + + pub fn with_version(mut self, version: Version) -> Self { + self.version = version; + self + } +} diff --git a/crates/smtp/src/inbound/milter/macros.rs b/crates/smtp/src/inbound/milter/macros.rs new file mode 100644 index 00000000..9986d7cc --- /dev/null +++ b/crates/smtp/src/inbound/milter/macros.rs @@ -0,0 +1,201 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::{borrow::Cow, net::IpAddr}; + +use super::{Macro, Macros}; + +pub trait IntoMacroValue<'x> { + fn into_macro_value(self) -> Cow<'x, [u8]>; +} + +impl<'x> Macros<'x> { + pub fn new() -> Self { + Macros::default() + } + + pub fn with_cmd_code(mut self, cmd_code: u8) -> Self { + self.cmdcode = cmd_code; + self + } + + pub fn with_macro(mut self, name: &'static [u8], value: impl IntoMacroValue<'x>) -> Self { + self.macros.push(Macro { + name, + value: value.into_macro_value(), + }); + self + } + + pub fn with_queue_id(self, queue_id: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"i", queue_id) + } + + pub fn with_local_hostname(self, my_hostname: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"j", my_hostname) + } + + pub fn with_validated_client_name(self, client_name: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"_", client_name) + } + + pub fn with_sasl_login_name(self, sasl_login_name: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{auth_authen}", sasl_login_name) + } + + pub fn with_sasl_sender(self, sasl_sender: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{auth_author}", sasl_sender) + } + + pub fn with_sasl_method(self, sasl_method: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{auth_type}", sasl_method) + } + + pub fn with_client_address(self, client_address: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{client_addr}", client_address) + } + + pub fn with_client_connections(self, client_connections: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{client_connections}", client_connections) + } + + pub fn with_client_name(self, client_name: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{client_name}", client_name) + } + + pub fn with_client_port(self, client_port: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{client_port}", client_port) + } + + pub fn with_client_ptr(self, client_ptr: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{client_ptr}", client_ptr) + } + + pub fn with_cert_issuer(self, cert_issuer: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{cert_issuer}", cert_issuer) + } + + pub fn with_cert_subject(self, cert_subject: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{cert_subject}", cert_subject) + } + + pub fn with_cipher_bits(self, cipher_bits: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{cipher_bits}", cipher_bits) + } + + pub fn with_cipher(self, cipher: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{cipher}", cipher) + } + + pub fn with_daemon_address(self, daemon_address: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{daemon_addr}", daemon_address) + } + + pub fn with_daemon_name(self, daemon_name: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{daemon_name}", daemon_name) + } + + pub fn with_daemon_port(self, daemon_port: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{daemon_port}", daemon_port) + } + + pub fn with_mail_address(self, mail_address: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{mail_addr}", mail_address) + } + + pub fn with_mail_host(self, mail_host_address: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{mail_host}", mail_host_address) + } + + pub fn with_mail_mailer(self, mail_mailer: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{mail_mailer}", mail_mailer) + } + + pub fn with_rcpt_address(self, rcpt_address: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{rcpt_addr}", rcpt_address) + } + + pub fn with_rcpt_host(self, rcpt_host: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{rcpt_host}", rcpt_host) + } + + pub fn with_rcpt_mailer(self, rcpt_mailer: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{rcpt_mailer}", rcpt_mailer) + } + + pub fn with_tls_version(self, tls_version: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{tls_version}", tls_version) + } + + pub fn with_version(self, version: impl IntoMacroValue<'x>) -> Self { + self.with_macro(b"{v}", version) + } +} + +impl<'x> IntoMacroValue<'x> for IpAddr { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Owned(self.to_string().into_bytes()) + } +} + +impl<'x> IntoMacroValue<'x> for u16 { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Owned(self.to_string().into_bytes()) + } +} + +impl<'x> IntoMacroValue<'x> for &'x [u8] { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Borrowed(self) + } +} + +impl<'x> IntoMacroValue<'x> for &'x str { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Borrowed(self.as_bytes()) + } +} + +impl<'x> IntoMacroValue<'x> for String { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Owned(self.into_bytes()) + } +} + +impl<'x> IntoMacroValue<'x> for &'x String { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Borrowed(self.as_bytes()) + } +} + +impl<'x> IntoMacroValue<'x> for Vec { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Owned(self) + } +} + +impl<'x> IntoMacroValue<'x> for &'x Vec { + fn into_macro_value(self) -> Cow<'x, [u8]> { + Cow::Borrowed(self) + } +} diff --git a/crates/smtp/src/inbound/milter/message.rs b/crates/smtp/src/inbound/milter/message.rs new file mode 100644 index 00000000..863f57c4 --- /dev/null +++ b/crates/smtp/src/inbound/milter/message.rs @@ -0,0 +1,496 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::borrow::Cow; + +use mail_auth::AuthenticatedMessage; +use smtp_proto::request::parser::Rfc5321Parser; +use tokio::io::{AsyncRead, AsyncWrite}; + +use crate::{ + config::Milter, + core::{Session, SessionAddress, SessionData}, + inbound::{milter::MilterClient, IsTls}, + queue::DomainPart, + DAEMON_NAME, +}; + +use super::{Action, Error, Macros, Modification}; + +enum Rejection { + Action(Action), + Error(Error), +} + +impl Session { + pub async fn run_milters( + &self, + message: &AuthenticatedMessage<'_>, + ) -> Result, Cow<'static, [u8]>> { + let milters = &self.core.session.config.data.milters; + if milters.is_empty() { + return Ok(Vec::new()); + } + + let mut modifications = Vec::new(); + for milter in milters { + if !*milter.enable.eval(self).await { + continue; + } + + match self.connect_and_run(milter, message).await { + Ok(new_modifications) => { + if !modifications.is_empty() { + // The message body can only be replaced once, so we need to remove + // any previous replacements. + modifications.retain(|m| !matches!(m, Modification::ReplaceBody { .. })); + modifications.extend(new_modifications); + } else { + modifications = new_modifications; + } + } + Err(Rejection::Action(action)) => { + tracing::debug!( + parent: &self.span, + milter.host = &milter.hostname, + milter.port = &milter.port, + context = "milter", + event = "reject", + action = ?action, + "Milter rejected message."); + + return Err(match action { + Action::Discard => { + (b"250 2.0.0 Message queued for delivery.\r\n"[..]).into() + } + Action::Reject => (b"503 5.5.3 Message rejected.\r\n"[..]).into(), + Action::TempFail => { + (b"451 4.3.5 Unable to accept message at this time.\r\n"[..]).into() + } + Action::ReplyCode { code, text } => { + let mut response = Vec::with_capacity(text.len() + 6); + response.extend_from_slice(code.as_slice()); + response.push(b' '); + response.extend_from_slice(text.as_bytes()); + if !text.ends_with('\n') { + response.extend_from_slice(b"\r\n"); + } + response.into() + } + Action::Shutdown => (b"421 4.3.0 Server shutting down.\r\n"[..]).into(), + Action::ConnectionFailure => (b""[..]).into(), // TODO: Not very elegant design, fix. + Action::Accept | Action::Continue => unreachable!(), + }); + } + Err(Rejection::Error(err)) => { + tracing::warn!( + parent: &self.span, + milter.host = &milter.hostname, + milter.port = &milter.port, + context = "milter", + event = "error", + reason = ?err, + "Milter filter failed"); + if milter.tempfail_on_error { + return Err( + (b"451 4.3.5 Unable to accept message at this time.\r\n"[..]).into(), + ); + } + } + } + } + + Ok(modifications) + } + + async fn connect_and_run( + &self, + milter: &Milter, + message: &AuthenticatedMessage<'_>, + ) -> Result, Rejection> { + // Build client + let client = MilterClient::connect(milter, self.span.clone()).await?; + if !milter.tls { + self.run(client, message).await + } else { + self.run( + client + .into_tls( + if !milter.tls_allow_invalid_certs { + &self.core.queue.connectors.pki_verify + } else { + &self.core.queue.connectors.dummy_verify + }, + &milter.hostname, + ) + .await?, + message, + ) + .await + } + } + + async fn run( + &self, + mut client: MilterClient, + message: &AuthenticatedMessage<'_>, + ) -> Result, Rejection> { + // Option negotiation + client.init().await?; + + // Connect stage + let client_ptr = self + .data + .iprev + .as_ref() + .and_then(|ip_rev| ip_rev.ptr.as_ref()) + .and_then(|ptrs| ptrs.first()); + client + .connection( + client_ptr.unwrap_or(&self.data.helo_domain), + self.data.remote_ip, + self.data.remote_port, + Macros::new() + .with_daemon_name(DAEMON_NAME) + .with_local_hostname(&self.instance.hostname) + .with_client_address(self.data.remote_ip) + .with_client_port(self.data.remote_port) + .with_client_ptr(client_ptr.map(|p| p.as_str()).unwrap_or("unknown")), + ) + .await? + .assert_continue()?; + + // EHLO/HELO + let (tls_version, tls_ciper) = self.stream.tls_version_and_cipher(); + client + .helo( + &self.data.helo_domain, + Macros::new() + .with_cipher(tls_ciper) + .with_tls_version(tls_version), + ) + .await? + .assert_continue()?; + + // Mail from + let addr = &self.data.mail_from.as_ref().unwrap().address_lcase; + client + .mail_from( + &format!("<{addr}>"), + None::<&[&str]>, + Macros::new() + .with_mail_address(addr) + .with_sasl_login_name(&self.data.authenticated_as), + ) + .await? + .assert_continue()?; + + // Rcpt to + for rcpt in &self.data.rcpt_to { + client + .rcpt_to( + &format!("<{}>", rcpt.address_lcase), + None::<&[&str]>, + Macros::new().with_rcpt_address(&rcpt.address_lcase), + ) + .await? + .assert_continue()?; + } + + // Headers + client + .headers(message.raw_parsed_headers().iter().cloned()) + .await? + .assert_continue()?; + + // Data + client.data().await?.assert_continue()?; + + // Message body + let (action, modifications) = client.body(message.raw_message()).await?; + action.assert_continue()?; + + // Quit + let _ = client.quit().await; + + // Return modifications + Ok(modifications) + } +} + +impl SessionData { + pub fn apply_modifications( + &mut self, + modifications: Vec, + message: &AuthenticatedMessage<'_>, + ) -> Option> { + let mut body = Vec::new(); + let mut header_changes = Vec::new(); + let mut needs_rewrite = false; + + for modification in modifications { + match modification { + Modification::ChangeFrom { sender, mut args } => { + // Change sender + let sender = strip_brackets(&sender); + let address_lcase = sender.to_lowercase(); + let mut mail_from = SessionAddress { + domain: address_lcase.domain_part().to_string(), + address_lcase, + address: sender, + flags: 0, + dsn_info: None, + }; + if !args.is_empty() { + args.push('\n'); + match Rfc5321Parser::new(&mut args.as_bytes().iter()) + .mail_from_parameters(String::new()) + { + Ok(addr) => { + mail_from.flags = addr.flags; + mail_from.dsn_info = addr.env_id; + } + Err(err) => { + tracing::debug!( + context = "milter", + event = "error", + reason = ?err, + "Failed to parse milter mailFrom parameters."); + } + } + } + self.mail_from = Some(mail_from); + } + Modification::AddRcpt { + recipient, + mut args, + } => { + // Add recipient + let recipient = strip_brackets(&recipient); + if recipient.contains('@') { + let address_lcase = recipient.to_lowercase(); + let mut rcpt = SessionAddress { + domain: address_lcase.domain_part().to_string(), + address_lcase, + address: recipient, + flags: 0, + dsn_info: None, + }; + if !args.is_empty() { + args.push('\n'); + match Rfc5321Parser::new(&mut args.as_bytes().iter()) + .rcpt_to_parameters(String::new()) + { + Ok(addr) => { + rcpt.flags = addr.flags; + rcpt.dsn_info = addr.orcpt; + } + Err(err) => { + tracing::debug!( + context = "milter", + event = "error", + reason = ?err, + "Failed to parse milter rcptTo parameters."); + } + } + } + + if !self.rcpt_to.contains(&rcpt) { + self.rcpt_to.push(rcpt); + } + } + } + Modification::DeleteRcpt { recipient } => { + let recipient = strip_brackets(&recipient); + self.rcpt_to.retain(|r| r.address_lcase != recipient); + } + Modification::ReplaceBody { value } => { + body.extend(value); + } + Modification::AddHeader { name, value } => { + header_changes.push((0, name, value, false)); + } + Modification::InsertHeader { index, name, value } => { + header_changes.push((index, name, value, false)); + needs_rewrite = true; + } + Modification::ChangeHeader { index, name, value } => { + if message + .raw_parsed_headers() + .iter() + .any(|(n, _)| n.eq_ignore_ascii_case(name.as_bytes())) + { + header_changes.push((index, name, value, true)); + needs_rewrite = true; + } else { + header_changes.push((0, name, value, false)); + } + } + Modification::Quarantine { reason } => { + header_changes.push((0, "X-Quarantine".to_string(), reason, false)); + } + } + } + + // If there are no header changes return + if header_changes.is_empty() { + return if !body.is_empty() { + let mut new_message = Vec::with_capacity(body.len() + message.raw_headers().len()); + new_message.extend_from_slice(message.raw_headers()); + new_message.extend(body); + Some(new_message) + } else { + None + }; + } + + let new_body = if !body.is_empty() { + &body[..] + } else { + message.raw_body() + }; + + if needs_rewrite { + let mut headers = message + .raw_parsed_headers() + .iter() + .map(|(h, v)| (Cow::from(*h), Cow::from(*v))) + .collect::>(); + + // Perform changes + for (index, header_name, header_value, is_change) in header_changes { + if is_change { + let mut header_count = 0; + for (pos, (name, value)) in headers.iter_mut().enumerate() { + if name.eq_ignore_ascii_case(header_name.as_bytes()) { + header_count += 1; + if header_count == index { + if !header_value.is_empty() { + *value = Cow::from(header_value.into_bytes()); + } else { + headers.remove(pos); + } + break; + } + } + } + } else { + let mut header_pos = 0; + if index > 0 { + let mut header_count = 0; + for (pos, (name, _)) in headers.iter().enumerate() { + if name.eq_ignore_ascii_case(header_name.as_bytes()) { + header_pos = pos; + header_count += 1; + if header_count == index { + break; + } + } + } + } + + headers.insert( + header_pos, + ( + Cow::from(header_name.into_bytes()), + Cow::from(header_value.into_bytes()), + ), + ); + } + } + + // Write new headers + let mut new_message = Vec::with_capacity( + new_body.len() + + message.raw_headers().len() + + headers + .iter() + .map(|(h, v)| h.len() + v.len() + 4) + .sum::(), + ); + for (header, value) in headers { + new_message.extend_from_slice(header.as_ref()); + if value.first().map_or(false, |c| c.is_ascii_whitespace()) { + new_message.extend_from_slice(b":"); + } else { + new_message.extend_from_slice(b": "); + } + new_message.extend_from_slice(value.as_ref()); + if !value.last().map_or(false, |c| *c == b'\n') { + new_message.extend_from_slice(b"\r\n"); + } + } + new_message.extend_from_slice(b"\r\n"); + new_message.extend(new_body); + Some(new_message) + } else { + let mut new_message = Vec::with_capacity( + new_body.len() + + message.raw_headers().len() + + header_changes + .iter() + .map(|(_, h, v, _)| h.len() + v.len() + 4) + .sum::(), + ); + for (_, header, value, _) in header_changes { + new_message.extend_from_slice(header.as_bytes()); + new_message.extend_from_slice(b": "); + new_message.extend_from_slice(value.as_bytes()); + if !value.ends_with('\n') { + new_message.extend_from_slice(b"\r\n"); + } + } + new_message.extend_from_slice(message.raw_headers()); + new_message.extend(new_body); + Some(new_message) + } + } +} + +impl Action { + fn assert_continue(self) -> Result<(), Rejection> { + match self { + Action::Continue | Action::Accept => Ok(()), + action => Err(Rejection::Action(action)), + } + } +} + +impl From for Rejection { + fn from(err: Error) -> Self { + Rejection::Error(err) + } +} + +fn strip_brackets(addr: &str) -> String { + let addr = addr.trim(); + if let Some(addr) = addr.strip_prefix('<') { + if let Some((addr, _)) = addr.rsplit_once('>') { + addr.trim().to_string() + } else { + addr.trim().to_string() + } + } else { + addr.to_string() + } +} diff --git a/crates/smtp/src/inbound/milter/mod.rs b/crates/smtp/src/inbound/milter/mod.rs new file mode 100644 index 00000000..e7c346fa --- /dev/null +++ b/crates/smtp/src/inbound/milter/mod.rs @@ -0,0 +1,508 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::{borrow::Cow, fmt::Display, net::IpAddr, time::Duration}; + +use serde::{Deserialize, Serialize}; +use tokio::io::{AsyncRead, AsyncWrite}; + +use self::receiver::Receiver; + +pub mod client; +pub mod macros; +pub mod message; +pub mod protocol; +pub mod receiver; + +pub struct MilterClient { + stream: T, + buf: Vec, + bytes_read: usize, + timeout_cmd: Duration, + timeout_data: Duration, + receiver: Receiver, + version: Version, + options: u32, + span: tracing::Span, +} + +#[derive(Debug)] +pub enum Error { + Io(std::io::Error), + FrameTooLarge(usize), + FrameInvalid(Vec), + Unexpected(Response), + Timeout, + TLSInvalidName, + Disconnected, +} + +impl From for Error { + fn from(err: std::io::Error) -> Self { + Error::Io(err) + } +} + +pub enum Command<'x> { + Abort, + Body { + value: &'x [u8], + }, + EndOfBody, + Data, + Connect { + hostname: &'x [u8], + port: u16, + address: IpAddr, + }, + Macro { + macros: Macros<'x>, + }, + Header { + name: &'x [u8], + value: &'x [u8], + }, + EndOfHeader, + Helo { + hostname: &'x [u8], + }, + MailFrom { + sender: &'x [u8], + args: Option>, + }, + Rcpt { + recipient: &'x [u8], + args: Option>, + }, + OptionNegotiation(Options), + Quit, + QuitNewConnection, +} + +#[derive(Debug)] +pub enum Response { + Action(Action), + Modification(Modification), + Progress, + Skip, + SetSymbols, + OptionNegotiation(Options), +} + +#[derive(Debug)] +pub enum Action { + Accept, + Continue, + Discard, + Reject, + TempFail, + ReplyCode { code: [u8; 3], text: String }, + Shutdown, + ConnectionFailure, +} + +#[derive(Clone, Copy)] +pub enum Version { + V2, + V6, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum Modification { + ChangeFrom { + sender: String, + args: String, + }, + AddRcpt { + recipient: String, + args: String, + }, + DeleteRcpt { + recipient: String, + }, + ReplaceBody { + value: Vec, + }, + AddHeader { + name: String, + value: String, + }, + InsertHeader { + index: u32, + name: String, + value: String, + }, + ChangeHeader { + index: u32, + name: String, + value: String, + }, + Quarantine { + reason: String, + }, +} + +#[derive(Debug)] +pub struct Options { + pub version: u32, + pub actions: u32, + pub protocol: u32, +} + +#[derive(Default)] +pub struct Macros<'x> { + cmdcode: u8, + macros: Vec>, +} + +pub struct Macro<'x> { + name: &'x [u8], + value: Cow<'x, [u8]>, +} + +pub const SMFIF_NONE: u32 = 0x00000000; /* no flags */ +pub const SMFIF_ADDHDRS: u32 = 0x00000001; /* filter may add headers */ +pub const SMFIF_CHGBODY: u32 = 0x00000002; /* filter may replace body */ +pub const SMFIF_MODBODY: u32 = SMFIF_CHGBODY; /* backwards compatible */ +pub const SMFIF_ADDRCPT: u32 = 0x00000004; /* filter may add recipients */ +pub const SMFIF_DELRCPT: u32 = 0x00000008; /* filter may delete recipients */ +pub const SMFIF_CHGHDRS: u32 = 0x00000010; /* filter may change/delete headers */ +pub const SMFIF_QUARANTINE: u32 = 0x00000020; /* filter may quarantine envelope */ +pub const SMFIF_CHGFROM: u32 = 0x00000040; /* filter may change "from" (envelope sender) */ +pub const SMFIF_ADDRCPT_PAR: u32 = 0x00000080; /* add recipients incl. args */ +pub const SMFIF_SETSYMLIST: u32 = 0x00000100; /* filter can send set of symbols (macros) that it wants */ + +pub const SMFIP_NOCONNECT: u32 = 0x00000001; /* MTA should not send connect info */ +pub const SMFIP_NOHELO: u32 = 0x00000002; /* MTA should not send HELO info */ +pub const SMFIP_NOMAIL: u32 = 0x00000004; /* MTA should not send MAIL info */ +pub const SMFIP_NORCPT: u32 = 0x00000008; /* MTA should not send RCPT info */ +pub const SMFIP_NOBODY: u32 = 0x00000010; /* MTA should not send body */ +pub const SMFIP_NOHDRS: u32 = 0x00000020; /* MTA should not send headers */ +pub const SMFIP_NOEOH: u32 = 0x00000040; /* MTA should not send EOH */ +pub const SMFIP_NR_HDR: u32 = 0x00000080; /* No reply for headers */ +pub const SMFIP_NOHREPL: u32 = SMFIP_NR_HDR; /* No reply for headers */ +pub const SMFIP_NOUNKNOWN: u32 = 0x00000100; /* MTA should not send unknown commands */ +pub const SMFIP_NODATA: u32 = 0x00000200; /* MTA should not send DATA */ +pub const SMFIP_SKIP: u32 = 0x00000400; /* MTA understands SMFIS_SKIP */ +pub const SMFIP_RCPT_REJ: u32 = 0x00000800; /* MTA should also send rejected RCPTs */ +pub const SMFIP_NR_CONN: u32 = 0x00001000; /* No reply for connect */ +pub const SMFIP_NR_HELO: u32 = 0x00002000; /* No reply for HELO */ +pub const SMFIP_NR_MAIL: u32 = 0x00004000; /* No reply for MAIL */ +pub const SMFIP_NR_RCPT: u32 = 0x00008000; /* No reply for RCPT */ +pub const SMFIP_NR_DATA: u32 = 0x00010000; /* No reply for DATA */ +pub const SMFIP_NR_UNKN: u32 = 0x00020000; /* No reply for UNKN */ +pub const SMFIP_NR_EOH: u32 = 0x00040000; /* No reply for eoh */ +pub const SMFIP_NR_BODY: u32 = 0x00080000; /* No reply for body chunk */ +pub const SMFIP_HDR_LEADSPC: u32 = 0x00100000; /* header value leading space */ +pub const SMFIP_MDS_256K: u32 = 0x10000000; /* MILTER_MAX_DATA_SIZE=256K */ +pub const SMFIP_MDS_1M: u32 = 0x20000000; /* MILTER_MAX_DATA_SIZE=1M */ + +pub type Result = std::result::Result; + +impl Display for Command<'_> { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Command::Abort => write!(f, "ABORT"), + Command::Body { value } => write!(f, "BODY [{} bytes]", value.len()), + Command::EndOfBody => write!(f, "EOB"), + Command::Connect { + hostname, + port, + address, + } => write!( + f, + "CONNECT (host: {:?}, port: {}, address: {})", + std::str::from_utf8(hostname).unwrap_or_default(), + port, + address + ), + Command::Macro { macros } => { + write!(f, "MACRO (code: {}, params: ", macros.cmdcode)?; + for macro_ in ¯os.macros { + write!( + f, + "({:?}, {:?})", + std::str::from_utf8(macro_.name).unwrap_or_default(), + std::str::from_utf8(macro_.value.as_ref()).unwrap_or_default() + )?; + } + write!(f, ")") + } + Command::Header { name, value } => { + write!( + f, + "HEADER ({}: {:?})", + std::str::from_utf8(name).unwrap_or_default(), + std::str::from_utf8(value).unwrap_or_default() + ) + } + Command::EndOfHeader => write!(f, "EOH"), + Command::Helo { hostname } => write!( + f, + "HELO {:?}", + std::str::from_utf8(hostname).unwrap_or_default() + ), + Command::MailFrom { sender, args } => { + write!( + f, + "MAIL (from: {}, params: ", + std::str::from_utf8(sender).unwrap_or_default() + )?; + if let Some(args) = args { + for arg in args { + write!(f, " {}", std::str::from_utf8(arg).unwrap_or_default())?; + } + } + write!(f, ")") + } + Command::Rcpt { recipient, args } => { + write!( + f, + "RCPT (to: {}, params: ", + std::str::from_utf8(recipient).unwrap_or_default() + )?; + if let Some(args) = args { + for arg in args { + write!(f, " {}", std::str::from_utf8(arg).unwrap_or_default())?; + } + } + write!(f, ")") + } + Command::OptionNegotiation(opt) => write!(f, "OPTNEG ({})", opt), + Command::Quit => write!(f, "QUIT"), + Command::Data => write!(f, "DATA"), + Command::QuitNewConnection => write!(f, "QUIT_NC"), + } + } +} + +impl Display for Response { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Response::Action(action) => write!(f, "ACTION ({})", action), + Response::Modification(modification) => write!(f, "MODIFICATION ({})", modification), + Response::Progress => write!(f, "PROGRESS"), + Response::OptionNegotiation(opt) => write!(f, "OPTNEG ({})", opt), + Response::Skip => write!(f, "SKIP"), + Response::SetSymbols => write!(f, "SET_SYMBOLS"), + } + } +} + +impl Display for Action { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Action::Accept => write!(f, "ACCEPT"), + Action::Continue => write!(f, "CONTINUE"), + Action::Discard => write!(f, "DISCARD"), + Action::Reject => write!(f, "REJECT"), + Action::TempFail => write!(f, "TEMPFAIL"), + Action::ReplyCode { code, text } => { + write!(f, "REPLYCODE (code: {:?}, text: {})", code, text) + } + Action::Shutdown => write!(f, "SHUTDOWN"), + Action::ConnectionFailure => write!(f, "CONN_FAIL"), + } + } +} + +impl Display for Modification { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Modification::AddRcpt { recipient, args } => { + write!(f, "ADD_RCPT (recipient: {}, args: {})", recipient, args) + } + Modification::DeleteRcpt { recipient } => { + write!(f, "DEL_RCPT (recipient: {})", recipient) + } + Modification::ReplaceBody { value } => { + write!(f, "REPLACE_BODY ({} bytes)", value.len()) + } + Modification::AddHeader { name, value } => { + write!(f, "ADD_HEADER ({}: {})", name, value) + } + Modification::ChangeHeader { index, name, value } => { + write!(f, "CHANGE_HEADER (index: {}, {}: {})", index, name, value) + } + Modification::Quarantine { reason } => write!(f, "QUARANTINE ({})", reason), + Modification::ChangeFrom { sender, args } => { + write!(f, "CHANGE_FROM (<{}> {})", sender, args) + } + Modification::InsertHeader { index, name, value } => { + write!(f, "INSERT_HEADER (index: {}, {}: {})", index, name, value) + } + } + } +} + +impl Display for Options { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "version: {}, actions: [", self.version,)?; + + if self.actions & SMFIF_ADDHDRS != 0 { + write!(f, "ADDHDRS ")?; + } + if self.actions & SMFIF_CHGBODY != 0 { + write!(f, "CHGBODY ")?; + } + if self.actions & SMFIF_CHGHDRS != 0 { + write!(f, "CHGHDRS ")?; + } + if self.actions & SMFIF_ADDRCPT != 0 { + write!(f, "ADDRCPT ")?; + } + if self.actions & SMFIF_DELRCPT != 0 { + write!(f, "DELRCPT ")?; + } + if self.actions & SMFIF_CHGFROM != 0 { + write!(f, "CHGFROM ")?; + } + if self.actions & SMFIF_QUARANTINE != 0 { + write!(f, "QUARANTINE ")?; + } + if self.actions & SMFIF_CHGFROM != 0 { + write!(f, "CHGFROM ")?; + } + if self.actions & SMFIF_ADDRCPT_PAR != 0 { + write!(f, "ADDRCPT_PAR ")?; + } + if self.actions & SMFIF_SETSYMLIST != 0 { + write!(f, "SETSYMLIST ")?; + } + write!(f, "], options: [",)?; + + if self.protocol & SMFIP_NOCONNECT != 0 { + write!(f, "NOCONNECT ")?; + } + + if self.protocol & SMFIP_NOHELO != 0 { + write!(f, "NOHELO ")?; + } + + if self.protocol & SMFIP_NOMAIL != 0 { + write!(f, "NOMAIL ")?; + } + + if self.protocol & SMFIP_NORCPT != 0 { + write!(f, "NORCPT ")?; + } + + if self.protocol & SMFIP_NOBODY != 0 { + write!(f, "NOBODY ")?; + } + + if self.protocol & SMFIP_NOHDRS != 0 { + write!(f, "NOHDRS ")?; + } + + if self.protocol & SMFIP_NOEOH != 0 { + write!(f, "NOEOH ")?; + } + + if self.protocol & SMFIP_NR_HDR != 0 { + write!(f, "NR_HDR ")?; + } + + if self.protocol & SMFIP_NOUNKNOWN != 0 { + write!(f, "NOUNKNOWN ")?; + } + + if self.protocol & SMFIP_NODATA != 0 { + write!(f, "NODATA ")?; + } + + if self.protocol & SMFIP_SKIP != 0 { + write!(f, "SKIP ")?; + } + + if self.protocol & SMFIP_RCPT_REJ != 0 { + write!(f, "RCPT_REJ ")?; + } + + if self.protocol & SMFIP_NR_CONN != 0 { + write!(f, "NR_CONN ")?; + } + + if self.protocol & SMFIP_NR_HELO != 0 { + write!(f, "NR_HELO ")?; + } + + if self.protocol & SMFIP_NR_MAIL != 0 { + write!(f, "NR_MAIL ")?; + } + + if self.protocol & SMFIP_NR_RCPT != 0 { + write!(f, "NR_RCPT ")?; + } + + if self.protocol & SMFIP_NR_DATA != 0 { + write!(f, "NR_DATA ")?; + } + + if self.protocol & SMFIP_NR_UNKN != 0 { + write!(f, "NR_UNKN ")?; + } + + if self.protocol & SMFIP_NR_EOH != 0 { + write!(f, "NR_EOH ")?; + } + + if self.protocol & SMFIP_NR_BODY != 0 { + write!(f, "NR_BODY ")?; + } + + if self.protocol & SMFIP_HDR_LEADSPC != 0 { + write!(f, "HDR_LEADSPC ")?; + } + + if self.protocol & SMFIP_MDS_256K != 0 { + write!(f, "MDS_256K ")?; + } + + if self.protocol & SMFIP_MDS_1M != 0 { + write!(f, "MDS_1M ")?; + } + + write!(f, "]") + } +} + +impl Display for Error { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Error::Io(err) => write!(f, "IO error: {}", err), + Error::FrameTooLarge(size) => { + write!(f, "Milter response of {} bytes is too large.", size) + } + Error::FrameInvalid(frame) => write!( + f, + "Invalid milter response: {:?}", + frame.get(0..100).unwrap_or(frame.as_ref()) + ), + Error::Unexpected(response) => write!(f, "Unexpected response: {}", response), + Error::Timeout => write!(f, "Connection timed out"), + Error::TLSInvalidName => write!(f, "Invalid TLS name"), + Error::Disconnected => write!(f, "Disconnected unexpectedly"), + } + } +} diff --git a/crates/smtp/src/inbound/milter/protocol.rs b/crates/smtp/src/inbound/milter/protocol.rs new file mode 100644 index 00000000..1af6a4e4 --- /dev/null +++ b/crates/smtp/src/inbound/milter/protocol.rs @@ -0,0 +1,549 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::net::IpAddr; + +use crate::inbound::milter::Action; + +use super::{Command, Error, Modification, Options, Response}; + +pub const SMFIR_ADDRCPT: u8 = b'+'; /* add recipient */ +pub const SMFIR_DELRCPT: u8 = b'-'; /* remove recipient */ +pub const SMFIR_ADDRCPT_PAR: u8 = b'2'; /* add recipient (incl. ESMTP args) */ +pub const SMFIR_SHUTDOWN: u8 = b'4'; /* 421: shutdown (internal to MTA) */ +pub const SMFIR_ACCEPT: u8 = b'a'; /* accept */ +pub const SMFIR_REPLBODY: u8 = b'b'; /* replace body (chunk) */ +pub const SMFIR_CONTINUE: u8 = b'c'; /* continue */ +pub const SMFIR_DISCARD: u8 = b'd'; /* discard */ +pub const SMFIR_CHGFROM: u8 = b'e'; /* change envelope sender (from) */ +pub const SMFIR_CONN_FAIL: u8 = b'f'; /* cause a connection failure */ +pub const SMFIR_ADDHEADER: u8 = b'h'; /* add header */ +pub const SMFIR_INSHEADER: u8 = b'i'; /* insert header */ +pub const SMFIR_SETSYMLIST: u8 = b'l'; /* set list of symbols (macros) */ +pub const SMFIR_CHGHEADER: u8 = b'm'; /* change header */ +pub const SMFIR_PROGRESS: u8 = b'p'; /* progress */ +pub const SMFIR_QUARANTINE: u8 = b'q'; /* quarantine */ +pub const SMFIR_REJECT: u8 = b'r'; /* reject */ +pub const SMFIR_SKIP: u8 = b's'; /* skip */ +pub const SMFIR_TEMPFAIL: u8 = b't'; /* tempfail */ +pub const SMFIR_REPLYCODE: u8 = b'y'; /* reply code etc */ + +pub const SMFIC_ABORT: u8 = b'A'; /* Abort */ +pub const SMFIC_BODY: u8 = b'B'; /* Body chunk */ +pub const SMFIC_CONNECT: u8 = b'C'; /* Connection information */ +pub const SMFIC_MACRO: u8 = b'D'; /* Define macro */ +pub const SMFIC_BODYEOB: u8 = b'E'; /* final body chunk (End) */ +pub const SMFIC_HELO: u8 = b'H'; /* HELO/EHLO */ +pub const SMFIC_QUIT_NC: u8 = b'K'; /* QUIT but new connection follows */ +pub const SMFIC_HEADER: u8 = b'L'; /* Header */ +pub const SMFIC_MAIL: u8 = b'M'; /* MAIL from */ +pub const SMFIC_EOH: u8 = b'N'; /* EOH */ +pub const SMFIC_OPTNEG: u8 = b'O'; /* Option negotiation */ +pub const SMFIC_QUIT: u8 = b'Q'; /* QUIT */ +pub const SMFIC_RCPT: u8 = b'R'; /* RCPT to */ +pub const SMFIC_DATA: u8 = b'T'; /* DATA */ +pub const SMFIC_UNKNOWN: u8 = b'U'; /* Any unknown command */ + +impl<'x> Command<'x> { + fn build(command: u8, len: u32) -> Vec { + let mut buf = Vec::with_capacity(len as usize + 1 + std::mem::size_of::()); + buf.extend_from_slice((len + 1).to_be_bytes().as_ref()); + buf.push(command); + buf + } + + pub fn serialize(self) -> Vec { + match self { + Command::Abort => Command::build(SMFIC_ABORT, 0), + Command::Body { value } => { + let mut buf = Command::build(SMFIC_BODY, value.len() as u32); + buf.extend(value); + buf + } + Command::EndOfBody => Command::build(SMFIC_BODYEOB, 0), + Command::Connect { + hostname, + port, + address, + } => { + /* + + char hostname[] Hostname, NUL terminated + char family Protocol family (see below) + uint16 port Port number (SMFIA_INET or SMFIA_INET6 only) + char address[] IP address (ASCII) or unix socket path, NUL terminated + + */ + + let (address, family) = match address { + IpAddr::V4(address) => (address.to_string(), b'4'), + IpAddr::V6(address) => (address.to_string(), b'6'), + }; + + let mut buf = Command::build( + SMFIC_CONNECT, + hostname.len() as u32 // hostname + + 1 // NUL + + 1 // family + + std::mem::size_of::() as u32 // port + + address.len() as u32 // address + + 1, // NUL + ); + buf.extend(hostname); + buf.push(0x00); + buf.push(family); + buf.extend(port.to_be_bytes().as_ref()); + buf.extend(address.as_bytes()); + buf.push(0x00); + buf + } + Command::Macro { macros } => { + let mut buf = Command::build( + SMFIC_MACRO, + macros.macros.iter().fold(1, |acc, macro_| { + acc + macro_.name.len() as u32 + 1 + macro_.value.len() as u32 + 1 + }), + ); + buf.push(macros.cmdcode); + for macro_ in macros.macros { + buf.extend(macro_.name); + buf.push(0x00); + buf.extend(macro_.value.as_ref()); + buf.push(0x00); + } + buf + } + Command::Header { name, value } => { + let mut buf = + Command::build(SMFIC_HEADER, name.len() as u32 + 1 + value.len() as u32 + 1); + buf.extend(name); + buf.push(0x00); + buf.extend(value); + buf.push(0x00); + buf + } + Command::EndOfHeader => Command::build(SMFIC_EOH, 0), + Command::Helo { hostname } => { + let mut buf = Command::build(SMFIC_HELO, hostname.len() as u32 + 1); + buf.extend(hostname); + buf.push(0x00); + buf + } + Command::MailFrom { sender, args } => { + let mut buf = Command::build( + SMFIC_MAIL, + sender.len() as u32 // sender + + 1 // NUL + + args.as_ref().map_or(0, |args| args.iter().fold(0, |acc, arg| acc + arg.len() as u32) + 1), // args + ); + buf.extend(sender); + buf.push(0x00); + if let Some(args) = args { + for arg in args { + buf.extend(arg); + buf.push(0x00); + } + } + buf + } + Command::Rcpt { recipient, args } => { + let mut buf = Command::build( + SMFIC_RCPT, + recipient.len() as u32 // recipient + + 1 // NUL + + args.as_ref().map_or(0, |args| args.iter().fold(0, |acc, arg| acc + arg.len() as u32) + 1), // args + ); + buf.extend(recipient); + buf.push(0x00); + if let Some(args) = args { + for arg in args { + buf.extend(arg); + buf.push(0x00); + } + } + buf + } + Command::OptionNegotiation(opt) => { + let mut buf = Command::build(SMFIC_OPTNEG, 3 * std::mem::size_of::() as u32); + buf.extend(opt.version.to_be_bytes().as_ref()); + buf.extend(opt.actions.to_be_bytes().as_ref()); + buf.extend(opt.protocol.to_be_bytes().as_ref()); + buf + } + Command::Quit => Command::build(SMFIC_QUIT, 0), + + // Version 6 + Command::Data => Command::build(SMFIC_DATA, 0), + Command::QuitNewConnection => Command::build(SMFIC_QUIT_NC, 0), + } + } + + #[cfg(feature = "test_mode")] + pub fn deserialize(bytes: &'x [u8]) -> Command<'x> { + let mut reader = PacketReader::new(bytes); + match reader.byte() { + SMFIC_ABORT => Command::Abort, + SMFIC_BODY => Command::Body { value: &bytes[1..] }, + SMFIC_BODYEOB => Command::EndOfBody, + SMFIC_CONNECT => { + let hostname = reader.read_nul_terminated().unwrap(); + let family = reader.byte(); + let port = reader.read_u16(); + let address = std::str::from_utf8(reader.read_nul_terminated().unwrap()).unwrap(); + Command::Connect { + hostname, + port, + address: match family { + b'4' => IpAddr::V4(address.parse().unwrap()), + b'6' => IpAddr::V6(address.parse().unwrap()), + _ => unreachable!(), + }, + } + } + SMFIC_MACRO => { + let cmdcode = reader.byte(); + let mut macros = Vec::new(); + while let Some(name) = reader.read_nul_terminated() { + let value = reader.read_nul_terminated().unwrap(); + macros.push(super::Macro { + name, + value: value.into(), + }); + } + Command::Macro { + macros: super::Macros { cmdcode, macros }, + } + } + SMFIC_HEADER => { + let name = reader.read_nul_terminated().unwrap(); + let value = reader.read_nul_terminated().unwrap(); + Command::Header { name, value } + } + SMFIC_EOH => Command::EndOfHeader, + SMFIC_HELO => { + let hostname = reader.read_nul_terminated().unwrap(); + Command::Helo { hostname } + } + SMFIC_MAIL => { + let sender = reader.read_nul_terminated().unwrap(); + let mut args = Vec::new(); + while let Some(arg) = reader.read_nul_terminated() { + args.push(arg); + } + Command::MailFrom { + sender, + args: Some(args), + } + } + SMFIC_RCPT => { + let recipient = reader.read_nul_terminated().unwrap(); + let mut args = Vec::new(); + while let Some(arg) = reader.read_nul_terminated() { + args.push(arg); + } + Command::Rcpt { + recipient, + args: Some(args), + } + } + SMFIC_OPTNEG => Command::OptionNegotiation(super::Options { + version: reader.read_u32(), + actions: reader.read_u32(), + protocol: reader.read_u32(), + }), + SMFIC_QUIT => Command::Quit, + SMFIC_DATA => Command::Data, + SMFIC_QUIT_NC => Command::QuitNewConnection, + c => panic!("Unknown command: {}", char::from(c)), + } + } +} + +impl Response { + pub fn deserialize(bytes: &[u8]) -> Option { + let frame_len = bytes.len().saturating_sub(1); + let mut bytes = bytes.iter(); + match *bytes.next()? { + SMFIR_ADDRCPT => Response::Modification(Modification::AddRcpt { + recipient: read_nul_terminated(&mut bytes, frame_len)?, + args: String::new(), + }), + SMFIR_DELRCPT => Response::Modification(Modification::DeleteRcpt { + recipient: read_nul_terminated(&mut bytes, frame_len)?, + }), + SMFIR_ACCEPT => Response::Action(Action::Accept), + SMFIR_REPLBODY => { + let mut body = Vec::with_capacity(frame_len); + body.extend(bytes); + Response::Modification(Modification::ReplaceBody { value: body }) + } + SMFIR_CONTINUE => Response::Action(Action::Continue), + SMFIR_DISCARD => Response::Action(Action::Discard), + SMFIR_ADDHEADER => Response::Modification(Modification::AddHeader { + name: read_nul_terminated(&mut bytes, 16)?, + value: read_nul_terminated(&mut bytes, frame_len)?, + }), + SMFIR_CHGHEADER => Response::Modification(Modification::ChangeHeader { + index: read_u32(&mut bytes)?, + name: read_nul_terminated(&mut bytes, 16)?, + value: read_nul_terminated(&mut bytes, frame_len)?, + }), + SMFIR_PROGRESS => Response::Progress, + SMFIR_QUARANTINE => Response::Modification(Modification::Quarantine { + reason: read_nul_terminated(&mut bytes, frame_len)?, + }), + SMFIR_REJECT => Response::Action(Action::Reject), + SMFIR_TEMPFAIL => Response::Action(Action::TempFail), + SMFIR_REPLYCODE => { + let code = [*bytes.next()?, *bytes.next()?, *bytes.next()?]; + bytes.next()?; // Space + Response::Action(Action::ReplyCode { + code, + text: read_nul_terminated(&mut bytes, frame_len)?, + }) + } + SMFIC_OPTNEG => Response::OptionNegotiation(Options { + version: read_u32(&mut bytes)?, + actions: read_u32(&mut bytes)?, + protocol: read_u32(&mut bytes)?, + }), + + // V6 + SMFIR_ADDRCPT_PAR => Response::Modification(Modification::AddRcpt { + recipient: read_nul_terminated(&mut bytes, frame_len)?, + args: read_nul_terminated(&mut bytes, frame_len)?, + }), + SMFIR_CHGFROM => Response::Modification(Modification::ChangeFrom { + sender: read_nul_terminated(&mut bytes, frame_len)?, + args: read_nul_terminated(&mut bytes, frame_len)?, + }), + SMFIR_SKIP => Response::Skip, + SMFIR_SETSYMLIST => Response::SetSymbols, + SMFIR_SHUTDOWN => Response::Action(Action::Shutdown), + SMFIR_CONN_FAIL => Response::Action(Action::ConnectionFailure), + SMFIR_INSHEADER => Response::Modification(Modification::InsertHeader { + index: read_u32(&mut bytes)?, + name: read_nul_terminated(&mut bytes, 16)?, + value: read_nul_terminated(&mut bytes, frame_len)?, + }), + _ => return None, + } + .into() + } + + pub fn can_continue(&self) -> bool { + matches!( + self, + Response::Progress | Response::Action(Action::Accept | Action::Continue) + ) + } + + pub fn into_action(self) -> super::Result { + match self { + Response::Action(action) => Ok(action), + response => Err(Error::Unexpected(response)), + } + } + + #[cfg(feature = "test_mode")] + pub fn serialize(&self) -> Vec { + match self { + Response::Action(action) => match action { + Action::Accept => Command::build(SMFIR_ACCEPT, 0), + Action::Continue => Command::build(SMFIR_CONTINUE, 0), + Action::Discard => Command::build(SMFIR_DISCARD, 0), + Action::Reject => Command::build(SMFIR_REJECT, 0), + Action::TempFail => Command::build(SMFIR_TEMPFAIL, 0), + Action::ReplyCode { code, text } => { + let mut buf = Command::build(SMFIR_REPLYCODE, text.len() as u32 + 4 + 1); + buf.extend(code); + buf.push(b' '); + buf.extend(text.as_bytes()); + buf.push(0x00); + buf + } + Action::Shutdown => Command::build(SMFIR_SHUTDOWN, 0), + Action::ConnectionFailure => Command::build(SMFIR_CONN_FAIL, 0), + }, + Response::Modification(modif) => match modif { + Modification::ChangeFrom { sender, args } => { + let mut buf = + Command::build(SMFIR_CHGFROM, sender.len() as u32 + args.len() as u32 + 2); + buf.extend(sender.as_bytes()); + buf.push(0x00); + buf.extend(args.as_bytes()); + buf.push(0x00); + buf + } + Modification::AddRcpt { recipient, args } => { + let mut buf = Command::build( + SMFIR_ADDRCPT_PAR, + recipient.len() as u32 + args.len() as u32 + 2, + ); + buf.extend(recipient.as_bytes()); + buf.push(0x00); + buf.extend(args.as_bytes()); + buf.push(0x00); + buf + } + Modification::DeleteRcpt { recipient } => { + let mut buf = Command::build(SMFIR_DELRCPT, recipient.len() as u32 + 1); + buf.extend(recipient.as_bytes()); + buf.push(0x00); + buf + } + Modification::ReplaceBody { value } => { + let mut buf = Command::build(SMFIR_REPLBODY, value.len() as u32 + 1); + buf.extend(value); + buf + } + Modification::AddHeader { name, value } => { + let mut buf = + Command::build(SMFIR_ADDHEADER, name.len() as u32 + value.len() as u32 + 2); + buf.extend(name.as_bytes()); + buf.push(0x00); + buf.extend(value.as_bytes()); + buf.push(0x00); + buf + } + Modification::InsertHeader { index, name, value } => { + let mut buf = Command::build( + SMFIR_INSHEADER, + name.len() as u32 + + value.len() as u32 + + std::mem::size_of::() as u32 + + 2, + ); + buf.extend(index.to_be_bytes().as_ref()); + buf.extend(name.as_bytes()); + buf.push(0x00); + buf.extend(value.as_bytes()); + buf.push(0x00); + buf + } + Modification::ChangeHeader { index, name, value } => { + let mut buf = Command::build( + SMFIR_CHGHEADER, + name.len() as u32 + + value.len() as u32 + + std::mem::size_of::() as u32 + + 2, + ); + buf.extend(index.to_be_bytes().as_ref()); + buf.extend(name.as_bytes()); + buf.push(0x00); + buf.extend(value.as_bytes()); + buf.push(0x00); + buf + } + Modification::Quarantine { reason } => { + let mut buf = Command::build(SMFIR_QUARANTINE, reason.len() as u32 + 1); + buf.extend(reason.as_bytes()); + buf.push(0x00); + buf + } + }, + Response::Progress => Command::build(SMFIR_PROGRESS, 0), + Response::Skip => Command::build(SMFIR_SKIP, 0), + Response::SetSymbols => Command::build(SMFIR_SETSYMLIST, 0), + Response::OptionNegotiation(opt) => { + let mut buf = Command::build(SMFIC_OPTNEG, 3 * std::mem::size_of::() as u32); + buf.extend(opt.version.to_be_bytes().as_ref()); + buf.extend(opt.actions.to_be_bytes().as_ref()); + buf.extend(opt.protocol.to_be_bytes().as_ref()); + buf + } + } + } +} + +fn read_nul_terminated(bytes: &mut std::slice::Iter, expected_len: usize) -> Option { + let mut buf = Vec::with_capacity(expected_len); + loop { + match bytes.next()? { + 0x00 => break, + byte => buf.push(*byte), + } + } + String::from_utf8(buf).ok() +} + +fn read_u32(bytes: &mut std::slice::Iter) -> Option { + let mut buf = [0u8; 4]; + for byte in buf.iter_mut() { + *byte = *bytes.next()?; + } + Some(u32::from_be_bytes(buf)) +} + +#[cfg(feature = "test_mode")] +pub struct PacketReader<'x> { + bytes: &'x [u8], + iter: std::iter::Enumerate>, +} + +#[cfg(feature = "test_mode")] +impl<'x> PacketReader<'x> { + pub fn new(bytes: &'x [u8]) -> PacketReader<'x> { + Self { + bytes, + iter: bytes.iter().enumerate(), + } + } + + pub fn byte(&mut self) -> u8 { + *self.iter.next().unwrap().1 + } + + pub fn read_nul_terminated(&mut self) -> Option<&'x [u8]> { + let (start_pos, ch) = self.iter.next()?; + let mut end_pos = start_pos; + + if *ch != 0x00 { + loop { + match self.iter.next().unwrap().1 { + 0x00 => break, + _ => end_pos += 1, + } + } + } + + Some(&self.bytes[start_pos..end_pos + 1]) + } + + pub fn read_u32(&mut self) -> u32 { + let mut buf = [0u8; 4]; + for byte in buf.iter_mut() { + *byte = self.byte(); + } + u32::from_be_bytes(buf) + } + + pub fn read_u16(&mut self) -> u16 { + let mut buf = [0u8; 2]; + for byte in buf.iter_mut() { + *byte = self.byte(); + } + u16::from_be_bytes(buf) + } +} diff --git a/crates/smtp/src/inbound/milter/receiver.rs b/crates/smtp/src/inbound/milter/receiver.rs new file mode 100644 index 00000000..c3a48804 --- /dev/null +++ b/crates/smtp/src/inbound/milter/receiver.rs @@ -0,0 +1,127 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::borrow::Cow; + +enum State { + Len { + buf: [u8; std::mem::size_of::()], + bytes_read: usize, + }, + Frame { + buf: Vec, + frame_len: usize, + }, +} + +pub struct Receiver { + packet_pos: usize, + state: State, + max_frame_len: usize, +} + +pub enum FrameResult<'x> { + Frame(Cow<'x, [u8]>), + Incomplete, + TooLarge(usize), +} + +impl Default for State { + fn default() -> Self { + State::Len { + buf: [0; std::mem::size_of::()], + bytes_read: 0, + } + } +} + +impl Receiver { + pub fn with_max_frame_len(max_frame_len: usize) -> Self { + Receiver { + packet_pos: 0, + state: State::default(), + max_frame_len, + } + } + + pub fn read_frame<'x>(&mut self, packet: &'x [u8]) -> FrameResult<'x> { + if !packet.is_empty() { + match &mut self.state { + State::Len { buf, bytes_read } => { + while *bytes_read < std::mem::size_of::() { + if let Some(byte) = packet.get(self.packet_pos) { + buf[*bytes_read] = *byte; + *bytes_read += 1; + self.packet_pos += 1; + } else { + self.packet_pos = 0; + return FrameResult::Incomplete; + } + } + let length = u32::from_be_bytes(*buf) as usize; + if length <= self.max_frame_len { + if let Some(frame) = packet.get(self.packet_pos..self.packet_pos + length) { + self.packet_pos += length; + self.state = State::default(); + FrameResult::Frame(frame.into()) + } else { + let mut buf = Vec::with_capacity(length); + if let Some(bytes_available) = packet.get(self.packet_pos..) { + buf.extend(bytes_available); + } + self.state = State::Frame { + buf, + frame_len: length, + }; + self.packet_pos = 0; + FrameResult::Incomplete + } + } else { + FrameResult::TooLarge(length) + } + } + State::Frame { buf, frame_len } => { + let bytes_pending = *frame_len - buf.len(); + if let Some(bytes) = + packet.get(self.packet_pos..self.packet_pos + bytes_pending) + { + let mut buf = std::mem::take(buf); + buf.extend(bytes); + self.packet_pos += bytes_pending; + self.state = State::default(); + FrameResult::Frame(buf.into()) + } else if let Some(bytes_available) = packet.get(self.packet_pos..) { + buf.extend(bytes_available); + self.packet_pos = 0; + FrameResult::Incomplete + } else { + self.packet_pos = 0; + FrameResult::Incomplete + } + } + } + } else { + FrameResult::Incomplete + } + } +} diff --git a/crates/smtp/src/inbound/mod.rs b/crates/smtp/src/inbound/mod.rs index 27b3e4c2..0cefd584 100644 --- a/crates/smtp/src/inbound/mod.rs +++ b/crates/smtp/src/inbound/mod.rs @@ -33,6 +33,7 @@ pub mod auth; pub mod data; pub mod ehlo; pub mod mail; +pub mod milter; pub mod rcpt; pub mod session; pub mod spawn; @@ -41,13 +42,19 @@ pub mod vrfy; pub trait IsTls { fn is_tls(&self) -> bool; fn write_tls_header(&self, headers: &mut Vec); + fn tls_version_and_cipher(&self) -> (&'static str, &'static str); } impl IsTls for TcpStream { fn is_tls(&self) -> bool { false } + fn write_tls_header(&self, _headers: &mut Vec) {} + + fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { + ("", "") + } } impl IsTls for TlsStream { @@ -55,10 +62,10 @@ impl IsTls for TlsStream { true } - fn write_tls_header(&self, headers: &mut Vec) { + fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { let (_, conn) = self.get_ref(); - headers.extend_from_slice(b"(using "); - headers.extend_from_slice( + + ( match conn .protocol_version() .unwrap_or(rustls::ProtocolVersion::Unknown(0)) @@ -73,11 +80,7 @@ impl IsTls for TlsStream { rustls::ProtocolVersion::DTLSv1_2 => "DTLSv1.2", rustls::ProtocolVersion::DTLSv1_3 => "DTLSv1.3", _ => "unknown", - } - .as_bytes(), - ); - headers.extend_from_slice(b" with cipher "); - headers.extend_from_slice( + }, match conn.negotiated_cipher_suite() { Some(rustls::SupportedCipherSuite::Tls13(cs)) => { cs.common.suite.as_str().unwrap_or("unknown") @@ -86,9 +89,16 @@ impl IsTls for TlsStream { cs.common.suite.as_str().unwrap_or("unknown") } None => "unknown", - } - .as_bytes(), - ); + }, + ) + } + + fn write_tls_header(&self, headers: &mut Vec) { + let (version, cipher) = self.tls_version_and_cipher(); + headers.extend_from_slice(b"(using "); + headers.extend_from_slice(version.as_bytes()); + headers.extend_from_slice(b" with cipher "); + headers.extend_from_slice(cipher.as_bytes()); headers.extend_from_slice(b")\r\n\t"); } } diff --git a/crates/smtp/src/inbound/session.rs b/crates/smtp/src/inbound/session.rs index 5b1c8bae..b233c5bf 100644 --- a/crates/smtp/src/inbound/session.rs +++ b/crates/smtp/src/inbound/session.rs @@ -228,15 +228,20 @@ impl Session { if receiver.ingest(&mut iter, &mut self.data.message) { let num_rcpts = self.data.rcpt_to.len(); let message = self.queue_message().await; - if self.instance.protocol == ServerProtocol::Smtp { - self.write(message.as_ref()).await?; - } else { - for _ in 0..num_rcpts { + if !message.is_empty() { + if self.instance.protocol == ServerProtocol::Smtp { self.write(message.as_ref()).await?; + } else { + for _ in 0..num_rcpts { + self.write(message.as_ref()).await?; + } } + self.reset(); + state = State::default(); + } else { + // Disconnect requested + return Err(()); } - self.reset(); - state = State::default(); } else { break 'outer; } @@ -250,14 +255,19 @@ impl Session { if receiver.is_last { let num_rcpts = self.data.rcpt_to.len(); let message = self.queue_message().await; - if self.instance.protocol == ServerProtocol::Smtp { - self.write(message.as_ref()).await?; - } else { - for _ in 0..num_rcpts { + if !message.is_empty() { + if self.instance.protocol == ServerProtocol::Smtp { self.write(message.as_ref()).await?; + } else { + for _ in 0..num_rcpts { + self.write(message.as_ref()).await?; + } } + self.reset(); + } else { + // Disconnect requested + return Err(()); } - self.reset(); } else { self.write(b"250 2.6.0 Chunk accepted.\r\n").await?; } diff --git a/crates/smtp/src/inbound/spawn.rs b/crates/smtp/src/inbound/spawn.rs index 1dfb7ea3..265b97b9 100644 --- a/crates/smtp/src/inbound/spawn.rs +++ b/crates/smtp/src/inbound/spawn.rs @@ -49,7 +49,7 @@ impl SessionManager for SmtpSessionManager { span: session.span, stream: session.stream, in_flight: vec![session.in_flight], - data: SessionData::new(session.local_ip, session.remote_ip), + data: SessionData::new(session.local_ip, session.remote_ip, session.remote_port), params: SessionParameters::default(), }; diff --git a/crates/smtp/src/lib.rs b/crates/smtp/src/lib.rs index 3ce4f9c9..6f0f14ad 100644 --- a/crates/smtp/src/lib.rs +++ b/crates/smtp/src/lib.rs @@ -49,6 +49,7 @@ pub mod queue; pub mod reporting; pub static USER_AGENT: &str = concat!("StalwartSMTP/", env!("CARGO_PKG_VERSION"),); +pub static DAEMON_NAME: &str = concat!("Stalwart SMTP v", env!("CARGO_PKG_VERSION"),); impl SMTP { pub async fn init( diff --git a/crates/smtp/src/outbound/dane/verify.rs b/crates/smtp/src/outbound/dane/verify.rs index 0764ac3d..7487283e 100644 --- a/crates/smtp/src/outbound/dane/verify.rs +++ b/crates/smtp/src/outbound/dane/verify.rs @@ -158,150 +158,3 @@ impl Tlsa { } } } - -#[cfg(test)] -mod test { - use std::{ - collections::BTreeSet, - fs::{self, File}, - io::{BufRead, BufReader}, - num::ParseIntError, - path::PathBuf, - time::{Duration, Instant}, - }; - - use mail_auth::{ - common::lru::{DnsCache, LruCache}, - trust_dns_resolver::{ - config::{ResolverConfig, ResolverOpts}, - AsyncResolver, - }, - Resolver, - }; - use rustls::Certificate; - - use crate::{ - core::Resolvers, - outbound::dane::{DnssecResolver, Tlsa, TlsaEntry}, - queue::{Error, ErrorDetails, Status}, - }; - - #[tokio::test] - async fn dane_test() { - let conf = ResolverConfig::cloudflare_tls(); - let mut opts = ResolverOpts::default(); - opts.validate = true; - opts.try_tcp_on_error = true; - - let r = Resolvers { - dns: Resolver::new_cloudflare().unwrap(), - dnssec: DnssecResolver { - resolver: AsyncResolver::tokio(conf, opts).unwrap(), - }, - cache: crate::core::DnsCache { - tlsa: LruCache::with_capacity(10), - mta_sts: LruCache::with_capacity(10), - }, - }; - - // Add dns entries - let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); - path.push("resources"); - path.push("smtp"); - path.push("dane"); - let mut file = path.clone(); - file.push("dns.txt"); - - let mut hosts = BTreeSet::new(); - let mut tlsa = Tlsa { - entries: Vec::new(), - has_end_entities: false, - has_intermediates: false, - }; - let mut hostname = String::new(); - - for line in BufReader::new(File::open(file).unwrap()).lines() { - let line = line.unwrap(); - let mut is_end_entity = false; - for (pos, item) in line.split_whitespace().enumerate() { - match pos { - 0 => { - if hostname != item && !hostname.is_empty() { - r.tlsa_add(hostname, tlsa, Instant::now() + Duration::from_secs(30)); - tlsa = Tlsa { - entries: Vec::new(), - has_end_entities: false, - has_intermediates: false, - }; - } - hosts.insert(item.strip_prefix("_25._tcp.").unwrap().to_string()); - hostname = item.to_string(); - } - 1 => { - is_end_entity = item == "3"; - } - 4 => { - if is_end_entity { - tlsa.has_end_entities = true; - } else { - tlsa.has_intermediates = true; - } - tlsa.entries.push(TlsaEntry { - is_end_entity, - is_sha256: true, - is_spki: true, - data: decode_hex(item).unwrap(), - }); - } - _ => (), - } - } - } - r.tlsa_add(hostname, tlsa, Instant::now() + Duration::from_secs(30)); - - // Add certificates - assert!(!hosts.is_empty()); - for host in hosts { - // Add certificates - let mut certs = Vec::new(); - for num in 0..6 { - let mut file = path.clone(); - file.push(format!("{host}.{num}.cert")); - if file.exists() { - certs.push(Certificate(fs::read(file).unwrap())); - } else { - break; - } - } - - // Successful DANE verification - let tlsa = r - .tlsa_lookup(format!("_25._tcp.{host}.")) - .await - .unwrap() - .unwrap(); - - assert_eq!( - tlsa.verify(&tracing::info_span!("test_span"), &host, Some(&certs)), - Ok(()) - ); - - // Failed DANE verification - certs.remove(0); - assert_eq!( - tlsa.verify(&tracing::info_span!("test_span"), &host, Some(&certs)), - Err(Status::PermanentFailure(Error::DaneError(ErrorDetails { - entity: host.to_string(), - details: "No matching certificates found in TLSA records".to_string() - }))) - ); - } - } - - pub fn decode_hex(s: &str) -> Result, ParseIntError> { - (0..s.len()) - .step_by(2) - .map(|i| u8::from_str_radix(&s[i..i + 2], 16)) - .collect() - } -} diff --git a/crates/utils/src/listener/listen.rs b/crates/utils/src/listener/listen.rs index c93f26f8..80210633 100644 --- a/crates/utils/src/listener/listen.rs +++ b/crates/utils/src/listener/listen.rs @@ -98,6 +98,7 @@ impl Server { stream, local_ip, remote_ip: remote_addr.ip(), + remote_port: remote_addr.port(), span, in_flight, instance: instance.clone(), diff --git a/crates/utils/src/listener/mod.rs b/crates/utils/src/listener/mod.rs index f0ecdddb..b65dbde0 100644 --- a/crates/utils/src/listener/mod.rs +++ b/crates/utils/src/listener/mod.rs @@ -53,6 +53,7 @@ pub struct SessionData { pub stream: T, pub local_ip: IpAddr, pub remote_ip: IpAddr, + pub remote_port: u16, pub span: tracing::Span, pub in_flight: InFlight, pub instance: Arc, diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 2dbd4e5b..04b2ef7c 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -20,7 +20,7 @@ imap_proto = { path = "../crates/imap-proto" } smtp = { path = "../crates/smtp", features = ["test_mode", "local_delivery"] } managesieve = { path = "../crates/managesieve", features = ["test_mode"] } smtp-proto = { git = "https://github.com/stalwartlabs/smtp-proto" } -mail-send = { git = "https://github.com/stalwartlabs/mail-send" } +mail-send = { git = "https://github.com/stalwartlabs/mail-send", default-features = false, features = ["cram-md5", "skip-ehlo"] } mail-auth = { git = "https://github.com/stalwartlabs/mail-auth", features = ["test"] } sieve-rs = { git = "https://github.com/stalwartlabs/sieve" } utils = { path = "../crates/utils", features = ["test_mode"] } diff --git a/tests/resources/smtp/config/rules-eval.toml b/tests/resources/smtp/config/rules-eval.toml index 147464a1..cce78724 100644 --- a/tests/resources/smtp/config/rules-eval.toml +++ b/tests/resources/smtp/config/rules-eval.toml @@ -164,5 +164,8 @@ nested-none-of-false = { none-of = [ ]} ]} -[list] +[directory."list"] +type = "memory" + +[directory."list".lookup] domains = ["mydomain1.org", "foo.net", "otherdomain.net"] diff --git a/tests/resources/smtp/milter/message.eml b/tests/resources/smtp/milter/message.eml new file mode 100644 index 00000000..834aec87 --- /dev/null +++ b/tests/resources/smtp/milter/message.eml @@ -0,0 +1,11 @@ +From: John Doe +To: Mary Smith +References: a +References: b +X-Mailer: Test +X-1: 1 +X-2: 2 +X-3: 3 +Subject: Saying Hello + +This is a message just to say hello. diff --git a/tests/resources/smtp/milter/message.json b/tests/resources/smtp/milter/message.json new file mode 100644 index 00000000..03a7acc0 --- /dev/null +++ b/tests/resources/smtp/milter/message.json @@ -0,0 +1,149 @@ +[ + { + "modifications": [ + { + "AddHeader": { + "name": "X-Hello", + "value": "World" + } + }, + { + "AddHeader": { + "name": "X-CR", + "value": "LF\r\n" + } + } + ], + "result": "X-Hello: World\r\nX-CR: LF\r\nFrom: John Doe \r\nTo: Mary Smith \r\nReferences: a\r\nReferences: b\r\nX-Mailer: Test\r\nX-1: 1\r\nX-2: 2\r\nX-3: 3\r\nSubject: Saying Hello\r\n\r\nThis is a message just to say hello.\r\n" + }, + { + "modifications": [ + { + "ReplaceBody": { + "value": [ + 49, + 50, + 51 + ] + } + } + ], + "result": "From: John Doe \r\nTo: Mary Smith \r\nReferences: a\r\nReferences: b\r\nX-Mailer: Test\r\nX-1: 1\r\nX-2: 2\r\nX-3: 3\r\nSubject: Saying Hello\r\n\r\n123" + }, + { + "modifications": [ + { + "AddHeader": { + "name": "X-Spam", + "value": "Yes" + } + }, + { + "ReplaceBody": { + "value": [ + 49, + 50, + 51 + ] + } + }, + { + "ReplaceBody": { + "value": [ + 52, + 53, + 54 + ] + } + } + ], + "result": "X-Spam: Yes\r\nFrom: John Doe \r\nTo: Mary Smith \r\nReferences: a\r\nReferences: b\r\nX-Mailer: Test\r\nX-1: 1\r\nX-2: 2\r\nX-3: 3\r\nSubject: Saying Hello\r\n\r\n123456" + }, + { + "modifications": [ + { + "ChangeHeader": { + "index": 1, + "name": "References", + "value": "" + } + }, + { + "ChangeHeader": { + "index": 1, + "name": "References", + "value": "z" + } + }, + { + "ChangeHeader": { + "index": 1, + "name": "Subject", + "value": "[SPAM] Saying Hello" + } + } + ], + "result": "From: John Doe \r\nTo: Mary Smith \r\nReferences: z\r\nX-Mailer: Test\r\nX-1: 1\r\nX-2: 2\r\nX-3: 3\r\nSubject: [SPAM] Saying Hello\r\n\r\nThis is a message just to say hello.\r\n" + }, + { + "modifications": [ + { + "ChangeHeader": { + "index": 1, + "name": "X-Some-Header", + "value": "Some Value" + } + }, + { + "InsertHeader": { + "index": 2, + "name": "References", + "value": "" + } + }, + { + "InsertHeader": { + "index": 10, + "name": "X-3", + "value": "z" + } + }, + { + "ReplaceBody": { + "value": [ + 52, + 53, + 54 + ] + } + }, + { + "ReplaceBody": { + "value": [ + 49, + 50, + 51 + ] + } + } + ], + "result": "X-Some-Header: Some Value\r\nFrom: John Doe \r\nTo: Mary Smith \r\nReferences: a\r\nReferences: \r\nReferences: b\r\nX-Mailer: Test\r\nX-1: 1\r\nX-2: 2\r\nX-3: z\r\nX-3: 3\r\nSubject: Saying Hello\r\n\r\n456123" + }, + { + "modifications": [ + { + "Quarantine": { + "reason": "Virus found!" + } + }, + { + "InsertHeader": { + "index": 1, + "name": "References", + "value": "" + } + } + ], + "result": "X-Quarantine: Virus found!\r\nFrom: John Doe \r\nTo: Mary Smith \r\nReferences: \r\nReferences: a\r\nReferences: b\r\nX-Mailer: Test\r\nX-1: 1\r\nX-2: 2\r\nX-3: 3\r\nSubject: Saying Hello\r\n\r\nThis is a message just to say hello.\r\n" + } +] \ No newline at end of file diff --git a/tests/src/smtp/config.rs b/tests/src/smtp/config.rs index de790490..46c029c9 100644 --- a/tests/src/smtp/config.rs +++ b/tests/src/smtp/config.rs @@ -21,19 +21,22 @@ * for more details. */ -use std::{fs, path::PathBuf, sync::Arc, time::Duration}; +use std::{fs, net::IpAddr, path::PathBuf, sync::Arc, time::Duration}; use tokio::net::TcpSocket; use utils::config::{Config, Listener, Rate, Server, ServerProtocol}; use ahash::{AHashMap, AHashSet}; -use directory::Lookup; +use directory::{config::ConfigDirectory, Lookup}; -use smtp::config::{ - condition::ConfigCondition, if_block::ConfigIf, throttle::ConfigThrottle, Condition, - ConditionMatch, Conditions, ConfigContext, EnvelopeKey, IfBlock, IfThen, IpAddrMask, - StringMatch, Throttle, THROTTLE_AUTH_AS, THROTTLE_REMOTE_IP, THROTTLE_SENDER_DOMAIN, +use smtp::{ + config::{ + condition::ConfigCondition, if_block::ConfigIf, throttle::ConfigThrottle, Condition, + ConditionMatch, Conditions, ConfigContext, EnvelopeKey, IfBlock, IfThen, IpAddrMask, + StringMatch, Throttle, THROTTLE_AUTH_AS, THROTTLE_REMOTE_IP, THROTTLE_SENDER_DOMAIN, + }, + core::Envelope, }; use super::add_test_certs; @@ -516,3 +519,123 @@ fn parse_servers() { } } } + +struct TestEnvelope { + pub local_ip: IpAddr, + pub remote_ip: IpAddr, + pub sender_domain: String, + pub sender: String, + pub rcpt_domain: String, + pub rcpt: String, + pub helo_domain: String, + pub authenticated_as: String, + pub mx: String, + pub listener_id: u16, + pub priority: i16, +} + +impl Envelope for TestEnvelope { + fn local_ip(&self) -> IpAddr { + self.local_ip + } + + fn remote_ip(&self) -> IpAddr { + self.remote_ip + } + + fn sender_domain(&self) -> &str { + self.sender_domain.as_str() + } + + fn sender(&self) -> &str { + self.sender.as_str() + } + + fn rcpt_domain(&self) -> &str { + self.rcpt_domain.as_str() + } + + fn rcpt(&self) -> &str { + self.rcpt.as_str() + } + + fn helo_domain(&self) -> &str { + self.helo_domain.as_str() + } + + fn authenticated_as(&self) -> &str { + self.authenticated_as.as_str() + } + + fn mx(&self) -> &str { + self.mx.as_str() + } + + fn listener_id(&self) -> u16 { + self.listener_id + } + + fn priority(&self) -> i16 { + self.priority + } +} + +#[tokio::test] +async fn eval_if() { + let mut file = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + file.push("resources"); + file.push("smtp"); + file.push("config"); + file.push("rules-eval.toml"); + + let config = Config::parse(&fs::read_to_string(file).unwrap()).unwrap(); + let servers = vec![ + Server { + id: "smtp".to_string(), + internal_id: 123, + ..Default::default() + }, + Server { + id: "smtps".to_string(), + internal_id: 456, + ..Default::default() + }, + ]; + let mut context = ConfigContext::new(&servers); + context.directory = config.parse_directory().unwrap(); + let conditions = config.parse_conditions(&context).unwrap(); + + let envelope = TestEnvelope { + local_ip: config.property_require("envelope.local-ip").unwrap(), + remote_ip: config.property_require("envelope.remote-ip").unwrap(), + sender_domain: config.property_require("envelope.sender-domain").unwrap(), + sender: config.property_require("envelope.sender").unwrap(), + rcpt_domain: config.property_require("envelope.rcpt-domain").unwrap(), + rcpt: config.property_require("envelope.rcpt").unwrap(), + authenticated_as: config + .property_require("envelope.authenticated-as") + .unwrap(), + mx: config.property_require("envelope.mx").unwrap(), + listener_id: config.property_require("envelope.listener").unwrap(), + priority: config.property_require("envelope.priority").unwrap(), + helo_domain: config.property_require("envelope.helo-domain").unwrap(), + }; + + for (key, conditions) in conditions { + //println!("============= Testing {:?} ==================", key); + let (_, expected_result) = key.rsplit_once('-').unwrap(); + assert_eq!( + IfBlock { + if_then: vec![IfThen { + conditions, + then: true + }], + default: false, + } + .eval(&envelope) + .await, + &expected_result.parse::().unwrap(), + "failed for {key:?}" + ); + } +} diff --git a/tests/src/smtp/inbound/milter.rs b/tests/src/smtp/inbound/milter.rs new file mode 100644 index 00000000..3b80543a --- /dev/null +++ b/tests/src/smtp/inbound/milter.rs @@ -0,0 +1,533 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::{fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; + +use mail_auth::AuthenticatedMessage; +use mail_parser::Message; +use serde::Deserialize; +use smtp::{ + config::{ConfigContext, IfBlock, Milter}, + core::{Session, SessionData, SMTP}, + inbound::milter::{ + receiver::{FrameResult, Receiver}, + Action, Command, Macros, MilterClient, Modification, Options, Response, Version, + }, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + sync::watch, +}; + +use crate::smtp::{ + session::{load_test_message, TestSession}, + ParseTestConfig, TestConfig, TestSMTP, +}; + +#[derive(Debug, Deserialize)] +struct HeaderTest { + modifications: Vec, + result: String, +} + +#[tokio::test] +async fn milter_session() { + // Enable logging + let disable = "true"; + tracing::subscriber::set_global_default( + tracing_subscriber::FmtSubscriber::builder() + .with_max_level(tracing::Level::TRACE) + .finish(), + ) + .unwrap(); + + // Configure tests + let _rx = spawn_mock_milter_server(); + tokio::time::sleep(Duration::from_millis(100)).await; + let mut core = SMTP::test(); + let mut qr = core.init_test_queue("smtp_milter_test"); + let mut config = &mut core.session.config; + config.rcpt.relay = IfBlock::new(true); + config.data.milters = r#"[[session.data.milter]] + hostname = "127.0.0.1" + port = 9332 + enable = true + version = 6 + tls = false + "# + .parse_milters(&ConfigContext::new(&[])); + + // Build session + let mut session = Session::test(core); + session.data.remote_ip = "10.0.0.1".parse().unwrap(); + session.eval_session_params().await; + session.ehlo("mx.doe.org").await; + + // Test reject + session + .send_message( + "reject@doe.org", + &["bill@foobar.org"], + "test:no_dkim", + "503 5.5.3", + ) + .await; + qr.assert_empty_queue(); + + // Test discard + session + .send_message( + "discard@doe.org", + &["bill@foobar.org"], + "test:no_dkim", + "250 2.0.0", + ) + .await; + qr.assert_empty_queue(); + + // Test temp fail + session + .send_message( + "temp_fail@doe.org", + &["bill@foobar.org"], + "test:no_dkim", + "451 4.3.5", + ) + .await; + qr.assert_empty_queue(); + + // Test shutdown + session + .send_message( + "shutdown@doe.org", + &["bill@foobar.org"], + "test:no_dkim", + "421 4.3.0", + ) + .await; + qr.assert_empty_queue(); + + // Test reply code + session + .send_message( + "reply_code@doe.org", + &["bill@foobar.org"], + "test:no_dkim", + "321", + ) + .await; + qr.assert_empty_queue(); +} + +#[test] +fn milter_address_modifications() { + let test_message = fs::read_to_string( + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("resources") + .join("smtp") + .join("milter") + .join("message.eml"), + ) + .unwrap(); + let parsed_test_message = AuthenticatedMessage::parse(test_message.as_bytes()).unwrap(); + + let mut data = SessionData::new( + "127.0.0.1".parse().unwrap(), + "127.0.0.1".parse().unwrap(), + 0, + ); + + // ChangeFrom + assert!(data + .apply_modifications( + vec![Modification::ChangeFrom { + sender: "<>".to_string(), + args: String::new() + }], + &parsed_test_message + ) + .is_none()); + let addr = data.mail_from.as_ref().unwrap(); + assert_eq!(addr.address_lcase, ""); + assert_eq!(addr.dsn_info, None); + assert_eq!(addr.flags, 0); + + // ChangeFrom with parameters + assert!(data + .apply_modifications( + vec![Modification::ChangeFrom { + sender: "john@example.org".to_string(), + args: "REQUIRETLS ENVID=abc123".to_string(), //"NOTIFY=SUCCESS,FAILURE ENVID=abc123\n".to_string() + }], + &parsed_test_message + ) + .is_none()); + let addr = data.mail_from.as_ref().unwrap(); + assert_eq!(addr.address_lcase, "john@example.org"); + assert_ne!(addr.flags, 0); + assert_eq!(addr.dsn_info, Some("abc123".to_string())); + + // Add recipients + assert!(data + .apply_modifications( + vec![ + Modification::AddRcpt { + recipient: "bill@example.org".to_string(), + args: "".to_string(), + }, + Modification::AddRcpt { + recipient: "jane@foobar.org".to_string(), + args: "NOTIFY=SUCCESS,FAILURE ORCPT=rfc822;Jane.Doe@Foobar.org".to_string(), + }, + Modification::AddRcpt { + recipient: "".to_string(), + args: "".to_string(), + }, + Modification::AddRcpt { + recipient: "<>".to_string(), + args: "".to_string(), + }, + ], + &parsed_test_message + ) + .is_none()); + assert_eq!(data.rcpt_to.len(), 2); + let addr = data.rcpt_to.first().unwrap(); + assert_eq!(addr.address_lcase, "bill@example.org"); + assert_eq!(addr.dsn_info, None); + assert_eq!(addr.flags, 0); + let addr = data.rcpt_to.last().unwrap(); + assert_eq!(addr.address_lcase, "jane@foobar.org"); + assert_ne!(addr.flags, 0); + assert_eq!(addr.dsn_info, Some("Jane.Doe@Foobar.org".to_string())); + + // Remove recipients + assert!(data + .apply_modifications( + vec![ + Modification::DeleteRcpt { + recipient: "bill@example.org".to_string(), + }, + Modification::DeleteRcpt { + recipient: "<>".to_string(), + }, + ], + &parsed_test_message + ) + .is_none()); + assert_eq!(data.rcpt_to.len(), 1); + let addr = data.rcpt_to.last().unwrap(); + assert_eq!(addr.address_lcase, "jane@foobar.org"); + assert_ne!(addr.flags, 0); + assert_eq!(addr.dsn_info, Some("Jane.Doe@Foobar.org".to_string())); +} + +#[test] +fn milter_message_modifications() { + // Read test message + let milter_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("resources") + .join("smtp") + .join("milter"); + let test_message = fs::read_to_string(milter_path.join("message.eml")).unwrap(); + let tests = serde_json::from_str::>( + &fs::read_to_string(milter_path.join("message.json")).unwrap(), + ) + .unwrap(); + let parsed_test_message = AuthenticatedMessage::parse(test_message.as_bytes()).unwrap(); + let mut session_data = SessionData::new( + "127.0.0.1".parse().unwrap(), + "127.0.0.1".parse().unwrap(), + 0, + ); + + for test in tests { + assert_eq!( + test.result, + String::from_utf8( + session_data + .apply_modifications(test.modifications, &parsed_test_message) + .unwrap() + ) + .unwrap() + ) + } +} + +#[test] +fn milter_frame_receiver() { + let mut stream = Vec::new(); + + for i in 0u32..100u32 { + stream.extend_from_slice((i + 1).to_be_bytes().as_ref()); + stream.push(i as u8); + for v in 0..i { + stream.push(v as u8); + } + } + + for chunk_size in [stream.len(), 1, 2, 3, 4, 10, 20, 30, 40, 100, 200, 300, 400] { + let mut receiver = Receiver::with_max_frame_len(100); + let mut frame_num = 0; + + 'outer: for chunk in stream.chunks(chunk_size) { + loop { + match receiver.read_frame(chunk) { + FrameResult::Frame(bytes) => { + /*println!( + "frame {frame_num}, chunk: {chunk_size}, {}", + if matches!(bytes, std::borrow::Cow::Borrowed(_)) { + "borrowed" + } else { + "owned" + } + );*/ + assert_eq!(*bytes.first().unwrap(), frame_num); + assert_eq!(bytes.len(), frame_num as usize + 1); + frame_num += 1; + } + FrameResult::Incomplete => continue 'outer, + FrameResult::TooLarge(size) => { + panic!("Frame too large: {size}") + } + } + } + } + + assert_eq!(frame_num, 100, "chunk_size: {}", chunk_size); + } +} + +#[tokio::test] +#[ignore] +async fn milter_client_test() { + let mut client = MilterClient::connect( + &Milter { + enable: IfBlock::default(), + addrs: vec![SocketAddr::from(([127, 0, 0, 1], 1234))], + hostname: "localhost".to_string(), + port: 1234, + timeout_connect: Duration::from_secs(10), + timeout_command: Duration::from_secs(30), + timeout_data: Duration::from_secs(30), + tls: false, + tls_allow_invalid_certs: false, + tempfail_on_error: false, + max_frame_len: 5000000, + protocol_version: Version::V6, + }, + tracing::span!(tracing::Level::TRACE, "hi"), + ) + .await + .unwrap(); + client.init().await.unwrap(); + + let raw_message = load_test_message("arc", "messages"); + let message = Message::parse(raw_message.as_bytes()).unwrap(); + + let r = client + .connection( + "gmail.com", + "127.0.0.1".parse().unwrap(), + 1235, + Macros::new(), + ) + .await + .unwrap(); + println!("CONNECT: {:?}", r); + let r = client + .mail_from("john@gmail.com", None::<&[&str]>, Macros::new()) + .await + .unwrap(); + println!("MAIL FROM: {:?}", r); + let r = client + .rcpt_to("user@gmail.com", None::<&[&str]>, Macros::new()) + .await + .unwrap(); + println!("RCPT TO: {:?}", r); + + let r = client.data().await.unwrap(); + println!("DATA: {:?}", r); + let r = client.headers(message.headers_raw()).await.unwrap(); + println!("HEADERS: {:?}", r); + let r = client + .body(&message.raw_message()[message.root_part().raw_body_offset()..]) + .await + .unwrap(); + println!("BODY: {:?}", r); + + client.quit().await.unwrap(); +} + +pub fn spawn_mock_milter_server() -> watch::Sender { + let (tx, rx) = watch::channel(true); + let tests = Arc::new( + serde_json::from_str::>( + &fs::read_to_string( + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("resources") + .join("smtp") + .join("milter") + .join("message.json"), + ) + .unwrap(), + ) + .unwrap(), + ); + + tokio::spawn(async move { + let listener = TcpListener::bind("127.0.0.1:9332") + .await + .unwrap_or_else(|e| { + panic!("Failed to bind mock Milter server to 127.0.0.1:9332: {e}"); + }); + let mut rx_ = rx.clone(); + //println!("Mock Milter server listening on port 9332"); + loop { + tokio::select! { + stream = listener.accept() => { + match stream { + Ok((stream, _)) => { + tokio::spawn(accept_milter(stream, rx.clone(), tests.clone())); + } + Err(err) => { + panic!("Something went wrong: {err}" ); + } + } + }, + _ = rx_.changed() => { + //println!("Mock Milter server stopping"); + break; + } + }; + } + }); + + tx +} + +async fn accept_milter( + mut stream: TcpStream, + mut rx: watch::Receiver, + tests: Arc>, +) { + let mut buf = vec![0u8; 1024]; + let mut receiver = Receiver::with_max_frame_len(5000000); + let mut action = None; + + 'outer: loop { + let br = tokio::select! { + br = stream.read(&mut buf) => { + match br { + Ok(br) => { + br + } + Err(_) => { + break; + } + } + }, + _ = rx.changed() => { + break; + } + }; + + if br == 0 { + break; + } + + loop { + match receiver.read_frame(&buf[..br]) { + FrameResult::Frame(bytes) => { + let cmd = Command::deserialize(bytes.as_ref()); + println!("CMD: {cmd}"); + + let response = match cmd { + Command::Abort | Command::Macro { .. } => continue, + Command::Body { .. } + | Command::Data + | Command::Connect { .. } + | Command::Header { .. } + | Command::Helo { .. } + | Command::Rcpt { .. } + | Command::QuitNewConnection + | Command::EndOfHeader => Response::Action(Action::Accept), + Command::OptionNegotiation(_) => Response::OptionNegotiation(Options { + version: 6, + actions: 0, + protocol: 0, + }), + Command::MailFrom { sender, .. } => { + let sender = std::str::from_utf8(sender).unwrap(); + action = match sender + .strip_prefix('<') + .unwrap() + .split_once('@') + .unwrap() + .0 + { + "accept" => Action::Accept, + "reject" => Action::Reject, + "discard" => Action::Discard, + "temp_fail" => Action::TempFail, + "shutdown" => Action::Shutdown, + "conn_fail" => Action::ConnectionFailure, + "reply_code" => Action::ReplyCode { + code: [b'3', b'2', b'1'], + text: "test".to_string(), + }, + test_num => { + for modification in + &tests[test_num.parse::().unwrap()].modifications + { + // Write modifications + stream + .write_all( + &Response::Modification(modification.clone()) + .serialize(), + ) + .await + .unwrap(); + } + Action::Accept + } + } + .into(); + Response::Action(Action::Accept) + } + Command::Quit => break 'outer, + Command::EndOfBody => Response::Action(action.take().unwrap()), + }; + + // Write response + stream.write_all(&response.serialize()).await.unwrap(); + } + FrameResult::Incomplete => continue 'outer, + FrameResult::TooLarge(size) => { + panic!("Frame too large: {size}") + } + } + } + } +} diff --git a/tests/src/smtp/inbound/mod.rs b/tests/src/smtp/inbound/mod.rs index 2902fac9..eb0fad89 100644 --- a/tests/src/smtp/inbound/mod.rs +++ b/tests/src/smtp/inbound/mod.rs @@ -40,6 +40,7 @@ pub mod dnsrbl; pub mod ehlo; pub mod limits; pub mod mail; +pub mod milter; pub mod rcpt; pub mod scripts; pub mod sign; diff --git a/tests/src/smtp/mod.rs b/tests/src/smtp/mod.rs index b00c790e..dc7539d8 100644 --- a/tests/src/smtp/mod.rs +++ b/tests/src/smtp/mod.rs @@ -38,12 +38,12 @@ use tokio::sync::mpsc; use smtp::{ config::{ - if_block::ConfigIf, queue::ConfigQueue, throttle::ConfigThrottle, AggregateReport, - ArcAuthConfig, Auth, ConfigContext, Connect, Data, DkimAuthConfig, DmarcAuthConfig, - DnsBlConfig, Dsn, Ehlo, EnvelopeKey, Extensions, IfBlock, IpRevAuthConfig, Mail, - MailAuthConfig, QueueConfig, QueueOutboundSourceIp, QueueOutboundTimeout, QueueOutboundTls, - QueueQuotas, QueueThrottle, Rcpt, Report, ReportAnalysis, ReportConfig, SessionConfig, - SessionThrottle, SpfAuthConfig, Throttle, VerifyStrategy, + if_block::ConfigIf, queue::ConfigQueue, session::ConfigSession, throttle::ConfigThrottle, + AggregateReport, ArcAuthConfig, Auth, ConfigContext, Connect, Data, DkimAuthConfig, + DmarcAuthConfig, DnsBlConfig, Dsn, Ehlo, EnvelopeKey, Extensions, IfBlock, IpRevAuthConfig, + Mail, MailAuthConfig, Milter, QueueConfig, QueueOutboundSourceIp, QueueOutboundTimeout, + QueueOutboundTls, QueueQuotas, QueueThrottle, Rcpt, Report, ReportAnalysis, ReportConfig, + SessionConfig, SessionThrottle, SpfAuthConfig, Throttle, VerifyStrategy, }, core::{ throttle::ThrottleKeyHasherBuilder, QueueCore, ReportCore, Resolvers, SessionCore, @@ -67,6 +67,7 @@ pub trait ParseTestConfig { fn parse_throttle(&self, ctx: &ConfigContext) -> Vec; fn parse_quota(&self, ctx: &ConfigContext) -> QueueQuotas; fn parse_queue_throttle(&self, ctx: &ConfigContext) -> QueueThrottle; + fn parse_milters(&self, ctx: &ConfigContext) -> Vec; } impl ParseTestConfig for &str { @@ -128,6 +129,28 @@ impl ParseTestConfig for &str { .parse_queue_throttle(ctx) .unwrap() } + + fn parse_milters(&self, ctx: &ConfigContext) -> Vec { + Config::parse(self) + .unwrap() + .parse_milters( + ctx, + &[ + EnvelopeKey::Recipient, + EnvelopeKey::RecipientDomain, + EnvelopeKey::Sender, + EnvelopeKey::SenderDomain, + EnvelopeKey::Mx, + EnvelopeKey::HeloDomain, + EnvelopeKey::AuthenticatedAs, + EnvelopeKey::Listener, + EnvelopeKey::RemoteIp, + EnvelopeKey::LocalIp, + EnvelopeKey::Priority, + ], + ) + .unwrap() + } } pub trait TestConfig { @@ -237,6 +260,7 @@ impl TestConfig for SessionConfig { add_message_id: IfBlock::new(true), add_date: IfBlock::new(true), pipe_commands: vec![], + milters: vec![], }, } } diff --git a/tests/src/smtp/outbound/dane.rs b/tests/src/smtp/outbound/dane.rs index 7d4b63f8..2aea39e7 100644 --- a/tests/src/smtp/outbound/dane.rs +++ b/tests/src/smtp/outbound/dane.rs @@ -22,16 +22,29 @@ */ use std::{ + collections::BTreeSet, + fs::{self, File}, + io::{BufRead, BufReader}, + num::ParseIntError, + path::PathBuf, sync::Arc, time::{Duration, Instant}, }; use mail_auth::{ - common::parse::TxtRecordParser, + common::{ + lru::{DnsCache, LruCache}, + parse::TxtRecordParser, + }, mta_sts::{ReportUri, TlsRpt}, report::tlsrpt::ResultType, - MX, + trust_dns_resolver::{ + config::{ResolverConfig, ResolverOpts}, + AsyncResolver, + }, + Resolver, MX, }; +use rustls::Certificate; use utils::config::ServerProtocol; use crate::smtp::{ @@ -42,9 +55,9 @@ use crate::smtp::{ }; use smtp::{ config::{AggregateFrequency, IfBlock, RequireOptional}, - core::{Session, SMTP}, - outbound::dane::{Tlsa, TlsaEntry}, - queue::{manager::Queue, DeliveryAttempt}, + core::{Resolvers, Session, SMTP}, + outbound::dane::{DnssecResolver, Tlsa, TlsaEntry}, + queue::{manager::Queue, DeliveryAttempt, Error, ErrorDetails, Status}, reporting::PolicyType, }; @@ -208,3 +221,122 @@ async fn dane_verify() { assert_eq!(report.policy, PolicyType::Tlsa(tlsa.into())); assert!(report.failure.is_none()); } + +#[tokio::test] +async fn dane_test() { + let conf = ResolverConfig::cloudflare_tls(); + let mut opts = ResolverOpts::default(); + opts.validate = true; + opts.try_tcp_on_error = true; + + let r = Resolvers { + dns: Resolver::new_cloudflare().unwrap(), + dnssec: DnssecResolver { + resolver: AsyncResolver::tokio(conf, opts).unwrap(), + }, + cache: smtp::core::DnsCache { + tlsa: LruCache::with_capacity(10), + mta_sts: LruCache::with_capacity(10), + }, + }; + + // Add dns entries + let mut path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + path.push("resources"); + path.push("smtp"); + path.push("dane"); + let mut file = path.clone(); + file.push("dns.txt"); + + let mut hosts = BTreeSet::new(); + let mut tlsa = Tlsa { + entries: Vec::new(), + has_end_entities: false, + has_intermediates: false, + }; + let mut hostname = String::new(); + + for line in BufReader::new(File::open(file).unwrap()).lines() { + let line = line.unwrap(); + let mut is_end_entity = false; + for (pos, item) in line.split_whitespace().enumerate() { + match pos { + 0 => { + if hostname != item && !hostname.is_empty() { + r.tlsa_add(hostname, tlsa, Instant::now() + Duration::from_secs(30)); + tlsa = Tlsa { + entries: Vec::new(), + has_end_entities: false, + has_intermediates: false, + }; + } + hosts.insert(item.strip_prefix("_25._tcp.").unwrap().to_string()); + hostname = item.to_string(); + } + 1 => { + is_end_entity = item == "3"; + } + 4 => { + if is_end_entity { + tlsa.has_end_entities = true; + } else { + tlsa.has_intermediates = true; + } + tlsa.entries.push(TlsaEntry { + is_end_entity, + is_sha256: true, + is_spki: true, + data: decode_hex(item).unwrap(), + }); + } + _ => (), + } + } + } + r.tlsa_add(hostname, tlsa, Instant::now() + Duration::from_secs(30)); + + // Add certificates + assert!(!hosts.is_empty()); + for host in hosts { + // Add certificates + let mut certs = Vec::new(); + for num in 0..6 { + let mut file = path.clone(); + file.push(format!("{host}.{num}.cert")); + if file.exists() { + certs.push(Certificate(fs::read(file).unwrap())); + } else { + break; + } + } + + // Successful DANE verification + let tlsa = r + .tlsa_lookup(format!("_25._tcp.{host}.")) + .await + .unwrap() + .unwrap(); + + assert_eq!( + tlsa.verify(&tracing::info_span!("test_span"), &host, Some(&certs)), + Ok(()) + ); + + // Failed DANE verification + certs.remove(0); + assert_eq!( + tlsa.verify(&tracing::info_span!("test_span"), &host, Some(&certs)), + Err(Status::PermanentFailure(Error::DaneError(ErrorDetails { + entity: host.to_string(), + details: "No matching certificates found in TLSA records".to_string() + }))) + ); + } +} + +pub fn decode_hex(s: &str) -> Result, ParseIntError> { + (0..s.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&s[i..i + 2], 16)) + .collect() +} diff --git a/tests/src/smtp/session.rs b/tests/src/smtp/session.rs index 660290ec..1e0a08bd 100644 --- a/tests/src/smtp/session.rs +++ b/tests/src/smtp/session.rs @@ -92,6 +92,10 @@ impl IsTls for DummyIo { } fn write_tls_header(&self, _headers: &mut Vec) {} + + fn tls_version_and_cipher(&self) -> (&'static str, &'static str) { + ("", "") + } } impl Unpin for DummyIo {} @@ -125,7 +129,11 @@ impl TestSession for Session { tx_buf: vec![], tls: false, }, - data: SessionData::new("127.0.0.1".parse().unwrap(), "127.0.0.1".parse().unwrap()), + data: SessionData::new( + "127.0.0.1".parse().unwrap(), + "127.0.0.1".parse().unwrap(), + 0, + ), params: SessionParameters::default(), in_flight: vec![], }