Compare commits

...

4 Commits

Author SHA1 Message Date
Paolo Barbolini
efee4b5d72 Fix warnings 2024-03-19 16:21:10 +01:00
Paolo Barbolini
57d7bf25cc Replace try_smtp macro with more resilient method 2024-03-19 16:21:10 +01:00
Paolo Barbolini
c52c596458 Drop ref syntax 2024-03-10 15:28:54 +01:00
Paolo Barbolini
53dee3e31f Drop need for NetworkStream::None variant 2024-03-10 15:28:54 +01:00
14 changed files with 279 additions and 355 deletions

View File

@@ -134,7 +134,7 @@ impl Executor for Tokio1Executor {
#[allow(clippy::match_single_binding)] #[allow(clippy::match_single_binding)]
let tls_parameters = match tls { let tls_parameters = match tls {
#[cfg(any(feature = "tokio1-native-tls", feature = "tokio1-rustls-tls"))] #[cfg(any(feature = "tokio1-native-tls", feature = "tokio1-rustls-tls"))]
Tls::Wrapper(ref tls_parameters) => Some(tls_parameters.clone()), Tls::Wrapper(tls_parameters) => Some(tls_parameters.clone()),
_ => None, _ => None,
}; };
#[allow(unused_mut)] #[allow(unused_mut)]
@@ -149,13 +149,13 @@ impl Executor for Tokio1Executor {
#[cfg(any(feature = "tokio1-native-tls", feature = "tokio1-rustls-tls"))] #[cfg(any(feature = "tokio1-native-tls", feature = "tokio1-rustls-tls"))]
match tls { match tls {
Tls::Opportunistic(ref tls_parameters) => { Tls::Opportunistic(tls_parameters) => {
if conn.can_starttls() { if conn.can_starttls() {
conn.starttls(tls_parameters.clone(), hello_name).await?; conn = conn.starttls(tls_parameters.clone(), hello_name).await?;
} }
} }
Tls::Required(ref tls_parameters) => { Tls::Required(tls_parameters) => {
conn.starttls(tls_parameters.clone(), hello_name).await?; conn = conn.starttls(tls_parameters.clone(), hello_name).await?;
} }
_ => (), _ => (),
} }
@@ -231,7 +231,7 @@ impl Executor for AsyncStd1Executor {
#[allow(clippy::match_single_binding)] #[allow(clippy::match_single_binding)]
let tls_parameters = match tls { let tls_parameters = match tls {
#[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))] #[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))]
Tls::Wrapper(ref tls_parameters) => Some(tls_parameters.clone()), Tls::Wrapper(tls_parameters) => Some(tls_parameters.clone()),
_ => None, _ => None,
}; };
#[allow(unused_mut)] #[allow(unused_mut)]
@@ -245,13 +245,13 @@ impl Executor for AsyncStd1Executor {
#[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))] #[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))]
match tls { match tls {
Tls::Opportunistic(ref tls_parameters) => { Tls::Opportunistic(tls_parameters) => {
if conn.can_starttls() { if conn.can_starttls() {
conn.starttls(tls_parameters.clone(), hello_name).await?; conn = conn.starttls(tls_parameters.clone(), hello_name).await?;
} }
} }
Tls::Required(ref tls_parameters) => { Tls::Required(tls_parameters) => {
conn.starttls(tls_parameters.clone(), hello_name).await?; conn = conn.starttls(tls_parameters.clone(), hello_name).await?;
} }
_ => (), _ => (),
} }

View File

@@ -2,7 +2,6 @@ use std::{
borrow::Cow, borrow::Cow,
error::Error as StdError, error::Error as StdError,
fmt::{self, Display}, fmt::{self, Display},
iter::IntoIterator,
time::SystemTime, time::SystemTime,
}; };

View File

@@ -88,7 +88,7 @@ impl Mailbox {
impl Display for Mailbox { impl Display for Mailbox {
fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult { fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
if let Some(ref name) = self.name { if let Some(name) = &self.name {
let name = name.trim(); let name = name.trim();
if !name.is_empty() { if !name.is_empty() {
write_word(f, name)?; write_word(f, name)?;

View File

@@ -54,7 +54,7 @@ impl fmt::Debug for Error {
builder.field("kind", &self.inner.kind); builder.field("kind", &self.inner.kind);
if let Some(ref source) = self.inner.source { if let Some(source) = &self.inner.source {
builder.field("source", source); builder.field("source", source);
} }
@@ -70,7 +70,7 @@ impl fmt::Display for Error {
Kind::Envelope => f.write_str("internal client error")?, Kind::Envelope => f.write_str("internal client error")?,
}; };
if let Some(ref e) = self.inner.source { if let Some(e) = &self.inner.source {
write!(f, ": {e}")?; write!(f, ": {e}")?;
} }

View File

@@ -52,7 +52,7 @@ impl fmt::Debug for Error {
builder.field("kind", &self.inner.kind); builder.field("kind", &self.inner.kind);
if let Some(ref source) = self.inner.source { if let Some(source) = &self.inner.source {
builder.field("source", source); builder.field("source", source);
} }
@@ -67,7 +67,7 @@ impl fmt::Display for Error {
Kind::Client => f.write_str("internal client error")?, Kind::Client => f.write_str("internal client error")?,
}; };
if let Some(ref e) = self.inner.source { if let Some(e) = &self.inner.source {
write!(f, ": {e}")?; write!(f, ": {e}")?;
} }

View File

@@ -6,7 +6,7 @@ use futures_util::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use super::async_net::AsyncTokioStream; use super::async_net::AsyncTokioStream;
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
use super::escape_crlf; use super::escape_crlf;
use super::{AsyncNetworkStream, ClientCodec, TlsParameters}; use super::{AsyncNetworkStream, ClientCodec, ConnectionState, TlsParameters};
use crate::{ use crate::{
transport::smtp::{ transport::smtp::{
authentication::{Credentials, Mechanism}, authentication::{Credentials, Mechanism},
@@ -19,25 +19,11 @@ use crate::{
Envelope, Envelope,
}; };
macro_rules! try_smtp (
($err: expr, $client: ident) => ({
match $err {
Ok(val) => val,
Err(err) => {
$client.abort().await;
return Err(From::from(err))
},
}
})
);
/// Structure that implements the SMTP client /// Structure that implements the SMTP client
pub struct AsyncSmtpConnection { pub struct AsyncSmtpConnection {
/// TCP stream between client and server /// TCP stream between client and server
/// Value is None before connection /// Value is None before connection
stream: BufReader<AsyncNetworkStream>, stream: BufReader<AsyncNetworkStream>,
/// Panic state
panic: bool,
/// Information about the server /// Information about the server
server_info: ServerInfo, server_info: ServerInfo,
} }
@@ -126,7 +112,6 @@ impl AsyncSmtpConnection {
let stream = BufReader::new(stream); let stream = BufReader::new(stream);
let mut conn = AsyncSmtpConnection { let mut conn = AsyncSmtpConnection {
stream, stream,
panic: false,
server_info: ServerInfo::default(), server_info: ServerInfo::default(),
}; };
// TODO log // TODO log
@@ -170,30 +155,26 @@ impl AsyncSmtpConnection {
mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime)); mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
} }
try_smtp!( self.command(Mail::new(envelope.from().cloned(), mail_options))
self.command(Mail::new(envelope.from().cloned(), mail_options)) .await?;
.await,
self
);
// Recipient // Recipient
for to_address in envelope.to() { for to_address in envelope.to() {
try_smtp!( self.command(Rcpt::new(to_address.clone(), vec![])).await?;
self.command(Rcpt::new(to_address.clone(), vec![])).await,
self
);
} }
// Data // Data
try_smtp!(self.command(Data).await, self); self.command(Data).await?;
// Message content // Message content
let result = try_smtp!(self.message(email).await, self); self.message(email).await
Ok(result)
} }
pub fn has_broken(&self) -> bool { pub fn has_broken(&self) -> bool {
self.panic match self.stream.get_ref().state() {
ConnectionState::Ok => false,
ConnectionState::Broken | ConnectionState::Closed => true,
}
} }
pub fn can_starttls(&self) -> bool { pub fn can_starttls(&self) -> bool {
@@ -208,18 +189,20 @@ impl AsyncSmtpConnection {
/// [rfc8314]: https://www.rfc-editor.org/rfc/rfc8314 /// [rfc8314]: https://www.rfc-editor.org/rfc/rfc8314
#[allow(unused_variables)] #[allow(unused_variables)]
pub async fn starttls( pub async fn starttls(
&mut self, mut self,
tls_parameters: TlsParameters, tls_parameters: TlsParameters,
hello_name: &ClientId, hello_name: &ClientId,
) -> Result<(), Error> { ) -> Result<Self, Error> {
if self.server_info.supports_feature(Extension::StartTls) { if self.server_info.supports_feature(Extension::StartTls) {
try_smtp!(self.command(Starttls).await, self); self.command(Starttls).await?;
self.stream.get_mut().upgrade_tls(tls_parameters).await?; let stream = self.stream.into_inner();
let stream = stream.upgrade_tls(tls_parameters).await?;
self.stream = BufReader::new(stream);
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("connection encrypted"); tracing::debug!("connection encrypted");
// Send EHLO again // Send EHLO again
try_smtp!(self.ehlo(hello_name).await, self); self.ehlo(hello_name).await?;
Ok(()) Ok(self)
} else { } else {
Err(error::client("STARTTLS is not supported on this server")) Err(error::client("STARTTLS is not supported on this server"))
} }
@@ -227,22 +210,24 @@ impl AsyncSmtpConnection {
/// Send EHLO and update server info /// Send EHLO and update server info
async fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> { async fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> {
let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())).await, self); let ehlo_response = self.command(Ehlo::new(hello_name.clone())).await?;
self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self); self.server_info = ServerInfo::from_response(&ehlo_response)?;
Ok(()) Ok(())
} }
pub async fn quit(&mut self) -> Result<Response, Error> { pub async fn quit(&mut self) -> Result<Response, Error> {
Ok(try_smtp!(self.command(Quit).await, self)) self.command(Quit).await
} }
pub async fn abort(&mut self) { pub async fn abort(&mut self) {
// Only try to quit if we are not already broken match self.stream.get_ref().state() {
if !self.panic { ConnectionState::Ok | ConnectionState::Broken => {
self.panic = true; let _ = self.command(Quit).await;
let _ = self.command(Quit).await; let _ = self.stream.close().await;
self.stream.get_mut().set_state(ConnectionState::Closed);
}
ConnectionState::Closed => {}
} }
let _ = self.stream.close().await;
} }
/// Sets the underlying stream /// Sets the underlying stream
@@ -279,15 +264,13 @@ impl AsyncSmtpConnection {
while challenges > 0 && response.has_code(334) { while challenges > 0 && response.has_code(334) {
challenges -= 1; challenges -= 1;
response = try_smtp!( response = self
self.command(Auth::new_from_response( .command(Auth::new_from_response(
mechanism, mechanism,
credentials.clone(), credentials.clone(),
&response, &response,
)?) )?)
.await, .await?;
self
);
} }
if challenges == 0 { if challenges == 0 {
@@ -315,6 +298,9 @@ impl AsyncSmtpConnection {
/// Writes a string to the server /// Writes a string to the server
async fn write(&mut self, string: &[u8]) -> Result<(), Error> { async fn write(&mut self, string: &[u8]) -> Result<(), Error> {
self.stream.get_ref().state().verify()?;
self.stream.get_mut().set_state(ConnectionState::Broken);
self.stream self.stream
.get_mut() .get_mut()
.write_all(string) .write_all(string)
@@ -326,6 +312,8 @@ impl AsyncSmtpConnection {
.await .await
.map_err(error::network)?; .map_err(error::network)?;
self.stream.get_mut().set_state(ConnectionState::Ok);
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string))); tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string)));
Ok(()) Ok(())
@@ -333,6 +321,9 @@ impl AsyncSmtpConnection {
/// Gets the SMTP response /// Gets the SMTP response
pub async fn read_response(&mut self) -> Result<Response, Error> { pub async fn read_response(&mut self) -> Result<Response, Error> {
self.stream.get_ref().state().verify()?;
self.stream.get_mut().set_state(ConnectionState::Broken);
let mut buffer = String::with_capacity(100); let mut buffer = String::with_capacity(100);
while self while self
@@ -346,6 +337,8 @@ impl AsyncSmtpConnection {
tracing::debug!("<< {}", escape_crlf(&buffer)); tracing::debug!("<< {}", escape_crlf(&buffer));
match parse_response(&buffer) { match parse_response(&buffer) {
Ok((_remaining, response)) => { Ok((_remaining, response)) => {
self.stream.get_mut().set_state(ConnectionState::Ok);
return if response.is_positive() { return if response.is_positive() {
Ok(response) Ok(response)
} else { } else {
@@ -353,7 +346,7 @@ impl AsyncSmtpConnection {
response.code(), response.code(),
Some(response.message().collect()), Some(response.message().collect()),
)) ))
} };
} }
Err(nom::Err::Failure(e)) => { Err(nom::Err::Failure(e)) => {
return Err(error::response(e.to_string())); return Err(error::response(e.to_string()));

View File

@@ -11,8 +11,7 @@ use async_native_tls::TlsStream as AsyncStd1TlsStream;
#[cfg(feature = "async-std1")] #[cfg(feature = "async-std1")]
use async_std::net::{TcpStream as AsyncStd1TcpStream, ToSocketAddrs as AsyncStd1ToSocketAddrs}; use async_std::net::{TcpStream as AsyncStd1TcpStream, ToSocketAddrs as AsyncStd1ToSocketAddrs};
use futures_io::{ use futures_io::{
AsyncRead as FuturesAsyncRead, AsyncWrite as FuturesAsyncWrite, Error as IoError, ErrorKind, AsyncRead as FuturesAsyncRead, AsyncWrite as FuturesAsyncWrite, Result as IoResult,
Result as IoResult,
}; };
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
use futures_rustls::client::TlsStream as AsyncStd1RustlsTlsStream; use futures_rustls::client::TlsStream as AsyncStd1RustlsTlsStream;
@@ -40,7 +39,7 @@ use tokio1_rustls::client::TlsStream as Tokio1RustlsTlsStream;
feature = "async-std1-rustls-tls" feature = "async-std1-rustls-tls"
))] ))]
use super::InnerTlsParameters; use super::InnerTlsParameters;
use super::TlsParameters; use super::{ConnectionState, TlsParameters};
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
use crate::transport::smtp::client::net::resolved_address_filter; use crate::transport::smtp::client::net::resolved_address_filter;
use crate::transport::smtp::{error, Error}; use crate::transport::smtp::{error, Error};
@@ -49,6 +48,7 @@ use crate::transport::smtp::{error, Error};
#[derive(Debug)] #[derive(Debug)]
pub struct AsyncNetworkStream { pub struct AsyncNetworkStream {
inner: InnerAsyncNetworkStream, inner: InnerAsyncNetworkStream,
state: ConnectionState,
} }
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
@@ -91,45 +91,43 @@ enum InnerAsyncNetworkStream {
/// Encrypted Tokio 1.x TCP stream /// Encrypted Tokio 1.x TCP stream
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
AsyncStd1RustlsTls(AsyncStd1RustlsTlsStream<AsyncStd1TcpStream>), AsyncStd1RustlsTls(AsyncStd1RustlsTlsStream<AsyncStd1TcpStream>),
/// Can't be built
None,
} }
impl AsyncNetworkStream { impl AsyncNetworkStream {
fn new(inner: InnerAsyncNetworkStream) -> Self { fn new(inner: InnerAsyncNetworkStream) -> Self {
if let InnerAsyncNetworkStream::None = inner { AsyncNetworkStream {
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built"); inner,
state: ConnectionState::Ok,
} }
}
AsyncNetworkStream { inner } pub(super) fn state(&self) -> ConnectionState {
self.state
}
pub(super) fn set_state(&mut self, state: ConnectionState) {
self.state = state;
} }
/// Returns peer's address /// Returns peer's address
pub fn peer_addr(&self) -> IoResult<SocketAddr> { pub fn peer_addr(&self) -> IoResult<SocketAddr> {
match self.inner { match &self.inner {
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
InnerAsyncNetworkStream::Tokio1Tcp(ref s) => s.peer_addr(), InnerAsyncNetworkStream::Tokio1Tcp(s) => s.peer_addr(),
#[cfg(feature = "tokio1-native-tls")] #[cfg(feature = "tokio1-native-tls")]
InnerAsyncNetworkStream::Tokio1NativeTls(ref s) => { InnerAsyncNetworkStream::Tokio1NativeTls(s) => {
s.get_ref().get_ref().get_ref().peer_addr() s.get_ref().get_ref().get_ref().peer_addr()
} }
#[cfg(feature = "tokio1-rustls-tls")] #[cfg(feature = "tokio1-rustls-tls")]
InnerAsyncNetworkStream::Tokio1RustlsTls(ref s) => s.get_ref().0.peer_addr(), InnerAsyncNetworkStream::Tokio1RustlsTls(s) => s.get_ref().0.peer_addr(),
#[cfg(feature = "tokio1-boring-tls")] #[cfg(feature = "tokio1-boring-tls")]
InnerAsyncNetworkStream::Tokio1BoringTls(ref s) => s.get_ref().peer_addr(), InnerAsyncNetworkStream::Tokio1BoringTls(s) => s.get_ref().peer_addr(),
#[cfg(feature = "async-std1")] #[cfg(feature = "async-std1")]
InnerAsyncNetworkStream::AsyncStd1Tcp(ref s) => s.peer_addr(), InnerAsyncNetworkStream::AsyncStd1Tcp(s) => s.peer_addr(),
#[cfg(feature = "async-std1-native-tls")] #[cfg(feature = "async-std1-native-tls")]
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref s) => s.get_ref().peer_addr(), InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => s.get_ref().peer_addr(),
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref s) => s.get_ref().0.peer_addr(), InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => s.get_ref().0.peer_addr(),
InnerAsyncNetworkStream::None => {
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
Err(IoError::new(
ErrorKind::Other,
"InnerAsyncNetworkStream::None must never be built",
))
}
} }
} }
@@ -199,7 +197,7 @@ impl AsyncNetworkStream {
let mut stream = let mut stream =
AsyncNetworkStream::new(InnerAsyncNetworkStream::Tokio1Tcp(Box::new(tcp_stream))); AsyncNetworkStream::new(InnerAsyncNetworkStream::Tokio1Tcp(Box::new(tcp_stream)));
if let Some(tls_parameters) = tls_parameters { if let Some(tls_parameters) = tls_parameters {
stream.upgrade_tls(tls_parameters).await?; stream = stream.upgrade_tls(tls_parameters).await?;
} }
Ok(stream) Ok(stream)
} }
@@ -250,13 +248,13 @@ impl AsyncNetworkStream {
let mut stream = AsyncNetworkStream::new(InnerAsyncNetworkStream::AsyncStd1Tcp(tcp_stream)); let mut stream = AsyncNetworkStream::new(InnerAsyncNetworkStream::AsyncStd1Tcp(tcp_stream));
if let Some(tls_parameters) = tls_parameters { if let Some(tls_parameters) = tls_parameters {
stream.upgrade_tls(tls_parameters).await?; stream = stream.upgrade_tls(tls_parameters).await?;
} }
Ok(stream) Ok(stream)
} }
pub async fn upgrade_tls(&mut self, tls_parameters: TlsParameters) -> Result<(), Error> { pub async fn upgrade_tls(self, tls_parameters: TlsParameters) -> Result<Self, Error> {
match &self.inner { match self.inner {
#[cfg(all( #[cfg(all(
feature = "tokio1", feature = "tokio1",
not(any( not(any(
@@ -275,18 +273,14 @@ impl AsyncNetworkStream {
feature = "tokio1-rustls-tls", feature = "tokio1-rustls-tls",
feature = "tokio1-boring-tls" feature = "tokio1-boring-tls"
))] ))]
InnerAsyncNetworkStream::Tokio1Tcp(_) => { InnerAsyncNetworkStream::Tokio1Tcp(tcp_stream) => {
// get owned TcpStream let inner = Self::upgrade_tokio1_tls(tcp_stream, tls_parameters)
let tcp_stream = mem::replace(&mut self.inner, InnerAsyncNetworkStream::None);
let tcp_stream = match tcp_stream {
InnerAsyncNetworkStream::Tokio1Tcp(tcp_stream) => tcp_stream,
_ => unreachable!(),
};
self.inner = Self::upgrade_tokio1_tls(tcp_stream, tls_parameters)
.await .await
.map_err(error::connection)?; .map_err(error::connection)?;
Ok(()) Ok(Self {
inner,
state: ConnectionState::Ok,
})
} }
#[cfg(all( #[cfg(all(
feature = "async-std1", feature = "async-std1",
@@ -298,20 +292,16 @@ impl AsyncNetworkStream {
} }
#[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))] #[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))]
InnerAsyncNetworkStream::AsyncStd1Tcp(_) => { InnerAsyncNetworkStream::AsyncStd1Tcp(tcp_stream) => {
// get owned TcpStream let inner = Self::upgrade_asyncstd1_tls(tcp_stream, tls_parameters)
let tcp_stream = mem::replace(&mut self.inner, InnerAsyncNetworkStream::None);
let tcp_stream = match tcp_stream {
InnerAsyncNetworkStream::AsyncStd1Tcp(tcp_stream) => tcp_stream,
_ => unreachable!(),
};
self.inner = Self::upgrade_asyncstd1_tls(tcp_stream, tls_parameters)
.await .await
.map_err(error::connection)?; .map_err(error::connection)?;
Ok(()) Ok(Self {
inner,
state: ConnectionState::Ok,
})
} }
_ => Ok(()), _ => Ok(self),
} }
} }
@@ -445,7 +435,7 @@ impl AsyncNetworkStream {
} }
pub fn is_encrypted(&self) -> bool { pub fn is_encrypted(&self) -> bool {
match self.inner { match &self.inner {
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
InnerAsyncNetworkStream::Tokio1Tcp(_) => false, InnerAsyncNetworkStream::Tokio1Tcp(_) => false,
#[cfg(feature = "tokio1-native-tls")] #[cfg(feature = "tokio1-native-tls")]
@@ -460,7 +450,6 @@ impl AsyncNetworkStream {
InnerAsyncNetworkStream::AsyncStd1NativeTls(_) => true, InnerAsyncNetworkStream::AsyncStd1NativeTls(_) => true,
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
InnerAsyncNetworkStream::AsyncStd1RustlsTls(_) => true, InnerAsyncNetworkStream::AsyncStd1RustlsTls(_) => true,
InnerAsyncNetworkStream::None => false,
} }
} }
@@ -509,7 +498,6 @@ impl AsyncNetworkStream {
.first() .first()
.unwrap() .unwrap()
.to_vec()), .to_vec()),
InnerAsyncNetworkStream::None => panic!("InnerNetworkStream::None must never be built"),
} }
} }
} }
@@ -520,9 +508,9 @@ impl FuturesAsyncRead for AsyncNetworkStream {
cx: &mut Context<'_>, cx: &mut Context<'_>,
buf: &mut [u8], buf: &mut [u8],
) -> Poll<IoResult<usize>> { ) -> Poll<IoResult<usize>> {
match self.inner { match &mut self.inner {
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
InnerAsyncNetworkStream::Tokio1Tcp(ref mut s) => { InnerAsyncNetworkStream::Tokio1Tcp(s) => {
let mut b = Tokio1ReadBuf::new(buf); let mut b = Tokio1ReadBuf::new(buf);
match Pin::new(s).poll_read(cx, &mut b) { match Pin::new(s).poll_read(cx, &mut b) {
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())), Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
@@ -531,7 +519,7 @@ impl FuturesAsyncRead for AsyncNetworkStream {
} }
} }
#[cfg(feature = "tokio1-native-tls")] #[cfg(feature = "tokio1-native-tls")]
InnerAsyncNetworkStream::Tokio1NativeTls(ref mut s) => { InnerAsyncNetworkStream::Tokio1NativeTls(s) => {
let mut b = Tokio1ReadBuf::new(buf); let mut b = Tokio1ReadBuf::new(buf);
match Pin::new(s).poll_read(cx, &mut b) { match Pin::new(s).poll_read(cx, &mut b) {
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())), Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
@@ -540,7 +528,7 @@ impl FuturesAsyncRead for AsyncNetworkStream {
} }
} }
#[cfg(feature = "tokio1-rustls-tls")] #[cfg(feature = "tokio1-rustls-tls")]
InnerAsyncNetworkStream::Tokio1RustlsTls(ref mut s) => { InnerAsyncNetworkStream::Tokio1RustlsTls(s) => {
let mut b = Tokio1ReadBuf::new(buf); let mut b = Tokio1ReadBuf::new(buf);
match Pin::new(s).poll_read(cx, &mut b) { match Pin::new(s).poll_read(cx, &mut b) {
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())), Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
@@ -549,7 +537,7 @@ impl FuturesAsyncRead for AsyncNetworkStream {
} }
} }
#[cfg(feature = "tokio1-boring-tls")] #[cfg(feature = "tokio1-boring-tls")]
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => { InnerAsyncNetworkStream::Tokio1BoringTls(s) => {
let mut b = Tokio1ReadBuf::new(buf); let mut b = Tokio1ReadBuf::new(buf);
match Pin::new(s).poll_read(cx, &mut b) { match Pin::new(s).poll_read(cx, &mut b) {
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())), Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
@@ -558,19 +546,11 @@ impl FuturesAsyncRead for AsyncNetworkStream {
} }
} }
#[cfg(feature = "async-std1")] #[cfg(feature = "async-std1")]
InnerAsyncNetworkStream::AsyncStd1Tcp(ref mut s) => Pin::new(s).poll_read(cx, buf), InnerAsyncNetworkStream::AsyncStd1Tcp(s) => Pin::new(s).poll_read(cx, buf),
#[cfg(feature = "async-std1-native-tls")] #[cfg(feature = "async-std1-native-tls")]
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref mut s) => { InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => Pin::new(s).poll_read(cx, buf),
Pin::new(s).poll_read(cx, buf)
}
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => { InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_read(cx, buf),
Pin::new(s).poll_read(cx, buf)
}
InnerAsyncNetworkStream::None => {
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
Poll::Ready(Ok(0))
}
} }
} }
} }
@@ -581,75 +561,61 @@ impl FuturesAsyncWrite for AsyncNetworkStream {
cx: &mut Context<'_>, cx: &mut Context<'_>,
buf: &[u8], buf: &[u8],
) -> Poll<IoResult<usize>> { ) -> Poll<IoResult<usize>> {
match self.inner { match &mut self.inner {
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
InnerAsyncNetworkStream::Tokio1Tcp(ref mut s) => Pin::new(s).poll_write(cx, buf), InnerAsyncNetworkStream::Tokio1Tcp(s) => Pin::new(s).poll_write(cx, buf),
#[cfg(feature = "tokio1-native-tls")] #[cfg(feature = "tokio1-native-tls")]
InnerAsyncNetworkStream::Tokio1NativeTls(ref mut s) => Pin::new(s).poll_write(cx, buf), InnerAsyncNetworkStream::Tokio1NativeTls(s) => Pin::new(s).poll_write(cx, buf),
#[cfg(feature = "tokio1-rustls-tls")] #[cfg(feature = "tokio1-rustls-tls")]
InnerAsyncNetworkStream::Tokio1RustlsTls(ref mut s) => Pin::new(s).poll_write(cx, buf), InnerAsyncNetworkStream::Tokio1RustlsTls(s) => Pin::new(s).poll_write(cx, buf),
#[cfg(feature = "tokio1-boring-tls")] #[cfg(feature = "tokio1-boring-tls")]
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => Pin::new(s).poll_write(cx, buf), InnerAsyncNetworkStream::Tokio1BoringTls(s) => Pin::new(s).poll_write(cx, buf),
#[cfg(feature = "async-std1")] #[cfg(feature = "async-std1")]
InnerAsyncNetworkStream::AsyncStd1Tcp(ref mut s) => Pin::new(s).poll_write(cx, buf), InnerAsyncNetworkStream::AsyncStd1Tcp(s) => Pin::new(s).poll_write(cx, buf),
#[cfg(feature = "async-std1-native-tls")] #[cfg(feature = "async-std1-native-tls")]
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref mut s) => { InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => Pin::new(s).poll_write(cx, buf),
Pin::new(s).poll_write(cx, buf)
}
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => { InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_write(cx, buf),
Pin::new(s).poll_write(cx, buf)
}
InnerAsyncNetworkStream::None => {
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
Poll::Ready(Ok(0))
}
} }
} }
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> { fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
match self.inner { match &mut self.inner {
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
InnerAsyncNetworkStream::Tokio1Tcp(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::Tokio1Tcp(s) => Pin::new(s).poll_flush(cx),
#[cfg(feature = "tokio1-native-tls")] #[cfg(feature = "tokio1-native-tls")]
InnerAsyncNetworkStream::Tokio1NativeTls(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::Tokio1NativeTls(s) => Pin::new(s).poll_flush(cx),
#[cfg(feature = "tokio1-rustls-tls")] #[cfg(feature = "tokio1-rustls-tls")]
InnerAsyncNetworkStream::Tokio1RustlsTls(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::Tokio1RustlsTls(s) => Pin::new(s).poll_flush(cx),
#[cfg(feature = "tokio1-boring-tls")] #[cfg(feature = "tokio1-boring-tls")]
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::Tokio1BoringTls(s) => Pin::new(s).poll_flush(cx),
#[cfg(feature = "async-std1")] #[cfg(feature = "async-std1")]
InnerAsyncNetworkStream::AsyncStd1Tcp(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::AsyncStd1Tcp(s) => Pin::new(s).poll_flush(cx),
#[cfg(feature = "async-std1-native-tls")] #[cfg(feature = "async-std1-native-tls")]
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => Pin::new(s).poll_flush(cx),
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => Pin::new(s).poll_flush(cx), InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_flush(cx),
InnerAsyncNetworkStream::None => {
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
Poll::Ready(Ok(()))
}
} }
} }
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> { fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
match self.inner { self.state = ConnectionState::Closed;
match &mut self.inner {
#[cfg(feature = "tokio1")] #[cfg(feature = "tokio1")]
InnerAsyncNetworkStream::Tokio1Tcp(ref mut s) => Pin::new(s).poll_shutdown(cx), InnerAsyncNetworkStream::Tokio1Tcp(s) => Pin::new(s).poll_shutdown(cx),
#[cfg(feature = "tokio1-native-tls")] #[cfg(feature = "tokio1-native-tls")]
InnerAsyncNetworkStream::Tokio1NativeTls(ref mut s) => Pin::new(s).poll_shutdown(cx), InnerAsyncNetworkStream::Tokio1NativeTls(s) => Pin::new(s).poll_shutdown(cx),
#[cfg(feature = "tokio1-rustls-tls")] #[cfg(feature = "tokio1-rustls-tls")]
InnerAsyncNetworkStream::Tokio1RustlsTls(ref mut s) => Pin::new(s).poll_shutdown(cx), InnerAsyncNetworkStream::Tokio1RustlsTls(s) => Pin::new(s).poll_shutdown(cx),
#[cfg(feature = "tokio1-boring-tls")] #[cfg(feature = "tokio1-boring-tls")]
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => Pin::new(s).poll_shutdown(cx), InnerAsyncNetworkStream::Tokio1BoringTls(s) => Pin::new(s).poll_shutdown(cx),
#[cfg(feature = "async-std1")] #[cfg(feature = "async-std1")]
InnerAsyncNetworkStream::AsyncStd1Tcp(ref mut s) => Pin::new(s).poll_close(cx), InnerAsyncNetworkStream::AsyncStd1Tcp(s) => Pin::new(s).poll_close(cx),
#[cfg(feature = "async-std1-native-tls")] #[cfg(feature = "async-std1-native-tls")]
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref mut s) => Pin::new(s).poll_close(cx), InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => Pin::new(s).poll_close(cx),
#[cfg(feature = "async-std1-rustls-tls")] #[cfg(feature = "async-std1-rustls-tls")]
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => Pin::new(s).poll_close(cx), InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_close(cx),
InnerAsyncNetworkStream::None => {
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
Poll::Ready(Ok(()))
}
} }
} }
} }

View File

@@ -7,7 +7,7 @@ use std::{
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
use super::escape_crlf; use super::escape_crlf;
use super::{ClientCodec, NetworkStream, TlsParameters}; use super::{ClientCodec, ConnectionState, NetworkStream, TlsParameters};
use crate::{ use crate::{
address::Envelope, address::Envelope,
transport::smtp::{ transport::smtp::{
@@ -20,25 +20,11 @@ use crate::{
}, },
}; };
macro_rules! try_smtp (
($err: expr, $client: ident) => ({
match $err {
Ok(val) => val,
Err(err) => {
$client.abort();
return Err(From::from(err))
},
}
})
);
/// Structure that implements the SMTP client /// Structure that implements the SMTP client
pub struct SmtpConnection { pub struct SmtpConnection {
/// TCP stream between client and server /// TCP stream between client and server
/// Value is None before connection /// Value is None before connection
stream: BufReader<NetworkStream>, stream: BufReader<NetworkStream>,
/// Panic state
panic: bool,
/// Information about the server /// Information about the server
server_info: ServerInfo, server_info: ServerInfo,
} }
@@ -65,7 +51,6 @@ impl SmtpConnection {
let stream = BufReader::new(stream); let stream = BufReader::new(stream);
let mut conn = SmtpConnection { let mut conn = SmtpConnection {
stream, stream,
panic: false,
server_info: ServerInfo::default(), server_info: ServerInfo::default(),
}; };
conn.set_timeout(timeout).map_err(error::network)?; conn.set_timeout(timeout).map_err(error::network)?;
@@ -110,26 +95,25 @@ impl SmtpConnection {
mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime)); mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
} }
try_smtp!( self.command(Mail::new(envelope.from().cloned(), mail_options))?;
self.command(Mail::new(envelope.from().cloned(), mail_options)),
self
);
// Recipient // Recipient
for to_address in envelope.to() { for to_address in envelope.to() {
try_smtp!(self.command(Rcpt::new(to_address.clone(), vec![])), self); self.command(Rcpt::new(to_address.clone(), vec![]))?;
} }
// Data // Data
try_smtp!(self.command(Data), self); self.command(Data)?;
// Message content // Message content
let result = try_smtp!(self.message(email), self); self.message(email)
Ok(result)
} }
pub fn has_broken(&self) -> bool { pub fn has_broken(&self) -> bool {
self.panic match self.stream.get_ref().state() {
ConnectionState::Ok => false,
ConnectionState::Broken | ConnectionState::Closed => true,
}
} }
pub fn can_starttls(&self) -> bool { pub fn can_starttls(&self) -> bool {
@@ -138,20 +122,22 @@ impl SmtpConnection {
#[allow(unused_variables)] #[allow(unused_variables)]
pub fn starttls( pub fn starttls(
&mut self, mut self,
tls_parameters: &TlsParameters, tls_parameters: &TlsParameters,
hello_name: &ClientId, hello_name: &ClientId,
) -> Result<(), Error> { ) -> Result<Self, Error> {
if self.server_info.supports_feature(Extension::StartTls) { if self.server_info.supports_feature(Extension::StartTls) {
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
{ {
try_smtp!(self.command(Starttls), self); self.command(Starttls)?;
self.stream.get_mut().upgrade_tls(tls_parameters)?; let stream = self.stream.into_inner();
let stream = stream.upgrade_tls(tls_parameters)?;
self.stream = BufReader::new(stream);
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("connection encrypted"); tracing::debug!("connection encrypted");
// Send EHLO again // Send EHLO again
try_smtp!(self.ehlo(hello_name), self); self.ehlo(hello_name)?;
Ok(()) Ok(self)
} }
#[cfg(not(any( #[cfg(not(any(
feature = "native-tls", feature = "native-tls",
@@ -168,22 +154,24 @@ impl SmtpConnection {
/// Send EHLO and update server info /// Send EHLO and update server info
fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> { fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> {
let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())), self); let ehlo_response = self.command(Ehlo::new(hello_name.clone()))?;
self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self); self.server_info = ServerInfo::from_response(&ehlo_response)?;
Ok(()) Ok(())
} }
pub fn quit(&mut self) -> Result<Response, Error> { pub fn quit(&mut self) -> Result<Response, Error> {
Ok(try_smtp!(self.command(Quit), self)) self.command(Quit)
} }
pub fn abort(&mut self) { pub fn abort(&mut self) {
// Only try to quit if we are not already broken match self.stream.get_ref().state() {
if !self.panic { ConnectionState::Ok | ConnectionState::Broken => {
self.panic = true; let _ = self.command(Quit);
let _ = self.command(Quit); let _ = self.stream.get_mut().shutdown(std::net::Shutdown::Both);
self.stream.get_mut().set_state(ConnectionState::Closed);
}
ConnectionState::Closed => {}
} }
let _ = self.stream.get_mut().shutdown(std::net::Shutdown::Both);
} }
/// Sets the underlying stream /// Sets the underlying stream
@@ -224,14 +212,11 @@ impl SmtpConnection {
while challenges > 0 && response.has_code(334) { while challenges > 0 && response.has_code(334) {
challenges -= 1; challenges -= 1;
response = try_smtp!( response = self.command(Auth::new_from_response(
self.command(Auth::new_from_response( mechanism,
mechanism, credentials.clone(),
credentials.clone(), &response,
&response, )?)?;
)?),
self
);
} }
if challenges == 0 { if challenges == 0 {
@@ -260,12 +245,17 @@ impl SmtpConnection {
/// Writes a string to the server /// Writes a string to the server
fn write(&mut self, string: &[u8]) -> Result<(), Error> { fn write(&mut self, string: &[u8]) -> Result<(), Error> {
self.stream.get_ref().state().verify()?;
self.stream.get_mut().set_state(ConnectionState::Broken);
self.stream self.stream
.get_mut() .get_mut()
.write_all(string) .write_all(string)
.map_err(error::network)?; .map_err(error::network)?;
self.stream.get_mut().flush().map_err(error::network)?; self.stream.get_mut().flush().map_err(error::network)?;
self.stream.get_mut().set_state(ConnectionState::Ok);
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string))); tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string)));
Ok(()) Ok(())
@@ -273,6 +263,9 @@ impl SmtpConnection {
/// Gets the SMTP response /// Gets the SMTP response
pub fn read_response(&mut self) -> Result<Response, Error> { pub fn read_response(&mut self) -> Result<Response, Error> {
self.stream.get_ref().state().verify()?;
self.stream.get_mut().set_state(ConnectionState::Broken);
let mut buffer = String::with_capacity(100); let mut buffer = String::with_capacity(100);
while self.stream.read_line(&mut buffer).map_err(error::network)? > 0 { while self.stream.read_line(&mut buffer).map_err(error::network)? > 0 {
@@ -280,6 +273,8 @@ impl SmtpConnection {
tracing::debug!("<< {}", escape_crlf(&buffer)); tracing::debug!("<< {}", escape_crlf(&buffer));
match parse_response(&buffer) { match parse_response(&buffer) {
Ok((_remaining, response)) => { Ok((_remaining, response)) => {
self.stream.get_mut().set_state(ConnectionState::Ok);
return if response.is_positive() { return if response.is_positive() {
Ok(response) Ok(response)
} else { } else {

View File

@@ -40,6 +40,7 @@ pub use self::{
connection::SmtpConnection, connection::SmtpConnection,
tls::{Certificate, CertificateStore, Tls, TlsParameters, TlsParametersBuilder}, tls::{Certificate, CertificateStore, Tls, TlsParameters, TlsParametersBuilder},
}; };
use super::{error, Error};
#[cfg(any(feature = "tokio1", feature = "async-std1"))] #[cfg(any(feature = "tokio1", feature = "async-std1"))]
mod async_connection; mod async_connection;
@@ -49,6 +50,23 @@ mod connection;
mod net; mod net;
mod tls; mod tls;
#[derive(Debug, Copy, Clone)]
enum ConnectionState {
Ok,
Broken,
Closed,
}
impl ConnectionState {
fn verify(&mut self) -> Result<(), Error> {
match self {
Self::Ok => Ok(()),
Self::Broken => Err(error::connection("connection broken")),
Self::Closed => Err(error::connection("connection closed")),
}
}
}
/// The codec used for transparency /// The codec used for transparency
#[derive(Debug)] #[derive(Debug)]
struct ClientCodec { struct ClientCodec {

View File

@@ -2,8 +2,7 @@
use std::sync::Arc; use std::sync::Arc;
use std::{ use std::{
io::{self, Read, Write}, io::{self, Read, Write},
mem, net::{IpAddr, Shutdown, SocketAddr, TcpStream, ToSocketAddrs},
net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, SocketAddrV4, TcpStream, ToSocketAddrs},
time::Duration, time::Duration,
}; };
@@ -17,12 +16,13 @@ use socket2::{Domain, Protocol, Type};
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
use super::InnerTlsParameters; use super::InnerTlsParameters;
use super::TlsParameters; use super::{ConnectionState, TlsParameters};
use crate::transport::smtp::{error, Error}; use crate::transport::smtp::{error, Error};
/// A network stream /// A network stream
pub struct NetworkStream { pub struct NetworkStream {
inner: InnerNetworkStream, inner: InnerNetworkStream,
state: ConnectionState,
} }
/// Represents the different types of underlying network streams /// Represents the different types of underlying network streams
@@ -40,53 +40,49 @@ enum InnerNetworkStream {
RustlsTls(StreamOwned<ClientConnection, TcpStream>), RustlsTls(StreamOwned<ClientConnection, TcpStream>),
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
BoringTls(SslStream<TcpStream>), BoringTls(SslStream<TcpStream>),
/// Can't be built
None,
} }
impl NetworkStream { impl NetworkStream {
fn new(inner: InnerNetworkStream) -> Self { fn new(inner: InnerNetworkStream) -> Self {
if let InnerNetworkStream::None = inner { NetworkStream {
debug_assert!(false, "InnerNetworkStream::None must never be built"); inner,
state: ConnectionState::Ok,
} }
}
NetworkStream { inner } pub(super) fn state(&self) -> ConnectionState {
self.state
}
pub(super) fn set_state(&mut self, state: ConnectionState) {
self.state = state;
} }
/// Returns peer's address /// Returns peer's address
pub fn peer_addr(&self) -> io::Result<SocketAddr> { pub fn peer_addr(&self) -> io::Result<SocketAddr> {
match self.inner { match &self.inner {
InnerNetworkStream::Tcp(ref s) => s.peer_addr(), InnerNetworkStream::Tcp(s) => s.peer_addr(),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref s) => s.get_ref().peer_addr(), InnerNetworkStream::NativeTls(s) => s.get_ref().peer_addr(),
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref s) => s.get_ref().peer_addr(), InnerNetworkStream::RustlsTls(s) => s.get_ref().peer_addr(),
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref s) => s.get_ref().peer_addr(), InnerNetworkStream::BoringTls(s) => s.get_ref().peer_addr(),
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(SocketAddr::V4(SocketAddrV4::new(
Ipv4Addr::new(127, 0, 0, 1),
80,
)))
}
} }
} }
/// Shutdowns the connection /// Shutdowns the connection
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> { pub fn shutdown(&mut self, how: Shutdown) -> io::Result<()> {
match self.inner { self.state = ConnectionState::Closed;
InnerNetworkStream::Tcp(ref s) => s.shutdown(how),
match &self.inner {
InnerNetworkStream::Tcp(s) => s.shutdown(how),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref s) => s.get_ref().shutdown(how), InnerNetworkStream::NativeTls(s) => s.get_ref().shutdown(how),
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref s) => s.get_ref().shutdown(how), InnerNetworkStream::RustlsTls(s) => s.get_ref().shutdown(how),
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref s) => s.get_ref().shutdown(how), InnerNetworkStream::BoringTls(s) => s.get_ref().shutdown(how),
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(())
}
} }
} }
@@ -139,13 +135,13 @@ impl NetworkStream {
let tcp_stream = try_connect(server, timeout, local_addr)?; let tcp_stream = try_connect(server, timeout, local_addr)?;
let mut stream = NetworkStream::new(InnerNetworkStream::Tcp(tcp_stream)); let mut stream = NetworkStream::new(InnerNetworkStream::Tcp(tcp_stream));
if let Some(tls_parameters) = tls_parameters { if let Some(tls_parameters) = tls_parameters {
stream.upgrade_tls(tls_parameters)?; stream = stream.upgrade_tls(tls_parameters)?;
} }
Ok(stream) Ok(stream)
} }
pub fn upgrade_tls(&mut self, tls_parameters: &TlsParameters) -> Result<(), Error> { pub fn upgrade_tls(self, tls_parameters: &TlsParameters) -> Result<Self, Error> {
match &self.inner { match self.inner {
#[cfg(not(any( #[cfg(not(any(
feature = "native-tls", feature = "native-tls",
feature = "rustls-tls", feature = "rustls-tls",
@@ -157,18 +153,14 @@ impl NetworkStream {
} }
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
InnerNetworkStream::Tcp(_) => { InnerNetworkStream::Tcp(tcp_stream) => {
// get owned TcpStream let inner = Self::upgrade_tls_impl(tcp_stream, tls_parameters)?;
let tcp_stream = mem::replace(&mut self.inner, InnerNetworkStream::None); Ok(Self {
let tcp_stream = match tcp_stream { inner,
InnerNetworkStream::Tcp(tcp_stream) => tcp_stream, state: ConnectionState::Ok,
_ => unreachable!(), })
};
self.inner = Self::upgrade_tls_impl(tcp_stream, tls_parameters)?;
Ok(())
} }
_ => Ok(()), _ => Ok(self),
} }
} }
@@ -208,7 +200,7 @@ impl NetworkStream {
} }
pub fn is_encrypted(&self) -> bool { pub fn is_encrypted(&self) -> bool {
match self.inner { match &self.inner {
InnerNetworkStream::Tcp(_) => false, InnerNetworkStream::Tcp(_) => false,
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(_) => true, InnerNetworkStream::NativeTls(_) => true,
@@ -216,10 +208,6 @@ impl NetworkStream {
InnerNetworkStream::RustlsTls(_) => true, InnerNetworkStream::RustlsTls(_) => true,
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(_) => true, InnerNetworkStream::BoringTls(_) => true,
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
false
}
} }
} }
@@ -249,105 +237,72 @@ impl NetworkStream {
.unwrap() .unwrap()
.to_der() .to_der()
.map_err(error::tls)?), .map_err(error::tls)?),
InnerNetworkStream::None => panic!("InnerNetworkStream::None must never be built"),
} }
} }
pub fn set_read_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> { pub fn set_read_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> {
match self.inner { match &mut self.inner {
InnerNetworkStream::Tcp(ref mut stream) => stream.set_read_timeout(duration), InnerNetworkStream::Tcp(stream) => stream.set_read_timeout(duration),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref mut stream) => { InnerNetworkStream::NativeTls(stream) => stream.get_ref().set_read_timeout(duration),
stream.get_ref().set_read_timeout(duration)
}
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref mut stream) => { InnerNetworkStream::RustlsTls(stream) => stream.get_ref().set_read_timeout(duration),
stream.get_ref().set_read_timeout(duration)
}
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref mut stream) => { InnerNetworkStream::BoringTls(stream) => stream.get_ref().set_read_timeout(duration),
stream.get_ref().set_read_timeout(duration)
}
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(())
}
} }
} }
/// Set write timeout for IO calls /// Set write timeout for IO calls
pub fn set_write_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> { pub fn set_write_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> {
match self.inner { match &mut self.inner {
InnerNetworkStream::Tcp(ref mut stream) => stream.set_write_timeout(duration), InnerNetworkStream::Tcp(stream) => stream.set_write_timeout(duration),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref mut stream) => { InnerNetworkStream::NativeTls(stream) => stream.get_ref().set_write_timeout(duration),
stream.get_ref().set_write_timeout(duration)
}
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref mut stream) => { InnerNetworkStream::RustlsTls(stream) => stream.get_ref().set_write_timeout(duration),
stream.get_ref().set_write_timeout(duration)
}
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref mut stream) => { InnerNetworkStream::BoringTls(stream) => stream.get_ref().set_write_timeout(duration),
stream.get_ref().set_write_timeout(duration)
}
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(())
}
} }
} }
} }
impl Read for NetworkStream { impl Read for NetworkStream {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> { fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
match self.inner { match &mut self.inner {
InnerNetworkStream::Tcp(ref mut s) => s.read(buf), InnerNetworkStream::Tcp(s) => s.read(buf),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref mut s) => s.read(buf), InnerNetworkStream::NativeTls(s) => s.read(buf),
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref mut s) => s.read(buf), InnerNetworkStream::RustlsTls(s) => s.read(buf),
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref mut s) => s.read(buf), InnerNetworkStream::BoringTls(s) => s.read(buf),
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(0)
}
} }
} }
} }
impl Write for NetworkStream { impl Write for NetworkStream {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> { fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
match self.inner { match &mut self.inner {
InnerNetworkStream::Tcp(ref mut s) => s.write(buf), InnerNetworkStream::Tcp(s) => s.write(buf),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref mut s) => s.write(buf), InnerNetworkStream::NativeTls(s) => s.write(buf),
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref mut s) => s.write(buf), InnerNetworkStream::RustlsTls(s) => s.write(buf),
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref mut s) => s.write(buf), InnerNetworkStream::BoringTls(s) => s.write(buf),
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(0)
}
} }
} }
fn flush(&mut self) -> io::Result<()> { fn flush(&mut self) -> io::Result<()> {
match self.inner { match &mut self.inner {
InnerNetworkStream::Tcp(ref mut s) => s.flush(), InnerNetworkStream::Tcp(s) => s.flush(),
#[cfg(feature = "native-tls")] #[cfg(feature = "native-tls")]
InnerNetworkStream::NativeTls(ref mut s) => s.flush(), InnerNetworkStream::NativeTls(s) => s.flush(),
#[cfg(feature = "rustls-tls")] #[cfg(feature = "rustls-tls")]
InnerNetworkStream::RustlsTls(ref mut s) => s.flush(), InnerNetworkStream::RustlsTls(s) => s.flush(),
#[cfg(feature = "boring-tls")] #[cfg(feature = "boring-tls")]
InnerNetworkStream::BoringTls(ref mut s) => s.flush(), InnerNetworkStream::BoringTls(s) => s.flush(),
InnerNetworkStream::None => {
debug_assert!(false, "InnerNetworkStream::None must never be built");
Ok(())
}
} }
} }
} }

View File

@@ -119,7 +119,7 @@ impl fmt::Debug for Error {
builder.field("kind", &self.inner.kind); builder.field("kind", &self.inner.kind);
if let Some(ref source) = self.inner.source { if let Some(source) = &self.inner.source {
builder.field("source", source); builder.field("source", source);
} }
@@ -129,22 +129,22 @@ impl fmt::Debug for Error {
impl fmt::Display for Error { impl fmt::Display for Error {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.inner.kind { match &self.inner.kind {
Kind::Response => f.write_str("response error")?, Kind::Response => f.write_str("response error")?,
Kind::Client => f.write_str("internal client error")?, Kind::Client => f.write_str("internal client error")?,
Kind::Network => f.write_str("network error")?, Kind::Network => f.write_str("network error")?,
Kind::Connection => f.write_str("Connection error")?, Kind::Connection => f.write_str("Connection error")?,
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
Kind::Tls => f.write_str("tls error")?, Kind::Tls => f.write_str("tls error")?,
Kind::Transient(ref code) => { Kind::Transient(code) => {
write!(f, "transient error ({code})")?; write!(f, "transient error ({code})")?;
} }
Kind::Permanent(ref code) => { Kind::Permanent(code) => {
write!(f, "permanent error ({code})")?; write!(f, "permanent error ({code})")?;
} }
}; };
if let Some(ref e) = self.inner.source { if let Some(e) = &self.inner.source {
write!(f, ": {e}")?; write!(f, ": {e}")?;
} }

View File

@@ -4,7 +4,6 @@ use std::{
collections::HashSet, collections::HashSet,
fmt::{self, Display, Formatter}, fmt::{self, Display, Formatter},
net::{Ipv4Addr, Ipv6Addr}, net::{Ipv4Addr, Ipv6Addr},
result::Result,
}; };
use crate::transport::smtp::{ use crate::transport::smtp::{
@@ -53,10 +52,10 @@ impl Default for ClientId {
impl Display for ClientId { impl Display for ClientId {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match *self { match self {
Self::Domain(ref value) => f.write_str(value), Self::Domain(value) => f.write_str(value),
Self::Ipv4(ref value) => write!(f, "[{value}]"), Self::Ipv4(value) => write!(f, "[{value}]"),
Self::Ipv6(ref value) => write!(f, "[IPv6:{value}]"), Self::Ipv6(value) => write!(f, "[IPv6:{value}]"),
} }
} }
} }
@@ -93,11 +92,11 @@ pub enum Extension {
impl Display for Extension { impl Display for Extension {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match *self { match self {
Extension::EightBitMime => f.write_str("8BITMIME"), Extension::EightBitMime => f.write_str("8BITMIME"),
Extension::SmtpUtfEight => f.write_str("SMTPUTF8"), Extension::SmtpUtfEight => f.write_str("SMTPUTF8"),
Extension::StartTls => f.write_str("STARTTLS"), Extension::StartTls => f.write_str("STARTTLS"),
Extension::Authentication(ref mechanism) => write!(f, "AUTH {mechanism}"), Extension::Authentication(mechanism) => write!(f, "AUTH {mechanism}"),
} }
} }
} }
@@ -227,16 +226,16 @@ pub enum MailParameter {
impl Display for MailParameter { impl Display for MailParameter {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match *self { match self {
MailParameter::Body(ref value) => write!(f, "BODY={value}"), MailParameter::Body(value) => write!(f, "BODY={value}"),
MailParameter::Size(size) => write!(f, "SIZE={size}"), MailParameter::Size(size) => write!(f, "SIZE={size}"),
MailParameter::SmtpUtfEight => f.write_str("SMTPUTF8"), MailParameter::SmtpUtfEight => f.write_str("SMTPUTF8"),
MailParameter::Other { MailParameter::Other {
ref keyword, keyword,
value: Some(ref value), value: Some(value),
} => write!(f, "{}={}", keyword, XText(value)), } => write!(f, "{}={}", keyword, XText(value)),
MailParameter::Other { MailParameter::Other {
ref keyword, keyword,
value: None, value: None,
} => f.write_str(keyword), } => f.write_str(keyword),
} }
@@ -277,13 +276,13 @@ pub enum RcptParameter {
impl Display for RcptParameter { impl Display for RcptParameter {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match *self { match &self {
RcptParameter::Other { RcptParameter::Other {
ref keyword, keyword,
value: Some(ref value), value: Some(value),
} => write!(f, "{keyword}={}", XText(value)), } => write!(f, "{keyword}={}", XText(value)),
RcptParameter::Other { RcptParameter::Other {
ref keyword, keyword,
value: None, value: None,
} => f.write_str(keyword), } => f.write_str(keyword),
} }

View File

@@ -5,7 +5,6 @@ use std::{
fmt::{Display, Formatter, Result}, fmt::{Display, Formatter, Result},
result, result,
str::FromStr, str::FromStr,
string::ToString,
}; };
use nom::{ use nom::{

View File

@@ -317,9 +317,9 @@ impl SmtpClient {
/// Handles encryption and authentication /// Handles encryption and authentication
pub fn connection(&self) -> Result<SmtpConnection, Error> { pub fn connection(&self) -> Result<SmtpConnection, Error> {
#[allow(clippy::match_single_binding)] #[allow(clippy::match_single_binding)]
let tls_parameters = match self.info.tls { let tls_parameters = match &self.info.tls {
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
Tls::Wrapper(ref tls_parameters) => Some(tls_parameters), Tls::Wrapper(tls_parameters) => Some(tls_parameters),
_ => None, _ => None,
}; };
@@ -333,14 +333,14 @@ impl SmtpClient {
)?; )?;
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
match self.info.tls { match &self.info.tls {
Tls::Opportunistic(ref tls_parameters) => { Tls::Opportunistic(tls_parameters) => {
if conn.can_starttls() { if conn.can_starttls() {
conn.starttls(tls_parameters, &self.info.hello_name)?; conn = conn.starttls(tls_parameters, &self.info.hello_name)?;
} }
} }
Tls::Required(ref tls_parameters) => { Tls::Required(tls_parameters) => {
conn.starttls(tls_parameters, &self.info.hello_name)?; conn = conn.starttls(tls_parameters, &self.info.hello_name)?;
} }
_ => (), _ => (),
} }