Deuxfleurs
/
aerogramme
15
0
Fork 1
Code Issues 33 Pull Requests 1 Projects 11 Releases 4 Wiki Activity

LMTP refactoring, implement EXPUNGE

This commit is contained in:
Alex 2022-07-13 11:19:08 +02:00
parent a1ca6d9def
commit faca15f164
Signed by: lx
GPG Key ID: 0E496D15096376BE
4 changed files with 63 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/imap/mailbox_view.rs
View File

@ -160,7 +160,7 @@ impl MailboxView {
let flags = flags.iter().map(|x| x.to_string()).collect::<Vec<_>>();
let mails = self.get_mail_ids(sequence_set)?;
for (i, uid, uuid) in mails.iter() {
for (_i, _uid, uuid) in mails.iter() {
match kind {
StoreType::Add => {
self.mailbox.add_flags(*uuid, &flags[..]).await?;
@ -178,7 +178,19 @@ impl MailboxView {
}
pub async fn expunge(&mut self) -> Result<Vec<Body>> {
unimplemented!()
let deleted_flag = Flag::Deleted.to_string();
let msgs = self
.known_state
.table
.iter()
.filter(|(_uuid, (_uid, flags))| flags.iter().any(|x| *x == deleted_flag))
.map(|(uuid, _)| *uuid);
for msg in msgs {
self.mailbox.delete(msg).await?;
}
self.update().await
}
/// Looks up state changes in the mailbox and produces a set of IMAP

73
src/lmtp.rs
View File

@ -5,7 +5,11 @@ use anyhow::Result;
use async_trait::async_trait;
use duplexify::Duplex;
use futures::{io, AsyncRead, AsyncReadExt, AsyncWrite};
use futures::{stream, stream::FuturesUnordered, StreamExt};
use futures::{
stream,
stream::{FuturesOrdered, FuturesUnordered},
StreamExt,
};
use log::*;
use tokio::net::TcpListener;
use tokio::select;
@ -56,11 +60,12 @@ impl LmtpServer {
_ = wait_conn_finished => continue,
_ = must_exit.changed() => continue,
};
info!("LMTP: accepted connection from {}", remote_addr);
let conn = tokio::spawn(smtp_server::interact(
socket.compat(),
smtp_server::IsAlreadyTls::No,
Conn { remote_addr },
(),
self.clone(),
));
@ -77,10 +82,6 @@ impl LmtpServer {
// ----
pub struct Conn {
remote_addr: SocketAddr,
}
pub struct Message {
to: Vec<PublicCredentials>,
}
@ -89,21 +90,21 @@ pub struct Message {
impl Config for LmtpServer {
type Protocol = smtp_server::protocol::Lmtp;
type ConnectionUserMeta = Conn;
type ConnectionUserMeta = ();
type MailUserMeta = Message;
fn hostname(&self, _conn_meta: &ConnectionMetadata<Conn>) -> &str {
fn hostname(&self, _conn_meta: &ConnectionMetadata<()>) -> &str {
&self.hostname
}
async fn new_mail(&self, _conn_meta: &mut ConnectionMetadata<Conn>) -> Message {
async fn new_mail(&self, _conn_meta: &mut ConnectionMetadata<()>) -> Message {
Message { to: vec![] }
}
async fn tls_accept<IO>(
&self,
_io: IO,
_conn_meta: &mut ConnectionMetadata<Conn>,
_conn_meta: &mut ConnectionMetadata<()>,
) -> io::Result<Duplex<Pin<Box<dyn Send + AsyncRead>>, Pin<Box<dyn Send + AsyncWrite>>>>
where
IO: Send + AsyncRead + AsyncWrite,
@ -118,7 +119,7 @@ impl Config for LmtpServer {
&self,
from: Option<Email>,
_meta: &mut MailMetadata<Message>,
_conn_meta: &mut ConnectionMetadata<Conn>,
_conn_meta: &mut ConnectionMetadata<()>,
) -> Decision<Option<Email>> {
Decision::Accept {
reply: reply::okay_from().convert(),
@ -130,7 +131,7 @@ impl Config for LmtpServer {
&self,
to: Email,
meta: &mut MailMetadata<Message>,
_conn_meta: &mut ConnectionMetadata<Conn>,
_conn_meta: &mut ConnectionMetadata<()>,
) -> Decision<Email> {
let to_str = match to.hostname.as_ref() {
Some(h) => format!("{}@{}", to.localpart, h),
@ -158,7 +159,7 @@ impl Config for LmtpServer {
&'resp self,
reader: &mut EscapedDataReader<'_, R>,
meta: MailMetadata<Message>,
_conn_meta: &'resp mut ConnectionMetadata<Conn>,
_conn_meta: &'resp mut ConnectionMetadata<()>,
) -> Pin<Box<dyn futures::Stream<Item = Decision<()>> + Send + 'resp>>
where
R: Send + Unpin + AsyncRead,
@ -176,8 +177,8 @@ impl Config for LmtpServer {
};
let mut text = Vec::new();
if reader.read_to_end(&mut text).await.is_err() {
return err_response_stream(meta, "io error".into());
if let Err(e) = reader.read_to_end(&mut text).await {
return err_response_stream(meta, format!("io error: {}", e));
}
reader.complete();
@ -186,23 +187,29 @@ impl Config for LmtpServer {
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())],
},
},
}
}
}))
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<_>>(),
)
}
}

11
src/mail/uidindex.rs
View File

@ -214,9 +214,14 @@ impl FlagIndex {
});
}
fn remove(&mut self, uid: ImapUid, flags: &Vec<Flag>) -> () {
flags.iter().for_each(|flag| {
self.0.get_mut(flag).and_then(|set| set.remove(&uid));
});
for flag in flags.iter() {
if let Some(set) = self.0.get_mut(flag) {
set.remove(&uid);
if set.is_empty() {
self.0.remove(flag);
}
}
}
}
pub fn get(&self, f: &Flag) -> Option<&OrdSet<ImapUid>> {

2
src/server.rs
View File

@ -1,7 +1,7 @@
use std::sync::Arc;
use anyhow::{bail, Result};
use futures::{try_join, StreamExt};
use futures::try_join;
use log::*;
use tokio::sync::watch;