diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/config.rs | 1 | ||||
| -rw-r--r-- | src/lmtp.rs | 263 | ||||
| -rw-r--r-- | src/mail_uuid.rs | 8 | ||||
| -rw-r--r-- | src/main.rs | 12 | ||||
| -rw-r--r-- | src/server.rs | 67 |
5 files changed, 339 insertions, 12 deletions
diff --git a/src/config.rs b/src/config.rs index 3fd0bd4..9ec0ea1 100644 --- a/src/config.rs +++ b/src/config.rs @@ -68,6 +68,7 @@ pub struct LoginLdapConfig { #[derive(Deserialize, Debug, Clone)] pub struct LmtpConfig { pub bind_addr: SocketAddr, + pub hostname: String, } pub fn read_config(config_file: PathBuf) -> Result<Config> { diff --git a/src/lmtp.rs b/src/lmtp.rs new file mode 100644 index 0000000..4186d69 --- /dev/null +++ b/src/lmtp.rs @@ -0,0 +1,263 @@ +use std::collections::HashMap; +use std::net::SocketAddr; +use std::{pin::Pin, sync::Arc}; + +use anyhow::{bail, Result}; +use async_trait::async_trait; +use duplexify::Duplex; +use futures::{io, AsyncRead, AsyncReadExt, AsyncWrite}; +use futures::{stream, stream::FuturesUnordered, StreamExt}; +use log::*; +use rusoto_s3::{PutObjectRequest, S3Client, S3}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::select; +use tokio::sync::watch; +use tokio_util::compat::*; + +use smtp_message::{Email, EscapedDataReader, Reply, ReplyCode}; +use smtp_server::{reply, Config, ConnectionMetadata, Decision, MailMetadata, Protocol}; + +use crate::config::*; +use crate::cryptoblob::*; +use crate::login::*; +use crate::mail_uuid::*; + +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?; + 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, + }; + + let conn = tokio::spawn(smtp_server::interact( + socket.compat(), + smtp_server::IsAlreadyTls::No, + Conn { remote_addr }, + self.clone(), + )); + + connections.push(conn); + } + drop(tcp); + + info!("LMTP server shutting down, draining remaining connections..."); + while connections.next().await.is_some() {} + + Ok(()) + } +} + +// ---- + +pub struct Conn { + remote_addr: SocketAddr, +} + +pub struct Message { + to: Vec<PublicCredentials>, +} + +#[async_trait] +impl Config for LmtpServer { + const PROTOCOL: Protocol = Protocol::Lmtp; + + type ConnectionUserMeta = Conn; + type MailUserMeta = Message; + + fn hostname(&self, _conn_meta: &ConnectionMetadata<Conn>) -> &str { + &self.hostname + } + + async fn new_mail(&self, _conn_meta: &mut ConnectionMetadata<Conn>) -> Message { + Message { to: vec![] } + } + + async fn tls_accept<IO>( + &self, + _io: IO, + _conn_meta: &mut ConnectionMetadata<Conn>, + ) -> 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>, + meta: &mut MailMetadata<Message>, + _conn_meta: &mut ConnectionMetadata<Conn>, + ) -> Decision<Option<Email>> { + Decision::Accept { + reply: reply::okay_from().convert(), + res: from, + } + } + + async fn filter_to( + &self, + to: Email, + meta: &mut MailMetadata<Message>, + _conn_meta: &mut ConnectionMetadata<Conn>, + ) -> 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())], + }, + }, + } + } + + async fn handle_mail<'a, R>( + &self, + reader: &mut EscapedDataReader<'a, R>, + _mail: MailMetadata<Message>, + _conn_meta: &mut ConnectionMetadata<Conn>, + ) -> Decision<()> + where + R: Send + Unpin + AsyncRead, + { + unreachable!(); + } + + async fn handle_mail_multi<'a, 'slife0, 'slife1, 'stream, R>( + &'slife0 self, + reader: &mut EscapedDataReader<'a, R>, + meta: MailMetadata<Message>, + conn_meta: &'slife1 mut ConnectionMetadata<Conn>, + ) -> Pin<Box<dyn futures::Stream<Item = Decision<()>> + Send + 'stream>> + where + R: Send + Unpin + AsyncRead, + 'slife0: 'stream, + 'slife1: 'stream, + Self: 'stream, + { + 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(); + if reader.read_to_end(&mut text).await.is_err() { + return err_response_stream(meta, "io error".into()); + } + reader.complete(); + + let encrypted_message = match EncryptedMessage::new(text) { + Ok(x) => Arc::new(x), + Err(e) => return err_response_stream(meta, e.to_string()), + }; + + Box::pin(stream::iter(meta.user.to.into_iter()).then(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())], + }, + }, + } + } + })) + } +} + +// ---- + +struct EncryptedMessage { + key: Key, + encrypted_body: Vec<u8>, +} + +impl EncryptedMessage { + fn new(body: Vec<u8>) -> Result<Self> { + let key = gen_key(); + let encrypted_body = seal(&body, &key)?; + Ok(Self { + key, + encrypted_body, + }) + } + + async fn deliver_to(self: Arc<Self>, creds: PublicCredentials) -> Result<()> { + let s3_client = creds.storage.s3_client()?; + + let encrypted_key = + sodiumoxide::crypto::sealedbox::seal(self.key.as_ref(), &creds.public_key); + let key_header = base64::encode(&encrypted_key); + + let mut por = PutObjectRequest::default(); + por.bucket = creds.storage.bucket.clone(); + por.key = format!("incoming/{}", gen_uuid().to_string()); + por.metadata = Some( + [("Message-Key".to_string(), key_header)] + .into_iter() + .collect::<HashMap<_, _>>(), + ); + por.body = Some(self.encrypted_body.clone().into()); + s3_client.put_object(por).await?; + + Ok(()) + } +} diff --git a/src/mail_uuid.rs b/src/mail_uuid.rs index d0d582f..b784e78 100644 --- a/src/mail_uuid.rs +++ b/src/mail_uuid.rs @@ -71,6 +71,12 @@ impl Serialize for MailUuid { where S: Serializer, { - serializer.serialize_str(&hex::encode(self.0)) + serializer.serialize_str(&self.to_string()) + } +} + +impl ToString for MailUuid { + fn to_string(&self) -> String { + hex::encode(self.0) } } diff --git a/src/main.rs b/src/main.rs index b8231f1..33d3188 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,7 @@ mod bayou; mod config; mod cryptoblob; +mod lmtp; mod login; mod mail_uuid; mod mailbox; @@ -35,6 +36,11 @@ enum Command { #[clap(short, long, env = "CONFIG_FILE", default_value = "mailrage.toml")] config_file: PathBuf, }, + /// TEST TEST TEST + Test { + #[clap(short, long, env = "CONFIG_FILE", default_value = "mailrage.toml")] + config_file: PathBuf, + }, /// Initializes key pairs for a user and adds a key decryption password FirstLogin { #[clap(flatten)] @@ -125,6 +131,12 @@ async fn main() -> Result<()> { let server = Server::new(config)?; server.run().await?; } + Command::Test { config_file } => { + let config = read_config(config_file)?; + + let server = Server::new(config)?; + server.test().await?; + } Command::FirstLogin { creds, user_secrets, diff --git a/src/server.rs b/src/server.rs index e1ab599..1fd21b4 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,18 +1,23 @@ -use anyhow::{bail, Result}; use std::sync::Arc; +use anyhow::{bail, Result}; +use futures::{try_join, StreamExt}; +use log::*; use rusoto_signature::Region; +use tokio::sync::watch; use crate::config::*; +use crate::lmtp::*; use crate::login::{ldap_provider::*, static_provider::*, *}; use crate::mailbox::Mailbox; pub struct Server { - pub login_provider: Box<dyn LoginProvider>, + pub login_provider: Arc<dyn LoginProvider + Send + Sync>, + pub lmtp_server: Option<Arc<LmtpServer>>, } impl Server { - pub fn new(config: Config) -> Result<Arc<Self>> { + pub fn new(config: Config) -> Result<Self> { let s3_region = Region::Custom { name: config.aws_region.clone(), endpoint: config.s3_endpoint, @@ -21,17 +26,43 @@ impl Server { name: config.aws_region, endpoint: config.k2v_endpoint, }; - let login_provider: Box<dyn LoginProvider> = match (config.login_static, config.login_ldap) - { - (Some(st), None) => Box::new(StaticLoginProvider::new(st, k2v_region, s3_region)?), - (None, Some(ld)) => Box::new(LdapLoginProvider::new(ld, k2v_region, s3_region)?), - (Some(_), Some(_)) => bail!("A single login provider must be set up in config file"), - (None, None) => bail!("No login provider is set up in config file"), + let login_provider: Arc<dyn LoginProvider + Send + Sync> = + match (config.login_static, config.login_ldap) { + (Some(st), None) => Arc::new(StaticLoginProvider::new(st, k2v_region, s3_region)?), + (None, Some(ld)) => Arc::new(LdapLoginProvider::new(ld, k2v_region, s3_region)?), + (Some(_), Some(_)) => { + bail!("A single login provider must be set up in config file") + } + (None, None) => bail!("No login provider is set up in config file"), + }; + + let lmtp_server = config + .lmtp + .map(|cfg| LmtpServer::new(cfg, login_provider.clone())); + + Ok(Self { + login_provider, + lmtp_server, + }) + } + + pub async fn run(&self) -> Result<()> { + let (exit_signal, provoke_exit) = watch_ctrl_c(); + let exit_on_err = move |err: anyhow::Error| { + error!("Error: {}", err); + let _ = provoke_exit.send(true); }; - Ok(Arc::new(Self { login_provider })) + + try_join!(async { + match self.lmtp_server.as_ref() { + None => Ok(()), + Some(s) => s.run(exit_signal.clone()).await, + } + })?; + Ok(()) } - pub async fn run(self: &Arc<Self>) -> Result<()> { + pub async fn test(&self) -> Result<()> { let creds = self.login_provider.login("lx", "plop").await?; let mut mailbox = Mailbox::new(&creds, "TestMailbox".to_string()).await?; @@ -41,3 +72,17 @@ impl Server { Ok(()) } } + +pub fn watch_ctrl_c() -> (watch::Receiver<bool>, Arc<watch::Sender<bool>>) { + let (send_cancel, watch_cancel) = watch::channel(false); + let send_cancel = Arc::new(send_cancel); + let send_cancel_2 = send_cancel.clone(); + tokio::spawn(async move { + tokio::signal::ctrl_c() + .await + .expect("failed to install CTRL+C signal handler"); + info!("Received CTRL+C, shutting down."); + send_cancel.send(true).unwrap(); + }); + (watch_cancel, send_cancel_2) +} |