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

Message ingestion (not tested)

Browse Source
This commit is contained in:
Mauro D
2023-05-17 17:35:36 +00:00
parent e0e8347de1
commit 4e4632571c
53 changed files with 1363 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files

151
crates/jmap/src/auth/account.rs
View File

@@ -3,17 +3,22 @@ use crate::JMAP;
use super::{AclToken, AuthDatabase, SqlDatabase};
impl JMAP {
pub async fn authenticate(&self, account: &str, secret: &str) -> Option<AclToken> {
pub async fn authenticate(&self, account: &str, secret: &str) -> Option<u32> {
let account_id = self.get_account_id(account).await?;
let account_secret = self.get_account_secret(account_id).await?;
if secret == account_secret {
self.get_acl_token(account_id).await
account_id.into()
} else {
tracing::debug!(context = "auth", event = "failed", account = account);
None
}
}
pub async fn authenticate_with_token(&self, account: &str, secret: &str) -> Option<AclToken> {
self.get_acl_token(self.authenticate(account, secret).await?)
.await
}
pub async fn get_acl_token(&self, account_id: u32) -> Option<AclToken> {
self.update_acl_token(AclToken {
primary_id: account_id,
@@ -30,7 +35,7 @@ impl JMAP {
query_secret_by_uid,
..
} => {
db.fetch_string(query_secret_by_uid, account_id as i64)
db.fetch_uid_to_string(query_secret_by_uid, account_id as i64)
.await
}
AuthDatabase::Ldap => None,
@@ -44,7 +49,7 @@ impl JMAP {
query_uid_by_login,
..
} => db
.fetch_id(query_uid_by_login, account)
.fetch_string_to_id(query_uid_by_login, account)
.await
.map(|id| id as u32),
AuthDatabase::Ldap => None,
@@ -58,7 +63,7 @@ impl JMAP {
query_gids_by_uid,
..
} => db
.fetch_ids(query_gids_by_uid, account_id as i64)
.fetch_uid_to_uids(query_gids_by_uid, account_id as i64)
.await
.into_iter()
.map(|id| id as u32)
@@ -73,14 +78,66 @@ impl JMAP {
db,
query_login_by_uid,
..
} => db.fetch_string(query_login_by_uid, account_id as i64).await,
} => {
db.fetch_uid_to_string(query_login_by_uid, account_id as i64)
.await
}
AuthDatabase::Ldap => None,
}
}
pub async fn get_uids_by_address(&self, address: &str) -> Vec<u32> {
match &self.auth_db {
AuthDatabase::Sql {
db,
query_gids_by_uid,
..
} => db
.fetch_string_to_uids(query_gids_by_uid, address)
.await
.into_iter()
.map(|id| id as u32)
.collect(),
AuthDatabase::Ldap => vec![],
}
}
pub async fn get_addresses_by_uid(&self, account_id: u32) -> Vec<String> {
match &self.auth_db {
AuthDatabase::Sql {
db,
query_addresses_by_uid,
..
} => {
db.fetch_uid_to_strings(query_addresses_by_uid, account_id as i64)
.await
}
AuthDatabase::Ldap => vec![],
}
}
pub async fn vrfy_address(&self, address: &str) -> Vec<String> {
match &self.auth_db {
AuthDatabase::Sql { db, query_vrfy, .. } => {
db.fetch_string_to_strings(query_vrfy, address).await
}
AuthDatabase::Ldap => vec![],
}
}
pub async fn expn_address(&self, address: &str) -> Vec<String> {
match &self.auth_db {
AuthDatabase::Sql { db, query_expn, .. } => {
db.fetch_string_to_strings(query_expn, address).await
}
AuthDatabase::Ldap => vec![],
}
}
}
// TODO abstract this
impl SqlDatabase {
pub async fn fetch_string(&self, query: &str, uid: i64) -> Option<String> {
pub async fn fetch_uid_to_string(&self, query: &str, uid: i64) -> Option<String> {
let result = match &self {
SqlDatabase::Postgres(pool) => {
sqlx::query_scalar::<_, String>(query)
@@ -117,7 +174,7 @@ impl SqlDatabase {
}
}
pub async fn fetch_id(&self, query: &str, param: &str) -> Option<i64> {
pub async fn fetch_string_to_id(&self, query: &str, param: &str) -> Option<i64> {
let result = match &self {
SqlDatabase::Postgres(pool) => {
sqlx::query_scalar::<_, i64>(query)
@@ -154,7 +211,7 @@ impl SqlDatabase {
}
}
pub async fn fetch_strings(&self, query: &str, uid: i64) -> Vec<String> {
pub async fn fetch_uid_to_strings(&self, query: &str, uid: i64) -> Vec<String> {
let result = match &self {
SqlDatabase::Postgres(pool) => {
sqlx::query_scalar::<_, String>(query)
@@ -191,7 +248,7 @@ impl SqlDatabase {
}
}
pub async fn fetch_ids(&self, query: &str, uid: i64) -> Vec<i64> {
pub async fn fetch_uid_to_uids(&self, query: &str, uid: i64) -> Vec<i64> {
let result = match &self {
SqlDatabase::Postgres(pool) => {
sqlx::query_scalar::<_, i64>(query)
@@ -228,6 +285,80 @@ impl SqlDatabase {
}
}
pub async fn fetch_string_to_uids(&self, query: &str, param: &str) -> Vec<i64> {
let result = match &self {
SqlDatabase::Postgres(pool) => {
sqlx::query_scalar::<_, i64>(query)
.bind(param)
.fetch_all(pool)
.await
}
SqlDatabase::MySql(pool) => {
sqlx::query_scalar::<_, i64>(query)
.bind(param)
.fetch_all(pool)
.await
}
/*SqlDatabase::MsSql(pool) => {
sqlx::query_scalar::<_, i64>(query)
.bind(param)
.fetch_all(pool)
.await
}*/
SqlDatabase::SqlLite(pool) => {
sqlx::query_scalar::<_, i64>(query)
.bind(param)
.fetch_all(pool)
.await
}
};
match result {
Ok(result) => result,
Err(err) => {
tracing::warn!(context = "sql", event = "error", query = query, reason = ?err);
vec![]
}
}
}
pub async fn fetch_string_to_strings(&self, query: &str, param: &str) -> Vec<String> {
let result = match &self {
SqlDatabase::Postgres(pool) => {
sqlx::query_scalar::<_, String>(query)
.bind(param)
.fetch_all(pool)
.await
}
SqlDatabase::MySql(pool) => {
sqlx::query_scalar::<_, String>(query)
.bind(param)
.fetch_all(pool)
.await
}
/*SqlDatabase::MsSql(pool) => {
sqlx::query_scalar::<_, String>(query)
.bind(param)
.fetch_all(pool)
.await
}*/
SqlDatabase::SqlLite(pool) => {
sqlx::query_scalar::<_, String>(query)
.bind(param)
.fetch_all(pool)
.await
}
};
match result {
Ok(result) => result,
Err(err) => {
tracing::warn!(context = "sql", event = "error", query = query, reason = ?err);
vec![]
}
}
}
pub async fn execute(&self, query: &str, params: impl Iterator<Item = String>) -> bool {
let result = match self {
SqlDatabase::Postgres(pool) => {

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

@@ -56,7 +56,7 @@ impl JMAP {
})
})
{
self.authenticate(&account, &secret).await
self.authenticate_with_token(&account, &secret).await
} else {
tracing::debug!(
context = "authenticate_headers",

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

@@ -25,6 +25,10 @@ pub enum AuthDatabase {
query_login_by_uid: String,
query_secret_by_uid: String,
query_gids_by_uid: String,
query_uids_by_address: String,
query_addresses_by_uid: String,
query_vrfy: String,
query_expn: String,
},
Ldap,
}

2
crates/jmap/src/auth/oauth/device_auth.rs
View File

@@ -148,7 +148,7 @@ impl JMAP {
{
if let (Some(email), Some(password)) = (fields.get("email"), fields.get("password"))
{
if let Some(id) = self.authenticate(email, password).await {
if let Some(id) = self.authenticate_with_token(email, password).await {
oauth
.account_id
.store(id.primary_id(), atomic::Ordering::Relaxed);

2
crates/jmap/src/auth/oauth/user_code.rs
View File

@@ -109,7 +109,7 @@ impl JMAP {
// Authenticate user
if let (Some(email), Some(password)) = (params.get("email"), params.get("password")) {
if let Some(acl_token) = self.authenticate(email, password).await {
if let Some(acl_token) = self.authenticate_with_token(email, password).await {
// Generate client code
let client_code = thread_rng()
.sample_iter(Alphanumeric)

1
crates/jmap/src/email/import.rs
View File

@@ -113,6 +113,7 @@ impl JMAP {
mailbox_ids,
email.keywords,
email.received_at.map(|r| r.into()),
false,
)
.await
{

35
crates/jmap/src/email/ingest.rs
View File

@@ -32,6 +32,7 @@ pub struct IngestedEmail {
}
impl JMAP {
#[allow(clippy::blocks_in_if_conditions)]
pub async fn email_ingest(
&self,
raw_message: &[u8],
@@ -39,6 +40,7 @@ impl JMAP {
mailbox_ids: Vec<u32>,
keywords: Vec<Keyword>,
received_at: Option<u64>,
skip_duplicates: bool,
) -> Result<IngestedEmail, MaybeError> {
// Parse message
let message = Message::parse(raw_message)
@@ -83,6 +85,39 @@ impl JMAP {
_ => (),
}
}
// Check for duplicates
if !skip_duplicates
&& !self
.store
.filter(
account_id,
Collection::Email,
references
.iter()
.map(|id| Filter::eq(Property::MessageId, *id))
.collect(),
)
.await
.map_err(|err| {
tracing::error!(
event = "error",
context = "find_duplicates",
error = ?err,
"Duplicate message search failed.");
MaybeError::Temporary
})?
.results
.is_empty()
{
return Ok(IngestedEmail {
id: Id::default(),
change_id: u64::MAX,
blob_id: BlobId::default(),
size: 0,
});
}
let thread_id = if !references.is_empty() {
self.find_or_merge_thread(account_id, subject, &references)
.await?

15
crates/jmap/src/email/set.rs
View File

@@ -705,10 +705,17 @@ impl JMAP {
// Ingest message
response.created.insert(
id,
self.email_ingest(&raw_message, account_id, mailboxes, keywords, received_at)
.await
.map_err(|_| MethodError::ServerPartialFail)?
.into(),
self.email_ingest(
&raw_message,
account_id,
mailboxes,
keywords,
received_at,
false,
)
.await
.map_err(|_| MethodError::ServerPartialFail)?
.into(),
);
}

31
crates/jmap/src/lib.rs
View File

@@ -15,7 +15,10 @@ use jmap_proto::{
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 services::{
delivery::spawn_delivery_manager,
state::{self, init_state_manager, spawn_state_manager},
};
use sqlx::{mysql::MySqlPoolOptions, postgres::PgPoolOptions, sqlite::SqlitePoolOptions};
use store::{
fts::Language,
@@ -26,7 +29,7 @@ use store::{
BitmapKey, Deserialize, Serialize, Store, ValueKey,
};
use tokio::sync::mpsc;
use utils::{config::Rate, failed, UnwrapFailure};
use utils::{config::Rate, failed, ipc::DeliveryEvent, UnwrapFailure};
pub mod api;
pub mod auth;
@@ -102,7 +105,10 @@ pub enum MaybeError {
}
impl JMAP {
pub async fn init(config: &utils::config::Config) -> Arc<Self> {
pub async fn init(
config: &utils::config::Config,
delivery_rx: mpsc::Receiver<DeliveryEvent>,
) -> Arc<Self> {
let auth_db = match config
.value_require("jmap.auth.database.type")
.failed("Invalid property")
@@ -182,6 +188,22 @@ impl JMAP {
.value_require("jmap.auth.database.query.gids-by-uid")
.failed("Invalid property")
.to_string(),
query_uids_by_address: config
.value_require("jmap.auth.database.query.uids-by-address")
.failed("Invalid property")
.to_string(),
query_addresses_by_uid: config
.value_require("jmap.auth.database.query.addresses-by-uid")
.failed("Invalid property")
.to_string(),
query_vrfy: config
.value_require("jmap.auth.database.query.vrfy")
.failed("Invalid property")
.to_string(),
query_expn: config
.value_require("jmap.auth.database.query.expn")
.failed("Invalid property")
.to_string(),
}
}
_ => failed("Invalid auth database type"),
@@ -227,6 +249,9 @@ impl JMAP {
state_tx,
});
// Spawn delivery manager
spawn_delivery_manager(jmap_server.clone(), delivery_rx);
// Spawn state manager
spawn_state_manager(jmap_server.clone(), config, state_rx);

39
crates/jmap/src/services/delivery.rs Normal file
View File

@@ -0,0 +1,39 @@
use std::sync::Arc;
use mail_send::Credentials;
use tokio::sync::mpsc;
use utils::ipc::{DeliveryEvent, Item};
use crate::JMAP;
pub fn spawn_delivery_manager(core: Arc<JMAP>, mut delivery_rx: mpsc::Receiver<DeliveryEvent>) {
tokio::spawn(async move {
while let Some(event) = delivery_rx.recv().await {
match event {
DeliveryEvent::Ingest { message, result_tx } => {
result_tx.send(core.deliver_message(message).await).ok();
}
DeliveryEvent::Lookup(lookup) => {
lookup
.result
.send(match lookup.item {
Item::IsAccount(address) => {
(!core.get_uids_by_address(&address).await.is_empty()).into()
}
Item::Authenticate(credentials) => match credentials {
Credentials::Plain { username, secret } => {
core.authenticate(&username, &secret).await.is_some()
}
_ => false,
}
.into(),
Item::Verify(address) => core.vrfy_address(&address).await.into(),
Item::Expand(address) => core.expn_address(&address).await.into(),
})
.ok();
}
DeliveryEvent::Stop => break,
}
}
});
}

111
crates/jmap/src/services/ingest.rs Normal file
View File

@@ -0,0 +1,111 @@
use jmap_proto::types::{state::StateChange, type_state::TypeState};
use store::ahash::AHashMap;
use utils::ipc::{DeliveryResult, IngestMessage};
use crate::{mailbox::INBOX_ID, MaybeError, JMAP};
impl JMAP {
pub async fn deliver_message(&self, message: IngestMessage) -> Vec<DeliveryResult> {
// Read message
let raw_message = match message.read_message().await {
Ok(raw_message) => raw_message,
Err(_) => {
return (0..message.recipients.len())
.map(|_| DeliveryResult::TemporaryFailure {
reason: "Temporary I/O error.".into(),
})
.collect::<Vec<_>>();
}
};
// Obtain the UIDs for each recipient
let mut recipients = Vec::with_capacity(message.recipients.len());
let mut deliver_uids = AHashMap::with_capacity(message.recipients.len());
for rcpt in message.recipients {
let uids = self.get_uids_by_address(&rcpt).await;
for uid in &uids {
deliver_uids.insert(*uid, DeliveryResult::Success);
}
recipients.push(uids);
}
// Deliver to each recipient
for (uid, status) in &mut deliver_uids {
match self
.email_ingest(&raw_message, *uid, vec![INBOX_ID], vec![], None, true)
.await
{
Ok(ingested_message) => {
// Notify state change
if ingested_message.change_id != u64::MAX {
self.broadcast_state_change(
StateChange::new(*uid)
.with_change(TypeState::EmailDelivery, ingested_message.change_id)
.with_change(TypeState::Email, ingested_message.change_id)
.with_change(TypeState::Mailbox, ingested_message.change_id)
.with_change(TypeState::Thread, ingested_message.change_id),
)
.await;
}
}
Err(err) => match err {
MaybeError::Temporary => {
*status = DeliveryResult::TemporaryFailure {
reason: "Transient server failure.".into(),
}
}
MaybeError::Permanent(reason) => {
*status = DeliveryResult::PermanentFailure {
code: [5, 5, 0],
reason: reason.into(),
}
}
},
}
}
// Build result
recipients
.into_iter()
.map(|uids| {
match uids.len() {
1 => {
// Delivery to single recipient
deliver_uids.get(&uids[0]).unwrap().clone()
}
0 => {
// Something went wrong
DeliveryResult::TemporaryFailure {
reason: "Address lookup failed.".into(),
}
}
_ => {
// Delivery to list, count number of successes and failures
let mut success = 0;
let mut temp_failures = 0;
for uid in uids {
match deliver_uids.get(&uid).unwrap() {
DeliveryResult::Success => success += 1,
DeliveryResult::TemporaryFailure { .. } => temp_failures += 1,
DeliveryResult::PermanentFailure { .. } => {}
}
}
if success > temp_failures {
DeliveryResult::Success
} else if temp_failures > 0 {
DeliveryResult::TemporaryFailure {
reason: "Delivery to one or more recipients failed temporarily."
.into(),
}
} else {
DeliveryResult::PermanentFailure {
code: [5, 5, 0],
reason: "Delivery to all recipients failed.".into(),
}
}
}
}
})
.collect()
}
}

2
crates/jmap/src/services/mod.rs
View File

@@ -1,3 +1,5 @@
pub mod delivery;
pub mod ingest;
pub mod state;
pub const IPC_CHANNEL_BUFFER: usize = 1024;