Compare commits
4 Commits
prepare-v0
...
refactor
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
efee4b5d72 | ||
|
|
57d7bf25cc | ||
|
|
c52c596458 | ||
|
|
53dee3e31f |
@@ -134,7 +134,7 @@ impl Executor for Tokio1Executor {
|
||||
#[allow(clippy::match_single_binding)]
|
||||
let tls_parameters = match 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,
|
||||
};
|
||||
#[allow(unused_mut)]
|
||||
@@ -149,13 +149,13 @@ impl Executor for Tokio1Executor {
|
||||
|
||||
#[cfg(any(feature = "tokio1-native-tls", feature = "tokio1-rustls-tls"))]
|
||||
match tls {
|
||||
Tls::Opportunistic(ref tls_parameters) => {
|
||||
Tls::Opportunistic(tls_parameters) => {
|
||||
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) => {
|
||||
conn.starttls(tls_parameters.clone(), hello_name).await?;
|
||||
Tls::Required(tls_parameters) => {
|
||||
conn = conn.starttls(tls_parameters.clone(), hello_name).await?;
|
||||
}
|
||||
_ => (),
|
||||
}
|
||||
@@ -231,7 +231,7 @@ impl Executor for AsyncStd1Executor {
|
||||
#[allow(clippy::match_single_binding)]
|
||||
let tls_parameters = match 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,
|
||||
};
|
||||
#[allow(unused_mut)]
|
||||
@@ -245,13 +245,13 @@ impl Executor for AsyncStd1Executor {
|
||||
|
||||
#[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))]
|
||||
match tls {
|
||||
Tls::Opportunistic(ref tls_parameters) => {
|
||||
Tls::Opportunistic(tls_parameters) => {
|
||||
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) => {
|
||||
conn.starttls(tls_parameters.clone(), hello_name).await?;
|
||||
Tls::Required(tls_parameters) => {
|
||||
conn = conn.starttls(tls_parameters.clone(), hello_name).await?;
|
||||
}
|
||||
_ => (),
|
||||
}
|
||||
|
||||
@@ -2,7 +2,6 @@ use std::{
|
||||
borrow::Cow,
|
||||
error::Error as StdError,
|
||||
fmt::{self, Display},
|
||||
iter::IntoIterator,
|
||||
time::SystemTime,
|
||||
};
|
||||
|
||||
|
||||
@@ -88,7 +88,7 @@ impl Mailbox {
|
||||
|
||||
impl Display for Mailbox {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> FmtResult {
|
||||
if let Some(ref name) = self.name {
|
||||
if let Some(name) = &self.name {
|
||||
let name = name.trim();
|
||||
if !name.is_empty() {
|
||||
write_word(f, name)?;
|
||||
|
||||
@@ -54,7 +54,7 @@ impl fmt::Debug for Error {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -70,7 +70,7 @@ impl fmt::Display for 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}")?;
|
||||
}
|
||||
|
||||
|
||||
@@ -52,7 +52,7 @@ impl fmt::Debug for Error {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -67,7 +67,7 @@ impl fmt::Display for 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}")?;
|
||||
}
|
||||
|
||||
|
||||
@@ -6,7 +6,7 @@ use futures_util::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
use super::async_net::AsyncTokioStream;
|
||||
#[cfg(feature = "tracing")]
|
||||
use super::escape_crlf;
|
||||
use super::{AsyncNetworkStream, ClientCodec, TlsParameters};
|
||||
use super::{AsyncNetworkStream, ClientCodec, ConnectionState, TlsParameters};
|
||||
use crate::{
|
||||
transport::smtp::{
|
||||
authentication::{Credentials, Mechanism},
|
||||
@@ -19,25 +19,11 @@ use crate::{
|
||||
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
|
||||
pub struct AsyncSmtpConnection {
|
||||
/// TCP stream between client and server
|
||||
/// Value is None before connection
|
||||
stream: BufReader<AsyncNetworkStream>,
|
||||
/// Panic state
|
||||
panic: bool,
|
||||
/// Information about the server
|
||||
server_info: ServerInfo,
|
||||
}
|
||||
@@ -126,7 +112,6 @@ impl AsyncSmtpConnection {
|
||||
let stream = BufReader::new(stream);
|
||||
let mut conn = AsyncSmtpConnection {
|
||||
stream,
|
||||
panic: false,
|
||||
server_info: ServerInfo::default(),
|
||||
};
|
||||
// TODO log
|
||||
@@ -170,30 +155,26 @@ impl AsyncSmtpConnection {
|
||||
mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
|
||||
}
|
||||
|
||||
try_smtp!(
|
||||
self.command(Mail::new(envelope.from().cloned(), mail_options))
|
||||
.await,
|
||||
self
|
||||
);
|
||||
self.command(Mail::new(envelope.from().cloned(), mail_options))
|
||||
.await?;
|
||||
|
||||
// Recipient
|
||||
for to_address in envelope.to() {
|
||||
try_smtp!(
|
||||
self.command(Rcpt::new(to_address.clone(), vec![])).await,
|
||||
self
|
||||
);
|
||||
self.command(Rcpt::new(to_address.clone(), vec![])).await?;
|
||||
}
|
||||
|
||||
// Data
|
||||
try_smtp!(self.command(Data).await, self);
|
||||
self.command(Data).await?;
|
||||
|
||||
// Message content
|
||||
let result = try_smtp!(self.message(email).await, self);
|
||||
Ok(result)
|
||||
self.message(email).await
|
||||
}
|
||||
|
||||
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 {
|
||||
@@ -208,18 +189,20 @@ impl AsyncSmtpConnection {
|
||||
/// [rfc8314]: https://www.rfc-editor.org/rfc/rfc8314
|
||||
#[allow(unused_variables)]
|
||||
pub async fn starttls(
|
||||
&mut self,
|
||||
mut self,
|
||||
tls_parameters: TlsParameters,
|
||||
hello_name: &ClientId,
|
||||
) -> Result<(), Error> {
|
||||
) -> Result<Self, Error> {
|
||||
if self.server_info.supports_feature(Extension::StartTls) {
|
||||
try_smtp!(self.command(Starttls).await, self);
|
||||
self.stream.get_mut().upgrade_tls(tls_parameters).await?;
|
||||
self.command(Starttls).await?;
|
||||
let stream = self.stream.into_inner();
|
||||
let stream = stream.upgrade_tls(tls_parameters).await?;
|
||||
self.stream = BufReader::new(stream);
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("connection encrypted");
|
||||
// Send EHLO again
|
||||
try_smtp!(self.ehlo(hello_name).await, self);
|
||||
Ok(())
|
||||
self.ehlo(hello_name).await?;
|
||||
Ok(self)
|
||||
} else {
|
||||
Err(error::client("STARTTLS is not supported on this server"))
|
||||
}
|
||||
@@ -227,22 +210,24 @@ impl AsyncSmtpConnection {
|
||||
|
||||
/// Send EHLO and update server info
|
||||
async fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> {
|
||||
let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())).await, self);
|
||||
self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self);
|
||||
let ehlo_response = self.command(Ehlo::new(hello_name.clone())).await?;
|
||||
self.server_info = ServerInfo::from_response(&ehlo_response)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
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) {
|
||||
// Only try to quit if we are not already broken
|
||||
if !self.panic {
|
||||
self.panic = true;
|
||||
let _ = self.command(Quit).await;
|
||||
match self.stream.get_ref().state() {
|
||||
ConnectionState::Ok | ConnectionState::Broken => {
|
||||
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
|
||||
@@ -279,15 +264,13 @@ impl AsyncSmtpConnection {
|
||||
|
||||
while challenges > 0 && response.has_code(334) {
|
||||
challenges -= 1;
|
||||
response = try_smtp!(
|
||||
self.command(Auth::new_from_response(
|
||||
response = self
|
||||
.command(Auth::new_from_response(
|
||||
mechanism,
|
||||
credentials.clone(),
|
||||
&response,
|
||||
)?)
|
||||
.await,
|
||||
self
|
||||
);
|
||||
.await?;
|
||||
}
|
||||
|
||||
if challenges == 0 {
|
||||
@@ -315,6 +298,9 @@ impl AsyncSmtpConnection {
|
||||
|
||||
/// Writes a string to the server
|
||||
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
|
||||
.get_mut()
|
||||
.write_all(string)
|
||||
@@ -326,6 +312,8 @@ impl AsyncSmtpConnection {
|
||||
.await
|
||||
.map_err(error::network)?;
|
||||
|
||||
self.stream.get_mut().set_state(ConnectionState::Ok);
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string)));
|
||||
Ok(())
|
||||
@@ -333,6 +321,9 @@ impl AsyncSmtpConnection {
|
||||
|
||||
/// Gets the SMTP response
|
||||
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);
|
||||
|
||||
while self
|
||||
@@ -346,6 +337,8 @@ impl AsyncSmtpConnection {
|
||||
tracing::debug!("<< {}", escape_crlf(&buffer));
|
||||
match parse_response(&buffer) {
|
||||
Ok((_remaining, response)) => {
|
||||
self.stream.get_mut().set_state(ConnectionState::Ok);
|
||||
|
||||
return if response.is_positive() {
|
||||
Ok(response)
|
||||
} else {
|
||||
@@ -353,7 +346,7 @@ impl AsyncSmtpConnection {
|
||||
response.code(),
|
||||
Some(response.message().collect()),
|
||||
))
|
||||
}
|
||||
};
|
||||
}
|
||||
Err(nom::Err::Failure(e)) => {
|
||||
return Err(error::response(e.to_string()));
|
||||
|
||||
@@ -11,8 +11,7 @@ use async_native_tls::TlsStream as AsyncStd1TlsStream;
|
||||
#[cfg(feature = "async-std1")]
|
||||
use async_std::net::{TcpStream as AsyncStd1TcpStream, ToSocketAddrs as AsyncStd1ToSocketAddrs};
|
||||
use futures_io::{
|
||||
AsyncRead as FuturesAsyncRead, AsyncWrite as FuturesAsyncWrite, Error as IoError, ErrorKind,
|
||||
Result as IoResult,
|
||||
AsyncRead as FuturesAsyncRead, AsyncWrite as FuturesAsyncWrite, Result as IoResult,
|
||||
};
|
||||
#[cfg(feature = "async-std1-rustls-tls")]
|
||||
use futures_rustls::client::TlsStream as AsyncStd1RustlsTlsStream;
|
||||
@@ -40,7 +39,7 @@ use tokio1_rustls::client::TlsStream as Tokio1RustlsTlsStream;
|
||||
feature = "async-std1-rustls-tls"
|
||||
))]
|
||||
use super::InnerTlsParameters;
|
||||
use super::TlsParameters;
|
||||
use super::{ConnectionState, TlsParameters};
|
||||
#[cfg(feature = "tokio1")]
|
||||
use crate::transport::smtp::client::net::resolved_address_filter;
|
||||
use crate::transport::smtp::{error, Error};
|
||||
@@ -49,6 +48,7 @@ use crate::transport::smtp::{error, Error};
|
||||
#[derive(Debug)]
|
||||
pub struct AsyncNetworkStream {
|
||||
inner: InnerAsyncNetworkStream,
|
||||
state: ConnectionState,
|
||||
}
|
||||
|
||||
#[cfg(feature = "tokio1")]
|
||||
@@ -91,45 +91,43 @@ enum InnerAsyncNetworkStream {
|
||||
/// Encrypted Tokio 1.x TCP stream
|
||||
#[cfg(feature = "async-std1-rustls-tls")]
|
||||
AsyncStd1RustlsTls(AsyncStd1RustlsTlsStream<AsyncStd1TcpStream>),
|
||||
/// Can't be built
|
||||
None,
|
||||
}
|
||||
|
||||
impl AsyncNetworkStream {
|
||||
fn new(inner: InnerAsyncNetworkStream) -> Self {
|
||||
if let InnerAsyncNetworkStream::None = inner {
|
||||
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
|
||||
AsyncNetworkStream {
|
||||
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
|
||||
pub fn peer_addr(&self) -> IoResult<SocketAddr> {
|
||||
match self.inner {
|
||||
match &self.inner {
|
||||
#[cfg(feature = "tokio1")]
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(ref s) => s.peer_addr(),
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(s) => s.peer_addr(),
|
||||
#[cfg(feature = "tokio1-native-tls")]
|
||||
InnerAsyncNetworkStream::Tokio1NativeTls(ref s) => {
|
||||
InnerAsyncNetworkStream::Tokio1NativeTls(s) => {
|
||||
s.get_ref().get_ref().get_ref().peer_addr()
|
||||
}
|
||||
#[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")]
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(ref s) => s.get_ref().peer_addr(),
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(s) => s.get_ref().peer_addr(),
|
||||
#[cfg(feature = "async-std1")]
|
||||
InnerAsyncNetworkStream::AsyncStd1Tcp(ref s) => s.peer_addr(),
|
||||
InnerAsyncNetworkStream::AsyncStd1Tcp(s) => s.peer_addr(),
|
||||
#[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")]
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref 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",
|
||||
))
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => s.get_ref().0.peer_addr(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -199,7 +197,7 @@ impl AsyncNetworkStream {
|
||||
let mut stream =
|
||||
AsyncNetworkStream::new(InnerAsyncNetworkStream::Tokio1Tcp(Box::new(tcp_stream)));
|
||||
if let Some(tls_parameters) = tls_parameters {
|
||||
stream.upgrade_tls(tls_parameters).await?;
|
||||
stream = stream.upgrade_tls(tls_parameters).await?;
|
||||
}
|
||||
Ok(stream)
|
||||
}
|
||||
@@ -250,13 +248,13 @@ impl AsyncNetworkStream {
|
||||
|
||||
let mut stream = AsyncNetworkStream::new(InnerAsyncNetworkStream::AsyncStd1Tcp(tcp_stream));
|
||||
if let Some(tls_parameters) = tls_parameters {
|
||||
stream.upgrade_tls(tls_parameters).await?;
|
||||
stream = stream.upgrade_tls(tls_parameters).await?;
|
||||
}
|
||||
Ok(stream)
|
||||
}
|
||||
|
||||
pub async fn upgrade_tls(&mut self, tls_parameters: TlsParameters) -> Result<(), Error> {
|
||||
match &self.inner {
|
||||
pub async fn upgrade_tls(self, tls_parameters: TlsParameters) -> Result<Self, Error> {
|
||||
match self.inner {
|
||||
#[cfg(all(
|
||||
feature = "tokio1",
|
||||
not(any(
|
||||
@@ -275,18 +273,14 @@ impl AsyncNetworkStream {
|
||||
feature = "tokio1-rustls-tls",
|
||||
feature = "tokio1-boring-tls"
|
||||
))]
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(_) => {
|
||||
// get owned TcpStream
|
||||
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)
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(tcp_stream) => {
|
||||
let inner = Self::upgrade_tokio1_tls(tcp_stream, tls_parameters)
|
||||
.await
|
||||
.map_err(error::connection)?;
|
||||
Ok(())
|
||||
Ok(Self {
|
||||
inner,
|
||||
state: ConnectionState::Ok,
|
||||
})
|
||||
}
|
||||
#[cfg(all(
|
||||
feature = "async-std1",
|
||||
@@ -298,20 +292,16 @@ impl AsyncNetworkStream {
|
||||
}
|
||||
|
||||
#[cfg(any(feature = "async-std1-native-tls", feature = "async-std1-rustls-tls"))]
|
||||
InnerAsyncNetworkStream::AsyncStd1Tcp(_) => {
|
||||
// get owned TcpStream
|
||||
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)
|
||||
InnerAsyncNetworkStream::AsyncStd1Tcp(tcp_stream) => {
|
||||
let inner = Self::upgrade_asyncstd1_tls(tcp_stream, tls_parameters)
|
||||
.await
|
||||
.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 {
|
||||
match self.inner {
|
||||
match &self.inner {
|
||||
#[cfg(feature = "tokio1")]
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(_) => false,
|
||||
#[cfg(feature = "tokio1-native-tls")]
|
||||
@@ -460,7 +450,6 @@ impl AsyncNetworkStream {
|
||||
InnerAsyncNetworkStream::AsyncStd1NativeTls(_) => true,
|
||||
#[cfg(feature = "async-std1-rustls-tls")]
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(_) => true,
|
||||
InnerAsyncNetworkStream::None => false,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -509,7 +498,6 @@ impl AsyncNetworkStream {
|
||||
.first()
|
||||
.unwrap()
|
||||
.to_vec()),
|
||||
InnerAsyncNetworkStream::None => panic!("InnerNetworkStream::None must never be built"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -520,9 +508,9 @@ impl FuturesAsyncRead for AsyncNetworkStream {
|
||||
cx: &mut Context<'_>,
|
||||
buf: &mut [u8],
|
||||
) -> Poll<IoResult<usize>> {
|
||||
match self.inner {
|
||||
match &mut self.inner {
|
||||
#[cfg(feature = "tokio1")]
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(ref mut s) => {
|
||||
InnerAsyncNetworkStream::Tokio1Tcp(s) => {
|
||||
let mut b = Tokio1ReadBuf::new(buf);
|
||||
match Pin::new(s).poll_read(cx, &mut b) {
|
||||
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
|
||||
@@ -531,7 +519,7 @@ impl FuturesAsyncRead for AsyncNetworkStream {
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "tokio1-native-tls")]
|
||||
InnerAsyncNetworkStream::Tokio1NativeTls(ref mut s) => {
|
||||
InnerAsyncNetworkStream::Tokio1NativeTls(s) => {
|
||||
let mut b = Tokio1ReadBuf::new(buf);
|
||||
match Pin::new(s).poll_read(cx, &mut b) {
|
||||
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
|
||||
@@ -540,7 +528,7 @@ impl FuturesAsyncRead for AsyncNetworkStream {
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "tokio1-rustls-tls")]
|
||||
InnerAsyncNetworkStream::Tokio1RustlsTls(ref mut s) => {
|
||||
InnerAsyncNetworkStream::Tokio1RustlsTls(s) => {
|
||||
let mut b = Tokio1ReadBuf::new(buf);
|
||||
match Pin::new(s).poll_read(cx, &mut b) {
|
||||
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
|
||||
@@ -549,7 +537,7 @@ impl FuturesAsyncRead for AsyncNetworkStream {
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "tokio1-boring-tls")]
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => {
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(s) => {
|
||||
let mut b = Tokio1ReadBuf::new(buf);
|
||||
match Pin::new(s).poll_read(cx, &mut b) {
|
||||
Poll::Ready(Ok(())) => Poll::Ready(Ok(b.filled().len())),
|
||||
@@ -558,19 +546,11 @@ impl FuturesAsyncRead for AsyncNetworkStream {
|
||||
}
|
||||
}
|
||||
#[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")]
|
||||
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref mut s) => {
|
||||
Pin::new(s).poll_read(cx, buf)
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => Pin::new(s).poll_read(cx, buf),
|
||||
#[cfg(feature = "async-std1-rustls-tls")]
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => {
|
||||
Pin::new(s).poll_read(cx, buf)
|
||||
}
|
||||
InnerAsyncNetworkStream::None => {
|
||||
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
|
||||
Poll::Ready(Ok(0))
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_read(cx, buf),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -581,75 +561,61 @@ impl FuturesAsyncWrite for AsyncNetworkStream {
|
||||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<IoResult<usize>> {
|
||||
match self.inner {
|
||||
match &mut self.inner {
|
||||
#[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")]
|
||||
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")]
|
||||
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")]
|
||||
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")]
|
||||
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")]
|
||||
InnerAsyncNetworkStream::AsyncStd1NativeTls(ref mut s) => {
|
||||
Pin::new(s).poll_write(cx, buf)
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1NativeTls(s) => Pin::new(s).poll_write(cx, buf),
|
||||
#[cfg(feature = "async-std1-rustls-tls")]
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => {
|
||||
Pin::new(s).poll_write(cx, buf)
|
||||
}
|
||||
InnerAsyncNetworkStream::None => {
|
||||
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
|
||||
Poll::Ready(Ok(0))
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_write(cx, buf),
|
||||
}
|
||||
}
|
||||
|
||||
fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<IoResult<()>> {
|
||||
match self.inner {
|
||||
match &mut self.inner {
|
||||
#[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")]
|
||||
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")]
|
||||
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")]
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => Pin::new(s).poll_flush(cx),
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(s) => Pin::new(s).poll_flush(cx),
|
||||
#[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")]
|
||||
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")]
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => Pin::new(s).poll_flush(cx),
|
||||
InnerAsyncNetworkStream::None => {
|
||||
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_flush(cx),
|
||||
}
|
||||
}
|
||||
|
||||
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")]
|
||||
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")]
|
||||
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")]
|
||||
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")]
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(ref mut s) => Pin::new(s).poll_shutdown(cx),
|
||||
InnerAsyncNetworkStream::Tokio1BoringTls(s) => Pin::new(s).poll_shutdown(cx),
|
||||
#[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")]
|
||||
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")]
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(ref mut s) => Pin::new(s).poll_close(cx),
|
||||
InnerAsyncNetworkStream::None => {
|
||||
debug_assert!(false, "InnerAsyncNetworkStream::None must never be built");
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
InnerAsyncNetworkStream::AsyncStd1RustlsTls(s) => Pin::new(s).poll_close(cx),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::{
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
use super::escape_crlf;
|
||||
use super::{ClientCodec, NetworkStream, TlsParameters};
|
||||
use super::{ClientCodec, ConnectionState, NetworkStream, TlsParameters};
|
||||
use crate::{
|
||||
address::Envelope,
|
||||
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
|
||||
pub struct SmtpConnection {
|
||||
/// TCP stream between client and server
|
||||
/// Value is None before connection
|
||||
stream: BufReader<NetworkStream>,
|
||||
/// Panic state
|
||||
panic: bool,
|
||||
/// Information about the server
|
||||
server_info: ServerInfo,
|
||||
}
|
||||
@@ -65,7 +51,6 @@ impl SmtpConnection {
|
||||
let stream = BufReader::new(stream);
|
||||
let mut conn = SmtpConnection {
|
||||
stream,
|
||||
panic: false,
|
||||
server_info: ServerInfo::default(),
|
||||
};
|
||||
conn.set_timeout(timeout).map_err(error::network)?;
|
||||
@@ -110,26 +95,25 @@ impl SmtpConnection {
|
||||
mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
|
||||
}
|
||||
|
||||
try_smtp!(
|
||||
self.command(Mail::new(envelope.from().cloned(), mail_options)),
|
||||
self
|
||||
);
|
||||
self.command(Mail::new(envelope.from().cloned(), mail_options))?;
|
||||
|
||||
// Recipient
|
||||
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
|
||||
try_smtp!(self.command(Data), self);
|
||||
self.command(Data)?;
|
||||
|
||||
// Message content
|
||||
let result = try_smtp!(self.message(email), self);
|
||||
Ok(result)
|
||||
self.message(email)
|
||||
}
|
||||
|
||||
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 {
|
||||
@@ -138,20 +122,22 @@ impl SmtpConnection {
|
||||
|
||||
#[allow(unused_variables)]
|
||||
pub fn starttls(
|
||||
&mut self,
|
||||
mut self,
|
||||
tls_parameters: &TlsParameters,
|
||||
hello_name: &ClientId,
|
||||
) -> Result<(), Error> {
|
||||
) -> Result<Self, Error> {
|
||||
if self.server_info.supports_feature(Extension::StartTls) {
|
||||
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
|
||||
{
|
||||
try_smtp!(self.command(Starttls), self);
|
||||
self.stream.get_mut().upgrade_tls(tls_parameters)?;
|
||||
self.command(Starttls)?;
|
||||
let stream = self.stream.into_inner();
|
||||
let stream = stream.upgrade_tls(tls_parameters)?;
|
||||
self.stream = BufReader::new(stream);
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("connection encrypted");
|
||||
// Send EHLO again
|
||||
try_smtp!(self.ehlo(hello_name), self);
|
||||
Ok(())
|
||||
self.ehlo(hello_name)?;
|
||||
Ok(self)
|
||||
}
|
||||
#[cfg(not(any(
|
||||
feature = "native-tls",
|
||||
@@ -168,22 +154,24 @@ impl SmtpConnection {
|
||||
|
||||
/// Send EHLO and update server info
|
||||
fn ehlo(&mut self, hello_name: &ClientId) -> Result<(), Error> {
|
||||
let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())), self);
|
||||
self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self);
|
||||
let ehlo_response = self.command(Ehlo::new(hello_name.clone()))?;
|
||||
self.server_info = ServerInfo::from_response(&ehlo_response)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn quit(&mut self) -> Result<Response, Error> {
|
||||
Ok(try_smtp!(self.command(Quit), self))
|
||||
self.command(Quit)
|
||||
}
|
||||
|
||||
pub fn abort(&mut self) {
|
||||
// Only try to quit if we are not already broken
|
||||
if !self.panic {
|
||||
self.panic = true;
|
||||
let _ = self.command(Quit);
|
||||
match self.stream.get_ref().state() {
|
||||
ConnectionState::Ok | ConnectionState::Broken => {
|
||||
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
|
||||
@@ -224,14 +212,11 @@ impl SmtpConnection {
|
||||
|
||||
while challenges > 0 && response.has_code(334) {
|
||||
challenges -= 1;
|
||||
response = try_smtp!(
|
||||
self.command(Auth::new_from_response(
|
||||
mechanism,
|
||||
credentials.clone(),
|
||||
&response,
|
||||
)?),
|
||||
self
|
||||
);
|
||||
response = self.command(Auth::new_from_response(
|
||||
mechanism,
|
||||
credentials.clone(),
|
||||
&response,
|
||||
)?)?;
|
||||
}
|
||||
|
||||
if challenges == 0 {
|
||||
@@ -260,12 +245,17 @@ impl SmtpConnection {
|
||||
|
||||
/// Writes a string to the server
|
||||
fn write(&mut self, string: &[u8]) -> Result<(), Error> {
|
||||
self.stream.get_ref().state().verify()?;
|
||||
self.stream.get_mut().set_state(ConnectionState::Broken);
|
||||
|
||||
self.stream
|
||||
.get_mut()
|
||||
.write_all(string)
|
||||
.map_err(error::network)?;
|
||||
self.stream.get_mut().flush().map_err(error::network)?;
|
||||
|
||||
self.stream.get_mut().set_state(ConnectionState::Ok);
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
tracing::debug!("Wrote: {}", escape_crlf(&String::from_utf8_lossy(string)));
|
||||
Ok(())
|
||||
@@ -273,6 +263,9 @@ impl SmtpConnection {
|
||||
|
||||
/// Gets the SMTP response
|
||||
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);
|
||||
|
||||
while self.stream.read_line(&mut buffer).map_err(error::network)? > 0 {
|
||||
@@ -280,6 +273,8 @@ impl SmtpConnection {
|
||||
tracing::debug!("<< {}", escape_crlf(&buffer));
|
||||
match parse_response(&buffer) {
|
||||
Ok((_remaining, response)) => {
|
||||
self.stream.get_mut().set_state(ConnectionState::Ok);
|
||||
|
||||
return if response.is_positive() {
|
||||
Ok(response)
|
||||
} else {
|
||||
|
||||
@@ -40,6 +40,7 @@ pub use self::{
|
||||
connection::SmtpConnection,
|
||||
tls::{Certificate, CertificateStore, Tls, TlsParameters, TlsParametersBuilder},
|
||||
};
|
||||
use super::{error, Error};
|
||||
|
||||
#[cfg(any(feature = "tokio1", feature = "async-std1"))]
|
||||
mod async_connection;
|
||||
@@ -49,6 +50,23 @@ mod connection;
|
||||
mod net;
|
||||
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
|
||||
#[derive(Debug)]
|
||||
struct ClientCodec {
|
||||
|
||||
@@ -2,8 +2,7 @@
|
||||
use std::sync::Arc;
|
||||
use std::{
|
||||
io::{self, Read, Write},
|
||||
mem,
|
||||
net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, SocketAddrV4, TcpStream, ToSocketAddrs},
|
||||
net::{IpAddr, Shutdown, SocketAddr, TcpStream, ToSocketAddrs},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
@@ -17,12 +16,13 @@ use socket2::{Domain, Protocol, Type};
|
||||
|
||||
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
|
||||
use super::InnerTlsParameters;
|
||||
use super::TlsParameters;
|
||||
use super::{ConnectionState, TlsParameters};
|
||||
use crate::transport::smtp::{error, Error};
|
||||
|
||||
/// A network stream
|
||||
pub struct NetworkStream {
|
||||
inner: InnerNetworkStream,
|
||||
state: ConnectionState,
|
||||
}
|
||||
|
||||
/// Represents the different types of underlying network streams
|
||||
@@ -40,53 +40,49 @@ enum InnerNetworkStream {
|
||||
RustlsTls(StreamOwned<ClientConnection, TcpStream>),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
BoringTls(SslStream<TcpStream>),
|
||||
/// Can't be built
|
||||
None,
|
||||
}
|
||||
|
||||
impl NetworkStream {
|
||||
fn new(inner: InnerNetworkStream) -> Self {
|
||||
if let InnerNetworkStream::None = inner {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
NetworkStream {
|
||||
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
|
||||
pub fn peer_addr(&self) -> io::Result<SocketAddr> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref s) => s.peer_addr(),
|
||||
match &self.inner {
|
||||
InnerNetworkStream::Tcp(s) => s.peer_addr(),
|
||||
#[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")]
|
||||
InnerNetworkStream::RustlsTls(ref s) => s.get_ref().peer_addr(),
|
||||
InnerNetworkStream::RustlsTls(s) => s.get_ref().peer_addr(),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref 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,
|
||||
)))
|
||||
}
|
||||
InnerNetworkStream::BoringTls(s) => s.get_ref().peer_addr(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Shutdowns the connection
|
||||
pub fn shutdown(&self, how: Shutdown) -> io::Result<()> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref s) => s.shutdown(how),
|
||||
pub fn shutdown(&mut self, how: Shutdown) -> io::Result<()> {
|
||||
self.state = ConnectionState::Closed;
|
||||
|
||||
match &self.inner {
|
||||
InnerNetworkStream::Tcp(s) => s.shutdown(how),
|
||||
#[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")]
|
||||
InnerNetworkStream::RustlsTls(ref s) => s.get_ref().shutdown(how),
|
||||
InnerNetworkStream::RustlsTls(s) => s.get_ref().shutdown(how),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref s) => s.get_ref().shutdown(how),
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
Ok(())
|
||||
}
|
||||
InnerNetworkStream::BoringTls(s) => s.get_ref().shutdown(how),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -139,13 +135,13 @@ impl NetworkStream {
|
||||
let tcp_stream = try_connect(server, timeout, local_addr)?;
|
||||
let mut stream = NetworkStream::new(InnerNetworkStream::Tcp(tcp_stream));
|
||||
if let Some(tls_parameters) = tls_parameters {
|
||||
stream.upgrade_tls(tls_parameters)?;
|
||||
stream = stream.upgrade_tls(tls_parameters)?;
|
||||
}
|
||||
Ok(stream)
|
||||
}
|
||||
|
||||
pub fn upgrade_tls(&mut self, tls_parameters: &TlsParameters) -> Result<(), Error> {
|
||||
match &self.inner {
|
||||
pub fn upgrade_tls(self, tls_parameters: &TlsParameters) -> Result<Self, Error> {
|
||||
match self.inner {
|
||||
#[cfg(not(any(
|
||||
feature = "native-tls",
|
||||
feature = "rustls-tls",
|
||||
@@ -157,18 +153,14 @@ impl NetworkStream {
|
||||
}
|
||||
|
||||
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
|
||||
InnerNetworkStream::Tcp(_) => {
|
||||
// get owned TcpStream
|
||||
let tcp_stream = mem::replace(&mut self.inner, InnerNetworkStream::None);
|
||||
let tcp_stream = match tcp_stream {
|
||||
InnerNetworkStream::Tcp(tcp_stream) => tcp_stream,
|
||||
_ => unreachable!(),
|
||||
};
|
||||
|
||||
self.inner = Self::upgrade_tls_impl(tcp_stream, tls_parameters)?;
|
||||
Ok(())
|
||||
InnerNetworkStream::Tcp(tcp_stream) => {
|
||||
let inner = Self::upgrade_tls_impl(tcp_stream, tls_parameters)?;
|
||||
Ok(Self {
|
||||
inner,
|
||||
state: ConnectionState::Ok,
|
||||
})
|
||||
}
|
||||
_ => Ok(()),
|
||||
_ => Ok(self),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -208,7 +200,7 @@ impl NetworkStream {
|
||||
}
|
||||
|
||||
pub fn is_encrypted(&self) -> bool {
|
||||
match self.inner {
|
||||
match &self.inner {
|
||||
InnerNetworkStream::Tcp(_) => false,
|
||||
#[cfg(feature = "native-tls")]
|
||||
InnerNetworkStream::NativeTls(_) => true,
|
||||
@@ -216,10 +208,6 @@ impl NetworkStream {
|
||||
InnerNetworkStream::RustlsTls(_) => true,
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(_) => true,
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -249,105 +237,72 @@ impl NetworkStream {
|
||||
.unwrap()
|
||||
.to_der()
|
||||
.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<()> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref mut stream) => stream.set_read_timeout(duration),
|
||||
match &mut self.inner {
|
||||
InnerNetworkStream::Tcp(stream) => stream.set_read_timeout(duration),
|
||||
#[cfg(feature = "native-tls")]
|
||||
InnerNetworkStream::NativeTls(ref mut stream) => {
|
||||
stream.get_ref().set_read_timeout(duration)
|
||||
}
|
||||
InnerNetworkStream::NativeTls(stream) => stream.get_ref().set_read_timeout(duration),
|
||||
#[cfg(feature = "rustls-tls")]
|
||||
InnerNetworkStream::RustlsTls(ref mut stream) => {
|
||||
stream.get_ref().set_read_timeout(duration)
|
||||
}
|
||||
InnerNetworkStream::RustlsTls(stream) => stream.get_ref().set_read_timeout(duration),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref mut stream) => {
|
||||
stream.get_ref().set_read_timeout(duration)
|
||||
}
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
Ok(())
|
||||
}
|
||||
InnerNetworkStream::BoringTls(stream) => stream.get_ref().set_read_timeout(duration),
|
||||
}
|
||||
}
|
||||
|
||||
/// Set write timeout for IO calls
|
||||
pub fn set_write_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref mut stream) => stream.set_write_timeout(duration),
|
||||
match &mut self.inner {
|
||||
InnerNetworkStream::Tcp(stream) => stream.set_write_timeout(duration),
|
||||
|
||||
#[cfg(feature = "native-tls")]
|
||||
InnerNetworkStream::NativeTls(ref mut stream) => {
|
||||
stream.get_ref().set_write_timeout(duration)
|
||||
}
|
||||
InnerNetworkStream::NativeTls(stream) => stream.get_ref().set_write_timeout(duration),
|
||||
#[cfg(feature = "rustls-tls")]
|
||||
InnerNetworkStream::RustlsTls(ref mut stream) => {
|
||||
stream.get_ref().set_write_timeout(duration)
|
||||
}
|
||||
InnerNetworkStream::RustlsTls(stream) => stream.get_ref().set_write_timeout(duration),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref mut stream) => {
|
||||
stream.get_ref().set_write_timeout(duration)
|
||||
}
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
Ok(())
|
||||
}
|
||||
InnerNetworkStream::BoringTls(stream) => stream.get_ref().set_write_timeout(duration),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Read for NetworkStream {
|
||||
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref mut s) => s.read(buf),
|
||||
match &mut self.inner {
|
||||
InnerNetworkStream::Tcp(s) => s.read(buf),
|
||||
#[cfg(feature = "native-tls")]
|
||||
InnerNetworkStream::NativeTls(ref mut s) => s.read(buf),
|
||||
InnerNetworkStream::NativeTls(s) => s.read(buf),
|
||||
#[cfg(feature = "rustls-tls")]
|
||||
InnerNetworkStream::RustlsTls(ref mut s) => s.read(buf),
|
||||
InnerNetworkStream::RustlsTls(s) => s.read(buf),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref mut s) => s.read(buf),
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
Ok(0)
|
||||
}
|
||||
InnerNetworkStream::BoringTls(s) => s.read(buf),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for NetworkStream {
|
||||
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref mut s) => s.write(buf),
|
||||
match &mut self.inner {
|
||||
InnerNetworkStream::Tcp(s) => s.write(buf),
|
||||
#[cfg(feature = "native-tls")]
|
||||
InnerNetworkStream::NativeTls(ref mut s) => s.write(buf),
|
||||
InnerNetworkStream::NativeTls(s) => s.write(buf),
|
||||
#[cfg(feature = "rustls-tls")]
|
||||
InnerNetworkStream::RustlsTls(ref mut s) => s.write(buf),
|
||||
InnerNetworkStream::RustlsTls(s) => s.write(buf),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref mut s) => s.write(buf),
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
Ok(0)
|
||||
}
|
||||
InnerNetworkStream::BoringTls(s) => s.write(buf),
|
||||
}
|
||||
}
|
||||
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
match self.inner {
|
||||
InnerNetworkStream::Tcp(ref mut s) => s.flush(),
|
||||
match &mut self.inner {
|
||||
InnerNetworkStream::Tcp(s) => s.flush(),
|
||||
#[cfg(feature = "native-tls")]
|
||||
InnerNetworkStream::NativeTls(ref mut s) => s.flush(),
|
||||
InnerNetworkStream::NativeTls(s) => s.flush(),
|
||||
#[cfg(feature = "rustls-tls")]
|
||||
InnerNetworkStream::RustlsTls(ref mut s) => s.flush(),
|
||||
InnerNetworkStream::RustlsTls(s) => s.flush(),
|
||||
#[cfg(feature = "boring-tls")]
|
||||
InnerNetworkStream::BoringTls(ref mut s) => s.flush(),
|
||||
InnerNetworkStream::None => {
|
||||
debug_assert!(false, "InnerNetworkStream::None must never be built");
|
||||
Ok(())
|
||||
}
|
||||
InnerNetworkStream::BoringTls(s) => s.flush(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -119,7 +119,7 @@ impl fmt::Debug for Error {
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
@@ -129,22 +129,22 @@ impl fmt::Debug for Error {
|
||||
|
||||
impl fmt::Display for Error {
|
||||
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::Client => f.write_str("internal client error")?,
|
||||
Kind::Network => f.write_str("network error")?,
|
||||
Kind::Connection => f.write_str("Connection error")?,
|
||||
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
|
||||
Kind::Tls => f.write_str("tls error")?,
|
||||
Kind::Transient(ref code) => {
|
||||
Kind::Transient(code) => {
|
||||
write!(f, "transient error ({code})")?;
|
||||
}
|
||||
Kind::Permanent(ref code) => {
|
||||
Kind::Permanent(code) => {
|
||||
write!(f, "permanent error ({code})")?;
|
||||
}
|
||||
};
|
||||
|
||||
if let Some(ref e) = self.inner.source {
|
||||
if let Some(e) = &self.inner.source {
|
||||
write!(f, ": {e}")?;
|
||||
}
|
||||
|
||||
|
||||
@@ -4,7 +4,6 @@ use std::{
|
||||
collections::HashSet,
|
||||
fmt::{self, Display, Formatter},
|
||||
net::{Ipv4Addr, Ipv6Addr},
|
||||
result::Result,
|
||||
};
|
||||
|
||||
use crate::transport::smtp::{
|
||||
@@ -53,10 +52,10 @@ impl Default for ClientId {
|
||||
|
||||
impl Display for ClientId {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
|
||||
match *self {
|
||||
Self::Domain(ref value) => f.write_str(value),
|
||||
Self::Ipv4(ref value) => write!(f, "[{value}]"),
|
||||
Self::Ipv6(ref value) => write!(f, "[IPv6:{value}]"),
|
||||
match self {
|
||||
Self::Domain(value) => f.write_str(value),
|
||||
Self::Ipv4(value) => write!(f, "[{value}]"),
|
||||
Self::Ipv6(value) => write!(f, "[IPv6:{value}]"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -93,11 +92,11 @@ pub enum Extension {
|
||||
|
||||
impl Display for Extension {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
|
||||
match *self {
|
||||
match self {
|
||||
Extension::EightBitMime => f.write_str("8BITMIME"),
|
||||
Extension::SmtpUtfEight => f.write_str("SMTPUTF8"),
|
||||
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 {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
|
||||
match *self {
|
||||
MailParameter::Body(ref value) => write!(f, "BODY={value}"),
|
||||
match self {
|
||||
MailParameter::Body(value) => write!(f, "BODY={value}"),
|
||||
MailParameter::Size(size) => write!(f, "SIZE={size}"),
|
||||
MailParameter::SmtpUtfEight => f.write_str("SMTPUTF8"),
|
||||
MailParameter::Other {
|
||||
ref keyword,
|
||||
value: Some(ref value),
|
||||
keyword,
|
||||
value: Some(value),
|
||||
} => write!(f, "{}={}", keyword, XText(value)),
|
||||
MailParameter::Other {
|
||||
ref keyword,
|
||||
keyword,
|
||||
value: None,
|
||||
} => f.write_str(keyword),
|
||||
}
|
||||
@@ -277,13 +276,13 @@ pub enum RcptParameter {
|
||||
|
||||
impl Display for RcptParameter {
|
||||
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
|
||||
match *self {
|
||||
match &self {
|
||||
RcptParameter::Other {
|
||||
ref keyword,
|
||||
value: Some(ref value),
|
||||
keyword,
|
||||
value: Some(value),
|
||||
} => write!(f, "{keyword}={}", XText(value)),
|
||||
RcptParameter::Other {
|
||||
ref keyword,
|
||||
keyword,
|
||||
value: None,
|
||||
} => f.write_str(keyword),
|
||||
}
|
||||
|
||||
@@ -5,7 +5,6 @@ use std::{
|
||||
fmt::{Display, Formatter, Result},
|
||||
result,
|
||||
str::FromStr,
|
||||
string::ToString,
|
||||
};
|
||||
|
||||
use nom::{
|
||||
|
||||
@@ -317,9 +317,9 @@ impl SmtpClient {
|
||||
/// Handles encryption and authentication
|
||||
pub fn connection(&self) -> Result<SmtpConnection, Error> {
|
||||
#[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"))]
|
||||
Tls::Wrapper(ref tls_parameters) => Some(tls_parameters),
|
||||
Tls::Wrapper(tls_parameters) => Some(tls_parameters),
|
||||
_ => None,
|
||||
};
|
||||
|
||||
@@ -333,14 +333,14 @@ impl SmtpClient {
|
||||
)?;
|
||||
|
||||
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
|
||||
match self.info.tls {
|
||||
Tls::Opportunistic(ref tls_parameters) => {
|
||||
match &self.info.tls {
|
||||
Tls::Opportunistic(tls_parameters) => {
|
||||
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) => {
|
||||
conn.starttls(tls_parameters, &self.info.hello_name)?;
|
||||
Tls::Required(tls_parameters) => {
|
||||
conn = conn.starttls(tls_parameters, &self.info.hello_name)?;
|
||||
}
|
||||
_ => (),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user