2022-05-31 15:07:34 +00:00
|
|
|
use std::net::SocketAddr;
|
|
|
|
use std::{pin::Pin, sync::Arc};
|
|
|
|
|
2022-06-29 11:41:05 +00:00
|
|
|
use anyhow::Result;
|
2022-05-31 15:07:34 +00:00
|
|
|
use async_trait::async_trait;
|
|
|
|
use duplexify::Duplex;
|
|
|
|
use futures::{io, AsyncRead, AsyncReadExt, AsyncWrite};
|
2022-07-13 09:19:08 +00:00
|
|
|
use futures::{
|
|
|
|
stream,
|
|
|
|
stream::{FuturesOrdered, FuturesUnordered},
|
|
|
|
StreamExt,
|
|
|
|
};
|
2022-05-31 15:07:34 +00:00
|
|
|
use log::*;
|
2022-06-29 11:41:05 +00:00
|
|
|
use tokio::net::TcpListener;
|
2022-05-31 15:07:34 +00:00
|
|
|
use tokio::select;
|
|
|
|
use tokio::sync::watch;
|
|
|
|
use tokio_util::compat::*;
|
|
|
|
|
2024-02-10 11:11:01 +00:00
|
|
|
use smtp_message::{Email, EscapedDataReader, DataUnescaper, Reply, ReplyCode};
|
2022-06-29 10:52:58 +00:00
|
|
|
use smtp_server::{reply, Config, ConnectionMetadata, Decision, MailMetadata};
|
2022-05-31 15:07:34 +00:00
|
|
|
|
|
|
|
use crate::config::*;
|
|
|
|
use crate::login::*;
|
2022-06-30 15:40:59 +00:00
|
|
|
use crate::mail::incoming::EncryptedMessage;
|
2022-05-31 15:07:34 +00:00
|
|
|
|
|
|
|
pub struct LmtpServer {
|
|
|
|
bind_addr: SocketAddr,
|
|
|
|
hostname: String,
|
|
|
|
login_provider: Arc<dyn LoginProvider + Send + Sync>,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl LmtpServer {
|
|
|
|
pub fn new(
|
|
|
|
config: LmtpConfig,
|
|
|
|
login_provider: Arc<dyn LoginProvider + Send + Sync>,
|
|
|
|
) -> Arc<Self> {
|
|
|
|
Arc::new(Self {
|
|
|
|
bind_addr: config.bind_addr,
|
|
|
|
hostname: config.hostname,
|
|
|
|
login_provider,
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn run(self: &Arc<Self>, mut must_exit: watch::Receiver<bool>) -> Result<()> {
|
|
|
|
let tcp = TcpListener::bind(self.bind_addr).await?;
|
2022-06-29 10:50:44 +00:00
|
|
|
info!("LMTP server listening on {:#}", self.bind_addr);
|
|
|
|
|
2022-05-31 15:07:34 +00:00
|
|
|
let mut connections = FuturesUnordered::new();
|
|
|
|
|
|
|
|
while !*must_exit.borrow() {
|
|
|
|
let wait_conn_finished = async {
|
|
|
|
if connections.is_empty() {
|
|
|
|
futures::future::pending().await
|
|
|
|
} else {
|
|
|
|
connections.next().await
|
|
|
|
}
|
|
|
|
};
|
|
|
|
let (socket, remote_addr) = select! {
|
|
|
|
a = tcp.accept() => a?,
|
|
|
|
_ = wait_conn_finished => continue,
|
|
|
|
_ = must_exit.changed() => continue,
|
|
|
|
};
|
2022-07-13 09:19:08 +00:00
|
|
|
info!("LMTP: accepted connection from {}", remote_addr);
|
2022-05-31 15:07:34 +00:00
|
|
|
|
|
|
|
let conn = tokio::spawn(smtp_server::interact(
|
|
|
|
socket.compat(),
|
|
|
|
smtp_server::IsAlreadyTls::No,
|
2022-07-13 09:19:08 +00:00
|
|
|
(),
|
2022-05-31 15:07:34 +00:00
|
|
|
self.clone(),
|
|
|
|
));
|
|
|
|
|
|
|
|
connections.push(conn);
|
|
|
|
}
|
|
|
|
drop(tcp);
|
|
|
|
|
|
|
|
info!("LMTP server shutting down, draining remaining connections...");
|
|
|
|
while connections.next().await.is_some() {}
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// ----
|
|
|
|
|
|
|
|
pub struct Message {
|
|
|
|
to: Vec<PublicCredentials>,
|
|
|
|
}
|
|
|
|
|
|
|
|
#[async_trait]
|
|
|
|
impl Config for LmtpServer {
|
2022-06-21 09:26:04 +00:00
|
|
|
type Protocol = smtp_server::protocol::Lmtp;
|
2022-05-31 15:07:34 +00:00
|
|
|
|
2022-07-13 09:19:08 +00:00
|
|
|
type ConnectionUserMeta = ();
|
2022-05-31 15:07:34 +00:00
|
|
|
type MailUserMeta = Message;
|
|
|
|
|
2022-07-13 09:19:08 +00:00
|
|
|
fn hostname(&self, _conn_meta: &ConnectionMetadata<()>) -> &str {
|
2022-05-31 15:07:34 +00:00
|
|
|
&self.hostname
|
|
|
|
}
|
|
|
|
|
2022-07-13 09:19:08 +00:00
|
|
|
async fn new_mail(&self, _conn_meta: &mut ConnectionMetadata<()>) -> Message {
|
2022-05-31 15:07:34 +00:00
|
|
|
Message { to: vec![] }
|
|
|
|
}
|
|
|
|
|
|
|
|
async fn tls_accept<IO>(
|
|
|
|
&self,
|
|
|
|
_io: IO,
|
2022-07-13 09:19:08 +00:00
|
|
|
_conn_meta: &mut ConnectionMetadata<()>,
|
2022-05-31 15:07:34 +00:00
|
|
|
) -> io::Result<Duplex<Pin<Box<dyn Send + AsyncRead>>, Pin<Box<dyn Send + AsyncWrite>>>>
|
|
|
|
where
|
|
|
|
IO: Send + AsyncRead + AsyncWrite,
|
|
|
|
{
|
|
|
|
Err(io::Error::new(
|
|
|
|
io::ErrorKind::InvalidInput,
|
|
|
|
"TLS not implemented for LMTP server",
|
|
|
|
))
|
|
|
|
}
|
|
|
|
|
|
|
|
async fn filter_from(
|
|
|
|
&self,
|
|
|
|
from: Option<Email>,
|
2022-06-29 10:52:58 +00:00
|
|
|
_meta: &mut MailMetadata<Message>,
|
2022-07-13 09:19:08 +00:00
|
|
|
_conn_meta: &mut ConnectionMetadata<()>,
|
2022-05-31 15:07:34 +00:00
|
|
|
) -> Decision<Option<Email>> {
|
|
|
|
Decision::Accept {
|
|
|
|
reply: reply::okay_from().convert(),
|
|
|
|
res: from,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
async fn filter_to(
|
|
|
|
&self,
|
|
|
|
to: Email,
|
|
|
|
meta: &mut MailMetadata<Message>,
|
2022-07-13 09:19:08 +00:00
|
|
|
_conn_meta: &mut ConnectionMetadata<()>,
|
2022-05-31 15:07:34 +00:00
|
|
|
) -> Decision<Email> {
|
|
|
|
let to_str = match to.hostname.as_ref() {
|
|
|
|
Some(h) => format!("{}@{}", to.localpart, h),
|
|
|
|
None => to.localpart.to_string(),
|
|
|
|
};
|
|
|
|
match self.login_provider.public_login(&to_str).await {
|
|
|
|
Ok(creds) => {
|
|
|
|
meta.user.to.push(creds);
|
|
|
|
Decision::Accept {
|
|
|
|
reply: reply::okay_to().convert(),
|
|
|
|
res: to,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
Err(e) => Decision::Reject {
|
|
|
|
reply: Reply {
|
|
|
|
code: ReplyCode::POLICY_REASON,
|
|
|
|
ecode: None,
|
|
|
|
text: vec![smtp_message::MaybeUtf8::Utf8(e.to_string())],
|
|
|
|
},
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-06-29 10:50:44 +00:00
|
|
|
async fn handle_mail<'resp, R>(
|
|
|
|
&'resp self,
|
|
|
|
reader: &mut EscapedDataReader<'_, R>,
|
2022-05-31 15:07:34 +00:00
|
|
|
meta: MailMetadata<Message>,
|
2022-07-13 09:19:08 +00:00
|
|
|
_conn_meta: &'resp mut ConnectionMetadata<()>,
|
2022-06-29 10:50:44 +00:00
|
|
|
) -> Pin<Box<dyn futures::Stream<Item = Decision<()>> + Send + 'resp>>
|
2022-05-31 15:07:34 +00:00
|
|
|
where
|
|
|
|
R: Send + Unpin + AsyncRead,
|
|
|
|
{
|
|
|
|
let err_response_stream = |meta: MailMetadata<Message>, msg: String| {
|
|
|
|
Box::pin(
|
|
|
|
stream::iter(meta.user.to.into_iter()).map(move |_| Decision::Reject {
|
|
|
|
reply: Reply {
|
|
|
|
code: ReplyCode::POLICY_REASON,
|
|
|
|
ecode: None,
|
|
|
|
text: vec![smtp_message::MaybeUtf8::Utf8(msg.clone())],
|
|
|
|
},
|
|
|
|
}),
|
|
|
|
)
|
|
|
|
};
|
|
|
|
|
|
|
|
let mut text = Vec::new();
|
2022-07-13 09:19:08 +00:00
|
|
|
if let Err(e) = reader.read_to_end(&mut text).await {
|
|
|
|
return err_response_stream(meta, format!("io error: {}", e));
|
2022-05-31 15:07:34 +00:00
|
|
|
}
|
|
|
|
reader.complete();
|
2024-02-10 11:11:01 +00:00
|
|
|
let raw_size = text.len();
|
|
|
|
|
|
|
|
// Unescape email, shrink it also to remove last dot
|
|
|
|
let unesc_res = DataUnescaper::new(true).unescape(&mut text);
|
|
|
|
text.truncate(unesc_res.written);
|
|
|
|
tracing::debug!(prev_sz=raw_size, new_sz=text.len(), "unescaped");
|
2022-05-31 15:07:34 +00:00
|
|
|
|
|
|
|
let encrypted_message = match EncryptedMessage::new(text) {
|
|
|
|
Ok(x) => Arc::new(x),
|
|
|
|
Err(e) => return err_response_stream(meta, e.to_string()),
|
|
|
|
};
|
|
|
|
|
2022-07-13 09:19:08 +00:00
|
|
|
Box::pin(
|
|
|
|
meta.user
|
|
|
|
.to
|
|
|
|
.into_iter()
|
|
|
|
.map(move |creds| {
|
|
|
|
let encrypted_message = encrypted_message.clone();
|
|
|
|
async move {
|
|
|
|
match encrypted_message.deliver_to(creds).await {
|
|
|
|
Ok(()) => Decision::Accept {
|
|
|
|
reply: reply::okay_mail().convert(),
|
|
|
|
res: (),
|
|
|
|
},
|
|
|
|
Err(e) => Decision::Reject {
|
|
|
|
reply: Reply {
|
|
|
|
code: ReplyCode::POLICY_REASON,
|
|
|
|
ecode: None,
|
|
|
|
text: vec![smtp_message::MaybeUtf8::Utf8(e.to_string())],
|
|
|
|
},
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
})
|
|
|
|
.collect::<FuturesOrdered<_>>(),
|
|
|
|
)
|
2022-05-31 15:07:34 +00:00
|
|
|
}
|
|
|
|
}
|