daaki_smtp/connection/

lmtp.rs

//! LMTP sending and LMTP BDAT (RFC 2033 Section 4.2 + RFC 3030 Section 3).
//!
//! Per-recipient response handling for both DATA and BDAT paths.

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

impl SmtpConnection {
    // -----------------------------------------------------------------------
    // Sending — LMTP (RFC 2033)
    // -----------------------------------------------------------------------

    /// Send a message via LMTP and return per-recipient results.
    ///
    /// LMTP (RFC 2033 Section 4.2) differs from SMTP in that the server
    /// returns one response per accepted recipient after the DATA terminator,
    /// rather than a single aggregate response.
    ///
    /// Accepts optional [`MailFromParams`](crate::types::MailFromParams) to include ESMTP parameters
    /// such as `BODY=8BITMIME` (RFC 1652 Section 3) and `SMTPUTF8`
    /// (RFC 6531 Section 3.4). The SIZE parameter is always included
    /// automatically when the server advertises the SIZE extension
    /// (RFC 1870 Section 3).
    ///
    /// Addresses are pre-validated by their type constructors
    /// ([`ReversePath::new`], [`ForwardPath::new`]) per RFC 5321
    /// Section 4.1.2.
    ///
    /// Returns `Error::Protocol` if this connection is not LMTP.
    #[allow(clippy::significant_drop_tightening)]
    pub async fn send_lmtp(
        &self,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        params: Option<&crate::types::MailFromParams>,
        timeout: Duration,
    ) -> Result<crate::types::LmtpSendResult, Error> {
        if self.protocol != Protocol::Lmtp {
            return Err(Error::Protocol(
                "send_lmtp requires an LMTP connection (RFC 2033)".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,
            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 {
            Self::send_lmtp_inner(
                &mut inner,
                from,
                recipients,
                message,
                message_size,
                effective_mail_params.as_ref(),
                None,
            )
            .await
        })
        .await
        .map_err(|_| Error::Timeout)?
    }

    /// Send a message via LMTP with both MAIL FROM and per-recipient
    /// RCPT TO parameters, returning per-recipient results.
    ///
    /// Like [`send_lmtp`](Self::send_lmtp) 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).
    ///
    /// `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.
    ///
    /// Returns `Error::Protocol` if this connection is not LMTP.
    #[allow(clippy::significant_drop_tightening)]
    pub async fn send_lmtp_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::LmtpSendResult, 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(),
            )));
        }
        if self.protocol != Protocol::Lmtp {
            return Err(Error::Protocol(
                "send_lmtp_with_all_params requires an LMTP connection (RFC 2033)".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 {
            Self::send_lmtp_inner(
                &mut inner,
                from,
                recipients,
                message,
                message_size,
                effective_mail_params.as_ref(),
                Some(rcpt_params),
            )
            .await
        })
        .await
        .map_err(|_| Error::Timeout)?
    }

    /// Inner LMTP send logic (RFC 2033 Section 4.2).
    ///
    /// 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_lmtp_inner(
        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::LmtpSendResult, Error> {
        Self::send_mail_from(inner, from, message, message_size, params).await?;
        // RFC 2033 Section 4.2 / RFC 5321 Section 3.3: capture both accepted
        // and rejected recipients so callers have full visibility.
        let (accepted_recipients, rejected) =
            Self::send_rcpt_to_batch(inner, recipients, rcpt_params).await?;

        Self::send_data_body(inner, message).await?;
        // LMTP: one response per accepted recipient (RFC 2033 Section 4.2).
        let results = Self::collect_lmtp_results(inner, accepted_recipients).await?;
        Ok(crate::types::LmtpSendResult {
            results,
            rejected_recipients: rejected,
        })
    }

    // -----------------------------------------------------------------------
    // Sending — LMTP BDAT (RFC 3030 §3 + RFC 2033 §4.2)
    // -----------------------------------------------------------------------

    /// Send a message via LMTP using BDAT chunking and return per-recipient results.
    ///
    /// Combines LMTP per-recipient response handling (RFC 2033 Section 4.2)
    /// with BDAT chunking (RFC 3030 Section 3). Unlike DATA, BDAT does not
    /// require dot-stuffing, making it suitable for binary content.
    ///
    /// Addresses are pre-validated by their type constructors
    /// ([`ReversePath::new`], [`ForwardPath::new`]) per RFC 5321
    /// Section 4.1.2.
    ///
    /// The server must advertise CHUNKING (RFC 3030). Returns
    /// `Error::Protocol` if this is not an LMTP connection or if the server
    /// does not support CHUNKING.
    #[allow(clippy::significant_drop_tightening)]
    pub async fn send_lmtp_bdat(
        &self,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        params: Option<&crate::types::MailFromParams>,
        timeout: Duration,
    ) -> Result<crate::types::LmtpSendResult, Error> {
        // RFC 2033: LMTP connection required.
        if self.protocol != Protocol::Lmtp {
            return Err(Error::Protocol(
                "send_lmtp_bdat requires an LMTP connection (RFC 2033)".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,
            params,
        )?);
        Self::validate_send_addresses(from, recipients)?;
        Self::validate_bdat_prerequisites(
            &inner.capabilities,
            message,
            effective_mail_params.as_ref(),
            inner.stream.is_tls(),
        )?;
        tokio::time::timeout(timeout, async {
            Self::send_lmtp_bdat_inner(
                &mut inner,
                from,
                recipients,
                message,
                effective_mail_params.as_ref(),
                None,
            )
            .await
        })
        .await
        .map_err(|_| Error::Timeout)?
    }

    /// Send a message via LMTP using BDAT chunking with both MAIL FROM
    /// and per-recipient RCPT TO parameters, returning per-recipient results.
    ///
    /// Like [`send_lmtp_bdat`](Self::send_lmtp_bdat) 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).
    ///
    /// `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 server must advertise CHUNKING (RFC 3030). Returns
    /// `Error::Protocol` if this is not an LMTP connection or if the
    /// server does not support CHUNKING.
    #[allow(clippy::significant_drop_tightening)]
    pub async fn send_lmtp_bdat_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::LmtpSendResult, 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: LMTP connection required.
        if self.protocol != Protocol::Lmtp {
            return Err(Error::Protocol(
                "send_lmtp_bdat_with_all_params requires an LMTP connection (RFC 2033)".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)?;
        Self::validate_bdat_prerequisites(
            &inner.capabilities,
            message,
            effective_mail_params.as_ref(),
            inner.stream.is_tls(),
        )?;
        tokio::time::timeout(timeout, async {
            Self::send_lmtp_bdat_inner(
                &mut inner,
                from,
                recipients,
                message,
                effective_mail_params.as_ref(),
                Some(rcpt_params),
            )
            .await
        })
        .await
        .map_err(|_| Error::Timeout)?
    }

    /// Inner LMTP BDAT send logic (RFC 3030 Section 3 + RFC 2033 Section 4.2).
    ///
    /// 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_lmtp_bdat_inner(
        inner: &mut SmtpInner,
        from: &ReversePath,
        recipients: &[ForwardPath],
        message: &[u8],
        params: Option<&crate::types::MailFromParams>,
        rcpt_params: Option<&[crate::types::RcptToParams]>,
    ) -> Result<crate::types::LmtpSendResult, Error> {
        // RFC 2033 Section 4.2 / RFC 5321 Section 3.3: capture both accepted
        // and rejected recipients so callers have full visibility.
        let (accepted, rejected) =
            Self::send_bdat_envelope(inner, from, recipients, message, params, rcpt_params).await?;

        // LMTP: one response per accepted recipient after BDAT LAST
        // (RFC 2033 Section 4.2).
        let results = Self::collect_lmtp_results(inner, accepted).await?;
        Ok(crate::types::LmtpSendResult {
            results,
            rejected_recipients: rejected,
        })
    }

    /// Send the DATA command, dot-stuff the message body, and write the
    /// terminator (RFC 5321 Sections 4.1.1.4 / 4.5.2).
    ///
    /// On return the message body has been sent, but the final response(s)
    /// have NOT been read — the caller must read them (SMTP: one response;
    /// LMTP: one per accepted recipient per RFC 2033 Section 4.2).
    pub(super) async fn send_data_body(inner: &mut SmtpInner, message: &[u8]) -> Result<(), Error> {
        // DATA (RFC 5321 Section 4.1.1.4).
        let mut buf = BytesMut::new();
        encode::encode_data(&mut buf);
        inner.write_all(&buf).await?;
        let resp = inner.read_response().await?;
        // RFC 5321 Section 4.1.1.4: 354 is the only valid intermediate
        // response to DATA. Any other code means DATA was rejected.
        if 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(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?;
        Ok(())
    }

    /// Collect per-recipient LMTP responses (RFC 2033 Section 4.2).
    ///
    /// After DATA or BDAT LAST, LMTP returns one response per accepted
    /// recipient. This helper reads them all and pairs each with its
    /// recipient address.
    pub(super) async fn collect_lmtp_results(
        inner: &mut SmtpInner,
        accepted_recipients: Vec<ForwardPath>,
    ) -> Result<Vec<RecipientResult>, Error> {
        let mut results = Vec::with_capacity(accepted_recipients.len());
        for fp in accepted_recipients {
            let resp = inner.read_response().await?;
            results.push(RecipientResult {
                recipient: fp,
                response: resp,
            });
        }
        Ok(results)
    }
}