filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

IMAP mailbox synchronization

Browse Source
This commit is contained in:
Mauro D
2023-06-22 17:29:39 +00:00
parent 5fda550642
commit e1c3190b48
23 changed files with 1849 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
Cargo.lock generated
View File

@@ -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",
]

1
crates/imap-proto/Cargo.toml
View File

@@ -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"}

11
crates/imap-proto/src/lib.rs
View File

@@ -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<String>) -> Self {
self.tag = Some(tag.into());
self
}
@@ -224,4 +225,10 @@ impl StatusResponse {
}
}
impl From<MethodError> for StatusResponse {
fn from(_: MethodError) -> Self {
StatusResponse::database_failure()
}
}
pub type Result<T> = std::result::Result<T, StatusResponse>;

2
crates/imap-proto/src/protocol/status.rs
View File

@@ -32,7 +32,7 @@ pub struct Arguments {
pub items: Vec<Status>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Status {
Messages,
UidNext,

5
crates/imap/Cargo.toml
View File

@@ -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" }

2
crates/imap/src/core/client.rs
View File

@@ -32,7 +32,7 @@ use tokio::io::AsyncRead;
use super::{SelectedMailbox, Session, SessionData, State};
impl<T: AsyncRead> Session<T> {
pub async fn ingest(&mut self, bytes: &[u8]) -> Result<bool, ()> {
pub async fn ingest(&mut self, bytes: &[u8]) -> crate::Result<bool> {
/*let tmp = "dd";
for line in String::from_utf8_lossy(bytes).split("\r\n") {
println!("<- {:?}", &line[..std::cmp::min(line.len(), 100)]);

527
crates/imap/src/core/mailbox.rs Normal file
View File

@@ -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<T: AsyncRead>(
session: &Session<T>,
access_token: &AccessToken,
) -> crate::Result<Self> {
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<String>,
access_token: &AccessToken,
) -> crate::Result<Account> {
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::<Object<Value>>(
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<Option<MailboxSync>> {
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::<Vec<_>>();
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::<Vec<_>>();
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<MailboxId> {
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
}
}

56
crates/imap/src/core/mod.rs
View File

@@ -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<T: AsyncRead> {
pub is_qresync: bool,
pub writer: mpsc::Sender<writer::Event>,
pub stream_rx: ReadHalf<T>,
pub in_flight: Vec<InFlight>,
pub in_flight: InFlight,
pub remote_addr: RemoteAddress,
pub span: tracing::Span,
}
pub struct SessionData {
pub core: Arc<JMAP>,
pub account_id: u32,
pub jmap: Arc<JMAP>,
pub imap: Arc<IMAP>,
pub span: tracing::Span,
pub mailboxes: parking_lot::Mutex<Vec<Account>>,
pub writer: mpsc::Sender<writer::Event>,
pub access_token: Arc<AccessToken>,
pub state: AtomicU32,
}
#[derive(Debug, Default)]
pub struct Mailbox {
pub has_children: bool,
pub is_subscribed: bool,
pub special_use: Option<Attribute>,
pub total_messages: Option<u32>,
pub total_unseen: Option<u32>,
pub total_deleted: Option<u32>,
pub uid_validity: Option<u32>,
pub uid_next: Option<u32>,
pub size: Option<u32>,
}
#[derive(Debug)]
pub struct Account {
pub account_id: u32,
pub prefix: Option<String>,
pub mailbox_names: BTreeMap<String, u32>,
pub mailbox_data: AHashMap<u32, Mailbox>,
pub state: Option<u64>,
}
pub struct SelectedMailbox {
@@ -89,6 +119,13 @@ pub struct MailboxData {
pub last_state: u32,
}
#[derive(Debug, Default)]
pub struct MailboxSync {
pub added: Vec<String>,
pub changed: Vec<String>,
pub deleted: Vec<String>,
}
pub enum SavedSearch {
InFlight {
rx: watch::Receiver<Arc<Vec<ImapId>>>,
@@ -107,7 +144,6 @@ pub struct ImapId {
pub enum State {
NotAuthenticated {
remote_addr: RemoteAddress,
auth_failures: u32,
},
Authenticated {

19
crates/imap/src/core/session.rs
View File

@@ -87,7 +87,7 @@ impl<T: AsyncRead> Session<T> {
impl Session<TcpStream> {
pub async fn new(
mut session: utils::listener::SessionData<TcpStream>,
mut session: SessionData<TcpStream>,
manager: ImapSessionManager,
) -> Result<Session<TcpStream>, ()> {
// Write plain text greeting
@@ -102,10 +102,7 @@ impl Session<TcpStream> {
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<TcpStream> {
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<TcpStream> {
writer: self.writer,
span: self.span,
in_flight: self.in_flight,
remote_addr: self.remote_addr,
stream_rx,
})
}
@@ -191,10 +190,7 @@ impl Session<TlsStream<TcpStream>> {
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<TlsStream<TcpStream>> {
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,
})
}

2
crates/imap/src/core/writer.rs
View File

@@ -109,7 +109,7 @@ pub fn spawn_writer(mut stream: Event) -> mpsc::Sender<Event> {
}
impl<T: AsyncRead> Session<T> {
pub async fn write_bytes(&self, bytes: impl Into<Cow<'static, [u8]>>) -> Result<(), ()> {
pub async fn write_bytes(&self, bytes: impl Into<Cow<'static, [u8]>>) -> crate::OpResult {
/*let tmp = "dd";
println!(
"-> {:?}",

13
crates/imap/src/lib.rs
View File

@@ -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<Arc<Self>, String> {
pub async fn init(config: &Config) -> utils::config::Result<Arc<Self>> {
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<T> = std::result::Result<T, ()>;
pub type OpResult = std::result::Result<(), ()>;

286
crates/imap/src/op/authenticate.rs Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>.
*
* 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<T: AsyncRead> Session<T> {
pub async fn handle_authenticate(&mut self, request: Request<Command>) -> 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<String>,
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<Command>) -> 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<Credentials<String>, &'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<Credentials<String>, &'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()
);
}
}

447
crates/imap/src/op/list.rs Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>.
*
* 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<T: AsyncRead> Session<T> {
pub async fn handle_list(&mut self, request: Request<Command>) -> 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);
}
}
}

7
crates/imap/src/op/mod.rs Normal file
View File

@@ -0,0 +1,7 @@
use imap_proto::StatusResponse;
pub mod authenticate;
pub mod list;
pub mod status;
pub type Result<T> = std::result::Result<T, StatusResponse>;

373
crates/imap/src/op/status.rs Normal file
View File

@@ -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 <http://www.gnu.org/licenses/>.
*
* 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<T: AsyncRead> Session<T> {
pub async fn handle_status(&mut self, request: Request<Command>) -> 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<StatusItem> {
// 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<RoaringBitmap>,
) -> super::Result<u32> {
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()})
}
}

16
crates/jmap/src/api/session.rs
View File

@@ -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 {

62
crates/jmap/src/auth/authenticate.rs
View File

@@ -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<AccessToken>) {
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<Arc<AccessToken>> {
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<hyper::body::Incoming>,

17
crates/jmap/src/auth/mod.rs
View File

@@ -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<Collection>) -> impl Iterator<Item = &u32> {
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<Collection>) -> bool {
let to_collection = to_collection.into();
self.is_member(to_account_id)

30
crates/jmap/src/changes/get.rs
View File

@@ -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<Option<Changes>, MethodError> {
) -> Result<Changes, MethodError> {
self.store
.changes(account_id, collection, query)
.await

2
crates/jmap/src/mailbox/get.rs
View File

@@ -279,7 +279,7 @@ impl JMAP {
}
}
async fn mailbox_unread_tags(
pub async fn mailbox_unread_tags(
&self,
account_id: u32,
document_id: u32,

9
crates/jmap/src/sieve/ingest.rs
View File

@@ -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)
}

43
crates/store/src/query/get.rs
View File

@@ -150,4 +150,47 @@ impl Store {
.await
}
}
pub async fn index_values<T: Sync + Send + 'static>(
&self,
mut acc: T,
account_id: u32,
collection: impl Into<u8>,
field: impl Into<u8>,
ascending: bool,
cb: impl Fn(&mut T, u32, &[u8]) -> crate::Result<bool> + Sync + Send + 'static,
) -> crate::Result<T> {
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
}
}
}

4
crates/store/src/query/log.rs
View File

@@ -64,7 +64,7 @@ impl Store {
account_id: u32,
collection: impl Into<u8>,
query: Query,
) -> crate::Result<Option<Changes>> {
) -> crate::Result<Changes> {
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)
}
}