Compare commits

..

3 Commits

Author SHA1 Message Date
Paolo Barbolini
51794aa912 Prepare v0.11.10 (#1002) 2024-10-23 23:04:45 +02:00
Paolo Barbolini
eb42651401 Use case insensitive comparisons for login challenge requests (#1000) 2024-10-23 20:01:00 +02:00
Paolo Barbolini
99c6dc2a87 Replace quit with abort in transport connection drop code (#999) 2024-10-23 20:00:31 +02:00
11 changed files with 172 additions and 224 deletions

View File

@@ -1,3 +1,14 @@
<a name="v0.11.10"></a>
### v0.11.10 (2024-10-23)
#### Bug fixes
* Ignore disconnect errors when `pool` feature of SMTP transport is disabled ([#999])
* Use case insensitive comparisons for matching login challenge requests ([#1000])
[#999]: https://github.com/lettre/lettre/pull/999
[#1000]: https://github.com/lettre/lettre/pull/1000
<a name="v0.11.9"></a> <a name="v0.11.9"></a>
### v0.11.9 (2024-09-13) ### v0.11.9 (2024-09-13)

2
Cargo.lock generated
View File

@@ -1178,7 +1178,7 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
[[package]] [[package]]
name = "lettre" name = "lettre"
version = "0.11.9" version = "0.11.10"
dependencies = [ dependencies = [
"async-std", "async-std",
"async-trait", "async-trait",

View File

@@ -1,7 +1,7 @@
[package] [package]
name = "lettre" name = "lettre"
# remember to update html_root_url and README.md (Cargo.toml example and deps.rs badge) # remember to update html_root_url and README.md (Cargo.toml example and deps.rs badge)
version = "0.11.9" version = "0.11.10"
description = "Email client" description = "Email client"
readme = "README.md" readme = "README.md"
homepage = "https://lettre.rs" homepage = "https://lettre.rs"

View File

@@ -28,8 +28,8 @@
</div> </div>
<div align="center"> <div align="center">
<a href="https://deps.rs/crate/lettre/0.11.9"> <a href="https://deps.rs/crate/lettre/0.11.10">
<img src="https://deps.rs/crate/lettre/0.11.9/status.svg" <img src="https://deps.rs/crate/lettre/0.11.10/status.svg"
alt="dependency status" /> alt="dependency status" />
</a> </a>
</div> </div>

View File

@@ -109,7 +109,7 @@
//! [mime 0.3]: https://docs.rs/mime/0.3 //! [mime 0.3]: https://docs.rs/mime/0.3
//! [DKIM]: https://datatracker.ietf.org/doc/html/rfc6376 //! [DKIM]: https://datatracker.ietf.org/doc/html/rfc6376
#![doc(html_root_url = "https://docs.rs/crate/lettre/0.11.9")] #![doc(html_root_url = "https://docs.rs/crate/lettre/0.11.10")]
#![doc(html_favicon_url = "https://lettre.rs/favicon.ico")] #![doc(html_favicon_url = "https://lettre.rs/favicon.ico")]
#![doc(html_logo_url = "https://avatars0.githubusercontent.com/u/15113230?v=4")] #![doc(html_logo_url = "https://avatars0.githubusercontent.com/u/15113230?v=4")]
#![forbid(unsafe_code)] #![forbid(unsafe_code)]

View File

@@ -45,7 +45,7 @@ impl AsyncTransport for AsyncSmtpTransport<Tokio1Executor> {
let result = conn.send(envelope, email).await?; let result = conn.send(envelope, email).await?;
#[cfg(not(feature = "pool"))] #[cfg(not(feature = "pool"))]
conn.quit().await?; conn.abort().await;
Ok(result) Ok(result)
} }

View File

@@ -98,13 +98,17 @@ impl Mechanism {
let decoded_challenge = challenge let decoded_challenge = challenge
.ok_or_else(|| error::client("This mechanism does expect a challenge"))?; .ok_or_else(|| error::client("This mechanism does expect a challenge"))?;
if ["User Name", "Username:", "Username", "User Name\0"] if contains_ignore_ascii_case(
.contains(&decoded_challenge) decoded_challenge,
{ ["User Name", "Username:", "Username", "User Name\0"],
) {
return Ok(credentials.authentication_identity.clone()); return Ok(credentials.authentication_identity.clone());
} }
if ["Password", "Password:", "Password\0"].contains(&decoded_challenge) { if contains_ignore_ascii_case(
decoded_challenge,
["Password", "Password:", "Password\0"],
) {
return Ok(credentials.secret.clone()); return Ok(credentials.secret.clone());
} }
@@ -121,6 +125,15 @@ impl Mechanism {
} }
} }
fn contains_ignore_ascii_case<'a>(
haystack: &str,
needles: impl IntoIterator<Item = &'a str>,
) -> bool {
needles
.into_iter()
.any(|item| item.eq_ignore_ascii_case(haystack))
}
#[cfg(test)] #[cfg(test)]
mod test { mod test {
use super::{Credentials, Mechanism}; use super::{Credentials, Mechanism};
@@ -155,6 +168,23 @@ mod test {
assert!(mechanism.response(&credentials, None).is_err()); assert!(mechanism.response(&credentials, None).is_err());
} }
#[test]
fn test_login_case_insensitive() {
let mechanism = Mechanism::Login;
let credentials = Credentials::new("alice".to_owned(), "wonderland".to_owned());
assert_eq!(
mechanism.response(&credentials, Some("username")).unwrap(),
"alice"
);
assert_eq!(
mechanism.response(&credentials, Some("password")).unwrap(),
"wonderland"
);
assert!(mechanism.response(&credentials, None).is_err());
}
#[test] #[test]
fn test_xoauth2() { fn test_xoauth2() {
let mechanism = Mechanism::Xoauth2; let mechanism = Mechanism::Xoauth2;

View File

@@ -10,21 +10,34 @@ use super::{AsyncNetworkStream, ClientCodec, TlsParameters};
use crate::{ use crate::{
transport::smtp::{ transport::smtp::{
authentication::{Credentials, Mechanism}, authentication::{Credentials, Mechanism},
client::{ConnectionState, ConnectionWrapper},
commands::{Auth, Data, Ehlo, Mail, Noop, Quit, Rcpt, Starttls}, commands::{Auth, Data, Ehlo, Mail, Noop, Quit, Rcpt, Starttls},
error::{self, Error}, error,
error::Error,
extension::{ClientId, Extension, MailBodyParameter, MailParameter, ServerInfo}, extension::{ClientId, Extension, MailBodyParameter, MailParameter, ServerInfo},
response::{parse_response, Response}, response::{parse_response, Response},
}, },
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
stream: ConnectionWrapper<BufReader<AsyncNetworkStream>>, /// Value is None before connection
/// Whether QUIT has been sent stream: BufReader<AsyncNetworkStream>,
sent_quit: bool, /// Panic state
panic: bool,
/// Information about the server /// Information about the server
server_info: ServerInfo, server_info: ServerInfo,
} }
@@ -112,8 +125,8 @@ impl AsyncSmtpConnection {
) -> Result<AsyncSmtpConnection, Error> { ) -> Result<AsyncSmtpConnection, Error> {
let stream = BufReader::new(stream); let stream = BufReader::new(stream);
let mut conn = AsyncSmtpConnection { let mut conn = AsyncSmtpConnection {
stream: ConnectionWrapper::new(stream), stream,
sent_quit: false, panic: false,
server_info: ServerInfo::default(), server_info: ServerInfo::default(),
}; };
// TODO log // TODO log
@@ -157,28 +170,30 @@ impl AsyncSmtpConnection {
mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime)); mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
} }
self.command(Mail::new(envelope.from().cloned(), mail_options)) try_smtp!(
.await?; self.command(Mail::new(envelope.from().cloned(), mail_options))
.await,
self
);
// Recipient // Recipient
for to_address in envelope.to() { for to_address in envelope.to() {
self.command(Rcpt::new(to_address.clone(), vec![])).await?; try_smtp!(
self.command(Rcpt::new(to_address.clone(), vec![])).await,
self
);
} }
// Data // Data
self.command(Data).await?; try_smtp!(self.command(Data).await, self);
// Message content // Message content
let result = self.message(email).await?; let result = try_smtp!(self.message(email).await, self);
Ok(result) Ok(result)
} }
pub fn has_broken(&self) -> bool { pub fn has_broken(&self) -> bool {
self.sent_quit self.panic
|| matches!(
self.stream.state(),
ConnectionState::BrokenConnection | ConnectionState::BrokenResponse
)
} }
pub fn can_starttls(&self) -> bool { pub fn can_starttls(&self) -> bool {
@@ -198,14 +213,12 @@ impl AsyncSmtpConnection {
hello_name: &ClientId, hello_name: &ClientId,
) -> Result<(), Error> { ) -> Result<(), Error> {
if self.server_info.supports_feature(Extension::StartTls) { if self.server_info.supports_feature(Extension::StartTls) {
self.command(Starttls).await?; try_smtp!(self.command(Starttls).await, self);
self.stream self.stream.get_mut().upgrade_tls(tls_parameters).await?;
.async_op(|stream| stream.get_mut().upgrade_tls(tls_parameters))
.await?;
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("connection encrypted"); tracing::debug!("connection encrypted");
// Send EHLO again // Send EHLO again
self.ehlo(hello_name).await?; try_smtp!(self.ehlo(hello_name).await, self);
Ok(()) Ok(())
} else { } else {
Err(error::client("STARTTLS is not supported on this server")) Err(error::client("STARTTLS is not supported on this server"))
@@ -214,39 +227,32 @@ 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 = self.command(Ehlo::new(hello_name.clone())).await?; let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())).await, self);
self.server_info = ServerInfo::from_response(&ehlo_response)?; self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self);
Ok(()) Ok(())
} }
pub async fn quit(&mut self) -> Result<Response, Error> { pub async fn quit(&mut self) -> Result<Response, Error> {
self.sent_quit = true; 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 // Only try to quit if we are not already broken
// `write` already rejects writes if the connection state if bad if !self.panic {
if !self.sent_quit { self.panic = true;
let _ = self.quit().await; let _ = self.command(Quit).await;
}
if !matches!(self.stream.state(), ConnectionState::BrokenConnection) {
let _ = self
.stream
.async_op(|stream| async { stream.close().await.map_err(error::network) })
.await;
} }
let _ = self.stream.close().await;
} }
/// Sets the underlying stream /// Sets the underlying stream
pub fn set_stream(&mut self, stream: AsyncNetworkStream) { pub fn set_stream(&mut self, stream: AsyncNetworkStream) {
self.stream = ConnectionWrapper::new(BufReader::new(stream)); self.stream = BufReader::new(stream);
} }
/// Tells if the underlying stream is currently encrypted /// Tells if the underlying stream is currently encrypted
pub fn is_encrypted(&self) -> bool { pub fn is_encrypted(&self) -> bool {
self.stream.get_ref().get_ref().is_encrypted() self.stream.get_ref().is_encrypted()
} }
/// Checks if the server is connected using the NOOP SMTP command /// Checks if the server is connected using the NOOP SMTP command
@@ -273,13 +279,15 @@ impl AsyncSmtpConnection {
while challenges > 0 && response.has_code(334) { while challenges > 0 && response.has_code(334) {
challenges -= 1; challenges -= 1;
response = self response = try_smtp!(
.command(Auth::new_from_response( self.command(Auth::new_from_response(
mechanism, mechanism,
credentials.clone(), credentials.clone(),
&response, &response,
)?) )?)
.await?; .await,
self
);
} }
if challenges == 0 { if challenges == 0 {
@@ -308,17 +316,15 @@ 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 self.stream
.async_op(|stream| async { .get_mut()
stream .write_all(string)
.get_mut() .await
.write_all(string) .map_err(error::network)?;
.await
.map_err(error::network)
})
.await?;
self.stream self.stream
.async_op(|stream| async { stream.get_mut().flush().await.map_err(error::network) }) .get_mut()
.await?; .flush()
.await
.map_err(error::network)?;
#[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)));
@@ -331,10 +337,9 @@ impl AsyncSmtpConnection {
while self while self
.stream .stream
.async_op(|stream| async { .read_line(&mut buffer)
stream.read_line(&mut buffer).await.map_err(error::network) .await
}) .map_err(error::network)?
.await?
> 0 > 0
{ {
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
@@ -351,12 +356,10 @@ impl AsyncSmtpConnection {
} }
} }
Err(nom::Err::Failure(e)) => { Err(nom::Err::Failure(e)) => {
self.stream.set_state(ConnectionState::BrokenResponse);
return Err(error::response(e.to_string())); return Err(error::response(e.to_string()));
} }
Err(nom::Err::Incomplete(_)) => { /* read more */ } Err(nom::Err::Incomplete(_)) => { /* read more */ }
Err(nom::Err::Error(e)) => { Err(nom::Err::Error(e)) => {
self.stream.set_state(ConnectionState::BrokenResponse);
return Err(error::response(e.to_string())); return Err(error::response(e.to_string()));
} }
} }
@@ -368,12 +371,12 @@ impl AsyncSmtpConnection {
/// The X509 certificate of the server (DER encoded) /// The X509 certificate of the server (DER encoded)
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
pub fn peer_certificate(&self) -> Result<Vec<u8>, Error> { pub fn peer_certificate(&self) -> Result<Vec<u8>, Error> {
self.stream.get_ref().get_ref().peer_certificate() self.stream.get_ref().peer_certificate()
} }
/// All the X509 certificates of the chain (DER encoded) /// All the X509 certificates of the chain (DER encoded)
#[cfg(any(feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "rustls-tls", feature = "boring-tls"))]
pub fn certificate_chain(&self) -> Result<Vec<Vec<u8>>, Error> { pub fn certificate_chain(&self) -> Result<Vec<Vec<u8>>, Error> {
self.stream.get_ref().get_ref().certificate_chain() self.stream.get_ref().certificate_chain()
} }
} }

View File

@@ -7,25 +7,38 @@ use std::{
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
use super::escape_crlf; use super::escape_crlf;
use super::{ClientCodec, ConnectionWrapper, NetworkStream, TlsParameters}; use super::{ClientCodec, NetworkStream, TlsParameters};
use crate::{ use crate::{
address::Envelope, address::Envelope,
transport::smtp::{ transport::smtp::{
authentication::{Credentials, Mechanism}, authentication::{Credentials, Mechanism},
client::ConnectionState,
commands::{Auth, Data, Ehlo, Mail, Noop, Quit, Rcpt, Starttls}, commands::{Auth, Data, Ehlo, Mail, Noop, Quit, Rcpt, Starttls},
error::{self, Error}, error,
error::Error,
extension::{ClientId, Extension, MailBodyParameter, MailParameter, ServerInfo}, extension::{ClientId, Extension, MailBodyParameter, MailParameter, ServerInfo},
response::{parse_response, Response}, response::{parse_response, Response},
}, },
}; };
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
stream: ConnectionWrapper<BufReader<NetworkStream>>, /// Value is None before connection
/// Whether QUIT has been sent stream: BufReader<NetworkStream>,
sent_quit: bool, /// Panic state
panic: bool,
/// Information about the server /// Information about the server
server_info: ServerInfo, server_info: ServerInfo,
} }
@@ -51,8 +64,8 @@ impl SmtpConnection {
let stream = NetworkStream::connect(server, timeout, tls_parameters, local_address)?; let stream = NetworkStream::connect(server, timeout, tls_parameters, local_address)?;
let stream = BufReader::new(stream); let stream = BufReader::new(stream);
let mut conn = SmtpConnection { let mut conn = SmtpConnection {
stream: ConnectionWrapper::new(stream), stream,
sent_quit: false, 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)?;
@@ -97,27 +110,26 @@ impl SmtpConnection {
mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime)); mail_options.push(MailParameter::Body(MailBodyParameter::EightBitMime));
} }
self.command(Mail::new(envelope.from().cloned(), mail_options))?; try_smtp!(
self.command(Mail::new(envelope.from().cloned(), mail_options)),
self
);
// Recipient // Recipient
for to_address in envelope.to() { for to_address in envelope.to() {
self.command(Rcpt::new(to_address.clone(), vec![]))?; try_smtp!(self.command(Rcpt::new(to_address.clone(), vec![])), self);
} }
// Data // Data
self.command(Data)?; try_smtp!(self.command(Data), self);
// Message content // Message content
let result = self.message(email)?; let result = try_smtp!(self.message(email), self);
Ok(result) Ok(result)
} }
pub fn has_broken(&self) -> bool { pub fn has_broken(&self) -> bool {
self.sent_quit self.panic
|| matches!(
self.stream.state(),
ConnectionState::BrokenConnection | ConnectionState::BrokenResponse
)
} }
pub fn can_starttls(&self) -> bool { pub fn can_starttls(&self) -> bool {
@@ -133,13 +145,12 @@ impl SmtpConnection {
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"))]
{ {
self.command(Starttls)?; try_smtp!(self.command(Starttls), self);
self.stream self.stream.get_mut().upgrade_tls(tls_parameters)?;
.sync_op(|stream| stream.get_mut().upgrade_tls(tls_parameters))?;
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("connection encrypted"); tracing::debug!("connection encrypted");
// Send EHLO again // Send EHLO again
self.ehlo(hello_name)?; try_smtp!(self.ehlo(hello_name), self);
Ok(()) Ok(())
} }
#[cfg(not(any( #[cfg(not(any(
@@ -157,47 +168,38 @@ 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 = self.command(Ehlo::new(hello_name.clone()))?; let ehlo_response = try_smtp!(self.command(Ehlo::new(hello_name.clone())), self);
self.server_info = ServerInfo::from_response(&ehlo_response)?; self.server_info = try_smtp!(ServerInfo::from_response(&ehlo_response), self);
Ok(()) Ok(())
} }
pub fn quit(&mut self) -> Result<Response, Error> { pub fn quit(&mut self) -> Result<Response, Error> {
self.sent_quit = true; 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 // Only try to quit if we are not already broken
// `write` already rejects writes if the connection state if bad if !self.panic {
if !self.sent_quit { self.panic = true;
let _ = self.quit(); let _ = self.command(Quit);
}
if !matches!(self.stream.state(), ConnectionState::BrokenConnection) {
let _ = self.stream.sync_op(|stream| {
stream
.get_mut()
.shutdown(std::net::Shutdown::Both)
.map_err(error::network)
});
} }
let _ = self.stream.get_mut().shutdown(std::net::Shutdown::Both);
} }
/// Sets the underlying stream /// Sets the underlying stream
pub fn set_stream(&mut self, stream: NetworkStream) { pub fn set_stream(&mut self, stream: NetworkStream) {
self.stream = ConnectionWrapper::new(BufReader::new(stream)); self.stream = BufReader::new(stream);
} }
/// Tells if the underlying stream is currently encrypted /// Tells if the underlying stream is currently encrypted
pub fn is_encrypted(&self) -> bool { pub fn is_encrypted(&self) -> bool {
self.stream.get_ref().get_ref().is_encrypted() self.stream.get_ref().is_encrypted()
} }
/// Set timeout /// Set timeout
pub fn set_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> { pub fn set_timeout(&mut self, duration: Option<Duration>) -> io::Result<()> {
self.stream.get_mut().get_mut().set_read_timeout(duration)?; self.stream.get_mut().set_read_timeout(duration)?;
self.stream.get_mut().get_mut().set_write_timeout(duration) self.stream.get_mut().set_write_timeout(duration)
} }
/// Checks if the server is connected using the NOOP SMTP command /// Checks if the server is connected using the NOOP SMTP command
@@ -222,11 +224,14 @@ impl SmtpConnection {
while challenges > 0 && response.has_code(334) { while challenges > 0 && response.has_code(334) {
challenges -= 1; challenges -= 1;
response = self.command(Auth::new_from_response( response = try_smtp!(
mechanism, self.command(Auth::new_from_response(
credentials.clone(), mechanism,
&response, credentials.clone(),
)?)?; &response,
)?),
self
);
} }
if challenges == 0 { if challenges == 0 {
@@ -256,9 +261,10 @@ 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 self.stream
.sync_op(|stream| stream.get_mut().write_all(string).map_err(error::network))?; .get_mut()
self.stream .write_all(string)
.sync_op(|stream| stream.get_mut().flush().map_err(error::network))?; .map_err(error::network)?;
self.stream.get_mut().flush().map_err(error::network)?;
#[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)));
@@ -269,11 +275,7 @@ impl SmtpConnection {
pub fn read_response(&mut self) -> Result<Response, Error> { pub fn read_response(&mut self) -> Result<Response, Error> {
let mut buffer = String::with_capacity(100); let mut buffer = String::with_capacity(100);
while self while self.stream.read_line(&mut buffer).map_err(error::network)? > 0 {
.stream
.sync_op(|stream| stream.read_line(&mut buffer).map_err(error::network))?
> 0
{
#[cfg(feature = "tracing")] #[cfg(feature = "tracing")]
tracing::debug!("<< {}", escape_crlf(&buffer)); tracing::debug!("<< {}", escape_crlf(&buffer));
match parse_response(&buffer) { match parse_response(&buffer) {
@@ -288,12 +290,10 @@ impl SmtpConnection {
}; };
} }
Err(nom::Err::Failure(e)) => { Err(nom::Err::Failure(e)) => {
self.stream.set_state(ConnectionState::BrokenResponse);
return Err(error::response(e.to_string())); return Err(error::response(e.to_string()));
} }
Err(nom::Err::Incomplete(_)) => { /* read more */ } Err(nom::Err::Incomplete(_)) => { /* read more */ }
Err(nom::Err::Error(e)) => { Err(nom::Err::Error(e)) => {
self.stream.set_state(ConnectionState::BrokenResponse);
return Err(error::response(e.to_string())); return Err(error::response(e.to_string()));
} }
} }
@@ -305,12 +305,12 @@ impl SmtpConnection {
/// The X509 certificate of the server (DER encoded) /// The X509 certificate of the server (DER encoded)
#[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "native-tls", feature = "rustls-tls", feature = "boring-tls"))]
pub fn peer_certificate(&self) -> Result<Vec<u8>, Error> { pub fn peer_certificate(&self) -> Result<Vec<u8>, Error> {
self.stream.get_ref().get_ref().peer_certificate() self.stream.get_ref().peer_certificate()
} }
/// All the X509 certificates of the chain (DER encoded) /// All the X509 certificates of the chain (DER encoded)
#[cfg(any(feature = "rustls-tls", feature = "boring-tls"))] #[cfg(any(feature = "rustls-tls", feature = "boring-tls"))]
pub fn certificate_chain(&self) -> Result<Vec<Vec<u8>>, Error> { pub fn certificate_chain(&self) -> Result<Vec<Vec<u8>>, Error> {
self.stream.get_ref().get_ref().certificate_chain() self.stream.get_ref().certificate_chain()
} }
} }

View File

@@ -24,8 +24,6 @@
#[cfg(feature = "serde")] #[cfg(feature = "serde")]
use std::fmt::Debug; use std::fmt::Debug;
#[cfg(any(feature = "tokio1", feature = "async-std1"))]
use std::future::Future;
#[cfg(any(feature = "tokio1", feature = "async-std1"))] #[cfg(any(feature = "tokio1", feature = "async-std1"))]
pub use self::async_connection::AsyncSmtpConnection; pub use self::async_connection::AsyncSmtpConnection;
@@ -42,7 +40,6 @@ pub use self::{
connection::SmtpConnection, connection::SmtpConnection,
tls::{Certificate, CertificateStore, Identity, Tls, TlsParameters, TlsParametersBuilder}, tls::{Certificate, CertificateStore, Identity, 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;
@@ -52,99 +49,6 @@ mod connection;
mod net; mod net;
mod tls; mod tls;
#[derive(Debug)]
pub(super) struct ConnectionWrapper<C> {
conn: C,
state: ConnectionState,
}
impl<C> ConnectionWrapper<C> {
pub(super) fn new(conn: C) -> Self {
Self {
conn,
state: ConnectionState::ProbablyConnected,
}
}
pub(super) fn get_ref(&self) -> &C {
&self.conn
}
pub(super) fn get_mut(&mut self) -> &mut C {
&mut self.conn
}
pub(super) fn state(&self) -> ConnectionState {
self.state
}
pub(super) fn set_state(&mut self, state: ConnectionState) {
self.state = state;
}
pub(super) fn sync_op<F, T>(&mut self, f: F) -> Result<T, Error>
where
F: FnOnce(&mut C) -> Result<T, Error>,
{
if !matches!(
self.state,
ConnectionState::ProbablyConnected | ConnectionState::BrokenResponse
) {
return Err(error::client(
"attempted to send operation to broken connection",
));
}
self.state = ConnectionState::Writing;
match f(&mut self.conn) {
Ok(t) => {
self.state = ConnectionState::ProbablyConnected;
Ok(t)
}
Err(err) => {
self.state = ConnectionState::BrokenConnection;
Err(err)
}
}
}
#[cfg(any(feature = "tokio1", feature = "async-std1"))]
pub(super) async fn async_op<'a, F, Fut, T>(&'a mut self, f: F) -> Result<T, Error>
where
F: FnOnce(&'a mut C) -> Fut,
Fut: Future<Output = Result<T, Error>>,
{
if !matches!(
self.state,
ConnectionState::ProbablyConnected | ConnectionState::BrokenResponse
) {
return Err(error::client(
"attempted to send operation to broken connection",
));
}
self.state = ConnectionState::Writing;
match f(&mut self.conn).await {
Ok(t) => {
self.state = ConnectionState::ProbablyConnected;
Ok(t)
}
Err(err) => {
self.state = ConnectionState::BrokenConnection;
Err(err)
}
}
}
}
#[derive(Debug, Copy, Clone)]
pub(super) enum ConnectionState {
ProbablyConnected,
Writing,
BrokenResponse,
BrokenConnection,
}
/// The codec used for transparency /// The codec used for transparency
#[derive(Debug)] #[derive(Debug)]
struct ClientCodec { struct ClientCodec {

View File

@@ -32,7 +32,7 @@ impl Transport for SmtpTransport {
let result = conn.send(envelope, email)?; let result = conn.send(envelope, email)?;
#[cfg(not(feature = "pool"))] #[cfg(not(feature = "pool"))]
conn.quit()?; conn.abort();
Ok(result) Ok(result)
} }