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

Authenticate using registry - part 3

Browse Source
This commit is contained in:
mdecimus
2026-02-06 18:50:59 +01:00
parent 1a780ca294
commit 22e5f5a5ab
62 changed files with 2549 additions and 2156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

642
crates/common/src/auth/access_token.rs
View File

@@ -7,79 +7,56 @@
use super::AccessToken;
use crate::{
Server,
ipc::BroadcastEvent,
listener::limiter::{ConcurrencyLimiter, LimiterResult},
auth::{
AccessScope, AccessTo, AccessTokenInner, FALLBACK_ADMIN_ID, Permissions, PermissionsGroup,
},
network::limiter::{ConcurrencyLimiter, LimiterResult},
};
use registry::{
schema::{
enums::Permission,
structs::{self, Account},
},
types::EnumType,
};
use ahash::AHashSet;
use registry::schema::enums::Permission;
use std::{
hash::{DefaultHasher, Hash, Hasher},
sync::Arc,
};
use store::{query::acl::AclQuery, rand};
use store::{query::acl::AclQuery, rand, write::now};
use tinyvec::TinyVec;
use trc::AddContext;
use types::{acl::Acl, collection::Collection};
use utils::map::{
bitmap::{Bitmap, BitmapItem},
vec_map::VecMap,
};
use utils::map::bitmap::{Bitmap, BitmapItem};
impl Server {
async fn build_access_token_from_principal(
async fn build_account_access_token(
&self,
principal: Principal,
account: Account,
account_id: u32,
revision: u64,
) -> trc::Result<AccessToken> {
let mut role_permissions = PermissionsGroup::default();
// Extract data
let mut object_quota = self.core.email.max_objects;
let mut description = None;
let mut tenant_id = None;
let mut quota = None;
let mut locale = None;
let mut member_of = Vec::new();
let mut emails = Vec::new();
for data in principal.data {
match data {
PrincipalData::Tenant(v) => tenant_id = Some(v),
PrincipalData::MemberOf(v) => member_of.push(v),
PrincipalData::Role(v) => {
role_permissions.union(self.get_role_permissions(v).await?.as_ref());
}
PrincipalData::Permission {
permission_id,
grant,
} => {
if grant {
role_permissions.enabled.set(permission_id as usize);
} else {
role_permissions.disabled.set(permission_id as usize);
}
}
PrincipalData::DiskQuota(v) => quota = Some(v),
PrincipalData::ObjectQuota { quota, typ } => {
object_quota[typ as usize] = quota;
}
PrincipalData::Description(v) => description = Some(v),
PrincipalData::PrimaryEmail(v) => {
if emails.is_empty() {
emails.push(v);
} else {
emails.insert(0, v);
}
}
PrincipalData::EmailAlias(v) => {
emails.push(v);
}
PrincipalData::Locale(v) => locale = Some(v),
_ => (),
) -> trc::Result<AccessTokenInner> {
// Calculate effective permissions
let (mut permissions, roles) = match account.permissions {
structs::Permissions::Inherit => {
(PermissionsGroup::default(), account.role_ids.as_slice())
}
structs::Permissions::Merge(permissions) => (
PermissionsGroup::from(permissions),
account.role_ids.as_slice(),
),
structs::Permissions::Replace(permissions) => {
(PermissionsGroup::from(permissions), &[][..])
}
};
if !roles.is_empty() {
permissions = self
.add_role_permissions(permissions, roles.into_iter().map(|v| v.id() as u32))
.await
.caused_by(trc::location!())?
}
// Apply principal permissions
let mut permissions = role_permissions.finalize();
let mut tenant = None;
let tenant_id = account.member_tenant_id.map(|t| t.id() as u32);
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
@@ -87,92 +64,56 @@ impl Server {
#[cfg(feature = "enterprise")]
{
use directory::{QueryParams, ROLE_USER};
if let Some(tenant_id) = tenant_id {
if self.is_enterprise_edition() {
// Limit tenant permissions
permissions.intersection(&self.get_role_permissions(tenant_id).await?.enabled);
// Obtain tenant quota
tenant = Some(TenantInfo {
id: tenant_id,
quota: self
.store()
.query(QueryParams::id(tenant_id).with_return_member_of(false))
let tenant = self.tenant(tenant_id).await.caused_by(trc::location!())?;
let (mut tenant_permissions, tenant_roles) =
if let Some(permissions) = &tenant.permissions {
if permissions.merge {
((**permissions).clone(), tenant.id_roles.as_slice())
} else {
((**permissions).clone(), &[][..])
}
} else {
(PermissionsGroup::default(), tenant.id_roles.as_slice())
};
if !tenant_roles.is_empty() {
tenant_permissions = self
.add_role_permissions(tenant_permissions, tenant_roles.iter().copied())
.await
.caused_by(trc::location!())?
.ok_or_else(|| {
trc::SecurityEvent::Unauthorized
.into_err()
.details("Tenant not found")
.id(tenant_id)
.caused_by(trc::location!())
})?
.quota()
.unwrap_or_default(),
});
}
permissions.restrict(&tenant_permissions);
} else {
// Enterprise edition downgrade, remove any tenant administrator permissions
permissions.intersection(&self.get_role_permissions(ROLE_USER).await?.enabled);
permissions.restrict(&PermissionsGroup::user());
}
}
}
// SPDX-SnippetEnd
// Build member of and e-mail addresses
for &group_id in &member_of {
if let Some(group) = self
.store()
.query(QueryParams::id(group_id).with_return_member_of(false))
.await
.caused_by(trc::location!())?
&& group.typ == Type::Group
{
emails.extend(group.into_email_addresses());
}
}
// Build access token
let mut access_token = AccessToken {
account_id: principal.id,
member_of,
access_to: VecMap::new(),
tenant,
name: principal.name,
description,
emails,
quota: quota.unwrap_or_default(),
locale,
permissions,
object_quota,
concurrent_imap_requests: self.core.imap.rate_concurrent.map(ConcurrencyLimiter::new),
concurrent_http_requests: self
.core
.jmap
.request_max_concurrent
.map(ConcurrencyLimiter::new),
concurrent_uploads: self
.core
.jmap
.upload_max_concurrent
.map(ConcurrencyLimiter::new),
obj_size: 0,
revision,
};
for grant_account_id in [access_token.account_id]
.into_iter()
.chain(access_token.member_of.iter().copied())
{
let can_impersonate = permissions.enabled.get(Permission::Impersonate as usize)
&& !permissions.disabled.get(Permission::Impersonate as usize);
let member_of = account
.role_ids
.iter()
.map(|m| m.id() as u32)
.collect::<TinyVec<[u32; 3]>>();
let mut access_to: Vec<AccessTo> = Vec::new();
for grant_account_id in [account_id].into_iter().chain(member_of.iter().copied()) {
for acl_item in self
.store()
.acl_query(AclQuery::HasAccess { grant_account_id })
.await
.caused_by(trc::location!())?
{
if !access_token.is_member(acl_item.to_account_id) {
if acl_item.to_account_id != account_id
&& !member_of.contains(&acl_item.to_account_id)
&& !can_impersonate
{
let acl = Bitmap::<Acl>::from(acl_item.permissions);
let collection = acl_item.to_collection;
if !collection.is_valid() {
@@ -194,248 +135,186 @@ impl Server {
}
if !collections.is_empty() {
access_token
.access_to
.get_mut_or_insert_with(acl_item.to_account_id, Bitmap::new)
.union(&collections);
if let Some(idx) = access_to
.iter()
.position(|a| a.account_id == acl_item.to_account_id)
{
access_to[idx].collections.union(&collections);
} else {
access_to.push(AccessTo {
account_id: acl_item.to_account_id,
collections,
});
}
}
}
}
}
Ok(access_token.update_size())
}
let now = now();
let app_password_scopes = account
.app_passwords
.into_iter()
.filter_map(|pass| {
let expires_at = pass
.expires_at
.map(|v| v.timestamp() as u64)
.unwrap_or(u64::MAX);
if expires_at > now {
let permissions = match pass.permissions {
structs::Permissions::Inherit => permissions.clone().finalize(),
structs::Permissions::Merge(merge) => {
let mut permissions = permissions.clone();
permissions.union(&PermissionsGroup::from(merge));
permissions.finalize()
}
structs::Permissions::Replace(replace) => {
PermissionsGroup::from(replace).finalize()
}
};
Some(AccessScope {
permissions,
expires_at,
})
} else {
None
}
})
.collect::<Vec<_>>();
async fn build_access_token(&self, account_id: u32, revision: u64) -> trc::Result<AccessToken> {
let err = match self
.directory()
.query(QueryParams::id(account_id).with_return_member_of(true))
.await
{
Ok(Some(principal)) => {
return self
.build_access_token_from_principal(principal, revision)
.await;
}
Ok(None) => Err(trc::AuthEvent::Error
.into_err()
.details("Account not found.")
.caused_by(trc::location!())),
Err(err) => Err(err),
};
match &self.core.network.security.fallback_admin {
Some((_, secret)) if account_id == u32::MAX => {
self.build_access_token_from_principal(Principal::fallback_admin(secret), revision)
.await
}
_ => err,
Ok(AccessTokenInner {
concurrent_imap_requests: self.core.imap.rate_concurrent.map(ConcurrencyLimiter::new),
concurrent_http_requests: self
.core
.jmap
.request_max_concurrent
.map(ConcurrencyLimiter::new),
concurrent_uploads: self
.core
.jmap
.upload_max_concurrent
.map(ConcurrencyLimiter::new),
obj_size: 0,
revision,
account_id,
tenant_id,
member_of,
access_to: access_to.into_boxed_slice(),
scopes: [AccessScope::new(permissions.finalize())]
.into_iter()
.chain(app_password_scopes)
.collect::<Box<[AccessScope]>>(),
}
.update_size())
}
pub async fn get_access_token(
pub async fn account_access_token(
&self,
principal: impl Into<PrincipalOrId>,
) -> trc::Result<Arc<AccessToken>> {
let principal = principal.into();
// Obtain current revision
let principal_id = principal.id();
account_id: u32,
) -> trc::Result<Arc<AccessTokenInner>> {
match self
.inner
.cache
.access_tokens
.get_value_or_guard_async(&principal_id)
.get_value_or_guard_async(&account_id)
.await
{
Ok(token) => Ok(token),
Err(guard) => {
let revision = rand::random::<u64>();
let token: Arc<AccessToken> = match principal {
PrincipalOrId::Principal(principal) => {
self.build_access_token_from_principal(principal, revision)
.await?
}
PrincipalOrId::Id(account_id) => {
self.build_access_token(account_id, revision).await?
}
}
.into();
let account = self
.registry()
.object::<Account>(account_id)
.await?
.ok_or_else(|| {
trc::SecurityEvent::Unauthorized
.into_err()
.details("Account not found")
.account_id(account_id)
.caused_by(trc::location!())
})?;
let token: Arc<AccessTokenInner> = self
.build_account_access_token(account, account_id, revision)
.await?
.into();
let _ = guard.insert(token.clone());
Ok(token)
}
}
}
pub async fn invalidate_principal_caches(&self, changed_principals: ChangedPrincipals) {
let mut nested_principals = Vec::new();
let mut changed_ids = AHashSet::new();
let mut changed_names = Vec::new();
for (id, changed_principal) in changed_principals.iter() {
changed_ids.insert(*id);
if changed_principal.name_change {
self.inner.cache.files.remove(id);
self.inner.cache.contacts.remove(id);
self.inner.cache.events.remove(id);
self.inner.cache.scheduling.remove(id);
changed_names.push(*id);
async fn access_token_from_account(
&self,
account_id: u32,
account: Account,
) -> trc::Result<Arc<AccessTokenInner>> {
match self
.inner
.cache
.access_tokens
.get_value_or_guard_async(&account_id)
.await
{
Ok(token) => Ok(token),
Err(guard) => {
let revision = rand::random::<u64>();
let token: Arc<AccessTokenInner> = self
.build_account_access_token(account, account_id, revision)
.await?
.into();
let _ = guard.insert(token.clone());
Ok(token)
}
if changed_principal.member_change {
if changed_principal.typ == Type::Tenant {
match self
.store()
.list_principals(
None,
(*id).into(),
&[Type::Individual, Type::Group, Type::Role, Type::ApiKey],
false,
0,
0,
)
.await
{
Ok(principals) => {
for principal in principals.items {
changed_ids.insert(principal.id());
}
}
Err(err) => {
trc::error!(
err.details("Failed to list principals")
.caused_by(trc::location!())
.account_id(*id)
);
}
}
} else {
nested_principals.push(*id);
}
}
}
if !nested_principals.is_empty() {
let mut ids = nested_principals.into_iter();
let mut ids_stack = vec![];
loop {
if let Some(id) = ids.next() {
// Skip if already fetched
if !changed_ids.insert(id) {
continue;
}
// Obtain principal
match self.store().get_members(id).await {
Ok(members) => {
ids_stack.push(ids);
ids = members.into_iter();
}
Err(err) => {
trc::error!(
err.details("Failed to obtain principal")
.caused_by(trc::location!())
.account_id(id)
);
}
}
} else if let Some(prev_ids) = ids_stack.pop() {
ids = prev_ids;
} else {
break;
}
}
}
// Invalidate access tokens in cluster
if !changed_ids.is_empty() {
let mut ids = Vec::with_capacity(changed_ids.len());
for id in changed_ids {
self.inner.cache.permissions.remove(&id);
self.inner.cache.access_tokens.remove(&id);
ids.push(id);
}
self.cluster_broadcast(BroadcastEvent::InvalidateAccessTokens(ids))
.await;
}
// Invalidate DAV caches
if !changed_names.is_empty() {
self.cluster_broadcast(BroadcastEvent::InvalidateGroupwareCache(changed_names))
.await;
}
}
}
impl AccessToken {
pub fn from_id(account_id: u32) -> Self {
Self {
account_id,
..Default::default()
}
}
pub fn with_access_to(self, access_to: VecMap<u32, Bitmap<Collection>>) -> Self {
Self { access_to, ..self }
}
pub fn with_permission(mut self, permission: Permission) -> Self {
self.permissions.set(permission.id() as usize);
self
}
pub fn with_tenant_id(mut self, tenant_id: Option<u32>) -> Self {
self.tenant = tenant_id.map(|id| TenantInfo { id, quota: 0 });
self
}
pub fn state(&self) -> u32 {
// Hash state
let mut s = DefaultHasher::new();
self.member_of.hash(&mut s);
self.access_to.hash(&mut s);
self.inner.member_of.hash(&mut s);
self.inner.access_to.hash(&mut s);
s.finish() as u32
}
#[inline(always)]
pub fn account_id(&self) -> u32 {
self.account_id
self.inner.account_id
}
#[inline(always)]
pub fn tenant_id(&self) -> Option<u32> {
self.tenant.as_ref().map(|t| t.id)
self.inner.tenant_id
}
pub fn secondary_ids(&self) -> impl Iterator<Item = &u32> {
self.member_of
self.inner
.member_of
.iter()
.chain(self.access_to.iter().map(|(id, _)| id))
.chain(self.inner.access_to.iter().map(|a| &a.account_id))
}
pub fn member_ids(&self) -> impl Iterator<Item = u32> {
[self.account_id]
[self.inner.account_id]
.into_iter()
.chain(self.member_of.iter().copied())
.chain(self.inner.member_of.iter().copied())
}
pub fn all_ids(&self) -> impl Iterator<Item = u32> {
[self.account_id]
[self.inner.account_id]
.into_iter()
.chain(self.member_of.iter().copied())
.chain(self.access_to.iter().map(|(id, _)| *id))
.chain(self.inner.member_of.iter().copied())
.chain(self.inner.access_to.iter().map(|a| a.account_id))
}
pub fn all_ids_by_collection(&self, collection: Collection) -> impl Iterator<Item = u32> {
[self.account_id]
[self.inner.account_id]
.into_iter()
.chain(self.member_of.iter().copied())
.chain(self.access_to.iter().filter_map(move |(id, cols)| {
if cols.contains(collection) {
Some(*id)
.chain(self.inner.member_of.iter().copied())
.chain(self.inner.access_to.iter().filter_map(move |a| {
if a.collections.contains(collection) {
Some(a.account_id)
} else {
None
}
@@ -443,18 +322,29 @@ impl AccessToken {
}
pub fn is_member(&self, account_id: u32) -> bool {
self.account_id == account_id
|| self.member_of.contains(&account_id)
self.inner.account_id == account_id
|| self.inner.member_of.contains(&account_id)
|| self.has_permission(Permission::Impersonate)
}
pub fn is_account_id(&self, account_id: u32) -> bool {
self.account_id == account_id
self.inner.account_id == account_id
}
#[inline(always)]
pub fn has_permission(&self, permission: Permission) -> bool {
self.permissions.get(permission.id() as usize)
self.inner
.scopes
.get(self.scope_id as usize)
.map_or(false, |scope| scope.permissions.get(permission as usize))
}
pub fn is_valid(&self) -> bool {
let todo = "use this function";
self.inner
.scopes
.get(self.scope_id as usize)
.map_or(false, |scope| scope.expires_at > now())
}
pub fn assert_has_permission(&self, permission: Permission) -> trc::Result<bool> {
@@ -463,7 +353,7 @@ impl AccessToken {
} else {
Err(trc::SecurityEvent::Unauthorized
.into_err()
.details(permission.name()))
.details(permission.as_str()))
}
}
@@ -472,14 +362,18 @@ impl AccessToken {
const USIZE_MASK: u32 = USIZE_BITS as u32 - 1;
let mut permissions = Vec::new();
for (block_num, bytes) in self.permissions.inner().iter().enumerate() {
let Some(scope) = self.inner.scopes.get(self.scope_id as usize) else {
return permissions;
};
for (block_num, bytes) in scope.permissions.inner().iter().enumerate() {
let mut bytes = *bytes;
while bytes != 0 {
let item = USIZE_MASK - bytes.leading_zeros();
bytes ^= 1 << item;
if let Some(permission) =
Permission::from_id(((block_num * USIZE_BITS) + item as usize) as u32)
Permission::from_id(((block_num * USIZE_BITS) + item as usize) as u16)
{
permissions.push(permission);
}
@@ -488,21 +382,22 @@ impl AccessToken {
permissions
}
#[inline(always)]
pub fn object_quota(&self, collection: Collection) -> u32 {
self.object_quota[collection as usize]
}
pub fn is_shared(&self, account_id: u32) -> bool {
!self.is_member(account_id) && self.access_to.iter().any(|(id, _)| *id == account_id)
!self.is_member(account_id)
&& self
.inner
.access_to
.iter()
.any(|a| a.account_id == account_id)
}
pub fn shared_accounts(&self, collection: Collection) -> impl Iterator<Item = &u32> {
self.member_of
self.inner
.member_of
.iter()
.chain(self.access_to.iter().filter_map(move |(id, cols)| {
if cols.contains(collection) {
id.into()
.chain(self.inner.access_to.iter().filter_map(move |a| {
if a.collections.contains(collection) {
Some(&a.account_id)
} else {
None
}
@@ -512,49 +407,106 @@ impl AccessToken {
pub fn has_access(&self, to_account_id: u32, to_collection: impl Into<Collection>) -> bool {
let to_collection = to_collection.into();
self.is_member(to_account_id)
|| self.access_to.iter().any(|(id, collections)| {
*id == to_account_id && collections.contains(to_collection)
})
|| self
.inner
.access_to
.iter()
.any(|a| a.account_id == to_account_id && a.collections.contains(to_collection))
}
pub fn has_account_access(&self, to_account_id: u32) -> bool {
self.is_member(to_account_id) || self.access_to.iter().any(|(id, _)| *id == to_account_id)
}
pub fn as_resource_token(&self) -> ResourceToken {
ResourceToken {
account_id: self.account_id,
quota: self.quota,
tenant: self.tenant,
}
self.is_member(to_account_id)
|| self
.inner
.access_to
.iter()
.any(|a| a.account_id == to_account_id)
}
pub fn is_http_request_allowed(&self) -> LimiterResult {
self.concurrent_http_requests
self.inner
.concurrent_http_requests
.as_ref()
.map_or(LimiterResult::Disabled, |limiter| limiter.is_allowed())
}
pub fn is_imap_request_allowed(&self) -> LimiterResult {
self.concurrent_imap_requests
self.inner
.concurrent_imap_requests
.as_ref()
.map_or(LimiterResult::Disabled, |limiter| limiter.is_allowed())
}
pub fn is_upload_allowed(&self) -> LimiterResult {
self.concurrent_uploads
self.inner
.concurrent_uploads
.as_ref()
.map_or(LimiterResult::Disabled, |limiter| limiter.is_allowed())
}
}
impl AccessTokenInner {
pub fn from_id(account_id: u32) -> Self {
Self {
account_id,
..Default::default()
}
}
pub fn with_access_to(self, access_to: impl IntoIterator<Item = AccessTo>) -> Self {
Self {
access_to: access_to.into_iter().collect(),
..self
}
}
pub fn with_scopes(self, scopes: impl IntoIterator<Item = AccessScope>) -> Self {
Self {
scopes: scopes.into_iter().collect(),
..self
}
}
pub fn with_tenant_id(mut self, tenant_id: Option<u32>) -> Self {
self.tenant_id = tenant_id;
self
}
pub fn new_admin() -> Self {
AccessTokenInner {
account_id: FALLBACK_ADMIN_ID,
tenant_id: Default::default(),
member_of: Default::default(),
access_to: Default::default(),
scopes: Box::new([AccessScope::new(Permissions::all())]),
concurrent_http_requests: Default::default(),
concurrent_imap_requests: Default::default(),
concurrent_uploads: Default::default(),
revision: Default::default(),
obj_size: Default::default(),
}
}
pub fn update_size(mut self) -> Self {
self.obj_size = (std::mem::size_of::<AccessToken>()
+ (self.member_of.len() * std::mem::size_of::<u32>())
+ (self.access_to.len() * (std::mem::size_of::<u32>() + std::mem::size_of::<u64>()))
+ self.name.len()
+ self.description.as_ref().map_or(0, |v| v.len())
+ self.locale.as_ref().map_or(0, |v| v.len())
+ self.emails.iter().map(|v| v.len()).sum::<usize>()) as u64;
+ (self.scopes.len() * std::mem::size_of::<AccessScope>()))
as u64;
self
}
}
impl AccessScope {
pub fn new(permissions: Permissions) -> Self {
Self {
permissions,
expires_at: u64::MAX,
}
}
pub fn expires_at(mut self, expires_at: u64) -> Self {
self.expires_at = expires_at;
self
}
}

8
crates/common/src/auth/authentication.rs
View File

@@ -4,6 +4,8 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::Server;
impl Server {
pub async fn authenticate(&self, req: &AuthRequest<'_>) -> trc::Result<Arc<AccessToken>> {
// Resolve directory
@@ -239,12 +241,6 @@ impl<'x> AuthRequest<'x> {
}
}
impl CacheItemWeight for AccessToken {
fn weight(&self) -> u64 {
self.obj_size
}
}
pub(crate) trait CredentialsUsername {
fn login(&self) -> Option<&str>;
}

60
crates/common/src/auth/mod.rs
View File

@@ -4,14 +4,18 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{expr::if_block::IfBlock, listener::limiter::ConcurrencyLimiter};
use crate::{
expr::if_block::IfBlock,
network::limiter::ConcurrencyLimiter,
storage::{ObjectQuota, TenantQuota},
};
use arcstr::ArcStr;
use directory::Credentials;
use registry::{
schema::enums::{Locale, Permission, StorageQuota, TenantStorageQuota},
schema::enums::{Locale, Permission},
types::EnumType,
};
use std::net::IpAddr;
use std::{net::IpAddr, sync::Arc};
use tinyvec::TinyVec;
use trc::ipc::bitset::Bitset;
use types::collection::Collection;
@@ -24,10 +28,10 @@ pub mod permissions;
pub mod rate_limit;
pub mod sasl;
pub const FALLBACK_ADMIN_ID: u32 = u32::MAX;
const PERMISSIONS_BITSET_SIZE: usize = Permission::COUNT.div_ceil(std::mem::size_of::<usize>());
pub type Permissions = Bitset<PERMISSIONS_BITSET_SIZE>;
pub type ObjectQuota = [u32; StorageQuota::COUNT - 1];
pub type TenantQuota = [u32; TenantStorageQuota::COUNT - 1];
//pub type IdMap<V> = HashMap<u32, Arc<V>, nohash_hasher::BuildNoHashHasher<u32>>;
//pub type NameMap<V> = AHashMap<ArcStr, Arc<V>>;
@@ -41,9 +45,10 @@ pub enum EmailCache {
#[derive(Debug, Clone)]
pub struct DomainCache {
pub name: ArcStr,
pub id_tenant: u32,
pub id: u32,
pub id_directory: u32,
pub catch_all: Option<Box<str>>,
pub id_tenant: u32,
pub catch_all: Option<ArcStr>,
pub sub_addressing_custom: Option<Box<IfBlock>>,
pub flags: u8,
}
@@ -58,9 +63,8 @@ pub const DOMAIN_FLAG_ALIAS_LOGIN: u8 = 1 << 4;
pub struct AccountCache {
pub addresses: Box<[ArcStr]>,
pub addresses_temporary: Box<[TemporaryAddress]>,
pub id_member_of: TinyVec<[u32; 3]>,
pub id_tenant: u32,
pub id_roles: TinyVec<[u32; 3]>,
pub id_member_of: TinyVec<[u32; 3]>,
pub quota_disk: u64,
pub quota_objects: Option<Box<ObjectQuota>>,
pub description: Option<Box<str>>,
@@ -75,7 +79,6 @@ pub struct TemporaryAddress {
#[derive(Debug, Clone)]
pub struct RoleCache {
pub id_tenant: u32,
pub id_roles: TinyVec<[u32; 3]>,
pub permissions: PermissionsGroup,
}
@@ -84,8 +87,7 @@ pub struct RoleCache {
pub struct MailingListCache {
pub addresses: Box<[ArcStr]>,
pub addresses_temporary: Box<[TemporaryAddress]>,
pub id_tenant: u32,
pub recipients: Box<[ArcStr]>,
pub recipients: Arc<[ArcStr]>,
}
#[derive(Debug, Clone)]
@@ -96,15 +98,6 @@ pub struct TenantCache {
pub permissions: Option<Box<PermissionsGroup>>,
}
#[derive(Debug, Clone)]
pub struct ApiKeyCache {
pub id: u32,
pub id_tenant: u32,
pub id_roles: TinyVec<[u32; 3]>,
pub permissions: Option<Box<PermissionsGroup>>,
pub expires_at: u64,
}
#[derive(Debug, Clone, Default)]
pub struct PermissionsGroup {
pub enabled: Permissions,
@@ -114,9 +107,17 @@ pub struct PermissionsGroup {
#[derive(Debug, Default)]
pub struct AccessToken {
scope_id: u32,
inner: Arc<AccessTokenInner>,
}
#[derive(Debug, Default)]
pub struct AccessTokenInner {
pub account_id: u32,
pub tenant_id: Option<u32>,
pub member_of: TinyVec<[u32; 3]>,
pub access_to: Box<[AccessTo]>,
pub permissions: Permissions,
pub scopes: Box<[AccessScope]>,
pub concurrent_http_requests: Option<ConcurrencyLimiter>,
pub concurrent_imap_requests: Option<ConcurrencyLimiter>,
pub concurrent_uploads: Option<ConcurrencyLimiter>,
@@ -125,6 +126,12 @@ pub struct AccessToken {
}
#[derive(Debug, Default)]
struct AccessScope {
pub permissions: Permissions,
pub expires_at: u64,
}
#[derive(Debug, Default, Hash, PartialEq, Eq)]
pub struct AccessTo {
pub account_id: u32,
pub collections: Bitmap<Collection>,
@@ -138,7 +145,7 @@ pub struct AuthRequest {
allow_api_access: bool,
}
impl CacheItemWeight for AccessToken {
impl CacheItemWeight for AccessTokenInner {
fn weight(&self) -> u64 {
self.obj_size
}
@@ -196,10 +203,3 @@ impl CacheItemWeight for PermissionsGroup {
std::mem::size_of::<PermissionsGroup>() as u64
}
}
impl CacheItemWeight for ApiKeyCache {
fn weight(&self) -> u64 {
std::mem::size_of::<ApiKeyCache>() as u64
+ self.permissions.as_ref().map_or(0, |p| p.weight())
}
}

17
crates/common/src/auth/oauth/introspect.rs
View File

@@ -57,16 +57,13 @@ impl Server {
Ok(token_info) => Ok(OAuthIntrospect {
active: true,
client_id: Some(token_info.client_id),
username: if access_token.account_id() == token_info.account_id {
access_token.name.clone()
} else {
self.get_access_token(token_info.account_id)
.await
.caused_by(trc::location!())?
.name
.clone()
}
.into(),
username: self
.account(access_token.account_id())
.await
.caused_by(trc::location!())?
.addresses
.first()
.map(|v| v.to_string()),
token_type: Some("bearer".into()),
exp: Some(token_info.expiry as i64),
iat: Some(token_info.issued_at as i64),

25
crates/common/src/auth/oauth/token.rs
View File

@@ -8,6 +8,7 @@ use super::{CLIENT_ID_MAX_LEN, GrantType, RANDOM_CODE_LEN, crypto::SymmetricEncr
use crate::Server;
use mail_builder::encoders::base64::base64_encode;
use mail_parser::decoders::base64::base64_decode;
use registry::schema::structs::Account;
use std::time::SystemTime;
use store::{
blake3,
@@ -216,34 +217,18 @@ impl Server {
pub async fn password_hash(&self, account_id: u32) -> trc::Result<String> {
if account_id != u32::MAX {
self.core
.storage
.directory
.query(QueryParams::id(account_id).with_return_member_of(false))
self.registry()
.object::<Account>(account_id)
.await
.caused_by(trc::location!())?
.map(|account| account.secret)
.ok_or_else(|| {
trc::AuthEvent::Error
.into_err()
.details("Account no longer exists")
})?
.data
.into_iter()
.filter_map(|v| {
if let PrincipalData::Password(secret) = v {
Some(secret)
} else {
None
}
})
.next()
.ok_or(
trc::AuthEvent::Error
.into_err()
.details("Account does not contain secrets")
.caused_by(trc::location!()),
)
} else if let Some((_, secret)) = &self.core.network.security.fallback_admin {
let todo = "api keys?";
Ok(secret.into())
} else {
Err(trc::AuthEvent::Error

219
crates/common/src/auth/permissions.rs
View File

@@ -9,15 +9,16 @@ use crate::{
auth::{Permissions, PermissionsGroup},
};
use ahash::AHashSet;
use registry::schema::{enums::Permission, structs::PermissionsList};
use trc::AddContext;
impl Server {
pub async fn effective_permissions(
pub async fn add_role_permissions(
&self,
mut base_permissions: PermissionsGroup,
role_id: u32,
) -> trc::Result<Permissions> {
let mut role_ids = vec![role_id];
roles: impl IntoIterator<Item = u32>,
) -> trc::Result<PermissionsGroup> {
let mut role_ids = roles.into_iter().collect::<Vec<u32>>();
let mut fetched_role_ids = AHashSet::new();
while let Some(role_id) = role_ids.pop() {
@@ -29,7 +30,7 @@ impl Server {
}
}
Ok(base_permissions.finalize())
Ok(base_permissions)
}
}
@@ -39,6 +40,11 @@ impl PermissionsGroup {
self.disabled.union(&other.disabled);
}
pub fn restrict(&mut self, other: &PermissionsGroup) {
self.enabled.intersection(&other.enabled);
self.disabled.union(&other.disabled);
}
pub fn finalize(mut self) -> Permissions {
self.enabled.difference(&self.disabled);
self.enabled
@@ -49,4 +55,207 @@ impl PermissionsGroup {
enabled.difference(&self.disabled);
enabled
}
pub fn user() -> Self {
let mut permissions = PermissionsGroup::default();
for permission in [
Permission::Authenticate,
Permission::AuthenticateOauth,
Permission::EmailSend,
Permission::EmailReceive,
Permission::ManageEncryption,
Permission::ManagePasswords,
Permission::JmapEmailGet,
Permission::JmapMailboxGet,
Permission::JmapThreadGet,
Permission::JmapIdentityGet,
Permission::JmapEmailSubmissionGet,
Permission::JmapPushSubscriptionGet,
Permission::JmapSieveScriptGet,
Permission::JmapVacationResponseGet,
Permission::JmapQuotaGet,
Permission::JmapBlobGet,
Permission::JmapEmailSet,
Permission::JmapMailboxSet,
Permission::JmapIdentitySet,
Permission::JmapEmailSubmissionSet,
Permission::JmapPushSubscriptionSet,
Permission::JmapSieveScriptSet,
Permission::JmapVacationResponseSet,
Permission::JmapEmailChanges,
Permission::JmapMailboxChanges,
Permission::JmapThreadChanges,
Permission::JmapIdentityChanges,
Permission::JmapEmailSubmissionChanges,
Permission::JmapQuotaChanges,
Permission::JmapEmailCopy,
Permission::JmapBlobCopy,
Permission::JmapEmailImport,
Permission::JmapEmailParse,
Permission::JmapEmailQueryChanges,
Permission::JmapMailboxQueryChanges,
Permission::JmapEmailSubmissionQueryChanges,
Permission::JmapSieveScriptQueryChanges,
Permission::JmapQuotaQueryChanges,
Permission::JmapEmailQuery,
Permission::JmapMailboxQuery,
Permission::JmapEmailSubmissionQuery,
Permission::JmapSieveScriptQuery,
Permission::JmapQuotaQuery,
Permission::JmapSearchSnippet,
Permission::JmapSieveScriptValidate,
Permission::JmapBlobLookup,
Permission::JmapBlobUpload,
Permission::JmapEcho,
Permission::ImapAuthenticate,
Permission::ImapAclGet,
Permission::ImapAclSet,
Permission::ImapMyRights,
Permission::ImapListRights,
Permission::ImapAppend,
Permission::ImapCapability,
Permission::ImapId,
Permission::ImapCopy,
Permission::ImapMove,
Permission::ImapCreate,
Permission::ImapDelete,
Permission::ImapEnable,
Permission::ImapExpunge,
Permission::ImapFetch,
Permission::ImapIdle,
Permission::ImapList,
Permission::ImapLsub,
Permission::ImapNamespace,
Permission::ImapRename,
Permission::ImapSearch,
Permission::ImapSort,
Permission::ImapSelect,
Permission::ImapExamine,
Permission::ImapStatus,
Permission::ImapStore,
Permission::ImapSubscribe,
Permission::ImapThread,
Permission::Pop3Authenticate,
Permission::Pop3List,
Permission::Pop3Uidl,
Permission::Pop3Stat,
Permission::Pop3Retr,
Permission::Pop3Dele,
Permission::SieveAuthenticate,
Permission::SieveListScripts,
Permission::SieveSetActive,
Permission::SieveGetScript,
Permission::SievePutScript,
Permission::SieveDeleteScript,
Permission::SieveRenameScript,
Permission::SieveCheckScript,
Permission::SieveHaveSpace,
Permission::DavSyncCollection,
Permission::DavExpandProperty,
Permission::DavPrincipalAcl,
Permission::DavPrincipalList,
Permission::DavPrincipalSearch,
Permission::DavPrincipalMatch,
Permission::DavPrincipalSearchPropSet,
Permission::DavFilePropFind,
Permission::DavFilePropPatch,
Permission::DavFileGet,
Permission::DavFileMkCol,
Permission::DavFileDelete,
Permission::DavFilePut,
Permission::DavFileCopy,
Permission::DavFileMove,
Permission::DavFileLock,
Permission::DavFileAcl,
Permission::DavCardPropFind,
Permission::DavCardPropPatch,
Permission::DavCardGet,
Permission::DavCardMkCol,
Permission::DavCardDelete,
Permission::DavCardPut,
Permission::DavCardCopy,
Permission::DavCardMove,
Permission::DavCardLock,
Permission::DavCardAcl,
Permission::DavCardQuery,
Permission::DavCardMultiGet,
Permission::DavCalPropFind,
Permission::DavCalPropPatch,
Permission::DavCalGet,
Permission::DavCalMkCol,
Permission::DavCalDelete,
Permission::DavCalPut,
Permission::DavCalCopy,
Permission::DavCalMove,
Permission::DavCalLock,
Permission::DavCalAcl,
Permission::DavCalQuery,
Permission::DavCalMultiGet,
Permission::DavCalFreeBusyQuery,
Permission::CalendarAlarms,
Permission::CalendarSchedulingSend,
Permission::CalendarSchedulingReceive,
Permission::JmapAddressBookGet,
Permission::JmapAddressBookSet,
Permission::JmapAddressBookChanges,
Permission::JmapContactCardGet,
Permission::JmapContactCardChanges,
Permission::JmapContactCardQuery,
Permission::JmapContactCardQueryChanges,
Permission::JmapContactCardSet,
Permission::JmapContactCardCopy,
Permission::JmapContactCardParse,
Permission::JmapFileNodeGet,
Permission::JmapFileNodeSet,
Permission::JmapFileNodeChanges,
Permission::JmapFileNodeQuery,
Permission::JmapFileNodeQueryChanges,
Permission::JmapPrincipalGetAvailability,
Permission::JmapPrincipalChanges,
Permission::JmapPrincipalQuery,
Permission::JmapPrincipalGet,
Permission::JmapPrincipalQueryChanges,
Permission::JmapShareNotificationGet,
Permission::JmapShareNotificationSet,
Permission::JmapShareNotificationChanges,
Permission::JmapShareNotificationQuery,
Permission::JmapShareNotificationQueryChanges,
Permission::JmapCalendarGet,
Permission::JmapCalendarSet,
Permission::JmapCalendarChanges,
Permission::JmapCalendarEventGet,
Permission::JmapCalendarEventSet,
Permission::JmapCalendarEventChanges,
Permission::JmapCalendarEventQuery,
Permission::JmapCalendarEventQueryChanges,
Permission::JmapCalendarEventCopy,
Permission::JmapCalendarEventParse,
Permission::JmapCalendarEventNotificationGet,
Permission::JmapCalendarEventNotificationSet,
Permission::JmapCalendarEventNotificationChanges,
Permission::JmapCalendarEventNotificationQuery,
Permission::JmapCalendarEventNotificationQueryChanges,
Permission::JmapParticipantIdentityGet,
Permission::JmapParticipantIdentitySet,
Permission::JmapParticipantIdentityChanges,
] {
permissions.enabled.set(permission as usize);
}
permissions
}
}
impl From<PermissionsList> for PermissionsGroup {
fn from(value: PermissionsList) -> Self {
let mut permissions = PermissionsGroup::default();
for (permission, is_set) in value.permissions {
if is_set {
permissions.enabled.set(permission as usize);
} else {
permissions.disabled.set(permission as usize);
}
}
permissions
}
}

9
crates/common/src/auth/rate_limit.rs
View File

@@ -5,10 +5,9 @@
*/
use crate::auth::AccessToken;
use crate::{
KV_RATE_LIMIT_HTTP_ANONYMOUS, KV_RATE_LIMIT_HTTP_AUTHENTICATED, Server, ip_to_bytes,
listener::limiter::{InFlight, LimiterResult},
};
use crate::network::ip_to_bytes;
use crate::network::limiter::{InFlight, LimiterResult};
use crate::{KV_RATE_LIMIT_HTTP_ANONYMOUS, KV_RATE_LIMIT_HTTP_AUTHENTICATED, Server};
use registry::schema::enums::Permission;
use std::net::IpAddr;
use trc::AddContext;
@@ -24,7 +23,7 @@ impl Server {
.memory
.is_rate_allowed(
KV_RATE_LIMIT_HTTP_AUTHENTICATED,
&access_token.account_id.to_be_bytes(),
&access_token.account_id().to_be_bytes(),
rate,
false,
)

19
crates/common/src/cache/directory.rs vendored
View File

@@ -6,7 +6,7 @@
use crate::{
Server,
auth::{AccountCache, ApiKeyCache, DomainCache, EmailCache, RoleCache, TenantCache},
auth::{AccountCache, DomainCache, EmailCache, RoleCache, TenantCache},
config::smtp::auth::DkimSigner,
};
use std::sync::Arc;
@@ -21,10 +21,13 @@ impl Server {
}
pub async fn account(&self, id: u32) -> trc::Result<Arc<AccountCache>> {
todo!()
}
/*
pub async fn group(&self, id: u32) -> trc::Result<Arc<AccountCache>> {
Err(trc::AuthEvent::Error
.into_err()
.details("Account not found.")
.caused_by(trc::location!()))
*/
todo!()
}
@@ -32,15 +35,11 @@ impl Server {
todo!()
}
pub async fn tenant(&self, id: u32) -> trc::Result<Option<Arc<TenantCache>>> {
pub async fn tenant(&self, id: u32) -> trc::Result<Arc<TenantCache>> {
todo!()
}
pub async fn api_key(&self, id: u32) -> trc::Result<Option<Arc<ApiKeyCache>>> {
todo!()
}
pub async fn dkim_signers(&self, domain_id: u32) -> trc::Result<Option<Arc<[DkimSigner]>>> {
pub async fn dkim_signers(&self, domain: &str) -> trc::Result<Option<Arc<[DkimSigner]>>> {
todo!()
}
}

55
crates/common/src/cache/invalidate.rs vendored Normal file
View File

@@ -0,0 +1,55 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
Server,
ipc::{BroadcastEvent, CacheInvalidation},
};
impl Server {
pub async fn invalidate_caches(&self, changes: Vec<CacheInvalidation>, broadcast: bool) {
let cache = &self.inner.cache;
for change in &changes {
match change {
CacheInvalidation::AccessToken(id) => {
cache.access_tokens.remove(id);
}
CacheInvalidation::DavResources(id) => {
cache.files.remove(id);
cache.contacts.remove(id);
cache.events.remove(id);
cache.scheduling.remove(id);
}
CacheInvalidation::Domain(id) => {
cache.domains.remove(id);
}
CacheInvalidation::Account(id) => {
cache.accounts.remove(id);
}
CacheInvalidation::Group(id) => {
cache.accounts.remove(id);
}
CacheInvalidation::Tenant(id) => {
cache.tenants.remove(id);
}
CacheInvalidation::Role(id) => {
cache.roles.remove(id);
}
CacheInvalidation::List(id) => {
cache.lists.remove(id);
}
CacheInvalidation::PushServers(_) => {}
}
}
// Broadcast cache invalidation to other servers
if broadcast {
self.cluster_broadcast(BroadcastEvent::CacheInvalidation(changes))
.await;
}
}
}

44
crates/common/src/cache/mod.rs vendored
View File

@@ -4,5 +4,49 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{DavResources, HttpAuthCache, MailboxCache, MessageStoreCache};
use utils::cache::CacheItemWeight;
pub mod directory;
pub mod invalidate;
pub mod reload;
impl MailboxCache {
pub fn parent_id(&self) -> Option<u32> {
if self.parent_id != u32::MAX {
Some(self.parent_id)
} else {
None
}
}
pub fn sort_order(&self) -> Option<u32> {
if self.sort_order != u32::MAX {
Some(self.sort_order)
} else {
None
}
}
pub fn is_root(&self) -> bool {
self.parent_id == u32::MAX
}
}
impl CacheItemWeight for MessageStoreCache {
fn weight(&self) -> u64 {
self.size
}
}
impl CacheItemWeight for HttpAuthCache {
fn weight(&self) -> u64 {
std::mem::size_of::<HttpAuthCache>() as u64
}
}
impl CacheItemWeight for DavResources {
fn weight(&self) -> u64 {
self.size
}
}

4
crates/common/src/cache/reload.rs vendored
View File

@@ -7,13 +7,13 @@
use crate::{
Core, Server,
config::{server::Listeners, telemetry::Telemetry},
listener::blocked::BlockedIps,
};
use ahash::AHashMap;
use arc_swap::ArcSwap;
use store::registry::bootstrap::Bootstrap;
pub struct ReloadResult {
pub config: Config,
pub bp: Bootstrap,
pub new_core: Option<Core>,
pub tracers: Option<Telemetry>,
}

47
crates/common/src/config/inner.rs
View File

@@ -8,14 +8,16 @@ use super::server::tls::build_self_signed_cert;
use crate::{
Caches, Data, DavResource, DavResources, MailboxCache, MessageStoreCache, MessageUidCache,
TlsConnectors,
auth::AccessToken,
auth::{AccessTokenInner, AccountCache, DomainCache, MailingListCache, RoleCache, TenantCache},
config::{
mailstore::spamfilter::SpamClassifier,
server::tls::parse_certificates,
smtp::resolver::{Policy, Tlsa},
smtp::{
auth::DkimSigner,
resolver::{Policy, Tlsa},
},
},
listener::blocked::BlockedIps,
manager::webadmin::WebAdminManager,
network::security::BlockedIps,
};
use ahash::{AHashMap, AHashSet};
use arc_swap::ArcSwap;
@@ -94,7 +96,7 @@ impl Caches {
Caches {
access_tokens: Cache::new(
cache.access_tokens,
(std::mem::size_of::<AccessToken>() + 255) as u64,
(std::mem::size_of::<AccessTokenInner>() + 255) as u64,
),
http_auth: Cache::new(cache.http_auth, (50 + std::mem::size_of::<u32>()) as u64),
messages: Cache::new(
@@ -124,6 +126,40 @@ impl Caches {
(std::mem::size_of::<DavResources>() + (500 * std::mem::size_of::<DavResource>()))
as u64,
),
emails: Cache::new(cache.email_addresses, 255u64),
emails_negative: CacheWithTtl::new(
cache.email_addresses_negative,
(std::mem::size_of::<DomainCache>() + 255) as u64,
),
domain_names: Cache::new(
cache.domain_names,
(std::mem::size_of::<DomainCache>() + 255) as u64,
),
domain_names_negative: CacheWithTtl::new(
cache.domain_names_negative,
(std::mem::size_of::<DomainCache>() + 255) as u64,
),
domains: Cache::new(
cache.domains,
(std::mem::size_of::<DomainCache>() + 255) as u64,
),
accounts: Cache::new(
cache.accounts,
(std::mem::size_of::<AccountCache>() + 255) as u64,
),
roles: Cache::new(cache.roles, (std::mem::size_of::<RoleCache>() + 255) as u64),
tenants: Cache::new(
cache.tenants,
(std::mem::size_of::<TenantCache>() + 255) as u64,
),
lists: Cache::new(
cache.mailing_lists,
(std::mem::size_of::<MailingListCache>() + 255) as u64,
),
dkim_signers: Cache::new(
cache.dkim_signatures,
(std::mem::size_of::<DkimSigner>() + 255) as u64,
),
dns_txt: CacheWithTtl::new(cache.dns_txt, (std::mem::size_of::<Txt>() + 255) as u64),
dns_mx: CacheWithTtl::new(cache.dns_mx, ((std::mem::size_of::<MX>() + 255) * 2) as u64),
dns_ptr: CacheWithTtl::new(cache.dns_ptr, (std::mem::size_of::<IpAddr>() + 255) as u64),
@@ -144,6 +180,7 @@ impl Caches {
cache.dns_rbl,
((std::mem::size_of::<Ipv4Addr>() + 255) * 2) as u64,
),
negative_cache_ttl: cache.negative_ttl.into_inner(),
}
}

41
crates/common/src/config/mailstore/email.rs
View File

@@ -8,9 +8,10 @@ use ahash::{AHashMap, AHashSet};
use nlp::language::Language;
use registry::{
schema::{
enums::{SearchCalendarField, SearchContactField, SearchEmailField},
enums::{SearchCalendarField, SearchContactField, SearchEmailField, StorageQuota},
structs::{
AddressBook, Calendar, DataRetention, Email, Jmap, Search, SieveUserInterpreter,
AddressBook, Calendar, DataRetention, Email, Jmap, OidcProvider, Search,
SieveUserInterpreter,
},
},
types::EnumType,
@@ -21,9 +22,11 @@ use store::{
search::{CalendarSearchField, ContactSearchField, EmailSearchField, SearchField},
write::SearchIndex,
};
use types::{collection::Collection, special_use::SpecialUse};
use types::special_use::SpecialUse;
use utils::cron::SimpleCron;
use crate::storage::ObjectQuota;
#[derive(Clone)]
pub struct EmailConfig {
pub default_language: Language,
@@ -50,7 +53,7 @@ pub struct EmailConfig {
pub index_batch_size: usize,
pub index_fields: AHashMap<SearchIndex, AHashSet<SearchField>>,
pub max_objects: [u32; Collection::MAX],
pub max_objects: ObjectQuota,
pub account_purge_frequency: SimpleCron,
}
@@ -73,22 +76,28 @@ impl EmailConfig {
let jmap = bp.setting_infallible::<Jmap>().await;
let calendar = bp.setting_infallible::<Calendar>().await;
let address_book = bp.setting_infallible::<AddressBook>().await;
let oidc = bp.setting_infallible::<OidcProvider>().await;
// Parse default object quotas
let mut max_objects = [u32::MAX; Collection::MAX];
for (collection, max) in [
(Collection::Mailbox, email.max_mailboxes),
(Collection::SieveScript, sieve.max_scripts),
(Collection::Identity, email.max_identities),
(Collection::EmailSubmission, email.max_submissions),
(Collection::PushSubscription, jmap.max_subscriptions),
(Collection::Calendar, calendar.max_calendars),
(Collection::CalendarEvent, calendar.max_events),
(Collection::AddressBook, address_book.max_address_books),
(Collection::ContactCard, address_book.max_contacts),
let mut max_objects = ObjectQuota::default();
for (item, max) in [
(StorageQuota::MaxMailboxes, email.max_mailboxes),
(StorageQuota::MaxSieveScripts, sieve.max_scripts),
(StorageQuota::MaxIdentities, email.max_identities),
(StorageQuota::MaxEmailSubmissions, email.max_submissions),
(StorageQuota::MaxMaskedAddresses, email.max_masked_addresses),
(StorageQuota::MaxAppPasswords, oidc.max_app_passwords),
(StorageQuota::MaxPushSubscriptions, jmap.max_subscriptions),
(StorageQuota::MaxCalendars, calendar.max_calendars),
(StorageQuota::MaxCalendarEvents, calendar.max_events),
(
StorageQuota::MaxAddressBooks,
address_book.max_address_books,
),
(StorageQuota::MaxContactCards, address_book.max_contacts),
] {
if let Some(max) = max {
max_objects[collection as usize] = max as u32;
max_objects.set(item, max as u32);
}
}

2
crates/common/src/config/mod.rs
View File

@@ -6,7 +6,7 @@
use self::{mailstore::jmap::JmapConfig, smtp::SmtpConfig, storage::Storage};
use crate::{
Core, Network, Security,
Core, Network,
auth::oauth::config::OAuthConfig,
config::mailstore::{imap::ImapConfig, scripts::Scripting, spamfilter::SpamFilterConfig},
expr::*,

5
crates/common/src/config/network.rs
View File

@@ -5,7 +5,10 @@
*/
use super::*;
use crate::expr::if_block::{BootstrapExprExt, IfBlock};
use crate::{
expr::if_block::{BootstrapExprExt, IfBlock},
network::security::Security,
};
use ahash::AHashMap;
use registry::{
schema::{

2
crates/common/src/config/server/listener.rs
View File

@@ -10,7 +10,7 @@ use super::{
};
use crate::{
Inner,
listener::{TcpAcceptor, tls::CertificateResolver},
network::{TcpAcceptor, tls::CertificateResolver},
};
use registry::schema::{
enums::{NetworkListenerProtocol, TlsCipherSuite, TlsVersion},

2
crates/common/src/config/server/mod.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::listener::TcpAcceptor;
use crate::network::TcpAcceptor;
use ahash::AHashMap;
use registry::{
schema::structs::NetworkListener,

6
crates/common/src/config/server/tls.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{Server, listener::acme::AcmeProvider};
use crate::{Server, network::acme::AcmeProvider};
use ahash::{AHashMap, AHashSet};
use dns_update::{
Algorithm, DnsUpdater, TsigAlgorithm,
@@ -48,7 +48,7 @@ impl Server {
pub async fn build_acme_provider(&self, id: Id) -> trc::Result<AcmeProvider> {
if let Some(server) = self
.registry()
.get::<structs::AcmeProvider>(id)
.id::<structs::AcmeProvider>(id)
.await
.caused_by(trc::location!())?
{
@@ -66,7 +66,7 @@ impl Server {
pub async fn build_dns_updater(&self, id: Id) -> trc::Result<DnsUpdater> {
let Some(server) = self
.registry()
.get::<DnsServer>(id)
.id::<DnsServer>(id)
.await
.caused_by(trc::location!())?
else {

10
crates/common/src/config/smtp/resolver.rs
View File

@@ -251,18 +251,20 @@ impl Policy {
T: AsRef<str>,
{
if self.mx.is_empty() {
let mut mx = Vec::new();
for name in names {
let name = name.as_ref();
if let Some(domain) = name.strip_prefix('.') {
self.mx.push(MxPattern::StartsWith(domain.to_string()));
mx.push(MxPattern::StartsWith(domain.to_string()));
} else if name != "*" && !name.is_empty() {
self.mx.push(MxPattern::Equals(name.to_string()));
mx.push(MxPattern::Equals(name.to_string()));
}
}
if !self.mx.is_empty() {
self.mx.sort_unstable();
if !mx.is_empty() {
mx.sort_unstable();
self.id = self.hash().to_string();
self.mx = mx.into_boxed_slice();
Some(self)
} else {
None

2
crates/common/src/config/storage.rs
View File

@@ -6,6 +6,7 @@
use coordinator::Coordinator;
use directory::Directory;
use registry::schema::enums::CompressionAlgo;
use std::{collections::HashMap, sync::Arc};
use store::{BlobStore, InMemoryStore, RegistryStore, SearchStore, Store};
@@ -21,4 +22,5 @@ pub struct Storage {
pub coordinator: Coordinator,
pub directory: Option<Arc<Directory>>,
pub directories: IdMap<Directory>,
pub compression: CompressionAlgo,
}

1020
crates/common/src/core.rs
View File

File diff suppressed because it is too large Load Diff

163
crates/common/src/enterprise/mod.rs
View File

@@ -22,7 +22,10 @@ use ahash::{AHashMap, AHashSet};
use license::LicenseKey;
use llm::AiApiConfig;
use mail_parser::DateTime;
use registry::types::id::Id;
use registry::{
schema::structs::{Domain, Tenant},
types::id::Id,
};
use std::{sync::Arc, time::Duration};
use store::Store;
use trc::{AddContext, MetricType};
@@ -150,11 +153,7 @@ impl Server {
pub async fn can_create_account(&self) -> trc::Result<bool> {
if let Some(enterprise) = &self.core.enterprise {
let total_accounts = self
.store()
.count_principals(None, Type::Individual.into(), None)
.await
.caused_by(trc::location!())?;
let total_accounts = self.total_accounts().await.caused_by(trc::location!())?;
if total_accounts + 1 > enterprise.license.accounts as u64 {
trc::event!(
@@ -175,88 +174,78 @@ impl Server {
pub async fn logo_resource(&self, domain: &str) -> trc::Result<Option<Resource<Vec<u8>>>> {
const MAX_IMAGE_SIZE: usize = 1024 * 1024;
if self.is_enterprise_edition() {
let domain = psl::domain_str(domain).unwrap_or(domain);
let logo = { self.inner.data.logos.lock().get(domain).cloned() };
if let Some(logo) = logo {
Ok(logo)
} else {
// Try fetching the logo for the domain
let logo_url = if let Some(mut principal) = self
.store()
.query(QueryParams::name(domain).with_return_member_of(false))
.await
.caused_by(trc::location!())?
.filter(|p| p.typ() == Type::Domain)
{
if let Some(logo) = principal.picture_mut().filter(|l| l.starts_with("http")) {
std::mem::take(logo).into()
} else if let Some(tenant_id) = principal.tenant() {
if let Some(logo) = self
.store()
.query(QueryParams::id(tenant_id).with_return_member_of(false))
.await
.caused_by(trc::location!())?
.and_then(|mut p| p.picture_mut().map(std::mem::take))
.filter(|l| l.starts_with("http"))
{
logo.clone().into()
} else {
self.default_logo_url()
}
} else {
self.default_logo_url()
}
} else {
self.default_logo_url()
};
let mut logo = None;
if let Some(logo_url) = logo_url {
let response = reqwest::get(logo_url.as_str()).await.map_err(|err| {
trc::ResourceEvent::DownloadExternal
.into_err()
.details("Failed to download logo")
.reason(err)
})?;
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|ct| ct.to_str().ok())
.unwrap_or("image/svg+xml")
.to_string();
let contents = response
.bytes_with_limit(MAX_IMAGE_SIZE)
.await
.map_err(|err| {
trc::ResourceEvent::DownloadExternal
.into_err()
.details("Failed to download logo")
.reason(err)
})?
.ok_or_else(|| {
trc::ResourceEvent::DownloadExternal
.into_err()
.details("Download exceeded maximum size")
})?;
logo = Resource::new(content_type, contents).into();
}
self.inner
.data
.logos
.lock()
.insert(domain.to_string(), logo.clone());
Ok(logo)
}
} else {
Ok(None)
if !self.is_enterprise_edition() {
return Ok(None);
}
let domain = psl::domain_str(domain).unwrap_or(domain);
let logo = { self.inner.data.logos.lock().get(domain).cloned() };
if let Some(logo) = logo {
return Ok(logo);
}
let Some((domain_id, tenant_id)) = self.domain(domain).await?.map(|d| (d.id, d.id_tenant))
else {
self.inner.data.logos.lock().insert(domain.into(), None);
return Ok(None);
};
let Some(domain_record) = self.registry().object::<Domain>(domain_id).await? else {
return Ok(None);
};
let mut logo = domain_record.logo;
if logo.is_none() && tenant_id != u32::MAX {
logo = self
.registry()
.object::<Tenant>(tenant_id)
.await?
.and_then(|t| t.logo);
}
let logo_url = logo.or_else(|| self.default_logo_url());
let mut logo = None;
if let Some(logo_url) = logo_url {
let response = reqwest::get(logo_url.as_str()).await.map_err(|err| {
trc::ResourceEvent::DownloadExternal
.into_err()
.details("Failed to download logo")
.reason(err)
})?;
let content_type = response
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|ct| ct.to_str().ok())
.unwrap_or("image/svg+xml")
.to_string();
let contents = response
.bytes_with_limit(MAX_IMAGE_SIZE)
.await
.map_err(|err| {
trc::ResourceEvent::DownloadExternal
.into_err()
.details("Failed to download logo")
.reason(err)
})?
.ok_or_else(|| {
trc::ResourceEvent::DownloadExternal
.into_err()
.details("Download exceeded maximum size")
})?;
logo = Resource::new(content_type, contents).into();
}
self.inner
.data
.logos
.lock()
.insert(domain.into(), logo.clone());
Ok(logo)
}
fn default_logo_url(&self) -> Option<String> {

61
crates/common/src/expr/functions/asynch.rs
View File

@@ -5,7 +5,7 @@
*/
use super::*;
use crate::{Server, expr::StringCow};
use crate::{Server, expr::StringCow, network::RcptExpansion};
use compact_str::{CompactString, ToCompactString};
use mail_auth::IpLookupStrategy;
use std::{cmp::Ordering, net::IpAddr, vec::IntoIter};
@@ -25,47 +25,51 @@ impl Server {
F_IS_LOCAL_DOMAIN => {
let domain = params.next_as_string();
self.get_directory_or_default(directory.as_ref(), session_id)
.is_local_domain(domain.as_ref())
self.domain(domain.as_str())
.await
.caused_by(trc::location!())
.map(|v| v.into())
.map(|v| v.is_some().into())
}
F_IS_LOCAL_ADDRESS => {
let address = params.next_as_string();
self.get_directory_or_default(directory.as_ref(), session_id)
.rcpt(address.as_ref())
self.rcpt_expand(address.as_ref())
.await
.caused_by(trc::location!())
.map(|v| (v != RcptType::Invalid).into())
.map(|v| (v != RcptExpansion::Invalid).into())
}
F_KEY_GET => {
let store = params.next_as_string();
let Some(store) = self.get_lookup_store(params.next_as_string().as_str()) else {
return Ok(Variable::default());
};
let key = params.next_as_string();
self.get_in_memory_store_or_default(store.as_str(), session_id)
store
.key_get::<VariableWrapper>(key.as_str())
.await
.map(|value| value.map(|v| v.into_inner()).unwrap_or_default())
.caused_by(trc::location!())
}
F_KEY_EXISTS => {
let store = params.next_as_string();
let Some(store) = self.get_lookup_store(params.next_as_string().as_str()) else {
return Ok(Variable::default());
};
let key = params.next_as_string();
self.get_in_memory_store_or_default(store.as_str(), session_id)
store
.key_exists(key.as_str())
.await
.caused_by(trc::location!())
.map(|v| v.into())
}
F_KEY_SET => {
let store = params.next_as_string();
let Some(store) = self.get_lookup_store(params.next_as_string().as_str()) else {
return Ok(Variable::default());
};
let key = params.next_as_string();
let value = params.next_as_string();
self.get_in_memory_store_or_default(store.as_ref(), session_id)
store
.key_set(KeyValue::new(
key.as_bytes().to_vec(),
value.as_bytes().to_vec(),
@@ -76,21 +80,25 @@ impl Server {
.map(|v| v.into())
}
F_COUNTER_INCR => {
let store = params.next_as_string();
let Some(store) = self.get_lookup_store(params.next_as_string().as_str()) else {
return Ok(Variable::default());
};
let key = params.next_as_string();
let value = params.next_as_integer();
self.get_in_memory_store_or_default(store.as_ref(), session_id)
store
.counter_incr(KeyValue::new(key.into_owned(), value), true)
.await
.map(Variable::Integer)
.caused_by(trc::location!())
}
F_COUNTER_GET => {
let store = params.next_as_string();
let Some(store) = self.get_lookup_store(params.next_as_string().as_str()) else {
return Ok(Variable::default());
};
let key = params.next_as_string();
self.get_in_memory_store_or_default(store.as_ref(), session_id)
store
.counter_get(key.as_bytes().to_vec())
.await
.map(Variable::Integer)
@@ -107,13 +115,24 @@ impl Server {
mut arguments: FncParams<'x>,
session_id: u64,
) -> trc::Result<Variable<'x>> {
let store = self.get_data_store(arguments.next_as_string().as_ref(), session_id);
let store_name = arguments.next_as_string();
let Some(store) = self
.get_lookup_store(store_name.as_ref())
.and_then(|v| v.into_store())
else {
return Err(trc::EventType::Eval(trc::EvalEvent::Error)
.into_err()
.id(store_name.into_owned())
.span_id(session_id)
.details("Store not found or is not a SQL store"));
};
let query = arguments.next_as_string();
if query.is_empty() {
return Err(trc::EventType::Eval(trc::EvalEvent::Error)
.into_err()
.details("Empty query string"));
.details("Empty query string")
.span_id(session_id));
}
// Obtain arguments
@@ -203,9 +222,7 @@ impl Server {
.flat_map(|mx| {
mx.exchanges.iter().map(|host| {
Variable::String(StringCow::Owned(
host.strip_suffix('.')
.unwrap_or(host.as_str())
.to_compact_string(),
host.strip_suffix('.').unwrap_or(host).to_compact_string(),
))
})
})

4
crates/common/src/expr/if_block.rs
View File

@@ -122,9 +122,7 @@ impl BootstrapExprExt for Bootstrap {
) -> Option<IfBlock> {
// Parse conditions
let mut if_then = Vec::with_capacity(expr.match_.len());
let mut default = Expression {
items: Default::default(),
};
let default;
if expr.else_.is_empty() {
if !expr.match_.is_empty() {

21
crates/common/src/expr/mod.rs
View File

@@ -18,6 +18,9 @@ use std::{
time::Duration,
};
use trc::MetricType;
use utils::cache::CacheItemWeight;
use crate::expr::if_block::IfBlock;
pub mod eval;
pub mod functions;
@@ -498,3 +501,21 @@ where
.collect()
}
}
impl CacheItemWeight for Expression {
fn weight(&self) -> u64 {
self.items.len() as u64 * std::mem::size_of::<ExpressionItem>() as u64
}
}
impl CacheItemWeight for IfBlock {
fn weight(&self) -> u64 {
std::mem::size_of::<IfBlock>() as u64
+ self
.if_then
.iter()
.map(|if_then| if_then.expr.weight() + if_then.then.weight())
.sum::<u64>()
+ self.default.weight()
}
}

4
crates/common/src/expr/parser.rs
View File

@@ -222,7 +222,9 @@ impl<'x> ExpressionParser<'x> {
}
if self.operator_stack.is_empty() {
Ok(Expression { items: self.output })
Ok(Expression {
items: self.output.into_boxed_slice(),
})
} else {
Err("Invalid expression".to_string())
}

29
crates/common/src/ipc.rs
View File

@@ -15,6 +15,7 @@ use mail_auth::{
mta_sts::TlsRpt,
report::{Record, tlsrpt::FailureDetails},
};
use registry::{schema::prelude::Object, types::id::Id};
use std::{
sync::{
Arc,
@@ -101,12 +102,28 @@ pub struct CalendarAlert {
#[derive(Debug)]
pub enum BroadcastEvent {
PushNotification(PushNotification),
InvalidateAccessTokens(Vec<u32>),
InvalidateGroupwareCache(Vec<u32>),
ReloadPushServers(u32),
ReloadSettings,
ReloadBlockedIps,
ReloadSpamFilter,
RegistryChange(RegistryChange),
CacheInvalidation(Vec<CacheInvalidation>),
}
#[derive(Debug)]
pub enum RegistryChange {
Insert(Id),
Delete(Id),
Reload(Object),
}
#[derive(Debug, Clone, Copy)]
pub enum CacheInvalidation {
AccessToken(u32),
DavResources(u32),
Domain(u32),
Account(u32),
Group(u32),
Tenant(u32),
Role(u32),
List(u32),
PushServers(u32),
}
#[derive(Debug)]

503
crates/common/src/lib.rs
View File

@@ -6,11 +6,10 @@
#![warn(clippy::large_futures)]
use crate::auth::AccessTokenInner;
use crate::network::asn::AsnGeoLookupData;
use crate::{
auth::{
AccountCache, ApiKeyCache, DomainCache, EmailCache, MailingListCache, RoleCache,
TenantCache,
},
auth::{AccountCache, DomainCache, EmailCache, MailingListCache, RoleCache, TenantCache},
config::{
mailstore::{
email::EmailConfig,
@@ -21,12 +20,12 @@ use crate::{
smtp::auth::DkimSigner,
},
ipc::TrainTaskController,
listener::blocked::BlockedIps,
network::security::BlockedIps,
};
use ahash::{AHashMap, AHashSet};
use arc_swap::ArcSwap;
use arcstr::ArcStr;
use auth::{AccessToken, oauth::config::OAuthConfig};
use auth::oauth::config::OAuthConfig;
use calcard::common::timezone::Tz;
use config::{
groupware::GroupwareConfig,
@@ -40,40 +39,33 @@ use config::{
telemetry::Metrics,
};
use ipc::{BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent};
use listener::{asn::AsnGeoLookupData, blocked::Security};
use mail_auth::{MX, Txt};
use manager::webadmin::{Resource, WebAdminManager};
use parking_lot::{Mutex, RwLock};
use rustls::sign::CertifiedKey;
use std::{
hash::{BuildHasher, Hash, Hasher},
net::{IpAddr, Ipv4Addr, Ipv6Addr},
sync::{Arc, atomic::AtomicBool},
time::{Duration, Instant},
};
use store::{
InMemoryStore,
rand::{Rng, distr::Alphanumeric},
};
use store::InMemoryStore;
use tinyvec::TinyVec;
use tokio::sync::{Notify, Semaphore, mpsc};
use tokio_rustls::TlsConnector;
use types::{acl::AclGrant, special_use::SpecialUse};
use utils::{
cache::{Cache, CacheItemWeight, CacheWithTtl},
cache::{Cache, CacheWithTtl},
snowflake::SnowflakeIdGenerator,
};
pub mod auth;
pub mod cache;
pub mod config;
pub mod core;
pub mod dns;
pub mod expr;
pub mod i18n;
pub mod ipc;
pub mod listener;
pub mod manager;
pub mod network;
pub mod scripts;
pub mod sharing;
pub mod storage;
@@ -172,7 +164,7 @@ pub struct Data {
}
pub struct Caches {
pub access_tokens: Cache<u32, Arc<AccessToken>>,
pub access_tokens: Cache<u32, Arc<AccessTokenInner>>,
pub http_auth: Cache<Box<str>, HttpAuthCache>,
pub messages: Cache<u32, Arc<MessageStoreCache>>,
@@ -182,19 +174,17 @@ pub struct Caches {
pub scheduling: Cache<u32, Arc<DavResources>>,
pub emails: Cache<ArcStr, EmailCache>,
pub emails_temporary: CacheWithTtl<ArcStr, EmailCache>,
pub emails_negative: CacheWithTtl<ArcStr, ()>,
pub domains: Cache<ArcStr, Arc<DomainCache>>,
pub domains_negative: CacheWithTtl<ArcStr, ()>,
pub domain_names: Cache<ArcStr, u32>,
pub domain_names_negative: CacheWithTtl<ArcStr, ()>,
pub domains: Cache<u32, Arc<DomainCache>>,
pub accounts: Cache<u32, Arc<AccountCache>>,
pub groups: Cache<u32, Arc<AccountCache>>,
pub roles: Cache<u32, Arc<RoleCache>>,
pub tenants: Cache<u32, Arc<TenantCache>>,
pub lists: Cache<u32, Arc<MailingListCache>>,
pub api_keys: Cache<u32, Arc<ApiKeyCache>>,
pub dkim_signers: Cache<Box<str>, Arc<[DkimSigner]>>,
pub dkim_signers: Cache<u32, Arc<[DkimSigner]>>,
pub dns_txt: CacheWithTtl<Box<str>, Txt>,
pub dns_mx: CacheWithTtl<Box<str>, Arc<[MX]>>,
@@ -204,6 +194,8 @@ pub struct Caches {
pub dns_tlsa: CacheWithTtl<Box<str>, Arc<Tlsa>>,
pub dns_mta_sts: CacheWithTtl<Box<str>, Arc<Policy>>,
pub dns_rbl: CacheWithTtl<Box<str>, Option<Arc<IpResolver>>>,
pub negative_cache_ttl: Duration,
}
#[derive(Debug, Clone)]
@@ -385,21 +377,16 @@ pub struct Core {
// SPDX-SnippetEnd
}
impl CacheItemWeight for MessageStoreCache {
fn weight(&self) -> u64 {
self.size
}
pub trait BuildServer {
fn build_server(&self) -> Server;
}
impl CacheItemWeight for HttpAuthCache {
fn weight(&self) -> u64 {
std::mem::size_of::<HttpAuthCache>() as u64
}
}
impl CacheItemWeight for DavResources {
fn weight(&self) -> u64 {
self.size
impl BuildServer for Arc<Inner> {
fn build_server(&self) -> Server {
Server {
inner: self.clone(),
core: self.shared_core.load_full(),
}
}
}
@@ -419,458 +406,14 @@ pub struct ThrottleKey {
pub hash: [u8; 32],
}
impl PartialEq for ThrottleKey {
fn eq(&self, other: &Self) -> bool {
self.hash == other.hash
}
}
impl std::hash::Hash for ThrottleKey {
fn hash<H: Hasher>(&self, state: &mut H) {
self.hash.hash(state);
}
}
impl AsRef<[u8]> for ThrottleKey {
fn as_ref(&self) -> &[u8] {
&self.hash
}
}
#[derive(Default)]
pub struct ThrottleKeyHasher {
hash: u64,
}
impl Hasher for ThrottleKeyHasher {
fn finish(&self) -> u64 {
self.hash
}
fn write(&mut self, bytes: &[u8]) {
debug_assert!(
bytes.len() >= std::mem::size_of::<u64>(),
"ThrottleKeyHasher: input too short {bytes:?}"
);
self.hash = bytes
.get(0..std::mem::size_of::<u64>())
.map_or(0, |b| u64::from_ne_bytes(b.try_into().unwrap()));
}
}
#[derive(Clone, Default)]
pub struct ThrottleKeyHasherBuilder {}
impl BuildHasher for ThrottleKeyHasherBuilder {
type Hasher = ThrottleKeyHasher;
fn build_hasher(&self) -> Self::Hasher {
ThrottleKeyHasher::default()
}
}
pub fn ip_to_bytes(ip: &IpAddr) -> Vec<u8> {
match ip {
IpAddr::V4(ip) => ip.octets().to_vec(),
IpAddr::V6(ip) => ip.octets().to_vec(),
}
}
pub fn ip_to_bytes_prefix(prefix: u8, ip: &IpAddr) -> Vec<u8> {
match ip {
IpAddr::V4(ip) => {
let mut buf = Vec::with_capacity(5);
buf.push(prefix);
buf.extend_from_slice(&ip.octets());
buf
}
IpAddr::V6(ip) => {
let mut buf = Vec::with_capacity(17);
buf.push(prefix);
buf.extend_from_slice(&ip.octets());
buf
}
}
}
impl DavResourcePath<'_> {
#[inline(always)]
pub fn document_id(&self) -> u32 {
self.resource.document_id
}
#[inline(always)]
pub fn parent_id(&self) -> Option<u32> {
self.path.parent_id
}
#[inline(always)]
pub fn path(&self) -> &str {
self.path.path.as_str()
}
#[inline(always)]
pub fn is_container(&self) -> bool {
self.resource.is_container()
}
#[inline(always)]
pub fn hierarchy_seq(&self) -> u32 {
self.path.hierarchy_seq
}
#[inline(always)]
pub fn size(&self) -> u32 {
self.resource.size().unwrap_or_default()
}
}
impl DavResources {
pub fn by_path(&self, name: &str) -> Option<DavResourcePath<'_>> {
self.paths.get(name).map(|path| DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
}
pub fn container_resource_by_id(&self, id: u32) -> Option<&DavResource> {
self.resources
.iter()
.find(|res| res.document_id == id && res.is_container())
}
pub fn container_resource_path_by_id(&self, id: u32) -> Option<DavResourcePath<'_>> {
self.resources
.iter()
.enumerate()
.find(|(_, resource)| resource.document_id == id && resource.is_container())
.and_then(|(idx, resource)| {
self.paths
.iter()
.find(|path| path.resource_idx == idx)
.map(|path| DavResourcePath { path, resource })
})
}
pub fn any_resource_path_by_id(&self, id: u32) -> Option<DavResourcePath<'_>> {
self.resources
.iter()
.enumerate()
.find(|(_, resource)| resource.document_id == id)
.and_then(|(idx, resource)| {
self.paths
.iter()
.find(|path| path.resource_idx == idx)
.map(|path| DavResourcePath { path, resource })
})
}
pub fn subtree(&self, search_path: &str) -> impl Iterator<Item = DavResourcePath<'_>> {
let prefix = format!("{search_path}/");
self.paths.iter().filter_map(move |path| {
if path.path.starts_with(&prefix) || path.path == search_path {
Some(DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
} else {
None
}
})
}
pub fn subtree_with_depth(
&self,
search_path: &str,
depth: usize,
) -> impl Iterator<Item = DavResourcePath<'_>> {
let prefix = format!("{search_path}/");
self.paths.iter().filter_map(move |path| {
if path
.path
.strip_prefix(&prefix)
.is_some_and(|name| name.as_bytes().iter().filter(|&&c| c == b'/').count() < depth)
|| path.path.as_str() == search_path
{
Some(DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
} else {
None
}
})
}
pub fn tree_with_depth(&self, depth: usize) -> impl Iterator<Item = DavResourcePath<'_>> {
self.paths.iter().filter_map(move |path| {
if path.path.as_bytes().iter().filter(|&&c| c == b'/').count() <= depth {
Some(DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
} else {
None
}
})
}
pub fn children(&self, parent_id: u32) -> impl Iterator<Item = DavResourcePath<'_>> {
self.paths
.iter()
.filter(move |item| item.parent_id.is_some_and(|id| id == parent_id))
.map(|path| DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
}
pub fn children_ids(&self, parent_id: u32) -> impl Iterator<Item = u32> {
self.paths
.iter()
.filter(move |item| item.parent_id.is_some_and(|id| id == parent_id))
.map(|path| self.resources[path.resource_idx].document_id)
}
pub fn format_resource(&self, resource: DavResourcePath<'_>) -> String {
if resource.resource.is_container() {
format!("{}{}/", self.base_path, resource.path.path)
} else {
format!("{}{}", self.base_path, resource.path.path)
}
}
pub fn format_collection(&self, name: &str) -> String {
format!("{}{name}/", self.base_path)
}
pub fn format_item(&self, name: &str) -> String {
format!("{}{}", self.base_path, name)
}
}
const SCHEDULE_INBOX_ID: u32 = u32::MAX - 1;
impl DavResource {
pub fn is_child_of(&self, parent_id: u32) -> bool {
match &self.data {
DavResourceMetadata::File { parent_id: id, .. } => id.is_some_and(|id| id == parent_id),
DavResourceMetadata::CalendarEvent { names, .. } => {
names.iter().any(|name| name.parent_id == parent_id)
}
DavResourceMetadata::ContactCard { names } => {
names.iter().any(|name| name.parent_id == parent_id)
}
DavResourceMetadata::CalendarEventNotification { names } => {
names.is_empty() && parent_id == SCHEDULE_INBOX_ID
}
_ => false,
}
}
pub fn parent_id(&self) -> Option<u32> {
match &self.data {
DavResourceMetadata::File { parent_id, .. } => *parent_id,
DavResourceMetadata::CalendarEvent { names, .. } => {
names.first().map(|name| name.parent_id)
}
DavResourceMetadata::ContactCard { names } => names.first().map(|name| name.parent_id),
DavResourceMetadata::CalendarEventNotification { names } if names.is_empty() => {
Some(SCHEDULE_INBOX_ID)
}
_ => None,
}
}
pub fn child_names(&self) -> Option<&[DavName]> {
match &self.data {
DavResourceMetadata::CalendarEvent { names, .. } => Some(names.as_slice()),
DavResourceMetadata::ContactCard { names } => Some(names.as_slice()),
DavResourceMetadata::CalendarEventNotification { names } if !names.is_empty() => {
Some(names.as_slice())
}
_ => None,
}
}
pub fn container_name(&self) -> Option<&str> {
match &self.data {
DavResourceMetadata::File { name, .. } => Some(name.as_str()),
DavResourceMetadata::Calendar { name, .. } => Some(name.as_str()),
DavResourceMetadata::AddressBook { name, .. } => Some(name.as_str()),
DavResourceMetadata::CalendarEventNotification { names } if names.is_empty() => {
Some(if self.document_id == SCHEDULE_INBOX_ID {
"inbox"
} else {
"outbox"
})
}
_ => None,
}
}
pub fn has_hierarchy_changes(&self, other: &DavResource) -> bool {
match (&self.data, &other.data) {
(
DavResourceMetadata::File {
name: a,
parent_id: c,
..
},
DavResourceMetadata::File {
name: b,
parent_id: d,
..
},
) => a != b || c != d,
(
DavResourceMetadata::Calendar { name: a, .. },
DavResourceMetadata::Calendar { name: b, .. },
) => a != b,
(
DavResourceMetadata::AddressBook { name: a, .. },
DavResourceMetadata::AddressBook { name: b, .. },
) => a != b,
(
DavResourceMetadata::CalendarEvent { names: a, .. },
DavResourceMetadata::CalendarEvent { names: b, .. },
) => a != b,
(
DavResourceMetadata::ContactCard { names: a, .. },
DavResourceMetadata::ContactCard { names: b, .. },
) => a != b,
(
DavResourceMetadata::CalendarEventNotification { names: a, .. },
DavResourceMetadata::CalendarEventNotification { names: b, .. },
) => a != b,
_ => unreachable!(),
}
}
pub fn event_time_range(&self) -> Option<(i64, i64)> {
match &self.data {
DavResourceMetadata::CalendarEvent {
start, duration, ..
} => Some((*start, *start + *duration as i64)),
_ => None,
}
}
pub fn calendar_preferences(&self, account_id: u32) -> Option<&TinyCalendarPreferences> {
match &self.data {
DavResourceMetadata::Calendar { preferences, .. } => preferences
.iter()
.find(|pref| pref.account_id == account_id)
.or_else(|| preferences.first()),
_ => None,
}
}
pub fn is_container(&self) -> bool {
match &self.data {
DavResourceMetadata::File { size, .. } => size.is_none(),
DavResourceMetadata::Calendar { .. } | DavResourceMetadata::AddressBook { .. } => true,
DavResourceMetadata::CalendarEventNotification { names } => names.is_empty(),
_ => false,
}
}
pub fn size(&self) -> Option<u32> {
match &self.data {
DavResourceMetadata::File { size, .. } => *size,
_ => None,
}
}
pub fn acls(&self) -> Option<&[AclGrant]> {
match &self.data {
DavResourceMetadata::File { acls, .. } => Some(acls.as_slice()),
DavResourceMetadata::Calendar { acls, .. } => Some(acls.as_slice()),
DavResourceMetadata::AddressBook { acls, .. } => Some(acls.as_slice()),
_ => None,
}
}
}
impl Hash for DavPath {
fn hash<H: Hasher>(&self, state: &mut H) {
self.path.hash(state);
}
}
impl PartialEq for DavPath {
fn eq(&self, other: &Self) -> bool {
self.path == other.path
}
}
impl Eq for DavPath {}
impl std::borrow::Borrow<str> for DavPath {
fn borrow(&self) -> &str {
&self.path
}
}
impl std::hash::Hash for DavResource {
fn hash<H: Hasher>(&self, state: &mut H) {
self.document_id.hash(state);
}
}
impl PartialEq for DavResource {
fn eq(&self, other: &Self) -> bool {
self.document_id == other.document_id
}
}
impl Eq for DavResource {}
impl std::borrow::Borrow<u32> for DavResource {
fn borrow(&self) -> &u32 {
&self.document_id
}
}
impl DavName {
pub fn new(name: String, parent_id: u32) -> Self {
Self { name, parent_id }
}
pub fn new_with_rand_name(parent_id: u32) -> Self {
Self {
name: store::rand::rng()
.sample_iter(Alphanumeric)
.take(10)
.map(char::from)
.collect::<String>(),
parent_id,
}
}
}
impl MailboxCache {
pub fn parent_id(&self) -> Option<u32> {
if self.parent_id != u32::MAX {
Some(self.parent_id)
} else {
None
}
}
pub fn sort_order(&self) -> Option<u32> {
if self.sort_order != u32::MAX {
Some(self.sort_order)
} else {
None
}
}
pub fn is_root(&self) -> bool {
self.parent_id == u32::MAX
}
}
pub const DEFAULT_LOGO_RAW: &str = r#"<svg xmlns="http://www.w3.org/2000/svg" xml:space="preserve" id="Layer_1" x="0" y="0" style="enable-background:new 0 0 680.5 252.1" version="1.1" viewBox="0 0 680.5 252.1">
<style>
.st0{fill:#100e42}.st1{fill:#db2d54}

8
crates/common/src/manager/boot.rs
View File

@@ -8,7 +8,6 @@ use super::{WEBADMIN_KEY, backup::BackupParams, console::store_console};
use crate::{
Caches, Core, Data, IPC_CHANNEL_BUFFER, Inner, Ipc,
config::{network::AsnGeoLookupConfig, server::Listeners, telemetry::Telemetry},
core::BuildServer,
ipc::{
BroadcastEvent, HousekeeperEvent, PushEvent, QueueEvent, ReportingEvent,
TrainTaskController,
@@ -21,12 +20,15 @@ use std::{
path::PathBuf,
sync::Arc,
};
use store::rand::{Rng, distr::Alphanumeric, rng};
use store::{
rand::{Rng, distr::Alphanumeric, rng},
registry::bootstrap::Bootstrap,
};
use tokio::sync::{Notify, mpsc};
use utils::{UnwrapFailure, failed};
pub struct BootManager {
pub config: Config,
pub bp: Bootstrap,
pub inner: Arc<Inner>,
pub servers: Listeners,
pub ipc_rxs: IpcReceivers,

0
crates/common/src/listener/acme/cache.rs → crates/common/src/network/acme/cache.rs
View File

0
crates/common/src/listener/acme/directory.rs → crates/common/src/network/acme/directory.rs
View File

0
crates/common/src/listener/acme/jose.rs → crates/common/src/network/acme/jose.rs
View File

0
crates/common/src/listener/acme/mod.rs → crates/common/src/network/acme/mod.rs
View File

4
crates/common/src/listener/acme/order.rs → crates/common/src/network/acme/order.rs
View File

@@ -15,8 +15,8 @@ use store::dispatch::lookup::KeyValue;
use trc::{AcmeEvent, DnsEvent, EventType};
use x509_parser::parse_x509_certificate;
use crate::listener::acme::ChallengeSettings;
use crate::listener::acme::directory::Identifier;
use crate::network::acme::ChallengeSettings;
use crate::network::acme::directory::Identifier;
use crate::{KV_ACME, Server};
use super::AcmeProvider;

7
crates/common/src/listener/acme/resolver.rs → crates/common/src/network/acme/resolver.rs
View File

@@ -29,17 +29,14 @@ impl Server {
let mut certificates = self.inner.data.tls_certificates.load().as_ref().clone();
for domain in provider.domains.iter() {
certificates.insert(
domain
.strip_prefix("*.")
.unwrap_or(domain.as_str())
.to_string(),
domain.strip_prefix("*.").unwrap_or(domain.as_str()).into(),
cert.clone(),
);
}
// Add default certificate
if provider.default {
certificates.insert("*".to_string(), cert);
certificates.insert("*".into(), cert);
}
self.inner.data.tls_certificates.store(certificates.into());

0
crates/common/src/listener/asn.rs → crates/common/src/network/asn.rs
View File

0
crates/common/src/dns.rs → crates/common/src/network/dns.rs
View File

61
crates/common/src/listener/limiter.rs → crates/common/src/network/limiter.rs
View File

@@ -4,22 +4,25 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::sync::{
Arc,
atomic::{AtomicU64, Ordering},
use crate::{ThrottleKey, ThrottleKeyHasher, ThrottleKeyHasherBuilder};
use std::{
hash::{BuildHasher, Hasher},
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
};
#[derive(Debug, Clone)]
#[repr(transparent)]
pub struct ConcurrencyLimiter(Arc<ConcurrencyLimiterInner>);
#[derive(Debug)]
#[derive(Debug, Default)]
pub struct ConcurrencyLimiterInner {
max_concurrent: u64,
concurrent: AtomicU64,
}
#[derive(Default)]
pub struct InFlight(Arc<ConcurrencyLimiterInner>);
impl Drop for InFlight {
@@ -53,6 +56,10 @@ impl ConcurrencyLimiter {
pub fn is_active(&self) -> bool {
self.0.concurrent.load(Ordering::Relaxed) > 0
}
pub fn max_concurrent(&self) -> u64 {
self.0.max_concurrent
}
}
impl InFlight {
@@ -72,7 +79,49 @@ impl From<LimiterResult> for Option<InFlight> {
match result {
LimiterResult::Allowed(in_flight) => Some(in_flight),
LimiterResult::Forbidden => None,
LimiterResult::Disabled => Some(InFlight::default()),
LimiterResult::Disabled => Some(InFlight(Arc::new(ConcurrencyLimiterInner::default()))),
}
}
}
impl PartialEq for ThrottleKey {
fn eq(&self, other: &Self) -> bool {
self.hash == other.hash
}
}
impl std::hash::Hash for ThrottleKey {
fn hash<H: Hasher>(&self, state: &mut H) {
self.hash.hash(state);
}
}
impl AsRef<[u8]> for ThrottleKey {
fn as_ref(&self) -> &[u8] {
&self.hash
}
}
impl Hasher for ThrottleKeyHasher {
fn finish(&self) -> u64 {
self.hash
}
fn write(&mut self, bytes: &[u8]) {
debug_assert!(
bytes.len() >= std::mem::size_of::<u64>(),
"ThrottleKeyHasher: input too short {bytes:?}"
);
self.hash = bytes
.get(0..std::mem::size_of::<u64>())
.map_or(0, |b| u64::from_ne_bytes(b.try_into().unwrap()));
}
}
impl BuildHasher for ThrottleKeyHasherBuilder {
type Hasher = ThrottleKeyHasher;
fn build_hasher(&self) -> Self::Hasher {
ThrottleKeyHasher::default()
}
}

5
crates/common/src/listener/listen.rs → crates/common/src/network/listen.rs
View File

@@ -9,9 +9,8 @@ use super::{
limiter::{ConcurrencyLimiter, LimiterResult},
};
use crate::{
Inner, Server,
BuildServer, Inner, Server,
config::server::{Listener, Listeners, ServerProtocol, TcpListener},
core::BuildServer,
};
use proxy_header::io::ProxiedStream;
use rustls::crypto::ring::cipher_suite::TLS13_AES_128_GCM_SHA256;
@@ -253,7 +252,7 @@ impl BuildSession for Arc<ServerInstance> {
LocalPort = local_addr.port(),
RemoteIp = remote_ip,
RemotePort = remote_port,
Limit = self.limiter.max_concurrent,
Limit = self.limiter.max_concurrent(),
);
None

38
crates/common/src/listener/mod.rs → crates/common/src/network/mod.rs
View File

@@ -10,6 +10,7 @@ use crate::{
config::server::ServerProtocol,
expr::{functions::ResolveVariable, *},
};
use arcstr::ArcStr;
use compact_str::ToCompactString;
use registry::{schema::enums::ExpressionVariable, types::ipmask::IpAddrOrMask};
use rustls::ServerConfig;
@@ -25,12 +26,23 @@ use utils::snowflake::SnowflakeIdGenerator;
pub mod acme;
pub mod asn;
pub mod blocked;
pub mod dns;
pub mod limiter;
pub mod listen;
pub mod mta;
pub mod security;
pub mod stream;
pub mod tls;
#[derive(Debug, Default, Clone, PartialEq, Eq, Hash)]
pub enum RcptExpansion {
Mailbox(u32),
List(Arc<[ArcStr]>),
External(ArcStr),
#[default]
Invalid,
}
pub struct ServerInstance {
pub id: String,
pub protocol: ServerProtocol,
@@ -252,3 +264,27 @@ impl Debug for TcpAcceptor {
}
}
}
pub fn ip_to_bytes(ip: &IpAddr) -> Vec<u8> {
match ip {
IpAddr::V4(ip) => ip.octets().to_vec(),
IpAddr::V6(ip) => ip.octets().to_vec(),
}
}
pub fn ip_to_bytes_prefix(prefix: u8, ip: &IpAddr) -> Vec<u8> {
match ip {
IpAddr::V4(ip) => {
let mut buf = Vec::with_capacity(5);
buf.push(prefix);
buf.extend_from_slice(&ip.octets());
buf
}
IpAddr::V6(ip) => {
let mut buf = Vec::with_capacity(17);
buf.push(prefix);
buf.extend_from_slice(&ip.octets());
buf
}
}
}

268
crates/common/src/network/mta.rs Normal file
View File

@@ -0,0 +1,268 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
Server,
config::{
mailstore::spamfilter::SpamClassifier,
smtp::{
auth::DkimSigner,
queue::{
ConnectionStrategy, DEFAULT_QUEUE_NAME, MxConfig, QueueExpiry, QueueName,
QueueStrategy, RequireOptional, RoutingStrategy, TlsStrategy, VirtualQueue,
},
},
},
manager::SPAM_CLASSIFIER_KEY,
network::RcptExpansion,
};
use mail_auth::IpLookupStrategy;
use sieve::Sieve;
use std::{
sync::{Arc, LazyLock},
time::Duration,
};
use store::{
Deserialize, IterateParams, ValueKey,
write::{AlignedBytes, Archive, QueueClass, ValueClass},
};
use trc::{AddContext, SpamEvent};
impl Server {
pub async fn rcpt_expand(&self, address: &str) -> trc::Result<RcptExpansion> {
let todo = "TODO: RcptExpansion implementation";
todo!()
}
pub async fn get_dkim_signers(
&self,
domain: &str,
session_id: u64,
) -> trc::Result<Option<Arc<[DkimSigner]>>> {
if let Some(signers) = self.dkim_signers(domain).await? {
Ok(Some(signers))
} else {
trc::event!(
Dkim(trc::DkimEvent::SignerNotFound),
Id = domain.to_string(),
SpanId = session_id,
);
Ok(None)
}
}
pub fn get_trusted_sieve_script(&self, name: &str, session_id: u64) -> Option<&Arc<Sieve>> {
self.core.sieve.trusted_scripts.get(name).or_else(|| {
trc::event!(
Sieve(trc::SieveEvent::ScriptNotFound),
Id = name.to_string(),
SpanId = session_id,
);
None
})
}
pub fn get_untrusted_sieve_script(&self, name: &str, session_id: u64) -> Option<&Arc<Sieve>> {
self.core.sieve.untrusted_scripts.get(name).or_else(|| {
trc::event!(
Sieve(trc::SieveEvent::ScriptNotFound),
Id = name.to_string(),
SpanId = session_id,
);
None
})
}
pub fn get_route_or_default(&self, name: &str, session_id: u64) -> &RoutingStrategy {
static LOCAL_GATEWAY: RoutingStrategy = RoutingStrategy::Local;
static MX_GATEWAY: RoutingStrategy = RoutingStrategy::Mx(MxConfig {
max_mx: 5,
max_multi_homed: 2,
ip_lookup_strategy: IpLookupStrategy::Ipv4thenIpv6,
});
self.core
.smtp
.queue
.routing_strategy
.get(name)
.unwrap_or_else(|| match name {
"local" => &LOCAL_GATEWAY,
"mx" => &MX_GATEWAY,
_ => {
trc::event!(
Smtp(trc::SmtpEvent::IdNotFound),
Id = name.to_string(),
Details = "Gateway not found",
SpanId = session_id,
);
&MX_GATEWAY
}
})
}
pub fn get_virtual_queue_or_default(&self, name: &QueueName) -> &VirtualQueue {
static DEFAULT_QUEUE: VirtualQueue = VirtualQueue { threads: 25 };
self.core
.smtp
.queue
.virtual_queues
.get(name)
.unwrap_or_else(|| {
if name != &DEFAULT_QUEUE_NAME {
trc::event!(
Smtp(trc::SmtpEvent::IdNotFound),
Id = name.to_string(),
Details = "Virtual queue not found",
);
}
&DEFAULT_QUEUE
})
}
pub fn get_queue_or_default(&self, name: &str, session_id: u64) -> &QueueStrategy {
static DEFAULT_SCHEDULE: LazyLock<QueueStrategy> = LazyLock::new(|| QueueStrategy {
retry: vec![
120, // 2 minutes
300, // 5 minutes
600, // 10 minutes
900, // 15 minutes
1800, // 30 minutes
3600, // 1 hour
7200, // 2 hours
],
notify: vec![
86400, // 1 day
259200, // 3 days
],
expiry: QueueExpiry::Ttl(432000), // 5 days
virtual_queue: QueueName::default(),
});
self.core
.smtp
.queue
.queue_strategy
.get(name)
.unwrap_or_else(|| {
if name != "default" {
trc::event!(
Smtp(trc::SmtpEvent::IdNotFound),
Id = name.to_string(),
Details = "Queue strategy not found",
SpanId = session_id,
);
}
&DEFAULT_SCHEDULE
})
}
pub fn get_tls_or_default(&self, name: &str, session_id: u64) -> &TlsStrategy {
static DEFAULT_TLS: TlsStrategy = TlsStrategy {
dane: RequireOptional::Optional,
mta_sts: RequireOptional::Optional,
tls: RequireOptional::Optional,
allow_invalid_certs: false,
timeout_tls: Duration::from_secs(3 * 60),
timeout_mta_sts: Duration::from_secs(5 * 60),
};
self.core
.smtp
.queue
.tls_strategy
.get(name)
.unwrap_or_else(|| {
if name != "default" {
trc::event!(
Smtp(trc::SmtpEvent::IdNotFound),
Id = name.to_string(),
Details = "TLS strategy not found",
SpanId = session_id,
);
}
&DEFAULT_TLS
})
}
pub fn get_connection_or_default(&self, name: &str, session_id: u64) -> &ConnectionStrategy {
static DEFAULT_CONNECTION: ConnectionStrategy = ConnectionStrategy {
source_ipv4: Vec::new(),
source_ipv6: Vec::new(),
ehlo_hostname: None,
timeout_connect: Duration::from_secs(5 * 60),
timeout_greeting: Duration::from_secs(5 * 60),
timeout_ehlo: Duration::from_secs(5 * 60),
timeout_mail: Duration::from_secs(5 * 60),
timeout_rcpt: Duration::from_secs(5 * 60),
timeout_data: Duration::from_secs(10 * 60),
};
self.core
.smtp
.queue
.connection_strategy
.get(name)
.unwrap_or_else(|| {
if name != "default" {
trc::event!(
Smtp(trc::SmtpEvent::IdNotFound),
Id = name.to_string(),
Details = "Connection strategy not found",
SpanId = session_id,
);
}
&DEFAULT_CONNECTION
})
}
pub async fn spam_model_reload(&self) -> trc::Result<()> {
if self.core.spam.classifier.is_some() {
if let Some(model) = self
.blob_store()
.get_blob(SPAM_CLASSIFIER_KEY, 0..usize::MAX)
.await
.and_then(|archive| match archive {
Some(archive) => <Archive<AlignedBytes> as Deserialize>::deserialize(&archive)
.and_then(|archive| archive.deserialize_untrusted::<SpamClassifier>())
.map(Some),
None => Ok(None),
})
.caused_by(trc::location!())?
{
self.inner.data.spam_classifier.store(Arc::new(model));
} else {
trc::event!(Spam(SpamEvent::ModelNotFound));
}
}
Ok(())
}
pub async fn total_queued_messages(&self) -> trc::Result<u64> {
let mut total = 0;
self.store()
.iterate(
IterateParams::new(
ValueKey::from(ValueClass::Queue(QueueClass::Message(0))),
ValueKey::from(ValueClass::Queue(QueueClass::Message(u64::MAX))),
)
.no_values(),
|_, _| {
total += 1;
Ok(true)
},
)
.await
.caused_by(trc::location!())
.map(|_| total)
}
}

8
crates/common/src/listener/blocked.rs → crates/common/src/network/security.rs
View File

@@ -6,7 +6,8 @@
use crate::{
KV_RATE_LIMIT_AUTH, KV_RATE_LIMIT_LOITER, KV_RATE_LIMIT_RCPT, KV_RATE_LIMIT_SCAN, Server,
ip_to_bytes, ipc::BroadcastEvent,
ipc::{BroadcastEvent, RegistryChange},
network::ip_to_bytes,
};
use ahash::AHashSet;
use registry::{
@@ -233,7 +234,8 @@ impl Server {
// Write blocked IP to config
let now = now() as i64;
self.registry()
let id = self
.registry()
.insert(&BlockedIp {
address: IpAddrOrMask::from_ip(ip),
created_at: UTCDateTime::from_timestamp(now),
@@ -249,7 +251,7 @@ impl Server {
.caused_by(trc::location!())?;
// Increment version
self.cluster_broadcast(BroadcastEvent::ReloadBlockedIps)
self.cluster_broadcast(BroadcastEvent::RegistryChange(RegistryChange::Insert(id)))
.await;
Ok(())

0
crates/common/src/listener/stream.rs → crates/common/src/network/stream.rs
View File

0
crates/common/src/listener/tls.rs → crates/common/src/network/tls.rs
View File

22
crates/common/src/scripts/plugins/lookup.rs
View File

@@ -102,23 +102,13 @@ pub async fn exec_local_domain(ctx: PluginContext<'_>) -> trc::Result<Variable>
let domain = ctx.arguments[0].to_string();
if !domain.is_empty() {
return match &ctx.arguments[0] {
Variable::String(v) if !v.is_empty() => {
ctx.server.core.storage.directories.get(v.as_ref())
}
_ => Some(&ctx.server.core.storage.directory),
}
.ok_or_else(|| {
trc::SieveEvent::RuntimeError
.ctx(trc::Key::Id, ctx.arguments[0].to_string().into_owned())
.details("Unknown directory")
})?
.is_local_domain(domain.as_ref())
.await
.map(Into::into);
ctx.server
.domain(domain.as_ref())
.await
.map(|result| Variable::from(result.is_some()))
} else {
Ok(Variable::default())
}
Ok(Variable::default())
}
#[derive(Debug, PartialEq, Eq)]

2
crates/common/src/scripts/plugins/query.rs
View File

@@ -20,7 +20,7 @@ pub async fn exec(ctx: PluginContext<'_>) -> trc::Result<Variable> {
Variable::String(v) if !v.is_empty() => ctx
.server
.get_lookup_store(v.as_str())
.and_then(|v| v.as_store().cloned()),
.and_then(|v| v.into_store()),
_ => Some(ctx.server.core.storage.data.clone()),
}
.ok_or_else(|| {

19
crates/common/src/sharing/acl.rs
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::Server;
use crate::{Server, ipc::CacheInvalidation};
use types::acl::{AclGrant, ArchivedAclGrant};
impl Server {
@@ -20,7 +20,8 @@ impl Server {
}
}
if invalidate {
changed_principals.push(current_item.account_id);
changed_principals
.push(CacheInvalidation::AccessToken(current_item.account_id));
}
}
@@ -33,16 +34,16 @@ impl Server {
}
}
if invalidate {
changed_principals.push(change_item.account_id);
changed_principals.push(CacheInvalidation::AccessToken(change_item.account_id));
}
}
} else {
for value in acl_changes {
changed_principals.push(value.account_id);
changed_principals.push(CacheInvalidation::AccessToken(value.account_id));
}
}
self.invalidate_principal_caches(changed_principals).await;
self.invalidate_caches(changed_principals, true).await;
}
pub async fn refresh_archived_acls(
@@ -60,7 +61,9 @@ impl Server {
}
}
if invalidate {
changed_principals.push(current_item.account_id.to_native());
changed_principals.push(CacheInvalidation::AccessToken(
current_item.account_id.to_native(),
));
}
}
@@ -73,10 +76,10 @@ impl Server {
}
}
if invalidate {
changed_principals.push(change_item.account_id);
changed_principals.push(CacheInvalidation::AccessToken(change_item.account_id));
}
}
self.invalidate_principal_caches(changed_principals).await;
self.invalidate_caches(changed_principals, true).await;
}
}

111
crates/common/src/storage/archive.rs Normal file
View File

@@ -0,0 +1,111 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::Server;
use store::{
Deserialize, IterateParams, U32_LEN, ValueKey,
dispatch::DocumentSet,
write::{AlignedBytes, Archive, ValueClass, key::DeserializeBigEndian},
};
use trc::AddContext;
use types::{collection::Collection, field::Field};
impl Server {
pub async fn archives<I, CB>(
&self,
account_id: u32,
collection: Collection,
documents: &I,
mut cb: CB,
) -> trc::Result<()>
where
I: DocumentSet + Send + Sync,
CB: FnMut(u32, Archive<AlignedBytes>) -> trc::Result<bool> + Send + Sync,
{
let collection: u8 = collection.into();
self.core
.storage
.data
.iterate(
IterateParams::new(
ValueKey {
account_id,
collection,
document_id: documents.min(),
class: ValueClass::Property(Field::ARCHIVE.into()),
},
ValueKey {
account_id,
collection,
document_id: documents.max(),
class: ValueClass::Property(Field::ARCHIVE.into()),
},
),
|key, value| {
let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?;
if documents.contains(document_id) {
<Archive<AlignedBytes> as Deserialize>::deserialize(value)
.and_then(|archive| cb(document_id, archive))
} else {
Ok(true)
}
},
)
.await
.add_context(|err| {
err.caused_by(trc::location!())
.account_id(account_id)
.collection(collection)
})
}
pub async fn all_archives<CB>(
&self,
account_id: u32,
collection: Collection,
field: u8,
mut cb: CB,
) -> trc::Result<()>
where
CB: FnMut(u32, Archive<AlignedBytes>) -> trc::Result<()> + Send + Sync,
{
let collection: u8 = collection.into();
self.core
.storage
.data
.iterate(
IterateParams::new(
ValueKey {
account_id,
collection,
document_id: 0,
class: ValueClass::Property(field),
},
ValueKey {
account_id,
collection,
document_id: u32::MAX,
class: ValueClass::Property(field),
},
),
|key, value| {
let document_id = key.deserialize_be_u32(key.len() - U32_LEN)?;
let archive = <Archive<AlignedBytes> as Deserialize>::deserialize(value)?;
cb(document_id, archive)?;
Ok(true)
},
)
.await
.add_context(|err| {
err.caused_by(trc::location!())
.account_id(account_id)
.collection(collection)
})
}
}

140
crates/common/src/storage/blob.rs
View File

@@ -9,9 +9,147 @@ use mail_parser::{
Encoding,
decoders::{base64::base64_decode, quoted_printable::quoted_printable_decode},
};
use types::{blob::BlobSection, blob_hash::BlobHash};
use store::{
SerializeInfallible,
write::{BatchBuilder, BlobLink, BlobOp, now},
};
use trc::AddContext;
use types::{
blob::{BlobClass, BlobId, BlobSection},
blob_hash::BlobHash,
};
impl Server {
#[allow(clippy::blocks_in_conditions)]
pub async fn put_jmap_blob(&self, account_id: u32, data: &[u8]) -> trc::Result<BlobId> {
// First reserve the hash
let hash = BlobHash::generate(data);
let mut batch = BatchBuilder::new();
let until = now() + self.core.jmap.upload_tmp_ttl;
batch
.with_account_id(account_id)
.set(
BlobOp::Link {
hash: hash.clone(),
to: BlobLink::Temporary { until },
},
vec![BlobLink::QUOTA_LINK],
)
.set(
BlobOp::Quota {
hash: hash.clone(),
until,
},
(data.len() as u32).serialize(),
);
self.core
.storage
.data
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
if !self
.core
.storage
.data
.blob_exists(&hash)
.await
.caused_by(trc::location!())?
{
// Upload blob to store
self.core
.storage
.blob
.put_blob(hash.as_ref(), data, self.core.storage.compression)
.await
.caused_by(trc::location!())?;
// Commit blob
let mut batch = BatchBuilder::new();
batch.set(BlobOp::Commit { hash: hash.clone() }, Vec::new());
self.core
.storage
.data
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
Ok(BlobId {
hash,
class: BlobClass::Reserved {
account_id,
expires: until,
},
section: None,
})
}
pub async fn put_temporary_blob(
&self,
account_id: u32,
data: &[u8],
hold_for: u64,
) -> trc::Result<(BlobHash, BlobOp)> {
// First reserve the hash
let hash = BlobHash::generate(data);
let mut batch = BatchBuilder::new();
let until = now() + hold_for;
batch.with_account_id(account_id).set(
BlobOp::Link {
hash: hash.clone(),
to: BlobLink::Temporary { until },
},
vec![],
);
self.core
.storage
.data
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
if !self
.core
.storage
.data
.blob_exists(&hash)
.await
.caused_by(trc::location!())?
{
// Upload blob to store
self.core
.storage
.blob
.put_blob(hash.as_ref(), data, self.core.storage.compression)
.await
.caused_by(trc::location!())?;
// Commit blob
let mut batch = BatchBuilder::new();
batch.set(BlobOp::Commit { hash: hash.clone() }, Vec::new());
self.core
.storage
.data
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
Ok((
hash.clone(),
BlobOp::Link {
hash,
to: BlobLink::Temporary { until },
},
))
}
pub async fn get_blob_section(
&self,
hash: &BlobHash,

369
crates/common/src/storage/dav.rs Normal file
View File

@@ -0,0 +1,369 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
DavName, DavPath, DavResource, DavResourceMetadata, DavResourcePath, DavResources,
TinyCalendarPreferences,
};
use std::hash::{Hash, Hasher};
use store::rand::{Rng, distr::Alphanumeric};
use types::acl::AclGrant;
const SCHEDULE_INBOX_ID: u32 = u32::MAX - 1;
impl DavResourcePath<'_> {
#[inline(always)]
pub fn document_id(&self) -> u32 {
self.resource.document_id
}
#[inline(always)]
pub fn parent_id(&self) -> Option<u32> {
self.path.parent_id
}
#[inline(always)]
pub fn path(&self) -> &str {
self.path.path.as_str()
}
#[inline(always)]
pub fn is_container(&self) -> bool {
self.resource.is_container()
}
#[inline(always)]
pub fn hierarchy_seq(&self) -> u32 {
self.path.hierarchy_seq
}
#[inline(always)]
pub fn size(&self) -> u32 {
self.resource.size().unwrap_or_default()
}
}
impl DavResources {
pub fn by_path(&self, name: &str) -> Option<DavResourcePath<'_>> {
self.paths.get(name).map(|path| DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
}
pub fn container_resource_by_id(&self, id: u32) -> Option<&DavResource> {
self.resources
.iter()
.find(|res| res.document_id == id && res.is_container())
}
pub fn container_resource_path_by_id(&self, id: u32) -> Option<DavResourcePath<'_>> {
self.resources
.iter()
.enumerate()
.find(|(_, resource)| resource.document_id == id && resource.is_container())
.and_then(|(idx, resource)| {
self.paths
.iter()
.find(|path| path.resource_idx == idx)
.map(|path| DavResourcePath { path, resource })
})
}
pub fn any_resource_path_by_id(&self, id: u32) -> Option<DavResourcePath<'_>> {
self.resources
.iter()
.enumerate()
.find(|(_, resource)| resource.document_id == id)
.and_then(|(idx, resource)| {
self.paths
.iter()
.find(|path| path.resource_idx == idx)
.map(|path| DavResourcePath { path, resource })
})
}
pub fn subtree(&self, search_path: &str) -> impl Iterator<Item = DavResourcePath<'_>> {
let prefix = format!("{search_path}/");
self.paths.iter().filter_map(move |path| {
if path.path.starts_with(&prefix) || path.path == search_path {
Some(DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
} else {
None
}
})
}
pub fn subtree_with_depth(
&self,
search_path: &str,
depth: usize,
) -> impl Iterator<Item = DavResourcePath<'_>> {
let prefix = format!("{search_path}/");
self.paths.iter().filter_map(move |path| {
if path
.path
.strip_prefix(&prefix)
.is_some_and(|name| name.as_bytes().iter().filter(|&&c| c == b'/').count() < depth)
|| path.path.as_str() == search_path
{
Some(DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
} else {
None
}
})
}
pub fn tree_with_depth(&self, depth: usize) -> impl Iterator<Item = DavResourcePath<'_>> {
self.paths.iter().filter_map(move |path| {
if path.path.as_bytes().iter().filter(|&&c| c == b'/').count() <= depth {
Some(DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
} else {
None
}
})
}
pub fn children(&self, parent_id: u32) -> impl Iterator<Item = DavResourcePath<'_>> {
self.paths
.iter()
.filter(move |item| item.parent_id.is_some_and(|id| id == parent_id))
.map(|path| DavResourcePath {
path,
resource: &self.resources[path.resource_idx],
})
}
pub fn children_ids(&self, parent_id: u32) -> impl Iterator<Item = u32> {
self.paths
.iter()
.filter(move |item| item.parent_id.is_some_and(|id| id == parent_id))
.map(|path| self.resources[path.resource_idx].document_id)
}
pub fn format_resource(&self, resource: DavResourcePath<'_>) -> String {
if resource.resource.is_container() {
format!("{}{}/", self.base_path, resource.path.path)
} else {
format!("{}{}", self.base_path, resource.path.path)
}
}
pub fn format_collection(&self, name: &str) -> String {
format!("{}{name}/", self.base_path)
}
pub fn format_item(&self, name: &str) -> String {
format!("{}{}", self.base_path, name)
}
}
impl DavResource {
pub fn is_child_of(&self, parent_id: u32) -> bool {
match &self.data {
DavResourceMetadata::File { parent_id: id, .. } => id.is_some_and(|id| id == parent_id),
DavResourceMetadata::CalendarEvent { names, .. } => {
names.iter().any(|name| name.parent_id == parent_id)
}
DavResourceMetadata::ContactCard { names } => {
names.iter().any(|name| name.parent_id == parent_id)
}
DavResourceMetadata::CalendarEventNotification { names } => {
names.is_empty() && parent_id == SCHEDULE_INBOX_ID
}
_ => false,
}
}
pub fn parent_id(&self) -> Option<u32> {
match &self.data {
DavResourceMetadata::File { parent_id, .. } => *parent_id,
DavResourceMetadata::CalendarEvent { names, .. } => {
names.first().map(|name| name.parent_id)
}
DavResourceMetadata::ContactCard { names } => names.first().map(|name| name.parent_id),
DavResourceMetadata::CalendarEventNotification { names } if names.is_empty() => {
Some(SCHEDULE_INBOX_ID)
}
_ => None,
}
}
pub fn child_names(&self) -> Option<&[DavName]> {
match &self.data {
DavResourceMetadata::CalendarEvent { names, .. } => Some(names.as_slice()),
DavResourceMetadata::ContactCard { names } => Some(names.as_slice()),
DavResourceMetadata::CalendarEventNotification { names } if !names.is_empty() => {
Some(names.as_slice())
}
_ => None,
}
}
pub fn container_name(&self) -> Option<&str> {
match &self.data {
DavResourceMetadata::File { name, .. } => Some(name.as_str()),
DavResourceMetadata::Calendar { name, .. } => Some(name.as_str()),
DavResourceMetadata::AddressBook { name, .. } => Some(name.as_str()),
DavResourceMetadata::CalendarEventNotification { names } if names.is_empty() => {
Some(if self.document_id == SCHEDULE_INBOX_ID {
"inbox"
} else {
"outbox"
})
}
_ => None,
}
}
pub fn has_hierarchy_changes(&self, other: &DavResource) -> bool {
match (&self.data, &other.data) {
(
DavResourceMetadata::File {
name: a,
parent_id: c,
..
},
DavResourceMetadata::File {
name: b,
parent_id: d,
..
},
) => a != b || c != d,
(
DavResourceMetadata::Calendar { name: a, .. },
DavResourceMetadata::Calendar { name: b, .. },
) => a != b,
(
DavResourceMetadata::AddressBook { name: a, .. },
DavResourceMetadata::AddressBook { name: b, .. },
) => a != b,
(
DavResourceMetadata::CalendarEvent { names: a, .. },
DavResourceMetadata::CalendarEvent { names: b, .. },
) => a != b,
(
DavResourceMetadata::ContactCard { names: a, .. },
DavResourceMetadata::ContactCard { names: b, .. },
) => a != b,
(
DavResourceMetadata::CalendarEventNotification { names: a, .. },
DavResourceMetadata::CalendarEventNotification { names: b, .. },
) => a != b,
_ => unreachable!(),
}
}
pub fn event_time_range(&self) -> Option<(i64, i64)> {
match &self.data {
DavResourceMetadata::CalendarEvent {
start, duration, ..
} => Some((*start, *start + *duration as i64)),
_ => None,
}
}
pub fn calendar_preferences(&self, account_id: u32) -> Option<&TinyCalendarPreferences> {
match &self.data {
DavResourceMetadata::Calendar { preferences, .. } => preferences
.iter()
.find(|pref| pref.account_id == account_id)
.or_else(|| preferences.first()),
_ => None,
}
}
pub fn is_container(&self) -> bool {
match &self.data {
DavResourceMetadata::File { size, .. } => size.is_none(),
DavResourceMetadata::Calendar { .. } | DavResourceMetadata::AddressBook { .. } => true,
DavResourceMetadata::CalendarEventNotification { names } => names.is_empty(),
_ => false,
}
}
pub fn size(&self) -> Option<u32> {
match &self.data {
DavResourceMetadata::File { size, .. } => *size,
_ => None,
}
}
pub fn acls(&self) -> Option<&[AclGrant]> {
match &self.data {
DavResourceMetadata::File { acls, .. } => Some(acls.as_slice()),
DavResourceMetadata::Calendar { acls, .. } => Some(acls.as_slice()),
DavResourceMetadata::AddressBook { acls, .. } => Some(acls.as_slice()),
_ => None,
}
}
}
impl Hash for DavPath {
fn hash<H: Hasher>(&self, state: &mut H) {
self.path.hash(state);
}
}
impl PartialEq for DavPath {
fn eq(&self, other: &Self) -> bool {
self.path == other.path
}
}
impl Eq for DavPath {}
impl std::borrow::Borrow<str> for DavPath {
fn borrow(&self) -> &str {
&self.path
}
}
impl std::hash::Hash for DavResource {
fn hash<H: Hasher>(&self, state: &mut H) {
self.document_id.hash(state);
}
}
impl PartialEq for DavResource {
fn eq(&self, other: &Self) -> bool {
self.document_id == other.document_id
}
}
impl Eq for DavResource {}
impl std::borrow::Borrow<u32> for DavResource {
fn borrow(&self) -> &u32 {
&self.document_id
}
}
impl DavName {
pub fn new(name: String, parent_id: u32) -> Self {
Self { name, parent_id }
}
pub fn new_with_rand_name(parent_id: u32) -> Self {
Self {
name: store::rand::rng()
.sample_iter(Alphanumeric)
.take(10)
.map(char::from)
.collect::<String>(),
parent_id,
}
}
}

136
crates/common/src/storage/document.rs Normal file
View File

@@ -0,0 +1,136 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use store::{
IndexKey, IndexKeyPrefix, IterateParams, U32_LEN, roaring::RoaringBitmap,
write::key::DeserializeBigEndian,
};
use trc::AddContext;
use types::collection::Collection;
use crate::Server;
impl Server {
pub async fn document_ids(
&self,
account_id: u32,
collection: Collection,
field: impl Into<u8>,
) -> trc::Result<RoaringBitmap> {
let field = field.into();
let mut results = RoaringBitmap::new();
self.store()
.iterate(
IterateParams::new(
IndexKeyPrefix {
account_id,
collection: collection.into(),
field,
},
IndexKeyPrefix {
account_id,
collection: collection.into(),
field: field + 1,
},
)
.no_values(),
|key, _| {
results.insert(key.deserialize_be_u32(key.len() - U32_LEN)?);
Ok(true)
},
)
.await
.caused_by(trc::location!())
.map(|_| results)
}
pub async fn document_exists(
&self,
account_id: u32,
collection: Collection,
field: impl Into<u8>,
filter: impl AsRef<[u8]>,
) -> trc::Result<bool> {
let field = field.into();
let mut exists = false;
let filter = filter.as_ref();
let key_len = IndexKeyPrefix::len() + filter.len() + U32_LEN;
self.store()
.iterate(
IterateParams::new(
IndexKey {
account_id,
collection: collection.into(),
document_id: 0,
field,
key: filter,
},
IndexKey {
account_id,
collection: collection.into(),
document_id: u32::MAX,
field,
key: filter,
},
)
.no_values(),
|key, _| {
exists = key.len() == key_len;
Ok(!exists)
},
)
.await
.caused_by(trc::location!())
.map(|_| exists)
}
pub async fn document_ids_matching(
&self,
account_id: u32,
collection: Collection,
field: impl Into<u8>,
filter: impl AsRef<[u8]>,
) -> trc::Result<RoaringBitmap> {
let field = field.into();
let filter = filter.as_ref();
let key_len = IndexKeyPrefix::len() + filter.len() + U32_LEN;
let mut results = RoaringBitmap::new();
self.store()
.iterate(
IterateParams::new(
IndexKey {
account_id,
collection: collection.into(),
document_id: 0,
field,
key: filter,
},
IndexKey {
account_id,
collection: collection.into(),
document_id: u32::MAX,
field,
key: filter,
},
)
.no_values(),
|key, _| {
if key.len() == key_len {
results.insert(key.deserialize_be_u32(key.len() - U32_LEN)?);
}
Ok(true)
},
)
.await
.caused_by(trc::location!())
.map(|_| results)
}
}

2
crates/common/src/storage/index.rs
View File

@@ -300,7 +300,7 @@ impl<C: IndexableObject, N: IndexableAndSerializableObject> ObjectIndexBuilder<C
}
pub fn with_access_token(mut self, access_token: &AccessToken) -> Self {
self.tenant_id = access_token.tenant.as_ref().map(|t| t.id);
self.tenant_id = access_token.tenant_id();
self.changed_by = access_token.account_id();
self
}

85
crates/common/src/storage/mod.rs
View File

@@ -4,6 +4,91 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::Server;
use directory::Directory;
use registry::{
schema::{
enums::{StorageQuota, TenantStorageQuota},
prelude::Object,
},
types::EnumType,
};
use std::sync::Arc;
use store::{BlobStore, InMemoryStore, RegistryStore, SearchStore, Store};
pub mod archive;
pub mod blob;
pub mod dav;
pub mod document;
pub mod index;
pub mod quota;
pub mod state;
pub mod transaction;
#[derive(Debug, Clone)]
pub struct ObjectQuota([u32; StorageQuota::COUNT - 1]);
#[derive(Debug, Clone)]
pub struct TenantQuota([u32; TenantStorageQuota::COUNT - 1]);
impl Server {
#[inline(always)]
pub fn registry(&self) -> &RegistryStore {
&self.core.storage.registry
}
#[inline(always)]
pub fn store(&self) -> &Store {
&self.core.storage.data
}
#[inline(always)]
pub fn blob_store(&self) -> &BlobStore {
&self.core.storage.blob
}
#[inline(always)]
pub fn search_store(&self) -> &SearchStore {
&self.core.storage.fts
}
#[inline(always)]
pub fn in_memory_store(&self) -> &InMemoryStore {
&self.core.storage.memory
}
#[inline(always)]
pub fn get_directory(&self, id: &u32) -> Option<&Arc<Directory>> {
self.core.storage.directories.get(id)
}
#[inline(always)]
pub fn get_default_directory(&self) -> Option<&Arc<Directory>> {
self.core.storage.directory.as_ref()
}
#[inline(always)]
pub fn get_lookup_store(&self, name: &str) -> Option<InMemoryStore> {
if !name.is_empty() && name != "*" {
self.inner.data.lookup_stores.load().get(name).cloned()
} else {
self.in_memory_store().clone().into()
}
}
pub async fn total_accounts(&self) -> trc::Result<u64> {
self.registry().count(Object::Account).await
}
pub async fn total_domains(&self) -> trc::Result<u64> {
self.registry().count(Object::Domain).await
}
#[cfg(not(feature = "enterprise"))]
pub async fn logo_resource(
&self,
_: &str,
) -> trc::Result<Option<crate::manager::webadmin::Resource<Vec<u8>>>> {
Ok(None)
}
}

105
crates/common/src/storage/quota.rs Normal file
View File

@@ -0,0 +1,105 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use registry::{
schema::enums::{StorageQuota, TenantStorageQuota},
types::EnumType,
};
use store::write::DirectoryClass;
use trc::AddContext;
use crate::{
Server,
storage::{ObjectQuota, TenantQuota},
};
impl Server {
pub async fn get_used_quota(&self, account_id: u32) -> trc::Result<i64> {
self.core
.storage
.data
.get_counter(DirectoryClass::UsedQuota(account_id))
.await
.add_context(|err| err.caused_by(trc::location!()).account_id(account_id))
}
pub async fn has_available_quota(&self, account_id: u32, item_size: u64) -> trc::Result<()> {
let account = self.account(account_id).await.caused_by(trc::location!())?;
if account.quota_disk != 0 {
let used_quota = self.get_used_quota(account_id).await? as u64;
if used_quota + item_size > account.quota_disk {
return Err(trc::LimitEvent::Quota
.into_err()
.ctx(trc::Key::Limit, account.quota_disk)
.ctx(trc::Key::Size, used_quota));
}
}
// SPDX-SnippetBegin
// SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
// SPDX-License-Identifier: LicenseRef-SEL
#[cfg(feature = "enterprise")]
if self.core.is_enterprise_edition() && account.id_tenant != u32::MAX {
let tenant = self
.tenant(account.id_tenant)
.await
.caused_by(trc::location!())?;
if tenant.quota_disk != 0 {
let used_quota = self.get_used_quota(account.id_tenant).await? as u64;
if used_quota + item_size > tenant.quota_disk {
return Err(trc::LimitEvent::TenantQuota
.into_err()
.ctx(trc::Key::Limit, tenant.quota_disk)
.ctx(trc::Key::Size, used_quota));
}
}
}
// SPDX-SnippetEnd
Ok(())
}
}
impl ObjectQuota {
#[inline(always)]
pub fn set(&mut self, item: StorageQuota, max: u32) {
self.0[item as usize] = max;
}
#[inline(always)]
pub fn get(&self, item: StorageQuota) -> u32 {
self.0[item as usize]
}
}
impl TenantQuota {
#[inline(always)]
pub fn set(&mut self, item: TenantStorageQuota, max: u32) {
self.0[item as usize] = max;
}
#[inline(always)]
pub fn get(&self, item: TenantStorageQuota) -> u32 {
self.0[item as usize]
}
}
impl Default for ObjectQuota {
fn default() -> Self {
Self([u32::MAX; StorageQuota::COUNT - 1])
}
}
impl Default for TenantQuota {
fn default() -> Self {
Self([u32::MAX; TenantStorageQuota::COUNT - 1])
}
}

45
crates/common/src/storage/state.rs
View File

@@ -7,7 +7,7 @@
use crate::{
IPC_CHANNEL_BUFFER, Server,
auth::AccessToken,
ipc::{PushEvent, PushNotification},
ipc::{BroadcastEvent, PushEvent, PushNotification},
};
use tokio::sync::mpsc;
use types::type_state::DataType;
@@ -37,4 +37,47 @@ impl Server {
Ok(rx)
}
#[inline(always)]
pub fn notify_task_queue(&self) {
self.inner.ipc.task_tx.notify_one();
}
pub async fn broadcast_push_notification(&self, notification: PushNotification) -> bool {
match self
.inner
.ipc
.push_tx
.clone()
.send(PushEvent::Publish {
notification,
broadcast: true,
})
.await
{
Ok(_) => true,
Err(_) => {
trc::event!(
Server(trc::ServerEvent::ThreadError),
Details = "Error sending state change.",
CausedBy = trc::location!()
);
false
}
}
}
pub async fn cluster_broadcast(&self, event: BroadcastEvent) {
let todo = "refactor event names";
if let Some(broadcast_tx) = &self.inner.ipc.broadcast_tx.clone()
&& broadcast_tx.send(event).await.is_err()
{
trc::event!(
Server(trc::ServerEvent::ThreadError),
Details = "Error sending broadcast event.",
CausedBy = trc::location!()
);
}
}
}

198
crates/common/src/storage/transaction.rs Normal file
View File

@@ -0,0 +1,198 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{Server, ipc::PushNotification};
use std::time::Duration;
use store::{
IterateParams, Key, LogKey, SUBSPACE_LOGS, U64_LEN,
write::{AnyClass, AssignedIds, BatchBuilder, ValueClass, key::DeserializeBigEndian},
};
use trc::AddContext;
use types::{
collection::SyncCollection,
type_state::{DataType, StateChange},
};
use utils::{map::bitmap::Bitmap, snowflake::SnowflakeIdGenerator};
impl Server {
pub async fn commit_batch(&self, mut builder: BatchBuilder) -> trc::Result<AssignedIds> {
let mut assigned_ids = AssignedIds::default();
let mut commit_points = builder.commit_points();
for commit_point in commit_points.iter() {
let batch = builder.build_one(commit_point);
assigned_ids
.ids
.extend(self.store().write(batch).await?.ids);
}
if let Some(changes) = builder.changes() {
for (account_id, changed_collections) in changes {
let mut state_change = StateChange::new(account_id);
for changed_collection in changed_collections.changed_containers {
if let Some(data_type) = DataType::try_from_sync(changed_collection, true) {
state_change.set_change(data_type);
}
}
for changed_collection in changed_collections.changed_items {
if let Some(data_type) = DataType::try_from_sync(changed_collection, false) {
state_change.set_change(data_type);
}
}
if state_change.has_changes() {
self.broadcast_push_notification(PushNotification::StateChange(
state_change.with_change_id(assigned_ids.last_change_id(account_id)?),
))
.await;
}
if let Some(change_id) = changed_collections.share_notification_id {
self.broadcast_push_notification(PushNotification::StateChange(StateChange {
account_id,
change_id,
types: Bitmap::from_iter([DataType::ShareNotification]),
}))
.await;
}
}
}
Ok(assigned_ids)
}
pub async fn delete_changes(
&self,
account_id: u32,
max_entries: Option<usize>,
max_duration: Option<Duration>,
) -> trc::Result<()> {
if let Some(max_entries) = max_entries {
for sync_collection in [
SyncCollection::Email,
SyncCollection::Thread,
SyncCollection::Identity,
SyncCollection::EmailSubmission,
SyncCollection::SieveScript,
SyncCollection::FileNode,
SyncCollection::AddressBook,
SyncCollection::Calendar,
SyncCollection::CalendarEventNotification,
] {
let collection = sync_collection.into();
let from_key = LogKey {
account_id,
collection,
change_id: 0,
};
let to_key = LogKey {
account_id,
collection,
change_id: u64::MAX,
};
let mut first_change_id = 0;
let mut num_changes = 0;
self.store()
.iterate(
IterateParams::new(from_key, to_key)
.descending()
.no_values(),
|key, _| {
first_change_id = key.deserialize_be_u64(key.len() - U64_LEN)?;
num_changes += 1;
Ok(num_changes <= max_entries)
},
)
.await
.caused_by(trc::location!())?;
if num_changes > max_entries {
self.store()
.delete_range(
LogKey {
account_id,
collection,
change_id: 0,
},
LogKey {
account_id,
collection,
change_id: first_change_id,
},
)
.await
.caused_by(trc::location!())?;
// Delete vanished items
if let Some(vanished_collection) =
sync_collection.vanished_collection().map(u8::from)
{
self.store()
.delete_range(
LogKey {
account_id,
collection: vanished_collection,
change_id: 0,
},
LogKey {
account_id,
collection: vanished_collection,
change_id: first_change_id,
},
)
.await
.caused_by(trc::location!())?;
}
// Write truncation entry for cache
let mut batch = BatchBuilder::new();
batch.with_account_id(account_id).set(
ValueClass::Any(AnyClass {
subspace: SUBSPACE_LOGS,
key: LogKey {
account_id,
collection,
change_id: first_change_id,
}
.serialize(0),
}),
Vec::new(),
);
self.store()
.write(batch.build_all())
.await
.caused_by(trc::location!())?;
}
}
}
if let Some(max_duration) = max_duration {
self.store()
.delete_range(
LogKey {
account_id,
collection: SyncCollection::ShareNotification.into(),
change_id: 0,
},
LogKey {
account_id,
collection: SyncCollection::ShareNotification.into(),
change_id: SnowflakeIdGenerator::from_duration(max_duration)
.unwrap_or_default(),
},
)
.await
.caused_by(trc::location!())?;
}
Ok(())
}
#[inline(always)]
pub fn generate_snowflake_id(&self) -> u64 {
self.inner.data.jmap_id_gen.generate()
}
}

7
crates/dav/src/calendar/copy_move.rs
View File

@@ -990,12 +990,7 @@ async fn copy_container(
if from_account_id != to_account_id && required_space > 0 {
server
.has_available_quota(
&server
.get_resource_token(access_token, to_account_id)
.await?,
required_space,
)
.has_available_quota(to_account_id, required_space)
.await?;
}

14
crates/directory/src/core/config.rs
View File

@@ -11,7 +11,7 @@ use crate::{
use ahash::AHashMap;
use registry::schema::{
prelude::Object,
structs::{self, DefaultDirectory},
structs::{self, Authentication},
};
use std::sync::Arc;
use store::registry::bootstrap::Bootstrap;
@@ -40,20 +40,20 @@ impl Directories {
}
}
let default_directory = match bp.setting_infallible::<DefaultDirectory>().await {
DefaultDirectory::Internal => Ok(None),
DefaultDirectory::Ldap(directory) => LdapDirectory::open(directory).map(Some),
DefaultDirectory::Sql(directory) => SqlDirectory::open(directory, &bp.data_store)
let default_directory = match bp.setting_infallible::<Authentication>().await {
Authentication::Internal => Ok(None),
Authentication::Ldap(directory) => LdapDirectory::open(directory).map(Some),
Authentication::Sql(directory) => SqlDirectory::open(directory, &bp.data_store)
.await
.map(Some),
DefaultDirectory::Oidc(directory) => OpenIdDirectory::open(directory).map(Some),
Authentication::Oidc(directory) => OpenIdDirectory::open(directory).map(Some),
};
Directories {
default_directory: match default_directory {
Ok(default_directory) => default_directory.map(Arc::new),
Err(err) => {
bp.build_error(Object::DefaultDirectory.singleton(), err);
bp.build_error(Object::Authentication.singleton(), err);
None
}
},

9
crates/registry/src/types/id.rs
View File

@@ -19,8 +19,11 @@ pub struct Id {
}
impl Id {
pub fn new(object: Object, id: u64) -> Self {
Self { object, id }
pub fn new(object: Object, id: impl Into<u64>) -> Self {
Self {
object,
id: id.into(),
}
}
pub fn id(&self) -> u64 {
@@ -50,7 +53,7 @@ impl Object {
}
pub fn singleton(&self) -> Id {
Id::new(*self, 20080258862541)
Id::new(*self, 20080258862541u64)
}
}

2
crates/store/src/dispatch/lookup.rs
View File

@@ -516,7 +516,7 @@ impl InMemoryStore {
}
}
pub fn as_store(&self) -> Option<&Store> {
pub fn into_store(self) -> Option<Store> {
match self {
InMemoryStore::Store(store) => Some(store),
_ => None,

12
crates/store/src/dispatch/registry.rs
View File

@@ -11,11 +11,19 @@ use registry::{
};
impl RegistryStore {
pub async fn get<T: ObjectType>(&self, id: Id) -> trc::Result<Option<T>> {
pub async fn id<T: ObjectType>(&self, id: Id) -> trc::Result<Option<T>> {
todo!()
}
pub async fn insert<T: ObjectType>(&self, object: &T) -> trc::Result<bool> {
pub async fn object<T: ObjectType>(&self, id: impl Into<u64>) -> trc::Result<Option<T>> {
self.id(Id::new(T::object(), id.into())).await
}
pub async fn singleton<T: ObjectType>(&self) -> trc::Result<Option<T>> {
self.id(T::object().singleton()).await
}
pub async fn insert<T: ObjectType>(&self, object: &T) -> trc::Result<Id> {
todo!()
}

4
crates/store/src/registry/bootstrap.rs
View File

@@ -43,7 +43,7 @@ impl Bootstrap {
pub async fn setting<T: ObjectType>(&mut self) -> trc::Result<T> {
let object_id = T::object().singleton();
if let Some(setting) = self.registry.get::<T>(object_id).await? {
if let Some(setting) = self.registry.id::<T>(object_id).await? {
let mut errors = Vec::new();
if setting.validate(&mut errors) {
return Ok(setting);
@@ -72,7 +72,7 @@ impl Bootstrap {
pub async fn get_infallible<T: ObjectType>(&mut self, id: Id) -> Option<T> {
if id.object() != T::object() {
match self.registry.get::<T>(id).await {
match self.registry.id::<T>(id).await {
Ok(Some(setting)) => {
let mut errors = Vec::new();
if setting.validate(&mut errors) {