From 4d44e2fa770d81768d078b373dc2e6faf3598b44 Mon Sep 17 00:00:00 2001 From: Mauro D Date: Mon, 15 May 2023 15:21:09 +0000 Subject: [PATCH] PushSubscription and EventSource tests passing. --- crates/jmap-proto/src/method/copy.rs | 5 +- crates/jmap-proto/src/method/get.rs | 3 +- crates/jmap-proto/src/method/import.rs | 17 +- crates/jmap-proto/src/method/set.rs | 76 +++- crates/jmap-proto/src/object/mod.rs | 15 +- crates/jmap-proto/src/response/mod.rs | 30 +- crates/jmap-proto/src/types/collection.rs | 17 + crates/jmap-proto/src/types/id.rs | 14 + crates/jmap-proto/src/types/property.rs | 2 +- crates/jmap-proto/src/types/state.rs | 42 ++- crates/jmap-proto/src/types/type_state.rs | 74 ++++ crates/jmap-proto/src/types/value.rs | 15 +- crates/jmap/Cargo.toml | 11 + crates/jmap/src/api/config.rs | 3 + crates/jmap/src/api/event_source.rs | 158 ++++++++ crates/jmap/src/api/http.rs | 2 +- crates/jmap/src/api/mod.rs | 38 +- crates/jmap/src/api/request.rs | 48 ++- crates/jmap/src/email/copy.rs | 15 + crates/jmap/src/email/get.rs | 2 +- crates/jmap/src/email/import.rs | 54 ++- crates/jmap/src/email/set.rs | 22 +- crates/jmap/src/lib.rs | 83 +++-- crates/jmap/src/mailbox/get.rs | 5 +- crates/jmap/src/mailbox/set.rs | 35 +- crates/jmap/src/principal/mod.rs | 15 - crates/jmap/src/principal/set.rs | 266 ------------- crates/jmap/src/push/ece.rs | 211 +++++++++++ crates/jmap/src/push/get.rs | 233 ++++++++++++ crates/jmap/src/push/manager.rs | 316 ++++++++++++++++ crates/jmap/src/push/mod.rs | 83 +++++ crates/jmap/src/push/set.rs | 271 ++++++++++++++ crates/jmap/src/services/mod.rs | 3 + crates/jmap/src/services/state.rs | 434 ++++++++++++++++++++++ crates/jmap/src/thread/get.rs | 2 +- tests/Cargo.toml | 7 +- tests/src/jmap/auth_acl.rs | 231 +++++------- tests/src/jmap/auth_limits.rs | 37 +- tests/src/jmap/auth_oauth.rs | 31 +- tests/src/jmap/event_source.rs | 135 +++++++ tests/src/jmap/mailbox.rs | 18 +- tests/src/jmap/mod.rs | 40 +- tests/src/jmap/push_subscription.rs | 347 +++++++++++++++++ 43 files changed, 2849 insertions(+), 617 deletions(-) create mode 100644 crates/jmap/src/api/event_source.rs delete mode 100644 crates/jmap/src/principal/mod.rs delete mode 100644 crates/jmap/src/principal/set.rs create mode 100644 crates/jmap/src/push/ece.rs create mode 100644 crates/jmap/src/push/get.rs create mode 100644 crates/jmap/src/push/manager.rs create mode 100644 crates/jmap/src/push/mod.rs create mode 100644 crates/jmap/src/push/set.rs create mode 100644 crates/jmap/src/services/mod.rs create mode 100644 crates/jmap/src/services/state.rs create mode 100644 tests/src/jmap/event_source.rs create mode 100644 tests/src/jmap/push_subscription.rs diff --git a/crates/jmap-proto/src/method/copy.rs b/crates/jmap-proto/src/method/copy.rs index e3dd0657..3434d140 100644 --- a/crates/jmap-proto/src/method/copy.rs +++ b/crates/jmap-proto/src/method/copy.rs @@ -9,7 +9,7 @@ use crate::{ types::{ blob::BlobId, id::Id, - state::State, + state::{State, StateChange}, value::{SetValue, Value}, }, }; @@ -47,6 +47,9 @@ pub struct CopyResponse { #[serde(rename = "notCreated")] #[serde(skip_serializing_if = "VecMap::is_empty")] pub not_created: VecMap, + + #[serde(skip)] + pub state_change: Option, } #[derive(Debug, Clone)] diff --git a/crates/jmap-proto/src/method/get.rs b/crates/jmap-proto/src/method/get.rs index 20098956..2a65f328 100644 --- a/crates/jmap-proto/src/method/get.rs +++ b/crates/jmap-proto/src/method/get.rs @@ -37,7 +37,8 @@ pub struct GetResponse { #[serde(skip_serializing_if = "Option::is_none")] pub account_id: Option, - pub state: State, + #[serde(skip_serializing_if = "Option::is_none")] + pub state: Option, pub list: Vec>, diff --git a/crates/jmap-proto/src/method/import.rs b/crates/jmap-proto/src/method/import.rs index 03661d4c..37f2542e 100644 --- a/crates/jmap-proto/src/method/import.rs +++ b/crates/jmap-proto/src/method/import.rs @@ -8,12 +8,14 @@ use crate::{ reference::{MaybeReference, ResultReference}, RequestProperty, }, + response::Response, types::{ blob::BlobId, date::UTCDate, id::Id, keyword::Keyword, - state::State, + property::Property, + state::{State, StateChange}, value::{SetValueMap, Value}, }, }; @@ -52,6 +54,9 @@ pub struct ImportEmailResponse { #[serde(rename = "notCreated")] #[serde(skip_serializing_if = "VecMap::is_empty")] pub not_created: VecMap, + + #[serde(skip)] + pub state_change: Option, } impl JsonObjectParser for ImportEmailRequest { @@ -139,3 +144,13 @@ impl JsonObjectParser for ImportEmail { Ok(request) } } + +impl ImportEmailResponse { + pub fn update_created_ids(&self, response: &mut Response) { + for (user_id, obj) in &self.created { + if let Some(id) = obj.get(&Property::Id).as_id() { + response.created_ids.insert(user_id.clone(), *id); + } + } + } +} diff --git a/crates/jmap-proto/src/method/set.rs b/crates/jmap-proto/src/method/set.rs index ce708a80..e160f99c 100644 --- a/crates/jmap-proto/src/method/set.rs +++ b/crates/jmap-proto/src/method/set.rs @@ -13,6 +13,7 @@ use crate::{ reference::{MaybeReference, ResultReference}, RequestProperty, RequestPropertyParser, }, + response::Response, types::{ acl::Acl, blob::BlobId, @@ -20,8 +21,7 @@ use crate::{ id::Id, keyword::Keyword, property::{HeaderForm, ObjectProperty, Property, SetProperty}, - state::State, - type_state::TypeState, + state::{State, StateChange}, value::{SetValue, SetValueMap, Value}, }, }; @@ -87,6 +87,9 @@ pub struct SetResponse { #[serde(rename = "notDestroyed")] #[serde(skip_serializing_if = "VecMap::is_empty")] pub not_destroyed: VecMap, + + #[serde(skip)] + pub state_change: Option, } impl JsonObjectParser for SetRequest { @@ -316,12 +319,11 @@ impl JsonObjectParser for Object { | Property::Sender | Property::SubParts | Property::To - | Property::UndoStatus => { - SetValue::Value(Value::parse::( - parser.next_token()?, - parser, - )?) - } + | Property::UndoStatus + | Property::Types => SetValue::Value(Value::parse::( + parser.next_token()?, + parser, + )?), Property::Members => SetValue::Value(Value::parse::( parser.next_token()?, parser, @@ -331,10 +333,7 @@ impl JsonObjectParser for Object { } else { Value::parse::(parser.next_token()?, parser) }?), - Property::Types => SetValue::Value(Value::parse::( - parser.next_token()?, - parser, - )?), + _ => { parser.skip_token(parser.depth_array, parser.depth_dict)?; SetValue::Value(Value::Null) @@ -427,6 +426,47 @@ impl SetRequest { } impl SetResponse { + pub fn from_request( + request: &SetRequest, + max_objects: usize, + ) -> Result { + let n_create = request.create.as_ref().map_or(0, |objs| objs.len()); + let n_update = request.update.as_ref().map_or(0, |objs| objs.len()); + let n_destroy = request.destroy.as_ref().map_or(0, |objs| { + if let MaybeReference::Value(ids) = objs { + ids.len() + } else { + 0 + } + }); + if n_create + n_update + n_destroy <= max_objects { + Ok(SetResponse { + account_id: if request.account_id.is_valid() { + request.account_id.into() + } else { + None + }, + new_state: None, + old_state: None, + created: AHashMap::with_capacity(n_create), + updated: VecMap::with_capacity(n_update), + destroyed: Vec::with_capacity(n_destroy), + not_created: VecMap::new(), + not_updated: VecMap::new(), + not_destroyed: VecMap::new(), + state_change: None, + }) + } else { + Err(MethodError::RequestTooLarge) + } + } + + pub fn with_state(mut self, state: State) -> Self { + self.old_state = Some(state.clone()); + self.new_state = Some(state); + self + } + pub fn created(&mut self, id: String, document_id: u32) { self.created.insert( id, @@ -451,4 +491,16 @@ impl SetResponse { .with_description("Invalid property or value.".to_string()), ); } + + pub fn update_created_ids(&self, response: &mut Response) { + for (user_id, obj) in &self.created { + if let Some(id) = obj.get(&Property::Id).as_id() { + response.created_ids.insert(user_id.clone(), *id); + } + } + } + + pub fn has_changes(&self) -> bool { + !self.created.is_empty() || !self.updated.is_empty() || !self.destroyed.is_empty() + } } diff --git a/crates/jmap-proto/src/object/mod.rs b/crates/jmap-proto/src/object/mod.rs index 94db26ad..a5ed393b 100644 --- a/crates/jmap-proto/src/object/mod.rs +++ b/crates/jmap-proto/src/object/mod.rs @@ -16,8 +16,7 @@ use utils::{ }; use crate::types::{ - blob::BlobId, date::UTCDate, id::Id, keyword::Keyword, property::Property, - type_state::TypeState, value::Value, + blob::BlobId, date::UTCDate, id::Id, keyword::Keyword, property::Property, value::Value, }; #[derive(Debug, Clone, Default, serde::Serialize, PartialEq, Eq)] @@ -96,10 +95,9 @@ const ID: u8 = 4; const DATE: u8 = 5; const BLOB_ID: u8 = 6; const KEYWORD: u8 = 7; -const TYPE_STATE: u8 = 8; -const LIST: u8 = 9; -const OBJECT: u8 = 10; -const NULL: u8 = 11; +const LIST: u8 = 8; +const OBJECT: u8 = 9; +const NULL: u8 = 10; impl Serialize for Value { fn serialize(self) -> Vec { @@ -189,10 +187,6 @@ impl SerializeInto for Value { buf.push(KEYWORD); v.serialize_into(buf); } - Value::TypeState(v) => { - buf.push(TYPE_STATE); - v.serialize_into(buf); - } Value::List(v) => { buf.push(LIST); buf.push_leb128(v.len()); @@ -224,7 +218,6 @@ impl DeserializeFrom for Value { ))), BLOB_ID => Some(Value::BlobId(BlobId::deserialize_from(bytes)?)), KEYWORD => Some(Value::Keyword(Keyword::deserialize_from(bytes)?)), - TYPE_STATE => Some(Value::TypeState(TypeState::deserialize_from(bytes)?)), LIST => { let len = bytes.next_leb128()?; let mut items = Vec::with_capacity(len); diff --git a/crates/jmap-proto/src/response/mod.rs b/crates/jmap-proto/src/response/mod.rs index acbcac19..bbbbe481 100644 --- a/crates/jmap-proto/src/response/mod.rs +++ b/crates/jmap-proto/src/response/mod.rs @@ -18,7 +18,7 @@ use crate::{ validate::ValidateSieveScriptResponse, }, request::{echo::Echo, method::MethodName, Call}, - types::{id::Id, property::Property}, + types::id::Id, }; use self::serialize::serialize_hex; @@ -69,29 +69,11 @@ impl Response { name: MethodName, method: impl Into, ) { - // Add created ids - let method = method.into(); - if !self.created_ids.is_empty() { - match &method { - ResponseMethod::Set(SetResponse { created, .. }) => { - for (user_id, obj) in created { - if let Some(id) = obj.get(&Property::Id).as_id() { - self.created_ids.insert(user_id.clone(), *id); - } - } - } - ResponseMethod::ImportEmail(ImportEmailResponse { created, .. }) => { - for (user_id, obj) in created { - if let Some(id) = obj.get(&Property::Id).as_id() { - self.created_ids.insert(user_id.clone(), *id); - } - } - } - _ => {} - } - } - - self.method_responses.push(Call { id, method, name }); + self.method_responses.push(Call { + id, + method: method.into(), + name, + }); } pub fn push_error(&mut self, id: String, err: MethodError) { diff --git a/crates/jmap-proto/src/types/collection.rs b/crates/jmap-proto/src/types/collection.rs index d9143282..c6f1d8e8 100644 --- a/crates/jmap-proto/src/types/collection.rs +++ b/crates/jmap-proto/src/types/collection.rs @@ -2,6 +2,8 @@ use std::fmt::{self, Display, Formatter}; use utils::map::bitmap::BitmapItem; +use super::type_state::TypeState; + #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)] #[repr(u8)] pub enum Collection { @@ -57,6 +59,21 @@ impl From for u64 { } } +impl TryFrom for TypeState { + type Error = (); + + fn try_from(value: Collection) -> Result { + match value { + Collection::Email => Ok(TypeState::Email), + Collection::Mailbox => Ok(TypeState::Mailbox), + Collection::Thread => Ok(TypeState::Thread), + Collection::Identity => Ok(TypeState::Identity), + Collection::EmailSubmission => Ok(TypeState::EmailSubmission), + _ => Err(()), + } + } +} + impl Display for Collection { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { match self { diff --git a/crates/jmap-proto/src/types/id.rs b/crates/jmap-proto/src/types/id.rs index 6507ec51..c96d7b1a 100644 --- a/crates/jmap-proto/src/types/id.rs +++ b/crates/jmap-proto/src/types/id.rs @@ -193,6 +193,10 @@ impl Id { pub fn is_singleton(&self) -> bool { self.id == 20080258862541 } + + pub fn is_valid(&self) -> bool { + self.id != u64::MAX + } } impl From for Id { @@ -260,6 +264,16 @@ impl serde::Serialize for Id { } } +impl<'de> serde::Deserialize<'de> for Id { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + Id::from_bytes(<&str>::deserialize(deserializer)?.as_bytes()) + .ok_or_else(|| serde::de::Error::custom("invalid JMAP ID")) + } +} + impl std::fmt::Display for Id { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.write_str(&self.as_string()) diff --git a/crates/jmap-proto/src/types/property.rs b/crates/jmap-proto/src/types/property.rs index cf7e7c18..484bd8a9 100644 --- a/crates/jmap-proto/src/types/property.rs +++ b/crates/jmap-proto/src/types/property.rs @@ -523,7 +523,7 @@ impl JsonObjectParser for ObjectProperty { let mut shift = 0; while let Some(ch) = parser.next_unescaped()? { - if ch.is_ascii_alphabetic() { + if ch.is_ascii_alphanumeric() { if first_char != 0 { if shift < 128 { hash |= (ch as u128) << shift; diff --git a/crates/jmap-proto/src/types/state.rs b/crates/jmap-proto/src/types/state.rs index 5efd6807..25e5f365 100644 --- a/crates/jmap-proto/src/types/state.rs +++ b/crates/jmap-proto/src/types/state.rs @@ -28,7 +28,7 @@ use utils::codec::{ use crate::parser::{base32::JsonBase32Reader, json::Parser, JsonObjectParser}; -use super::ChangeId; +use super::{type_state::TypeState, ChangeId}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct JMAPIntermediateState { @@ -45,6 +45,34 @@ pub enum State { Intermediate(JMAPIntermediateState), } +#[derive(Clone, Debug)] +pub struct StateChange { + pub account_id: u32, + pub types: Vec<(TypeState, u64)>, +} + +impl StateChange { + pub fn new(account_id: u32) -> Self { + Self { + account_id, + types: Vec::with_capacity(0), + } + } + + pub fn with_change(mut self, type_state: TypeState, change_id: u64) -> Self { + if let Some((_, last_change_id)) = self.types.iter_mut().find(|(ts, _)| ts == &type_state) { + *last_change_id = change_id; + } else { + self.types.push((type_state, change_id)); + } + self + } + + pub fn has_changes(&self) -> bool { + !self.types.is_empty() + } +} + impl From for State { fn from(change_id: ChangeId) -> Self { State::Exact(change_id) @@ -138,6 +166,18 @@ impl serde::Serialize for State { } } +impl<'de> serde::Deserialize<'de> for State { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + // This is inefficient, but serde deserialize on State is only used in test mode + let value = format!("{}\"", <&str>::deserialize(deserializer)?); + let mut parser = Parser::new(value.as_bytes()); + State::parse(&mut parser).map_err(|_| serde::de::Error::custom("invalid JMAP State")) + } +} + impl std::fmt::Display for State { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let mut writer = Base32Writer::with_capacity(10); diff --git a/crates/jmap-proto/src/types/type_state.rs b/crates/jmap-proto/src/types/type_state.rs index 5e3c2e61..e4e57d4c 100644 --- a/crates/jmap-proto/src/types/type_state.rs +++ b/crates/jmap-proto/src/types/type_state.rs @@ -2,6 +2,7 @@ use std::fmt::Display; use serde::Serialize; use store::write::{DeserializeFrom, SerializeInto}; +use utils::map::bitmap::BitmapItem; use crate::parser::{json::Parser, JsonObjectParser}; @@ -20,6 +21,40 @@ pub enum TypeState { Thread = 4, #[serde(rename = "Identity")] Identity = 5, + None = 6, +} + +impl BitmapItem for TypeState { + fn max() -> u64 { + TypeState::None as u64 + } + + fn is_valid(&self) -> bool { + !matches!(self, TypeState::None) + } +} + +impl From for TypeState { + fn from(value: u64) -> Self { + match value { + 0 => TypeState::Email, + 1 => TypeState::EmailDelivery, + 2 => TypeState::EmailSubmission, + 3 => TypeState::Mailbox, + 4 => TypeState::Thread, + 5 => TypeState::Identity, + _ => { + debug_assert!(false, "Invalid type_state value: {}", value); + TypeState::None + } + } + } +} + +impl From for u64 { + fn from(type_state: TypeState) -> u64 { + type_state as u64 + } } impl JsonObjectParser for TypeState { @@ -51,6 +86,34 @@ impl JsonObjectParser for TypeState { } } +impl TryFrom<&str> for TypeState { + type Error = (); + + fn try_from(value: &str) -> Result { + let mut hash = 0; + let mut shift = 0; + + for &ch in value.as_bytes() { + if shift < 128 { + hash |= (ch as u128) << shift; + shift += 8; + } else { + return Err(()); + } + } + + match hash { + 0x006c_6961_6d45 => Ok(TypeState::Email), + 0x0079_7265_7669_6c65_446c_6961_6d45 => Ok(TypeState::EmailDelivery), + 0x006e_6f69_7373_696d_6275_536c_6961_6d45 => Ok(TypeState::EmailSubmission), + 0x0078_6f62_6c69_614d => Ok(TypeState::Mailbox), + 0x6461_6572_6854 => Ok(TypeState::Thread), + 0x7974_6974_6e65_6449 => Ok(TypeState::Identity), + _ => Err(()), + } + } +} + impl TypeState { pub fn as_str(&self) -> &'static str { match self { @@ -60,6 +123,7 @@ impl TypeState { TypeState::Mailbox => "Mailbox", TypeState::Thread => "Thread", TypeState::Identity => "Identity", + TypeState::None => "", } } } @@ -89,3 +153,13 @@ impl DeserializeFrom for TypeState { } } } + +impl<'de> serde::Deserialize<'de> for TypeState { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + TypeState::try_from(<&str>::deserialize(deserializer)?) + .map_err(|_| serde::de::Error::custom("invalid JMAP type state")) + } +} diff --git a/crates/jmap-proto/src/types/value.rs b/crates/jmap-proto/src/types/value.rs index ed81912d..2f4a4765 100644 --- a/crates/jmap-proto/src/types/value.rs +++ b/crates/jmap-proto/src/types/value.rs @@ -16,7 +16,6 @@ use super::{ id::Id, keyword::Keyword, property::{HeaderForm, IntoProperty, ObjectProperty, Property}, - type_state::TypeState, }; #[derive(Debug, Default, Clone, PartialEq, Eq, Serialize)] @@ -29,7 +28,6 @@ pub enum Value { Date(UTCDate), BlobId(BlobId), Keyword(Keyword), - TypeState(TypeState), List(Vec), Object(Object), #[default] @@ -274,6 +272,13 @@ impl Value { } } + pub fn as_date(&self) -> Option<&UTCDate> { + match self { + Value::Date(d) => Some(d), + _ => None, + } + } + pub fn try_cast_uint(&self) -> Option { match self { Value::UnsignedInt(u) => Some(*u), @@ -322,12 +327,6 @@ impl IntoValue for UTCDate { } } -impl IntoValue for TypeState { - fn into_value(self) -> Value { - Value::TypeState(self) - } -} - impl From for Value { fn from(value: usize) -> Self { Value::UnsignedInt(value as u64) diff --git a/crates/jmap/Cargo.toml b/crates/jmap/Cargo.toml index 03d7f8a3..8597a673 100644 --- a/crates/jmap/Cargo.toml +++ b/crates/jmap/Cargo.toml @@ -19,11 +19,22 @@ http-body-util = "0.1.0-rc.2" form_urlencoded = "1.1.0" tracing = "0.1" tokio = { version = "1.23", features = ["rt"] } +aes-gcm = "0.10.1" aes-gcm-siv = "0.11.1" bincode = "1.3.3" form-data = { version = "0.4.2", features = ["sync"], default-features = false } mime = "0.3.17" sqlx = { git = "https://github.com/mdecimus/sqlx", features = [ "runtime-tokio-rustls", "postgres", "mysql", "sqlite" ] } +futures-util = "0.3.28" +async-stream = "0.3.5" +base64 = "0.21" +p256 = { version = "0.13", features = ["ecdh"] } +hkdf = "0.12.3" +sha2 = "0.10.1" +reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"]} + +[dev-dependencies] +ece = "2.2" [features] test_mode = [] diff --git a/crates/jmap/src/api/config.rs b/crates/jmap/src/api/config.rs index f32cd63c..b1c7c133 100644 --- a/crates/jmap/src/api/config.rs +++ b/crates/jmap/src/api/config.rs @@ -99,6 +99,9 @@ impl crate::Config { .property_or_static::("oauth.expiry.refresh-token-renew", "4d")? .as_secs(), oauth_max_auth_attempts: settings.property_or_static("oauth.max-auth-attempts", "3")?, + event_source_throttle: settings + .property_or_static("jmap.event-source.throttle", "1s")?, + push_max_total: settings.property_or_static("jmap.push.max-total", "100")?, }; config.add_capabilites(settings); Ok(config) diff --git a/crates/jmap/src/api/event_source.rs b/crates/jmap/src/api/event_source.rs new file mode 100644 index 00000000..22600c16 --- /dev/null +++ b/crates/jmap/src/api/event_source.rs @@ -0,0 +1,158 @@ +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; + +use http_body_util::{combinators::BoxBody, StreamBody}; +use hyper::{ + body::{Bytes, Frame}, + header, StatusCode, +}; +use jmap_proto::{error::request::RequestError, types::type_state::TypeState}; +use utils::map::bitmap::Bitmap; + +use crate::{auth::AclToken, JMAP, LONG_SLUMBER}; + +use super::{http::ToHttpResponse, HttpRequest, HttpResponse, StateChangeResponse}; + +struct Ping { + interval: Duration, + last_ping: Instant, + payload: Bytes, +} + +impl JMAP { + pub async fn handle_event_source( + &self, + req: &HttpRequest, + acl_token: Arc, + ) -> HttpResponse { + // Parse query + let mut ping = 0; + let mut types = Bitmap::default(); + let mut close_after_state = false; + + for (key, value) in form_urlencoded::parse(req.uri().query().unwrap_or_default().as_bytes()) + { + match key.as_ref() { + "types" => { + for type_state in value.split(',') { + if type_state == "*" { + types = Bitmap::all(); + break; + } else if let Ok(type_state) = TypeState::try_from(type_state) { + types.insert(type_state); + } else { + return RequestError::invalid_parameters().into_http_response(); + } + } + } + "closeafter" => match value.as_ref() { + "state" => { + close_after_state = true; + } + "no" => {} + _ => return RequestError::invalid_parameters().into_http_response(), + }, + "ping" => match value.parse::() { + Ok(value) => { + ping = value; + } + Err(_) => return RequestError::invalid_parameters().into_http_response(), + }, + _ => {} + } + } + + let mut ping = if ping > 0 { + #[cfg(not(feature = "test_mode"))] + let interval = std::cmp::max(ping, 30) * 1000; + #[cfg(feature = "test_mode")] + let interval = ping * 1000; + + Ping { + interval: Duration::from_millis(interval as u64), + last_ping: Instant::now() - Duration::from_millis(interval as u64), + payload: Bytes::from(format!( + "event: ping\ndata: {{\"interval\": {}}}\n\n", + interval + )), + } + .into() + } else { + None + }; + let mut response = StateChangeResponse::new(); + let throttle = self.config.event_source_throttle; + + // Register with state manager + let mut change_rx = if let Some(change_rx) = self + .subscribe_state_manager(acl_token.primary_id(), acl_token.primary_id(), types) + .await + { + change_rx + } else { + return RequestError::internal_server_error().into_http_response(); + }; + + hyper::Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "text/event-stream") + .header(header::CACHE_CONTROL, "no-store") + .body(BoxBody::new(StreamBody::new(async_stream::stream! { + let mut last_message = Instant::now() - throttle; + let mut timeout = + ping.as_ref().map(|p| p.interval).unwrap_or(LONG_SLUMBER); + + loop { + match tokio::time::timeout(timeout, change_rx.recv()).await { + Ok(Some(state_change)) => { + for (type_state, change_id) in state_change.types { + response + .changed + .get_mut_or_insert(state_change.account_id.into()) + .set(type_state, change_id.into()); + } + } + Ok(None) => { + tracing::debug!("Broadcast channel was closed."); + break; + } + Err(_) => (), + } + + timeout = if !response.changed.is_empty() { + let elapsed = last_message.elapsed(); + if elapsed >= throttle { + last_message = Instant::now(); + yield Ok(Frame::data(Bytes::from(format!( + "event: state\ndata: {}\n\n", + serde_json::to_string(&response).unwrap() + )))); + + if close_after_state { + break; + } + + response.changed.clear(); + ping.as_ref().map(|p| p.interval).unwrap_or(LONG_SLUMBER) + } else { + throttle - elapsed + } + } else if let Some(ping) = &mut ping { + let elapsed = ping.last_ping.elapsed(); + if elapsed >= ping.interval { + ping.last_ping = Instant::now(); + yield Ok(Frame::data(ping.payload.clone())); + ping.interval + } else { + ping.interval - elapsed + } + } else { + LONG_SLUMBER + }; + } + }))) + .unwrap() + } +} diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index ce30af32..30296cb8 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -122,7 +122,7 @@ impl JMAP { } } ("eventsource", &Method::GET) => { - todo!() + return self.handle_event_source(req, acl_token).await } ("ws", &Method::GET) => { todo!() diff --git a/crates/jmap/src/api/mod.rs b/crates/jmap/src/api/mod.rs index 6899ea48..36f9e8fd 100644 --- a/crates/jmap/src/api/mod.rs +++ b/crates/jmap/src/api/mod.rs @@ -1,11 +1,14 @@ use std::sync::Arc; use hyper::StatusCode; +use jmap_proto::types::{id::Id, state::State, type_state::TypeState}; use serde::Serialize; +use utils::map::vec_map::VecMap; use crate::JMAP; pub mod config; +pub mod event_source; pub mod http; pub mod request; pub mod session; @@ -15,11 +18,9 @@ pub struct SessionManager { pub inner: Arc, } -impl From for SessionManager { - fn from(jmap: JMAP) -> Self { - SessionManager { - inner: Arc::new(jmap), - } +impl From> for SessionManager { + fn from(inner: Arc) -> Self { + SessionManager { inner } } } @@ -36,3 +37,30 @@ pub struct HtmlResponse { pub type HttpRequest = hyper::Request; pub type HttpResponse = hyper::Response>; + +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub enum StateChangeType { + StateChange, +} + +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub struct StateChangeResponse { + #[serde(rename = "@type")] + pub type_: StateChangeType, + pub changed: VecMap>, +} + +impl StateChangeResponse { + pub fn new() -> Self { + Self { + type_: StateChangeType::StateChange, + changed: VecMap::new(), + } + } +} + +impl Default for StateChangeResponse { + fn default() -> Self { + Self::new() + } +} diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index 9151e93e..5d44d86c 100644 --- a/crates/jmap/src/api/request.rs +++ b/crates/jmap/src/api/request.rs @@ -2,7 +2,10 @@ use std::sync::Arc; use jmap_proto::{ error::{method::MethodError, request::RequestError}, - method::{get, query, set}, + method::{ + get, query, + set::{self}, + }, request::{method::MethodName, Call, Request, RequestMethod}, response::{Response, ResponseMethod}, types::collection::Collection, @@ -26,6 +29,7 @@ impl JMAP { request.created_ids.unwrap_or_default(), request.method_calls.len(), ); + let add_created_ids = !response.created_ids.is_empty(); for mut call in request.method_calls { // Resolve result and id references @@ -42,7 +46,39 @@ impl JMAP { .handle_method_call(call.method, &acl_token, &mut next_call) .await { - Ok(method_response) => { + Ok(mut method_response) => { + match &mut method_response { + ResponseMethod::Set(set_response) => { + // Add created ids + if add_created_ids { + set_response.update_created_ids(&mut response); + } + + // Publish state changes + if let Some(state_change) = set_response.state_change.take() { + self.broadcast_state_change(state_change).await; + } + } + ResponseMethod::ImportEmail(import_response) => { + // Add created ids + if add_created_ids { + import_response.update_created_ids(&mut response); + } + + // Publish state changes + if let Some(state_change) = import_response.state_change.take() { + self.broadcast_state_change(state_change).await; + } + } + ResponseMethod::Copy(copy_response) => { + // Publish state changes + if let Some(state_change) = copy_response.state_change.take() { + self.broadcast_state_change(state_change).await; + } + } + _ => {} + } + response.push_response(call.id, call.name, method_response); } Err(err) => { @@ -90,7 +126,9 @@ impl JMAP { } get::RequestArguments::Identity => todo!(), get::RequestArguments::EmailSubmission => todo!(), - get::RequestArguments::PushSubscription => todo!(), + get::RequestArguments::PushSubscription => { + self.push_subscription_get(req, acl_token).await?.into() + } get::RequestArguments::SieveScript => todo!(), get::RequestArguments::VacationResponse => todo!(), get::RequestArguments::Principal => todo!(), @@ -129,7 +167,9 @@ impl JMAP { } set::RequestArguments::Identity => todo!(), set::RequestArguments::EmailSubmission(_) => todo!(), - set::RequestArguments::PushSubscription => todo!(), + set::RequestArguments::PushSubscription => { + self.push_subscription_set(req, acl_token).await?.into() + } set::RequestArguments::SieveScript(_) => todo!(), set::RequestArguments::VacationResponse => todo!(), set::RequestArguments::Principal => todo!(), diff --git a/crates/jmap/src/email/copy.rs b/crates/jmap/src/email/copy.rs index 136462a8..4da4d4ed 100644 --- a/crates/jmap/src/email/copy.rs +++ b/crates/jmap/src/email/copy.rs @@ -17,6 +17,8 @@ use jmap_proto::{ collection::Collection, id::Id, property::Property, + state::{State, StateChange}, + type_state::TypeState, value::{MaybePatchValue, Value}, }, }; @@ -61,6 +63,7 @@ impl JMAP { old_state, created: VecMap::with_capacity(request.create.len()), not_created: VecMap::new(), + state_change: None, }; let from_message_ids = self @@ -344,6 +347,18 @@ impl JMAP { } } + // Update state + if !response.created.is_empty() { + response.new_state = self.get_state(account_id, Collection::Email).await?; + if let State::Exact(change_id) = &response.new_state { + response.state_change = StateChange::new(account_id) + .with_change(TypeState::Email, *change_id) + .with_change(TypeState::Mailbox, *change_id) + .with_change(TypeState::Thread, *change_id) + .into() + } + } + // Destroy ids if on_success_delete && !destroy_ids.is_empty() { *next_call = Call { diff --git a/crates/jmap/src/email/get.rs b/crates/jmap/src/email/get.rs index e03e2b4e..ac7deab0 100644 --- a/crates/jmap/src/email/get.rs +++ b/crates/jmap/src/email/get.rs @@ -90,7 +90,7 @@ impl JMAP { }; let mut response = GetResponse { account_id: Some(request.account_id), - state: self.get_state(account_id, Collection::Email).await?, + state: self.get_state(account_id, Collection::Email).await?.into(), list: Vec::with_capacity(ids.len()), not_found: vec![], }; diff --git a/crates/jmap/src/email/import.rs b/crates/jmap/src/email/import.rs index 146f637f..7cd5650c 100644 --- a/crates/jmap/src/email/import.rs +++ b/crates/jmap/src/email/import.rs @@ -4,7 +4,13 @@ use jmap_proto::{ set::{SetError, SetErrorType}, }, method::import::{ImportEmailRequest, ImportEmailResponse}, - types::{acl::Acl, collection::Collection, property::Property, state::State}, + types::{ + acl::Acl, + collection::Collection, + property::Property, + state::{State, StateChange}, + type_state::TypeState, + }, }; use utils::map::vec_map::VecMap; @@ -31,8 +37,14 @@ impl JMAP { None }; - let mut created = VecMap::with_capacity(request.emails.len()); - let mut not_created = VecMap::with_capacity(request.emails.len()); + let mut response = ImportEmailResponse { + account_id: request.account_id, + new_state: old_state.clone(), + old_state: old_state.into(), + created: VecMap::with_capacity(request.emails.len()), + not_created: VecMap::new(), + state_change: None, + }; 'outer: for (id, email) in request.emails { // Validate mailboxIds @@ -43,7 +55,7 @@ impl JMAP { .map(|m| m.unwrap().document_id()) .collect::>(); if mailbox_ids.is_empty() { - not_created.append( + response.not_created.append( id, SetError::invalid_properties() .with_property(Property::MailboxIds) @@ -53,7 +65,7 @@ impl JMAP { } for mailbox_id in &mailbox_ids { if !valid_mailbox_ids.contains(*mailbox_id) { - not_created.append( + response.not_created.append( id, SetError::invalid_properties() .with_property(Property::MailboxIds) @@ -61,7 +73,7 @@ impl JMAP { ); continue 'outer; } else if matches!(&can_add_mailbox_ids, Some(ids) if !ids.contains(*mailbox_id)) { - not_created.append( + response.not_created.append( id, SetError::forbidden().with_description(format!( "You are not allowed to add messages to mailbox {mailbox_id}." @@ -75,7 +87,7 @@ impl JMAP { let raw_message = match self.blob_download(&email.blob_id, acl_token).await { Ok(Some(raw_message)) => raw_message, Ok(None) => { - not_created.append( + response.not_created.append( id, SetError::new(SetErrorType::BlobNotFound) .with_description(format!("BlobId {} not found.", email.blob_id)), @@ -105,10 +117,10 @@ impl JMAP { .await { Ok(email) => { - created.append(id, email.into()); + response.created.append(id, email.into()); } Err(MaybeError::Permanent(reason)) => { - not_created.append( + response.not_created.append( id, SetError::new(SetErrorType::InvalidEmail).with_description(reason), ); @@ -119,16 +131,18 @@ impl JMAP { } } - Ok(ImportEmailResponse { - account_id: request.account_id, - new_state: if !created.is_empty() { - self.get_state(account_id, Collection::Email).await? - } else { - old_state.clone() - }, - old_state: old_state.into(), - created, - not_created, - }) + // Update state + if !response.created.is_empty() { + response.new_state = self.get_state(account_id, Collection::Email).await?; + if let State::Exact(change_id) = &response.new_state { + response.state_change = StateChange::new(account_id) + .with_change(TypeState::Email, *change_id) + .with_change(TypeState::Mailbox, *change_id) + .with_change(TypeState::Thread, *change_id) + .into() + } + } + + Ok(response) } } diff --git a/crates/jmap/src/email/set.rs b/crates/jmap/src/email/set.rs index 98dd1c01..db01ed28 100644 --- a/crates/jmap/src/email/set.rs +++ b/crates/jmap/src/email/set.rs @@ -14,6 +14,8 @@ use jmap_proto::{ id::Id, keyword::Keyword, property::Property, + state::{State, StateChange}, + type_state::TypeState, value::{MaybePatchValue, SetValue, Value}, }, }; @@ -971,10 +973,22 @@ impl JMAP { } } - if !changes.is_empty() { - response.new_state = self.commit_changes(account_id, changes).await?.into(); - } else if !response.created.is_empty() { - response.new_state = self.get_state(account_id, Collection::Email).await?.into(); + // Update state + if !changes.is_empty() || !response.created.is_empty() { + let new_state = if !changes.is_empty() { + self.commit_changes(account_id, changes).await? + } else { + self.get_state(account_id, Collection::Email).await? + }; + if let State::Exact(change_id) = &new_state { + response.state_change = StateChange::new(account_id) + .with_change(TypeState::Email, *change_id) + .with_change(TypeState::Mailbox, *change_id) + .with_change(TypeState::Thread, *change_id) + .into(); + } + + response.new_state = new_state.into(); } Ok(response) diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index d9e2d9cb..a601ac7f 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -12,21 +12,21 @@ use jmap_proto::{ query::{QueryRequest, QueryResponse}, set::{SetRequest, SetResponse}, }, - request::reference::MaybeReference, types::{collection::Collection, property::Property}, }; use mail_send::mail_auth::common::lru::{DnsCache, LruCache}; +use services::state::{self, init_state_manager, spawn_state_manager}; use sqlx::{mysql::MySqlPoolOptions, postgres::PgPoolOptions, sqlite::SqlitePoolOptions}; use store::{ - ahash::AHashMap, fts::Language, parking_lot::Mutex, query::{sort::Pagination, Comparator, Filter, ResultSet, SortedResultSet}, roaring::RoaringBitmap, - write::BitmapFamily, + write::{BatchBuilder, BitmapFamily}, BitmapKey, Deserialize, Serialize, Store, ValueKey, }; -use utils::{config::Rate, failed, map::vec_map::VecMap, UnwrapFailure}; +use tokio::sync::mpsc; +use utils::{config::Rate, failed, UnwrapFailure}; pub mod api; pub mod auth; @@ -34,7 +34,8 @@ pub mod blob; pub mod changes; pub mod email; pub mod mailbox; -//pub mod principal; +pub mod push; +pub mod services; pub mod thread; pub struct JMAP { @@ -46,6 +47,7 @@ pub struct JMAP { pub rate_limit_unauth: LruCache>>, pub oauth_codes: LruCache>, pub auth_db: AuthDatabase, + pub state_tx: mpsc::Sender, } pub struct Config { @@ -78,6 +80,9 @@ pub struct Config { pub rate_anonymous: Rate, pub rate_use_forwarded: bool, + pub event_source_throttle: Duration, + pub push_max_total: usize, + pub oauth_key: String, pub oauth_expiry_user_code: u64, pub oauth_expiry_auth_code: u64, @@ -90,6 +95,7 @@ pub struct Config { } pub const SUPERUSER_ID: u32 = 0; +pub const LONG_SLUMBER: Duration = Duration::from_secs(60 * 60 * 24); pub enum MaybeError { Temporary, @@ -97,7 +103,7 @@ pub enum MaybeError { } impl JMAP { - pub async fn new(config: &utils::config::Config) -> Self { + pub async fn init(config: &utils::config::Config) -> Arc { let auth_db = match config .value_require("jmap.auth.database.type") .failed("Invalid property") @@ -182,7 +188,10 @@ impl JMAP { _ => failed("Invalid auth database type"), }; - JMAP { + // Init state manager + let (state_tx, state_rx) = init_state_manager(); + + let jmap_server = Arc::new(JMAP { store: Store::open(config).await.failed("Unable to open database"), config: Config::new(config).failed("Invalid configuration file"), sessions: LruCache::with_capacity( @@ -216,7 +225,13 @@ impl JMAP { .unwrap_or(128), ), auth_db, - } + state_tx, + }); + + // Spawn state manager + spawn_state_manager(jmap_server.clone(), config, state_rx); + + jmap_server } pub async fn assign_document_id( @@ -388,37 +403,16 @@ impl JMAP { request: &SetRequest, collection: Collection, ) -> Result { - let n_create = request.create.as_ref().map_or(0, |objs| objs.len()); - let n_update = request.update.as_ref().map_or(0, |objs| objs.len()); - let n_destroy = request.destroy.as_ref().map_or(0, |objs| { - if let MaybeReference::Value(ids) = objs { - ids.len() - } else { - 0 - } - }); - if n_create + n_update + n_destroy > self.config.set_max_objects { - return Err(MethodError::RequestTooLarge); - } - let old_state = self - .assert_state( - request.account_id.document_id(), - collection, - &request.if_in_state, - ) - .await?; - - Ok(SetResponse { - account_id: request.account_id.into(), - new_state: old_state.clone().into(), - old_state: old_state.into(), - created: AHashMap::with_capacity(n_create), - updated: VecMap::with_capacity(n_update), - destroyed: Vec::with_capacity(n_destroy), - not_created: VecMap::new(), - not_updated: VecMap::new(), - not_destroyed: VecMap::new(), - }) + Ok( + SetResponse::from_request(request, self.config.set_max_objects)?.with_state( + self.assert_state( + request.account_id.document_id(), + collection, + &request.if_in_state, + ) + .await?, + ), + ) } pub async fn filter( @@ -518,6 +512,17 @@ impl JMAP { Ok(response) } + + pub async fn write_batch(&self, batch: BatchBuilder) -> Result<(), MethodError> { + self.store.write(batch.build()).await.map_err(|err| { + tracing::error!( + event = "error", + context = "write_batch", + error = ?err, + "Failed to write batch."); + MethodError::ServerPartialFail + }) + } } trait UpdateResults: Sized { diff --git a/crates/jmap/src/mailbox/get.rs b/crates/jmap/src/mailbox/get.rs index b52e4910..2bfa651c 100644 --- a/crates/jmap/src/mailbox/get.rs +++ b/crates/jmap/src/mailbox/get.rs @@ -61,7 +61,10 @@ impl JMAP { }); let mut response = GetResponse { account_id: Some(request.account_id), - state: self.get_state(account_id, Collection::Mailbox).await?, + state: self + .get_state(account_id, Collection::Mailbox) + .await? + .into(), list: Vec::with_capacity(ids.len()), not_found: vec![], }; diff --git a/crates/jmap/src/mailbox/set.rs b/crates/jmap/src/mailbox/set.rs index 1dcee8b4..82ce6bd2 100644 --- a/crates/jmap/src/mailbox/set.rs +++ b/crates/jmap/src/mailbox/set.rs @@ -15,6 +15,8 @@ use jmap_proto::{ collection::Collection, id::Id, property::Property, + state::StateChange, + type_state::TypeState, value::{MaybePatchValue, SetValue, Value}, }, }; @@ -95,16 +97,7 @@ impl JMAP { .custom(builder); changes.log_insert(Collection::Mailbox, document_id); ctx.mailbox_ids.insert(document_id); - self.store.write(batch.build()).await.map_err(|err| { - tracing::error!( - event = "error", - context = "mailbox_set", - account_id = account_id, - error = ?err, - "Failed to create mailbox(es)."); - MethodError::ServerPartialFail - })?; - + self.write_batch(batch).await?; ctx.set_response.created(id, document_id); } Err(err) => { @@ -116,6 +109,14 @@ impl JMAP { // Process updates 'update: for (id, object) in request.unwrap_update() { + // Make sure id won't be destroyed + if ctx.will_destroy.contains(&id) { + ctx.set_response + .not_updated + .append(id, SetError::will_destroy()); + continue 'update; + } + // Obtain mailbox let document_id = id.document_id(); if let Some(mut mailbox) = self @@ -198,6 +199,7 @@ impl JMAP { } // Process deletions + let mut did_remove_emails = false; 'destroy: for id in ctx.will_destroy { let document_id = id.document_id(); // Internal folders cannot be deleted @@ -242,6 +244,9 @@ impl JMAP { .await? { if on_destroy_remove_emails { + // Flag removal for state change notification + did_remove_emails = true; + // If the message is in multiple mailboxes, untag it from the current mailbox, // otherwise delete it. for message_id in message_ids { @@ -427,6 +432,16 @@ impl JMAP { // Write changes if !changes.is_empty() { + let state_change = + StateChange::new(account_id).with_change(TypeState::Mailbox, changes.change_id); + ctx.set_response.state_change = if did_remove_emails { + state_change + .with_change(TypeState::Email, changes.change_id) + .with_change(TypeState::Thread, changes.change_id) + } else { + state_change + } + .into(); ctx.set_response.new_state = self.commit_changes(account_id, changes).await?.into(); } diff --git a/crates/jmap/src/principal/mod.rs b/crates/jmap/src/principal/mod.rs deleted file mode 100644 index e3dcac25..00000000 --- a/crates/jmap/src/principal/mod.rs +++ /dev/null @@ -1,15 +0,0 @@ -pub mod set; - -/* - -user -uid -gid -secret -addresses -name -quota - - - - */ diff --git a/crates/jmap/src/principal/set.rs b/crates/jmap/src/principal/set.rs deleted file mode 100644 index 15fe91f0..00000000 --- a/crates/jmap/src/principal/set.rs +++ /dev/null @@ -1,266 +0,0 @@ -use jmap_proto::{ - error::{ - method::MethodError, - set::{SetError, SetErrorType}, - }, - method::set::{RequestArguments, SetRequest, SetResponse}, - object::{ - index::{IndexAs, IndexProperty, ObjectIndexBuilder}, - Object, - }, - types::{ - collection::Collection, - id::Id, - keyword::Keyword, - property::Property, - value::{MaybePatchValue, SetValue, Value}, - }, -}; -use store::{ - roaring::RoaringBitmap, - write::{ - assert::HashedValue, log::ChangeLogBuilder, BatchBuilder, DeserializeFrom, SerializeInto, - ToBitmaps, F_BITMAP, F_CLEAR, F_VALUE, - }, - BlobKind, Serialize, ValueKey, -}; - -use crate::{mailbox, JMAP, SUPERUSER_ID}; - -struct SetContext { - set_response: SetResponse, - principal_ids: RoaringBitmap, - will_destroy: Vec, -} - -pub static SCHEMA: &[IndexProperty] = &[ - IndexProperty::new(Property::Name) - .index_as(IndexAs::Text { - tokenize: true, - index: true, - }) - .required(), - IndexProperty::new(Property::Role).index_as(IndexAs::Text { - tokenize: false, - index: true, - }), - IndexProperty::new(Property::Role).index_as(IndexAs::HasProperty), - IndexProperty::new(Property::ParentId).index_as(IndexAs::Integer), - IndexProperty::new(Property::SortOrder).index_as(IndexAs::Integer), - IndexProperty::new(Property::IsSubscribed).index_as(IndexAs::IntegerList), -]; - -impl JMAP { - pub async fn principal_set( - &self, - mut request: SetRequest, - ) -> Result { - // Prepare response - let mut ctx = SetContext { - set_response: self - .prepare_set_response(&request, Collection::Principal) - .await?, - principal_ids: self - .get_document_ids(SUPERUSER_ID, Collection::Principal) - .await? - .unwrap_or_default(), - will_destroy: request.unwrap_destroy(), - }; - - // Process creates - let mut changes = ChangeLogBuilder::new(); - 'create: for (id, object) in request.unwrap_create() { - match self.principal_set_item(object, None, &ctx).await? { - Ok(builder) => { - let mut batch = BatchBuilder::new(); - let principal_id = self - .assign_document_id(SUPERUSER_ID, Collection::Principal) - .await?; - let create_mailboxes = matches!( - builder.get(&Property::Type), - Value::Text(t) if ["individual", "group"].contains(&t.as_str()) - ); - batch - .with_account_id(SUPERUSER_ID) - .with_collection(Collection::Principal) - .create_document(principal_id) - .custom(builder); - - // Create mailboxes - if create_mailboxes { - batch - .with_account_id(principal_id) - .with_collection(Collection::Mailbox); - for (name, role) in [ - ("Inbox", "inbox"), - ("Deleted Items", "trash"), - ("Drafts", "drafts"), - ("Sent Items", "sent"), - ("Junk Mail", "junk"), - ] { - batch - .create_document( - self.assign_document_id(principal_id, Collection::Mailbox) - .await?, - ) - .custom( - ObjectIndexBuilder::new(mailbox::set::SCHEMA).with_changes( - Object::with_capacity(3) - .with_property(Property::Name, name) - .with_property(Property::Role, role) - .with_property(Property::ParentId, 0u32), - ), - ); - } - } - - changes.log_insert(Collection::Principal, principal_id); - ctx.principal_ids.insert(principal_id); - self.store.write(batch.build()).await.map_err(|err| { - tracing::error!( - event = "error", - context = "principal_set", - error = ?err, - "Failed to create mailbox(es)."); - MethodError::ServerPartialFail - })?; - - ctx.set_response.created(id, principal_id); - } - Err(err) => { - ctx.set_response.not_created.append(id, err); - continue 'create; - } - } - } - - // Process updates - 'update: for (id, object) in request.unwrap_update() { - // Obtain mailbox - let principal_id = id.document_id(); - if let Some(mut principal) = self - .get_property::>>( - SUPERUSER_ID, - Collection::Principal, - principal_id, - Property::Value, - ) - .await? - { - match self - .principal_set_item(object, (principal_id, principal.take()).into(), &ctx) - .await? - { - Ok(builder) => { - let mut batch = BatchBuilder::new(); - batch - .with_account_id(SUPERUSER_ID) - .with_collection(Collection::Principal) - .create_document(principal_id) - .assert_value(Property::Value, &principal) - .custom(builder); - if !batch.is_empty() { - changes.log_update(Collection::Principal, principal_id); - match self.store.write(batch.build()).await { - Ok(_) => (), - Err(store::Error::AssertValueFailed) => { - ctx.set_response.not_updated.append(id, SetError::forbidden().with_description( - "Another process modified this principal, please try again.", - )); - continue 'update; - } - Err(err) => { - tracing::error!( - event = "error", - context = "principal_set", - error = ?err, - "Failed to update principal(s)."); - return Err(MethodError::ServerPartialFail); - } - } - } - ctx.set_response.updated.append(id, None); - } - Err(err) => { - ctx.set_response.not_updated.append(id, err); - continue 'update; - } - } - } else { - ctx.set_response - .not_updated - .append(id, SetError::not_found()); - } - } - - // Process deletions - 'destroy: for id in ctx.will_destroy { - let principal_id = id.document_id(); - // Obtain mailbox - if let Some(mailbox) = self - .get_property::>>( - SUPERUSER_ID, - Collection::Principal, - principal_id, - Property::Value, - ) - .await? - { - let delete_account_data = "todo"; - let mut batch = BatchBuilder::new(); - batch - .with_account_id(SUPERUSER_ID) - .with_collection(Collection::Principal) - .delete_document(principal_id) - .assert_value(Property::Value, &mailbox) - .custom(ObjectIndexBuilder::new(SCHEMA).with_current(mailbox.inner)); - - match self.store.write(batch.build()).await { - Ok(_) => { - changes.log_delete(Collection::Principal, principal_id); - ctx.set_response.destroyed.push(id); - } - Err(store::Error::AssertValueFailed) => { - ctx.set_response.not_destroyed.append( - id, - SetError::forbidden().with_description(concat!( - "Another process modified this mailbox ", - "while deleting it, please try again." - )), - ); - } - Err(err) => { - tracing::error!( - event = "error", - context = "mailbox_set", - document_id = principal_id, - error = ?err, - "Failed to delete principal."); - return Err(MethodError::ServerPartialFail); - } - } - } else { - ctx.set_response - .not_destroyed - .append(id, SetError::not_found()); - } - } - - // Write changes - if !changes.is_empty() { - ctx.set_response.new_state = self.commit_changes(SUPERUSER_ID, changes).await?.into(); - } - - Ok(ctx.set_response) - } - - #[allow(clippy::blocks_in_if_conditions)] - async fn principal_set_item( - &self, - changes_: Object, - update: Option<(u32, Object)>, - ctx: &SetContext, - ) -> Result, MethodError> { - todo!() - } -} diff --git a/crates/jmap/src/push/ece.rs b/crates/jmap/src/push/ece.rs new file mode 100644 index 00000000..33c4a598 --- /dev/null +++ b/crates/jmap/src/push/ece.rs @@ -0,0 +1,211 @@ +use aes_gcm::{ + aead::{generic_array::GenericArray, Aead}, + Aes128Gcm, Nonce, +}; +use hkdf::Hkdf; +use p256::{ + ecdh::EphemeralSecret, + elliptic_curve::{rand_core::OsRng, sec1::ToEncodedPoint}, + PublicKey, +}; +use sha2::Sha256; +use store::rand::Rng; + +/* + + From https://github.com/mozilla/rust-ece (MPL-2.0 license) + Adapted to use 'aes-gcm' and 'p256' crates instead of 'openssl'. + +*/ + +const ECE_WEBPUSH_AES128GCM_IKM_INFO_PREFIX: &str = "WebPush: info\0"; +const ECE_WEBPUSH_AES128GCM_IKM_INFO_LENGTH: usize = 144; +const ECE_WEBPUSH_IKM_LENGTH: usize = 32; +const ECE_WEBPUSH_PUBLIC_KEY_LENGTH: usize = 65; +const ECE_WEBPUSH_DEFAULT_RS: u32 = 4096; +const ECE_WEBPUSH_DEFAULT_PADDING_BLOCK_SIZE: usize = 128; + +const ECE_AES128GCM_PAD_SIZE: usize = 1; +const ECE_AES128GCM_KEY_INFO: &str = "Content-Encoding: aes128gcm\0"; +const ECE_AES128GCM_NONCE_INFO: &str = "Content-Encoding: nonce\0"; +const ECE_AES128GCM_HEADER_LENGTH: usize = 21; +const ECE_AES_KEY_LENGTH: usize = 16; + +const ECE_NONCE_LENGTH: usize = 12; +const ECE_TAG_LENGTH: usize = 16; + +pub fn ece_encrypt( + p256dh: &[u8], + client_auth_secret: &[u8], + mut data: &[u8], +) -> Result, String> { + let salt = store::rand::thread_rng().gen::<[u8; 16]>(); + let server_secret = EphemeralSecret::random(&mut OsRng); + let server_public_key = server_secret.public_key(); + let server_public_key_bytes = server_public_key.to_encoded_point(false); + + let client_public_key = PublicKey::from_sec1_bytes(p256dh).map_err(|e| e.to_string())?; + let shared_secret = server_secret.diffie_hellman(&client_public_key); + + let ikm_info = generate_info(p256dh, server_public_key_bytes.as_bytes()); + let ikm = hkdf_sha256( + client_auth_secret, + &shared_secret.raw_secret_bytes()[..], + &ikm_info, + ECE_WEBPUSH_IKM_LENGTH, + )?; + let key = hkdf_sha256( + &salt, + &ikm, + ECE_AES128GCM_KEY_INFO.as_bytes(), + ECE_AES_KEY_LENGTH, + )?; + let nonce = hkdf_sha256( + &salt, + &ikm, + ECE_AES128GCM_NONCE_INFO.as_bytes(), + ECE_NONCE_LENGTH, + )?; + + // Calculate pad length + let mut pad_length = ECE_WEBPUSH_DEFAULT_PADDING_BLOCK_SIZE + - (data.len() % ECE_WEBPUSH_DEFAULT_PADDING_BLOCK_SIZE); + if pad_length < ECE_AES128GCM_PAD_SIZE { + pad_length += ECE_WEBPUSH_DEFAULT_PADDING_BLOCK_SIZE; + } + + // Split into records + let rs = ECE_WEBPUSH_DEFAULT_RS as usize - ECE_TAG_LENGTH; + let mut min_num_records = data.len() / (rs - 1); + if data.len() % (rs - 1) != 0 { + min_num_records += 1; + } + let mut pad_length = std::cmp::max(pad_length, min_num_records); + let total_size = data.len() + pad_length; + let mut num_records = total_size / rs; + let size_of_final_record = total_size % rs; + if size_of_final_record > 0 { + num_records += 1; + } + let data_per_record = data.len() / num_records; + let mut extra_data = data.len() % num_records; + if size_of_final_record > 0 && data_per_record > size_of_final_record - 1 { + extra_data += data_per_record - (size_of_final_record - 1) + } + let mut sequence_number = 0; + let mut plain_text = + Vec::with_capacity(data_per_record + ECE_WEBPUSH_DEFAULT_PADDING_BLOCK_SIZE); + + // Write header + let key_id = server_public_key_bytes.as_bytes(); + debug_assert_eq!(key_id.len(), ECE_WEBPUSH_PUBLIC_KEY_LENGTH); + let mut output = Vec::with_capacity( + ECE_AES128GCM_HEADER_LENGTH + key_id.len() + total_size + num_records * ECE_TAG_LENGTH, + ); + output.extend_from_slice(&salt); + output.extend_from_slice(&ECE_WEBPUSH_DEFAULT_RS.to_be_bytes()); + output.push(key_id.len() as u8); + output.extend_from_slice(key_id); + + loop { + let records_remaining = num_records - sequence_number; + if records_remaining == 0 { + break; + } + let mut data_share = data_per_record; + if data_share > data.len() { + data_share = data.len(); + } else if extra_data > 0 { + let mut extra_share = extra_data / (records_remaining - 1); + if extra_data % (records_remaining - 1) != 0 { + extra_share += 1; + } + data_share += extra_share; + extra_data -= extra_share; + } + + let cur_data = &data[0..data_share]; + data = &data[data_share..]; + let padding = std::cmp::min(pad_length, rs - data_share); + pad_length -= padding; + let cur_sequence_number = sequence_number; + sequence_number += 1; + + let padded_plaintext_len = cur_data.len() + padding; + + plain_text.extend_from_slice(cur_data); + plain_text.push(if sequence_number == num_records { 2 } else { 1 }); + plain_text.resize(padded_plaintext_len, 0); + + output.extend_from_slice(&aes_gcm_128_encrypt( + &key, + &generate_iv(&nonce, cur_sequence_number), + &plain_text, + )?); + plain_text.clear(); + } + + Ok(output) +} + +fn hkdf_sha256(salt: &[u8], secret: &[u8], info: &[u8], len: usize) -> Result, String> { + let (_, hk) = Hkdf::::extract(Some(salt), secret); + let mut okm = vec![0u8; len]; + hk.expand(info, &mut okm).map_err(|e| e.to_string())?; + Ok(okm) +} + +fn aes_gcm_128_encrypt(key: &[u8], nonce: &[u8], data: &[u8]) -> Result, String> { + ::new(&GenericArray::clone_from_slice(key)) + .encrypt(Nonce::from_slice(nonce), data) + .map_err(|e| e.to_string()) +} + +fn generate_info( + client_public_key: &[u8], + server_public_key: &[u8], +) -> [u8; ECE_WEBPUSH_AES128GCM_IKM_INFO_LENGTH] { + let mut info = [0u8; ECE_WEBPUSH_AES128GCM_IKM_INFO_LENGTH]; + let prefix = ECE_WEBPUSH_AES128GCM_IKM_INFO_PREFIX.as_bytes(); + let mut offset = prefix.len(); + info[0..offset].copy_from_slice(prefix); + info[offset..offset + ECE_WEBPUSH_PUBLIC_KEY_LENGTH].copy_from_slice(client_public_key); + offset += ECE_WEBPUSH_PUBLIC_KEY_LENGTH; + info[offset..].copy_from_slice(server_public_key); + info +} + +pub fn generate_iv(nonce: &[u8], counter: usize) -> [u8; ECE_NONCE_LENGTH] { + let mut iv = [0u8; ECE_NONCE_LENGTH]; + let offset = ECE_NONCE_LENGTH - 8; + iv[0..offset].copy_from_slice(&nonce[0..offset]); + let mask = u64::from_be_bytes((&nonce[offset..]).try_into().unwrap()); + iv[offset..].copy_from_slice(&(mask ^ (counter as u64)).to_be_bytes()); + iv +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ece_roundtrip() { + for len in [1, 2, 5, 16, 256, 1024, 2048, 4096, 1024 * 1024] { + let (keypair, auth_secret) = ece::generate_keypair_and_auth_secret().unwrap(); + + let bytes: Vec = (0..len).map(|_| store::rand::random::()).collect(); + + let encrypted_bytes = + ece_encrypt(&keypair.pub_as_raw().unwrap(), &auth_secret, &bytes).unwrap(); + + let decrypted_bytes = ece::decrypt( + &keypair.raw_components().unwrap(), + &auth_secret, + &encrypted_bytes, + ) + .unwrap(); + + assert_eq!(bytes, decrypted_bytes, "len: {}", len); + } + } +} diff --git a/crates/jmap/src/push/get.rs b/crates/jmap/src/push/get.rs new file mode 100644 index 00000000..3d2c071d --- /dev/null +++ b/crates/jmap/src/push/get.rs @@ -0,0 +1,233 @@ +use base64::{engine::general_purpose, Engine}; +use jmap_proto::{ + error::method::MethodError, + method::get::{GetRequest, GetResponse, RequestArguments}, + object::Object, + types::{collection::Collection, property::Property, type_state::TypeState, value::Value}, +}; +use store::{write::now, BitmapKey, ValueKey}; +use utils::map::bitmap::Bitmap; + +use crate::{auth::AclToken, services::state, JMAP}; + +use super::{EncryptionKeys, PushSubscription, UpdateSubscription}; + +impl JMAP { + pub async fn push_subscription_get( + &self, + mut request: GetRequest, + acl_token: &AclToken, + ) -> Result { + let ids = request.unwrap_ids(self.config.get_max_objects)?; + let properties = request.unwrap_properties(&[ + Property::Id, + Property::DeviceClientId, + Property::VerificationCode, + Property::Expires, + Property::Types, + ]); + let account_id = acl_token.primary_id(); + let push_ids = self + .get_document_ids(account_id, Collection::PushSubscription) + .await? + .unwrap_or_default(); + let ids = if let Some(ids) = ids { + ids + } else { + push_ids + .iter() + .take(self.config.get_max_objects) + .map(Into::into) + .collect::>() + }; + let mut response = GetResponse { + account_id: None, + state: None, + list: Vec::with_capacity(ids.len()), + not_found: vec![], + }; + + for id in ids { + // Obtain the push subscription object + let document_id = id.document_id(); + if !push_ids.contains(document_id) { + response.not_found.push(id); + continue; + } + let mut push = if let Some(push) = self + .get_property::>( + account_id, + Collection::PushSubscription, + document_id, + Property::Value, + ) + .await? + { + push + } else { + response.not_found.push(id); + continue; + }; + let mut result = Object::with_capacity(properties.len()); + for property in &properties { + match property { + Property::Id => { + result.append(Property::Id, Value::Id(id)); + } + Property::Url | Property::Keys | Property::Value => { + return Err(MethodError::Forbidden( + "The 'url' and 'keys' properties are not readable".to_string(), + )); + } + property => { + result.append(property.clone(), push.remove(property)); + } + } + } + response.list.push(result); + } + + Ok(response) + } + + pub async fn fetch_push_subscriptions(&self, account_id: u32) -> store::Result { + let mut subscriptions = Vec::new(); + let document_ids = self + .store + .get_bitmap(BitmapKey::document_ids( + account_id, + Collection::PushSubscription, + )) + .await? + .unwrap_or_default(); + + let current_time = now(); + + for document_id in document_ids { + let mut subscription = self + .store + .get_value::>(ValueKey::new( + account_id, + Collection::PushSubscription, + document_id, + Property::Value, + )) + .await? + .ok_or_else(|| { + store::Error::InternalError(format!( + "Could not find push subscription {}", + document_id + )) + })?; + + let expires = subscription + .properties + .get(&Property::Expires) + .and_then(|p| p.as_date()) + .ok_or_else(|| { + store::Error::InternalError(format!( + "Missing expires property for push subscription {}", + document_id + )) + })? + .timestamp() as u64; + if expires > current_time { + let keys = if let Some((auth, p256dh)) = subscription + .properties + .remove(&Property::Keys) + .and_then(|value| value.try_unwrap_object()) + .and_then(|mut obj| { + ( + obj.properties + .remove(&Property::Auth) + .and_then(|value| value.try_unwrap_string())?, + obj.properties + .remove(&Property::P256dh) + .and_then(|value| value.try_unwrap_string())?, + ) + .into() + }) { + EncryptionKeys { + p256dh: general_purpose::URL_SAFE + .decode(&p256dh) + .unwrap_or_default(), + auth: general_purpose::URL_SAFE.decode(&auth).unwrap_or_default(), + } + .into() + } else { + None + }; + let verification_code = subscription + .properties + .remove(&Property::Value) + .and_then(|p| p.try_unwrap_string()) + .ok_or_else(|| { + store::Error::InternalError(format!( + "Missing verificationCode property for push subscription {}", + document_id + )) + })?; + let url = subscription + .properties + .remove(&Property::Url) + .and_then(|p| p.try_unwrap_string()) + .ok_or_else(|| { + store::Error::InternalError(format!( + "Missing Url property for push subscription {}", + document_id + )) + })?; + + if subscription + .properties + .get(&Property::VerificationCode) + .and_then(|p| p.as_string()) + .map_or(false, |v| v == verification_code) + { + let types = if let Some(Value::List(value)) = + subscription.properties.remove(&Property::Types) + { + if !value.is_empty() { + let mut type_states = Bitmap::new(); + for type_state in value { + if let Some(type_state) = type_state + .as_string() + .and_then(|type_state| TypeState::try_from(type_state).ok()) + { + type_states.insert(type_state); + } + } + type_states + } else { + Bitmap::all() + } + } else { + Bitmap::all() + }; + + // Add verified subscription + subscriptions.push(UpdateSubscription::Verified(PushSubscription { + id: document_id, + url, + expires, + types, + keys, + })); + } else { + // Add unverified subscription + subscriptions.push(UpdateSubscription::Unverified { + id: document_id, + url, + code: verification_code, + keys, + }); + } + } + } + + Ok(state::Event::UpdateSubscriptions { + account_id, + subscriptions, + }) + } +} diff --git a/crates/jmap/src/push/manager.rs b/crates/jmap/src/push/manager.rs new file mode 100644 index 00000000..57e127f6 --- /dev/null +++ b/crates/jmap/src/push/manager.rs @@ -0,0 +1,316 @@ +use base64::{engine::general_purpose, Engine}; +use jmap_proto::types::id::Id; +use store::ahash::{AHashMap, AHashSet}; +use tokio::sync::mpsc; +use utils::{config::Config, UnwrapFailure}; + +use crate::{api::StateChangeResponse, services::IPC_CHANNEL_BUFFER, LONG_SLUMBER}; + +use super::{ece::ece_encrypt, EncryptionKeys, Event, PushServer, PushUpdate}; + +use reqwest::header::{CONTENT_ENCODING, CONTENT_TYPE}; +use std::{ + collections::hash_map::Entry, + time::{Duration, Instant}, +}; + +pub fn spawn_push_manager(settings: &Config) -> mpsc::Sender { + let (push_tx_, mut push_rx) = mpsc::channel::(IPC_CHANNEL_BUFFER); + let push_tx = push_tx_.clone(); + + let push_attempt_interval: Duration = settings + .property_or_static("jmap.push.attempts.interval", "1m") + .failed("Invalid configuration"); + let push_attempts_max: u32 = settings + .property_or_static("jmap.push.attempts.max", "3") + .failed("Invalid configuration"); + let push_retry_interval: Duration = settings + .property_or_static("jmap.push.retry.interval", "1s") + .failed("Invalid configuration"); + let push_timeout: Duration = settings + .property_or_static("jmap.push.timeout.request", "10s") + .failed("Invalid configuration"); + let push_verify_timeout: Duration = settings + .property_or_static("jmap.push.timeout.verify", "1m") + .failed("Invalid configuration"); + let push_throttle: Duration = settings + .property_or_static("jmap.push.throttle", "1s") + .failed("Invalid configuration"); + + tokio::spawn(async move { + let mut subscriptions = AHashMap::default(); + let mut last_verify: AHashMap = AHashMap::default(); + let mut last_retry = Instant::now(); + let mut retry_timeout = LONG_SLUMBER; + let mut retry_ids = AHashSet::default(); + + loop { + match tokio::time::timeout(retry_timeout, push_rx.recv()).await { + Ok(Some(event)) => match event { + Event::Update { updates } => { + for update in updates { + match update { + PushUpdate::Verify { + id, + account_id, + url, + code, + keys, + } => { + let current_time = Instant::now(); + + #[cfg(feature = "test_mode")] + if url.contains("skip_checks") { + last_verify.insert( + account_id, + current_time + - (push_verify_timeout + Duration::from_millis(1)), + ); + } + + if last_verify + .get(&account_id) + .map(|last_verify| { + current_time - *last_verify > push_verify_timeout + }) + .unwrap_or(true) + { + tokio::spawn(async move { + http_request( + url, + format!( + concat!( + "{{\"@type\":\"PushVerification\",", + "\"pushSubscriptionId\":\"{}\",", + "\"verificationCode\":\"{}\"}}" + ), + Id::from(id), + code + ), + keys, + push_timeout, + ) + .await; + }); + + last_verify.insert(account_id, current_time); + } else { + tracing::debug!( + concat!( + "Failed to verify push subscription: ", + "Too many requests from accountId {}." + ), + account_id + ); + continue; + } + } + PushUpdate::Register { id, url, keys } => { + if let Entry::Vacant(entry) = subscriptions.entry(id) { + entry.insert(PushServer { + url, + keys, + num_attempts: 0, + last_request: Instant::now() + - (push_throttle + Duration::from_millis(1)), + state_changes: Vec::new(), + in_flight: false, + }); + } + } + PushUpdate::Unregister { id } => { + subscriptions.remove(&id); + } + } + } + } + Event::Push { ids, state_change } => { + for id in ids { + if let Some(subscription) = subscriptions.get_mut(&id) { + subscription.state_changes.push(state_change.clone()); + let last_request = subscription.last_request.elapsed(); + + if !subscription.in_flight + && ((subscription.num_attempts == 0 + && last_request > push_throttle) + || ((1..push_attempts_max) + .contains(&subscription.num_attempts) + && last_request > push_attempt_interval)) + { + subscription.send(id, push_tx.clone(), push_timeout); + retry_ids.remove(&id); + } else { + retry_ids.insert(id); + } + } else { + tracing::debug!("No push subscription found for id: {}", id); + } + } + } + Event::Reset => { + subscriptions.clear(); + } + Event::DeliverySuccess { id } => { + if let Some(subscription) = subscriptions.get_mut(&id) { + subscription.num_attempts = 0; + subscription.in_flight = false; + retry_ids.remove(&id); + } + } + Event::DeliveryFailure { id, state_changes } => { + if let Some(subscription) = subscriptions.get_mut(&id) { + subscription.last_request = Instant::now(); + subscription.num_attempts += 1; + subscription.state_changes.extend(state_changes); + subscription.in_flight = false; + retry_ids.insert(id); + } + } + }, + Ok(None) => { + break; + } + Err(_) => (), + } + + retry_timeout = if !retry_ids.is_empty() { + let last_retry_elapsed = last_retry.elapsed(); + + if last_retry_elapsed >= push_retry_interval { + let mut remove_ids = Vec::with_capacity(retry_ids.len()); + + for retry_id in &retry_ids { + if let Some(subscription) = subscriptions.get_mut(retry_id) { + let last_request = subscription.last_request.elapsed(); + + if !subscription.in_flight + && ((subscription.num_attempts == 0 + && last_request >= push_throttle) + || (subscription.num_attempts > 0 + && last_request >= push_attempt_interval)) + { + if subscription.num_attempts < push_attempts_max { + subscription.send(*retry_id, push_tx.clone(), push_timeout); + } else { + tracing::debug!( + concat!( + "Failed to deliver push subscription: ", + "Too many attempts for url {}." + ), + subscription.url + ); + subscription.state_changes.clear(); + subscription.num_attempts = 0; + } + remove_ids.push(*retry_id); + } + } else { + remove_ids.push(*retry_id); + } + } + + if remove_ids.len() < retry_ids.len() { + for remove_id in remove_ids { + retry_ids.remove(&remove_id); + } + last_retry = Instant::now(); + push_retry_interval + } else { + retry_ids.clear(); + LONG_SLUMBER + } + } else { + push_retry_interval - last_retry_elapsed + } + } else { + LONG_SLUMBER + }; + } + }); + + push_tx_ +} + +impl PushServer { + fn send(&mut self, id: Id, push_tx: mpsc::Sender, push_timeout: Duration) { + let url = self.url.clone(); + let keys = self.keys.clone(); + let state_changes = std::mem::take(&mut self.state_changes); + + self.in_flight = true; + self.last_request = Instant::now(); + + tokio::spawn(async move { + let mut response = StateChangeResponse::new(); + for state_change in &state_changes { + for (type_state, change_id) in &state_change.types { + response + .changed + .get_mut_or_insert(state_change.account_id.into()) + .set(*type_state, (*change_id).into()); + } + } + + push_tx + .send( + if http_request( + url, + serde_json::to_string(&response).unwrap(), + keys, + push_timeout, + ) + .await + { + Event::DeliverySuccess { id } + } else { + Event::DeliveryFailure { id, state_changes } + }, + ) + .await + .ok(); + }); + } +} + +async fn http_request( + url: String, + mut body: String, + keys: Option, + push_timeout: Duration, +) -> bool { + let client_builder = reqwest::Client::builder().timeout(push_timeout); + + #[cfg(feature = "test_mode")] + let client_builder = client_builder.danger_accept_invalid_certs(true); + + let mut client = client_builder + .build() + .unwrap_or_default() + .post(&url) + .header(CONTENT_TYPE, "application/json") + .header("TTL", "86400"); + + if let Some(keys) = keys { + match ece_encrypt(&keys.p256dh, &keys.auth, body.as_bytes()) + .map(|b| general_purpose::URL_SAFE.encode(b)) + { + Ok(body_) => { + body = body_; + client = client.header(CONTENT_ENCODING, "aes128gcm"); + } + Err(err) => { + // Do not reattempt if encryption fails. + tracing::debug!("Failed to encrypt push subscription to {}: {}", url, err); + return true; + } + } + } + + match client.body(body).send().await { + Ok(response) => response.status().is_success(), + Err(err) => { + tracing::debug!("HTTP post to {} failed with: {}", url, err); + false + } + } +} diff --git a/crates/jmap/src/push/mod.rs b/crates/jmap/src/push/mod.rs new file mode 100644 index 00000000..df078ed5 --- /dev/null +++ b/crates/jmap/src/push/mod.rs @@ -0,0 +1,83 @@ +pub mod ece; +pub mod get; +pub mod manager; +pub mod set; + +use std::time::Instant; + +use jmap_proto::types::{id::Id, state::StateChange, type_state::TypeState}; +use utils::map::bitmap::Bitmap; + +#[derive(Debug)] +pub enum UpdateSubscription { + Unverified { + id: u32, + url: String, + code: String, + keys: Option, + }, + Verified(PushSubscription), +} + +#[derive(Debug)] +pub struct PushSubscription { + pub id: u32, + pub url: String, + pub expires: u64, + pub types: Bitmap, + pub keys: Option, +} + +#[derive(Debug, Clone)] +pub struct EncryptionKeys { + pub p256dh: Vec, + pub auth: Vec, +} + +#[derive(Debug)] +pub enum Event { + Update { + updates: Vec, + }, + Push { + ids: Vec, + state_change: StateChange, + }, + DeliverySuccess { + id: Id, + }, + DeliveryFailure { + id: Id, + state_changes: Vec, + }, + Reset, +} + +#[derive(Debug)] +pub enum PushUpdate { + Verify { + id: u32, + account_id: u32, + url: String, + code: String, + keys: Option, + }, + Register { + id: Id, + url: String, + keys: Option, + }, + Unregister { + id: Id, + }, +} + +#[derive(Debug)] +pub struct PushServer { + url: String, + keys: Option, + num_attempts: u32, + last_request: Instant, + state_changes: Vec, + in_flight: bool, +} diff --git a/crates/jmap/src/push/set.rs b/crates/jmap/src/push/set.rs new file mode 100644 index 00000000..245c70d4 --- /dev/null +++ b/crates/jmap/src/push/set.rs @@ -0,0 +1,271 @@ +use base64::{engine::general_purpose, Engine}; +use jmap_proto::{ + error::{method::MethodError, set::SetError}, + method::set::{RequestArguments, SetRequest, SetResponse}, + object::Object, + response::references::EvalObjectReferences, + types::{ + collection::Collection, + date::UTCDate, + property::Property, + type_state::TypeState, + value::{MaybePatchValue, Value}, + }, +}; +use store::{ + rand::{distributions::Alphanumeric, thread_rng, Rng}, + write::{now, BatchBuilder, F_CLEAR, F_VALUE}, +}; + +use crate::{auth::AclToken, JMAP}; + +const EXPIRES_MAX: i64 = 7 * 24 * 3600; // 7 days +const VERIFICATION_CODE_LEN: usize = 32; + +impl JMAP { + pub async fn push_subscription_set( + &self, + mut request: SetRequest, + acl_token: &AclToken, + ) -> Result { + let account_id = acl_token.primary_id(); + let mut push_ids = self + .get_document_ids(account_id, Collection::PushSubscription) + .await? + .unwrap_or_default(); + let mut response = SetResponse::from_request(&request, self.config.set_max_objects)?; + let will_destroy = request.unwrap_destroy(); + + // Process creates + 'create: for (id, object) in request.unwrap_create() { + let mut push = Object::with_capacity(object.properties.len()); + + if push_ids.len() as usize >= self.config.push_max_total { + response.not_created.append(id, SetError::forbidden().with_description( + "There are too many subscriptions, please delete some before adding a new one.", + )); + continue 'create; + } + + for (property, value) in object.properties { + match response + .eval_object_references(value) + .and_then(|value| validate_push_value(&property, value, None)) + { + Ok(Value::Null) => (), + Ok(value) => { + push.set(property, value); + } + Err(err) => { + response.not_created.append(id, err); + continue 'create; + } + } + } + + if !push.properties.contains_key(&Property::DeviceClientId) + || !push.properties.contains_key(&Property::Url) + { + response.not_created.append( + id, + SetError::invalid_properties() + .with_properties([Property::DeviceClientId, Property::Url]) + .with_description("Missing required properties"), + ); + continue 'create; + } + + // Add expiry time if missing + if !push.properties.contains_key(&Property::Expires) { + push.append( + Property::Expires, + Value::Date(UTCDate::from_timestamp(now() as i64 + EXPIRES_MAX)), + ) + } + + // Generate random verification code + push.append( + Property::Value, + Value::Text( + thread_rng() + .sample_iter(Alphanumeric) + .take(VERIFICATION_CODE_LEN) + .map(char::from) + .collect::(), + ), + ); + + // Insert record + let mut batch = BatchBuilder::new(); + let document_id = self + .assign_document_id(account_id, Collection::PushSubscription) + .await?; + batch + .with_account_id(account_id) + .with_collection(Collection::PushSubscription) + .create_document(document_id) + .value(Property::Value, push, F_VALUE); + push_ids.insert(document_id); + self.write_batch(batch).await?; + response.created(id, document_id); + } + + // Process updates + 'update: for (id, object) in request.unwrap_update() { + // Make sure id won't be destroyed + if will_destroy.contains(&id) { + response.not_updated.append(id, SetError::will_destroy()); + continue 'update; + } + + // Obtain push subscription + let document_id = id.document_id(); + let mut push = if let Some(push) = self + .get_property::>( + account_id, + Collection::PushSubscription, + document_id, + Property::Value, + ) + .await? + { + push + } else { + response.not_updated.append(id, SetError::not_found()); + continue 'update; + }; + + for (property, value) in object.properties { + match response + .eval_object_references(value) + .and_then(|value| validate_push_value(&property, value, Some(&push))) + { + Ok(Value::Null) => { + push.remove(&property); + } + Ok(value) => { + push.set(property, value); + } + Err(err) => { + response.not_updated.append(id, err); + continue 'update; + } + }; + } + + // Update record + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::PushSubscription) + .update_document(document_id) + .value(Property::Value, push, F_VALUE); + self.write_batch(batch).await?; + response.updated.append(id, None); + } + + // Process deletions + for id in will_destroy { + let document_id = id.document_id(); + if push_ids.contains(document_id) { + // Update record + let mut batch = BatchBuilder::new(); + batch + .with_account_id(account_id) + .with_collection(Collection::PushSubscription) + .delete_document(document_id) + .value(Property::Value, (), F_VALUE | F_CLEAR); + self.write_batch(batch).await?; + response.destroyed.push(id); + } else { + response.not_destroyed.append(id, SetError::not_found()); + } + } + + // Update push subscriptions + if response.has_changes() { + self.update_push_subscriptions(account_id).await; + } + + Ok(response) + } +} + +fn validate_push_value( + property: &Property, + value: MaybePatchValue, + current: Option<&Object>, +) -> Result { + Ok(match (property, value) { + (Property::DeviceClientId, MaybePatchValue::Value(Value::Text(value))) + if current.is_none() && value.len() < 255 => + { + Value::Text(value) + } + (Property::Url, MaybePatchValue::Value(Value::Text(value))) + if current.is_none() && value.len() < 512 && value.starts_with("https://") => + { + Value::Text(value) + } + (Property::Keys, MaybePatchValue::Value(Value::Object(value))) + if current.is_none() + && value.properties.len() == 2 + && matches!(value.get(&Property::Auth), Value::Text(auth) if auth.len() < 1024 && + general_purpose::URL_SAFE.decode(auth).is_ok()) + && matches!(value.get(&Property::P256dh), Value::Text(p256dh) if p256dh.len() < 1024 && + general_purpose::URL_SAFE.decode(p256dh).is_ok()) => + { + Value::Object(value) + } + (Property::Expires, MaybePatchValue::Value(Value::Date(value))) => { + let current_time = now() as i64; + let expires = value.timestamp(); + Value::Date(UTCDate::from_timestamp( + if expires > current_time && (expires - current_time) > EXPIRES_MAX { + current_time + EXPIRES_MAX + } else { + expires + }, + )) + } + (Property::Expires, MaybePatchValue::Value(Value::Null)) => { + Value::Date(UTCDate::from_timestamp(now() as i64 + EXPIRES_MAX)) + } + (Property::Types, MaybePatchValue::Value(Value::List(value))) + if value.iter().all(|value| { + value + .as_string() + .and_then(|value| TypeState::try_from(value).ok()) + .is_some() + }) => + { + Value::List(value) + } + (Property::VerificationCode, MaybePatchValue::Value(Value::Text(value))) + if current.is_some() => + { + if current + .as_ref() + .unwrap() + .properties + .get(&Property::Value) + .map_or(false, |v| matches!(v, Value::Text(v) if v == &value)) + { + Value::Text(value) + } else { + return Err(SetError::invalid_properties() + .with_property(property.clone()) + .with_description("Verification code does not match.".to_string())); + } + } + ( + Property::Keys | Property::Types | Property::VerificationCode, + MaybePatchValue::Value(Value::Null), + ) => Value::Null, + (property, _) => { + return Err(SetError::invalid_properties() + .with_property(property.clone()) + .with_description("Field could not be set.")); + } + }) +} diff --git a/crates/jmap/src/services/mod.rs b/crates/jmap/src/services/mod.rs new file mode 100644 index 00000000..06e0179f --- /dev/null +++ b/crates/jmap/src/services/mod.rs @@ -0,0 +1,3 @@ +pub mod state; + +pub const IPC_CHANNEL_BUFFER: usize = 1024; diff --git a/crates/jmap/src/services/state.rs b/crates/jmap/src/services/state.rs new file mode 100644 index 00000000..e20e9111 --- /dev/null +++ b/crates/jmap/src/services/state.rs @@ -0,0 +1,434 @@ +use std::{ + sync::Arc, + time::{Duration, Instant, SystemTime}, +}; + +use jmap_proto::types::{id::Id, state::StateChange, type_state::TypeState}; +use store::ahash::AHashMap; +use tokio::sync::mpsc; +use utils::{config::Config, map::bitmap::Bitmap}; + +use crate::{ + push::{manager::spawn_push_manager, UpdateSubscription}, + JMAP, +}; + +use super::IPC_CHANNEL_BUFFER; + +#[derive(Debug)] +pub enum Event { + Subscribe { + id: u32, + account_id: u32, + types: Bitmap, + tx: mpsc::Sender, + }, + Publish { + state_change: StateChange, + }, + UpdateSharedAccounts { + account_id: u32, + }, + UpdateSubscriptions { + account_id: u32, + subscriptions: Vec, + }, + Stop, +} + +#[derive(Debug)] +struct Subscriber { + types: Bitmap, + subscription: SubscriberType, +} + +#[derive(Debug)] +pub enum SubscriberType { + Ipc { tx: mpsc::Sender }, + Push { expires: u64 }, +} + +impl Subscriber { + fn is_valid(&self, current_time: u64) -> bool { + match &self.subscription { + SubscriberType::Ipc { tx } => !tx.is_closed(), + SubscriberType::Push { expires } => expires > ¤t_time, + } + } +} + +const PURGE_EVERY_SECS: u64 = 3600; +const SEND_TIMEOUT_MS: u64 = 500; + +pub fn init_state_manager() -> (mpsc::Sender, mpsc::Receiver) { + mpsc::channel::(IPC_CHANNEL_BUFFER) +} + +pub fn spawn_state_manager( + core: Arc, + settings: &Config, + mut change_rx: mpsc::Receiver, +) { + let push_tx = spawn_push_manager(settings); + + tokio::spawn(async move { + let mut subscribers: AHashMap> = AHashMap::default(); + let mut shared_accounts: AHashMap> = AHashMap::default(); + let mut shared_accounts_map: AHashMap>> = + AHashMap::default(); + + let mut last_purge = Instant::now(); + + while let Some(event) = change_rx.recv().await { + let mut purge_needed = last_purge.elapsed() >= Duration::from_secs(PURGE_EVERY_SECS); + + match event { + Event::Stop => { + if let Err(err) = push_tx.send(crate::push::Event::Reset).await { + tracing::debug!("Error sending push reset: {}", err); + } + break; + } + Event::UpdateSharedAccounts { account_id } => { + // Obtain account membership and shared mailboxes + let acl = match core.get_acl_token(account_id).await { + Some(result) => result, + None => { + continue; + } + }; + + // Delete any removed sharings + if let Some(shared_account_ids) = shared_accounts.get(&account_id) { + for shared_account_id in shared_account_ids { + if *shared_account_id != acl.primary_id + && !acl.member_of.contains(shared_account_id) + && !acl + .access_to + .iter() + .any(|(id, _)| *id == *shared_account_id) + { + if let Some(shared_list) = + shared_accounts_map.get_mut(shared_account_id) + { + shared_list.remove(&account_id); + if shared_list.is_empty() { + shared_accounts_map.remove(shared_account_id); + } + } + } + } + } + + // Update lists + let mut shared_account_ids = + Vec::with_capacity(acl.member_of.len() + 1 + acl.access_to.len()); + for member_id in [acl.primary_id].iter().chain(acl.member_of.iter()) { + shared_account_ids.push(*member_id); + shared_accounts_map + .entry(*member_id) + .or_insert_with(AHashMap::new) + .insert(account_id, Bitmap::all()); + } + for (shared_account_id, shared_collections) in acl.access_to.iter() { + let mut types: Bitmap = Bitmap::new(); + for collection in *shared_collections { + if let Ok(type_state) = TypeState::try_from(collection) { + types.insert(type_state); + if type_state == TypeState::Email { + types.insert(TypeState::EmailDelivery); + types.insert(TypeState::Thread); + } + } + } + if !types.is_empty() { + shared_account_ids.push(*shared_account_id); + shared_accounts_map + .entry(*shared_account_id) + .or_insert_with(AHashMap::new) + .insert(account_id, types); + } + } + shared_accounts.insert(account_id, shared_account_ids); + } + Event::Subscribe { + id, + account_id, + types, + tx, + } => { + subscribers + .entry(account_id) + .or_insert_with(AHashMap::default) + .insert( + u32::MAX - id, + Subscriber { + types, + subscription: SubscriberType::Ipc { tx }, + }, + ); + } + Event::Publish { state_change } => { + if let Some(shared_accounts) = shared_accounts_map.get(&state_change.account_id) + { + let current_time = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + let mut push_ids = Vec::new(); + + for (owner_account_id, allowed_types) in shared_accounts { + if let Some(subscribers) = subscribers.get(owner_account_id) { + for (subscriber_id, subscriber) in subscribers { + let mut types = Vec::with_capacity(state_change.types.len()); + for (state_type, change_id) in &state_change.types { + if subscriber.types.contains(*state_type) + && allowed_types.contains(*state_type) + { + types.push((*state_type, *change_id)); + } + } + if !types.is_empty() { + match &subscriber.subscription { + SubscriberType::Ipc { tx } if !tx.is_closed() => { + let subscriber_tx = tx.clone(); + let state_change = state_change.clone(); + + tokio::spawn(async move { + // Timeout after 500ms in case there is a blocked client + if let Err(err) = subscriber_tx + .send_timeout( + StateChange { + account_id: state_change.account_id, + types, + }, + Duration::from_millis(SEND_TIMEOUT_MS), + ) + .await + { + tracing::debug!( + "Error sending state change to subscriber: {}", + err + ); + } + }); + } + SubscriberType::Push { expires } + if expires > ¤t_time => + { + push_ids.push(Id::from_parts( + *owner_account_id, + *subscriber_id, + )); + } + _ => { + purge_needed = true; + } + } + } + } + } + } + + if !push_ids.is_empty() { + if let Err(err) = push_tx + .send(crate::push::Event::Push { + ids: push_ids, + state_change, + }) + .await + { + tracing::debug!("Error sending push updates: {}", err); + } + } + } + } + Event::UpdateSubscriptions { + account_id, + subscriptions, + } => { + let mut updated_ids = Vec::with_capacity(subscriptions.len()); + let mut push_updates = Vec::with_capacity(subscriptions.len()); + + if let Some(subscribers) = subscribers.get_mut(&account_id) { + let mut remove_ids = Vec::new(); + + for subscriber_id in subscribers.keys() { + #[allow(clippy::match_like_matches_macro)] + if (*subscriber_id < u32::MAX / 2) + && !subscriptions.iter().any(|s| match s { + UpdateSubscription::Verified( + crate::push::PushSubscription { id, .. }, + ) if id == subscriber_id => true, + _ => false, + }) + { + remove_ids.push(*subscriber_id); + } + } + + for remove_id in remove_ids { + push_updates.push(crate::push::PushUpdate::Unregister { + id: Id::from_parts(account_id, remove_id), + }); + subscribers.remove(&remove_id); + } + } + + for subscription in subscriptions { + match subscription { + UpdateSubscription::Unverified { + id, + url, + code, + keys, + } => { + push_updates.push(crate::push::PushUpdate::Verify { + id, + account_id, + url, + code, + keys, + }); + } + UpdateSubscription::Verified(verified) => { + updated_ids.push(verified.id); + subscribers + .entry(account_id) + .or_insert_with(AHashMap::default) + .insert( + verified.id, + Subscriber { + types: verified.types, + subscription: SubscriberType::Push { + expires: verified.expires, + }, + }, + ); + + push_updates.push(crate::push::PushUpdate::Register { + id: Id::from_parts(account_id, verified.id), + url: verified.url, + keys: verified.keys, + }); + } + } + } + + if !push_updates.is_empty() { + if let Err(err) = push_tx + .send(crate::push::Event::Update { + updates: push_updates, + }) + .await + { + tracing::debug!("Error sending push updates: {}", err); + } + } + } + } + + if purge_needed { + let mut remove_account_ids = Vec::new(); + let current_time = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0); + + for (account_id, subscriber_map) in &mut subscribers { + let mut remove_subscription_ids = Vec::new(); + for (id, subscriber) in subscriber_map.iter() { + if !subscriber.is_valid(current_time) { + remove_subscription_ids.push(*id); + } + } + if !remove_subscription_ids.is_empty() { + if remove_subscription_ids.len() < subscriber_map.len() { + for remove_subscription_id in remove_subscription_ids { + subscriber_map.remove(&remove_subscription_id); + } + } else { + remove_account_ids.push(*account_id); + } + } + } + + for remove_account_id in remove_account_ids { + subscribers.remove(&remove_account_id); + } + + last_purge = Instant::now(); + } + } + }); +} + +impl JMAP { + pub async fn subscribe_state_manager( + &self, + id: u32, + account_id: u32, + types: Bitmap, + ) -> Option> { + let (change_tx, change_rx) = mpsc::channel::(IPC_CHANNEL_BUFFER); + let state_tx = self.state_tx.clone(); + + for event in [ + Event::UpdateSharedAccounts { account_id }, + Event::Subscribe { + id, + account_id, + types, + tx: change_tx, + }, + ] { + if let Err(err) = state_tx.send(event).await { + tracing::error!( + "Channel failure while subscribing to state manager: {}", + err + ); + return None; + } + } + + change_rx.into() + } + + pub async fn broadcast_state_change(&self, state_change: StateChange) -> bool { + match self + .state_tx + .clone() + .send(Event::Publish { state_change }) + .await + { + Ok(_) => true, + Err(err) => { + tracing::error!("Channel failure while publishing state change: {}", err); + false + } + } + } + + pub async fn update_push_subscriptions(&self, account_id: u32) -> bool { + let push_subs = match self.fetch_push_subscriptions(account_id).await { + Ok(push_subs) => push_subs, + Err(err) => { + tracing::error!(context = "update_push_subscriptions", + event = "error", + reason = %err, + "Error fetching push subscriptions."); + return false; + } + }; + + let state_tx = self.state_tx.clone(); + for event in [Event::UpdateSharedAccounts { account_id }, push_subs] { + if let Err(err) = state_tx.send(event).await { + tracing::error!("Channel failure while publishing state change: {}", err); + return false; + } + } + + true + } +} diff --git a/crates/jmap/src/thread/get.rs b/crates/jmap/src/thread/get.rs index e0a9f7ee..7445f9d5 100644 --- a/crates/jmap/src/thread/get.rs +++ b/crates/jmap/src/thread/get.rs @@ -30,7 +30,7 @@ impl JMAP { .map_or(true, |p| p.unwrap().contains(&Property::EmailIds)); let mut response = GetResponse { account_id: Some(request.account_id), - state: self.get_state(account_id, Collection::Thread).await?, + state: self.get_state(account_id, Collection::Thread).await?.into(), list: Vec::with_capacity(ids.len()), not_found: vec![], }; diff --git a/tests/Cargo.toml b/tests/Cargo.toml index 8cdb1a55..a60598bb 100644 --- a/tests/Cargo.toml +++ b/tests/Cargo.toml @@ -8,6 +8,7 @@ resolver = "2" store = { path = "../crates/store", features = ["test_mode"] } jmap = { path = "../crates/jmap", features = ["test_mode"] } jmap_proto = { path = "../crates/jmap-proto" } +mail-send = { git = "https://github.com/stalwartlabs/mail-send" } utils = { path = "../crates/utils" } #jmap-client = { git = "https://github.com/stalwartlabs/jmap-client", features = ["websockets", "debug", "async"] } jmap-client = { path = "/home/vagrant/code/jmap-client", features = ["websockets", "debug", "async"] } @@ -22,4 +23,8 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"]} bytes = "1.4.0" - +futures = "0.3" +ece = "2.2" +hyper = { version = "1.0.0-rc.3", features = ["server", "http1", "http2"] } +http-body-util = "0.1.0-rc.2" +base64 = "0.21" diff --git a/tests/src/jmap/auth_acl.rs b/tests/src/jmap/auth_acl.rs index fdbca2a6..427bf68b 100644 --- a/tests/src/jmap/auth_acl.rs +++ b/tests/src/jmap/auth_acl.rs @@ -1,21 +1,16 @@ -use std::{sync::Arc, time::Duration}; +use std::sync::Arc; use jmap::{ mailbox::{INBOX_ID, TRASH_ID}, JMAP, }; use jmap_client::{ - client::{Client, Credentials}, + client::Client, core::{ error::{MethodError, MethodErrorType}, set::{SetError, SetErrorType}, }, - email::{ - self, - import::EmailImportResponse, - query::{Comparator, Filter}, - Property, - }, + email::{self, import::EmailImportResponse, query::Filter, Property}, mailbox::{self, Role}, principal::ACL, }; @@ -28,56 +23,17 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Create a group and three test accounts let inbox_id = Id::new(INBOX_ID as u64).to_string(); let trash_id = Id::new(TRASH_ID as u64).to_string(); - const JOHN_ID: u64 = 1; - const JANE_ID: u64 = 2; - const BILL_ID: u64 = 3; - const SALES_ID: u64 = 4; - let john_id = Id::from(JOHN_ID).to_string(); - let jane_id = Id::from(JANE_ID).to_string(); - let bill_id = Id::from(BILL_ID).to_string(); - let sales_id = Id::from(SALES_ID).to_string(); - for (login, secret, name) in [ - ("jdoe@example.com", "12345", "John Doe"), - ("jane.smith@example.com", "abcde", "Jane Smith"), - ("bill@example.com", "098765", "Bill Foobar"), - ("sales@example.com", "Sales Group", ""), - ] { - assert!( - server - .auth_db - .execute( - "INSERT OR REPLACE INTO users (login, secret, name) VALUES (?, ?, ?)", - vec![login.to_string(), secret.to_string(), name.to_string()].into_iter() - ) - .await - ); - } + let john_id = test_account_create(&server, "jdoe@example.com", "12345", "John Doe").await; + let jane_id = + test_account_create(&server, "jane.smith@example.com", "abcde", "Jane Smith").await; + let bill_id = test_account_create(&server, "bill@example.com", "098765", "Bill Foobar").await; + let sales_id = test_account_create(&server, "sales@example.com", "", "Sales Group").await; // Authenticate all accounts - let mut john_client = Client::new() - .credentials(Credentials::basic("jdoe@example.com", "12345")) - .timeout(Duration::from_secs(60)) - .accept_invalid_certs(true) - .connect("https://127.0.0.1:8899") - .await - .unwrap(); - - let mut jane_client = Client::new() - .credentials(Credentials::basic("jane.smith@example.com", "abcde")) - .timeout(Duration::from_secs(60)) - .accept_invalid_certs(true) - .connect("https://127.0.0.1:8899") - .await - .unwrap(); - - let mut bill_client = Client::new() - .credentials(Credentials::basic("bill@example.com", "098765")) - .timeout(Duration::from_secs(60)) - .accept_invalid_certs(true) - .connect("https://127.0.0.1:8899") - .await - .unwrap(); + let mut john_client = test_account_login("jdoe@example.com", "12345").await; + let mut jane_client = test_account_login("jane.smith@example.com", "abcde").await; + let mut bill_client = test_account_login("bill@example.com", "098765").await; // Insert two emails in each account let mut email_ids = AHashMap::default(); @@ -92,7 +48,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { for (mailbox_id, mailbox_name) in [(&inbox_id, "inbox"), (&trash_id, "trash")] { ids.push( client - .set_default_account_id(account_id) + .set_default_account_id(account_id.to_string()) .email_import( format!( concat!( @@ -133,7 +89,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_get( email_ids.get("jane").unwrap().first().unwrap(), [Property::Subject].into(), @@ -142,13 +98,13 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_get(&inbox_id, None::>) .await, ); assert_forbidden( john_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .email_get( email_ids.get("sales").unwrap().first().unwrap(), [Property::Subject].into(), @@ -157,13 +113,13 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_forbidden( john_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .mailbox_get(&inbox_id, None::>) .await, ); assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_query(None::, None::>) .await, ); @@ -177,7 +133,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // John shoud have ReadItems access to Inbox assert_eq!( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_get( email_ids.get("jane").unwrap().first().unwrap(), [Property::Subject].into(), @@ -191,7 +147,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_eq!( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_query(None::, None::>) .await .unwrap() @@ -202,13 +158,17 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // John's session resource should contain Jane's account details john_client.refresh_session().await.unwrap(); assert_eq!( - john_client.session().account(&jane_id).unwrap().name(), + john_client + .session() + .account(&jane_id.to_string()) + .unwrap() + .name(), "jane.smith@example.com" ); // John should not have access to emails in Jane's Trash folder assert!(john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_get( email_ids.get("jane").unwrap().last().unwrap(), [Property::Subject].into(), @@ -228,8 +188,8 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .unwrap() .take_blob_id(); john_client - .set_default_account_id(&john_id) - .blob_copy(&jane_id, &blob_id) + .set_default_account_id(&john_id.to_string()) + .blob_copy(&jane_id.to_string(), &blob_id) .await .unwrap(); let blob_id = jane_client @@ -243,15 +203,15 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .take_blob_id(); assert_forbidden( john_client - .set_default_account_id(&john_id) - .blob_copy(&jane_id, &blob_id) + .set_default_account_id(&john_id.to_string()) + .blob_copy(&jane_id.to_string(), &blob_id) .await, ); // John only has ReadItems access to Inbox but no Read access assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_get(&inbox_id, [mailbox::Property::MyRights].into()) .await, ); @@ -261,7 +221,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .unwrap(); assert_eq!( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_get(&inbox_id, [mailbox::Property::MyRights].into()) .await .unwrap() @@ -274,9 +234,9 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try to add items using import and copy let blob_id = john_client - .set_default_account_id(&john_id) + .set_default_account_id(&john_id.to_string()) .upload( - Some(&john_id), + Some(&john_id.to_string()), concat!( "From: acl_test@example.com\r\n", "To: jane.smith@example.com\r\n", @@ -291,7 +251,9 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap() .take_blob_id(); - let mut request = john_client.set_default_account_id(&jane_id).build(); + let mut request = john_client + .set_default_account_id(&jane_id.to_string()) + .build(); let email_id = request .import_email() .email(&blob_id) @@ -306,9 +268,9 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_copy( - &john_id, + &john_id.to_string(), email_ids.get("john").unwrap().last().unwrap(), [&inbox_id], None::>, @@ -327,7 +289,9 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); - let mut request = john_client.set_default_account_id(&jane_id).build(); + let mut request = john_client + .set_default_account_id(&jane_id.to_string()) + .build(); let email_id = request .import_email() .email(&blob_id) @@ -341,9 +305,9 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .unwrap() .take_id(); let email_id_2 = john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_copy( - &john_id, + &john_id.to_string(), email_ids.get("john").unwrap().last().unwrap(), [&inbox_id], None::>, @@ -377,7 +341,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try removing items assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_destroy(&email_id) .await, ); @@ -390,7 +354,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_destroy(&email_id) .await .unwrap(); @@ -398,7 +362,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try to set keywords assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_set_keyword(&email_id_2, "$seen", true) .await, ); @@ -417,12 +381,12 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_set_keyword(&email_id_2, "$seen", true) .await .unwrap(); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_set_keyword(&email_id_2, "my-keyword", true) .await .unwrap(); @@ -430,7 +394,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try to create a child assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_create("John's mailbox", None::<&str>, Role::None) .await, ); @@ -450,7 +414,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); let mailbox_id = john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_create("John's mailbox", Some(&inbox_id), Role::None) .await .unwrap() @@ -459,7 +423,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try renaming a mailbox assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_rename(&mailbox_id, "John's private mailbox") .await, ); @@ -472,7 +436,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_rename(&mailbox_id, "John's private mailbox") .await .unwrap(); @@ -480,7 +444,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try moving a message assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_set_mailbox(&email_id_2, &mailbox_id, true) .await, ); @@ -493,7 +457,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_set_mailbox(&email_id_2, &mailbox_id, true) .await .unwrap(); @@ -501,7 +465,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try deleting a mailbox assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_destroy(&mailbox_id, true) .await, ); @@ -521,7 +485,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .unwrap(); assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_destroy(&mailbox_id, true) .await, ); @@ -541,7 +505,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await .unwrap(); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_destroy(&mailbox_id, true) .await .unwrap(); @@ -549,13 +513,13 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Try changing ACL assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_update_acl(&inbox_id, "bill@example.com", [ACL::Read, ACL::ReadItems]) .await, ); assert_forbidden( bill_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_query(None::, None::>) .await, ); @@ -578,7 +542,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .unwrap(); assert_eq!( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_get(&inbox_id, [mailbox::Property::MyRights].into()) .await .unwrap() @@ -596,13 +560,13 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ] ); john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .mailbox_update_acl(&inbox_id, "bill@example.com", [ACL::Read, ACL::ReadItems]) .await .unwrap(); assert_eq!( bill_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_query( None::, vec![email::query::Comparator::subject()].into() @@ -623,7 +587,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .unwrap(); assert_forbidden( john_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_get( email_ids.get("jane").unwrap().first().unwrap(), [Property::Subject].into(), @@ -631,10 +595,13 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .await, ); john_client.refresh_session().await.unwrap(); - assert!(john_client.session().account(&jane_id).is_none()); + assert!(john_client + .session() + .account(&jane_id.to_string()) + .is_none()); assert_eq!( bill_client - .set_default_account_id(&jane_id) + .set_default_account_id(&jane_id.to_string()) .email_get( email_ids.get("jane").unwrap().first().unwrap(), [Property::Subject].into(), @@ -648,14 +615,15 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); // Add John and Jane to the Sales group - for id in [JANE_ID, JOHN_ID] { + for id in [jane_id.id(), john_id.id()] { assert!( server .auth_db .execute( &format!( "INSERT INTO groups (uid, gid) VALUES ({}, {})", - id, SALES_ID + id, + sales_id.id() ), Vec::::new().into_iter(), ) @@ -667,25 +635,36 @@ pub async fn test(server: Arc, admin_client: &mut Client) { jane_client.refresh_session().await.unwrap(); bill_client.refresh_session().await.unwrap(); assert_eq!( - john_client.session().account(&sales_id).unwrap().name(), + john_client + .session() + .account(&sales_id.to_string()) + .unwrap() + .name(), "sales@example.com" ); assert!(!john_client .session() - .account(&sales_id) + .account(&sales_id.to_string()) .unwrap() .is_personal()); assert_eq!( - jane_client.session().account(&sales_id).unwrap().name(), + jane_client + .session() + .account(&sales_id.to_string()) + .unwrap() + .name(), "sales@example.com" ); - assert!(bill_client.session().account(&sales_id).is_none()); + assert!(bill_client + .session() + .account(&sales_id.to_string()) + .is_none()); // Insert a message in Sales's inbox let blob_id = john_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .upload( - Some(&sales_id), + Some(&sales_id.to_string()), concat!( "From: acl_test@example.com\r\n", "To: sales@example.com\r\n", @@ -717,7 +696,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { // Both Jane and John should be able to see this message, but not Bill assert_eq!( john_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .email_get(&email_id, [Property::Subject].into(),) .await .unwrap() @@ -728,7 +707,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_eq!( jane_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .email_get(&email_id, [Property::Subject].into(),) .await .unwrap() @@ -739,7 +718,7 @@ pub async fn test(server: Arc, admin_client: &mut Client) { ); assert_forbidden( bill_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .email_get(&email_id, [Property::Subject].into()) .await, ); @@ -751,7 +730,8 @@ pub async fn test(server: Arc, admin_client: &mut Client) { .execute( &format!( "DELETE FROM groups WHERE uid = {} AND gid ={}", - JOHN_ID, SALES_ID + john_id.id(), + sales_id.id() ), Vec::::new().into_iter(), ) @@ -760,31 +740,22 @@ pub async fn test(server: Arc, admin_client: &mut Client) { server.sessions.lock().clear(); assert_forbidden( john_client - .set_default_account_id(&sales_id) + .set_default_account_id(&sales_id.to_string()) .email_get(&email_id, [Property::Subject].into()) .await, ); - let coco = "fd"; - // Check that Jane's id is not assigned to new accounts before the - // purge has taken place. - /*server.store.id_assigner.invalidate_all(); - let tom_id = admin_client - .individual_create("tom@example.com", "098765", "Tom Foobar") - .await - .unwrap() - .take_id(); - assert_ne!(tom_id, jane_id); - - // Destroy test accounts - for principal_id in [tom_id, john_id, bill_id, sales_id, domain_id] { - admin_client.principal_destroy(&principal_id).await.unwrap(); + // Destroy test account data + for id in [john_id, bill_id, jane_id, sales_id] { + admin_client.set_default_account_id(&id.to_string()); + destroy_all_mailboxes(admin_client).await; } - server.store.principal_purge().unwrap(); - server.store.assert_is_empty();*/ + server.store.assert_is_empty().await; } use std::fmt::Debug; + +use crate::jmap::{mailbox::destroy_all_mailboxes, test_account_create, test_account_login}; pub fn assert_forbidden(result: Result) { if !matches!( result, diff --git a/tests/src/jmap/auth_limits.rs b/tests/src/jmap/auth_limits.rs index f58279ad..05782a66 100644 --- a/tests/src/jmap/auth_limits.rs +++ b/tests/src/jmap/auth_limits.rs @@ -5,30 +5,19 @@ use jmap_client::{ client::{Client, Credentials}, mailbox::{self}, }; -use jmap_proto::types::id::Id; -pub async fn test(server: Arc, _client: &mut Client) { +use crate::jmap::{mailbox::destroy_all_mailboxes, test_account_create}; + +pub async fn test(server: Arc, admin_client: &mut Client) { println!("Running Authorization tests..."); // Create test account - assert!( - server - .auth_db - .execute( - "INSERT OR REPLACE INTO users (login, secret, name) VALUES (?, ?, ?)", - vec![ - "jdoe@example.com".to_string(), - "12345".to_string(), - "John Doe".to_string() - ] - .into_iter() - ) - .await - ); - let account_id = Id::from(1u64).to_string(); + let account_id = test_account_create(&server, "jdoe@example.com", "12345", "John Doe") + .await + .to_string(); // Wait for rate limit to be restored after running previous tests - //tokio::time::sleep(Duration::from_secs(1)).await; + tokio::time::sleep(Duration::from_secs(1)).await; // Incorrect passwords should be rejected with a 401 error assert!(matches!( @@ -164,13 +153,7 @@ pub async fn test(server: Arc, _client: &mut Client) { Err(jmap_client::Error::Problem(err)) if err.status() == Some(400))); // Destroy test accounts - let implement = "true"; - /*admin_client - .set_default_account_id(Id::new(SUPERUSER_ID as u64)) - .principal_destroy(&account_id) - .await - .unwrap(); - admin_client.principal_destroy(&domain_id).await.unwrap(); - server.store.principal_purge().unwrap(); - server.store.assert_is_empty();*/ + admin_client.set_default_account_id(&account_id); + destroy_all_mailboxes(admin_client).await; + server.store.assert_is_empty().await; } diff --git a/tests/src/jmap/auth_oauth.rs b/tests/src/jmap/auth_oauth.rs index 2fcb1502..bb73a82c 100644 --- a/tests/src/jmap/auth_oauth.rs +++ b/tests/src/jmap/auth_oauth.rs @@ -9,30 +9,19 @@ use jmap_client::{ client::{Client, Credentials}, mailbox::query::Filter, }; -use jmap_proto::types::id::Id; use reqwest::{header, redirect::Policy}; use serde::de::DeserializeOwned; use store::ahash::AHashMap; +use crate::jmap::{mailbox::destroy_all_mailboxes, test_account_create}; + pub async fn test(server: Arc, _client: &mut Client) { println!("Running OAuth tests..."); // Create test account - assert!( - server - .auth_db - .execute( - "INSERT OR REPLACE INTO users (login, secret, name) VALUES (?, ?, ?)", - vec![ - "jdoe@example.com".to_string(), - "abcde".to_string(), - "John Doe".to_string() - ] - .into_iter() - ) - .await - ); - let john_id = Id::from(1u64).to_string(); + let john_id = test_account_create(&server, "jdoe@example.com", "abcde", "John Doe") + .await + .to_string(); // Obtain OAuth metadata let metadata: OAuthMetadata = @@ -216,7 +205,7 @@ pub async fn test(server: Arc, _client: &mut Client) { ); // Connect to account using token and attempt to search - let john_client = Client::new() + let mut john_client = Client::new() .credentials(Credentials::bearer(&token)) .accept_invalid_certs(true) .connect("https://127.0.0.1:8899") @@ -281,12 +270,8 @@ pub async fn test(server: Arc, _client: &mut Client) { ); // Destroy test accounts - let cleanup = "true"; - /*for principal_id in [john_id, domain_id] { - admin_client.principal_destroy(&principal_id).await.unwrap(); - } - server.store.principal_purge().unwrap(); - server.store.assert_is_empty();*/ + destroy_all_mailboxes(&mut john_client).await; + server.store.assert_is_empty().await; } async fn post_bytes(url: &str, params: &AHashMap) -> Bytes { diff --git a/tests/src/jmap/event_source.rs b/tests/src/jmap/event_source.rs new file mode 100644 index 00000000..8feaf1e9 --- /dev/null +++ b/tests/src/jmap/event_source.rs @@ -0,0 +1,135 @@ +use std::{sync::Arc, time::Duration}; + +use futures::StreamExt; +use jmap::JMAP; +use jmap_client::{ + client::{Client, Credentials}, + event_source::Changes, + mailbox::Role, + TypeState, +}; +use jmap_proto::types::id::Id; +use store::ahash::AHashSet; +use tokio::sync::mpsc; + +use crate::jmap::{mailbox::destroy_all_mailboxes, test_account_create, test_account_login}; + +pub async fn test(server: Arc, admin_client: &mut Client) { + println!("Running EventSource tests..."); + + // Create test account + test_account_create(&server, "jdoe@example.com", "12345", "John Doe").await; + let mut client = test_account_login("jdoe@example.com", "12345").await; + + let mut changes = client + .event_source(None::>, false, 1.into(), None) + .await + .unwrap(); + + let (event_tx, mut event_rx) = mpsc::channel::(100); + + tokio::spawn(async move { + while let Some(change) = changes.next().await { + if let Err(_err) = event_tx.send(change.unwrap()).await { + //println!("Error sending event: {}", _err); + break; + } + } + }); + + assert_ping(&mut event_rx).await; + + // Create mailbox and expect state change + let mailbox_id = client + .set_default_account_id(Id::new(1).to_string()) + .mailbox_create("EventSource Test", None::, Role::None) + .await + .unwrap() + .take_id(); + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + + // Multiple changes should be grouped and delivered in intervals + for num in 0..5 { + client + .mailbox_update_sort_order(&mailbox_id, num) + .await + .unwrap(); + } + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + assert_ping(&mut event_rx).await; // Pings are only received in cfg(test) + + // Ingest email and expect state change + let implement = "true"; + /*let mut lmtp = SmtpConnection::connect().await; + lmtp.ingest( + "bill@example.com", + &["jdoe@example.com"], + concat!( + "From: bill@example.com\r\n", + "To: jdoe@example.com\r\n", + "Subject: TPS Report\r\n", + "\r\n", + "I'm going to need those TPS reports ASAP. ", + "So, if you could do that, that'd be great." + ), + ) + .await; + lmtp.quit().await; + + assert_state( + &mut event_rx, + &[ + TypeState::EmailDelivery, + TypeState::Email, + TypeState::Thread, + TypeState::Mailbox, + ], + ) + .await; + assert_ping(&mut event_rx).await;*/ + + // Destroy mailbox + client.mailbox_destroy(&mailbox_id, true).await.unwrap(); + + /*assert_state( + &mut event_rx, + &[TypeState::Email, TypeState::Thread, TypeState::Mailbox], + ) + .await;*/ + let fix = "true"; + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + assert_ping(&mut event_rx).await; + assert_ping(&mut event_rx).await; + + destroy_all_mailboxes(admin_client).await; + server.store.assert_is_empty().await; +} + +async fn assert_state(event_rx: &mut mpsc::Receiver, state: &[TypeState]) { + match tokio::time::timeout(Duration::from_millis(700), event_rx.recv()).await { + Ok(Some(changes)) => { + assert_eq!( + changes + .changes(&Id::new(1).to_string()) + .unwrap() + .map(|x| x.0) + .collect::>(), + state.iter().collect::>() + ); + } + result => { + panic!("Timeout waiting for event {:?}: {:?}", state, result); + } + } +} + +async fn assert_ping(event_rx: &mut mpsc::Receiver) { + match tokio::time::timeout(Duration::from_millis(1100), event_rx.recv()).await { + Ok(Some(changes)) => { + assert!(changes.changes("ping").is_some(),); + } + _ => { + panic!("Did not receive ping."); + } + } +} diff --git a/tests/src/jmap/mailbox.rs b/tests/src/jmap/mailbox.rs index ad22582f..c56802ab 100644 --- a/tests/src/jmap/mailbox.rs +++ b/tests/src/jmap/mailbox.rs @@ -603,13 +603,7 @@ pub async fn test(server: Arc, client: &mut Client) { ["inbox", "sent", "spam"] ); - let mut request = client.build(); - request.query_mailbox().arguments().sort_as_tree(true); - let mut ids = request.send_query_mailbox().await.unwrap().take_ids(); - ids.reverse(); - for id in ids { - client.mailbox_destroy(&id, true).await.unwrap(); - } + destroy_all_mailboxes(client).await; server.store.assert_is_empty().await; } @@ -659,6 +653,16 @@ fn build_create_query( } } +pub async fn destroy_all_mailboxes(client: &mut Client) { + let mut request = client.build(); + request.query_mailbox().arguments().sort_as_tree(true); + let mut ids = request.send_query_mailbox().await.unwrap().take_ids(); + ids.reverse(); + for id in ids { + client.mailbox_destroy(&id, true).await.unwrap(); + } +} + #[derive(Serialize, Deserialize)] struct TestMailbox { id: String, diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index 0e54cfb1..1a807664 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -18,7 +18,9 @@ pub mod email_query; pub mod email_query_changes; pub mod email_search_snippet; pub mod email_set; +pub mod event_source; pub mod mailbox; +pub mod push_subscription; pub mod thread_get; pub mod thread_merge; @@ -63,6 +65,16 @@ account.rate = '100/1m' authentication.rate = '100/1m' anonymous.rate = '1000/1m' +[jmap.event-source] +throttle = '500ms' + +[jmap.web-sockets] +throttle = '500ms' + +[jmap.push] +throttle = '500ms' +attempts.interval = '500ms' + [jmap.auth.database] type = 'sql' address = 'sqlite::memory:' @@ -108,7 +120,9 @@ pub async fn jmap_tests() { //mailbox::test(params.server.clone(), &mut params.client).await; //auth_acl::test(params.server.clone(), &mut params.client).await; //auth_limits::test(params.server.clone(), &mut params.client).await; - auth_oauth::test(params.server.clone(), &mut params.client).await; + //auth_oauth::test(params.server.clone(), &mut params.client).await; + //event_source::test(params.server.clone(), &mut params.client).await; + push_subscription::test(params.server.clone(), &mut params.client).await; if delete { params.temp_dir.delete(); @@ -133,7 +147,7 @@ async fn init_jmap_tests(delete_if_exists: bool) -> JMAPTest { let servers = settings.parse_servers().unwrap(); // Start JMAP server - let manager = SessionManager::from(JMAP::new(&settings).await); + let manager = SessionManager::from(JMAP::init(&settings).await); let shutdown_tx = servers.spawn(&settings, |server, shutdown_rx| { server.spawn(manager.clone(), shutdown_rx); }); @@ -228,3 +242,25 @@ pub fn replace_blob_ids(string: String) -> String { string } } + +pub async fn test_account_create(jmap: &JMAP, login: &str, secret: &str, name: &str) -> Id { + assert!( + jmap.auth_db + .execute( + "INSERT OR REPLACE INTO users (login, secret, name) VALUES (?, ?, ?)", + vec![login.to_string(), secret.to_string(), name.to_string()].into_iter() + ) + .await + ); + Id::new(jmap.get_account_id(login).await.unwrap() as u64) +} + +pub async fn test_account_login(login: &str, secret: &str) -> Client { + Client::new() + .credentials(Credentials::basic(login, secret)) + .timeout(Duration::from_secs(5)) + .accept_invalid_certs(true) + .connect("https://127.0.0.1:8899") + .await + .unwrap() +} diff --git a/tests/src/jmap/push_subscription.rs b/tests/src/jmap/push_subscription.rs new file mode 100644 index 00000000..6c6a8025 --- /dev/null +++ b/tests/src/jmap/push_subscription.rs @@ -0,0 +1,347 @@ +use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + time::Duration, +}; + +use base64::{engine::general_purpose, Engine}; +use ece::EcKeyComponents; + +use hyper::{body, server::conn::http1, service::service_fn, StatusCode}; +use jmap::{ + api::{ + http::{fetch_body, ToHttpResponse}, + HtmlResponse, StateChangeResponse, + }, + JMAP, +}; +use jmap_client::{client::Client, mailbox::Role, push_subscription::Keys}; +use jmap_proto::types::{id::Id, type_state::TypeState}; +use reqwest::header::CONTENT_ENCODING; +use store::ahash::AHashSet; +use tokio::{net::TcpStream, sync::mpsc}; +use utils::listener::SessionData; + +use crate::{ + add_test_certs, + jmap::{mailbox::destroy_all_mailboxes, test_account_create, test_account_login}, +}; + +const SERVER: &str = " +[server] +hostname = 'jmap-push.example.org' + +[server.listener.jmap] +bind = ['127.0.0.1:9000'] +url = 'https://127.0.0.1:9000' +protocol = 'jmap' + +[server.socket] +reuse-addr = true + +[server.tls] +enable = true +implicit = false +certificate = 'default' + +[certificate.default] +cert = 'file://{CERT}' +private-key = 'file://{PK}' +"; + +pub async fn test(server: Arc, admin_client: &mut Client) { + println!("Running Push Subscription tests..."); + + // Create test account + let account_id = test_account_create(&server, "jdoe@example.com", "12345", "John Doe").await; + admin_client.set_default_account_id(account_id); + let mut client = test_account_login("jdoe@example.com", "12345").await; + + // Create channels + let (event_tx, mut event_rx) = mpsc::channel::(100); + + // Create subscription keys + let (keypair, auth_secret) = ece::generate_keypair_and_auth_secret().unwrap(); + let pubkey = keypair.pub_as_raw().unwrap(); + let keys = Keys::new(&pubkey, &auth_secret); + + let push_server = Arc::new(PushServer { + keypair: keypair.raw_components().unwrap(), + auth_secret: auth_secret.to_vec(), + tx: event_tx, + fail_requests: false.into(), + }); + + // Start mock push server + let settings = utils::config::Config::parse(&add_test_certs(SERVER)).unwrap(); + let servers = settings.parse_servers().unwrap(); + + // Start JMAP server + let manager = SessionManager::from(push_server.clone()); + let _shutdown_tx = servers.spawn(&settings, |server, shutdown_rx| { + server.spawn(manager.clone(), shutdown_rx); + }); + + // Register push notification (no encryption) + let push_id = client + .push_subscription_create("123", "https://127.0.0.1:9000/push", None) + .await + .unwrap() + .take_id(); + + // Expect push verification + let verification = expect_push(&mut event_rx).await.unwrap_verification(); + assert_eq!(verification.push_subscription_id, push_id); + + // Update verification code + client + .push_subscription_verify(&push_id, verification.verification_code) + .await + .unwrap(); + + // Create a mailbox and expect a state change + let mailbox_id = client + .set_default_account_id(Id::new(1).to_string()) + .mailbox_create("PushSubscription Test", None::, Role::None) + .await + .unwrap() + .take_id(); + + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + + // Receive states just for the requested types + client + .push_subscription_update_types(&push_id, [jmap_client::TypeState::Email].into()) + .await + .unwrap(); + client + .mailbox_update_sort_order(&mailbox_id, 123) + .await + .unwrap(); + expect_nothing(&mut event_rx).await; + + // Destroy subscription + client.push_subscription_destroy(&push_id).await.unwrap(); + + // Only one verification per minute is allowed + let push_id = client + .push_subscription_create("invalid", "https://127.0.0.1:9000/push", None) + .await + .unwrap() + .take_id(); + expect_nothing(&mut event_rx).await; + client.push_subscription_destroy(&push_id).await.unwrap(); + + // Register push notification (with encryption) + let push_id = client + .push_subscription_create( + "123", + "https://127.0.0.1:9000/push?skip_checks=true", // skip_checks only works in cfg(test) + keys.into(), + ) + .await + .unwrap() + .take_id(); + + // Expect push verification + let verification = expect_push(&mut event_rx).await.unwrap_verification(); + assert_eq!(verification.push_subscription_id, push_id); + + // Update verification code + client + .push_subscription_verify(&push_id, verification.verification_code) + .await + .unwrap(); + + // Failed deliveries should be re-attempted + push_server.fail_requests.store(true, Ordering::Relaxed); + client + .mailbox_update_sort_order(&mailbox_id, 101) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + push_server.fail_requests.store(false, Ordering::Relaxed); + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + + // Make a mailbox change and expect state change + client + .mailbox_rename(&mailbox_id, "My Mailbox") + .await + .unwrap(); + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + //expect_nothing(&mut event_rx).await; + + // Multiple change updates should be grouped and pushed in intervals + for num in 0..25 { + client + .mailbox_update_sort_order(&mailbox_id, num) + .await + .unwrap(); + } + assert_state(&mut event_rx, &[TypeState::Mailbox]).await; + expect_nothing(&mut event_rx).await; + + // Destroy mailbox + client.push_subscription_destroy(&push_id).await.unwrap(); + client.mailbox_destroy(&mailbox_id, true).await.unwrap(); + expect_nothing(&mut event_rx).await; + + destroy_all_mailboxes(admin_client).await; + + server.store.assert_is_empty().await; +} + +#[derive(Clone)] +pub struct SessionManager { + pub inner: Arc, +} + +impl From> for SessionManager { + fn from(inner: Arc) -> Self { + SessionManager { inner } + } +} +pub struct PushServer { + keypair: EcKeyComponents, + auth_secret: Vec, + tx: mpsc::Sender, + fail_requests: AtomicBool, +} + +#[derive(serde::Deserialize, Debug)] +#[serde(untagged)] +enum PushMessage { + StateChange(StateChangeResponse), + Verification(PushVerification), +} + +impl PushMessage { + pub fn unwrap_state_change(self) -> StateChangeResponse { + match self { + PushMessage::StateChange(state_change) => state_change, + _ => panic!("Expected StateChange"), + } + } + + pub fn unwrap_verification(self) -> PushVerification { + match self { + PushMessage::Verification(verification) => verification, + _ => panic!("Expected Verification"), + } + } +} + +#[derive(serde::Deserialize, Debug)] +enum PushVerificationType { + PushVerification, +} + +#[derive(serde::Deserialize, Debug)] +struct PushVerification { + #[serde(rename = "@type")] + _type: PushVerificationType, + #[serde(rename = "pushSubscriptionId")] + pub push_subscription_id: String, + #[serde(rename = "verificationCode")] + pub verification_code: String, +} + +impl utils::listener::SessionManager for SessionManager { + fn spawn(&self, session: SessionData) { + let push = self.inner.clone(); + + tokio::spawn(async move { + let _ = http1::Builder::new() + .keep_alive(false) + .serve_connection( + session + .instance + .tls_acceptor + .as_ref() + .unwrap() + .accept(session.stream) + .await + .unwrap(), + service_fn(|mut req: hyper::Request| { + let push = push.clone(); + + async move { + if push.fail_requests.load(Ordering::Relaxed) { + return Ok(HtmlResponse::with_status( + StatusCode::TOO_MANY_REQUESTS, + "too many requests".to_string(), + ) + .into_http_response()); + } + let is_encrypted = req + .headers() + .get(CONTENT_ENCODING) + .map_or(false, |encoding| { + encoding.to_str().unwrap() == "aes128gcm" + }); + let body = fetch_body(&mut req, 1024 * 1024).await.unwrap(); + let message = serde_json::from_slice::(&if is_encrypted { + ece::decrypt( + &push.keypair, + &push.auth_secret, + &general_purpose::URL_SAFE.decode(body).unwrap(), + ) + .unwrap() + } else { + body + }) + .unwrap(); + + //println!("Push received ({}): {:?}", is_encrypted, message); + + push.tx.send(message).await.unwrap(); + + Ok::<_, hyper::Error>( + HtmlResponse::new("ok".to_string()).into_http_response(), + ) + } + }), + ) + .await; + }); + } + + fn max_concurrent(&self) -> u64 { + 100 + } +} + +async fn expect_push(event_rx: &mut mpsc::Receiver) -> PushMessage { + match tokio::time::timeout(Duration::from_millis(1500), event_rx.recv()).await { + Ok(Some(push)) => push, + result => { + panic!("Timeout waiting for push: {:?}", result); + } + } +} + +async fn expect_nothing(event_rx: &mut mpsc::Receiver) { + match tokio::time::timeout(Duration::from_millis(1000), event_rx.recv()).await { + Err(_) => {} + message => { + panic!("Received a message when expecting nothing: {:?}", message); + } + } +} + +async fn assert_state(event_rx: &mut mpsc::Receiver, state: &[TypeState]) { + assert_eq!( + expect_push(event_rx) + .await + .unwrap_state_change() + .changed + .get(&Id::new(1)) + .unwrap() + .iter() + .map(|x| x.0) + .collect::>(), + state.iter().collect::>() + ); +}