minco_plugin_notifications/
mailpit.rs

1use crate::{
2    MailAddress, MailError, MailErrorKind, MailMessage, MailReceipt, MailTransport, render_mime,
3};
4use async_trait::async_trait;
5use chrono::Utc;
6use std::{fmt, net::SocketAddr, time::Duration};
7use tokio::{
8    io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
9    net::TcpStream,
10    time::timeout,
11};
12
13const MAX_SMTP_RESPONSE_BYTES: usize = 16 * 1024;
14
15#[derive(Debug, Clone)]
16pub struct MailpitTransportConfig {
17    pub endpoint: SocketAddr,
18    pub from: MailAddress,
19    pub connect_timeout: Duration,
20    pub command_timeout: Duration,
21}
22
23impl MailpitTransportConfig {
24    pub fn new(endpoint: SocketAddr, from: MailAddress) -> Result<Self, MailError> {
25        if !endpoint.ip().is_loopback() {
26            return Err(MailError::new(
27                MailErrorKind::Configuration,
28                "mailpit",
29                "Mailpit SMTP endpoint must be loopback-only",
30            ));
31        }
32        from.validate()?;
33        Ok(Self {
34            endpoint,
35            from,
36            connect_timeout: Duration::from_secs(2),
37            command_timeout: Duration::from_secs(3),
38        })
39    }
40}
41
42impl Default for MailpitTransportConfig {
43    fn default() -> Self {
44        Self::new(
45            SocketAddr::from(([127, 0, 0, 1], 1025)),
46            MailAddress::new("minco@localhost").expect("static local address"),
47        )
48        .expect("static loopback Mailpit configuration")
49    }
50}
51
52#[derive(Clone)]
53pub struct MailpitTransport {
54    config: MailpitTransportConfig,
55}
56
57impl MailpitTransport {
58    pub fn new(config: MailpitTransportConfig) -> Result<Self, MailError> {
59        if !config.endpoint.ip().is_loopback()
60            || config.connect_timeout.is_zero()
61            || config.command_timeout.is_zero()
62        {
63            return Err(MailError::new(
64                MailErrorKind::Configuration,
65                "mailpit",
66                "Mailpit transport configuration is invalid",
67            ));
68        }
69        config.from.validate()?;
70        Ok(Self { config })
71    }
72}
73
74impl Default for MailpitTransport {
75    fn default() -> Self {
76        Self::new(MailpitTransportConfig::default()).expect("static Mailpit configuration")
77    }
78}
79
80impl fmt::Debug for MailpitTransport {
81    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
82        formatter
83            .debug_struct("MailpitTransport")
84            .field("endpoint", &self.config.endpoint)
85            .field("from", &"[REDACTED]")
86            .field("connect_timeout", &self.config.connect_timeout)
87            .field("command_timeout", &self.config.command_timeout)
88            .finish()
89    }
90}
91
92#[async_trait]
93impl MailTransport for MailpitTransport {
94    fn name(&self) -> &'static str {
95        "mailpit"
96    }
97
98    async fn send(&self, message: &MailMessage, attempt: u32) -> Result<MailReceipt, MailError> {
99        message.validate()?;
100        let mime = render_mime(message, &self.config.from)?;
101        let stream = timeout(
102            self.config.connect_timeout,
103            TcpStream::connect(self.config.endpoint),
104        )
105        .await
106        .map_err(|_| {
107            MailError::new(
108                MailErrorKind::Unavailable,
109                self.name(),
110                "Mailpit SMTP connection timed out",
111            )
112        })?
113        .map_err(|_| {
114            MailError::new(
115                MailErrorKind::Unavailable,
116                self.name(),
117                "Mailpit SMTP endpoint is unavailable",
118            )
119        })?;
120        let mut connection = SmtpConnection::new(stream, self.config.command_timeout);
121
122        connection.expect_response(220, false).await?;
123        connection
124            .command("EHLO minco.local\r\n", 250, false)
125            .await?;
126        connection
127            .command(
128                &format!("MAIL FROM:<{}>\r\n", self.config.from.address),
129                250,
130                false,
131            )
132            .await?;
133        for recipient in message.recipients() {
134            connection
135                .command(&format!("RCPT TO:<{}>\r\n", recipient.address), 250, false)
136                .await?;
137        }
138        connection.command("DATA\r\n", 354, false).await?;
139        connection.write_message(&mime).await?;
140        connection.expect_response(250, true).await?;
141        let _ = connection.command("QUIT\r\n", 221, true).await;
142
143        Ok(MailReceipt {
144            message_id: message.id,
145            transport: self.name().into(),
146            provider_message_id: format!("mailpit:{}", message.id),
147            accepted_at: Utc::now(),
148            attempt,
149        })
150    }
151}
152
153struct SmtpConnection {
154    stream: BufReader<TcpStream>,
155    command_timeout: Duration,
156}
157
158impl SmtpConnection {
159    fn new(stream: TcpStream, command_timeout: Duration) -> Self {
160        Self {
161            stream: BufReader::new(stream),
162            command_timeout,
163        }
164    }
165
166    async fn command(
167        &mut self,
168        command: &str,
169        expected: u16,
170        after_data: bool,
171    ) -> Result<String, MailError> {
172        self.write_all(command.as_bytes(), after_data).await?;
173        self.expect_response(expected, after_data).await
174    }
175
176    async fn write_message(&mut self, mime: &[u8]) -> Result<(), MailError> {
177        let mut body = dot_stuff(mime);
178        if !body.ends_with(b"\r\n") {
179            body.extend_from_slice(b"\r\n");
180        }
181        body.extend_from_slice(b".\r\n");
182        self.write_all(&body, true).await
183    }
184
185    async fn write_all(&mut self, bytes: &[u8], after_data: bool) -> Result<(), MailError> {
186        timeout(self.command_timeout, async {
187            self.stream.get_mut().write_all(bytes).await?;
188            self.stream.get_mut().flush().await
189        })
190        .await
191        .map_err(|_| smtp_io_error(after_data, "SMTP write timed out"))?
192        .map_err(|_| smtp_io_error(after_data, "SMTP connection closed during write"))
193    }
194
195    async fn expect_response(
196        &mut self,
197        expected: u16,
198        after_data: bool,
199    ) -> Result<String, MailError> {
200        let (status, response) = timeout(self.command_timeout, self.read_response())
201            .await
202            .map_err(|_| smtp_io_error(after_data, "SMTP response timed out"))?
203            .map_err(|_| smtp_io_error(after_data, "SMTP connection closed during response"))?;
204        if status == expected {
205            return Ok(response);
206        }
207        let kind = match status / 100 {
208            4 if after_data => MailErrorKind::Unavailable,
209            4 => MailErrorKind::Unavailable,
210            5 => MailErrorKind::Rejected,
211            _ if after_data => MailErrorKind::Ambiguous,
212            _ => MailErrorKind::Protocol,
213        };
214        Err(MailError::new(
215            kind,
216            "mailpit",
217            format!("SMTP server returned status {status}"),
218        ))
219    }
220
221    async fn read_response(&mut self) -> std::io::Result<(u16, String)> {
222        let mut complete = String::new();
223        let mut status = None;
224        loop {
225            let mut line = String::new();
226            if self.stream.read_line(&mut line).await? == 0 {
227                return Err(std::io::Error::new(
228                    std::io::ErrorKind::UnexpectedEof,
229                    "SMTP response ended unexpectedly",
230                ));
231            }
232            if complete.len() + line.len() > MAX_SMTP_RESPONSE_BYTES {
233                return Err(std::io::Error::new(
234                    std::io::ErrorKind::InvalidData,
235                    "SMTP response exceeds the bounded size",
236                ));
237            }
238            let bytes = line.as_bytes();
239            if bytes.len() < 4 || !bytes[..3].iter().all(u8::is_ascii_digit) {
240                return Err(std::io::Error::new(
241                    std::io::ErrorKind::InvalidData,
242                    "SMTP response is malformed",
243                ));
244            }
245            let code = line[..3].parse::<u16>().map_err(|_| {
246                std::io::Error::new(std::io::ErrorKind::InvalidData, "SMTP status is malformed")
247            })?;
248            if status
249                .replace(code)
250                .is_some_and(|previous| previous != code)
251            {
252                return Err(std::io::Error::new(
253                    std::io::ErrorKind::InvalidData,
254                    "SMTP multiline status changed",
255                ));
256            }
257            let continuation = bytes[3] == b'-';
258            if !continuation && bytes[3] != b' ' {
259                return Err(std::io::Error::new(
260                    std::io::ErrorKind::InvalidData,
261                    "SMTP response separator is malformed",
262                ));
263            }
264            complete.push_str(line.trim_end_matches(['\r', '\n']));
265            complete.push('\n');
266            if !continuation {
267                return Ok((code, complete));
268            }
269        }
270    }
271}
272
273fn smtp_io_error(after_data: bool, message: &str) -> MailError {
274    MailError::new(
275        if after_data {
276            MailErrorKind::Ambiguous
277        } else {
278            MailErrorKind::Unavailable
279        },
280        "mailpit",
281        message,
282    )
283}
284
285fn dot_stuff(message: &[u8]) -> Vec<u8> {
286    let mut output = Vec::with_capacity(message.len() + 16);
287    let mut at_line_start = true;
288    for byte in message {
289        if at_line_start && *byte == b'.' {
290            output.push(b'.');
291        }
292        output.push(*byte);
293        at_line_start = *byte == b'\n';
294    }
295    output
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301    use crate::MailMessage;
302    use tokio::{
303        io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
304        net::TcpListener,
305    };
306
307    #[tokio::test]
308    async fn loopback_smtp_captures_rich_mime_without_bcc_header() {
309        let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
310        let endpoint = listener.local_addr().unwrap();
311        let server = tokio::spawn(async move {
312            let (stream, _) = listener.accept().await.unwrap();
313            let mut stream = BufReader::new(stream);
314            stream
315                .get_mut()
316                .write_all(b"220 mailpit ESMTP\r\n")
317                .await
318                .unwrap();
319            let mut captured = Vec::new();
320            let mut recipients = Vec::new();
321            loop {
322                let mut line = String::new();
323                if stream.read_line(&mut line).await.unwrap() == 0 {
324                    break;
325                }
326                if line.starts_with("EHLO") {
327                    stream
328                        .get_mut()
329                        .write_all(b"250-mailpit\r\n250 8BITMIME\r\n")
330                        .await
331                        .unwrap();
332                } else if line.starts_with("MAIL FROM") {
333                    stream.get_mut().write_all(b"250 ok\r\n").await.unwrap();
334                } else if line.starts_with("RCPT TO") {
335                    recipients.push(line.trim().to_owned());
336                    stream.get_mut().write_all(b"250 ok\r\n").await.unwrap();
337                } else if line == "DATA\r\n" {
338                    stream
339                        .get_mut()
340                        .write_all(b"354 send data\r\n")
341                        .await
342                        .unwrap();
343                    loop {
344                        let mut data_line = Vec::new();
345                        stream.read_until(b'\n', &mut data_line).await.unwrap();
346                        if data_line == b".\r\n" {
347                            break;
348                        }
349                        captured.extend_from_slice(&data_line);
350                    }
351                    stream
352                        .get_mut()
353                        .write_all(b"250 accepted\r\n")
354                        .await
355                        .unwrap();
356                } else if line.starts_with("QUIT") {
357                    stream.get_mut().write_all(b"221 bye\r\n").await.unwrap();
358                    break;
359                }
360            }
361            (captured, recipients)
362        });
363
364        let transport = MailpitTransport::new(
365            MailpitTransportConfig::new(
366                endpoint,
367                MailAddress::new("no-reply@example.com").unwrap(),
368            )
369            .unwrap(),
370        )
371        .unwrap();
372        let message = MailMessage::builder("account.welcome", "Welcome")
373            .to(MailAddress::new("person@example.com").unwrap())
374            .bcc(MailAddress::new("audit@example.com").unwrap())
375            .text("Hello")
376            .html("<p>Hello</p>")
377            .build()
378            .unwrap();
379        let receipt = transport.send(&message, 1).await.unwrap();
380        assert_eq!(receipt.transport, "mailpit");
381        let (captured, recipients) = server.await.unwrap();
382        let captured = String::from_utf8(captured).unwrap();
383        assert!(
384            recipients
385                .iter()
386                .any(|recipient| recipient == "RCPT TO:<audit@example.com>")
387        );
388        assert!(!captured.contains("Bcc:"));
389        assert!(!captured.contains("audit@example.com"));
390        assert!(captured.contains("multipart/alternative"));
391    }
392
393    #[test]
394    fn remote_plaintext_smtp_is_rejected() {
395        let endpoint = SocketAddr::from(([192, 0, 2, 1], 1025));
396        assert!(
397            MailpitTransportConfig::new(
398                endpoint,
399                MailAddress::new("no-reply@example.com").unwrap()
400            )
401            .is_err()
402        );
403    }
404
405    #[test]
406    fn dot_stuffing_covers_every_line_start() {
407        assert_eq!(dot_stuff(b".a\r\n.b\r\n"), b"..a\r\n..b\r\n");
408    }
409}
minco_plugin_notifications/mailpit.rs

minco_plugin_notifications/
mailpit.rs
