deltachat/smtp/
connect.rs

1//! SMTP connection establishment.
2
3use std::net::SocketAddr;
4
5use anyhow::{Context as _, Result, bail};
6use async_smtp::{SmtpClient, SmtpTransport};
7use tokio::io::{AsyncBufRead, AsyncWrite, BufStream};
8
9use crate::context::Context;
10use crate::log::warn;
11use crate::net::dns::{lookup_host_with_cache, update_connect_timestamp};
12use crate::net::proxy::ProxyConfig;
13use crate::net::session::SessionBufStream;
14use crate::net::tls::{TlsSessionStore, wrap_tls};
15use crate::net::{
16    connect_tcp_inner, connect_tls_inner, run_connection_attempts, update_connection_history,
17};
18use crate::oauth2::get_oauth2_access_token;
19use crate::tools::time;
20use crate::transport::ConnectionCandidate;
21use crate::transport::ConnectionSecurity;
22
23/// Converts port number to ALPN.
24fn alpn(port: u16) -> &'static str {
25    if port == 465 {
26        // Do not request ALPN on standard port.
27        ""
28    } else {
29        "smtp"
30    }
31}
32
33// Constructs a new SMTP transport
34// over a stream with already skipped SMTP greeting.
35async fn new_smtp_transport<S: AsyncBufRead + AsyncWrite + Unpin>(
36    stream: S,
37) -> Result<SmtpTransport<S>> {
38    // We always read the greeting manually to unify
39    // the cases of STARTTLS where the greeting is
40    // sent outside the encrypted channel and implicit TLS
41    // where the greeting is sent after establishing TLS channel.
42    let client = SmtpClient::new().smtp_utf8(true).without_greeting();
43
44    let transport = SmtpTransport::new(client, stream)
45        .await
46        .context("Failed to send EHLO command")?;
47    Ok(transport)
48}
49
50#[expect(clippy::too_many_arguments)]
51pub(crate) async fn connect_and_auth(
52    context: &Context,
53    proxy_config: &Option<ProxyConfig>,
54    strict_tls: bool,
55    candidate: ConnectionCandidate,
56    oauth2: bool,
57    addr: &str,
58    user: &str,
59    password: &str,
60) -> Result<SmtpTransport<Box<dyn SessionBufStream>>> {
61    let session_stream = connect_stream(context, proxy_config.clone(), strict_tls, candidate)
62        .await
63        .context("SMTP failed to connect")?;
64    let mut transport = new_smtp_transport(session_stream).await?;
65
66    // Authenticate.
67    let (creds, mechanism) = if oauth2 {
68        // oauth2
69        let access_token = get_oauth2_access_token(context, addr, password, false)
70            .await
71            .context("SMTP failed to get OAUTH2 access token")?;
72        if access_token.is_none() {
73            bail!("SMTP OAuth 2 error {addr}");
74        }
75        (
76            async_smtp::authentication::Credentials::new(
77                user.to_string(),
78                access_token.unwrap_or_default(),
79            ),
80            vec![async_smtp::authentication::Mechanism::Xoauth2],
81        )
82    } else {
83        // plain
84        (
85            async_smtp::authentication::Credentials::new(user.to_string(), password.to_string()),
86            vec![
87                async_smtp::authentication::Mechanism::Plain,
88                async_smtp::authentication::Mechanism::Login,
89            ],
90        )
91    };
92    transport
93        .try_login(&creds, &mechanism)
94        .await
95        .context("SMTP failed to login")?;
96    Ok(transport)
97}
98
99async fn connection_attempt(
100    context: Context,
101    host: String,
102    security: ConnectionSecurity,
103    resolved_addr: SocketAddr,
104    strict_tls: bool,
105) -> Result<Box<dyn SessionBufStream>> {
106    let context = &context;
107    let host = &host;
108    info!(
109        context,
110        "Attempting SMTP connection to {host} ({resolved_addr})."
111    );
112    let res = match security {
113        ConnectionSecurity::Tls => {
114            connect_secure(resolved_addr, host, strict_tls, &context.tls_session_store).await
115        }
116        ConnectionSecurity::Starttls => {
117            connect_starttls(resolved_addr, host, strict_tls, &context.tls_session_store).await
118        }
119        ConnectionSecurity::Plain => connect_insecure(resolved_addr).await,
120    };
121    match res {
122        Ok(stream) => {
123            let ip_addr = resolved_addr.ip().to_string();
124            let port = resolved_addr.port();
125
126            let save_cache = match security {
127                ConnectionSecurity::Tls | ConnectionSecurity::Starttls => strict_tls,
128                ConnectionSecurity::Plain => false,
129            };
130            if save_cache {
131                update_connect_timestamp(context, host, &ip_addr).await?;
132            }
133            update_connection_history(context, "smtp", host, port, &ip_addr, time()).await?;
134            Ok(stream)
135        }
136        Err(err) => {
137            warn!(
138                context,
139                "Failed to connect to {host} ({resolved_addr}): {err:#}."
140            );
141            Err(err)
142        }
143    }
144}
145
146/// Returns TLS, STARTTLS or plaintext connection
147/// using SOCKS5 or direct connection depending on the given configuration.
148///
149/// Connection is returned after skipping the welcome message
150/// and is ready for sending commands. Because SMTP STARTTLS
151/// does not send welcome message over TLS connection
152/// after establishing it, welcome message is always ignored
153/// to unify the result regardless of whether TLS or STARTTLS is used.
154async fn connect_stream(
155    context: &Context,
156    proxy_config: Option<ProxyConfig>,
157    strict_tls: bool,
158    candidate: ConnectionCandidate,
159) -> Result<Box<dyn SessionBufStream>> {
160    let host = &candidate.host;
161    let port = candidate.port;
162    let security = candidate.security;
163
164    if let Some(proxy_config) = proxy_config {
165        let stream = match security {
166            ConnectionSecurity::Tls => {
167                connect_secure_proxy(context, host, port, strict_tls, proxy_config.clone()).await?
168            }
169            ConnectionSecurity::Starttls => {
170                connect_starttls_proxy(context, host, port, strict_tls, proxy_config.clone())
171                    .await?
172            }
173            ConnectionSecurity::Plain => {
174                connect_insecure_proxy(context, host, port, proxy_config.clone()).await?
175            }
176        };
177        update_connection_history(context, "smtp", host, port, host, time()).await?;
178        Ok(stream)
179    } else {
180        let load_cache = match security {
181            ConnectionSecurity::Tls | ConnectionSecurity::Starttls => strict_tls,
182            ConnectionSecurity::Plain => false,
183        };
184
185        let connection_futures = lookup_host_with_cache(context, host, port, "smtp", load_cache)
186            .await?
187            .into_iter()
188            .map(|resolved_addr| {
189                let context = context.clone();
190                let host = host.to_string();
191                connection_attempt(context, host, security, resolved_addr, strict_tls)
192            });
193        run_connection_attempts(connection_futures).await
194    }
195}
196
197/// Reads and ignores SMTP greeting.
198///
199/// This function is used to unify
200/// TLS, STARTTLS and plaintext connection setup
201/// by skipping the greeting in case of TLS
202/// and STARTTLS connection setup.
203async fn skip_smtp_greeting<R: tokio::io::AsyncBufReadExt + Unpin>(stream: &mut R) -> Result<()> {
204    let mut line = String::with_capacity(512);
205    loop {
206        line.clear();
207        let read = stream
208            .read_line(&mut line)
209            .await
210            .context("Failed to read from stream while waiting for SMTP greeting")?;
211        if read == 0 {
212            bail!("Unexpected EOF while reading SMTP greeting");
213        }
214        if line.starts_with("220-") {
215            continue;
216        } else if line.starts_with("220 ") {
217            return Ok(());
218        } else {
219            bail!("Unexpected greeting: {line:?}");
220        }
221    }
222}
223
224async fn connect_secure_proxy(
225    context: &Context,
226    hostname: &str,
227    port: u16,
228    strict_tls: bool,
229    proxy_config: ProxyConfig,
230) -> Result<Box<dyn SessionBufStream>> {
231    let proxy_stream = proxy_config
232        .connect(context, hostname, port, strict_tls)
233        .await?;
234    let tls_stream = wrap_tls(
235        strict_tls,
236        hostname,
237        port,
238        alpn(port),
239        proxy_stream,
240        &context.tls_session_store,
241    )
242    .await?;
243    let mut buffered_stream = BufStream::new(tls_stream);
244    skip_smtp_greeting(&mut buffered_stream).await?;
245    let session_stream: Box<dyn SessionBufStream> = Box::new(buffered_stream);
246    Ok(session_stream)
247}
248
249async fn connect_starttls_proxy(
250    context: &Context,
251    hostname: &str,
252    port: u16,
253    strict_tls: bool,
254    proxy_config: ProxyConfig,
255) -> Result<Box<dyn SessionBufStream>> {
256    let proxy_stream = proxy_config
257        .connect(context, hostname, port, strict_tls)
258        .await?;
259
260    // Run STARTTLS command and convert the client back into a stream.
261    let mut buffered_stream = BufStream::new(proxy_stream);
262    skip_smtp_greeting(&mut buffered_stream).await?;
263    let transport = new_smtp_transport(buffered_stream).await?;
264    let tcp_stream = transport.starttls().await?.into_inner();
265    let tls_stream = wrap_tls(
266        strict_tls,
267        hostname,
268        port,
269        "",
270        tcp_stream,
271        &context.tls_session_store,
272    )
273    .await
274    .context("STARTTLS upgrade failed")?;
275    let buffered_stream = BufStream::new(tls_stream);
276    let session_stream: Box<dyn SessionBufStream> = Box::new(buffered_stream);
277    Ok(session_stream)
278}
279
280async fn connect_insecure_proxy(
281    context: &Context,
282    hostname: &str,
283    port: u16,
284    proxy_config: ProxyConfig,
285) -> Result<Box<dyn SessionBufStream>> {
286    let proxy_stream = proxy_config.connect(context, hostname, port, false).await?;
287    let mut buffered_stream = BufStream::new(proxy_stream);
288    skip_smtp_greeting(&mut buffered_stream).await?;
289    let session_stream: Box<dyn SessionBufStream> = Box::new(buffered_stream);
290    Ok(session_stream)
291}
292
293async fn connect_secure(
294    addr: SocketAddr,
295    hostname: &str,
296    strict_tls: bool,
297    tls_session_store: &TlsSessionStore,
298) -> Result<Box<dyn SessionBufStream>> {
299    let tls_stream = connect_tls_inner(
300        addr,
301        hostname,
302        strict_tls,
303        alpn(addr.port()),
304        tls_session_store,
305    )
306    .await?;
307    let mut buffered_stream = BufStream::new(tls_stream);
308    skip_smtp_greeting(&mut buffered_stream).await?;
309    let session_stream: Box<dyn SessionBufStream> = Box::new(buffered_stream);
310    Ok(session_stream)
311}
312
313async fn connect_starttls(
314    addr: SocketAddr,
315    host: &str,
316    strict_tls: bool,
317    tls_session_store: &TlsSessionStore,
318) -> Result<Box<dyn SessionBufStream>> {
319    let tcp_stream = connect_tcp_inner(addr).await?;
320
321    // Run STARTTLS command and convert the client back into a stream.
322    let mut buffered_stream = BufStream::new(tcp_stream);
323    skip_smtp_greeting(&mut buffered_stream).await?;
324    let transport = new_smtp_transport(buffered_stream).await?;
325    let tcp_stream = transport.starttls().await?.into_inner();
326    let tls_stream = wrap_tls(
327        strict_tls,
328        host,
329        addr.port(),
330        "",
331        tcp_stream,
332        tls_session_store,
333    )
334    .await
335    .context("STARTTLS upgrade failed")?;
336
337    let buffered_stream = BufStream::new(tls_stream);
338    let session_stream: Box<dyn SessionBufStream> = Box::new(buffered_stream);
339    Ok(session_stream)
340}
341
342async fn connect_insecure(addr: SocketAddr) -> Result<Box<dyn SessionBufStream>> {
343    let tcp_stream = connect_tcp_inner(addr).await?;
344    let mut buffered_stream = BufStream::new(tcp_stream);
345    skip_smtp_greeting(&mut buffered_stream).await?;
346    let session_stream: Box<dyn SessionBufStream> = Box::new(buffered_stream);
347    Ok(session_stream)
348}
349
350#[cfg(test)]
351mod tests {
352    use tokio::io::BufReader;
353
354    use super::*;
355
356    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
357    async fn test_skip_smtp_greeting() -> Result<()> {
358        let greeting = b"220-server261.web-hosting.com ESMTP Exim 4.96.2 #2 Sat, 24 Aug 2024 12:25:53 -0400 \r\n\
359                         220-We do not authorize the use of this system to transport unsolicited,\r\n\
360                         220 and/or bulk e-mail.\r\n";
361        let mut buffered_stream = BufReader::new(&greeting[..]);
362        skip_smtp_greeting(&mut buffered_stream).await
363    }
364}
deltachat/smtp/connect.rs

deltachat/smtp/
connect.rs
