add basic smtp client

Cargo.lock

@@ -1491,6 +1491,7 @@ version = "0.1.0"
 dependencies = [
  "pest",
  "pest_derive",
+ "thiserror",
  "tokio",
  "tokio-rustls",
 ]

crates/rfc5321/Cargo.toml

@@ -8,5 +8,6 @@ edition = "2021"
 [dependencies]
 pest = "2.5"
 pest_derive = "2.5"
+thiserror = "1.0"
 tokio = {version="1.25", features=["full"]}
 tokio-rustls = "0.23"

crates/rfc5321/src/client.rs

@@ -0,0 +1,363 @@
+use crate::{AsyncReadAndWrite, BoxedAsyncReadAndWrite, Command, Domain, ForwardPath, ReversePath};
+use std::collections::HashMap;
+use thiserror::Error;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+const MAX_LINE_LEN: usize = 4096;
+
+#[derive(Error, Debug)]
+pub enum ClientError {
+    #[error("response is not UTF8")]
+    Utf8(#[from] std::string::FromUtf8Error),
+    #[error("IO error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("Malformed Response: {0}")]
+    MalformedResponseLine(String),
+    #[error("Response line is too long")]
+    ResponseTooLong,
+    #[error("Not connected")]
+    NotConnected,
+    #[error("Command rejected {0:?}")]
+    Rejected(Response),
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct EsmtpCapability {
+    pub name: String,
+    pub param: Option<String>,
+}
+
+pub struct SmtpClient {
+    socket: Option<BoxedAsyncReadAndWrite>,
+    hostname: String,
+    capabilities: HashMap<String, EsmtpCapability>,
+    read_buffer: Vec<u8>,
+}
+
+impl SmtpClient {
+    pub fn with_stream<S: AsyncReadAndWrite + 'static, H: AsRef<str>>(
+        stream: S,
+        peer_hostname: H,
+    ) -> Self {
+        Self {
+            socket: Some(Box::new(stream)),
+            hostname: peer_hostname.as_ref().to_string(),
+            capabilities: HashMap::new(),
+            read_buffer: Vec::with_capacity(1024),
+        }
+    }
+
+    async fn read_line(&mut self) -> Result<String, ClientError> {
+        let mut too_long = false;
+        loop {
+            let mut iter = self.read_buffer.iter().enumerate();
+            while let Some((i, &b)) = iter.next() {
+                if b != b'\r' {
+                    continue;
+                }
+                if let Some((_, b'\n')) = iter.next() {
+                    if too_long {
+                        self.read_buffer.drain(0..i + 2);
+                        return Err(ClientError::ResponseTooLong);
+                    }
+
+                    let line = String::from_utf8(self.read_buffer[0..i].to_vec());
+                    self.read_buffer.drain(0..i + 2);
+                    return Ok(line?);
+                }
+            }
+            if self.read_buffer.len() > MAX_LINE_LEN {
+                self.read_buffer.clear();
+                too_long = true;
+            }
+
+            // Didn't find a complete line, fill up the rest of the buffer
+            let mut data = [0u8; MAX_LINE_LEN];
+            let size = self.socket.as_mut().unwrap().read(&mut data).await?;
+            self.read_buffer.extend_from_slice(&data[0..size]);
+        }
+    }
+
+    pub async fn read_response(&mut self) -> Result<Response, ClientError> {
+        if let Some(sock) = self.socket.as_mut() {
+            sock.flush().await?;
+        }
+
+        let mut line = self.read_line().await?;
+        let mut parsed = parse_response_line(&line)?;
+        let code = parsed.code;
+        let (enhanced_code, mut response_string) = match parse_enhanced_status_code(parsed.content)
+        {
+            Some((enhanced, content)) => (Some(enhanced), content.to_string()),
+            None => (None, parsed.content.to_string()),
+        };
+        while !parsed.is_final {
+            line = self.read_line().await?;
+            parsed = parse_response_line(&line)?;
+            if parsed.code != code {
+                return Err(ClientError::MalformedResponseLine(line.to_string()));
+            }
+            response_string.push('\n');
+            response_string.push_str(parsed.content);
+        }
+
+        Ok(Response {
+            code,
+            content: response_string,
+            enhanced_code,
+        })
+    }
+
+    pub async fn send_command(&mut self, command: &Command) -> Result<Response, ClientError> {
+        let line = command.encode();
+        match self.socket.as_mut() {
+            Some(socket) => {
+                socket.write_all(line.as_bytes()).await?;
+            }
+            None => return Err(ClientError::NotConnected),
+        };
+
+        self.read_response().await
+    }
+
+    pub async fn ehlo(
+        &mut self,
+        ehlo_name: &str,
+    ) -> Result<&HashMap<String, EsmtpCapability>, ClientError> {
+        let response = self
+            .send_command(&Command::Ehlo(Domain::Name(ehlo_name.to_string())))
+            .await?;
+        if response.code != 250 {
+            return Err(ClientError::Rejected(response));
+        }
+
+        let mut capabilities = HashMap::new();
+
+        for line in response.content.lines().skip(1) {
+            let mut fields = line.splitn(2, ' ');
+            if let Some(name) = fields.next() {
+                let param = fields.next().map(|s| s.to_string());
+                let cap = EsmtpCapability {
+                    name: name.to_string(),
+                    param,
+                };
+                capabilities.insert(name.to_ascii_uppercase(), cap);
+            }
+        }
+
+        self.capabilities = capabilities;
+        Ok(&self.capabilities)
+    }
+
+    pub async fn send_mail<B: AsRef<[u8]>, SENDER: Into<ReversePath>, RECIP: Into<ForwardPath>>(
+        &mut self,
+        sender: SENDER,
+        recipient: RECIP,
+        data: B,
+    ) -> Result<Response, ClientError> {
+        let mail_resp = self
+            .send_command(&Command::MailFrom {
+                address: sender.into(),
+                parameters: vec![],
+            })
+            .await?;
+        if mail_resp.code != 250 {
+            return Err(ClientError::Rejected(mail_resp));
+        }
+
+        let rcpt_resp = self
+            .send_command(&Command::RcptTo {
+                address: recipient.into(),
+                parameters: vec![],
+            })
+            .await?;
+        if rcpt_resp.code != 250 {
+            return Err(ClientError::Rejected(rcpt_resp));
+        }
+
+        let data_resp = self.send_command(&Command::Data).await?;
+        if data_resp.code != 354 {
+            return Err(ClientError::Rejected(data_resp));
+        }
+
+        let mut needs_stuffing = false;
+        let data: &[u8] = data.as_ref();
+
+        for line in data.split(|&b| b == b'\n') {
+            if line.starts_with(b".") {
+                needs_stuffing = true;
+                break;
+            }
+        }
+
+        if needs_stuffing {
+            let mut stuffed = vec![];
+            for line in data.split(|&b| b == b'\n') {
+                if line.starts_with(b".") {
+                    stuffed.push(b'.');
+                }
+                stuffed.extend_from_slice(line);
+            }
+            match self.socket.as_mut() {
+                Some(sock) => sock.write_all(&stuffed).await?,
+                None => return Err(ClientError::NotConnected),
+            }
+        } else {
+            match self.socket.as_mut() {
+                Some(sock) => sock.write_all(data).await?,
+                None => return Err(ClientError::NotConnected),
+            }
+        }
+
+        let needs_newline = data.last().map(|&b| b != b'\n').unwrap_or(true);
+        let marker = if needs_newline { "\r\n.\r\n" } else { ".\r\n" };
+
+        match self.socket.as_mut() {
+            Some(sock) => sock.write_all(marker.as_bytes()).await?,
+            None => return Err(ClientError::NotConnected),
+        }
+
+        let resp = self.read_response().await?;
+        if resp.code != 250 {
+            return Err(ClientError::Rejected(resp));
+        }
+
+        Ok(resp)
+    }
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub struct EnhancedStatusCode {
+    pub class: u8,
+    pub subject: u16,
+    pub detail: u16,
+}
+
+fn parse_enhanced_status_code(line: &str) -> Option<(EnhancedStatusCode, &str)> {
+    let mut fields = line.splitn(3, '.');
+    let class = fields.next()?.parse::<u8>().ok()?;
+    if !matches!(class, 2 | 4 | 5) {
+        // No other classes are defined
+        return None;
+    }
+    let subject = fields.next()?.parse::<u16>().ok()?;
+
+    let remainder = fields.next()?;
+    let mut fields = remainder.splitn(2, ' ');
+    let detail = fields.next()?.parse::<u16>().ok()?;
+    let remainder = fields.next()?;
+
+    Some((
+        EnhancedStatusCode {
+            class,
+            subject,
+            detail,
+        },
+        remainder,
+    ))
+}
+
+#[derive(Debug, PartialEq, Eq)]
+pub struct Response {
+    pub code: u16,
+    pub enhanced_code: Option<EnhancedStatusCode>,
+    pub content: String,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct ResponseLine<'a> {
+    pub code: u16,
+    pub is_final: bool,
+    pub content: &'a str,
+}
+
+fn parse_response_line(line: &str) -> Result<ResponseLine, ClientError> {
+    if line.len() < 4 {
+        return Err(ClientError::MalformedResponseLine(line.to_string()));
+    }
+
+    match line.as_bytes()[3] {
+        b' ' | b'-' => match line[0..3].parse::<u16>() {
+            Ok(code) => Ok(ResponseLine {
+                code,
+                is_final: line.as_bytes()[3] == b' ',
+                content: &line[4..],
+            }),
+            Err(_) => Err(ClientError::MalformedResponseLine(line.to_string())),
+        },
+        _ => Err(ClientError::MalformedResponseLine(line.to_string())),
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_against_sink() {
+        use tokio::net::TcpStream;
+        let stream = TcpStream::connect("127.0.0.1:25").await.unwrap();
+        let mut client = SmtpClient::with_stream(stream, "localhost");
+        dbg!(client.read_response().await).unwrap();
+        dbg!(client.ehlo("localhost").await).unwrap();
+        let resp = client
+            .send_mail(
+                ReversePath::try_from("wez@wez").unwrap(),
+                ForwardPath::try_from("wez@wez").unwrap(),
+                "Subject: hello\r\n\r\nwoot\r\n",
+            )
+            .await
+            .unwrap();
+        panic!("{resp:#?}");
+    }
+
+    #[test]
+    fn response_parsing() {
+        assert_eq!(
+            parse_enhanced_status_code("2.0.1 w00t"),
+            Some((
+                EnhancedStatusCode {
+                    class: 2,
+                    subject: 0,
+                    detail: 1
+                },
+                "w00t"
+            ))
+        );
+
+        assert_eq!(parse_enhanced_status_code("3.0.0 w00t"), None);
+
+        assert_eq!(parse_enhanced_status_code("2.0.0.1 w00t"), None);
+
+        assert_eq!(parse_enhanced_status_code("2.0.0.1w00t"), None);
+    }
+
+    #[test]
+    fn response_line_parsing() {
+        assert_eq!(
+            parse_response_line("220 woot").unwrap(),
+            ResponseLine {
+                code: 220,
+                is_final: true,
+                content: "woot"
+            }
+        );
+        assert_eq!(
+            parse_response_line("220-woot").unwrap(),
+            ResponseLine {
+                code: 220,
+                is_final: false,
+                content: "woot"
+            }
+        );
+
+        assert!(matches!(
+            parse_response_line("220_woot"),
+            Err(ClientError::MalformedResponseLine(_))
+        ));
+        assert!(matches!(
+            parse_response_line("not really"),
+            Err(ClientError::MalformedResponseLine(_))
+        ));
+    }
+}

crates/rfc5321/src/lib.rs

@@ -1,5 +1,7 @@
+pub mod client;
 pub mod parser;
 pub mod traits;
 
+pub use client::*;
 pub use parser::*;
 pub use traits::*;

crates/rfc5321/src/parser.rs

@@ -171,6 +171,28 @@ pub enum ReversePath {
     NullSender,
 }
 
+impl TryFrom<&str> for ReversePath {
+    type Error = &'static str;
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        if s.is_empty() {
+            Ok(Self::NullSender)
+        } else {
+            let fields: Vec<&str> = s.split('@').collect();
+            if fields.len() == 2 {
+                Ok(Self::Path(MailPath {
+                    at_domain_list: vec![],
+                    mailbox: Mailbox {
+                        local_part: fields[0].to_string(),
+                        domain: Domain::Name(fields[1].to_string()),
+                    },
+                }))
+            } else {
+                Err("wrong number of @ signs")
+            }
+        }
+    }
+}
+
 impl ToString for ReversePath {
     fn to_string(&self) -> String {
         match self {
@@ -186,6 +208,30 @@ pub enum ForwardPath {
     Postmaster,
 }
 
+impl TryFrom<&str> for ForwardPath {
+    type Error = &'static str;
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        if s.is_empty() {
+            Err("cannot send to null sender")
+        } else if s.eq_ignore_ascii_case("postmaster") {
+            Ok(Self::Postmaster)
+        } else {
+            let fields: Vec<&str> = s.split('@').collect();
+            if fields.len() == 2 {
+                Ok(Self::Path(MailPath {
+                    at_domain_list: vec![],
+                    mailbox: Mailbox {
+                        local_part: fields[0].to_string(),
+                        domain: Domain::Name(fields[1].to_string()),
+                    },
+                }))
+            } else {
+                Err("wrong number of @ signs")
+            }
+        }
+    }
+}
+
 impl ToString for ForwardPath {
     fn to_string(&self) -> String {
         match self {
@@ -221,7 +267,8 @@ pub struct Mailbox {
 
 impl ToString for Mailbox {
     fn to_string(&self) -> String {
-        format!("{}@{}", self.local_part, self.domain.to_string())
+        let domain = self.domain.to_string();
+        format!("{}@{}", self.local_part, domain)
     }
 }
 
@@ -250,6 +297,15 @@ pub struct EsmtpParameter {
     pub value: Option<String>,
 }
 
+impl ToString for EsmtpParameter {
+    fn to_string(&self) -> String {
+        match &self.value {
+            Some(value) => format!("{}={}", self.name, value),
+            None => self.name.to_string(),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Command {
     Ehlo(Domain),
@@ -276,6 +332,47 @@ impl Command {
     pub fn parse(line: &str) -> Result<Self, String> {
         Parser::parse_command(line)
     }
+
+    pub fn encode(&self) -> String {
+        match self {
+            Self::Ehlo(domain) => format!("EHLO {}\r\n", domain.to_string()),
+            Self::Helo(domain) => format!("HELO {}\r\n", domain.to_string()),
+            Self::MailFrom {
+                address,
+                parameters,
+            } => {
+                let mut params = String::new();
+                for p in parameters {
+                    params.push(' ');
+                    params.push_str(&p.to_string());
+                }
+
+                format!("MAIL FROM:<{}>{params}\r\n", address.to_string())
+            }
+            Self::RcptTo {
+                address,
+                parameters,
+            } => {
+                let mut params = String::new();
+                for p in parameters {
+                    params.push(' ');
+                    params.push_str(&p.to_string());
+                }
+
+                format!("RCPT TO:<{}>{params}\r\n", address.to_string())
+            }
+            Self::Data => "DATA\r\n".to_string(),
+            Self::Rset => "RSET\r\n".to_string(),
+            Self::Quit => "QUIT\r\n".to_string(),
+            Self::StartTls => "STARTTLS\r\n".to_string(),
+            Self::Vrfy(param) => format!("VRFY {param}\r\n"),
+            Self::Expn(param) => format!("EXPN {param}\r\n"),
+            Self::Help(Some(param)) => format!("HELP {param}\r\n"),
+            Self::Help(None) => format!("HELP\r\n"),
+            Self::Noop(Some(param)) => format!("NOOP {param}\r\n"),
+            Self::Noop(None) => format!("NOOP\r\n"),
+        }
+    }
 }
 
 pub fn is_valid_domain(text: &str) -> bool {