aerogramme/src/lmtp.rs

216 lines
6.4 KiB
Rust
Raw Normal View History

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::*;
use smtp_message::{Email, EscapedDataReader, 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::*;
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 {
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();
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
}
}