daaki_smtp/connection/

sending.rs

//! SMTP sending: MAIL FROM / RCPT TO / DATA sequence.
//!
//! RFC 5321 Section 3.3 (SMTP mail transaction), RFC 1854 (pipelining).

#[allow(clippy::wildcard_imports)]
use super::*;

impl SmtpConnection {
    // -----------------------------------------------------------------------
    // Sending — SMTP (RFC 5321)
    // -----------------------------------------------------------------------

    /// Send a message to one or more recipients.
    ///
    /// Performs the full MAIL FROM / RCPT TO / DATA sequence (RFC 5321
    /// Section 3.3). When the server advertises PIPELINING (RFC 1854),
    /// commands are batched for better throughput.
    ///
    /// Returns a [`SendResult`](crate::types::SendResult) containing any rejected recipients
    /// (RFC 5321 Section 3.3). When some RCPT TO commands are rejected
    /// but at least one succeeds, the message is delivered to the accepted
    /// recipients and the rejected ones are listed in
    /// [`SendResult::rejected_recipients`](crate::types::SendResult::rejected_recipients).
    ///
    /// BCC recipients should be included in `recipients` but must NOT appear
    /// in the message headers — that is the caller's responsibility.
    ///
    /// Addresses are pre-validated by their type constructors
    /// ([`ReversePath::new`], [`ForwardPath::new`]) per RFC 5321
    /// Section 4.1.2.
    ///
    /// `timeout` applies to the overall send operation. Per RFC 5321 Section
    /// 4.5.3.2, the recommended DATA timeout is 600 seconds.
    pub async fn send(
        &self,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        timeout: Duration,
    ) -> Result<crate::types::SendResult, Error> {
        self.send_with_params(from, recipients, message, None, timeout)
            .await
    }

    /// Send a message with extended MAIL FROM parameters.
    ///
    /// Like [`send`](Self::send) but accepts optional [`MailFromParams`](crate::types::MailFromParams)
    /// to include ESMTP parameters such as `BODY=` (RFC 1652 Section 3,
    /// RFC 3030 Section 2) and `SMTPUTF8` (RFC 6531 Section 3.4).
    ///
    /// Returns a [`SendResult`](crate::types::SendResult) containing any rejected recipients
    /// (RFC 5321 Section 3.3). When some RCPT TO commands are rejected
    /// but at least one succeeds, the message is delivered to the accepted
    /// recipients and the rejected ones are listed in
    /// [`SendResult::rejected_recipients`](crate::types::SendResult::rejected_recipients).
    ///
    /// Addresses are pre-validated by their type constructors
    /// ([`ReversePath::new`], [`ForwardPath::new`]) per RFC 5321
    /// Section 4.1.2.
    ///
    /// The SIZE parameter is always included automatically when the
    /// server advertises the SIZE extension (RFC 1870 Section 3).
    #[allow(clippy::significant_drop_tightening)]
    pub async fn send_with_params(
        &self,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        params: Option<&crate::types::MailFromParams>,
        timeout: Duration,
    ) -> Result<crate::types::SendResult, Error> {
        // RFC 2033 Section 4.2: LMTP returns one response per accepted
        // recipient after DATA, not a single aggregate response like SMTP.
        // send/send_with_params reads only one DATA response; using it on
        // an LMTP connection would leave per-recipient responses in the
        // buffer, corrupting subsequent operations. LMTP connections must
        // use send_lmtp() or send_lmtp_bdat() instead.
        if self.protocol == Protocol::Lmtp {
            return Err(Error::Protocol(
                "send/send_with_params does not support LMTP per-recipient \
                 responses (RFC 2033 Section 4.2); LMTP connections must use \
                 send_lmtp or send_lmtp_bdat"
                    .into(),
            ));
        }
        let mut inner = self.inner.lock().await;
        // RFC 5321 Section 3.8: after a 421 response the server will close
        // the transmission channel. Fail immediately instead of writing to
        // a doomed connection.
        if inner.server_shutting_down {
            return Err(Error::Protocol(
                "connection is shutting down after 421 (RFC 5321 Section 3.8)".into(),
            ));
        }
        let effective_mail_params = Some(Self::effective_mail_from_params(
            &inner.capabilities,
            from,
            recipients,
            message,
            params,
        )?);
        Self::validate_send_addresses(from, recipients)?;
        let message_size = Self::validate_data_prerequisites(
            &inner.capabilities,
            message,
            effective_mail_params.as_ref(),
            inner.stream.is_tls(),
        )?;
        tokio::time::timeout(timeout, async {
            if inner.capabilities.supports_pipelining() {
                Self::send_pipelined(
                    &mut inner,
                    from,
                    recipients,
                    message,
                    message_size,
                    effective_mail_params.as_ref(),
                    None,
                )
                .await
            } else {
                Self::send_sequential(
                    &mut inner,
                    from,
                    recipients,
                    message,
                    message_size,
                    effective_mail_params.as_ref(),
                    None,
                )
                .await
            }
        })
        .await
        .map_err(|_| Error::Timeout)?
    }

    /// Send a message with both MAIL FROM and per-recipient RCPT TO parameters.
    ///
    /// Like [`send_with_params`](Self::send_with_params) but also accepts
    /// per-recipient [`RcptToParams`](crate::types::RcptToParams) to include DSN parameters (NOTIFY,
    /// ORCPT) on each RCPT TO command (RFC 3461 Sections 4.1–4.2).
    ///
    /// Returns a [`SendResult`](crate::types::SendResult) containing any rejected recipients
    /// (RFC 5321 Section 3.3).
    ///
    /// `rcpt_params` must have the same length as `recipients` — each
    /// entry is paired by index. Returns `Error::Protocol` if the
    /// lengths do not match.
    ///
    /// Addresses are pre-validated by their type constructors
    /// ([`ReversePath::new`], [`ForwardPath::new`]) per RFC 5321
    /// Section 4.1.2.
    ///
    /// The SIZE parameter is always included automatically when the
    /// server advertises the SIZE extension (RFC 1870 Section 3).
    #[allow(clippy::significant_drop_tightening)]
    pub async fn send_with_all_params(
        &self,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        mail_params: Option<&crate::types::MailFromParams>,
        rcpt_params: &[crate::types::RcptToParams],
        timeout: Duration,
    ) -> Result<crate::types::SendResult, Error> {
        // RFC 3461 Sections 4.1–4.2: each recipient must have a
        // corresponding RcptToParams entry.
        if recipients.len() != rcpt_params.len() {
            return Err(Error::Protocol(format!(
                "rcpt_params length ({}) must match recipients length ({}) \
                 (RFC 3461 Sections 4.1–4.2)",
                rcpt_params.len(),
                recipients.len(),
            )));
        }
        // RFC 2033 Section 4.2: LMTP returns per-recipient responses;
        // use send_lmtp_with_all_params instead.
        if self.protocol == Protocol::Lmtp {
            return Err(Error::Protocol(
                "send_with_all_params does not support LMTP per-recipient \
                 responses (RFC 2033 Section 4.2); LMTP connections must use \
                 send_lmtp_with_all_params"
                    .into(),
            ));
        }
        let mut inner = self.inner.lock().await;
        // RFC 5321 Section 3.8: after a 421 response the server will close
        // the transmission channel. Fail immediately.
        if inner.server_shutting_down {
            return Err(Error::Protocol(
                "connection is shutting down after 421 (RFC 5321 Section 3.8)".into(),
            ));
        }
        let effective_mail_params = Some(Self::effective_mail_from_params(
            &inner.capabilities,
            from,
            recipients,
            message,
            mail_params,
        )?);
        Self::validate_send_addresses(from, recipients)?;
        Self::validate_rcpt_params(&inner.capabilities, rcpt_params)?;
        let message_size = Self::validate_data_prerequisites(
            &inner.capabilities,
            message,
            effective_mail_params.as_ref(),
            inner.stream.is_tls(),
        )?;
        tokio::time::timeout(timeout, async {
            if inner.capabilities.supports_pipelining() {
                Self::send_pipelined(
                    &mut inner,
                    from,
                    recipients,
                    message,
                    message_size,
                    effective_mail_params.as_ref(),
                    Some(rcpt_params),
                )
                .await
            } else {
                Self::send_sequential(
                    &mut inner,
                    from,
                    recipients,
                    message,
                    message_size,
                    effective_mail_params.as_ref(),
                    Some(rcpt_params),
                )
                .await
            }
        })
        .await
        .map_err(|_| Error::Timeout)?
    }

    /// Sequential send — issues each command and waits for its response
    /// before sending the next (RFC 5321 Section 3.3).
    ///
    /// Returns a [`SendResult`] containing any rejected recipients so callers
    /// can see which addresses were refused (RFC 5321 Section 3.3).
    ///
    /// When `rcpt_params` is `Some`, per-recipient DSN parameters are
    /// included on each RCPT TO command (RFC 3461 Sections 4.1–4.2).
    async fn send_sequential(
        inner: &mut SmtpInner,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        message_size: usize,
        params: Option<&crate::types::MailFromParams>,
        rcpt_params: Option<&[crate::types::RcptToParams]>,
    ) -> Result<crate::types::SendResult, Error> {
        Self::send_mail_from(inner, from, message, message_size, params).await?;
        let (_accepted, rejected) =
            Self::send_rcpt_to_batch(inner, recipients, rcpt_params).await?;

        Self::send_data_body(inner, message).await?;
        // SMTP: single response after the terminator (RFC 5321 Section 3.3).
        let resp = inner.read_response().await?;
        if !resp.is_success() {
            return Err(Self::response_to_error(resp));
        }

        Ok(crate::types::SendResult {
            rejected_recipients: rejected,
        })
    }

    /// Pipelined send — batches MAIL FROM, all RCPT TOs, and DATA into a
    /// single write, then reads responses in order (RFC 1854 Section 3).
    ///
    /// Returns a [`SendResult`] containing any rejected recipients so callers
    /// can see which addresses were refused (RFC 5321 Section 3.3).
    ///
    /// When `rcpt_params` is `Some`, per-recipient DSN parameters are
    /// included on each RCPT TO command (RFC 3461 Sections 4.1–4.2).
    #[allow(clippy::too_many_lines)]
    async fn send_pipelined(
        inner: &mut SmtpInner,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        message_size: usize,
        params: Option<&crate::types::MailFromParams>,
        rcpt_params: Option<&[crate::types::RcptToParams]>,
    ) -> Result<crate::types::SendResult, Error> {
        // Build the pipeline: MAIL FROM + RCPT TO(s) + DATA.
        let mut buf = BytesMut::new();
        let is_8bit = Self::message_contains_8bit(message);
        Self::encode_mail_from_cmd(
            &inner.capabilities,
            &mut buf,
            from,
            message_size,
            params,
            is_8bit,
        )?;
        // RFC 5321 Section 4.5.3.1.4 / RFC 1870 Section 4 / RFC 6152
        // Section 7 / RFC 6531 Section 3.4: MAIL FROM has an extended
        // command line limit to accommodate ESMTP parameters.
        Self::validate_mail_from_line_length(buf.len())?;
        for (i, fp) in recipients.iter().enumerate() {
            let start = buf.len();
            if let Some(rp) = rcpt_params {
                // RFC 3461 Sections 4.1–4.2: encode with DSN parameters.
                encode::encode_rcpt_to_full(&mut buf, fp, &rp[i])?;
            } else {
                encode::encode_rcpt_to(&mut buf, fp)?;
            }
            // RFC 5321 Section 4.5.3.1.4 / RFC 3461 Section 5: validate
            // each RCPT TO line, using the extended 1012-octet limit when
            // DSN parameters are present.
            let has_dsn = rcpt_params.is_some_and(|rp| !rp[i].is_empty());
            Self::validate_rcpt_to_line_length(buf.len() - start, has_dsn)?;
        }
        encode::encode_data(&mut buf);

        // Send all commands at once (RFC 1854 Section 3).
        inner.write_all(&buf).await?;

        // Read responses: 1 MAIL FROM + N RCPT TOs + 1 DATA.
        // MAIL FROM response (RFC 5321 Section 4.1.1.2).
        let mail_resp = inner.read_response().await?;
        if !mail_resp.is_success() {
            tracing::debug!(code = mail_resp.code, "pipelined MAIL FROM rejected");
            // Drain remaining responses: N RCPT TOs + 1 DATA.
            // RFC 1854 Section 3 / RFC 5321 Section 3.3: the server may
            // have already processed the pipelined DATA command and sent
            // a 354 intermediate reply. If so, it is waiting for message
            // data. We must send the dot terminator to exit the DATA
            // state, otherwise the session is left in an inconsistent
            // state where the server expects data and the client expects
            // command responses.
            // Track the DATA response specifically — it's the last of
            // the N+1 responses (N RCPT TOs + 1 DATA). We identify it by
            // index rather than using the last successfully read response,
            // because a mid-drain read failure would leave `last_resp`
            // pointing at an RCPT TO response, not the DATA response
            // (RFC 1854 Section 3).
            let drain_count = recipients.len() + 1;
            let mut data_resp: Option<SmtpResponse> = None;
            for i in 0..drain_count {
                match inner.read_response().await {
                    Ok(resp) => {
                        // The DATA response is the last one
                        // (index = drain_count - 1).
                        if i == drain_count - 1 {
                            data_resp = Some(resp);
                        }
                    }
                    Err(_) => break,
                }
            }
            // RFC 5321 Section 3.3: if the DATA command received 354
            // ("Start mail input"), we must send the dot terminator
            // to exit DATA state.
            // RFC 5321 Section 4.1.1.4: 354 is the only valid
            // intermediate response to DATA.
            if let Some(ref data_resp) = data_resp {
                if data_resp.code == 354 {
                    buf.clear();
                    encode::encode_data_end(&mut buf, b"");
                    let _ = inner.write_all(&buf).await;
                    // Read and discard the response to the empty DATA.
                    let _ = inner.read_response().await;
                }
            }
            return Err(Self::response_to_error(mail_resp));
        }

        // RCPT TO responses (RFC 5321 Section 4.1.1.3).
        let mut accepted = 0usize;
        let mut rejected_recipients = Vec::new();
        for fp in recipients {
            let resp = inner.read_response().await?;
            if resp.is_success() {
                accepted += 1;
            } else {
                tracing::debug!(
                    recipient = fp.as_str(),
                    code = resp.code,
                    "pipelined RCPT TO rejected"
                );
                rejected_recipients.push(crate::types::RejectedRecipient {
                    recipient: fp.clone(),
                    response: resp,
                });
            }
        }

        // DATA response (RFC 5321 Section 4.1.1.4).
        let data_resp = inner.read_response().await?;

        if accepted == 0 {
            // All RCPT TOs failed — send terminator to exit DATA state
            // per RFC 1854 Section 3, then return error.
            // RFC 5321 Section 4.1.1.4: 354 is the only valid
            // intermediate response to DATA.
            if data_resp.code == 354 {
                buf.clear();
                // No message data was sent — pass empty slice so the leading
                // CRLF is included (RFC 5321 Section 4.1.1.4).
                encode::encode_data_end(&mut buf, b"");
                // Best-effort write — the server may have already
                // closed the connection after rejecting all recipients
                // (RFC 1854 Section 3). Propagating the I/O error here
                // would shadow the AllRecipientsFailed error below.
                let _ = inner.write_all(&buf).await;
                // Read and discard the response to the empty DATA.
                let _ = inner.read_response().await;
            } else {
                // RFC 5321 Section 3.3: DATA was rejected but the mail
                // transaction (MAIL FROM) is still open. RSET to clean up.
                inner.rset_best_effort().await;
            }
            return Err(Error::AllRecipientsFailed {
                count: recipients.len(),
                // Extract just the responses for the error variant.
                responses: rejected_recipients
                    .into_iter()
                    .map(|r| r.response)
                    .collect(),
            });
        }

        // RFC 5321 Section 4.1.1.4: 354 is the only valid intermediate
        // response to DATA. Any other code means DATA was rejected.
        if data_resp.code != 354 {
            // RFC 5321 Section 3.3: DATA was rejected but the mail
            // transaction (MAIL FROM) is still open. RSET to clean up.
            inner.rset_best_effort().await;
            return Err(Self::response_to_error(data_resp));
        }

        // Send dot-stuffed message body + terminator in a single write
        // (RFC 5321 Section 4.5.2 / Section 4.1.1.4).
        let body = encode::dot_stuff_and_terminate(message);
        inner.write_all(&body).await?;
        let resp = inner.read_response().await?;
        if !resp.is_success() {
            return Err(Self::response_to_error(resp));
        }

        Ok(crate::types::SendResult {
            rejected_recipients,
        })
    }
}