diff --git a/Cargo.lock b/Cargo.lock index 720c61d3..50002399 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1491,6 +1491,7 @@ version = "0.1.0" dependencies = [ "pest", "pest_derive", + "thiserror", "tokio", "tokio-rustls", ] diff --git a/crates/rfc5321/Cargo.toml b/crates/rfc5321/Cargo.toml index d076119d..73694946 100644 --- a/crates/rfc5321/Cargo.toml +++ b/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" diff --git a/crates/rfc5321/src/client.rs b/crates/rfc5321/src/client.rs new file mode 100644 index 00000000..26762422 --- /dev/null +++ b/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, +} + +pub struct SmtpClient { + socket: Option, + hostname: String, + capabilities: HashMap, + read_buffer: Vec, +} + +impl SmtpClient { + pub fn with_stream>( + 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 { + 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 { + 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 { + 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, 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, SENDER: Into, RECIP: Into>( + &mut self, + sender: SENDER, + recipient: RECIP, + data: B, + ) -> Result { + 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::().ok()?; + if !matches!(class, 2 | 4 | 5) { + // No other classes are defined + return None; + } + let subject = fields.next()?.parse::().ok()?; + + let remainder = fields.next()?; + let mut fields = remainder.splitn(2, ' '); + let detail = fields.next()?.parse::().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, + 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 { + if line.len() < 4 { + return Err(ClientError::MalformedResponseLine(line.to_string())); + } + + match line.as_bytes()[3] { + b' ' | b'-' => match line[0..3].parse::() { + 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(_)) + )); + } +} diff --git a/crates/rfc5321/src/lib.rs b/crates/rfc5321/src/lib.rs index 9effb0fe..2faf1ac9 100644 --- a/crates/rfc5321/src/lib.rs +++ b/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::*; diff --git a/crates/rfc5321/src/parser.rs b/crates/rfc5321/src/parser.rs index bfe11b3f..b7670b61 100644 --- a/crates/rfc5321/src/parser.rs +++ b/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 { + 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 { + 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, } +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 { 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 {