From e1c3190b486f3e89c33bbaeb6f26693fcc197a15 Mon Sep 17 00:00:00 2001 From: Mauro D Date: Thu, 22 Jun 2023 17:29:39 +0000 Subject: [PATCH] IMAP mailbox synchronization --- Cargo.lock | 6 + crates/imap-proto/Cargo.toml | 1 + crates/imap-proto/src/lib.rs | 11 +- crates/imap-proto/src/protocol/status.rs | 2 +- crates/imap/Cargo.toml | 5 + crates/imap/src/core/client.rs | 2 +- crates/imap/src/core/mailbox.rs | 527 +++++++++++++++++++++++ crates/imap/src/core/mod.rs | 56 ++- crates/imap/src/core/session.rs | 19 +- crates/imap/src/core/writer.rs | 2 +- crates/imap/src/lib.rs | 13 +- crates/imap/src/op/authenticate.rs | 286 ++++++++++++ crates/imap/src/op/list.rs | 447 +++++++++++++++++++ crates/imap/src/op/mod.rs | 7 + crates/imap/src/op/status.rs | 373 ++++++++++++++++ crates/jmap/src/api/session.rs | 16 +- crates/jmap/src/auth/authenticate.rs | 62 +-- crates/jmap/src/auth/mod.rs | 17 + crates/jmap/src/changes/get.rs | 30 +- crates/jmap/src/mailbox/get.rs | 2 +- crates/jmap/src/sieve/ingest.rs | 9 +- crates/store/src/query/get.rs | 43 ++ crates/store/src/query/log.rs | 4 +- 23 files changed, 1849 insertions(+), 91 deletions(-) create mode 100644 crates/imap/src/core/mailbox.rs create mode 100644 crates/imap/src/op/authenticate.rs create mode 100644 crates/imap/src/op/list.rs create mode 100644 crates/imap/src/op/mod.rs create mode 100644 crates/imap/src/op/status.rs diff --git a/Cargo.lock b/Cargo.lock index 1306ca89..e0790527 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1708,12 +1708,17 @@ dependencies = [ name = "imap" version = "0.1.0" dependencies = [ + "ahash 0.8.3", "directory", "imap_proto", "jmap", + "jmap_proto", + "mail-parser", + "mail-send", "parking_lot", "rustls 0.21.1", "rustls-pemfile", + "store", "tokio", "tokio-rustls 0.24.0", "tracing", @@ -1726,6 +1731,7 @@ version = "0.1.0" dependencies = [ "ahash 0.8.3", "chrono", + "jmap_proto", "mail-parser", ] diff --git a/crates/imap-proto/Cargo.toml b/crates/imap-proto/Cargo.toml index c12d6347..faa16d59 100644 --- a/crates/imap-proto/Cargo.toml +++ b/crates/imap-proto/Cargo.toml @@ -5,6 +5,7 @@ edition = "2021" resolver = "2" [dependencies] +jmap_proto = { path = "../jmap-proto" } mail-parser = { git = "https://github.com/stalwartlabs/mail-parser", features = ["full_encoding", "serde_support", "ludicrous_mode"] } ahash = { version = "0.8" } chrono = { version = "0.4"} diff --git a/crates/imap-proto/src/lib.rs b/crates/imap-proto/src/lib.rs index 44d8026e..de46327a 100644 --- a/crates/imap-proto/src/lib.rs +++ b/crates/imap-proto/src/lib.rs @@ -1,5 +1,6 @@ use std::borrow::Cow; +use jmap_proto::error::method::MethodError; use protocol::capability::Capability; pub mod parser; @@ -191,8 +192,8 @@ impl StatusResponse { self } - pub fn with_tag(mut self, tag: String) -> Self { - self.tag = Some(tag); + pub fn with_tag(mut self, tag: impl Into) -> Self { + self.tag = Some(tag.into()); self } @@ -224,4 +225,10 @@ impl StatusResponse { } } +impl From for StatusResponse { + fn from(_: MethodError) -> Self { + StatusResponse::database_failure() + } +} + pub type Result = std::result::Result; diff --git a/crates/imap-proto/src/protocol/status.rs b/crates/imap-proto/src/protocol/status.rs index b50de1b4..2bee93f8 100644 --- a/crates/imap-proto/src/protocol/status.rs +++ b/crates/imap-proto/src/protocol/status.rs @@ -32,7 +32,7 @@ pub struct Arguments { pub items: Vec, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum Status { Messages, UidNext, diff --git a/crates/imap/Cargo.toml b/crates/imap/Cargo.toml index fe6a03d1..00e32295 100644 --- a/crates/imap/Cargo.toml +++ b/crates/imap/Cargo.toml @@ -7,11 +7,16 @@ resolver = "2" [dependencies] imap_proto = { path = "../imap-proto" } jmap = { path = "../jmap" } +jmap_proto = { path = "../jmap-proto" } directory = { path = "../directory" } +store = { path = "../store" } utils = { path = "../utils" } +mail-parser = { git = "https://github.com/stalwartlabs/mail-parser", features = ["full_encoding", "ludicrous_mode"] } +mail-send = { git = "https://github.com/stalwartlabs/mail-send", default-features = false, features = ["cram-md5", "skip-ehlo"] } rustls = "0.21.0" rustls-pemfile = "1.0" tokio = { version = "1.23", features = ["full"] } tokio-rustls = { version = "0.24.0"} parking_lot = "0.12" tracing = "0.1" +ahash = { version = "0.8" } diff --git a/crates/imap/src/core/client.rs b/crates/imap/src/core/client.rs index 6c83d69c..a455e9a6 100644 --- a/crates/imap/src/core/client.rs +++ b/crates/imap/src/core/client.rs @@ -32,7 +32,7 @@ use tokio::io::AsyncRead; use super::{SelectedMailbox, Session, SessionData, State}; impl Session { - pub async fn ingest(&mut self, bytes: &[u8]) -> Result { + pub async fn ingest(&mut self, bytes: &[u8]) -> crate::Result { /*let tmp = "dd"; for line in String::from_utf8_lossy(bytes).split("\r\n") { println!("<- {:?}", &line[..std::cmp::min(line.len(), 100)]); diff --git a/crates/imap/src/core/mailbox.rs b/crates/imap/src/core/mailbox.rs new file mode 100644 index 00000000..d31b8be8 --- /dev/null +++ b/crates/imap/src/core/mailbox.rs @@ -0,0 +1,527 @@ +use std::{collections::BTreeMap, sync::atomic::Ordering}; + +use ahash::AHashMap; +use imap_proto::{protocol::list::Attribute, StatusResponse}; +use jmap::{auth::AccessToken, mailbox::INBOX_ID, SUPERUSER_ID}; +use jmap_proto::{ + object::Object, + types::{acl::Acl, collection::Collection, property::Property, value::Value}, +}; +use parking_lot::Mutex; +use store::query::log::{Change, Query}; +use tokio::io::AsyncRead; + +use super::{Account, Mailbox, MailboxId, MailboxSync, Session, SessionData}; + +impl SessionData { + pub async fn new( + session: &Session, + access_token: &AccessToken, + ) -> crate::Result { + let mut session = SessionData { + writer: session.writer.clone(), + jmap: session.jmap.clone(), + imap: session.imap.clone(), + account_id: access_token.primary_id(), + span: session.span.clone(), + mailboxes: Mutex::new(vec![]), + state: access_token.state().into(), + }; + + // Fetch mailboxes for the main account + let mut mailboxes = vec![session + .fetch_account_mailboxes(session.account_id, None, access_token) + .await + .map_err(|_| tracing::warn!(parent: &session.span, account_id = session.account_id, event = "error", "Failed to retrieve mailboxes."))?]; + + // Fetch shared mailboxes + for &account_id in access_token.shared_accounts(Collection::Mailbox) { + if account_id != SUPERUSER_ID { + match session + .fetch_account_mailboxes( + account_id, + format!( + "{}/{}", + session.imap.name_shared, + session.jmap.get_account_name(account_id).await + ) + .into(), + access_token, + ) + .await + { + Ok(account_mailboxes) => { + mailboxes.push(account_mailboxes); + } + Err(_) => { + tracing::warn!(parent: &session.span, account_id = account_id, event = "error", "Failed to retrieve mailboxes."); + } + } + } + } + + session.mailboxes = Mutex::new(mailboxes); + + Ok(session) + } + + async fn fetch_account_mailboxes( + &self, + account_id: u32, + mailbox_prefix: Option, + access_token: &AccessToken, + ) -> crate::Result { + let mailbox_ids = if access_token.is_primary_id(account_id) { + self.jmap + .mailbox_get_or_create(account_id) + .await + .map_err(|_| {})? + } else if access_token.member_of.contains(&account_id) { + self.jmap + .get_document_ids(account_id, Collection::Mailbox) + .await + .map_err(|_| {})? + .unwrap_or_default() + } else { + self.jmap + .shared_documents(access_token, account_id, Collection::Mailbox, Acl::Read) + .await + .map_err(|_| {})? + }; + + // Fetch mailboxes + let mut mailboxes = Vec::with_capacity(10); + for mailbox_id in mailbox_ids { + mailboxes.push( + match self + .jmap + .get_property::>( + account_id, + Collection::Mailbox, + mailbox_id, + &Property::Value, + ) + .await + .map_err(|_| {})? + { + Some(values) => ( + mailbox_id, + values + .properties + .get(&Property::ParentId) + .map(|parent_id| match parent_id { + Value::Id(value) => value.document_id(), + _ => 0, + }) + .unwrap_or(0), + values, + ), + None => { + continue; + } + }, + ); + } + + // Build tree + let mut iter = mailboxes.iter(); + let mut parent_id = 0; + let mut path = Vec::new(); + let mut iter_stack = Vec::new(); + let message_ids = self + .jmap + .get_document_ids(account_id, Collection::Email) + .await + .map_err(|_| {})?; + + if let Some(mailbox_prefix) = &mailbox_prefix { + path.push(mailbox_prefix.to_string()); + }; + + let mut account = Account { + account_id, + prefix: mailbox_prefix, + mailbox_names: BTreeMap::new(), + mailbox_data: AHashMap::with_capacity(mailboxes.len()), + state: self + .jmap + .store + .get_last_change_id(account_id, Collection::Mailbox) + .await + .map_err(|_| {})?, + }; + + loop { + while let Some((mailbox_id, mailbox_parent_id, mailbox)) = iter.next() { + if *mailbox_parent_id == parent_id { + let mut mailbox_path = path.clone(); + if *mailbox_id != INBOX_ID || account.prefix.is_some() { + mailbox_path.push( + mailbox + .get(&Property::Name) + .as_string() + .unwrap_or_default() + .to_string(), + ); + } else { + mailbox_path.push("INBOX".to_string()); + } + let has_children = mailboxes + .iter() + .any(|(_, child_parent_id, _)| child_parent_id == mailbox_id); + + account.mailbox_data.insert( + *mailbox_id, + Mailbox { + has_children, + is_subscribed: mailbox + .properties + .get(&Property::IsSubscribed) + .map(|parent_id| match parent_id { + Value::List(values) => values + .contains(&Value::Id(access_token.primary_id().into())), + _ => false, + }) + .unwrap_or(false), + special_use: mailbox.properties.get(&Property::Role).and_then( + |parent_id| match parent_id { + Value::Text(role) => match role.as_str() { + "archive" => Some(Attribute::Archive), + "drafts" => Some(Attribute::Drafts), + "junk" => Some(Attribute::Junk), + "sent" => Some(Attribute::Sent), + "trash" => Some(Attribute::Trash), + "important" => Some(Attribute::Important), + _ => None, + }, + _ => None, + }, + ), + total_messages: self + .jmap + .get_tag( + account_id, + Collection::Email, + Property::MailboxIds, + *mailbox_id, + ) + .await + .map_err(|_| {})? + .map(|v| v.len() as u32) + .unwrap_or(0) + .into(), + total_unseen: self + .jmap + .mailbox_unread_tags(account_id, *mailbox_id, &message_ids) + .await + .map_err(|_| {})? + .map(|v| v.len() as u32) + .unwrap_or(0) + .into(), + ..Default::default() + }, + ); + account + .mailbox_names + .insert(mailbox_path.join("/"), *mailbox_id); + + if has_children && iter_stack.len() < 100 { + iter_stack.push((iter, parent_id, path)); + parent_id = *mailbox_id; + path = mailbox_path; + iter = mailboxes.iter(); + } + } + } + + if let Some((prev_iter, prev_parent_id, prev_path)) = iter_stack.pop() { + iter = prev_iter; + parent_id = prev_parent_id; + path = prev_path; + } else { + break; + } + } + + Ok(account) + } + + pub async fn synchronize_mailboxes( + &self, + return_changes: bool, + ) -> crate::Result> { + let mut changes = if return_changes { + MailboxSync::default().into() + } else { + None + }; + + // Obtain access token + let access_token = self + .jmap + .get_cached_access_token(self.account_id) + .await + .ok_or(())?; + let state = access_token.state(); + + // Shared mailboxes might have changed + let mut added_accounts = Vec::new(); + if self.state.load(Ordering::Relaxed) != state { + // Remove unlinked shared accounts + let mut added_account_ids = Vec::new(); + { + let mut mailboxes = self.mailboxes.lock(); + let mut new_accounts = Vec::with_capacity(mailboxes.len()); + let has_access_to = access_token + .shared_accounts(Collection::Mailbox) + .copied() + .collect::>(); + for account in mailboxes.drain(..) { + if access_token.is_primary_id(account.account_id) + || has_access_to.contains(&account.account_id) + { + new_accounts.push(account); + } else { + tracing::debug!(parent: &self.span, "Removed unlinked shared account {}", account.account_id); + + // Add unshared mailboxes to deleted list + if let Some(changes) = &mut changes { + for (mailbox_name, _) in account.mailbox_names { + changes.deleted.push(mailbox_name); + } + } + } + } + + // Add new shared account ids + for account_id in has_access_to { + if account_id != SUPERUSER_ID + && !new_accounts + .iter() + .skip(1) + .any(|m| m.account_id == account_id) + { + tracing::debug!(parent: &self.span, "Adding shared account {}", account_id); + added_account_ids.push(account_id); + } + } + *mailboxes = new_accounts; + } + + // Fetch mailboxes for each new shared account + for account_id in added_account_ids { + let prefix = format!( + "{}/{}", + self.imap.name_shared, + self.jmap.get_account_name(account_id).await + ); + match self + .fetch_account_mailboxes(account_id, prefix.into(), &access_token) + .await + { + Ok(account) => { + added_accounts.push(account); + } + Err(_) => { + tracing::debug!(parent: &self.span, "Failed to fetch shared mailbox."); + } + } + } + + // Update state + self.state.store(state, Ordering::Relaxed); + } + + // Fetch mailbox changes for all accounts + let mut changed_accounts = Vec::new(); + let account_states = self + .mailboxes + .lock() + .iter() + .map(|m| (m.account_id, m.state)) + .collect::>(); + for (account_id, last_state) in account_states { + let changelog = self + .jmap + .changes_( + account_id, + Collection::Mailbox, + last_state.map(Query::Since).unwrap_or(Query::All), + ) + .await + .map_err(|_| {})?; + if !changelog.changes.is_empty() { + let mut has_changes = false; + let mut has_child_changes = false; + + for change in changelog.changes { + match change { + Change::Insert(_) | Change::Update(_) | Change::Delete(_) => { + has_changes = true + } + Change::ChildUpdate(_) => has_child_changes = true, + } + } + + if has_child_changes && !has_changes && changes.is_none() { + // Only child changes, no need to re-fetch mailboxes + for account in self.mailboxes.lock().iter_mut() { + if account.account_id == account_id { + account.mailbox_data.values_mut().for_each(|v| { + v.total_deleted = None; + v.total_unseen = None; + v.total_messages = None; + v.size = None; + v.uid_next = None; + }); + account.state = changelog.to_change_id.into(); + break; + } + } + } else { + // Refresh mailboxes for changed account + let mailbox_prefix = if !access_token.is_primary_id(account_id) { + format!( + "{}/{}", + self.imap.name_shared, + self.jmap.get_account_name(account_id).await + ) + .into() + } else { + None + }; + match self + .fetch_account_mailboxes(account_id, mailbox_prefix, &access_token) + .await + { + Ok(account_mailboxes) => { + changed_accounts.push(account_mailboxes); + } + Err(_) => { + tracing::debug!(parent: &self.span, "Failed to fetch mailboxes:."); + } + } + } + } + } + + // Update mailboxes + if !changed_accounts.is_empty() || !added_accounts.is_empty() { + let mut mailboxes = self.mailboxes.lock(); + + for changed_account in changed_accounts { + if let Some(pos) = mailboxes + .iter() + .position(|a| a.account_id == changed_account.account_id) + { + // Add changes and deletions + if let Some(changes) = &mut changes { + let old_account = &mailboxes[pos]; + let new_account = &changed_account; + + // Add new mailboxes + for (mailbox_name, mailbox_id) in new_account.mailbox_names.iter() { + if let Some(old_mailbox) = old_account.mailbox_data.get(mailbox_id) { + if let Some(mailbox) = new_account.mailbox_data.get(mailbox_id) { + if mailbox.total_messages.unwrap_or(0) + != old_mailbox.total_messages.unwrap_or(0) + || mailbox.total_unseen.unwrap_or(0) + != old_mailbox.total_unseen.unwrap_or(0) + { + changes.changed.push(mailbox_name.to_string()); + } + } + } else { + changes.added.push(mailbox_name.to_string()); + } + } + + // Add deleted mailboxes + for (mailbox_name, mailbox_id) in &old_account.mailbox_names { + if !new_account.mailbox_data.contains_key(mailbox_id) { + changes.deleted.push(mailbox_name.to_string()); + } + } + } + + mailboxes[pos] = changed_account; + } else { + // Add newly shared accounts + if let Some(changes) = &mut changes { + changes + .added + .extend(changed_account.mailbox_names.keys().cloned()); + } + + mailboxes.push(changed_account); + } + } + + if !added_accounts.is_empty() { + // Add newly shared accounts + if let Some(changes) = &mut changes { + for added_account in &added_accounts { + changes + .added + .extend(added_account.mailbox_names.keys().cloned()); + } + } + mailboxes.extend(added_accounts); + } + } + + Ok(changes) + } + + pub async fn try_synchronize_mailboxes(&self, tag: &str) -> bool { + if self.synchronize_mailboxes(false).await.is_ok() { + true + } else { + tracing::warn!(parent: &self.span, + event = "error", + context = "synchronize_mailboxes", + account_id = self.account_id, + "Failed to synchronize mailboxes."); + self.write_bytes( + StatusResponse::database_failure() + .with_tag(tag) + .into_bytes(), + ) + .await; + false + } + } + + pub fn get_mailbox_by_name(&self, mailbox_name: &str) -> Option { + if !self.is_all_mailbox(mailbox_name) { + for account in self.mailboxes.lock().iter() { + if account + .prefix + .as_ref() + .map_or(true, |p| mailbox_name.starts_with(p)) + { + for (mailbox_name_, mailbox_id_) in account.mailbox_names.iter() { + if mailbox_name_ == mailbox_name { + return MailboxId { + account_id: account.account_id, + mailbox_id: Some(*mailbox_id_), + } + .into(); + } + } + } + } + None + } else { + MailboxId { + account_id: self.account_id, + mailbox_id: None, + } + .into() + } + } + + pub fn is_all_mailbox(&self, mailbox_name: &str) -> bool { + self.imap.name_all == mailbox_name + } +} diff --git a/crates/imap/src/core/mod.rs b/crates/imap/src/core/mod.rs index 0382fa79..b9b6bb2d 100644 --- a/crates/imap/src/core/mod.rs +++ b/crates/imap/src/core/mod.rs @@ -1,14 +1,16 @@ use std::{ - net::{IpAddr, SocketAddr}, - sync::Arc, + collections::BTreeMap, + sync::{atomic::AtomicU32, Arc}, time::Duration, }; -use imap_proto::{protocol::ProtocolVersion, receiver::Receiver, Command}; -use jmap::{ - auth::{rate_limit::RemoteAddress, AccessToken}, - JMAP, +use ahash::AHashMap; +use imap_proto::{ + protocol::{list::Attribute, ProtocolVersion}, + receiver::Receiver, + Command, }; +use jmap::{auth::rate_limit::RemoteAddress, JMAP}; use tokio::{ io::{AsyncRead, ReadHalf}, sync::{mpsc, watch}, @@ -16,6 +18,7 @@ use tokio::{ use utils::listener::{limiter::InFlight, ServerInstance}; pub mod client; +pub mod mailbox; pub mod session; pub mod writer; @@ -55,14 +58,41 @@ pub struct Session { pub is_qresync: bool, pub writer: mpsc::Sender, pub stream_rx: ReadHalf, - pub in_flight: Vec, + pub in_flight: InFlight, + pub remote_addr: RemoteAddress, pub span: tracing::Span, } pub struct SessionData { - pub core: Arc, + pub account_id: u32, + pub jmap: Arc, + pub imap: Arc, + pub span: tracing::Span, + pub mailboxes: parking_lot::Mutex>, pub writer: mpsc::Sender, - pub access_token: Arc, + pub state: AtomicU32, +} + +#[derive(Debug, Default)] +pub struct Mailbox { + pub has_children: bool, + pub is_subscribed: bool, + pub special_use: Option, + pub total_messages: Option, + pub total_unseen: Option, + pub total_deleted: Option, + pub uid_validity: Option, + pub uid_next: Option, + pub size: Option, +} + +#[derive(Debug)] +pub struct Account { + pub account_id: u32, + pub prefix: Option, + pub mailbox_names: BTreeMap, + pub mailbox_data: AHashMap, + pub state: Option, } pub struct SelectedMailbox { @@ -89,6 +119,13 @@ pub struct MailboxData { pub last_state: u32, } +#[derive(Debug, Default)] +pub struct MailboxSync { + pub added: Vec, + pub changed: Vec, + pub deleted: Vec, +} + pub enum SavedSearch { InFlight { rx: watch::Receiver>>, @@ -107,7 +144,6 @@ pub struct ImapId { pub enum State { NotAuthenticated { - remote_addr: RemoteAddress, auth_failures: u32, }, Authenticated { diff --git a/crates/imap/src/core/session.rs b/crates/imap/src/core/session.rs index 4ae87d58..567a1a53 100644 --- a/crates/imap/src/core/session.rs +++ b/crates/imap/src/core/session.rs @@ -87,7 +87,7 @@ impl Session { impl Session { pub async fn new( - mut session: utils::listener::SessionData, + mut session: SessionData, manager: ImapSessionManager, ) -> Result, ()> { // Write plain text greeting @@ -102,10 +102,7 @@ impl Session { Ok(Session { receiver: Receiver::with_max_request_size(manager.imap.max_request_size), version: ProtocolVersion::Rev1, - state: State::NotAuthenticated { - auth_failures: 0, - remote_addr: RemoteAddress::IpAddress(session.remote_ip), - }, + state: State::NotAuthenticated { auth_failures: 0 }, writer: writer::spawn_writer(writer::Event::Stream(stream_tx)), is_tls: false, is_condstore: false, @@ -114,7 +111,8 @@ impl Session { jmap: manager.jmap, instance: session.instance, span: session.span, - in_flight: vec![session.in_flight], + in_flight: session.in_flight, + remote_addr: RemoteAddress::IpAddress(session.remote_ip), stream_rx, }) } @@ -162,6 +160,7 @@ impl Session { writer: self.writer, span: self.span, in_flight: self.in_flight, + remote_addr: self.remote_addr, stream_rx, }) } @@ -191,10 +190,7 @@ impl Session> { Ok(Session { receiver: Receiver::with_max_request_size(manager.imap.max_request_size), version: ProtocolVersion::Rev1, - state: State::NotAuthenticated { - auth_failures: 0, - remote_addr: RemoteAddress::IpAddress(session.remote_ip), - }, + state: State::NotAuthenticated { auth_failures: 0 }, writer: writer::spawn_writer(writer::Event::StreamTls(stream_tx)), is_tls: true, is_condstore: false, @@ -203,7 +199,8 @@ impl Session> { jmap: manager.jmap, instance: session.instance, span, - in_flight: vec![session.in_flight], + in_flight: session.in_flight, + remote_addr: RemoteAddress::IpAddress(session.remote_ip), stream_rx, }) } diff --git a/crates/imap/src/core/writer.rs b/crates/imap/src/core/writer.rs index 65200743..2b369088 100644 --- a/crates/imap/src/core/writer.rs +++ b/crates/imap/src/core/writer.rs @@ -109,7 +109,7 @@ pub fn spawn_writer(mut stream: Event) -> mpsc::Sender { } impl Session { - pub async fn write_bytes(&self, bytes: impl Into>) -> Result<(), ()> { + pub async fn write_bytes(&self, bytes: impl Into>) -> crate::OpResult { /*let tmp = "dd"; println!( "-> {:?}", diff --git a/crates/imap/src/lib.rs b/crates/imap/src/lib.rs index 814538e3..a67d7438 100644 --- a/crates/imap/src/lib.rs +++ b/crates/imap/src/lib.rs @@ -2,11 +2,11 @@ use std::sync::Arc; use crate::core::IMAP; -use directory::DirectoryConfig; use imap_proto::{protocol::capability::Capability, ResponseCode, StatusResponse}; use utils::config::Config; pub mod core; +pub mod op; static SERVER_GREETING: &str = concat!( "Stalwart IMAP4rev2 v", @@ -15,7 +15,7 @@ static SERVER_GREETING: &str = concat!( ); impl IMAP { - pub async fn init(config: &Config, directory: &DirectoryConfig) -> Result, String> { + pub async fn init(config: &Config) -> utils::config::Result> { Ok(Arc::new(IMAP { max_request_size: config.property_or_static("imap.request.max-size", "52428800")?, name_shared: config @@ -26,13 +26,13 @@ impl IMAP { .value("imap.folders.name.all") .unwrap_or("All Mail") .to_string(), + timeout_auth: config.property_or_static("imap.timeout.authenticated", "30m")?, + timeout_unauth: config.property_or_static("imap.timeout.anonymous", "1m")?, greeting_plain: StatusResponse::ok(SERVER_GREETING) .with_code(ResponseCode::Capability { capabilities: Capability::all_capabilities(false, false), }) .into_bytes(), - timeout_auth: config.property_or_static("imap.timeout.authenticated", "30m")?, - timeout_unauth: config.property_or_static("imap.timeout.anonymous", "1m")?, greeting_tls: StatusResponse::ok(SERVER_GREETING) .with_code(ResponseCode::Capability { capabilities: Capability::all_capabilities(false, true), @@ -41,3 +41,8 @@ impl IMAP { })) } } + +pub struct ImapError; + +pub type Result = std::result::Result; +pub type OpResult = std::result::Result<(), ()>; diff --git a/crates/imap/src/op/authenticate.rs b/crates/imap/src/op/authenticate.rs new file mode 100644 index 00000000..99c2b224 --- /dev/null +++ b/crates/imap/src/op/authenticate.rs @@ -0,0 +1,286 @@ +/* + * Copyright (c) 2020-2022, Stalwart Labs Ltd. + * + * This file is part of the Stalwart IMAP Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::sync::Arc; + +use imap_proto::{ + protocol::{authenticate::Mechanism, capability::Capability}, + receiver::{self, Request}, + Command, ResponseCode, StatusResponse, +}; +use mail_parser::decoders::base64::base64_decode; +use mail_send::Credentials; +use tokio::io::AsyncRead; + +use crate::core::{Session, SessionData, State}; + +impl Session { + pub async fn handle_authenticate(&mut self, request: Request) -> crate::OpResult { + match request.parse_authenticate() { + Ok(mut args) => match args.mechanism { + Mechanism::Plain | Mechanism::OAuthBearer => { + if !args.params.is_empty() { + match base64_decode(args.params.pop().unwrap().as_bytes()) { + Some(challenge) => { + let result = if args.mechanism == Mechanism::Plain { + decode_challenge_plain(&challenge) + } else { + decode_challenge_oauth(&challenge) + }; + + match result { + Ok(credentials) => { + self.authenticate(credentials, args.tag).await + } + Err(err) => { + self.write_bytes( + StatusResponse::no(err).with_tag(args.tag).into_bytes(), + ) + .await + } + } + } + None => { + self.write_bytes( + StatusResponse::no("Failed to decode challenge.") + .with_tag(args.tag) + .with_code(ResponseCode::Parse) + .into_bytes(), + ) + .await + } + } + } else { + self.receiver.request = receiver::Request { + tag: args.tag, + command: Command::Authenticate, + tokens: vec![receiver::Token::Argument(args.mechanism.into_bytes())], + }; + self.receiver.state = receiver::State::Argument { last_ch: b' ' }; + self.write_bytes(b"+ \"\"\r\n".to_vec()).await + } + } + _ => { + self.write_bytes( + StatusResponse::no("Authentication mechanism not supported.") + .with_tag(args.tag) + .with_code(ResponseCode::Cannot) + .into_bytes(), + ) + .await + } + }, + Err(response) => self.write_bytes(response.into_bytes()).await, + } + } + + pub async fn authenticate( + &mut self, + credentials: Credentials, + tag: String, + ) -> crate::Result<()> { + // Throttle authentication requests + if self.jmap.is_auth_allowed(self.remote_addr.clone()).is_err() { + self.write_bytes( + StatusResponse::bye("Too many authentication requests from this IP address.") + .into_bytes(), + ) + .await?; + tracing::debug!(parent: &self.span, + event = "disconnect", + "Too many authentication attempts, disconnecting.", + ); + return Err(()); + } + + // Authenticate + let access_token = match credentials { + Credentials::Plain { username, secret } | Credentials::XOauth2 { username, secret } => { + self.jmap.authenticate_plain(&username, &secret).await + } + Credentials::OAuthBearer { token } => { + match self + .jmap + .validate_access_token("access_token", &token) + .await + { + Ok((account_id, _, _)) => self.jmap.get_access_token(account_id).await, + Err(err) => { + tracing::debug!( + parent: &self.span, + context = "authenticate", + err = err, + "Failed to validate access token." + ); + None + } + } + } + }; + + if let Some(access_token) = access_token { + // Cache access token + let access_token = Arc::new(access_token); + self.jmap.cache_access_token(access_token.clone()); + + // Create session + self.state = State::Authenticated { + data: Arc::new(SessionData::new(self, &access_token).await?), + }; + self.write_bytes( + StatusResponse::ok("Authentication successful") + .with_code(ResponseCode::Capability { + capabilities: Capability::all_capabilities(true, self.is_tls), + }) + .with_tag(tag) + .into_bytes(), + ) + .await?; + Ok(()) + } else { + self.write_bytes( + StatusResponse::no("Authentication failed") + .with_tag(tag) + .with_code(ResponseCode::AuthenticationFailed) + .into_bytes(), + ) + .await?; + + let auth_failures = self.state.auth_failures(); + if auth_failures < 3 { + self.state = State::NotAuthenticated { + auth_failures: auth_failures + 1, + }; + Ok(()) + } else { + self.write_bytes( + StatusResponse::bye("Too many authentication failures").into_bytes(), + ) + .await?; + tracing::debug!( + parent: &self.span, + event = "disconnect", + "Too many authentication failures, disconnecting.", + ); + Err(()) + } + } + } + + pub async fn handle_unauthenticate(&mut self, request: Request) -> crate::OpResult { + self.state = State::NotAuthenticated { auth_failures: 0 }; + + self.write_bytes( + StatusResponse::completed(Command::Unauthenticate) + .with_tag(request.tag) + .into_bytes(), + ) + .await + } +} + +pub fn decode_challenge_plain(challenge: &[u8]) -> Result, &'static str> { + let mut username = Vec::new(); + let mut secret = Vec::new(); + let mut arg_num = 0; + for &ch in challenge { + if ch != 0 { + if arg_num == 1 { + username.push(ch); + } else if arg_num == 2 { + secret.push(ch); + } + } else { + arg_num += 1; + } + } + + match (String::from_utf8(username), String::from_utf8(secret)) { + (Ok(username), Ok(secret)) if !username.is_empty() && !secret.is_empty() => { + Ok((username, secret).into()) + } + _ => Err("Invalid AUTH=PLAIN challenge."), + } +} + +pub fn decode_challenge_oauth(challenge: &[u8]) -> Result, &'static str> { + let mut saw_marker = true; + for (pos, &ch) in challenge.iter().enumerate() { + if saw_marker { + if challenge + .get(pos..) + .map_or(false, |b| b.starts_with(b"auth=Bearer ")) + { + let pos = pos + 12; + return Ok(Credentials::OAuthBearer { + token: String::from_utf8( + challenge + .get( + pos..pos + + challenge + .get(pos..) + .and_then(|c| c.iter().position(|&ch| ch == 0x01)) + .unwrap_or(challenge.len()), + ) + .ok_or("Failed to find end of bearer token")? + .to_vec(), + ) + .map_err(|_| "Bearer token is not a valid UTF-8 string.")?, + }); + } else { + saw_marker = false; + } + } else if ch == 0x01 { + saw_marker = true; + } + } + + Err("Failed to find 'auth=Bearer' in challenge.") +} + +#[cfg(test)] +mod tests { + use mail_parser::decoders::base64::base64_decode; + use mail_send::Credentials; + + #[test] + fn decode_challenge_oauth() { + assert!( + Credentials::OAuthBearer { + token: "vF9dft4qmTc2Nvb3RlckBhbHRhdmlzdGEuY29tCg==".to_string() + } == super::decode_challenge_oauth( + &base64_decode( + concat!( + "bixhPXVzZXJAZXhhbXBsZS5jb20sAWhv", + "c3Q9c2VydmVyLmV4YW1wbGUuY29tAXBvcnQ9MTQzAWF1dGg9QmVhcmVyI", + "HZGOWRmdDRxbVRjMk52YjNSbGNrQmhiSFJoZG1semRHRXVZMjl0Q2c9PQ", + "EB" + ) + .as_bytes(), + ) + .unwrap(), + ) + .unwrap() + ); + } +} diff --git a/crates/imap/src/op/list.rs b/crates/imap/src/op/list.rs new file mode 100644 index 00000000..efd33172 --- /dev/null +++ b/crates/imap/src/op/list.rs @@ -0,0 +1,447 @@ +/* + * Copyright (c) 2020-2022, Stalwart Labs Ltd. + * + * This file is part of the Stalwart IMAP Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use imap_proto::{ + protocol::{ + list::{ + self, Arguments, Attribute, ChildInfo, ListItem, ReturnOption, SelectionOption, Tag, + }, + ImapResponse, ProtocolVersion, + }, + receiver::Request, + Command, StatusResponse, +}; + +use tokio::io::AsyncRead; + +use crate::core::{Session, SessionData}; + +impl Session { + pub async fn handle_list(&mut self, request: Request) -> crate::OpResult { + let command = request.command; + let is_lsub = command == Command::Lsub; + match if !is_lsub { + request.parse_list(self.version) + } else { + request.parse_lsub() + } { + Ok(arguments) => { + if !arguments.is_separator_query() { + let data = self.state.session_data(); + let version = self.version; + tokio::spawn(async move { + data.list(arguments, is_lsub, version).await; + }); + Ok(()) + } else { + self.write_bytes( + StatusResponse::completed(command) + .with_tag(arguments.unwrap_tag()) + .serialize( + list::Response { + is_rev2: self.version.is_rev2(), + is_lsub, + list_items: vec![ListItem { + mailbox_name: String::new(), + attributes: vec![Attribute::NoSelect], + tags: vec![], + }], + status_items: Vec::new(), + } + .serialize(), + ), + ) + .await + } + } + Err(response) => self.write_bytes(response.into_bytes()).await, + } + } +} + +impl SessionData { + pub async fn list(&self, arguments: Arguments, is_lsub: bool, version: ProtocolVersion) { + let (tag, reference_name, mut patterns, selection_options, return_options) = match arguments + { + Arguments::Basic { + tag, + reference_name, + mailbox_name, + } => ( + tag, + reference_name, + vec![mailbox_name], + Vec::new(), + Vec::new(), + ), + Arguments::Extended { + tag, + reference_name, + mailbox_name, + selection_options, + return_options, + } => ( + tag, + reference_name, + mailbox_name, + selection_options, + return_options, + ), + }; + + // Refresh mailboxes + if !self.try_synchronize_mailboxes(&tag).await { + return; + } + + // Process arguments + let mut filter_subscribed = false; + let mut filter_special_use = false; + let mut recursive_match = false; + let mut include_special_use = version.is_rev2(); + let mut include_subscribed = false; + let mut include_children = false; + let mut include_status = None; + for selection_option in &selection_options { + match selection_option { + SelectionOption::Subscribed => { + filter_subscribed = true; + include_subscribed = true; + } + SelectionOption::Remote => (), + SelectionOption::SpecialUse => { + filter_special_use = true; + include_special_use = true; + } + SelectionOption::RecursiveMatch => { + recursive_match = true; + } + } + } + for return_option in &return_options { + match return_option { + ReturnOption::Subscribed => { + include_subscribed = true; + } + ReturnOption::Children => { + include_children = true; + } + ReturnOption::Status(status) => { + include_status = status.into(); + } + ReturnOption::SpecialUse => { + include_special_use = true; + } + } + } + if recursive_match && !filter_subscribed { + self.write_bytes( + StatusResponse::bad("RECURSIVEMATCH cannot be the only selection option.") + .with_tag(tag) + .into_bytes(), + ) + .await; + return; + } + + // Append reference name + if !patterns.is_empty() && !reference_name.is_empty() { + patterns.iter_mut().for_each(|item| { + *item = format!("{}{}", reference_name, item); + }) + } + + let mut list_items = Vec::with_capacity(10); + + // Add "All Mail" folder + if !filter_subscribed && matches_pattern(&patterns, &self.imap.name_all) { + list_items.push(ListItem { + mailbox_name: self.imap.name_all.clone(), + attributes: vec![Attribute::All, Attribute::NoInferiors], + tags: vec![], + }); + } + + // Add mailboxes + let mut added_shared_folder = false; + for account in self.mailboxes.lock().iter() { + if let Some(prefix) = &account.prefix { + if !added_shared_folder { + if !filter_subscribed && matches_pattern(&patterns, &self.imap.name_shared) { + list_items.push(ListItem { + mailbox_name: self.imap.name_shared.clone(), + attributes: if include_children { + vec![Attribute::HasChildren, Attribute::NoSelect] + } else { + vec![Attribute::NoSelect] + }, + tags: vec![], + }); + } + added_shared_folder = true; + } + if !filter_subscribed && matches_pattern(&patterns, prefix) { + list_items.push(ListItem { + mailbox_name: prefix.clone(), + attributes: if include_children { + vec![Attribute::HasChildren, Attribute::NoSelect] + } else { + vec![Attribute::NoSelect] + }, + tags: vec![], + }); + } + } + + for (mailbox_name, mailbox_id) in &account.mailbox_names { + if matches_pattern(&patterns, mailbox_name) { + let mailbox = account.mailbox_data.get(mailbox_id).unwrap(); + let mut has_recursive_match = false; + if recursive_match { + let prefix = format!("{}/", mailbox_name); + for (mailbox_name, mailbox_id) in &account.mailbox_names { + if mailbox_name.starts_with(&prefix) + && account.mailbox_data.get(mailbox_id).unwrap().is_subscribed + { + has_recursive_match = true; + break; + } + } + } + if !filter_subscribed || mailbox.is_subscribed || has_recursive_match { + let mut attributes = Vec::with_capacity(2); + if include_children { + attributes.push(if mailbox.has_children { + Attribute::HasChildren + } else { + Attribute::HasNoChildren + }); + } + if include_subscribed && mailbox.is_subscribed { + attributes.push(Attribute::Subscribed); + } + if include_special_use { + if let Some(special_use) = &mailbox.special_use { + attributes.push(special_use.clone()); + } else if filter_special_use { + continue; + } + } + list_items.push(ListItem { + mailbox_name: mailbox_name.clone(), + attributes, + tags: if !has_recursive_match { + vec![] + } else { + vec![Tag::ChildInfo(vec![ChildInfo::Subscribed])] + }, + }); + } + } + } + } + + // Add status response + let mut status_items = Vec::new(); + if let Some(include_status) = include_status { + for list_item in &list_items { + match self + .status(list_item.mailbox_name.to_string(), include_status) + .await + { + Ok(status) => { + status_items.push(status); + } + Err(_) => { + tracing::debug!(parent: &self.span, "Failed to get mailbox status."); + } + } + } + } + + // Write response + self.write_bytes( + StatusResponse::completed(if !is_lsub { + Command::List + } else { + Command::Lsub + }) + .with_tag(tag) + .serialize( + list::Response { + is_rev2: version.is_rev2(), + is_lsub, + list_items, + status_items, + } + .serialize(), + ), + ) + .await; + } +} + +#[allow(clippy::while_let_on_iterator)] +fn matches_pattern(patterns: &[String], mailbox_name: &str) -> bool { + if patterns.is_empty() { + return true; + } + + 'outer: for pattern in patterns { + let mut pattern_bytes = pattern.as_bytes().iter().enumerate().peekable(); + let mut mailbox_name = mailbox_name.as_bytes().iter().peekable(); + + 'inner: while let Some((pos, &ch)) = pattern_bytes.next() { + if ch == b'%' || ch == b'*' { + let mut end_pos = pos; + while let Some((_, &next_ch)) = pattern_bytes.peek() { + if next_ch == b'%' || next_ch == b'*' { + break; + } else { + end_pos = pattern_bytes.next().unwrap().0; + } + } + if end_pos > pos { + let match_bytes = &pattern.as_bytes()[pos + 1..end_pos + 1]; + let mut match_count = 0; + let pattern_eof = end_pos == pattern.len() - 1; + + loop { + match mailbox_name.next() { + Some(&ch) => { + if match_bytes[match_count] == ch { + match_count += 1; + if match_count == match_bytes.len() { + if !pattern_eof { + continue 'inner; + } else if mailbox_name.peek().is_none() { + return true; + } else { + // Match needs to be at the end of the string, + // reset counter. + match_count = 0; + } + } + } else if match_count > 0 { + match_count = 0; + } + } + None => continue 'outer, + } + } + } else if ch == b'*' || !mailbox_name.any(|&ch| ch == b'/') { + return true; + } else { + continue 'outer; + } + } else { + match mailbox_name.next() { + Some(&mch) if mch == ch => (), + _ => continue 'outer, + } + } + } + + if mailbox_name.next().is_none() { + return true; + } + } + + false +} + +#[cfg(test)] +mod tests { + + #[test] + fn matches_pattern() { + let mailboxes = [ + "imaptest", + "imaptest/test", + "imaptest/test2", + "imaptest/test3", + "imaptest/test3/test4", + "imaptest/test3/test4/test5", + "foobar/test", + "foobar/test/test", + "foobar/test1/test1", + ]; + + for (pattern, expected_match) in [ + ( + "imaptest/%", + vec!["imaptest/test", "imaptest/test2", "imaptest/test3"], + ), + ("imaptest/%/%", vec!["imaptest/test3/test4"]), + ( + "imaptest/*", + vec![ + "imaptest/test", + "imaptest/test2", + "imaptest/test3", + "imaptest/test3/test4", + "imaptest/test3/test4/test5", + ], + ), + ("imaptest/*test4", vec!["imaptest/test3/test4"]), + ( + "imaptest/*test*", + vec![ + "imaptest/test", + "imaptest/test2", + "imaptest/test3", + "imaptest/test3/test4", + "imaptest/test3/test4/test5", + ], + ), + ("imaptest/%3/%", vec!["imaptest/test3/test4"]), + ("imaptest/%3/%4", vec!["imaptest/test3/test4"]), + ("imaptest/%t*4", vec!["imaptest/test3/test4"]), + ("*st/%3/%4/%5", vec!["imaptest/test3/test4/test5"]), + ( + "*%*%*%", + vec![ + "imaptest", + "imaptest/test", + "imaptest/test2", + "imaptest/test3", + "imaptest/test3/test4", + "imaptest/test3/test4/test5", + "foobar/test", + "foobar/test/test", + "foobar/test1/test1", + ], + ), + ("foobar*test", vec!["foobar/test", "foobar/test/test"]), + ] { + let patterns = vec![pattern.to_string()]; + let mut matched_mailboxes = Vec::new(); + for mailbox in mailboxes { + if super::matches_pattern(&patterns, mailbox) { + matched_mailboxes.push(mailbox); + } + } + assert_eq!(matched_mailboxes, expected_match, "for pattern {}", pattern); + } + } +} diff --git a/crates/imap/src/op/mod.rs b/crates/imap/src/op/mod.rs new file mode 100644 index 00000000..d9ce4e7c --- /dev/null +++ b/crates/imap/src/op/mod.rs @@ -0,0 +1,7 @@ +use imap_proto::StatusResponse; + +pub mod authenticate; +pub mod list; +pub mod status; + +pub type Result = std::result::Result; diff --git a/crates/imap/src/op/status.rs b/crates/imap/src/op/status.rs new file mode 100644 index 00000000..8085346a --- /dev/null +++ b/crates/imap/src/op/status.rs @@ -0,0 +1,373 @@ +/* + * Copyright (c) 2020-2022, Stalwart Labs Ltd. + * + * This file is part of the Stalwart IMAP Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::sync::Arc; + +use ahash::AHashSet; +use imap_proto::{ + protocol::status::{Status, StatusItem, StatusItemType}, + receiver::Request, + Command, ResponseCode, StatusResponse, +}; +use jmap_proto::types::{collection::Collection, id::Id, keyword::Keyword, property::Property}; +use store::{roaring::RoaringBitmap}; +use tokio::io::AsyncRead; +use store::Deserialize; + +use crate::core::{Mailbox, Session, SessionData}; + +impl Session { + pub async fn handle_status(&mut self, request: Request) -> crate::OpResult { + match request.parse_status(self.version) { + Ok(arguments) => { + let version = self.version; + let data = self.state.session_data(); + tokio::spawn(async move { + // Refresh mailboxes + if !data.try_synchronize_mailboxes(&arguments.tag).await { + return; + } + + // Fetch status + match data.status(arguments.mailbox_name, &arguments.items).await { + Ok(status) => { + let mut buf = Vec::with_capacity(32); + status.serialize(&mut buf, version.is_rev2()); + data.write_bytes( + StatusResponse::completed(Command::Status) + .with_tag(arguments.tag) + .serialize(buf), + ) + .await; + } + Err(mut response) => { + response.tag = arguments.tag.into(); + data.write_bytes(response.into_bytes()).await; + } + } + }); + Ok(()) + } + Err(response) => self.write_bytes(response.into_bytes()).await, + } + } +} + +impl SessionData { + pub async fn status( + &self, + mailbox_name: String, + items: &[Status], + ) -> super::Result { + // Get mailbox id + let mailbox = if let Some(mailbox) = self.get_mailbox_by_name(&mailbox_name) { + Arc::new(mailbox) + } else { + return Err( + StatusResponse::no("Mailbox does not exist.").with_code(ResponseCode::NonExistent) + ); + }; + + // Make sure all requested fields are up to date + let mut items_update = AHashSet::with_capacity(items.len()); + let mut items_response = Vec::with_capacity(items.len()); + + for account in self.mailboxes.lock().iter_mut() { + if account.account_id == mailbox.account_id { + let mailbox_data = account + .mailbox_data + .entry(mailbox.mailbox_id.as_ref().cloned().unwrap_or_default()) + .or_insert_with(Mailbox::default); + for item in items { + match item { + Status::Messages => { + if let Some(value) = mailbox_data.total_messages { + items_response.push((*item, StatusItemType::Number(value))); + } else { + items_update.insert(*item); + } + } + Status::UidNext => { + if let Some(value) = mailbox_data.uid_next { + items_response.push((*item, StatusItemType::Number(value))); + } else { + items_update.insert(*item); + } + } + Status::UidValidity => { + if let Some(value) = mailbox_data.uid_validity { + items_response.push((*item, StatusItemType::Number(value))); + } else { + items_update.insert(*item); + } + } + Status::Unseen => { + if let Some(value) = mailbox_data.total_unseen { + items_response.push((*item, StatusItemType::Number(value))); + } else { + items_update.insert(*item); + } + } + Status::Deleted => { + if let Some(value) = mailbox_data.total_deleted { + items_response.push((*item, StatusItemType::Number(value))); + } else { + items_update.insert(*item); + } + } + Status::Size => { + if let Some(value) = mailbox_data.size { + items_response.push((*item, StatusItemType::Number(value))); + } else { + items_update.insert(*item); + } + } + Status::HighestModSeq => { + items_response.push(( + *item, + StatusItemType::Number(account.state.unwrap_or_default() as u32), + )); + } + Status::MailboxId => { + items_response.push(( + *item, + StatusItemType::String( + Id::from_parts( + mailbox.account_id, + mailbox.mailbox_id.unwrap_or(u32::MAX), + ) + .to_string(), + ), + )); + } + Status::Recent => { + items_response.push((*item, StatusItemType::Number(0))); + } + } + } + break; + } + } + + if !items_update.is_empty() { + // Retrieve latest values + let mut values_update = Vec::with_capacity(items_update.len()); + + if let Some(mailbox_id) = mailbox.mailbox_id { + let mailbox_message_ids = self + .jmap + .get_tag( + mailbox.account_id, + Collection::Email, + Property::MailboxIds, + mailbox_id, + ) + .await? + .map(Arc::new); + let message_ids = self + .jmap + .get_document_ids(mailbox.account_id, Collection::Email) + .await?; + + for item in items_update { + let result = match item { + Status::Messages => { + message_ids.as_ref().map(|v| v.len()).unwrap_or(0) as u32 + } + Status::UidNext => todo!(), + Status::UidValidity => todo!(), + Status::Unseen => { + if let (Some(message_ids), Some(mailbox_message_ids), Some(mut seen)) = ( + &message_ids, + &mailbox_message_ids, + self.jmap + .get_tag( + mailbox.account_id, + Collection::Email, + Property::Keywords, + Keyword::Seen, + ) + .await?, + ) { + seen ^= message_ids; + seen &= mailbox_message_ids.as_ref(); + seen.len() as u32 + } else { + 0 + } + } + Status::Deleted => { + if let (Some(mailbox_message_ids), Some(mut deleted)) = ( + &mailbox_message_ids, + self.jmap + .get_tag( + mailbox.account_id, + Collection::Email, + Property::Keywords, + Keyword::Deleted, + ) + .await?, + ) { + deleted &= mailbox_message_ids.as_ref(); + deleted.len() as u32 + } else { + 0 + } + } + Status::Size => { + if let Some(mailbox_message_ids) = &mailbox_message_ids { + self.calculate_mailbox_size(mailbox.account_id, mailbox_message_ids) + .await? + } else { + 0 + } + } + Status::HighestModSeq | Status::MailboxId | Status::Recent => { + unreachable!() + } + }; + + items_response.push((item, StatusItemType::Number(result))); + values_update.push((item, result)); + } + } else { + let message_ids = Arc::new( + self.jmap + .get_document_ids(mailbox.account_id, Collection::Email) + .await? + .unwrap_or_default(), + ); + for item in items_update { + let result = match item { + Status::Messages => message_ids.len() as u32, + Status::UidNext => todo!(), + Status::UidValidity => todo!(), + Status::Unseen => self + .jmap + .get_tag( + mailbox.account_id, + Collection::Email, + Property::Keywords, + Keyword::Seen, + ) + .await? + .map(|mut seen| { + seen ^= message_ids.as_ref(); + seen.len() + }) + .unwrap_or(0) as u32, + Status::Deleted => self + .jmap + .get_tag( + mailbox.account_id, + Collection::Email, + Property::Keywords, + Keyword::Deleted, + ) + .await? + .map(|v| v.len()) + .unwrap_or(0) as u32, + Status::Size => { + if !message_ids.is_empty() { + self.calculate_mailbox_size(mailbox.account_id, &message_ids) + .await? + } else { + 0 + } + } + Status::HighestModSeq | Status::MailboxId | Status::Recent => { + unreachable!() + } + }; + + items_response.push((item, StatusItemType::Number(result))); + values_update.push((item, result)); + } + } + + // Update cache + for account in self.mailboxes.lock().iter_mut() { + if account.account_id == mailbox.account_id { + let mailbox_data = account + .mailbox_data + .entry(mailbox.mailbox_id.as_ref().cloned().unwrap_or_default()) + .or_insert_with(Mailbox::default); + + for (item, value) in values_update { + match item { + Status::Messages => mailbox_data.total_messages = value.into(), + Status::UidNext => mailbox_data.uid_next = value.into(), + Status::UidValidity => mailbox_data.uid_validity = value.into(), + Status::Unseen => mailbox_data.total_unseen = value.into(), + Status::Deleted => mailbox_data.total_deleted = value.into(), + Status::Size => mailbox_data.size = value.into(), + Status::HighestModSeq | Status::MailboxId | Status::Recent => { + unreachable!() + } + } + } + + break; + } + } + } + + // Generate response + Ok(StatusItem { + mailbox_name, + items: items_response, + }) + } + + async fn calculate_mailbox_size( + &self, + account_id: u32, + message_ids: &Arc, + ) -> super::Result { + self.jmap + .store + .index_values( + (message_ids.clone(), 0u32), + account_id, + Collection::Email, + Property::Size, + true, + |(message_ids, total_size), document_id, bytes| { + if message_ids.contains(document_id) { + u32::deserialize(bytes).map(|size| { + *total_size += size; + })?; + } + Ok(true) + }, + ) + .await + .map(|(_, size)| size ) + .map_err(|err| { + tracing::warn!(parent: &self.span, + event = "error", + reason = ?err, + "Failed to calculate mailbox size"); + StatusResponse::database_failure()}) + } +} diff --git a/crates/jmap/src/api/session.rs b/crates/jmap/src/api/session.rs index d323517b..a5a6266b 100644 --- a/crates/jmap/src/api/session.rs +++ b/crates/jmap/src/api/session.rs @@ -192,12 +192,7 @@ impl JMAP { session.add_account( (*id).into(), - self.directory - .principal_by_id(*id) - .await - .unwrap_or_default() - .map(|p| p.name) - .unwrap_or_else(|| Id::from(*id).to_string()), + self.get_account_name(*id).await, is_personal, is_readonly, Some(&[Capability::Core, Capability::Mail, Capability::WebSocket]), @@ -206,6 +201,15 @@ impl JMAP { Ok(session) } + + pub async fn get_account_name(&self, account_id: u32) -> String { + self.directory + .principal_by_id(account_id) + .await + .unwrap_or_default() + .map(|p| p.name) + .unwrap_or_else(|| Id::from(account_id).to_string()) + } } impl crate::Config { diff --git a/crates/jmap/src/auth/authenticate.rs b/crates/jmap/src/auth/authenticate.rs index 308f8209..4f05b1b0 100644 --- a/crates/jmap/src/auth/authenticate.rs +++ b/crates/jmap/src/auth/authenticate.rs @@ -50,20 +50,7 @@ impl JMAP { .and_then(|h| h.split_once(' ').map(|(l, t)| (l, t.trim().to_string()))) { let session = if let Some(account_id) = self.sessions.get(&token) { - if let Some(access_token) = self.access_tokens.get(&account_id) { - access_token.into() - } else { - // Refresh ACL token - self.get_access_token(account_id).await.map(|access_token| { - let access_token = Arc::new(access_token); - self.access_tokens.insert( - account_id, - access_token.clone(), - Instant::now() + self.config.session_cache_ttl, - ); - access_token - }) - } + self.get_cached_access_token(account_id).await } else { let addr = self.build_remote_addr(req, remote_ip); if mechanism.eq_ignore_ascii_case("basic") { @@ -108,19 +95,11 @@ impl JMAP { self.is_anonymous_allowed(addr)?; None } - .map(|session| { - let session = Arc::new(session); - self.sessions.insert( - token, - session.primary_id(), - Instant::now() + self.config.session_cache_ttl, - ); - self.access_tokens.insert( - session.primary_id(), - session.clone(), - Instant::now() + self.config.session_cache_ttl, - ); - session + .map(|access_token| { + let access_token = Arc::new(access_token); + self.cache_session(token, &access_token); + self.cache_access_token(access_token.clone()); + access_token }) }; @@ -138,6 +117,35 @@ impl JMAP { } } + pub fn cache_session(&self, session_id: String, access_token: &AccessToken) { + self.sessions.insert( + session_id, + access_token.primary_id(), + Instant::now() + self.config.session_cache_ttl, + ); + } + + pub fn cache_access_token(&self, access_token: Arc) { + self.access_tokens.insert( + access_token.primary_id(), + access_token, + Instant::now() + self.config.session_cache_ttl, + ); + } + + pub async fn get_cached_access_token(&self, primary_id: u32) -> Option> { + if let Some(access_token) = self.access_tokens.get(&primary_id) { + access_token.into() + } else { + // Refresh ACL token + self.get_access_token(primary_id).await.map(|access_token| { + let access_token = Arc::new(access_token); + self.cache_access_token(access_token.clone()); + access_token + }) + } + } + pub fn build_remote_addr( &self, req: &hyper::Request, diff --git a/crates/jmap/src/auth/mod.rs b/crates/jmap/src/auth/mod.rs index 8244adfc..c88a1ae3 100644 --- a/crates/jmap/src/auth/mod.rs +++ b/crates/jmap/src/auth/mod.rs @@ -101,6 +101,10 @@ impl AccessToken { || self.member_of.contains(&SUPERUSER_ID) } + pub fn is_primary_id(&self, account_id: u32) -> bool { + self.primary_id == account_id + } + pub fn is_super_user(&self) -> bool { self.primary_id == SUPERUSER_ID || self.member_of.contains(&SUPERUSER_ID) } @@ -109,6 +113,19 @@ impl AccessToken { !self.is_member(account_id) && self.access_to.iter().any(|(id, _)| *id == account_id) } + pub fn shared_accounts(&self, collection: impl Into) -> impl Iterator { + let collection = collection.into(); + self.member_of + .iter() + .chain(self.access_to.iter().filter_map(move |(id, cols)| { + if cols.contains(collection) { + id.into() + } else { + None + } + })) + } + pub fn has_access(&self, to_account_id: u32, to_collection: impl Into) -> bool { let to_collection = to_collection.into(); self.is_member(to_account_id) diff --git a/crates/jmap/src/changes/get.rs b/crates/jmap/src/changes/get.rs index b09aabc0..dbdd80ba 100644 --- a/crates/jmap/src/changes/get.rs +++ b/crates/jmap/src/changes/get.rs @@ -85,10 +85,7 @@ impl JMAP { let (items_sent, mut changelog) = match &request.since_state { State::Initial => { - let changelog = self - .changes_(account_id, collection, Query::All) - .await? - .unwrap(); + let changelog = self.changes_(account_id, collection, Query::All).await?; if changelog.changes.is_empty() && changelog.from_change_id == 0 { return Ok(response); } @@ -98,12 +95,7 @@ impl JMAP { State::Exact(change_id) => ( 0, self.changes_(account_id, collection, Query::Since(*change_id)) - .await? - .ok_or_else(|| { - MethodError::InvalidArguments( - "The specified stateId does could not be found.".to_string(), - ) - })?, + .await?, ), State::Intermediate(intermediate_state) => { let mut changelog = self @@ -112,12 +104,7 @@ impl JMAP { collection, Query::RangeInclusive(intermediate_state.from_id, intermediate_state.to_id), ) - .await? - .ok_or_else(|| { - MethodError::InvalidArguments( - "The specified stateId does could not be found.".to_string(), - ) - })?; + .await?; if intermediate_state.items_sent >= changelog.changes.len() { ( 0, @@ -126,12 +113,7 @@ impl JMAP { collection, Query::Since(intermediate_state.to_id), ) - .await? - .ok_or_else(|| { - MethodError::InvalidArguments( - "The specified stateId does could not be found.".to_string(), - ) - })?, + .await?, ) } else { changelog.changes.drain( @@ -189,12 +171,12 @@ impl JMAP { Ok(response) } - async fn changes_( + pub async fn changes_( &self, account_id: u32, collection: Collection, query: Query, - ) -> Result, MethodError> { + ) -> Result { self.store .changes(account_id, collection, query) .await diff --git a/crates/jmap/src/mailbox/get.rs b/crates/jmap/src/mailbox/get.rs index 885cd8ca..d567ebe0 100644 --- a/crates/jmap/src/mailbox/get.rs +++ b/crates/jmap/src/mailbox/get.rs @@ -279,7 +279,7 @@ impl JMAP { } } - async fn mailbox_unread_tags( + pub async fn mailbox_unread_tags( &self, account_id: u32, document_id: u32, diff --git a/crates/jmap/src/sieve/ingest.rs b/crates/jmap/src/sieve/ingest.rs index a1ff03c2..511ef236 100644 --- a/crates/jmap/src/sieve/ingest.rs +++ b/crates/jmap/src/sieve/ingest.rs @@ -495,7 +495,14 @@ impl JMAP { #[inline(always)] pub fn is_valid_role(role: &str) -> bool { [ - "inbox", "trash", "spam", "junk", "drafts", "archive", "sent", + "inbox", + "trash", + "spam", + "junk", + "drafts", + "archive", + "sent", + "important", ] .contains(&role) } diff --git a/crates/store/src/query/get.rs b/crates/store/src/query/get.rs index b90ef2e4..8b7f2646 100644 --- a/crates/store/src/query/get.rs +++ b/crates/store/src/query/get.rs @@ -150,4 +150,47 @@ impl Store { .await } } + + pub async fn index_values( + &self, + mut acc: T, + account_id: u32, + collection: impl Into, + field: impl Into, + ascending: bool, + cb: impl Fn(&mut T, u32, &[u8]) -> crate::Result + Sync + Send + 'static, + ) -> crate::Result { + let collection = collection.into(); + let field = field.into(); + #[cfg(not(feature = "is_sync"))] + { + self.read_transaction() + .await? + .sort_index( + account_id, + collection, + field, + ascending, + |value, document_id| cb(&mut acc, document_id, value).unwrap_or(false), + ) + .await + .map(|_| acc) + } + + #[cfg(feature = "is_sync")] + { + let trx = self.read_transaction()?; + self.spawn_worker(move || { + trx.sort_index( + account_id, + collection, + field, + ascending, + |value, document_id| cb(&mut acc, document_id, value).unwrap_or(false), + ) + .map(|_| acc) + }) + .await + } + } } diff --git a/crates/store/src/query/log.rs b/crates/store/src/query/log.rs index ea2776d9..12405dd1 100644 --- a/crates/store/src/query/log.rs +++ b/crates/store/src/query/log.rs @@ -64,7 +64,7 @@ impl Store { account_id: u32, collection: impl Into, query: Query, - ) -> crate::Result> { + ) -> crate::Result { let collection = collection.into(); let (is_inclusive, from_change_id, to_change_id) = match query { Query::All => (true, 0, u64::MAX), @@ -121,7 +121,7 @@ impl Store { }; } - Ok(Some(changelog)) + Ok(changelog) } }