Milter implementation.

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"] }

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"]

tests/resources/smtp/milter/message.eml

@@ -0,0 +1,11 @@
+From: John Doe <john@example.org>
+To: Mary Smith <mary.smith@example.org>
+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.

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 <john@example.org>\r\nTo: Mary Smith <mary.smith@example.org>\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 <john@example.org>\r\nTo: Mary Smith <mary.smith@example.org>\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 <john@example.org>\r\nTo: Mary Smith <mary.smith@example.org>\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 <john@example.org>\r\nTo: Mary Smith <mary.smith@example.org>\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": "<my-new-ref>"
+                }
+            },
+            {
+                "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 <john@example.org>\r\nTo: Mary Smith <mary.smith@example.org>\r\nReferences: a\r\nReferences: <my-new-ref>\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": "<my-new-ref>"
+                }
+            }
+        ],
+        "result": "X-Quarantine: Virus found!\r\nFrom: John Doe <john@example.org>\r\nTo: Mary Smith <mary.smith@example.org>\r\nReferences: <my-new-ref>\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"
+    }
+]

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::<bool>().unwrap(),
+            "failed for {key:?}"
+        );
+    }
+}

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 <http://www.gnu.org/licenses/>.
+ *
+ * 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<Modification>,
+    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: "<bill@example.org>".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::<Vec<HeaderTest>>(
+        &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<bool> {
+    let (tx, rx) = watch::channel(true);
+    let tests = Arc::new(
+        serde_json::from_str::<Vec<HeaderTest>>(
+            &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<bool>,
+    tests: Arc<Vec<HeaderTest>>,
+) {
+    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::<usize>().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}")
+                }
+            }
+        }
+    }
+}

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;

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<Throttle>;
     fn parse_quota(&self, ctx: &ConfigContext) -> QueueQuotas;
     fn parse_queue_throttle(&self, ctx: &ConfigContext) -> QueueThrottle;
+    fn parse_milters(&self, ctx: &ConfigContext) -> Vec<Milter>;
 }
 
 impl ParseTestConfig for &str {
@@ -128,6 +129,28 @@ impl ParseTestConfig for &str {
             .parse_queue_throttle(ctx)
             .unwrap()
     }
+
+    fn parse_milters(&self, ctx: &ConfigContext) -> Vec<Milter> {
+        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![],
             },
         }
     }

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<Vec<u8>, ParseIntError> {
+    (0..s.len())
+        .step_by(2)
+        .map(|i| u8::from_str_radix(&s[i..i + 2], 16))
+        .collect()
+}

tests/src/smtp/session.rs

@@ -92,6 +92,10 @@ impl IsTls for DummyIo {
     }
 
     fn write_tls_header(&self, _headers: &mut Vec<u8>) {}
+
+    fn tls_version_and_cipher(&self) -> (&'static str, &'static str) {
+        ("", "")
+    }
 }
 
 impl Unpin for DummyIo {}
@@ -125,7 +129,11 @@ impl TestSession for Session<DummyIo> {
                 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![],
         }