deltachat/
smtp.rs

1//! # SMTP transport module.
2
3mod connect;
4pub mod send;
5
6use anyhow::{Context as _, Error, Result, bail, format_err};
7use async_smtp::response::{Category, Code, Detail};
8use async_smtp::{EmailAddress, SmtpTransport};
9use tokio::task;
10
11use crate::chat::{ChatId, add_info_msg_with_cmd};
12use crate::config::Config;
13use crate::contact::{Contact, ContactId};
14use crate::context::Context;
15use crate::events::EventType;
16use crate::log::warn;
17use crate::message::Message;
18use crate::message::{self, MsgId};
19use crate::mimefactory::MimeFactory;
20use crate::net::proxy::ProxyConfig;
21use crate::net::session::SessionBufStream;
22use crate::scheduler::connectivity::ConnectivityStore;
23use crate::stock_str::unencrypted_email;
24use crate::tools::{self, time_elapsed};
25use crate::transport::{
26    ConfiguredLoginParam, ConfiguredServerLoginParam, prioritize_server_login_params,
27};
28
29#[derive(Default)]
30pub(crate) struct Smtp {
31    /// SMTP connection.
32    transport: Option<SmtpTransport<Box<dyn SessionBufStream>>>,
33
34    /// Email address we are sending from.
35    from: Option<EmailAddress>,
36
37    /// Timestamp of last successful send/receive network interaction
38    /// (eg connect or send succeeded). On initialization and disconnect
39    /// it is set to None.
40    last_success: Option<tools::Time>,
41
42    pub(crate) connectivity: ConnectivityStore,
43
44    /// If sending the last message failed, contains the error message.
45    pub(crate) last_send_error: Option<String>,
46}
47
48impl Smtp {
49    /// Create a new Smtp instances.
50    pub fn new() -> Self {
51        Default::default()
52    }
53
54    /// Disconnect the SMTP transport and drop it entirely.
55    pub fn disconnect(&mut self) {
56        if let Some(mut transport) = self.transport.take() {
57            // Closing connection with a QUIT command may take some time, especially if it's a
58            // stale connection and an attempt to send the command times out. Send a command in a
59            // separate task to avoid waiting for reply or timeout.
60            task::spawn(async move { transport.quit().await });
61        }
62        self.last_success = None;
63    }
64
65    /// Return true if smtp was connected but is not known to
66    /// have been successfully used the last 60 seconds
67    pub fn has_maybe_stale_connection(&self) -> bool {
68        if let Some(last_success) = self.last_success {
69            time_elapsed(&last_success).as_secs() > 60
70        } else {
71            false
72        }
73    }
74
75    /// Check whether we are connected.
76    pub fn is_connected(&self) -> bool {
77        self.transport.is_some()
78    }
79
80    /// Connect using configured parameters.
81    pub async fn connect_configured(&mut self, context: &Context) -> Result<()> {
82        if self.has_maybe_stale_connection() {
83            info!(context, "Closing stale connection.");
84            self.disconnect();
85        }
86
87        if self.is_connected() {
88            return Ok(());
89        }
90
91        self.connectivity.set_connecting(context);
92        let lp = ConfiguredLoginParam::load(context)
93            .await?
94            .context("Not configured")?;
95        let proxy_config = ProxyConfig::load(context).await?;
96        self.connect(
97            context,
98            &lp.smtp,
99            &lp.smtp_password,
100            &proxy_config,
101            &lp.addr,
102            lp.strict_tls(proxy_config.is_some()),
103            lp.oauth2,
104        )
105        .await
106    }
107
108    /// Connect using the provided login params.
109    #[expect(clippy::too_many_arguments)]
110    pub async fn connect(
111        &mut self,
112        context: &Context,
113        login_params: &[ConfiguredServerLoginParam],
114        password: &str,
115        proxy_config: &Option<ProxyConfig>,
116        addr: &str,
117        strict_tls: bool,
118        oauth2: bool,
119    ) -> Result<()> {
120        if self.is_connected() {
121            warn!(context, "SMTP already connected.");
122            return Ok(());
123        }
124
125        let from = EmailAddress::new(addr.to_string())
126            .with_context(|| format!("Invalid address {addr:?}"))?;
127        self.from = Some(from);
128
129        let login_params =
130            prioritize_server_login_params(&context.sql, login_params, "smtp").await?;
131        let mut first_error = None;
132        for lp in login_params {
133            info!(context, "SMTP trying to connect to {}.", &lp.connection);
134            let transport = match connect::connect_and_auth(
135                context,
136                proxy_config,
137                strict_tls,
138                lp.connection.clone(),
139                oauth2,
140                addr,
141                &lp.user,
142                password,
143            )
144            .await
145            {
146                Ok(transport) => transport,
147                Err(err) => {
148                    warn!(context, "SMTP failed to connect and authenticate: {err:#}.");
149                    first_error.get_or_insert(err);
150                    continue;
151                }
152            };
153
154            self.transport = Some(transport);
155            self.last_success = Some(tools::Time::now());
156
157            context.emit_event(EventType::SmtpConnected(format!(
158                "SMTP-LOGIN as {} ok",
159                lp.user,
160            )));
161            return Ok(());
162        }
163
164        Err(first_error.unwrap_or_else(|| format_err!("No SMTP connection candidates provided")))
165    }
166}
167
168pub(crate) enum SendResult {
169    /// Message was sent successfully.
170    Success,
171
172    /// Permanent error, message sending has failed.
173    Failure(Error),
174
175    /// Temporary error, the message should be retried later.
176    Retry,
177}
178
179/// Tries to send a message.
180pub(crate) async fn smtp_send(
181    context: &Context,
182    recipients: &[async_smtp::EmailAddress],
183    message: &str,
184    smtp: &mut Smtp,
185    msg_id: Option<MsgId>,
186) -> SendResult {
187    if std::env::var(crate::DCC_MIME_DEBUG).is_ok() {
188        info!(context, "SMTP-sending out mime message:\n{message}");
189    }
190
191    smtp.connectivity.set_working(context);
192
193    if let Err(err) = smtp
194        .connect_configured(context)
195        .await
196        .context("Failed to open SMTP connection")
197    {
198        smtp.last_send_error = Some(format!("{err:#}"));
199        return SendResult::Retry;
200    }
201
202    let send_result = smtp.send(context, recipients, message.as_bytes()).await;
203    smtp.last_send_error = send_result.as_ref().err().map(|e| e.to_string());
204
205    let status = match send_result {
206        Err(crate::smtp::send::Error::SmtpSend(err)) => {
207            // Remote error, retry later.
208            info!(context, "SMTP failed to send: {:?}.", &err);
209
210            let res = match err {
211                async_smtp::error::Error::Permanent(ref response) => {
212                    // Workaround for incorrectly configured servers returning permanent errors
213                    // instead of temporary ones.
214                    let maybe_transient = match response.code {
215                        // Sometimes servers send a permanent error when actually it is a temporary error
216                        // For documentation see <https://tools.ietf.org/html/rfc3463>
217                        Code {
218                            category: Category::MailSystem,
219                            detail: Detail::Zero,
220                            ..
221                        } => {
222                            // Ignore status code 5.5.0, see <https://support.delta.chat/t/every-other-message-gets-stuck/877/2>
223                            // Maybe incorrectly configured Postfix milter with "reject" instead of "tempfail", which returns
224                            // "550 5.5.0 Service unavailable" instead of "451 4.7.1 Service unavailable - try again later".
225                            //
226                            // Other enhanced status codes, such as Postfix
227                            // "550 5.1.1 <foobar@example.org>: Recipient address rejected: User unknown in local recipient table"
228                            // are not ignored.
229                            response.first_word() == Some("5.5.0")
230                        }
231                        _ => false,
232                    };
233
234                    if maybe_transient {
235                        info!(
236                            context,
237                            "Permanent error that is likely to actually be transient, postponing retry for later."
238                        );
239                        SendResult::Retry
240                    } else {
241                        info!(context, "Permanent error, message sending failed.");
242                        // If we do not retry, add an info message to the chat.
243                        // Yandex error "554 5.7.1 [2] Message rejected under suspicion of SPAM; https://ya.cc/..."
244                        // should definitely go here, because user has to open the link to
245                        // resume message sending.
246                        SendResult::Failure(format_err!("Permanent SMTP error: {err}"))
247                    }
248                }
249                async_smtp::error::Error::Transient(ref response) => {
250                    // We got a transient 4xx response from SMTP server.
251                    // Give some time until the server-side error maybe goes away.
252                    //
253                    // One particular case is
254                    // `450 4.1.2 <alice@example.org>: Recipient address rejected: Domain not found`.
255                    // known to be returned by Postfix.
256                    //
257                    // [RFC 3463](https://tools.ietf.org/html/rfc3463#section-3.2)
258                    // says "This code is only useful for permanent failures."
259                    // in X.1.1, X.1.2 and X.1.3 descriptions.
260                    //
261                    // Previous Delta Chat core versions
262                    // from 1.51.0 to 1.151.1
263                    // were treating such errors as permanent.
264                    //
265                    // This was later reverted because such errors were observed
266                    // for existing domains and turned out to be actually transient,
267                    // likely caused by nameserver downtime.
268                    info!(
269                        context,
270                        "Transient error {response:?}, postponing retry for later."
271                    );
272                    SendResult::Retry
273                }
274                _ => {
275                    info!(
276                        context,
277                        "Message sending failed without error returned by the server, retry later."
278                    );
279                    SendResult::Retry
280                }
281            };
282
283            // this clears last_success info
284            info!(context, "Failed to send message over SMTP, disconnecting.");
285            smtp.disconnect();
286
287            res
288        }
289        Err(crate::smtp::send::Error::Envelope(err)) => {
290            // Local error, job is invalid, do not retry.
291            smtp.disconnect();
292            warn!(context, "SMTP job is invalid: {err:#}.");
293            SendResult::Failure(err)
294        }
295        Err(crate::smtp::send::Error::NoTransport) => {
296            // Should never happen.
297            // It does not even make sense to disconnect here.
298            error!(context, "SMTP job failed because SMTP has no transport.");
299            SendResult::Failure(format_err!("SMTP has not transport"))
300        }
301        Err(crate::smtp::send::Error::Other(err)) => {
302            // Local error, job is invalid, do not retry.
303            smtp.disconnect();
304            warn!(context, "Unable to load SMTP job: {err:#}.");
305            SendResult::Failure(err)
306        }
307        Ok(()) => SendResult::Success,
308    };
309
310    if let SendResult::Failure(err) = &status
311        && let Some(msg_id) = msg_id
312    {
313        // We couldn't send the message, so mark it as failed
314        match Message::load_from_db(context, msg_id).await {
315            Ok(mut msg) => {
316                if let Err(err) = message::set_msg_failed(context, &mut msg, &err.to_string()).await
317                {
318                    error!(context, "Failed to mark {msg_id} as failed: {err:#}.");
319                }
320            }
321            Err(err) => {
322                error!(
323                    context,
324                    "Failed to load {msg_id} to mark it as failed: {err:#}."
325                );
326            }
327        }
328    }
329    status
330}
331
332/// Sends message identified by `smtp` table rowid over SMTP connection.
333///
334/// Removes row if the message should not be retried, otherwise increments retry count.
335pub(crate) async fn send_msg_to_smtp(
336    context: &Context,
337    smtp: &mut Smtp,
338    rowid: i64,
339) -> anyhow::Result<()> {
340    if let Err(err) = smtp
341        .connect_configured(context)
342        .await
343        .context("SMTP connection failure")
344    {
345        smtp.last_send_error = Some(format!("{err:#}"));
346        return Err(err);
347    }
348
349    // Increase retry count as soon as we have an SMTP connection. This ensures that the message is
350    // eventually removed from the queue by exceeding retry limit even in case of an error that
351    // keeps happening early in the message sending code, e.g. failure to read the message from the
352    // database.
353    context
354        .sql
355        .execute("UPDATE smtp SET retries=retries+1 WHERE id=?", (rowid,))
356        .await
357        .context("failed to update retries count")?;
358
359    let Some((body, recipients, msg_id, retries)) = context
360        .sql
361        .query_row_optional(
362            "SELECT mime, recipients, msg_id, retries FROM smtp WHERE id=?",
363            (rowid,),
364            |row| {
365                let mime: String = row.get(0)?;
366                let recipients: String = row.get(1)?;
367                let msg_id: MsgId = row.get(2)?;
368                let retries: i64 = row.get(3)?;
369                Ok((mime, recipients, msg_id, retries))
370            },
371        )
372        .await?
373    else {
374        return Ok(());
375    };
376    if retries > 6 {
377        if let Some(mut msg) = Message::load_from_db_optional(context, msg_id).await? {
378            message::set_msg_failed(context, &mut msg, "Number of retries exceeded the limit.")
379                .await?;
380        }
381        context
382            .sql
383            .execute("DELETE FROM smtp WHERE id=?", (rowid,))
384            .await
385            .context("Failed to remove message with exceeded retry limit from smtp table")?;
386        return Ok(());
387    }
388    info!(
389        context,
390        "Try number {retries} to send message {msg_id} (entry {rowid}) over SMTP."
391    );
392
393    let recipients_list = recipients
394        .split(' ')
395        .filter_map(
396            |addr| match async_smtp::EmailAddress::new(addr.to_string()) {
397                Ok(addr) => Some(addr),
398                Err(err) => {
399                    warn!(context, "Invalid recipient: {} {:?}.", addr, err);
400                    None
401                }
402            },
403        )
404        .collect::<Vec<_>>();
405
406    let status = smtp_send(context, &recipients_list, body.as_str(), smtp, Some(msg_id)).await;
407
408    match status {
409        SendResult::Retry => {}
410        SendResult::Success => {
411            context
412                .sql
413                .execute("DELETE FROM smtp WHERE id=?", (rowid,))
414                .await?;
415        }
416        SendResult::Failure(ref err) => {
417            if err
418                .to_string()
419                .to_lowercase()
420                .contains("invalid unencrypted mail")
421            {
422                let res = context
423                    .sql
424                    .query_row_optional(
425                        "SELECT chat_id, timestamp FROM msgs WHERE id=?;",
426                        (msg_id,),
427                        |row| Ok((row.get::<_, ChatId>(0)?, row.get::<_, i64>(1)?)),
428                    )
429                    .await?;
430
431                if let Some((chat_id, timestamp_sort)) = res {
432                    let addr = context.get_config(Config::ConfiguredAddr).await?;
433                    let text = unencrypted_email(
434                        context,
435                        addr.unwrap_or_default()
436                            .split('@')
437                            .nth(1)
438                            .unwrap_or_default(),
439                    )
440                    .await;
441                    add_info_msg_with_cmd(
442                        context,
443                        chat_id,
444                        &text,
445                        crate::mimeparser::SystemMessage::InvalidUnencryptedMail,
446                        Some(timestamp_sort),
447                        timestamp_sort,
448                        None,
449                        None,
450                        None,
451                    )
452                    .await?;
453                };
454            }
455            context
456                .sql
457                .execute("DELETE FROM smtp WHERE id=?", (rowid,))
458                .await?;
459        }
460    };
461
462    match status {
463        SendResult::Retry => Err(format_err!("Retry")),
464        SendResult::Success => {
465            if !context
466                .sql
467                .exists("SELECT COUNT(*) FROM smtp WHERE msg_id=?", (msg_id,))
468                .await?
469            {
470                msg_id.set_delivered(context).await?;
471            }
472            Ok(())
473        }
474        SendResult::Failure(err) => Err(format_err!("{err}")),
475    }
476}
477
478/// Attempts to send queued MDNs.
479async fn send_mdns(context: &Context, connection: &mut Smtp) -> Result<()> {
480    loop {
481        if !context.ratelimit.read().await.can_send() {
482            info!(context, "Ratelimiter does not allow sending MDNs now.");
483            return Ok(());
484        }
485
486        let more_mdns = send_mdn(context, connection).await?;
487        if !more_mdns {
488            // No more MDNs to send or one of them failed.
489            return Ok(());
490        }
491    }
492}
493
494/// Tries to send all messages currently in `smtp`, `smtp_status_updates` and `smtp_mdns` tables.
495pub(crate) async fn send_smtp_messages(context: &Context, connection: &mut Smtp) -> Result<()> {
496    let ratelimited = if context.ratelimit.read().await.can_send() {
497        // add status updates and sync messages to end of sending queue
498        context.flush_status_updates().await?;
499        false
500    } else {
501        true
502    };
503
504    let rowids = context
505        .sql
506        .query_map_vec("SELECT id FROM smtp ORDER BY id ASC", (), |row| {
507            let rowid: i64 = row.get(0)?;
508            Ok(rowid)
509        })
510        .await?;
511
512    info!(context, "Selected rows from SMTP queue: {rowids:?}.");
513    for rowid in rowids {
514        send_msg_to_smtp(context, connection, rowid)
515            .await
516            .context("Failed to send message")?;
517    }
518
519    // although by slow sending, ratelimit may have been expired meanwhile,
520    // do not attempt to send MDNs if ratelimited happened before on status-updates/sync:
521    // instead, let the caller recall this function so that more important status-updates/sync are sent out.
522    if !ratelimited {
523        send_mdns(context, connection)
524            .await
525            .context("Failed to send MDNs")?;
526    }
527    Ok(())
528}
529
530/// Tries to send MDN for message identified by `rfc724_mdn` to `contact_id`.
531///
532/// Attempts to aggregate additional MDNs for `contact_id` into sent MDN.
533///
534/// On failure returns an error without removing any `smtp_mdns` entries, the caller is responsible
535/// for removing the corresponding entry to prevent endless loop in case the entry is invalid, e.g.
536/// points to non-existent message or contact.
537///
538/// Returns true on success, false on temporary error.
539async fn send_mdn_rfc724_mid(
540    context: &Context,
541    rfc724_mid: &str,
542    contact_id: ContactId,
543    smtp: &mut Smtp,
544) -> Result<bool> {
545    let contact = Contact::get_by_id(context, contact_id).await?;
546    if contact.is_blocked() {
547        return Err(format_err!("Contact is blocked"));
548    }
549
550    // Try to aggregate additional MDNs into this MDN.
551    let additional_rfc724_mids = context
552        .sql
553        .query_map_vec(
554            "SELECT rfc724_mid
555             FROM smtp_mdns
556             WHERE from_id=? AND rfc724_mid!=?",
557            (contact_id, &rfc724_mid),
558            |row| {
559                let rfc724_mid: String = row.get(0)?;
560                Ok(rfc724_mid)
561            },
562        )
563        .await?;
564
565    let mimefactory = MimeFactory::from_mdn(
566        context,
567        contact_id,
568        rfc724_mid.to_string(),
569        additional_rfc724_mids.clone(),
570    )
571    .await?;
572    let rendered_msg = mimefactory.render(context).await?;
573    let body = rendered_msg.message;
574
575    let addr = contact.get_addr();
576    let recipient = async_smtp::EmailAddress::new(addr.to_string())
577        .map_err(|err| format_err!("invalid recipient: {addr} {err:?}"))?;
578    let recipients = vec![recipient];
579
580    match smtp_send(context, &recipients, &body, smtp, None).await {
581        SendResult::Success => {
582            info!(context, "Successfully sent MDN for {rfc724_mid}.");
583            context
584                .sql
585                .transaction(|transaction| {
586                    let mut stmt =
587                        transaction.prepare("DELETE FROM smtp_mdns WHERE rfc724_mid = ?")?;
588                    stmt.execute((rfc724_mid,))?;
589                    for additional_rfc724_mid in additional_rfc724_mids {
590                        stmt.execute((additional_rfc724_mid,))?;
591                    }
592                    Ok(())
593                })
594                .await?;
595            Ok(true)
596        }
597        SendResult::Retry => {
598            info!(
599                context,
600                "Temporary SMTP failure while sending an MDN for {rfc724_mid}."
601            );
602            Ok(false)
603        }
604        SendResult::Failure(err) => Err(err),
605    }
606}
607
608/// Tries to send a single MDN. Returns true if more MDNs should be sent.
609async fn send_mdn(context: &Context, smtp: &mut Smtp) -> Result<bool> {
610    if !context.should_send_mdns().await? {
611        context.sql.execute("DELETE FROM smtp_mdns", []).await?;
612        return Ok(false);
613    }
614    info!(context, "Sending MDNs.");
615
616    context
617        .sql
618        .execute("DELETE FROM smtp_mdns WHERE retries > 6", [])
619        .await?;
620    let Some(msg_row) = context
621        .sql
622        .query_row_optional(
623            "SELECT rfc724_mid, from_id FROM smtp_mdns ORDER BY retries LIMIT 1",
624            [],
625            |row| {
626                let rfc724_mid: String = row.get(0)?;
627                let from_id: ContactId = row.get(1)?;
628                Ok((rfc724_mid, from_id))
629            },
630        )
631        .await?
632    else {
633        return Ok(false);
634    };
635    let (rfc724_mid, contact_id) = msg_row;
636
637    context
638        .sql
639        .execute(
640            "UPDATE smtp_mdns SET retries=retries+1 WHERE rfc724_mid=?",
641            (rfc724_mid.clone(),),
642        )
643        .await
644        .context("Failed to update MDN retries count")?;
645
646    match send_mdn_rfc724_mid(context, &rfc724_mid, contact_id, smtp).await {
647        Err(err) => {
648            // If there is an error, for example there is no message corresponding to the msg_id in the
649            // database, do not try to send this MDN again.
650            warn!(
651                context,
652                "Error sending MDN for {rfc724_mid}, removing it: {err:#}."
653            );
654            context
655                .sql
656                .execute("DELETE FROM smtp_mdns WHERE rfc724_mid = ?", (rfc724_mid,))
657                .await?;
658            Err(err)
659        }
660        Ok(false) => {
661            bail!("Temporary error while sending an MDN");
662        }
663        Ok(true) => {
664            // Successfully sent MDN.
665            Ok(true)
666        }
667    }
668}
deltachat/smtp.rs

deltachat/
smtp.rs
