rustlavel_mail/smtp/
stream.rs

1//! The socket an SMTP conversation runs over, plain or wrapped in TLS.
2//!
3//! An enum rather than a boxed trait object, for the same reason the HTTP
4//! client uses one: there are exactly two cases, and the read path is hot
5//! enough that dynamic dispatch would be a needless cost.
6//!
7//! The client itself is generic over any `AsyncRead + AsyncWrite`, which is
8//! what lets the whole protocol be tested against an in-memory duplex pipe
9//! with no socket, no port, and no server.
10
11use rustlavel_core::{Error, Result};
12use std::pin::Pin;
13use std::sync::Arc;
14use std::task::{Context, Poll};
15use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
16use tokio::net::TcpStream;
17
18pub enum SmtpStream {
19    Plain(TcpStream),
20    Tls(Box<tokio_rustls::client::TlsStream<TcpStream>>),
21}
22
23impl SmtpStream {
24    pub fn is_encrypted(&self) -> bool {
25        matches!(self, SmtpStream::Tls(_))
26    }
27
28    /// Negotiate TLS over an already-open connection, for STARTTLS.
29    pub async fn upgrade(self, host: &str) -> Result<SmtpStream> {
30        let tcp = match self {
31            SmtpStream::Plain(tcp) => tcp,
32            already => return Ok(already),
33        };
34
35        let name = rustls::pki_types::ServerName::try_from(host.to_string())
36            .map_err(|_| Error::msg(format!("`{host}` is not a valid TLS server name")))?;
37
38        let tls = connector()
39            .connect(name, tcp)
40            .await
41            .map_err(|e| Error::msg(format!("TLS handshake with {host} failed: {e}")))?;
42
43        Ok(SmtpStream::Tls(Box::new(tls)))
44    }
45}
46
47/// Open a TCP connection, wrapping it in TLS straight away when asked.
48///
49/// Implicit TLS is what port 465 means: the server expects a handshake before
50/// it says anything at all.
51pub async fn connect(
52    host: &str,
53    port: u16,
54    implicit_tls: bool,
55    timeout: std::time::Duration,
56) -> Result<SmtpStream> {
57    let address = format!("{host}:{port}");
58
59    let tcp = tokio::time::timeout(timeout, TcpStream::connect(&address))
60        .await
61        .map_err(|_| {
62            Error::msg(format!(
63                "timed out connecting to {address}. Is the mail server reachable, and is \
64                 mail.port right?"
65            ))
66        })?
67        .map_err(|e| Error::msg(format!("cannot connect to {address}: {e}")))?;
68
69    let _ = tcp.set_nodelay(true);
70
71    let stream = SmtpStream::Plain(tcp);
72    if implicit_tls { stream.upgrade(host).await } else { Ok(stream) }
73}
74
75/// The TLS configuration, built once and shared.
76///
77/// Trust anchors come from webpki-roots rather than the OS store, so behaviour
78/// is identical on a laptop and in a scratch container.
79fn connector() -> tokio_rustls::TlsConnector {
80    use std::sync::OnceLock;
81    static CONNECTOR: OnceLock<tokio_rustls::TlsConnector> = OnceLock::new();
82
83    CONNECTOR
84        .get_or_init(|| {
85            let roots = rustls::RootCertStore { roots: webpki_roots::TLS_SERVER_ROOTS.to_vec() };
86            let config =
87                rustls::ClientConfig::builder().with_root_certificates(roots).with_no_client_auth();
88            tokio_rustls::TlsConnector::from(Arc::new(config))
89        })
90        .clone()
91}
92
93impl AsyncRead for SmtpStream {
94    fn poll_read(
95        self: Pin<&mut Self>,
96        cx: &mut Context<'_>,
97        buf: &mut ReadBuf<'_>,
98    ) -> Poll<std::io::Result<()>> {
99        match self.get_mut() {
100            SmtpStream::Plain(stream) => Pin::new(stream).poll_read(cx, buf),
101            SmtpStream::Tls(stream) => Pin::new(stream).poll_read(cx, buf),
102        }
103    }
104}
105
106impl AsyncWrite for SmtpStream {
107    fn poll_write(
108        self: Pin<&mut Self>,
109        cx: &mut Context<'_>,
110        buf: &[u8],
111    ) -> Poll<std::io::Result<usize>> {
112        match self.get_mut() {
113            SmtpStream::Plain(stream) => Pin::new(stream).poll_write(cx, buf),
114            SmtpStream::Tls(stream) => Pin::new(stream).poll_write(cx, buf),
115        }
116    }
117
118    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
119        match self.get_mut() {
120            SmtpStream::Plain(stream) => Pin::new(stream).poll_flush(cx),
121            SmtpStream::Tls(stream) => Pin::new(stream).poll_flush(cx),
122        }
123    }
124
125    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
126        match self.get_mut() {
127            SmtpStream::Plain(stream) => Pin::new(stream).poll_shutdown(cx),
128            SmtpStream::Tls(stream) => Pin::new(stream).poll_shutdown(cx),
129        }
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::*;
136    use std::time::Duration;
137
138    #[tokio::test]
139    async fn a_refused_connection_names_the_host_and_the_port_setting() {
140        let message = match connect("127.0.0.1", 1, false, Duration::from_secs(2)).await {
141            Err(error) => error.to_string(),
142            Ok(_) => panic!("nothing should be listening on port 1"),
143        };
144
145        assert!(message.contains("127.0.0.1:1"), "{message}");
146    }
147}
rustlavel_mail/smtp/stream.rs

rustlavel_mail/smtp/
stream.rs
