mail_send/smtp/
client.rs

1/*
2 * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
3 *
4 * SPDX-License-Identifier: Apache-2.0 OR MIT
5 */
6
7use std::{
8    net::{IpAddr, SocketAddr},
9    time::Duration,
10};
11
12use smtp_proto::{Response, response::parser::ResponseReceiver};
13use tokio::{
14    io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
15    net::{TcpSocket, TcpStream},
16};
17
18use crate::SmtpClient;
19
20impl<T: AsyncRead + AsyncWrite + Unpin> SmtpClient<T> {
21    pub async fn read(&mut self) -> crate::Result<Response<String>> {
22        let mut buf = vec![0u8; 1024];
23        let mut parser = ResponseReceiver::default();
24
25        loop {
26            let br = self.stream.read(&mut buf).await?;
27
28            if br > 0 {
29                match parser.parse(&mut buf[..br].iter()) {
30                    Ok(reply) => return Ok(reply),
31                    Err(err) => match err {
32                        smtp_proto::Error::NeedsMoreData { .. } => (),
33                        _ => {
34                            return Err(crate::Error::UnparseableReply);
35                        }
36                    },
37                }
38            } else {
39                return Err(crate::Error::UnparseableReply);
40            }
41        }
42    }
43
44    pub async fn read_many(&mut self, num: usize) -> crate::Result<Vec<Response<String>>> {
45        let mut buf = vec![0u8; 1024];
46        let mut response = Vec::with_capacity(num);
47        let mut parser = ResponseReceiver::default();
48
49        'outer: loop {
50            let br = self.stream.read(&mut buf).await?;
51
52            if br > 0 {
53                let mut iter = buf[..br].iter();
54
55                loop {
56                    match parser.parse(&mut iter) {
57                        Ok(reply) => {
58                            response.push(reply);
59                            if response.len() != num {
60                                parser.reset();
61                            } else {
62                                break 'outer;
63                            }
64                        }
65                        Err(err) => match err {
66                            smtp_proto::Error::NeedsMoreData { .. } => break,
67                            _ => {
68                                return Err(crate::Error::UnparseableReply);
69                            }
70                        },
71                    }
72                }
73            } else {
74                return Err(crate::Error::UnparseableReply);
75            }
76        }
77
78        Ok(response)
79    }
80
81    /// Sends a command to the SMTP server and waits for a reply.
82    pub async fn cmd(&mut self, cmd: impl AsRef<[u8]>) -> crate::Result<Response<String>> {
83        tokio::time::timeout(self.timeout, async {
84            self.stream.write_all(cmd.as_ref()).await?;
85            self.stream.flush().await?;
86            self.read().await
87        })
88        .await
89        .map_err(|_| crate::Error::Timeout)?
90    }
91
92    /// Pipelines multiple command to the SMTP server and waits for a reply.
93    pub async fn cmds(
94        &mut self,
95        cmds: impl IntoIterator<Item = impl AsRef<[u8]>>,
96    ) -> crate::Result<Vec<Response<String>>> {
97        tokio::time::timeout(self.timeout, async {
98            let mut num_replies = 0;
99            for cmd in cmds {
100                self.stream.write_all(cmd.as_ref()).await?;
101                num_replies += 1;
102            }
103            self.stream.flush().await?;
104            self.read_many(num_replies).await
105        })
106        .await
107        .map_err(|_| crate::Error::Timeout)?
108    }
109}
110
111impl SmtpClient<TcpStream> {
112    /// Connects to a remote host address
113    pub async fn connect(remote_addr: SocketAddr, timeout: Duration) -> crate::Result<Self> {
114        tokio::time::timeout(timeout, async {
115            Ok(SmtpClient {
116                stream: TcpStream::connect(remote_addr).await?,
117                timeout,
118            })
119        })
120        .await
121        .map_err(|_| crate::Error::Timeout)?
122    }
123
124    /// Connects to a remote host address using the provided local IP
125    pub async fn connect_using(
126        local_ip: IpAddr,
127        remote_addr: SocketAddr,
128        timeout: Duration,
129    ) -> crate::Result<Self> {
130        tokio::time::timeout(timeout, async {
131            let socket = if local_ip.is_ipv4() {
132                TcpSocket::new_v4()?
133            } else {
134                TcpSocket::new_v6()?
135            };
136            socket.bind(SocketAddr::new(local_ip, 0))?;
137
138            Ok(SmtpClient {
139                stream: socket.connect(remote_addr).await?,
140                timeout,
141            })
142        })
143        .await
144        .map_err(|_| crate::Error::Timeout)?
145    }
146}
147
148#[cfg(test)]
149mod test {
150    use std::time::Duration;
151
152    use tokio::io::{AsyncRead, AsyncWrite};
153
154    use crate::{SmtpClient, SmtpClientBuilder};
155
156    #[tokio::test]
157    async fn smtp_basic() {
158        // StartTLS test
159        env_logger::init();
160        let client = SmtpClientBuilder::new("mail.smtp2go.com", 2525)
161            .unwrap()
162            .implicit_tls(false)
163            .connect()
164            .await
165            .unwrap();
166        client.quit().await.unwrap();
167        let client = SmtpClientBuilder::new("mail.smtp2go.com", 2525)
168            .unwrap()
169            .allow_invalid_certs()
170            .implicit_tls(false)
171            .connect()
172            .await
173            .unwrap();
174        client.quit().await.unwrap();
175
176        // Say hello to Google over TLS and quit
177        let client = SmtpClientBuilder::new("smtp.gmail.com", 465)
178            .unwrap()
179            .connect()
180            .await
181            .unwrap();
182        client.quit().await.unwrap();
183
184        // Say hello to Google over TLS and quit
185        let client = SmtpClientBuilder::new("smtp.gmail.com", 465)
186            .unwrap()
187            .allow_invalid_certs()
188            .connect()
189            .await
190            .unwrap();
191        client.quit().await.unwrap();
192    }
193
194    #[derive(Default)]
195    struct AsyncBufWriter {
196        buf: Vec<u8>,
197    }
198
199    impl AsyncRead for AsyncBufWriter {
200        fn poll_read(
201            self: std::pin::Pin<&mut Self>,
202            _cx: &mut std::task::Context<'_>,
203            _buf: &mut tokio::io::ReadBuf<'_>,
204        ) -> std::task::Poll<std::io::Result<()>> {
205            unreachable!()
206        }
207    }
208
209    impl AsyncWrite for AsyncBufWriter {
210        fn poll_write(
211            mut self: std::pin::Pin<&mut Self>,
212            _cx: &mut std::task::Context<'_>,
213            buf: &[u8],
214        ) -> std::task::Poll<Result<usize, std::io::Error>> {
215            self.buf.extend_from_slice(buf);
216            std::task::Poll::Ready(Ok(buf.len()))
217        }
218
219        fn poll_flush(
220            self: std::pin::Pin<&mut Self>,
221            _cx: &mut std::task::Context<'_>,
222        ) -> std::task::Poll<Result<(), std::io::Error>> {
223            std::task::Poll::Ready(Ok(()))
224        }
225
226        fn poll_shutdown(
227            self: std::pin::Pin<&mut Self>,
228            _cx: &mut std::task::Context<'_>,
229        ) -> std::task::Poll<Result<(), std::io::Error>> {
230            std::task::Poll::Ready(Ok(()))
231        }
232    }
233
234    #[tokio::test]
235    async fn transparency_procedure() {
236        const SMUGGLER: &str = r#"From: Joe SixPack <john@foobar.net>
237To: Suzie Q <suzie@foobar.org>
238Subject: Is dinner ready?
239
240Hi.
241
242We lost the game. Are you hungry yet?
243
244Joe.
245
246<SEP>.
247MAIL FROM:<admin@foobar.net>
248RCPT TO:<ok@foobar.org>
249DATA
250From: Joe SixPack <admin@foobar.net>
251To: Suzie Q <suzie@foobar.org>
252Subject: smuggled message
253
254This is a smuggled message
255"#;
256
257        for (test, result) in [
258            (
259                "A: b\r\n.\r\n".to_string(),
260                "A: b\r\n..\r\n\r\n.\r\n".to_string(),
261            ),
262            ("A: b\r\n.".to_string(), "A: b\r\n..\r\n.\r\n".to_string()),
263            (
264                "A: b\r\n..\r\n".to_string(),
265                "A: b\r\n...\r\n\r\n.\r\n".to_string(),
266            ),
267            ("A: ...b".to_string(), "A: ...b\r\n.\r\n".to_string()),
268            (
269                "A: \n.\r\nMAIL FROM:<>".to_string(),
270                "A: \n..\r\nMAIL FROM:<>\r\n.\r\n".to_string(),
271            ),
272            (
273                "A: \r.\r\nMAIL FROM:<>".to_string(),
274                "A: \r..\r\nMAIL FROM:<>\r\n.\r\n".to_string(),
275            ),
276            (
277                SMUGGLER
278                    .replace('\r', "")
279                    .replace('\n', "\r\n")
280                    .replace("<SEP>", "\r"),
281                SMUGGLER
282                    .replace('\r', "")
283                    .replace('\n', "\r\n")
284                    .replace("<SEP>", "\r.")
285                    + "\r\n.\r\n",
286            ),
287            (
288                SMUGGLER
289                    .replace('\r', "")
290                    .replace('\n', "\r\n")
291                    .replace("<SEP>", "\n"),
292                SMUGGLER
293                    .replace('\r', "")
294                    .replace('\n', "\r\n")
295                    .replace("<SEP>", "\n.")
296                    + "\r\n.\r\n",
297            ),
298        ] {
299            let mut client = SmtpClient {
300                stream: AsyncBufWriter::default(),
301                timeout: Duration::from_secs(30),
302            };
303            client.write_message(test.as_bytes()).await.unwrap();
304            assert_eq!(String::from_utf8(client.stream.buf).unwrap(), result);
305        }
306    }
307}
mail_send/smtp/client.rs

mail_send/smtp/
client.rs
