diff options
Diffstat (limited to 'src/mail')
| -rw-r--r-- | src/mail/mailbox.rs | 189 | ||||
| -rw-r--r-- | src/mail/mod.rs | 11 |
2 files changed, 176 insertions, 24 deletions
diff --git a/src/mail/mailbox.rs b/src/mail/mailbox.rs index 6801ab7..3ebaa0b 100644 --- a/src/mail/mailbox.rs +++ b/src/mail/mailbox.rs @@ -1,14 +1,20 @@ -use anyhow::Result; +use anyhow::{anyhow, bail, Result}; use k2v_client::K2vClient; -use rusoto_s3::S3Client; +use k2v_client::{BatchReadOp, Filter, K2vValue}; +use rusoto_s3::{ + DeleteObjectRequest, GetObjectRequest, ListObjectsV2Request, PutObjectRequest, S3Client, S3, +}; +use serde::{Deserialize, Serialize}; +use tokio::io::AsyncReadExt; use tokio::sync::RwLock; use crate::bayou::Bayou; -use crate::cryptoblob::Key; +use crate::cryptoblob::{self, gen_key, open_deserialize, seal_serialize, Key}; use crate::login::Credentials; use crate::mail::uidindex::*; use crate::mail::unique_ident::*; use crate::mail::IMF; +use crate::time::now_msec; pub struct Mailbox { id: UniqueIdent, @@ -48,8 +54,8 @@ impl Mailbox { } /// Insert an email in the mailbox - pub async fn append<'a>(&self, _msg: IMF<'a>) -> Result<()> { - unimplemented!() + pub async fn append<'a>(&self, msg: IMF<'a>) -> Result<()> { + self.mbox.write().await.append(msg, None).await } /// Copy an email from an other Mailbox to this mailbox @@ -58,21 +64,15 @@ impl Mailbox { unimplemented!() } - /// Delete all emails with the \Delete flag in the mailbox - /// Can be called by CLOSE and EXPUNGE - /// @FIXME do we want to implement this feature or a simpler "delete" command - /// The controller could then "fetch \Delete" and call delete on each email? - pub async fn expunge(&self) -> Result<()> { - unimplemented!() - } - - /// Update flags of a range of emails - pub async fn store(&self) -> Result<()> { - unimplemented!() + /// Fetch the metadata (headers + some more info) of the specified + /// mail IDs + pub async fn fetch_meta(&self, ids: &[UniqueIdent]) -> Result<Vec<MailMeta>> { + self.mbox.read().await.fetch_meta(ids).await } - pub async fn fetch(&self) -> Result<()> { - unimplemented!() + /// Fetch an entire e-mail + pub async fn fetch_full(&self, id: UniqueIdent, message_key: &Key) -> Result<Vec<u8>> { + self.mbox.read().await.fetch_full(id, message_key).await } /// Test procedure TODO WILL REMOVE THIS @@ -98,17 +98,142 @@ struct MailboxInternal { } impl MailboxInternal { - pub async fn test(&mut self) -> Result<()> { - self.uid_index.sync().await?; + async fn fetch_meta(&self, ids: &[UniqueIdent]) -> Result<Vec<MailMeta>> { + let ids = ids.iter().map(|x| x.to_string()).collect::<Vec<_>>(); + let ops = ids + .iter() + .map(|id| BatchReadOp { + partition_key: &self.mail_path, + filter: Filter { + start: Some(id), + end: None, + prefix: None, + limit: None, + reverse: false, + }, + single_item: true, + conflicts_only: false, + tombstones: false, + }) + .collect::<Vec<_>>(); + let res_vec = self.k2v.read_batch(&ops).await?; - dump(&self.uid_index); + let mut meta_vec = vec![]; + for res in res_vec { + if res.items.len() != 1 { + bail!("Expected 1 item, got {}", res.items.len()); + } + let (_, cv) = res.items.iter().next().unwrap(); + if cv.value.len() != 1 { + bail!("Expected 1 value, got {}", cv.value.len()); + } + match &cv.value[0] { + K2vValue::Tombstone => bail!("Expected value, got tombstone"), + K2vValue::Value(v) => { + let meta = open_deserialize::<MailMeta>(v, &self.encryption_key)?; + meta_vec.push(meta); + } + } + } + + Ok(meta_vec) + } + + async fn fetch_full(&self, id: UniqueIdent, message_key: &Key) -> Result<Vec<u8>> { + let mut gor = GetObjectRequest::default(); + gor.bucket = self.bucket.clone(); + gor.key = format!("{}/{}", self.mail_path, id); + + let obj_res = self.s3.get_object(gor).await?; + + let obj_body = obj_res.body.ok_or(anyhow!("Missing object body"))?; + let mut buf = Vec::with_capacity(obj_res.content_length.unwrap_or(128) as usize); + obj_body.into_async_read().read_to_end(&mut buf).await?; + + Ok(cryptoblob::open(&buf, &message_key)?) + } + + async fn append(&mut self, mail: IMF<'_>, ident: Option<UniqueIdent>) -> Result<()> { + let ident = ident.unwrap_or_else(|| gen_ident()); + let message_key = gen_key(); + + futures::try_join!( + async { + // Encrypt and save mail body + let message_blob = cryptoblob::seal(mail.raw, &message_key)?; + let mut por = PutObjectRequest::default(); + por.bucket = self.bucket.clone(); + por.key = format!("{}/{}", self.mail_path, ident); + por.body = Some(message_blob.into()); + self.s3.put_object(por).await?; + Ok::<_, anyhow::Error>(()) + }, + async { + // Save mail meta + let meta = MailMeta { + internaldate: now_msec(), + headers: mail.raw[..mail.parsed.offset_body].to_vec(), + message_key: message_key.clone(), + rfc822_size: mail.raw.len(), + }; + let meta_blob = cryptoblob::seal_serialize(&meta, &self.encryption_key)?; + self.k2v + .insert_item(&self.mail_path, &ident.to_string(), meta_blob, None) + .await?; + Ok::<_, anyhow::Error>(()) + } + )?; + // Add mail to Bayou mail index let add_mail_op = self .uid_index .state() - .op_mail_add(gen_ident(), vec!["\\Unseen".into()]); + .op_mail_add(ident, vec!["\\Unseen".into()]); self.uid_index.push(add_mail_op).await?; + Ok(()) + } + + async fn delete(&mut self, ident: UniqueIdent) -> Result<()> { + let del_mail_op = self.uid_index.state().op_mail_del(ident); + self.uid_index.push(del_mail_op).await?; + + futures::try_join!( + async { + // Delete mail body from S3 + let mut dor = DeleteObjectRequest::default(); + dor.bucket = self.bucket.clone(); + dor.key = format!("{}/{}", self.mail_path, ident); + self.s3.delete_object(dor).await?; + Ok::<_, anyhow::Error>(()) + }, + async { + // Delete mail meta from K2V + let sk = ident.to_string(); + let v = self.k2v.read_item(&self.mail_path, &sk).await?; + self.k2v + .delete_item(&self.mail_path, &sk, v.causality) + .await?; + Ok::<_, anyhow::Error>(()) + } + )?; + Ok(()) + } + + // ---- + + async fn test(&mut self) -> Result<()> { + self.uid_index.sync().await?; + + dump(&self.uid_index); + + let mail = br#"From: Garage team <garagehq@deuxfleurs.fr> +Subject: Welcome to Aerogramme!! + +This is just a test email, feel free to ignore."#; + let mail = IMF::try_from(&mail[..]).unwrap(); + self.append(mail, None).await?; + dump(&self.uid_index); if self.uid_index.state().idx_by_uid.len() > 6 { @@ -121,8 +246,8 @@ impl MailboxInternal { .skip(3 + i) .next() .unwrap(); - let del_mail_op = self.uid_index.state().op_mail_del(*ident); - self.uid_index.push(del_mail_op).await?; + + self.delete(*ident).await?; dump(&self.uid_index); } @@ -148,3 +273,19 @@ fn dump(uid_index: &Bayou<UidIndex>) { } println!(""); } + +// ---- + +/// The metadata of a message that is stored in K2V +/// at pk = mail/<mailbox uuid>, sk = <message uuid> +#[derive(Serialize, Deserialize)] +pub struct MailMeta { + /// INTERNALDATE field (milliseconds since epoch) + pub internaldate: u64, + /// Headers of the message + pub headers: Vec<u8>, + /// Secret key for decrypting entire message + pub message_key: Key, + /// RFC822 size + pub rfc822_size: usize, +} diff --git a/src/mail/mod.rs b/src/mail/mod.rs index bbb553a..8dc42c4 100644 --- a/src/mail/mod.rs +++ b/src/mail/mod.rs @@ -1,3 +1,5 @@ +use std::convert::TryFrom; + pub mod mailbox; pub mod uidindex; pub mod unique_ident; @@ -9,3 +11,12 @@ pub struct IMF<'a> { raw: &'a [u8], parsed: mail_parser::Message<'a>, } + +impl<'a> TryFrom<&'a [u8]> for IMF<'a> { + type Error = (); + + fn try_from(body: &'a [u8]) -> Result<IMF<'a>, ()> { + let parsed = mail_parser::Message::parse(body).ok_or(())?; + Ok(Self { raw: body, parsed }) + } +} |