diff --git a/crates/cli/src/modules/cli.rs b/crates/cli/src/modules/cli.rs index 249cfc99..b5ffbb67 100644 --- a/crates/cli/src/modules/cli.rs +++ b/crates/cli/src/modules/cli.rs @@ -402,8 +402,32 @@ pub enum ExportCommands { pub enum ServerCommands { /// Perform database maintenance DatabaseMaintenance {}, + /// Reload TLS certificates ReloadCertificates {}, + + /// Reload configuration + ReloadConfig {}, + + /// Create a new configuration key + AddConfig { + /// Key to add + key: String, + /// Value to set + value: Option, + }, + + /// Delete a configuration key or prefix + DeleteConfig { + /// Configuration key or prefix to delete + key: String, + }, + + /// List all configuration entries + ListConfig { + /// Prefix to filter configuration entries by + prefix: Option, + }, } #[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)] diff --git a/crates/cli/src/modules/database.rs b/crates/cli/src/modules/database.rs index 7317951d..44d0bb0d 100644 --- a/crates/cli/src/modules/database.rs +++ b/crates/cli/src/modules/database.rs @@ -21,6 +21,7 @@ * for more details. */ +use prettytable::{Attr, Cell, Row, Table}; use reqwest::Method; use serde_json::Value; @@ -37,10 +38,67 @@ impl ServerCommands { } ServerCommands::ReloadCertificates {} => { client - .http_request::(Method::GET, "/admin/certificates/reload", None) + .http_request::(Method::GET, "/admin/reload/certificates", None) .await; eprintln!("Success."); } + ServerCommands::ReloadConfig {} => { + client + .http_request::(Method::GET, "/admin/reload/config", None) + .await; + eprintln!("Success."); + } + ServerCommands::AddConfig { key, value } => { + client + .http_request::( + Method::POST, + "/admin/config", + Some(vec![(key.clone(), value.unwrap_or_default())]), + ) + .await; + eprintln!("Successfully added key {key}."); + } + ServerCommands::DeleteConfig { key } => { + client + .http_request::( + Method::DELETE, + &format!("/admin/config/{key}"), + None, + ) + .await; + eprintln!("Successfully deleted key {key}."); + } + ServerCommands::ListConfig { prefix } => { + let results = client + .http_request::, String>( + Method::GET, + &format!("/admin/config/{}", prefix.unwrap_or_default()), + None, + ) + .await; + + if !results.is_empty() { + let mut table = Table::new(); + table.add_row(Row::new(vec![ + Cell::new("Key").with_style(Attr::Bold), + Cell::new("Value").with_style(Attr::Bold), + ])); + + for (key, value) in &results { + table.add_row(Row::new(vec![Cell::new(key), Cell::new(value)])); + } + + eprintln!(); + table.printstd(); + eprintln!(); + } + + eprintln!( + "\n\n{} key{} found.\n", + results.len(), + if results.len() == 1 { "" } else { "s" } + ); + } } } } diff --git a/crates/directory/src/backend/imap/config.rs b/crates/directory/src/backend/imap/config.rs index cd621a92..81db4e96 100644 --- a/crates/directory/src/backend/imap/config.rs +++ b/crates/directory/src/backend/imap/config.rs @@ -22,6 +22,7 @@ */ use mail_send::smtp::tls::build_tls_connector; +use store::Store; use utils::config::{utils::AsKey, Config}; use crate::core::config::build_pool; @@ -29,7 +30,11 @@ use crate::core::config::build_pool; use super::{ImapConnectionManager, ImapDirectory}; impl ImapDirectory { - pub fn from_config(config: &Config, prefix: impl AsKey) -> utils::config::Result { + pub fn from_config( + config: &Config, + prefix: impl AsKey, + data_store: Store, + ) -> utils::config::Result { let prefix = prefix.as_key(); let address = config.value_require((&prefix, "address"))?; let tls_implicit: bool = config.property_or_static((&prefix, "tls.implicit"), "false")?; @@ -53,6 +58,7 @@ impl ImapDirectory { .values((&prefix, "lookup.domains")) .map(|(_, v)| v.to_lowercase()) .collect(), + data_store, }) } } diff --git a/crates/directory/src/backend/imap/mod.rs b/crates/directory/src/backend/imap/mod.rs index 7556bab7..987d369e 100644 --- a/crates/directory/src/backend/imap/mod.rs +++ b/crates/directory/src/backend/imap/mod.rs @@ -31,12 +31,14 @@ use std::{fmt::Display, sync::atomic::AtomicU64, time::Duration}; use ahash::AHashSet; use deadpool::managed::Pool; +use store::Store; use tokio::io::{AsyncRead, AsyncWrite}; use tokio_rustls::TlsConnector; pub struct ImapDirectory { pool: Pool, domains: AHashSet, + pub(crate) data_store: Store, } pub struct ImapConnectionManager { diff --git a/crates/directory/src/backend/ldap/config.rs b/crates/directory/src/backend/ldap/config.rs index c4bf563e..d0eda70c 100644 --- a/crates/directory/src/backend/ldap/config.rs +++ b/crates/directory/src/backend/ldap/config.rs @@ -33,7 +33,7 @@ impl LdapDirectory { pub fn from_config( config: &Config, prefix: impl AsKey, - id_store: Option, + data_store: Store, ) -> utils::config::Result { let prefix = prefix.as_key(); let bind_dn = if let Some(dn) = config.value((&prefix, "bind.dn")) { @@ -123,7 +123,7 @@ impl LdapDirectory { mappings, pool: build_pool(config, &prefix, manager)?, auth_bind, - id_store, + data_store, }) } } diff --git a/crates/directory/src/backend/ldap/lookup.rs b/crates/directory/src/backend/ldap/lookup.rs index 39578225..cb503fa0 100644 --- a/crates/directory/src/backend/ldap/lookup.rs +++ b/crates/directory/src/backend/ldap/lookup.rs @@ -23,7 +23,6 @@ use ldap3::{Ldap, LdapConnAsync, LdapError, Scope, SearchEntry}; use mail_send::Credentials; -use store::Store; use crate::{backend::internal::manage::ManageDirectory, DirectoryError, Principal, QueryBy, Type}; @@ -53,7 +52,7 @@ impl LdapDirectory { } } QueryBy::Id(uid) => { - if let Some(username) = self.unwrap_id_store().get_account_name(uid).await? { + if let Some(username) = self.data_store.get_account_name(uid).await? { account_name = username; } else { return Ok(None); @@ -127,16 +126,16 @@ impl LdapDirectory { // Obtain account ID if not available if let Some(account_id) = account_id { principal.id = account_id; - } else if self.has_id_store() { + } else { principal.id = self - .unwrap_id_store() + .data_store .get_or_create_account_id(&account_name) .await?; } principal.name = account_name; // Obtain groups - if return_member_of && !principal.member_of.is_empty() && self.has_id_store() { + if return_member_of && !principal.member_of.is_empty() { for member_of in principal.member_of.iter_mut() { if member_of.contains('=') { let (rs, _res) = conn @@ -164,7 +163,7 @@ impl LdapDirectory { } // Map ids - self.unwrap_id_store() + self.data_store .map_group_names(principal, true) .await .map(Some) @@ -195,11 +194,7 @@ impl LdapDirectory { 'outer: for attr in &self.mappings.attr_name { if let Some(name) = entry.attrs.get(attr).and_then(|v| v.first()) { if !name.is_empty() { - ids.push( - self.unwrap_id_store() - .get_or_create_account_id(name) - .await?, - ); + ids.push(self.data_store.get_or_create_account_id(name).await?); break 'outer; } } @@ -326,14 +321,6 @@ impl LdapDirectory { }) .map_err(Into::into) } - - pub fn has_id_store(&self) -> bool { - self.id_store.is_some() - } - - pub fn unwrap_id_store(&self) -> &Store { - self.id_store.as_ref().unwrap() - } } impl LdapMappings { diff --git a/crates/directory/src/backend/ldap/mod.rs b/crates/directory/src/backend/ldap/mod.rs index b1a56721..5a2be0f3 100644 --- a/crates/directory/src/backend/ldap/mod.rs +++ b/crates/directory/src/backend/ldap/mod.rs @@ -33,7 +33,7 @@ pub struct LdapDirectory { pool: Pool, mappings: LdapMappings, auth_bind: Option, - id_store: Option, + pub(crate) data_store: Store, } #[derive(Debug, Default)] diff --git a/crates/directory/src/backend/memory/config.rs b/crates/directory/src/backend/memory/config.rs index 2e7acc51..c446ca8e 100644 --- a/crates/directory/src/backend/memory/config.rs +++ b/crates/directory/src/backend/memory/config.rs @@ -24,7 +24,7 @@ use store::Store; use utils::config::{utils::AsKey, Config}; -use crate::{Principal, Type}; +use crate::{backend::internal::manage::ManageDirectory, Principal, Type}; use super::{EmailType, MemoryDirectory}; @@ -32,15 +32,17 @@ impl MemoryDirectory { pub async fn from_config( config: &Config, prefix: impl AsKey, - id_store: Option, + data_store: Store, ) -> utils::config::Result { let prefix = prefix.as_key(); let mut directory = MemoryDirectory { - names_to_ids: id_store.into(), - ..Default::default() + data_store, + principals: Default::default(), + emails_to_ids: Default::default(), + domains: Default::default(), }; - for lookup_id in config.sub_keys((prefix.as_str(), "principals")) { + for lookup_id in config.sub_keys((prefix.as_str(), "principals"), ".name") { let name = config .value_require((prefix.as_str(), "principals", lookup_id, "name"))? .to_string(); @@ -53,8 +55,8 @@ impl MemoryDirectory { // Obtain id let id = directory - .names_to_ids - .get_or_insert(&name) + .data_store + .get_or_create_account_id(&name) .await .map_err(|err| { format!( @@ -67,14 +69,18 @@ impl MemoryDirectory { let mut member_of = Vec::new(); for (_, group) in config.values((prefix.as_str(), "principals", lookup_id, "member-of")) { - member_of.push(directory.names_to_ids.get_or_insert(group).await.map_err( - |err| { - format!( - "Failed to obtain id for principal {} ({}): {:?}", - name, lookup_id, err - ) - }, - )?); + member_of.push( + directory + .data_store + .get_or_create_account_id(group) + .await + .map_err(|err| { + format!( + "Failed to obtain id for principal {} ({}): {:?}", + name, lookup_id, err + ) + })?, + ); } // Parse email addresses diff --git a/crates/directory/src/backend/memory/mod.rs b/crates/directory/src/backend/memory/mod.rs index d1a0c79e..97e8ceff 100644 --- a/crates/directory/src/backend/memory/mod.rs +++ b/crates/directory/src/backend/memory/mod.rs @@ -26,60 +26,17 @@ use store::Store; use crate::Principal; -use super::internal::manage::ManageDirectory; - pub mod config; pub mod lookup; -#[derive(Default, Debug)] +#[derive(Debug)] pub struct MemoryDirectory { principals: Vec>, emails_to_ids: AHashMap>, - names_to_ids: NameToId, + pub(crate) data_store: Store, domains: AHashSet, } -pub enum NameToId { - Internal(AHashMap), - Store(Store), -} - -impl Default for NameToId { - fn default() -> Self { - Self::Internal(AHashMap::new()) - } -} - -impl std::fmt::Debug for NameToId { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Internal(arg0) => f.debug_tuple("Internal").field(arg0).finish(), - Self::Store(_) => f.debug_tuple("Store").finish(), - } - } -} - -impl From> for NameToId { - fn from(store: Option) -> Self { - match store { - Some(store) => Self::Store(store), - None => Self::Internal(AHashMap::new()), - } - } -} - -impl NameToId { - pub async fn get_or_insert(&mut self, name: &str) -> crate::Result { - match self { - Self::Internal(map) => { - let next_id = map.len() as u32; - Ok(*map.entry(name.to_string()).or_insert(next_id)) - } - Self::Store(store) => store.get_or_create_account_id(name).await, - } - } -} - #[derive(Debug)] enum EmailType { Primary(u32), diff --git a/crates/directory/src/backend/smtp/config.rs b/crates/directory/src/backend/smtp/config.rs index 418a5160..a417be08 100644 --- a/crates/directory/src/backend/smtp/config.rs +++ b/crates/directory/src/backend/smtp/config.rs @@ -22,6 +22,7 @@ */ use mail_send::{smtp::tls::build_tls_connector, SmtpClientBuilder}; +use store::Store; use utils::config::{utils::AsKey, Config}; use crate::core::config::build_pool; @@ -33,6 +34,7 @@ impl SmtpDirectory { config: &Config, prefix: impl AsKey, is_lmtp: bool, + data_store: Store, ) -> utils::config::Result { let prefix = prefix.as_key(); let address = config.value_require((&prefix, "address"))?; @@ -67,6 +69,7 @@ impl SmtpDirectory { .values((&prefix, "lookup.domains")) .map(|(_, v)| v.to_lowercase()) .collect(), + data_store, }) } } diff --git a/crates/directory/src/backend/smtp/mod.rs b/crates/directory/src/backend/smtp/mod.rs index 2e10833e..2a139d62 100644 --- a/crates/directory/src/backend/smtp/mod.rs +++ b/crates/directory/src/backend/smtp/mod.rs @@ -29,12 +29,14 @@ use ahash::AHashSet; use deadpool::managed::Pool; use mail_send::SmtpClientBuilder; use smtp_proto::EhloResponse; +use store::Store; use tokio::net::TcpStream; use tokio_rustls::client::TlsStream; pub struct SmtpDirectory { pool: Pool, domains: AHashSet, + pub(crate) data_store: Store, } pub struct SmtpConnectionManager { diff --git a/crates/directory/src/backend/sql/config.rs b/crates/directory/src/backend/sql/config.rs index 0c351a6b..1da5510b 100644 --- a/crates/directory/src/backend/sql/config.rs +++ b/crates/directory/src/backend/sql/config.rs @@ -31,7 +31,7 @@ impl SqlDirectory { config: &Config, prefix: impl AsKey, stores: &Stores, - id_store: Option, + data_store: Store, ) -> utils::config::Result { let prefix = prefix.as_key(); let store_id = config.value_require((&prefix, "store"))?; @@ -81,7 +81,7 @@ impl SqlDirectory { Ok(SqlDirectory { store, mappings, - id_store, + data_store, }) } } diff --git a/crates/directory/src/backend/sql/lookup.rs b/crates/directory/src/backend/sql/lookup.rs index 14855099..8118a026 100644 --- a/crates/directory/src/backend/sql/lookup.rs +++ b/crates/directory/src/backend/sql/lookup.rs @@ -22,7 +22,7 @@ */ use mail_send::Credentials; -use store::{NamedRows, Rows, Store, Value}; +use store::{NamedRows, Rows, Value}; use crate::{backend::internal::manage::ManageDirectory, Principal, QueryBy, Type}; @@ -47,7 +47,7 @@ impl SqlDirectory { .await? } QueryBy::Id(uid) => { - if let Some(username) = self.unwrap_id_store().get_account_name(uid).await? { + if let Some(username) = self.data_store.get_account_name(uid).await? { account_name = username; } else { return Ok(None); @@ -100,47 +100,43 @@ impl SqlDirectory { // Obtain account ID if not available if let Some(account_id) = account_id { principal.id = account_id; - } else if self.has_id_store() { + } else { principal.id = self - .unwrap_id_store() + .data_store .get_or_create_account_id(&account_name) .await?; } principal.name = account_name; - if self.has_id_store() { - // Obtain members - if return_member_of && !self.mappings.query_members.is_empty() { - for row in self - .store - .query::( - &self.mappings.query_members, - vec![principal.name.clone().into()], - ) - .await? - .rows - { - if let Some(Value::Text(account_id)) = row.values.first() { - principal.member_of.push( - self.unwrap_id_store() - .get_or_create_account_id(account_id) - .await?, - ); - } + // Obtain members + if return_member_of && !self.mappings.query_members.is_empty() { + for row in self + .store + .query::( + &self.mappings.query_members, + vec![principal.name.clone().into()], + ) + .await? + .rows + { + if let Some(Value::Text(account_id)) = row.values.first() { + principal + .member_of + .push(self.data_store.get_or_create_account_id(account_id).await?); } } + } - // Obtain emails - if !self.mappings.query_emails.is_empty() { - principal.emails = self - .store - .query::( - &self.mappings.query_emails, - vec![principal.name.clone().into()], - ) - .await? - .into(); - } + // Obtain emails + if !self.mappings.query_emails.is_empty() { + principal.emails = self + .store + .query::( + &self.mappings.query_emails, + vec![principal.name.clone().into()], + ) + .await? + .into(); } Ok(Some(principal)) @@ -156,11 +152,7 @@ impl SqlDirectory { for row in names.rows { if let Some(Value::Text(name)) = row.values.first() { - ids.push( - self.unwrap_id_store() - .get_or_create_account_id(name) - .await?, - ); + ids.push(self.data_store.get_or_create_account_id(name).await?); } } @@ -207,16 +199,6 @@ impl SqlDirectory { } } -impl SqlDirectory { - pub fn has_id_store(&self) -> bool { - self.id_store.is_some() - } - - pub fn unwrap_id_store(&self) -> &Store { - self.id_store.as_ref().unwrap() - } -} - impl SqlMappings { pub fn row_to_principal(&self, rows: NamedRows) -> crate::Result> { let mut principal = Principal::default(); diff --git a/crates/directory/src/backend/sql/mod.rs b/crates/directory/src/backend/sql/mod.rs index 46b6a1fb..d3f169f5 100644 --- a/crates/directory/src/backend/sql/mod.rs +++ b/crates/directory/src/backend/sql/mod.rs @@ -29,7 +29,7 @@ pub mod lookup; pub struct SqlDirectory { store: LookupStore, mappings: SqlMappings, - id_store: Option, + pub(crate) data_store: Store, } #[derive(Debug, Default)] diff --git a/crates/directory/src/core/config.rs b/crates/directory/src/core/config.rs index fd00da1e..94007a6b 100644 --- a/crates/directory/src/core/config.rs +++ b/crates/directory/src/core/config.rs @@ -27,10 +27,10 @@ use deadpool::{ }; use regex::Regex; use std::{sync::Arc, time::Duration}; -use store::Stores; +use store::{Store, Stores}; use utils::config::{ utils::{AsKey, ParseValue}, - Config, + Config, Servers, }; use ahash::AHashMap; @@ -50,7 +50,8 @@ pub trait ConfigDirectory { async fn parse_directory( &self, stores: &Stores, - id_store: Option<&str>, + servers: &Servers, + data_store: Store, ) -> utils::config::Result; } @@ -58,15 +59,19 @@ impl ConfigDirectory for Config { async fn parse_directory( &self, stores: &Stores, - id_store: Option<&str>, + servers: &Servers, + data_store: Store, ) -> utils::config::Result { let mut config = Directories { directories: AHashMap::new(), lookups: AHashMap::new(), }; - let id_store = id_store.and_then(|id| stores.stores.get(id).cloned()); - for id in self.sub_keys("directory") { + for id in self.sub_keys("directory", ".type") { + if id.ends_with(".columns") || id.ends_with(".attributes") || id.contains(".principals") + { + continue; + } // Parse directory if self.property_or_static::(("directory", id, "disable"), "false")? { tracing::debug!("Skipping disabled directory {id:?}."); @@ -101,19 +106,33 @@ impl ConfigDirectory for Config { "ldap" => DirectoryInner::Ldap(LdapDirectory::from_config( self, prefix, - id_store.clone(), + data_store.clone(), )?), "sql" => DirectoryInner::Sql(SqlDirectory::from_config( self, prefix, stores, - id_store.clone(), + data_store.clone(), + )?), + "imap" => DirectoryInner::Imap(ImapDirectory::from_config( + self, + prefix, + data_store.clone(), + )?), + "smtp" => DirectoryInner::Smtp(SmtpDirectory::from_config( + self, + prefix, + false, + data_store.clone(), + )?), + "lmtp" => DirectoryInner::Smtp(SmtpDirectory::from_config( + self, + prefix, + true, + data_store.clone(), )?), - "imap" => DirectoryInner::Imap(ImapDirectory::from_config(self, prefix)?), - "smtp" => DirectoryInner::Smtp(SmtpDirectory::from_config(self, prefix, false)?), - "lmtp" => DirectoryInner::Smtp(SmtpDirectory::from_config(self, prefix, true)?), "memory" => DirectoryInner::Memory( - MemoryDirectory::from_config(self, prefix, id_store.clone()).await?, + MemoryDirectory::from_config(self, prefix, data_store.clone()).await?, ), unknown => { return Err(format!("Unknown directory type: {unknown:?}")); @@ -132,6 +151,7 @@ impl ConfigDirectory for Config { ("directory", id, "options.subaddressing"), )?, cache: CachedDirectory::try_from_config(self, ("directory", id))?, + blocked_ips: servers.blocked_ips.clone(), }); // Add lookups diff --git a/crates/directory/src/core/dispatch.rs b/crates/directory/src/core/dispatch.rs index 6771f74f..73a11614 100644 --- a/crates/directory/src/core/dispatch.rs +++ b/crates/directory/src/core/dispatch.rs @@ -21,11 +21,58 @@ * for more details. */ +use std::net::IpAddr; + +use mail_send::Credentials; +use store::Store; + use crate::{ - backend::internal::lookup::DirectoryStore, Directory, DirectoryInner, Principal, QueryBy, + backend::internal::lookup::DirectoryStore, AuthResult, Directory, DirectoryInner, Principal, + QueryBy, }; impl Directory { + pub async fn authenticate( + &self, + credentials: &Credentials, + remote_ip: IpAddr, + return_member_of: bool, + ) -> crate::Result>> { + if let Some(principal) = self + .query(QueryBy::Credentials(credentials), return_member_of) + .await? + { + Ok(AuthResult::Success(principal)) + } else if self.blocked_ips.has_fail2ban() { + let login = match credentials { + Credentials::Plain { username, .. } + | Credentials::XOauth2 { username, .. } + | Credentials::OAuthBearer { token: username } => username, + }; + if let Some(banned) = self + .blocked_ips + .is_fail2banned(remote_ip, login.to_string()) + { + tracing::info!( + context = "directory", + event = "fail2ban", + remote_ip = ?remote_ip, + login = ?login, + "IP address blocked after too many failed login attempts", + ); + + // Write blocked address to config + self.store().config_set(vec![banned].into_iter()).await?; + + Ok(AuthResult::Banned) + } else { + Ok(AuthResult::Failure) + } + } else { + Ok(AuthResult::Failure) + } + } + pub async fn query( &self, by: QueryBy<'_>, @@ -161,4 +208,15 @@ impl Directory { DirectoryInner::Memory(store) => store.expn(address.as_ref()).await, } } + + fn store(&self) -> &Store { + match &self.store { + DirectoryInner::Internal(store) => store, + DirectoryInner::Ldap(store) => &store.data_store, + DirectoryInner::Sql(store) => &store.data_store, + DirectoryInner::Imap(store) => &store.data_store, + DirectoryInner::Smtp(store) => &store.data_store, + DirectoryInner::Memory(store) => &store.data_store, + } + } } diff --git a/crates/directory/src/core/mod.rs b/crates/directory/src/core/mod.rs index 8742eb5e..683aa978 100644 --- a/crates/directory/src/core/mod.rs +++ b/crates/directory/src/core/mod.rs @@ -21,20 +21,7 @@ * for more details. */ -use crate::{backend::memory::MemoryDirectory, AddressMapping, Directory, DirectoryInner}; - pub mod cache; pub mod config; pub mod dispatch; pub mod secret; - -impl Default for Directory { - fn default() -> Self { - Directory { - store: DirectoryInner::Memory(MemoryDirectory::default()), - catch_all: AddressMapping::Disable, - subaddressing: AddressMapping::Disable, - cache: None, - } - } -} diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 87f54077..bb5cbe6c 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -37,16 +37,17 @@ use deadpool::managed::PoolError; use ldap3::LdapError; use mail_send::Credentials; use store::Store; -use utils::config::DynValue; +use utils::{config::DynValue, listener::blocked::BlockedIps}; pub mod backend; pub mod core; pub struct Directory { - store: DirectoryInner, - catch_all: AddressMapping, - subaddressing: AddressMapping, - cache: Option, + pub store: DirectoryInner, + pub catch_all: AddressMapping, + pub subaddressing: AddressMapping, + pub cache: Option, + pub blocked_ips: Arc, } #[derive(Debug, Default, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -125,6 +126,12 @@ pub enum QueryBy<'x> { Credentials(&'x Credentials), } +pub enum AuthResult { + Success(T), + Failure, + Banned, +} + impl Principal { pub fn name(&self) -> &str { &self.name diff --git a/crates/imap/src/core/client.rs b/crates/imap/src/core/client.rs index b5ff97e0..bb371dea 100644 --- a/crates/imap/src/core/client.rs +++ b/crates/imap/src/core/client.rs @@ -28,7 +28,6 @@ use imap_proto::{ Command, ResponseCode, StatusResponse, }; use jmap::auth::rate_limit::AuthenticatedLimiter; -use parking_lot::Mutex; use utils::listener::{ limiter::{ConcurrencyLimiter, RateLimiter}, SessionStream, @@ -231,9 +230,8 @@ impl Session { if !data .imap .get_authenticated_limiter(data.account_id) - .lock() .request_limiter - .is_allowed() + .is_allowed(&self.imap.rate_requests) { return Err(StatusResponse::no("Too many requests") .with_tag(request.tag) @@ -392,19 +390,16 @@ impl State { } impl IMAP { - pub fn get_authenticated_limiter(&self, account_id: u32) -> Arc> { + pub fn get_authenticated_limiter(&self, account_id: u32) -> Arc { self.rate_limiter .get(&account_id) .map(|limiter| limiter.clone()) .unwrap_or_else(|| { - let limiter = Arc::new(Mutex::new(AuthenticatedLimiter { - request_limiter: RateLimiter::new( - self.rate_requests.requests, - self.rate_requests.period, - ), + let limiter = Arc::new(AuthenticatedLimiter { + request_limiter: RateLimiter::new(&self.rate_requests), concurrent_requests: ConcurrencyLimiter::new(self.rate_concurrent), concurrent_uploads: ConcurrencyLimiter::new(self.rate_concurrent), - })); + }); self.rate_limiter.insert(account_id, limiter.clone()); limiter }) diff --git a/crates/imap/src/core/mod.rs b/crates/imap/src/core/mod.rs index dea1f3d5..b3dc64c3 100644 --- a/crates/imap/src/core/mod.rs +++ b/crates/imap/src/core/mod.rs @@ -23,6 +23,7 @@ use std::{ collections::BTreeMap, + net::IpAddr, sync::{atomic::AtomicU32, Arc}, time::Duration, }; @@ -35,10 +36,7 @@ use imap_proto::{ Command, ResponseCode, StatusResponse, }; use jmap::{ - auth::{ - rate_limit::{AuthenticatedLimiter, RemoteAddress}, - AccessToken, - }, + auth::{rate_limit::AuthenticatedLimiter, AccessToken}, JMAP, }; use store::roaring::RoaringBitmap; @@ -82,7 +80,7 @@ pub struct IMAP { pub greeting_plain: Vec, pub greeting_tls: Vec, - pub rate_limiter: DashMap>>, + pub rate_limiter: DashMap>, pub rate_requests: Rate, pub rate_concurrent: u64, } @@ -100,7 +98,7 @@ pub struct Session { pub stream_rx: ReadHalf, pub stream_tx: Arc>>, pub in_flight: InFlight, - pub remote_addr: RemoteAddress, + pub remote_addr: IpAddr, pub span: tracing::Span, } diff --git a/crates/imap/src/core/session.rs b/crates/imap/src/core/session.rs index 765a9591..447d5e3b 100644 --- a/crates/imap/src/core/session.rs +++ b/crates/imap/src/core/session.rs @@ -24,7 +24,6 @@ use std::{borrow::Cow, sync::Arc}; use imap_proto::{protocol::ProtocolVersion, receiver::Receiver}; -use jmap::auth::rate_limit::RemoteAddress; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_rustls::server::TlsStream; use utils::listener::{stream::NullIo, SessionManager, SessionStream}; @@ -139,7 +138,7 @@ impl Session { instance: session.instance, span: session.span, in_flight: session.in_flight, - remote_addr: RemoteAddress::IpAddress(session.remote_ip), + remote_addr: session.remote_ip, stream_rx, stream_tx: Arc::new(tokio::sync::Mutex::new(stream_tx)), }) diff --git a/crates/imap/src/op/authenticate.rs b/crates/imap/src/op/authenticate.rs index fef64abf..875bf5c2 100644 --- a/crates/imap/src/op/authenticate.rs +++ b/crates/imap/src/op/authenticate.rs @@ -23,6 +23,7 @@ use std::sync::Arc; +use directory::AuthResult; use imap_proto::{ protocol::{authenticate::Mechanism, capability::Capability}, receiver::{self, Request}, @@ -116,9 +117,15 @@ impl Session { // Authenticate let access_token = match credentials { Credentials::Plain { username, secret } | Credentials::XOauth2 { username, secret } => { - self.jmap - .authenticate_plain(&username, &secret, &self.remote_addr) + match self + .jmap + .authenticate_plain(&username, &secret, self.remote_addr) .await + { + AuthResult::Success(token) => Some(token), + AuthResult::Failure => None, + AuthResult::Banned => return Err(()), + } } Credentials::OAuthBearer { token } => { match self @@ -145,7 +152,6 @@ impl Session { let in_flight = self .imap .get_authenticated_limiter(access_token.primary_id()) - .lock() .concurrent_requests .is_allowed(); if let Some(in_flight) = in_flight { diff --git a/crates/install/src/main.rs b/crates/install/src/main.rs index 07d4f591..9f6448dd 100644 --- a/crates/install/src/main.rs +++ b/crates/install/src/main.rs @@ -277,13 +277,18 @@ fn main() -> std::io::Result<()> { ("__DIRECTORY__", directory.id()), ], ); - sed( - cfg_path.join("jmap").join("store.toml"), - &[ - ("__BLOB_STORE__", blob.id().unwrap_or("%{DEFAULT_STORE}%")), - ("__FTS_STORE__", fts.id().unwrap_or("%{DEFAULT_STORE}%")), - ], - ); + if let Some(blob) = blob.id() { + sed( + cfg_path.join("common").join("store.toml"), + &[("blob = \"%{DEFAULT_STORE}%", &format!("blob = \"{blob}"))], + ); + } + if let Some(fts) = fts.id() { + sed( + cfg_path.join("common").join("store.toml"), + &[("fts = \"%{DEFAULT_STORE}%", &format!("fts = \"{fts}"))], + ); + } if let Some(id) = spamdb.id() { sed( cfg_path.join("common").join("sieve.toml"), @@ -364,13 +369,6 @@ fn main() -> std::io::Result<()> { ("__DIRECTORY__", smtp_directory.id()), ], ); - sed( - cfg_path.join("jmap").join("store.toml"), - &[ - ("__BLOB_STORE__", "%{DEFAULT_STORE}%"), - ("__FTS_STORE__", "%{DEFAULT_STORE}%"), - ], - ); if let Some(id) = spamdb.id() { sed( cfg_path.join("common").join("sieve.toml"), diff --git a/crates/jmap/src/api/admin.rs b/crates/jmap/src/api/admin.rs index a5cc8823..06687a70 100644 --- a/crates/jmap/src/api/admin.rs +++ b/crates/jmap/src/api/admin.rs @@ -29,6 +29,7 @@ use http_body_util::combinators::BoxBody; use hyper::{body::Bytes, Method, StatusCode}; use jmap_proto::error::request::RequestError; use serde_json::json; +use utils::config::ConfigKey; use crate::{services::housekeeper, JMAP}; @@ -299,7 +300,18 @@ impl JMAP { .into_http_response(), } } - ("certificates", Some("reload"), &Method::GET) => { + ("reload", Some("config"), &Method::GET) => { + let _ = self + .housekeeper_tx + .send(housekeeper::Event::ReloadConfig) + .await; + + JsonResponse::new(json!({ + "data": [], + })) + .into_http_response() + } + ("reload", Some("certificates"), &Method::GET) => { let _ = self .housekeeper_tx .send(housekeeper::Event::ReloadCertificates) @@ -310,6 +322,73 @@ impl JMAP { })) .into_http_response() } + ("config", key, &Method::GET) => { + match self.store.config_list(key.unwrap_or_default()).await { + Ok(config) => JsonResponse::new(json!({ + "data": config.keys.into_iter().collect::>(), + })) + .into_http_response(), + Err(err) => RequestError::blank( + StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + "Config fetch failed", + err.to_string(), + ) + .into_http_response(), + } + } + ("config", Some(prefix), &Method::DELETE) if !prefix.is_empty() => { + let result = match prefix.strip_suffix('.') { + Some(prefix) if !prefix.is_empty() => { + self.store.config_clear_prefix(prefix).await + } + _ => self.store.config_clear(prefix).await, + }; + match result { + Ok(_) => JsonResponse::new(json!({ + "data": [], + })) + .into_http_response(), + Err(err) => RequestError::blank( + StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + "Config fetch failed", + err.to_string(), + ) + .into_http_response(), + } + } + ("config", None, &Method::POST) => { + if let Some(changes) = body + .and_then(|body| serde_json::from_slice::>(&body).ok()) + { + match self + .store + .config_set( + changes + .into_iter() + .map(|(key, value)| ConfigKey { key, value }), + ) + .await + { + Ok(_) => JsonResponse::new(json!({ + "data": [], + })) + .into_http_response(), + Err(err) => RequestError::blank( + StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + "Config update failed", + err.to_string(), + ) + .into_http_response(), + } + } else { + RequestError::blank( + StatusCode::BAD_REQUEST.as_u16(), + "Invalid parameters", + "Failed to deserialize config update request", + ) + .into_http_response() + } + } (path_1 @ ("queue" | "report"), Some(path_2), &Method::GET) => { self.smtp .handle_manage_request(req.uri(), req.method(), path_1, path_2) diff --git a/crates/jmap/src/api/config.rs b/crates/jmap/src/api/config.rs index d1422709..59c79e68 100644 --- a/crates/jmap/src/api/config.rs +++ b/crates/jmap/src/api/config.rs @@ -32,7 +32,9 @@ impl crate::Config { pub fn new(settings: &utils::config::Config) -> Result { let mut config = Self { default_language: Language::from_iso_639( - settings.value("jmap.fts.default-language").unwrap_or("en"), + settings + .value("storage.fts.default-language") + .unwrap_or("en"), ) .unwrap_or(Language::English), query_max_results: settings @@ -139,9 +141,9 @@ impl crate::Config { principal_allow_lookups: settings .property("jmap.principal.allow-lookups")? .unwrap_or(true), - encrypt: settings.property_or_static("jmap.encryption.enable", "true")?, - encrypt_append: settings.property_or_static("jmap.encryption.append", "false")?, - spam_header: settings.value("jmap.spam.header").and_then(|v| { + encrypt: settings.property_or_static("storage.encryption.enable", "true")?, + encrypt_append: settings.property_or_static("storage.encryption.append", "false")?, + spam_header: settings.value("storage.spam.header").and_then(|v| { v.split_once(':').map(|(k, v)| { ( mail_parser::HeaderName::parse(k.trim().to_string()).unwrap(), diff --git a/crates/jmap/src/api/http.rs b/crates/jmap/src/api/http.rs index a4789290..6cde1ce3 100644 --- a/crates/jmap/src/api/http.rs +++ b/crates/jmap/src/api/http.rs @@ -215,7 +215,7 @@ pub async fn parse_jmap_request( ("", &Method::POST) => { return match jmap.is_auth_allowed_soft(&remote_addr) { Ok(_) => { - jmap.handle_user_device_auth_post(&mut req, &remote_addr) + jmap.handle_user_device_auth_post(&mut req, remote_addr) .await } Err(err) => err.into_http_response(), @@ -229,10 +229,7 @@ pub async fn parse_jmap_request( } ("code", &Method::POST) => { return match jmap.is_auth_allowed_soft(&remote_addr) { - Ok(_) => { - jmap.handle_user_code_auth_post(&mut req, &remote_addr) - .await - } + Ok(_) => jmap.handle_user_code_auth_post(&mut req, remote_addr).await, Err(err) => err.into_http_response(), } } @@ -259,11 +256,11 @@ pub async fn parse_jmap_request( match *req.method() { Method::GET => { - return jmap.handle_crypto_update(&mut req, &remote_addr).await; + return jmap.handle_crypto_update(&mut req, remote_addr).await; } Method::POST => { return match jmap.is_auth_allowed_soft(&remote_addr) { - Ok(_) => jmap.handle_crypto_update(&mut req, &remote_addr).await, + Ok(_) => jmap.handle_crypto_update(&mut req, remote_addr).await, Err(err) => err.into_http_response(), } } diff --git a/crates/jmap/src/api/session.rs b/crates/jmap/src/api/session.rs index 001f414e..de72df7a 100644 --- a/crates/jmap/src/api/session.rs +++ b/crates/jmap/src/api/session.rs @@ -425,7 +425,7 @@ impl CoreCapabilities { pub fn new(config: &crate::Config) -> Self { CoreCapabilities { max_size_upload: config.upload_max_size, - max_concurrent_upload: config.upload_max_concurrent, + max_concurrent_upload: config.upload_max_concurrent as usize, max_size_request: config.request_max_size, max_concurrent_requests: config.request_max_concurrent as usize, max_calls_in_request: config.request_max_calls, diff --git a/crates/jmap/src/auth/authenticate.rs b/crates/jmap/src/auth/authenticate.rs index 311bba69..5e58b842 100644 --- a/crates/jmap/src/auth/authenticate.rs +++ b/crates/jmap/src/auth/authenticate.rs @@ -21,13 +21,9 @@ * for more details. */ -use std::{ - net::{IpAddr, Ipv4Addr}, - sync::Arc, - time::Instant, -}; +use std::{net::IpAddr, sync::Arc, time::Instant}; -use directory::QueryBy; +use directory::{AuthResult, QueryBy}; use hyper::header; use jmap_proto::error::request::RequestError; use mail_parser::decoders::base64::base64_decode; @@ -36,7 +32,7 @@ use utils::{listener::limiter::InFlight, map::ttl_dashmap::TtlMap}; use crate::JMAP; -use super::{rate_limit::RemoteAddress, AccessToken}; +use super::AccessToken; impl JMAP { pub async fn authenticate_headers( @@ -67,7 +63,13 @@ impl JMAP { }) }) { - self.authenticate_plain(&account, &secret, &addr).await + if let AuthResult::Success(access_token) = + self.authenticate_plain(&account, &secret, addr).await + { + Some(access_token) + } else { + None + } } else { tracing::debug!( context = "authenticate_headers", @@ -151,18 +153,19 @@ impl JMAP { &self, req: &hyper::Request, remote_ip: IpAddr, - ) -> RemoteAddress { + ) -> IpAddr { if !self.config.rate_use_forwarded { - RemoteAddress::IpAddress(remote_ip) + remote_ip } else if let Some(forwarded_for) = req .headers() .get(header::FORWARDED) .and_then(|h| h.to_str().ok()) + .and_then(|h| h.parse::().ok()) { - RemoteAddress::IpAddressFwd(forwarded_for.trim().to_string()) + forwarded_for } else { - tracing::debug!("Warning: No remote address found in request, using loopback."); - RemoteAddress::IpAddress(Ipv4Addr::new(127, 0, 0, 1).into()) + tracing::warn!("Warning: No remote address found in request, using remote ip."); + remote_ip } } @@ -170,25 +173,27 @@ impl JMAP { &self, username: &str, secret: &str, - remote_addr: &RemoteAddress, - ) -> Option { + remote_ip: IpAddr, + ) -> AuthResult { match self .directory - .query( - QueryBy::Credentials(&Credentials::Plain { + .authenticate( + &Credentials::Plain { username: username.to_string(), secret: secret.to_string(), - }), + }, + remote_ip, true, ) .await { - Ok(Some(principal)) => AccessToken::new(principal).into(), - Ok(None) => { - let _ = self.is_auth_allowed_hard(remote_addr); - None + Ok(AuthResult::Success(principal)) => AuthResult::Success(AccessToken::new(principal)), + Ok(AuthResult::Failure) => { + let _ = self.is_auth_allowed_hard(&remote_ip); + AuthResult::Failure } - Err(_) => None, + Ok(AuthResult::Banned) => AuthResult::Banned, + Err(_) => AuthResult::Failure, } } diff --git a/crates/jmap/src/auth/oauth/device_auth.rs b/crates/jmap/src/auth/oauth/device_auth.rs index 658e761a..c77d5b59 100644 --- a/crates/jmap/src/auth/oauth/device_auth.rs +++ b/crates/jmap/src/auth/oauth/device_auth.rs @@ -22,10 +22,12 @@ */ use std::{ + net::IpAddr, sync::{atomic, Arc}, time::{Duration, Instant}, }; +use directory::AuthResult; use hyper::StatusCode; use store::rand::{ distributions::{Alphanumeric, Standard}, @@ -35,12 +37,9 @@ use utils::{listener::ServerInstance, map::ttl_dashmap::TtlMap}; use crate::{ api::{http::ToHttpResponse, HtmlResponse, HttpRequest, HttpResponse, JsonResponse}, - auth::{ - oauth::{ - MAX_POST_LEN, OAUTH_HTML_ERROR, OAUTH_HTML_LOGIN_HEADER_FAILED, - OAUTH_HTML_LOGIN_SUCCESS, STATUS_AUTHORIZED, - }, - rate_limit::RemoteAddress, + auth::oauth::{ + MAX_POST_LEN, OAUTH_HTML_ERROR, OAUTH_HTML_LOGIN_HEADER_FAILED, OAUTH_HTML_LOGIN_SUCCESS, + STATUS_AUTHORIZED, }, JMAP, }; @@ -154,7 +153,7 @@ impl JMAP { pub async fn handle_user_device_auth_post( &self, req: &mut HttpRequest, - remote_addr: &RemoteAddress, + remote_addr: IpAddr, ) -> HttpResponse { // Parse form let fields = match FormData::from_request(req, MAX_POST_LEN).await { @@ -177,7 +176,9 @@ impl JMAP { { if let (Some(email), Some(password)) = (fields.get("email"), fields.get("password")) { - if let Some(id) = self.authenticate_plain(email, password, remote_addr).await { + if let AuthResult::Success(id) = + self.authenticate_plain(email, password, remote_addr).await + { oauth .account_id .store(id.primary_id(), atomic::Ordering::Relaxed); diff --git a/crates/jmap/src/auth/oauth/user_code.rs b/crates/jmap/src/auth/oauth/user_code.rs index e62792d8..9403ecc6 100644 --- a/crates/jmap/src/auth/oauth/user_code.rs +++ b/crates/jmap/src/auth/oauth/user_code.rs @@ -23,10 +23,12 @@ use std::{ collections::HashMap, + net::IpAddr, sync::Arc, time::{Duration, Instant}, }; +use directory::AuthResult; use http_body_util::{BodyExt, Full}; use hyper::{body::Bytes, header, StatusCode}; use mail_builder::encoders::base64::base64_encode; @@ -37,7 +39,6 @@ use utils::map::ttl_dashmap::TtlMap; use crate::{ api::{http::ToHttpResponse, HtmlResponse, HttpRequest, HttpResponse}, - auth::rate_limit::RemoteAddress, JMAP, }; @@ -111,7 +112,7 @@ impl JMAP { pub async fn handle_user_code_auth_post( &self, req: &mut HttpRequest, - remote_addr: &RemoteAddress, + remote_addr: IpAddr, ) -> HttpResponse { // Parse form let params = match FormData::from_request(req, MAX_POST_LEN).await { @@ -137,7 +138,8 @@ impl JMAP { // Authenticate user if let (Some(email), Some(password)) = (params.get("email"), params.get("password")) { - if let Some(access_token) = self.authenticate_plain(email, password, remote_addr).await + if let AuthResult::Success(access_token) = + self.authenticate_plain(email, password, remote_addr).await { // Generate client code let client_code = thread_rng() diff --git a/crates/jmap/src/auth/rate_limit.rs b/crates/jmap/src/auth/rate_limit.rs index 88a3e699..3d8b6905 100644 --- a/crates/jmap/src/auth/rate_limit.rs +++ b/crates/jmap/src/auth/rate_limit.rs @@ -24,19 +24,12 @@ use std::{net::IpAddr, sync::Arc}; use jmap_proto::error::request::{RequestError, RequestLimitError}; -use store::parking_lot::Mutex; use utils::listener::limiter::{ConcurrencyLimiter, InFlight, RateLimiter}; use crate::JMAP; use super::AccessToken; -#[derive(Debug, Clone, Eq, PartialEq, Hash)] -pub enum RemoteAddress { - IpAddress(IpAddr), - IpAddressFwd(String), -} - pub struct AuthenticatedLimiter { pub request_limiter: RateLimiter, pub concurrent_requests: ConcurrencyLimiter, @@ -50,53 +43,44 @@ pub struct AnonymousLimiter { } impl JMAP { - pub fn get_authenticated_limiter(&self, account_id: u32) -> Arc> { + pub fn get_authenticated_limiter(&self, account_id: u32) -> Arc { self.rate_limit_auth .get(&account_id) .map(|limiter| limiter.clone()) .unwrap_or_else(|| { - let limiter = Arc::new(Mutex::new(AuthenticatedLimiter { - request_limiter: RateLimiter::new( - self.config.rate_authenticated.requests, - self.config.rate_authenticated.period, - ), + let limiter = Arc::new(AuthenticatedLimiter { + request_limiter: RateLimiter::new(&self.config.rate_authenticated), concurrent_requests: ConcurrencyLimiter::new( self.config.request_max_concurrent, ), - concurrent_uploads: ConcurrencyLimiter::new( - self.config.upload_max_concurrent as u64, - ), - })); + concurrent_uploads: ConcurrencyLimiter::new(self.config.upload_max_concurrent), + }); self.rate_limit_auth.insert(account_id, limiter.clone()); limiter }) } - pub fn get_anonymous_limiter(&self, addr: &RemoteAddress) -> Arc> { + pub fn get_anonymous_limiter(&self, addr: &IpAddr) -> Arc { self.rate_limit_unauth .get(addr) .map(|limiter| limiter.clone()) .unwrap_or_else(|| { - let limiter = Arc::new(Mutex::new(AnonymousLimiter { - request_limiter: RateLimiter::new( - self.config.rate_anonymous.requests, - self.config.rate_anonymous.period, - ), - auth_limiter: RateLimiter::new( - self.config.rate_authenticate_req.requests, - self.config.rate_authenticate_req.period, - ), - })); - self.rate_limit_unauth.insert(addr.clone(), limiter.clone()); + let limiter = Arc::new(AnonymousLimiter { + request_limiter: RateLimiter::new(&self.config.rate_anonymous), + auth_limiter: RateLimiter::new(&self.config.rate_authenticate_req), + }); + self.rate_limit_unauth.insert(*addr, limiter.clone()); limiter }) } pub fn is_account_allowed(&self, access_token: &AccessToken) -> Result { - let limiter_ = self.get_authenticated_limiter(access_token.primary_id()); - let mut limiter = limiter_.lock(); + let limiter = self.get_authenticated_limiter(access_token.primary_id()); - if limiter.request_limiter.is_allowed() { + if limiter + .request_limiter + .is_allowed(&self.config.rate_authenticated) + { if let Some(in_flight_request) = limiter.concurrent_requests.is_allowed() { Ok(in_flight_request) } else if access_token.is_super_user() { @@ -111,12 +95,11 @@ impl JMAP { } } - pub fn is_anonymous_allowed(&self, addr: &RemoteAddress) -> Result<(), RequestError> { + pub fn is_anonymous_allowed(&self, addr: &IpAddr) -> Result<(), RequestError> { if self .get_anonymous_limiter(addr) - .lock() .request_limiter - .is_allowed() + .is_allowed(&self.config.rate_anonymous) { Ok(()) } else { @@ -127,7 +110,6 @@ impl JMAP { pub fn is_upload_allowed(&self, access_token: &AccessToken) -> Result { if let Some(in_flight_request) = self .get_authenticated_limiter(access_token.primary_id()) - .lock() .concurrent_uploads .is_allowed() { @@ -139,21 +121,24 @@ impl JMAP { } } - pub fn is_auth_allowed_soft(&self, addr: &RemoteAddress) -> Result<(), RequestError> { + pub fn is_auth_allowed_soft(&self, addr: &IpAddr) -> Result<(), RequestError> { match self.rate_limit_unauth.get(addr) { - Some(limiter) if !limiter.lock().auth_limiter.is_allowed_soft() => { + Some(limiter) + if !limiter + .auth_limiter + .is_allowed_soft(&self.config.rate_authenticate_req) => + { Err(RequestError::too_many_auth_attempts()) } _ => Ok(()), } } - pub fn is_auth_allowed_hard(&self, addr: &RemoteAddress) -> Result<(), RequestError> { + pub fn is_auth_allowed_hard(&self, addr: &IpAddr) -> Result<(), RequestError> { if self .get_anonymous_limiter(addr) - .lock() .auth_limiter - .is_allowed() + .is_allowed(&self.config.rate_authenticate_req) { Ok(()) } else { diff --git a/crates/jmap/src/email/crypto.rs b/crates/jmap/src/email/crypto.rs index 3020721e..a2a34d8e 100644 --- a/crates/jmap/src/email/crypto.rs +++ b/crates/jmap/src/email/crypto.rs @@ -21,14 +21,15 @@ * for more details. */ -use std::{borrow::Cow, collections::BTreeSet, fmt::Display, io::Cursor}; +use std::{borrow::Cow, collections::BTreeSet, fmt::Display, io::Cursor, net::IpAddr}; use crate::{ api::{http::ToHttpResponse, HtmlResponse, HttpRequest, HttpResponse}, - auth::{oauth::FormData, rate_limit::RemoteAddress}, + auth::oauth::FormData, JMAP, }; use aes::cipher::{block_padding::Pkcs7, BlockEncryptMut, KeyIvInit}; +use directory::AuthResult; use jmap_proto::types::{collection::Collection, property::Property}; use mail_builder::{encoders::base64::base64_encode_mime, mime::make_boundary}; use mail_parser::{decoders::base64::base64_decode, Message, MessageParser, MimeHeaders}; @@ -619,7 +620,7 @@ impl JMAP { pub async fn handle_crypto_update( &self, req: &mut HttpRequest, - remote_addr: &RemoteAddress, + remote_addr: IpAddr, ) -> HttpResponse { let mut response = String::with_capacity( CRYPT_HTML_HEADER.len() + CRYPT_HTML_FOOTER.len() + CRYPT_HTML_FORM.len(), @@ -668,7 +669,7 @@ impl JMAP { async fn validate_form( &self, mut form: FormData, - remote_addr: &RemoteAddress, + remote_addr: IpAddr, ) -> Result, Cow> { let certificate = form.remove_bytes("certificate"); if let (Some(email), Some(password), Some(encryption)) = ( @@ -685,10 +686,14 @@ impl JMAP { } // Authenticate - let token = self - .authenticate_plain(email, password, remote_addr) - .await - .ok_or_else(|| Cow::from("Invalid login or password"))?; + let token = if let AuthResult::Success(token) = + self.authenticate_plain(email, password, remote_addr).await + { + token + } else { + return Err(Cow::from("Invalid login or password")); + }; + if encryption != "disable" { let (method, certs) = try_parse_certs(certificate.unwrap_or_default()).map_err(Cow::from)?; diff --git a/crates/jmap/src/lib.rs b/crates/jmap/src/lib.rs index db79ec32..79ce7b89 100644 --- a/crates/jmap/src/lib.rs +++ b/crates/jmap/src/lib.rs @@ -21,13 +21,15 @@ * for more details. */ -use std::{collections::hash_map::RandomState, fmt::Display, sync::Arc, time::Duration}; +use std::{ + collections::hash_map::RandomState, fmt::Display, net::IpAddr, sync::Arc, time::Duration, +}; use ::sieve::{Compiler, Runtime}; use api::session::BaseCapabilities; use auth::{ oauth::OAuthCode, - rate_limit::{AnonymousLimiter, AuthenticatedLimiter, RemoteAddress}, + rate_limit::{AnonymousLimiter, AuthenticatedLimiter}, AccessToken, }; use dashmap::DashMap; @@ -50,7 +52,6 @@ use services::{ use smtp::core::SMTP; use store::{ fts::FtsFilter, - parking_lot::Mutex, query::{sort::Pagination, Comparator, Filter, ResultSet, SortedResultSet}, roaring::RoaringBitmap, write::{BatchBuilder, BitmapClass, DirectoryClass, TagValue, ToBitmaps, ValueClass}, @@ -58,9 +59,8 @@ use store::{ }; use tokio::sync::mpsc; use utils::{ - config::Rate, + config::{Rate, Servers}, ipc::DeliveryEvent, - listener::tls::Certificate, map::ttl_dashmap::{TtlDashMap, TtlMap}, snowflake::SnowflakeIdGenerator, UnwrapFailure, @@ -96,8 +96,8 @@ pub struct JMAP { pub access_tokens: TtlDashMap>, pub snowflake_id: SnowflakeIdGenerator, - pub rate_limit_auth: DashMap>>, - pub rate_limit_unauth: DashMap>>, + pub rate_limit_auth: DashMap>, + pub rate_limit_unauth: DashMap>, pub oauth_codes: TtlDashMap>, @@ -123,7 +123,7 @@ pub struct Config { pub set_max_objects: usize, pub upload_max_size: usize, - pub upload_max_concurrent: usize, + pub upload_max_concurrent: u64, pub upload_tmp_quota_size: usize, pub upload_tmp_quota_amount: usize, @@ -187,7 +187,7 @@ impl JMAP { config: &utils::config::Config, stores: &Stores, directories: &Directories, - certificates: Vec>, + servers: &mut Servers, delivery_rx: mpsc::Receiver, smtp: Arc, ) -> Result, String> { @@ -202,40 +202,19 @@ impl JMAP { let jmap_server = Arc::new(JMAP { directory: directories .directories - .get(config.value_require("jmap.directory")?) + .get(config.value_require("storage.directory")?) .failed(&format!( "Unable to find directory '{}'", - config.value_require("jmap.directory")? + config.value_require("storage.directory")? )) .clone(), snowflake_id: config - .property::("jmap.cluster.node-id")? + .property::("storage.cluster.node-id")? .map(SnowflakeIdGenerator::with_node_id) .unwrap_or_else(SnowflakeIdGenerator::new), - store: stores - .stores - .get(config.value_require("jmap.store.data")?) - .failed(&format!( - "Unable to find data store '{}'", - config.value_require("jmap.store.data")? - )) - .clone(), - fts_store: stores - .fts_stores - .get(config.value_require("jmap.store.fts")?) - .failed(&format!( - "Unable to find full text store '{}'", - config.value_require("jmap.store.fts")? - )) - .clone(), - blob_store: stores - .blob_stores - .get(config.value_require("jmap.store.blob")?) - .failed(&format!( - "Unable to find blob store '{}'", - config.value_require("jmap.store.blob")? - )) - .clone(), + store: stores.get_store(config, "storage.data")?, + fts_store: stores.get_fts_store(config, "storage.fts")?, + blob_store: stores.get_blob_store(config, "storage.blob")?, config: Config::new(config).failed("Invalid configuration file"), sessions: TtlDashMap::with_capacity( config.property("jmap.session.cache.size")?.unwrap_or(100), @@ -422,7 +401,7 @@ impl JMAP { spawn_state_manager(jmap_server.clone(), config, state_rx); // Spawn housekeeper - spawn_housekeeper(jmap_server.clone(), config, certificates, housekeeper_rx); + spawn_housekeeper(jmap_server.clone(), config, servers, housekeeper_rx); Ok(jmap_server) } diff --git a/crates/jmap/src/services/housekeeper.rs b/crates/jmap/src/services/housekeeper.rs index 63977fa5..6a968aaf 100644 --- a/crates/jmap/src/services/housekeeper.rs +++ b/crates/jmap/src/services/housekeeper.rs @@ -25,8 +25,8 @@ use std::sync::Arc; use tokio::sync::mpsc; use utils::{ - config::{cron::SimpleCron, Config}, - listener::tls::Certificate, + config::{cron::SimpleCron, Config, Servers}, + listener::blocked::BLOCKED_IP_KEY, map::ttl_dashmap::TtlMap, UnwrapFailure, }; @@ -38,6 +38,7 @@ use super::IPC_CHANNEL_BUFFER; pub enum Event { PurgeSessions, ReloadCertificates, + ReloadConfig, IndexStart, IndexDone, #[cfg(feature = "test_mode")] @@ -48,13 +49,16 @@ pub enum Event { pub fn spawn_housekeeper( core: Arc, settings: &Config, - certificates: Vec>, + servers: &mut Servers, mut rx: mpsc::Receiver, ) { let purge_cache = settings .property_or_static::("jmap.session.purge.frequency", "15 * *") .failed("Initialize housekeeper"); + let certificates = std::mem::take(&mut servers.certificates); + let blocked_ips = servers.blocked_ips.clone(); + tokio::spawn(async move { tracing::debug!("Housekeeper task started."); @@ -102,6 +106,34 @@ pub fn spawn_housekeeper( } }); } + Event::ReloadConfig => { + // Future releases will support reloading the configuration + // for now, we just reload the blocked IP addresses + let core = core.clone(); + let blocked_ips = blocked_ips.clone(); + tokio::spawn(async move { + match core.store.config_list(BLOCKED_IP_KEY).await { + Ok(config) => { + if let Err(err) = blocked_ips.reload_blocked_ips(&config) { + tracing::error!( + context = "store", + event = "error", + error = ?err, + "Failed to reload configuration." + ); + } + } + Err(err) => { + tracing::error!( + context = "store", + event = "error", + error = ?err, + "Failed to reload configuration." + ); + } + } + }); + } Event::IndexStart => { if !index_busy { index_busy = true; @@ -144,15 +176,17 @@ pub fn spawn_housekeeper( if do_purge { let core = core.clone(); + let blocked_ips = blocked_ips.clone(); tokio::spawn(async move { tracing::info!("Purging session cache."); + blocked_ips.cleanup(); core.sessions.cleanup(); core.access_tokens.cleanup(); core.oauth_codes.cleanup(); core.rate_limit_auth - .retain(|_, limiter| limiter.lock().is_active()); + .retain(|_, limiter| limiter.is_active()); core.rate_limit_unauth - .retain(|_, limiter| limiter.lock().is_active()); + .retain(|_, limiter| limiter.is_active()); }); } } diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index 640a4cf2..7e0ccebd 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -44,7 +44,7 @@ static GLOBAL: Jemalloc = Jemalloc; #[tokio::main] async fn main() -> std::io::Result<()> { - let config = Config::init(); + let mut config = Config::init(); // Enable tracing let _tracer = enable_tracing( @@ -60,17 +60,31 @@ async fn main() -> std::io::Result<()> { let mut servers = config.parse_servers().failed("Invalid configuration"); servers.bind(&config); - // Parse stores and directories + // Parse stores let stores = config.parse_stores().await.failed("Invalid configuration"); + let data_store = stores + .get_store(&config, "storage.data") + .failed("Invalid configuration"); + + // Update configuration + config.update(data_store.config_list("").await.failed("Storage error")); + servers + .blocked_ips + .reload(&config) + .failed("Invalid configuration"); + + let todo = "Update config.zip"; + + // Parse directories let directory = config - .parse_directory(&stores, config.value("jmap.store.data")) + .parse_directory(&stores, &servers, data_store) .await .failed("Invalid configuration"); let schedulers = config .parse_purge_schedules( &stores, - config.value("jmap.store.data"), - config.value("jmap.store.blob"), + config.value("storage.data"), + config.value("storage.blob"), ) .await .failed("Invalid configuration"); @@ -84,7 +98,7 @@ async fn main() -> std::io::Result<()> { &config, &stores, &directory, - std::mem::take(&mut servers.certificates), + &mut servers, delivery_rx, smtp.clone(), ) diff --git a/crates/managesieve/src/core/client.rs b/crates/managesieve/src/core/client.rs index 9207c557..9a05c4ab 100644 --- a/crates/managesieve/src/core/client.rs +++ b/crates/managesieve/src/core/client.rs @@ -227,9 +227,8 @@ impl ValidateRequest for Request { if let State::Authenticated { access_token, .. } = state { if imap .get_authenticated_limiter(access_token.primary_id()) - .lock() .request_limiter - .is_allowed() + .is_allowed(&imap.rate_requests) { Ok(self) } else { diff --git a/crates/managesieve/src/core/mod.rs b/crates/managesieve/src/core/mod.rs index d3dff2ae..5c9a2e64 100644 --- a/crates/managesieve/src/core/mod.rs +++ b/crates/managesieve/src/core/mod.rs @@ -24,14 +24,11 @@ pub mod client; pub mod session; -use std::{borrow::Cow, sync::Arc}; +use std::{borrow::Cow, net::IpAddr, sync::Arc}; use imap::core::IMAP; use imap_proto::receiver::{CommandParser, Receiver}; -use jmap::{ - auth::{rate_limit::RemoteAddress, AccessToken}, - JMAP, -}; +use jmap::{auth::AccessToken, JMAP}; use tokio::io::{AsyncRead, AsyncWrite}; use utils::listener::{limiter::InFlight, ServerInstance}; @@ -41,7 +38,7 @@ pub struct Session { pub instance: Arc, pub receiver: Receiver, pub state: State, - pub remote_addr: RemoteAddress, + pub remote_addr: IpAddr, pub stream: T, pub span: tracing::Span, pub in_flight: InFlight, diff --git a/crates/managesieve/src/core/session.rs b/crates/managesieve/src/core/session.rs index 6d4242a8..735e39d1 100644 --- a/crates/managesieve/src/core/session.rs +++ b/crates/managesieve/src/core/session.rs @@ -22,7 +22,6 @@ */ use imap_proto::receiver::{self, Receiver}; -use jmap::auth::rate_limit::RemoteAddress; use tokio_rustls::server::TlsStream; use utils::listener::{SessionManager, SessionStream}; @@ -48,7 +47,7 @@ impl SessionManager for ManageSieveSessionManager { span: session.span, stream: session.stream, in_flight: session.in_flight, - remote_addr: RemoteAddress::IpAddress(session.remote_ip), + remote_addr: session.remote_ip, }; if session diff --git a/crates/managesieve/src/op/authenticate.rs b/crates/managesieve/src/op/authenticate.rs index fc99a191..8608abf4 100644 --- a/crates/managesieve/src/op/authenticate.rs +++ b/crates/managesieve/src/op/authenticate.rs @@ -23,6 +23,7 @@ use std::sync::Arc; +use directory::AuthResult; use imap::op::authenticate::{decode_challenge_oauth, decode_challenge_plain}; use imap_proto::{ protocol::authenticate::Mechanism, @@ -89,9 +90,19 @@ impl Session { // Authenticate let access_token = match credentials { Credentials::Plain { username, secret } | Credentials::XOauth2 { username, secret } => { - self.jmap - .authenticate_plain(&username, &secret, &self.remote_addr) + match self + .jmap + .authenticate_plain(&username, &secret, self.remote_addr) .await + { + AuthResult::Success(token) => Some(token), + AuthResult::Failure => None, + AuthResult::Banned => { + return Err(StatusResponse::bye( + "Too many authentication requests from this IP address.", + )) + } + } } Credentials::OAuthBearer { token } => { match self @@ -118,7 +129,6 @@ impl Session { let in_flight = self .imap .get_authenticated_limiter(access_token.primary_id()) - .lock() .concurrent_requests .is_allowed(); if let Some(in_flight) = in_flight { diff --git a/crates/smtp/src/config/auth.rs b/crates/smtp/src/config/auth.rs index d0ae9c50..4f57c10c 100644 --- a/crates/smtp/src/config/auth.rs +++ b/crates/smtp/src/config/auth.rs @@ -111,7 +111,7 @@ impl ConfigAuth for Config { #[allow(clippy::type_complexity)] fn parse_signatures(&self, ctx: &mut ConfigContext) -> super::Result<()> { - for id in self.sub_keys("signature") { + for id in self.sub_keys("signature", ".algorithm") { let (signer, sealer) = match self.property_require::(("signature", id, "algorithm"))? { Algorithm::RsaSha256 => { diff --git a/crates/smtp/src/config/condition.rs b/crates/smtp/src/config/condition.rs index 44b608ce..163b9311 100644 --- a/crates/smtp/src/config/condition.rs +++ b/crates/smtp/src/config/condition.rs @@ -63,7 +63,7 @@ impl ConfigCondition for Config { 'outer: loop { let mut op_str = ""; - for key in self.sub_keys(prefix.as_str()) { + for key in self.sub_keys(prefix.as_str(), "") { if !["if", "then"].contains(&key) { if op_str.is_empty() { op_str = key; @@ -81,7 +81,9 @@ impl ConfigCondition for Config { stack.push(( std::mem::replace( &mut iter, - self.sub_keys((&prefix, op_str).as_key()).peekable().into(), + self.sub_keys((&prefix, op_str).as_key(), "") + .peekable() + .into(), ), (&prefix, op_str).as_key(), std::mem::take(&mut jmp_pos), @@ -301,7 +303,7 @@ impl ConfigCondition for Config { EnvelopeKey::Mx, ]; - for rule_name in self.sub_keys("rule") { + for rule_name in self.sub_keys("rule", "") { conditions.insert( rule_name.to_string(), self.parse_condition(("rule", rule_name), ctx, &available_keys)?, diff --git a/crates/smtp/src/config/mod.rs b/crates/smtp/src/config/mod.rs index a8057ca4..1eb24bb7 100644 --- a/crates/smtp/src/config/mod.rs +++ b/crates/smtp/src/config/mod.rs @@ -50,7 +50,7 @@ use mail_send::Credentials; use regex::Regex; use sieve::Sieve; use smtp_proto::MtPriority; -use store::Stores; +use store::{LookupStore, Store, Stores}; use utils::config::{ipmask::IpAddrMask, DynValue, Rate, Server, ServerProtocol}; use crate::{core::Lookup, inbound::milter}; @@ -342,7 +342,11 @@ pub struct QueueConfig { // Throttle and Quotas pub throttle: QueueThrottle, pub quota: QueueQuotas, - pub management_lookup: Arc, + + // Default store and directory + pub directory: Arc, + pub data_store: Store, + pub lookup_store: LookupStore, } pub struct QueueOutboundSourceIp { diff --git a/crates/smtp/src/config/queue.rs b/crates/smtp/src/config/queue.rs index 93a1c668..7ede3adb 100644 --- a/crates/smtp/src/config/queue.rs +++ b/crates/smtp/src/config/queue.rs @@ -203,17 +203,29 @@ impl ConfigQueue for Config { .unwrap_or_default() .map_if_block(&ctx.signers, "report.dsn.sign", "signature")?, }, - management_lookup: if let Some(id) = self.value("management.directory") { - ctx.directory - .directories - .get(id) - .ok_or_else(|| { - format!("Directory {id:?} not found for key \"management.directory\".") - })? - .clone() - } else { - Arc::new(Directory::default()) - }, + directory: ctx + .directory + .directories + .get(self.value_require("storage.directory")?) + .ok_or_else(|| { + format!( + "Directory {:?} not found for key \"storage.directory\".", + self.value_require("storage.directory").unwrap() + ) + })? + .clone(), + data_store: ctx.stores.get_store(self, "storage.data")?, + lookup_store: self + .value_or_default("storage.lookup", "storage.data") + .and_then(|id| ctx.stores.lookup_stores.get(id)) + .ok_or_else(|| { + format!( + "Lookup store {:?} not found for key \"storage.lookup\".", + self.value_or_default("storage.lookup", "storage.data") + .unwrap() + ) + })? + .clone(), }; if config.retry.has_empty_list() { @@ -292,7 +304,7 @@ impl ConfigQueue for Config { rcpt_domain: Vec::new(), }; - for array_pos in self.sub_keys("queue.quota") { + for array_pos in self.sub_keys("queue.quota", "") { let quota = self.parse_queue_quota_item(("queue.quota", array_pos), ctx)?; if (quota.keys & THROTTLE_RCPT) != 0 diff --git a/crates/smtp/src/config/remote.rs b/crates/smtp/src/config/remote.rs index 1ea5292d..68739ad5 100644 --- a/crates/smtp/src/config/remote.rs +++ b/crates/smtp/src/config/remote.rs @@ -34,7 +34,7 @@ pub trait ConfigHost { impl ConfigHost for Config { fn parse_remote_hosts(&self, ctx: &mut ConfigContext) -> super::Result<()> { - for id in self.sub_keys("remote") { + for id in self.sub_keys("remote", ".address") { ctx.hosts.insert(id.to_string(), self.parse_host(id)?); } diff --git a/crates/smtp/src/config/scripts.rs b/crates/smtp/src/config/scripts.rs index 1ddbf1a2..35ab1d51 100644 --- a/crates/smtp/src/config/scripts.rs +++ b/crates/smtp/src/config/scripts.rs @@ -154,12 +154,12 @@ impl ConfigSieve for Config { runtime.set_local_hostname(hostname.to_string()); // Parse scripts - for id in self.sub_keys("sieve.trusted.scripts") { + for id in self.sub_keys("sieve.trusted.scripts", "") { let key = ("sieve.trusted.scripts", id); let script = if !self.contains_key(key) { let mut script = Vec::new(); - for sub_key in self.sub_keys(key) { + for sub_key in self.sub_keys(key, "") { script.extend(self.file_contents(("sieve.trusted.scripts", id, sub_key))?); } script @@ -195,14 +195,6 @@ impl ConfigSieve for Config { scripts: ctx.scripts.clone(), lookup_stores: ctx.stores.lookup_stores.clone(), directories: ctx.directory.directories.clone(), - default_directory: self - .value("sieve.trusted.default.directory") - .and_then(|id| ctx.directory.directories.get(id)) - .cloned(), - default_lookup_store: self - .value("sieve.trusted.default.store") - .and_then(|id| ctx.stores.lookup_stores.get(id)) - .cloned(), from_addr: self .value("sieve.trusted.from-addr") .map(|a| a.to_string()) diff --git a/crates/smtp/src/config/session.rs b/crates/smtp/src/config/session.rs index c86a348f..99cd9db2 100644 --- a/crates/smtp/src/config/session.rs +++ b/crates/smtp/src/config/session.rs @@ -440,7 +440,7 @@ impl ConfigSession for Config { available_keys: &[EnvelopeKey], ) -> super::Result> { let mut pipes = Vec::new(); - for id in self.sub_keys("session.data.pipe") { + for id in self.sub_keys("session.data.pipe", "") { pipes.push(Pipe { command: self .parse_if_block(("session.data.pipe", id, "command"), ctx, available_keys)? @@ -462,7 +462,7 @@ impl ConfigSession for Config { available_keys: &[EnvelopeKey], ) -> super::Result> { let mut milters = Vec::new(); - for id in self.sub_keys("session.data.milter") { + for id in self.sub_keys("session.data.milter", "") { let hostname = self .value_require(("session.data.milter", id, "hostname"))? .to_string(); diff --git a/crates/smtp/src/config/throttle.rs b/crates/smtp/src/config/throttle.rs index 237026d6..d04d1e99 100644 --- a/crates/smtp/src/config/throttle.rs +++ b/crates/smtp/src/config/throttle.rs @@ -55,7 +55,7 @@ impl ConfigThrottle for Config { ) -> super::Result> { let prefix_ = prefix.as_key(); let mut throttles = Vec::new(); - for array_pos in self.sub_keys(prefix) { + for array_pos in self.sub_keys(prefix, "") { throttles.push(self.parse_throttle_item( (&prefix_, array_pos), ctx, diff --git a/crates/smtp/src/core/management.rs b/crates/smtp/src/core/management.rs index f57c0b47..8d4b7057 100644 --- a/crates/smtp/src/core/management.rs +++ b/crates/smtp/src/core/management.rs @@ -23,7 +23,7 @@ use std::{borrow::Cow, fmt::Display, net::IpAddr, sync::Arc, time::Instant}; -use directory::{QueryBy, Type}; +use directory::{AuthResult, Type}; use http_body_util::{combinators::BoxBody, BodyExt, Empty, Full}; use hyper::{ body::{self, Bytes}, @@ -186,7 +186,7 @@ async fn handle_request( let core = core.clone(); async move { - let response = core.parse_request(&req).await; + let response = core.parse_request(&req, remote_addr).await; tracing::debug!( context = "management", @@ -218,6 +218,7 @@ impl SMTP { async fn parse_request( &self, req: &hyper::Request, + remote_addr: IpAddr, ) -> Result>, hyper::Error> { // Authenticate request let mut is_authenticated = false; @@ -240,24 +241,21 @@ impl SMTP { match self .queue .config - .management_lookup - .query( - QueryBy::Credentials(&Credentials::Plain { username, secret }), - false, - ) + .directory + .authenticate(&Credentials::Plain { username, secret }, remote_addr, false) .await { - Ok(Some(principal)) if principal.typ == Type::Superuser => { + Ok(AuthResult::Success(principal)) if principal.typ == Type::Superuser => { is_authenticated = true; } - Ok(Some(_)) => { + Ok(AuthResult::Success(_)) => { tracing::debug!( context = "management", event = "auth-error", "Insufficient privileges." ); } - Ok(None) => { + Ok(AuthResult::Failure | AuthResult::Banned) => { tracing::debug!( context = "management", event = "auth-error", diff --git a/crates/smtp/src/core/mod.rs b/crates/smtp/src/core/mod.rs index 04003099..718309c5 100644 --- a/crates/smtp/src/core/mod.rs +++ b/crates/smtp/src/core/mod.rs @@ -119,8 +119,6 @@ pub struct SieveCore { pub sign: Vec>, pub directories: AHashMap>, pub lookup_stores: AHashMap, - pub default_lookup_store: Option, - pub default_directory: Option>, } pub struct Resolvers { @@ -428,7 +426,8 @@ static ref SIEVE: Arc = Arc::new(utils::listener::ServerInstance acceptor: TcpAcceptor::Plain, limiter: utils::listener::limiter::ConcurrencyLimiter::new(0), shutdown_rx: tokio::sync::watch::channel(false).1, - proxy_networks: vec![] + proxy_networks: vec![], + blocked_ips: Arc::new(Default::default()), }); } diff --git a/crates/smtp/src/core/throttle.rs b/crates/smtp/src/core/throttle.rs index 783efdb3..34bf3f4c 100644 --- a/crates/smtp/src/core/throttle.rs +++ b/crates/smtp/src/core/throttle.rs @@ -29,7 +29,6 @@ use utils::config::{KeyLookup, Rate}; use std::{ hash::{BuildHasher, Hash, Hasher}, net::IpAddr, - time::Duration, }; use crate::config::*; @@ -253,14 +252,14 @@ impl Session { return false; } } - if let Some(limiter) = &mut limiter.rate { - if !limiter.is_allowed() { + if let (Some(limiter), Some(rate)) = (&mut limiter.rate, &t.rate) { + if !limiter.is_allowed(rate) { tracing::debug!( parent: &self.span, context = "throttle", event = "rate-limit-exceeded", - max_requests = limiter.max_requests, - max_interval = limiter.max_interval.as_secs(), + max_requests = rate.requests, + max_interval = rate.period.as_secs(), "Rate limit exceeded." ); return false; @@ -276,11 +275,8 @@ impl Session { limiter }); let rate = t.rate.as_ref().map(|rate| { - let mut r = RateLimiter::new( - rate.requests, - std::cmp::min(rate.period, Duration::from_secs(1)), - ); - r.is_allowed(); + let r = RateLimiter::new(rate); + r.is_allowed(rate); r }); @@ -306,14 +302,14 @@ impl Session { match self.core.session.throttle.entry(key) { Entry::Occupied(mut e) => { if let Some(limiter) = &mut e.get_mut().rate { - limiter.is_allowed() + limiter.is_allowed(rate) } else { false } } Entry::Vacant(e) => { - let mut limiter = RateLimiter::new(rate.requests, rate.period); - limiter.is_allowed(); + let limiter = RateLimiter::new(rate); + limiter.is_allowed(rate); e.insert(Limiter { rate: limiter.into(), concurrency: None, diff --git a/crates/smtp/src/core/worker.rs b/crates/smtp/src/core/worker.rs index 57841516..4dafc1e5 100644 --- a/crates/smtp/src/core/worker.rs +++ b/crates/smtp/src/core/worker.rs @@ -58,9 +58,7 @@ impl SMTP { v.concurrency .as_ref() .map_or(false, |c| c.concurrent.load(Ordering::Relaxed) > 0) - || v.rate - .as_ref() - .map_or(false, |r| r.elapsed() < r.max_interval) + || v.rate.as_ref().map_or(false, |r| r.is_active()) }); } self.queue.quota.retain(|_, v| { diff --git a/crates/smtp/src/inbound/auth.rs b/crates/smtp/src/inbound/auth.rs index 66e05092..3be783d4 100644 --- a/crates/smtp/src/inbound/auth.rs +++ b/crates/smtp/src/inbound/auth.rs @@ -21,7 +21,7 @@ * for more details. */ -use directory::QueryBy; +use directory::AuthResult; use mail_parser::decoders::base64::base64_decode; use mail_send::Credentials; use smtp_proto::{IntoString, AUTH_LOGIN, AUTH_OAUTHBEARER, AUTH_PLAIN, AUTH_XOAUTH2}; @@ -181,17 +181,19 @@ impl Session { | Credentials::XOauth2 { username, .. } | Credentials::OAuthBearer { token: username } => username.to_string(), }; - if let Ok(principal) = lookup - .query(QueryBy::Credentials(&credentials), false) + + match lookup + .authenticate(&credentials, self.data.remote_ip, false) .await { - tracing::debug!( - parent: &self.span, - context = "auth", - event = "authenticate", - result = if principal.is_some() {"success"} else {"failed"} - ); - return if let Some(principal) = principal { + Ok(AuthResult::Success(principal)) => { + tracing::debug!( + parent: &self.span, + context = "auth", + event = "authenticate", + result = "success" + ); + self.data.authenticated_as = authenticated_as.to_lowercase(); self.data.authenticated_emails = principal .emails @@ -201,11 +203,31 @@ impl Session { self.eval_post_auth_params().await; self.write(b"235 2.7.0 Authentication succeeded.\r\n") .await?; - Ok(false) - } else { - self.auth_error(b"535 5.7.8 Authentication credentials invalid.\r\n") - .await - }; + return Ok(false); + } + Ok(AuthResult::Failure) => { + tracing::debug!( + parent: &self.span, + context = "auth", + event = "authenticate", + result = "failed" + ); + + return self + .auth_error(b"535 5.7.8 Authentication credentials invalid.\r\n") + .await; + } + Ok(AuthResult::Banned) => { + tracing::debug!( + parent: &self.span, + context = "auth", + event = "authenticate", + result = "banned" + ); + + return Err(()); + } + _ => (), } } else { tracing::warn!( diff --git a/crates/smtp/src/queue/throttle.rs b/crates/smtp/src/queue/throttle.rs index 0699f78b..a0915260 100644 --- a/crates/smtp/src/queue/throttle.rs +++ b/crates/smtp/src/queue/throttle.rs @@ -21,7 +21,7 @@ * for more details. */ -use std::time::Instant; +use std::time::{Duration, Instant}; use dashmap::mapref::entry::Entry; use utils::{ @@ -70,18 +70,19 @@ impl QueueCore { }); } } - if let Some(limiter) = &mut limiter.rate { - if !limiter.is_allowed() { + if let (Some(limiter), Some(rate)) = (&mut limiter.rate, &throttle.rate) { + if !limiter.is_allowed(rate) { tracing::info!( parent: span, context = "throttle", event = "rate-limit-exceeded", - max_requests = limiter.max_requests, - max_interval = limiter.max_interval.as_secs(), + max_requests = rate.requests, + max_interval = rate.period.as_secs(), "Queue rate limit exceeded." ); return Err(Error::Rate { - retry_at: limiter.retry_at(), + retry_at: Instant::now() + + Duration::from_secs(limiter.secs_to_refill()), }); } } @@ -95,8 +96,8 @@ impl QueueCore { limiter }); let rate = throttle.rate.as_ref().map(|rate| { - let mut r = RateLimiter::new(rate.requests, rate.period); - r.is_allowed(); + let r = RateLimiter::new(rate); + r.is_allowed(rate); r }); diff --git a/crates/smtp/src/scripts/plugins/bayes.rs b/crates/smtp/src/scripts/plugins/bayes.rs index ddf04c1c..b7566dd1 100644 --- a/crates/smtp/src/scripts/plugins/bayes.rs +++ b/crates/smtp/src/scripts/plugins/bayes.rs @@ -64,7 +64,7 @@ fn train(ctx: PluginContext<'_>, is_train: bool) -> Variable { let span: &tracing::Span = ctx.span; let store = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), - _ => ctx.core.sieve.default_lookup_store.as_ref(), + _ => Some(&ctx.core.queue.config.lookup_store), }; let store = if let Some(store) = store { @@ -163,7 +163,7 @@ pub fn exec_classify(ctx: PluginContext<'_>) -> Variable { let span = ctx.span; let store = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), - _ => ctx.core.sieve.default_lookup_store.as_ref(), + _ => Some(&ctx.core.queue.config.lookup_store), }; let store = if let Some(store) = store { store @@ -262,7 +262,7 @@ pub fn exec_is_balanced(ctx: PluginContext<'_>) -> Variable { let span = ctx.span; let store = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), - _ => ctx.core.sieve.default_lookup_store.as_ref(), + _ => Some(&ctx.core.queue.config.lookup_store), }; let store = if let Some(store) = store { store diff --git a/crates/smtp/src/scripts/plugins/lookup.rs b/crates/smtp/src/scripts/plugins/lookup.rs index 8b5c0792..c4e1d4a9 100644 --- a/crates/smtp/src/scripts/plugins/lookup.rs +++ b/crates/smtp/src/scripts/plugins/lookup.rs @@ -62,7 +62,7 @@ pub fn register_local_domain(plugin_id: u32, fnc_map: &mut FunctionMap) -> Variable { let store = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), - _ => ctx.core.sieve.default_lookup_store.as_ref(), + _ => Some(&ctx.core.queue.config.lookup_store), }; if let Some(store) = store { @@ -108,7 +108,7 @@ pub fn exec(ctx: PluginContext<'_>) -> Variable { pub fn exec_get(ctx: PluginContext<'_>) -> Variable { let store = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), - _ => ctx.core.sieve.default_lookup_store.as_ref(), + _ => Some(&ctx.core.queue.config.lookup_store), }; if let Some(store) = store { @@ -137,7 +137,7 @@ pub fn exec_get(ctx: PluginContext<'_>) -> Variable { pub fn exec_set(ctx: PluginContext<'_>) -> Variable { let store = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.lookup_stores.get(v.as_ref()), - _ => ctx.core.sieve.default_lookup_store.as_ref(), + _ => Some(&ctx.core.queue.config.lookup_store), }; if let Some(store) = store { @@ -400,7 +400,7 @@ pub fn exec_local_domain(ctx: PluginContext<'_>) -> Variable { if !domain.is_empty() { let directory = match &ctx.arguments[0] { Variable::String(v) if !v.is_empty() => ctx.core.sieve.directories.get(v.as_ref()), - _ => ctx.core.sieve.default_directory.as_ref(), + _ => Some(&ctx.core.queue.config.directory), }; if let Some(directory) = directory { diff --git a/crates/store/src/backend/sqlite/main.rs b/crates/store/src/backend/sqlite/main.rs index 781883b2..f89ed166 100644 --- a/crates/store/src/backend/sqlite/main.rs +++ b/crates/store/src/backend/sqlite/main.rs @@ -76,6 +76,23 @@ impl SqliteStore { Ok(db) } + #[cfg(feature = "test_mode")] + pub fn open_memory() -> crate::Result { + let db = Self { + conn_pool: Pool::builder() + .max_size(1) + .build(SqliteConnectionManager::memory())?, + worker_pool: rayon::ThreadPoolBuilder::new() + .num_threads(num_cpus::get()) + .build() + .map_err(|err| { + crate::Error::InternalError(format!("Failed to build worker pool: {}", err)) + })?, + }; + db.create_tables()?; + Ok(db) + } + pub(super) fn create_tables(&self) -> crate::Result<()> { let conn = self.conn_pool.get()?; diff --git a/crates/store/src/config.rs b/crates/store/src/config.rs index 5af1c978..a727f5a4 100644 --- a/crates/store/src/config.rs +++ b/crates/store/src/config.rs @@ -72,7 +72,7 @@ impl ConfigStore for Config { async fn parse_stores(&self) -> utils::config::Result { let mut config = Stores::default(); - for id in self.sub_keys("store") { + for id in self.sub_keys("store", ".type") { // Parse store if self.property_or_static::(("store", id, "disable"), "false")? { tracing::debug!("Skipping disabled store {id:?}."); @@ -190,7 +190,7 @@ impl ConfigStore for Config { // Add queries as lookup stores let lookup_store: LookupStore = lookup_store.into(); - for lookup_id in self.sub_keys(("store", id, "query")) { + for lookup_id in self.sub_keys(("store", id, "query"), "") { config.lookup_stores.insert( format!("{store_id}/{lookup_id}"), LookupStore::Query(Arc::new(QueryStore { diff --git a/crates/store/src/dispatch/config.rs b/crates/store/src/dispatch/config.rs new file mode 100644 index 00000000..0d610474 --- /dev/null +++ b/crates/store/src/dispatch/config.rs @@ -0,0 +1,90 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use utils::config::{Config, ConfigKey}; + +use crate::{ + write::{BatchBuilder, ValueClass}, + Deserialize, IterateParams, Store, ValueKey, +}; + +impl Store { + pub async fn config_get(&self, key: impl Into) -> crate::Result> { + self.get_value(ValueKey::from(ValueClass::Config(key.into().into_bytes()))) + .await + } + + pub async fn config_list(&self, key: impl AsRef) -> crate::Result { + let key = key.as_ref().as_bytes(); + let from_key = ValueKey::from(ValueClass::Config(key.to_vec())); + let to_key = ValueKey::from(ValueClass::Config( + key.iter() + .copied() + .chain([u8::MAX, u8::MAX, u8::MAX, u8::MAX, u8::MAX]) + .collect::>(), + )); + let mut config = Config::default(); + self.iterate( + IterateParams::new(from_key, to_key).ascending(), + |key, value| { + config.keys.insert( + String::deserialize(key.get(1..).unwrap_or_default())?, + String::deserialize(value)?, + ); + Ok(true) + }, + ) + .await?; + + Ok(config) + } + + pub async fn config_set(&self, keys: impl Iterator) -> crate::Result<()> { + let mut batch = BatchBuilder::new(); + for key in keys { + batch.set(ValueClass::Config(key.key.into_bytes()), key.value); + } + self.write(batch.build()).await + } + + pub async fn config_clear(&self, key: impl Into) -> crate::Result<()> { + let mut batch = BatchBuilder::new(); + batch.clear(ValueClass::Config(key.into().into_bytes())); + self.write(batch.build()).await + } + + pub async fn config_clear_prefix(&self, key: impl AsRef) -> crate::Result<()> { + self.delete_range( + ValueKey::from(ValueClass::Config(key.as_ref().as_bytes().to_vec())), + ValueKey::from(ValueClass::Config( + key.as_ref() + .as_bytes() + .iter() + .copied() + .chain([u8::MAX, u8::MAX, u8::MAX, u8::MAX, u8::MAX]) + .collect::>(), + )), + ) + .await + } +} diff --git a/crates/store/src/dispatch/mod.rs b/crates/store/src/dispatch/mod.rs index dbc34c81..c4641e59 100644 --- a/crates/store/src/dispatch/mod.rs +++ b/crates/store/src/dispatch/mod.rs @@ -22,6 +22,7 @@ */ pub mod blob; +pub mod config; pub mod fts; pub mod lookup; pub mod store; diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 06f2aca0..96e6ea40 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -650,3 +650,87 @@ impl From for Vec { .collect() } } + +impl std::fmt::Debug for Store { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + #[cfg(feature = "sqlite")] + Self::SQLite(_) => f.debug_tuple("SQLite").finish(), + #[cfg(feature = "foundation")] + Self::FoundationDb(_) => f.debug_tuple("FoundationDb").finish(), + #[cfg(feature = "postgres")] + Self::PostgreSQL(_) => f.debug_tuple("PostgreSQL").finish(), + #[cfg(feature = "mysql")] + Self::MySQL(_) => f.debug_tuple("MySQL").finish(), + #[cfg(feature = "rocks")] + Self::RocksDb(_) => f.debug_tuple("RocksDb").finish(), + } + } +} + +#[cfg(feature = "test_mode")] +impl Default for Store { + fn default() -> Self { + #[cfg(feature = "sqlite")] + { + Self::SQLite(Arc::new(SqliteStore::open_memory().unwrap())) + } + #[cfg(not(feature = "sqlite"))] + { + unreachable!("No default store available") + } + } +} + +impl Stores { + pub fn get_store( + &self, + config: &utils::config::Config, + key: &str, + ) -> utils::config::Result { + self.stores + .get(config.value_require(key)?) + .cloned() + .ok_or_else(|| { + format!( + "Unable to find data store '{}' defined in key '{}'", + config.value_require(key).unwrap(), + key + ) + }) + } + + pub fn get_blob_store( + &self, + config: &utils::config::Config, + key: &str, + ) -> utils::config::Result { + self.blob_stores + .get(config.value_require(key)?) + .cloned() + .ok_or_else(|| { + format!( + "Unable to find blob store '{}' defined in key '{}'", + config.value_require(key).unwrap(), + key + ) + }) + } + + pub fn get_fts_store( + &self, + config: &utils::config::Config, + key: &str, + ) -> utils::config::Result { + self.fts_stores + .get(config.value_require(key)?) + .cloned() + .ok_or_else(|| { + format!( + "Unable to find FTS store '{}' defined in key '{}'", + config.value_require(key).unwrap(), + key + ) + }) + } +} diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index dd1834f9..89e9cd3e 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -275,6 +275,7 @@ impl + Sync + Send> Key for ValueKey { .write(self.collection) .write(self.document_id), }, + ValueClass::Config(key) => serializer.write(8u8).write(key.as_slice()), ValueClass::Directory(directory) => match directory { DirectoryClass::NameToId(name) => serializer.write(20u8).write(name.as_slice()), DirectoryClass::EmailToId(email) => serializer.write(21u8).write(email.as_slice()), @@ -424,7 +425,7 @@ impl ValueClass { U32_LEN * 2 + 3 } ValueClass::Acl(_) => U32_LEN * 3 + 2, - ValueClass::Key(v) => v.len(), + ValueClass::Key(v) | ValueClass::Config(v) => v.len(), ValueClass::Directory(d) => match d { DirectoryClass::NameToId(v) | DirectoryClass::EmailToId(v) diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index 64517bf1..47b2cadb 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -137,6 +137,7 @@ pub enum ValueClass { Directory(DirectoryClass), Blob(BlobOp), IndexEmail(u64), + Config(Vec), } #[derive(Debug, PartialEq, Clone, Eq, Hash)] diff --git a/crates/utils/src/config/ipmask.rs b/crates/utils/src/config/ipmask.rs index 5f6a50b2..e7a9d43e 100644 --- a/crates/utils/src/config/ipmask.rs +++ b/crates/utils/src/config/ipmask.rs @@ -129,3 +129,30 @@ impl ParseValue for IpAddrMask { )) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_ipaddrmask() { + for (mask, ip) in [ + ("10.0.0.0/8", "10.30.20.11"), + ("10.0.0.0/8", "10.0.13.73"), + ("192.168.1.1", "192.168.1.1"), + ] { + let mask = IpAddrMask::parse_value("test", mask).unwrap(); + let ip = ip.parse::().unwrap(); + assert!(mask.matches(&ip)); + } + + for (mask, ip) in [ + ("10.0.0.0/8", "11.30.20.11"), + ("192.168.1.1", "193.168.1.1"), + ] { + let mask = IpAddrMask::parse_value("test", mask).unwrap(); + let ip = ip.parse::().unwrap(); + assert!(!mask.matches(&ip)); + } + } +} diff --git a/crates/utils/src/config/listener.rs b/crates/utils/src/config/listener.rs index f2bcb90c..c506e0f2 100644 --- a/crates/utils/src/config/listener.rs +++ b/crates/utils/src/config/listener.rs @@ -43,6 +43,7 @@ use tokio_rustls::TlsAcceptor; use crate::{ acme::{directory::ACME_TLS_ALPN_NAME, AcmeManager}, listener::{ + blocked::BlockedIps, tls::{Certificate, CertificateResolver}, TcpAcceptor, }, @@ -57,14 +58,16 @@ use super::{ impl Config { pub fn parse_servers(&self) -> super::Result { + let mut servers = Servers::default(); + // Parse certificates and ACME managers let certificates = self.parse_certificates()?; let acmes = self.parse_acmes()?; // Parse servers - let mut servers = Servers::default(); - for (internal_id, id) in self.sub_keys("server.listener").enumerate() { - let mut server = self.parse_server(id, &certificates, &acmes)?; + for (internal_id, id) in self.sub_keys("server.listener", ".protocol").enumerate() { + let mut server = + self.parse_server(id, &certificates, &acmes, servers.blocked_ips.clone())?; if !servers.inner.iter().any(|s| s.id == server.id) { server.internal_id = internal_id as u16; servers.inner.push(server); @@ -113,6 +116,7 @@ impl Config { id: &str, certificates: &AHashMap>, acmes: &AHashMap>, + blocked_ips: Arc, ) -> super::Result { // Build listeners let mut listeners = Vec::new(); @@ -330,11 +334,11 @@ impl Config { // Parse proxy networks let mut proxy_networks = Vec::new(); - for (key, protocol) in self.values_or_default( - ("server.listener", id, "proxy-trusted-networks"), - "server.proxy-trusted-networks", + for network in self.set_values_or_default( + ("server.listener", id, "proxy.trusted-networks"), + "server.proxy.trusted-networks", ) { - proxy_networks.push(protocol.parse_key(key)?); + proxy_networks.push(network.parse_key("server.proxy.trusted-networks")?); } Ok(Server { @@ -374,6 +378,7 @@ impl Config { acceptor, tls_implicit, proxy_networks, + blocked_ips, }) } } diff --git a/crates/utils/src/config/mod.rs b/crates/utils/src/config/mod.rs index a0c71805..3069aaa6 100644 --- a/crates/utils/src/config/mod.rs +++ b/crates/utils/src/config/mod.rs @@ -44,7 +44,7 @@ use tokio::net::TcpSocket; use crate::{ acme::AcmeManager, failed, - listener::{tls::Certificate, TcpAcceptor}, + listener::{blocked::BlockedIps, tls::Certificate, TcpAcceptor}, UnwrapFailure, }; @@ -55,6 +55,12 @@ pub struct Config { pub keys: BTreeMap, } +#[derive(Debug, Default, PartialEq, Eq)] +pub struct ConfigKey { + pub key: String, + pub value: String, +} + #[derive(Debug, Default)] pub struct Server { pub id: String, @@ -64,6 +70,7 @@ pub struct Server { pub protocol: ServerProtocol, pub listeners: Vec, pub proxy_networks: Vec, + pub blocked_ips: Arc, pub acceptor: TcpAcceptor, pub tls_implicit: bool, pub max_connections: u64, @@ -74,6 +81,7 @@ pub struct Servers { pub inner: Vec, pub certificates: Vec>, pub acme_managers: Vec>, + pub blocked_ips: Arc, } #[derive(Debug)] @@ -219,6 +227,10 @@ impl Config { config } + + pub fn update(&mut self, config: Self) { + self.keys.extend(config.keys); + } } trait ReplaceMacros: Sized { diff --git a/crates/utils/src/config/parser.rs b/crates/utils/src/config/parser.rs index b29c42b1..ff9779ec 100644 --- a/crates/utils/src/config/parser.rs +++ b/crates/utils/src/config/parser.rs @@ -108,11 +108,14 @@ impl Config { parser.skip_line(); } 'a'..='z' | 'A'..='Z' | '0'..='9' | '\"' => { - let key = parser.key(if !table_name.is_empty() { - format!("{table_name}.") - } else { - String::with_capacity(10) - })?; + let (key, _) = parser.key( + if !table_name.is_empty() { + format!("{table_name}.") + } else { + String::with_capacity(10) + }, + false, + )?; parser.value(key, &['\n'], 0)?; } '#' => { @@ -204,12 +207,19 @@ impl<'x, 'y> TomlParser<'x, 'y> { } #[allow(clippy::while_let_on_iterator)] - fn key(&mut self, mut key: String) -> Result { + fn key(&mut self, mut key: String, in_curly: bool) -> Result<(String, char)> { while let Some(ch) = self.iter.next() { match ch { '=' => { if !key.is_empty() { - return Ok(key); + return Ok((key, ch)); + } else { + return Err(format!("Empty key at line: {}", self.line)); + } + } + ',' | '}' if in_curly => { + if !key.is_empty() { + return Ok((key, ch)); } else { return Err(format!("Empty key at line: {}", self.line)); } @@ -281,23 +291,47 @@ impl<'x, 'y> TomlParser<'x, 'y> { } } } - '{' => loop { - let sub_key = self.key(format!("{key}."))?; - self.seek_next_char(); + '{' => { + let base_key = format!("{key}."); + let base_key_len = base_key.len(); - match self.value(sub_key, &[',', '}'], nest_level + 1)? { - ',' => { - self.seek_next_char(); - } - '}' => break, - ch => { - return Err(format!( - "Unexpected character {:?} found in inline table for property {:?} at line {}.", - ch, key, self.line - )); + loop { + let (sub_key, stop_char) = self.key(base_key.clone(), true)?; + match stop_char { + '=' => { + // Key value + self.seek_next_char(); + + match self.value(sub_key, &[',', '}'], nest_level + 1)? { + ',' => { + self.seek_next_char(); + } + '}' => break, + ch => { + return Err(format!( + "Unexpected character {:?} found in inline table for property {:?} at line {}.", + ch, key, self.line + )); + } + } + } + ',' => { + // Set + if sub_key.len() > base_key_len { + self.insert_key(sub_key, String::new())?; + } + } + '}' => { + // Set + if sub_key.len() > base_key_len { + self.insert_key(sub_key, String::new())?; + } + break; + } + _ => unreachable!(), } } - }, + } qch @ ('\'' | '\"') => { let mut value = String::new(); if matches!(self.iter.peek(), Some(ch) if ch == &qch) { @@ -455,136 +489,86 @@ mod tests { let mut config = Config::default(); config.parse(&fs::read_to_string(file).unwrap()).unwrap(); - let expected = BTreeMap::from_iter([ - ("arrays.colors.0000".to_string(), "red".to_string()), - ("arrays.colors.0001".to_string(), "yellow".to_string()), - ("arrays.colors.0002".to_string(), "green".to_string()), - ( - "arrays.contributors.0000".to_string(), - "Foo Bar ".to_string(), - ), - ( - "arrays.contributors.0001.email".to_string(), - "bazqux@example.com".to_string(), - ), - ( - "arrays.contributors.0001.name".to_string(), - "Baz Qux".to_string(), - ), - ( - "arrays.contributors.0001.url".to_string(), - "https://example.com/bazqux".to_string(), - ), - ("arrays.integers.0000".to_string(), "1".to_string()), - ("arrays.integers.0001".to_string(), "2".to_string()), - ("arrays.integers.0002".to_string(), "3".to_string()), - ("arrays.integers2.0000".to_string(), "1".to_string()), - ("arrays.integers2.0001".to_string(), "2".to_string()), - ("arrays.integers2.0002".to_string(), "3".to_string()), - ("arrays.integers3.0000".to_string(), "4".to_string()), - ("arrays.integers3.0001".to_string(), "5".to_string()), - ( - "arrays.nested_arrays_of_ints.0000.0000".to_string(), - "1".to_string(), - ), - ( - "arrays.nested_arrays_of_ints.0000.0001".to_string(), - "2".to_string(), - ), - ( - "arrays.nested_arrays_of_ints.0001.0000".to_string(), - "3".to_string(), - ), - ( - "arrays.nested_arrays_of_ints.0001.0001".to_string(), - "4".to_string(), - ), - ( - "arrays.nested_arrays_of_ints.0001.0002".to_string(), - "5".to_string(), - ), - ( - "arrays.nested_mixed_array.0000.0000".to_string(), - "1".to_string(), - ), - ( - "arrays.nested_mixed_array.0000.0001".to_string(), - "2".to_string(), - ), - ( - "arrays.nested_mixed_array.0001.0000".to_string(), - "a".to_string(), - ), - ( - "arrays.nested_mixed_array.0001.0001".to_string(), - "b".to_string(), - ), - ( - "arrays.nested_mixed_array.0001.0002".to_string(), - "c".to_string(), - ), - ("arrays.numbers.0000".to_string(), "0.1".to_string()), - ("arrays.numbers.0001".to_string(), "0.2".to_string()), - ("arrays.numbers.0002".to_string(), "0.5".to_string()), - ("arrays.numbers.0003".to_string(), "1".to_string()), - ("arrays.numbers.0004".to_string(), "2".to_string()), - ("arrays.numbers.0005".to_string(), "5".to_string()), - ("arrays.string_array.0000".to_string(), "all".to_string()), - ( - "arrays.string_array.0001".to_string(), - "strings".to_string(), - ), - ( - "arrays.string_array.0002".to_string(), - "are the same".to_string(), - ), - ("arrays.string_array.0003".to_string(), "type".to_string()), - ("database.data.0000.0000".to_string(), "delta".to_string()), - ("database.data.0000.0001".to_string(), "phi".to_string()), - ("database.data.0001.0000".to_string(), "3.14".to_string()), - ("database.enabled".to_string(), "true".to_string()), - ("database.ports.0000".to_string(), "8000".to_string()), - ("database.ports.0001".to_string(), "8001".to_string()), - ("database.ports.0002".to_string(), "8002".to_string()), - ("database.temp_targets.case".to_string(), "72.0".to_string()), - ("database.temp_targets.cpu".to_string(), "79.5".to_string()), - ("products.0000.name".to_string(), "Hammer".to_string()), - ("products.0000.sku".to_string(), "738594937".to_string()), - ("products.0002.color".to_string(), "gray".to_string()), - ("products.0002.name".to_string(), "Nail".to_string()), - ("products.0002.sku".to_string(), "284758393".to_string()), - ("servers.127.0.0.1".to_string(), "value".to_string()), - ("servers.alpha.ip".to_string(), "10.0.0.1".to_string()), - ("servers.alpha.role".to_string(), "frontend".to_string()), - ("servers.beta.ip".to_string(), "10.0.0.2".to_string()), - ("servers.beta.role".to_string(), "backend".to_string()), - ( - "servers.character encoding".to_string(), - "value".to_string(), - ), - ( - "strings.my \"string\" test.lines".to_string(), - concat!( - "The first newline is\ntrimmed in raw strings.\n", - "All other whitespace\nis preserved.\n" - ) - .to_string(), - ), - ( - "strings.my \"string\" test.str1".to_string(), - "I'm a string.".to_string(), - ), - ( - "strings.my \"string\" test.str2".to_string(), - "You can \"quote\" me.".to_string(), - ), - ( - "strings.my \"string\" test.str3".to_string(), - "Name\tTabs\nNew Line.".to_string(), - ), - ("env.var1".to_string(), "utils".to_string()), - ("env.var2".to_string(), "utils".to_string()), - ]); + let expected = BTreeMap::from_iter( + [ + ("arrays.colors.0000", "red"), + ("arrays.colors.0001", "yellow"), + ("arrays.colors.0002", "green"), + ("arrays.contributors.0000", "Foo Bar "), + ("arrays.contributors.0001.email", "bazqux@example.com"), + ("arrays.contributors.0001.name", "Baz Qux"), + ("arrays.contributors.0001.url", "https://example.com/bazqux"), + ("arrays.integers.0000", "1"), + ("arrays.integers.0001", "2"), + ("arrays.integers.0002", "3"), + ("arrays.integers2.0000", "1"), + ("arrays.integers2.0001", "2"), + ("arrays.integers2.0002", "3"), + ("arrays.integers3.0000", "4"), + ("arrays.integers3.0001", "5"), + ("arrays.nested_arrays_of_ints.0000.0000", "1"), + ("arrays.nested_arrays_of_ints.0000.0001", "2"), + ("arrays.nested_arrays_of_ints.0001.0000", "3"), + ("arrays.nested_arrays_of_ints.0001.0001", "4"), + ("arrays.nested_arrays_of_ints.0001.0002", "5"), + ("arrays.nested_mixed_array.0000.0000", "1"), + ("arrays.nested_mixed_array.0000.0001", "2"), + ("arrays.nested_mixed_array.0001.0000", "a"), + ("arrays.nested_mixed_array.0001.0001", "b"), + ("arrays.nested_mixed_array.0001.0002", "c"), + ("arrays.numbers.0000", "0.1"), + ("arrays.numbers.0001", "0.2"), + ("arrays.numbers.0002", "0.5"), + ("arrays.numbers.0003", "1"), + ("arrays.numbers.0004", "2"), + ("arrays.numbers.0005", "5"), + ("arrays.string_array.0000", "all"), + ("arrays.string_array.0001", "strings"), + ("arrays.string_array.0002", "are the same"), + ("arrays.string_array.0003", "type"), + ("database.data.0000.0000", "delta"), + ("database.data.0000.0001", "phi"), + ("database.data.0001.0000", "3.14"), + ("database.enabled", "true"), + ("database.ports.0000", "8000"), + ("database.ports.0001", "8001"), + ("database.ports.0002", "8002"), + ("database.temp_targets.case", "72.0"), + ("database.temp_targets.cpu", "79.5"), + ("products.0000.name", "Hammer"), + ("products.0000.sku", "738594937"), + ("products.0002.color", "gray"), + ("products.0002.name", "Nail"), + ("products.0002.sku", "284758393"), + ("servers.127.0.0.1", "value"), + ("servers.alpha.ip", "10.0.0.1"), + ("servers.alpha.role", "frontend"), + ("servers.beta.ip", "10.0.0.2"), + ("servers.beta.role", "backend"), + ("servers.character encoding", "value"), + ( + "strings.my \"string\" test.lines", + concat!( + "The first newline is\ntrimmed in raw strings.\n", + "All other whitespace\nis preserved.\n" + ), + ), + ("strings.my \"string\" test.str1", "I'm a string."), + ("strings.my \"string\" test.str2", "You can \"quote\" me."), + ("strings.my \"string\" test.str3", "Name\tTabs\nNew Line."), + ("env.var1", "utils"), + ("env.var2", "utils"), + ("sets.integer.1", ""), + ("sets.integers.1", ""), + ("sets.integers.2", ""), + ("sets.integers.3", ""), + ("sets.string.red", ""), + ("sets.strings.red", ""), + ("sets.strings.yellow", ""), + ("sets.strings.green", ""), + ] + .map(|(k, v)| (k.to_string(), v.to_string())), + ); if config.keys != expected { for (key, value) in &config.keys { @@ -619,5 +603,20 @@ mod tests { } } } + + assert_eq!( + config.set_values("sets.strings").collect::>(), + vec!["green", "red", "yellow"] + ); + + assert_eq!( + config.sub_keys("sets.strings", "").collect::>(), + vec!["green", "red", "yellow"] + ); + + assert_eq!( + config.sub_keys("sets", ".red").collect::>(), + vec!["string", "strings"] + ); } } diff --git a/crates/utils/src/config/tls.rs b/crates/utils/src/config/tls.rs index a48fb495..ca54f571 100644 --- a/crates/utils/src/config/tls.rs +++ b/crates/utils/src/config/tls.rs @@ -48,7 +48,7 @@ pub static TLS12_VERSION: &[&SupportedProtocolVersion] = &[&TLS12]; impl Config { pub fn parse_certificates(&self) -> super::Result>> { let mut certs = AHashMap::new(); - for cert_id in self.sub_keys("certificate") { + for cert_id in self.sub_keys("certificate", ".cert") { let key_cert = ("certificate", cert_id, "cert"); let key_pk = ("certificate", cert_id, "private-key"); @@ -75,7 +75,7 @@ impl Config { pub fn parse_acmes(&self) -> super::Result>> { let mut acmes = AHashMap::new(); - for acme_id in self.sub_keys("acme") { + for acme_id in self.sub_keys("acme", ".cache") { let directory = self .value(("acme", acme_id, "directory")) .unwrap_or(LETS_ENCRYPT_PRODUCTION_DIRECTORY) @@ -106,7 +106,7 @@ impl Config { // Find which domains are covered by this ACME manager let mut domains = Vec::new(); - for id in self.sub_keys("server.listener") { + for id in self.sub_keys("server.listener", ".protocol") { match ( self.value_or_default(("server.listener", id, "tls.acme"), "server.tls.acme"), self.value_or_default(("server.listener", id, "hostname"), "server.hostname"), diff --git a/crates/utils/src/config/utils.rs b/crates/utils/src/config/utils.rs index c5947a89..4a5db0e7 100644 --- a/crates/utils/src/config/utils.rs +++ b/crates/utils/src/config/utils.rs @@ -75,13 +75,19 @@ impl Config { } } - pub fn sub_keys<'x, 'y: 'x>(&'y self, prefix: impl AsKey) -> impl Iterator + 'x { + pub fn sub_keys<'x, 'y: 'x>( + &'y self, + prefix: impl AsKey, + suffix: &'y str, + ) -> impl Iterator + 'x { let mut last_key = ""; let prefix = prefix.as_prefix(); self.keys.keys().filter_map(move |key| { let key = key.strip_prefix(&prefix)?; - let key = if let Some((key, _)) = key.split_once('.') { + let key = if !suffix.is_empty() { + key.strip_suffix(suffix)? + } else if let Some((key, _)) = key.split_once('.') { key } else { key @@ -95,6 +101,29 @@ impl Config { }) } + pub fn set_values<'x, 'y: 'x>(&'y self, prefix: impl AsKey) -> impl Iterator + 'x { + let prefix = prefix.as_prefix(); + + self.keys + .keys() + .filter_map(move |key| key.strip_prefix(&prefix)) + } + + pub fn set_values_or_default( + &self, + prefix: impl AsKey, + default: impl AsKey, + ) -> impl Iterator { + let mut prefix = prefix.as_prefix(); + + self.set_values(if self.keys.keys().any(|k| k.starts_with(&prefix)) { + prefix.truncate(prefix.len() - 1); + prefix + } else { + default.as_key() + }) + } + pub fn properties( &self, prefix: impl AsKey, @@ -555,7 +584,7 @@ impl ParseValue for Rate { key.as_key() ) })?, - period: period.parse_key(key)?, + period: std::cmp::max(period.parse_key(key)?, Duration::from_secs(1)), }) } else if ["false", "none", "unlimited"].contains(&value) { Ok(Rate::default()) @@ -672,15 +701,15 @@ ip = "a:b::1:1" config.parse(toml).unwrap(); assert_eq!( - config.sub_keys("queues").collect::>(), + config.sub_keys("queues", "").collect::>(), ["a", "x", "z"] ); assert_eq!( - config.sub_keys("servers").collect::>(), + config.sub_keys("servers", "").collect::>(), ["my relay", "submissions"] ); assert_eq!( - config.sub_keys("queues.z.retry").collect::>(), + config.sub_keys("queues.z.retry", "").collect::>(), ["0000", "0001", "0002", "0003", "0004"] ); assert_eq!( diff --git a/crates/utils/src/listener/blocked.rs b/crates/utils/src/listener/blocked.rs new file mode 100644 index 00000000..7e33fdcb --- /dev/null +++ b/crates/utils/src/listener/blocked.rs @@ -0,0 +1,160 @@ +/* + * Copyright (c) 2023 Stalwart Labs Ltd. + * + * This file is part of the Stalwart Mail Server. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * in the LICENSE file at the top-level directory of this distribution. + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * You can be released from the requirements of the AGPLv3 license by + * purchasing a commercial license. Please contact licensing@stalw.art + * for more details. +*/ + +use std::{ + fmt::Debug, + net::IpAddr, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, +}; + +use ahash::{AHashMap, AHashSet}; +use arc_swap::{ArcSwap, ArcSwapOption}; +use parking_lot::{Mutex, RwLock}; + +use crate::config::{ipmask::IpAddrMask, utils::ParseKey, Config, ConfigKey, Rate}; + +use super::limiter::RateLimiter; + +pub struct BlockedIps { + ip_addresses: RwLock>, + ip_networks: ArcSwap>, + has_networks: AtomicBool, + limiters: Mutex>, + limiter_rate: ArcSwapOption, +} + +#[derive(Debug, PartialEq, Eq, Hash)] +enum LimitBy { + IpAddr(IpAddr), + Login(String), +} + +pub const BLOCKED_IP_KEY: &str = "server.security.blocked-networks"; + +impl BlockedIps { + pub fn new() -> Self { + Self { + ip_addresses: RwLock::new(AHashSet::new()), + ip_networks: ArcSwap::new(Arc::new(Vec::new())), + limiters: Mutex::new(Default::default()), + limiter_rate: ArcSwapOption::empty(), + has_networks: AtomicBool::new(false), + } + } + + pub fn reload(&self, config: &Config) -> crate::config::Result<()> { + self.limiter_rate.store( + config + .property::("server.security.fail2ban")? + .map(Arc::new), + ); + self.reload_blocked_ips(config) + } + + pub fn reload_blocked_ips(&self, config: &Config) -> crate::config::Result<()> { + let mut ip_addresses = AHashSet::new(); + let mut ip_networks = Vec::new(); + + for ip in config.set_values(BLOCKED_IP_KEY) { + if ip.contains('/') { + ip_networks.push(ip.parse_key(BLOCKED_IP_KEY)?); + } else { + ip_addresses.insert(ip.parse_key(BLOCKED_IP_KEY)?); + } + } + + self.has_networks + .store(!ip_networks.is_empty(), Ordering::Relaxed); + *self.ip_addresses.write() = ip_addresses; + self.ip_networks.store(Arc::new(ip_networks)); + + Ok(()) + } + + pub fn is_fail2banned(&self, ip: IpAddr, login: String) -> Option { + if let Some(rate) = self.limiter_rate.load().as_ref() { + let is_allowed = self + .limiters + .lock() + .entry(LimitBy::IpAddr(ip)) + .or_insert_with(|| RateLimiter::new(rate)) + .is_allowed(rate) + && self + .limiters + .lock() + .entry(LimitBy::Login(login)) + .or_insert_with(|| RateLimiter::new(rate)) + .is_allowed(rate); + + if !is_allowed { + self.ip_addresses.write().insert(ip); + return Some(ConfigKey { + key: format!("{}.{}", BLOCKED_IP_KEY, ip), + value: String::new(), + }); + } + } + + None + } + + pub fn has_fail2ban(&self) -> bool { + self.limiter_rate.load().is_some() + } + + pub fn cleanup(&self) { + self.limiters + .lock() + .retain(|_, limiter| limiter.is_active()); + } + + pub fn is_blocked(&self, ip: &IpAddr) -> bool { + self.ip_addresses.read().contains(ip) + || (self.has_networks.load(Ordering::Relaxed) + && self + .ip_networks + .load() + .iter() + .any(|network| network.matches(ip))) + } +} + +impl Debug for BlockedIps { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BlockedIps") + .field("ip_addresses", &self.ip_addresses) + .field("ip_networks", &self.ip_networks) + .field("limiters", &self.limiters) + .field("limiter_rate", &self.limiter_rate) + .finish() + } +} + +impl Default for BlockedIps { + fn default() -> Self { + Self::new() + } +} diff --git a/crates/utils/src/listener/limiter.rs b/crates/utils/src/listener/limiter.rs index e21689aa..592d1c56 100644 --- a/crates/utils/src/listener/limiter.rs +++ b/crates/utils/src/listener/limiter.rs @@ -26,15 +26,15 @@ use std::{ atomic::{AtomicU64, Ordering}, Arc, }, - time::{Duration, Instant}, + time::SystemTime, }; +use crate::config::Rate; + #[derive(Debug)] pub struct RateLimiter { - pub max_requests: u64, - pub max_interval: Duration, - last_refill: Instant, - tokens: u64, + next_refill: AtomicU64, + used_tokens: AtomicU64, } #[derive(Debug, Clone)] @@ -55,53 +55,43 @@ impl Drop for InFlight { } impl RateLimiter { - pub fn new(max_requests: u64, max_interval: Duration) -> Self { + pub fn new(rate: &Rate) -> Self { RateLimiter { - max_requests, - max_interval, - last_refill: Instant::now(), - tokens: max_requests, + next_refill: (now() + rate.period.as_secs()).into(), + used_tokens: 0.into(), } } - pub fn is_allowed(&mut self) -> bool { + pub fn is_allowed(&self, rate: &Rate) -> bool { // Check rate limit - if self.last_refill.elapsed() >= self.max_interval { - self.last_refill = Instant::now(); - self.tokens = self.max_requests; - } - - if self.tokens >= 1 { - self.tokens -= 1; + if self.used_tokens.fetch_add(1, Ordering::Relaxed) < rate.requests { true } else { - false + let now = now(); + if self.next_refill.load(Ordering::Relaxed) <= now { + self.next_refill + .store(now + rate.period.as_secs(), Ordering::Relaxed); + self.used_tokens.store(1, Ordering::Relaxed); + true + } else { + false + } } } - pub fn is_allowed_soft(&self) -> bool { - self.tokens >= 1 || self.last_refill.elapsed() >= self.max_interval + pub fn is_allowed_soft(&self, rate: &Rate) -> bool { + self.used_tokens.load(Ordering::Relaxed) < rate.requests + || self.next_refill.load(Ordering::Relaxed) <= now() } - pub fn retry_at(&self) -> Instant { - Instant::now() - + (self - .max_interval - .checked_sub(self.last_refill.elapsed()) - .unwrap_or_default()) - } - - pub fn elapsed(&self) -> Duration { - self.last_refill.elapsed() - } - - pub fn reset(&mut self) { - self.last_refill = Instant::now(); - self.tokens = self.max_requests; + pub fn secs_to_refill(&self) -> u64 { + self.next_refill + .load(Ordering::Relaxed) + .saturating_sub(now()) } pub fn is_active(&self) -> bool { - self.tokens < self.max_requests || self.last_refill.elapsed() < self.max_interval + self.next_refill.load(Ordering::Relaxed) > now() } } @@ -139,3 +129,10 @@ impl InFlight { self.concurrent.load(Ordering::Relaxed) } } + +fn now() -> u64 { + SystemTime::UNIX_EPOCH + .elapsed() + .unwrap_or_default() + .as_secs() +} diff --git a/crates/utils/src/listener/listen.rs b/crates/utils/src/listener/listen.rs index 6651d782..b140ae6a 100644 --- a/crates/utils/src/listener/listen.rs +++ b/crates/utils/src/listener/listen.rs @@ -63,6 +63,7 @@ impl Server { hostname: self.hostname, acceptor: self.acceptor, proxy_networks: self.proxy_networks, + blocked_ips: self.blocked_ips, limiter: ConcurrencyLimiter::new(self.max_connections), shutdown_rx, }); @@ -189,8 +190,20 @@ impl BuildSession for Arc { }; let remote_port = remote_addr.port(); - // Enforce concurrency - if let Some(in_flight) = self.limiter.is_allowed() { + // Check if blocked + if self.blocked_ips.is_blocked(&remote_ip) { + tracing::debug!( + context = "listener", + event = "blocked", + instance = self.id, + protocol = ?self.protocol, + remote.ip = remote_ip.to_string(), + remote.port = remote_port, + "Dropping connection from blocked IP." + ); + None + } else if let Some(in_flight) = self.limiter.is_allowed() { + // Enforce concurrency SessionData { stream, in_flight, diff --git a/crates/utils/src/listener/mod.rs b/crates/utils/src/listener/mod.rs index 6522b952..d8e52ac2 100644 --- a/crates/utils/src/listener/mod.rs +++ b/crates/utils/src/listener/mod.rs @@ -35,8 +35,12 @@ use tokio::{ }; use tokio_rustls::{Accept, TlsAcceptor}; -use self::limiter::{ConcurrencyLimiter, InFlight}; +use self::{ + blocked::BlockedIps, + limiter::{ConcurrencyLimiter, InFlight}, +}; +pub mod blocked; pub mod limiter; pub mod listen; pub mod stream; @@ -51,6 +55,7 @@ pub struct ServerInstance { pub acceptor: TcpAcceptor, pub limiter: ConcurrencyLimiter, pub proxy_networks: Vec, + pub blocked_ips: Arc, pub shutdown_rx: watch::Receiver, } diff --git a/resources/config/common/server.toml b/resources/config/common/server.toml index 273c82de..4de0ce0a 100644 --- a/resources/config/common/server.toml +++ b/resources/config/common/server.toml @@ -5,7 +5,13 @@ [server] hostname = "%{HOST}%" max-connections = 8192 -#proxy-trusted-networks = ["127.0.0.0/8", "::1", "10.0.0.0/8"] + +#[server.proxy] +#trusted-networks = {"127.0.0.0/8", "::1", "10.0.0.0/8"} + +[server.security] +blocked-networks = {} +fail2ban = "100/1d" [server.run-as] user = "stalwart-mail" @@ -25,5 +31,3 @@ backlog = 1024 [global] shared-map = {shard = 32, capacity = 10} #thread-pool = 8 - - diff --git a/resources/config/common/sieve.toml b/resources/config/common/sieve.toml index 322c78f5..d41cee31 100644 --- a/resources/config/common/sieve.toml +++ b/resources/config/common/sieve.toml @@ -47,10 +47,6 @@ return-path = "" no-capability-check = true sign = ["rsa"] -[sieve.trusted.default] -directory = "%{DEFAULT_DIRECTORY}%" -store = "%{DEFAULT_STORE}%" - [sieve.trusted.limits] redirects = 3 out-messages = 5 diff --git a/resources/config/common/store.toml b/resources/config/common/store.toml new file mode 100644 index 00000000..6924769f --- /dev/null +++ b/resources/config/common/store.toml @@ -0,0 +1,23 @@ +############################################# +# Storage configuration +############################################# + +[storage] +data = "%{DEFAULT_STORE}%" +fts = "%{DEFAULT_STORE}%" +blob = "%{DEFAULT_STORE}%" +lookup = "%{DEFAULT_STORE}%" +directory = "%{DEFAULT_DIRECTORY}%" + +[storage.encryption] +enable = true +append = false + +[storage.spam] +header = "X-Spam-Status: Yes" + +[storage.fts] +default-language = "en" + +[storage.cluster] +node-id = 1 diff --git a/resources/config/config.toml b/resources/config/config.toml index b79c9f0c..a2b14cd5 100644 --- a/resources/config/config.toml +++ b/resources/config/config.toml @@ -12,6 +12,7 @@ default_store = "__STORE__" [include] files = [ "%{BASE_PATH}%/etc/common/server.toml", "%{BASE_PATH}%/etc/common/tls.toml", + "%{BASE_PATH}%/etc/common/store.toml", "%{BASE_PATH}%/etc/common/tracing.toml", "%{BASE_PATH}%/etc/common/sieve.toml", "%{BASE_PATH}%/etc/directory/imap.toml", @@ -37,7 +38,6 @@ files = [ "%{BASE_PATH}%/etc/common/server.toml", "%{BASE_PATH}%/etc/jmap/protocol.toml", "%{BASE_PATH}%/etc/jmap/push.toml", "%{BASE_PATH}%/etc/jmap/ratelimit.toml", - "%{BASE_PATH}%/etc/jmap/store.toml", "%{BASE_PATH}%/etc/jmap/websockets.toml", "%{BASE_PATH}%/etc/smtp/auth.toml", "%{BASE_PATH}%/etc/smtp/listener.toml", diff --git a/resources/config/jmap/auth.toml b/resources/config/jmap/auth.toml index 6640ea42..a222d5e6 100644 --- a/resources/config/jmap/auth.toml +++ b/resources/config/jmap/auth.toml @@ -2,9 +2,6 @@ # JMAP authentication & session configuration ############################################# -[jmap] -directory = "%{DEFAULT_DIRECTORY}%" - [jmap.session.cache] ttl = "1h" size = 100 diff --git a/resources/config/jmap/store.toml b/resources/config/jmap/store.toml deleted file mode 100644 index f639bccd..00000000 --- a/resources/config/jmap/store.toml +++ /dev/null @@ -1,21 +0,0 @@ -############################################# -# JMAP server store configuration -############################################# - -[jmap.store] -data = "%{DEFAULT_STORE}%" -fts = "__FTS_STORE__" -blob = "__BLOB_STORE__" - -[jmap.encryption] -enable = true -append = false - -[jmap.spam] -header = "X-Spam-Status: Yes" - -[jmap.fts] -default-language = "en" - -[jmap.cluster] -node-id = 1 diff --git a/resources/config/smtp/listener.toml b/resources/config/smtp/listener.toml index c646c83b..fc4aa45d 100644 --- a/resources/config/smtp/listener.toml +++ b/resources/config/smtp/listener.toml @@ -19,6 +19,3 @@ tls.implicit = true [server.listener."management"] bind = ["127.0.0.1:8080"] protocol = "http" - -[management] -directory = "%{DEFAULT_DIRECTORY}%" diff --git a/tests/resources/smtp/config/toml-parser.toml b/tests/resources/smtp/config/toml-parser.toml index 6b6662f8..c0a536fb 100644 --- a/tests/resources/smtp/config/toml-parser.toml +++ b/tests/resources/smtp/config/toml-parser.toml @@ -38,6 +38,12 @@ All other whitespace is preserved. ''' +[sets] +integer = { 1 } +integers = { 1, 2, 3 } +string = { "red" } +strings = { "red", "yellow", "green" } + [arrays] integers = [ 1, 2, 3 ] colors = [ "red", "yellow", "green" ] diff --git a/tests/src/directory/mod.rs b/tests/src/directory/mod.rs index 36a50114..c4280b1d 100644 --- a/tests/src/directory/mod.rs +++ b/tests/src/directory/mod.rs @@ -39,6 +39,7 @@ use rustls_pki_types::PrivateKeyDer; use std::{borrow::Cow, io::BufReader, path::PathBuf, sync::Arc}; use store::{config::ConfigStore, LookupStore, Store, Stores}; use tokio_rustls::TlsAcceptor; +use utils::config::Servers; use crate::store::TempDir; @@ -312,7 +313,16 @@ impl DirectoryTest { let stores = config.parse_stores().await.unwrap(); DirectoryTest { - directories: config.parse_directory(&stores, id_store).await.unwrap(), + directories: config + .parse_directory( + &stores, + &Servers::default(), + id_store + .map(|id| stores.stores.get(id).unwrap().clone()) + .unwrap_or_default(), + ) + .await + .unwrap(), stores, temp_dir, } diff --git a/tests/src/imap/mod.rs b/tests/src/imap/mod.rs index c729ad23..a0d7254a 100644 --- a/tests/src/imap/mod.rs +++ b/tests/src/imap/mod.rs @@ -174,13 +174,11 @@ private-key = "file://{PK}" [imap.protocol] uidplus = true -[jmap] -directory = "auth" - -[jmap.store] +[storage] data = "{STORE}" fts = "{STORE}" blob = "{STORE}" +directory = "auth" [jmap.protocol] set.max-objects = 100000 @@ -279,7 +277,11 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { let mut servers = config.parse_servers().unwrap(); let stores = config.parse_stores().await.failed("Invalid configuration"); let directory = config - .parse_directory(&stores, store_id.into()) + .parse_directory( + &stores, + &servers, + stores.stores.get(store_id).unwrap().clone(), + ) .await .unwrap(); @@ -293,7 +295,7 @@ async fn init_imap_tests(store_id: &str, delete_if_exists: bool) -> IMAPTest { &config, &stores, &directory, - std::mem::take(&mut servers.certificates), + &mut servers, delivery_rx, smtp.clone(), ) @@ -486,6 +488,19 @@ impl ImapConnection { } } + pub async fn assert_disconnect(&mut self) { + match tokio::time::timeout(Duration::from_millis(1500), self.reader.next_line()).await { + Ok(Ok(None)) => {} + Ok(Ok(Some(line))) => { + panic!("Expected connection to be closed, but got {:?}", line); + } + Ok(Err(err)) => { + panic!("Connection broken: {:?}", err); + } + Err(_) => panic!("Timeout while waiting for server response."), + } + } + pub async fn read(&mut self, t: Type) -> Vec { let mut lines = Vec::new(); loop { diff --git a/tests/src/jmap/auth_limits.rs b/tests/src/jmap/auth_limits.rs index b1775af5..e61d6b05 100644 --- a/tests/src/jmap/auth_limits.rs +++ b/tests/src/jmap/auth_limits.rs @@ -24,14 +24,20 @@ use std::{sync::Arc, time::Duration}; use directory::backend::internal::manage::ManageDirectory; +use imap_proto::ResponseType; +use jmap::services::housekeeper::Event; use jmap_client::{ client::{Client, Credentials}, core::set::{SetError, SetErrorType}, mailbox::{self}, }; use jmap_proto::types::id::Id; +use utils::listener::blocked::BLOCKED_IP_KEY; -use crate::jmap::{assert_is_empty, mailbox::destroy_all_mailboxes}; +use crate::{ + imap::{ImapConnection, Type}, + jmap::{assert_is_empty, mailbox::destroy_all_mailboxes}, +}; use super::JMAPTest; @@ -104,6 +110,52 @@ pub async fn test(params: &mut JMAPTest) { // Limit should be restored after 1 second tokio::time::sleep(Duration::from_millis(1500)).await; + // Test fail2ban + assert_eq!( + server + .store + .config_get(format!("{BLOCKED_IP_KEY}.127.0.0.1")) + .await + .unwrap(), + None + ); + let mut imap = ImapConnection::connect(b"_x ").await; + imap.send("AUTHENTICATE PLAIN AGpvaG4AY2hpbWljaGFuZ2Fz") + .await; + imap.assert_read(Type::Tagged, ResponseType::No).await; + + // There are already 100 failed login attempts for this IP address + // so the next one should be rejected, even if done over IMAP + imap.send("AUTHENTICATE PLAIN AGpvaG4AY2hpbWljaGFuZ2Fz") + .await; + imap.assert_disconnect().await; + + // Make sure the IP address is blocked + assert_eq!( + server + .store + .config_get(format!("{BLOCKED_IP_KEY}.127.0.0.1")) + .await + .unwrap(), + Some(String::new()) + ); + ImapConnection::connect(b"_y ") + .await + .assert_disconnect() + .await; + + // Lift ban + server + .store + .config_clear(format!("{BLOCKED_IP_KEY}.127.0.0.1")) + .await + .unwrap(); + server + .housekeeper_tx + .send(Event::ReloadConfig) + .await + .unwrap(); + // Valid authentication requests should not be rate limited for _ in 0..110 { Client::new() diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index fd0b9201..b356dbf8 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -25,6 +25,7 @@ use std::{sync::Arc, time::Duration}; use base64::{engine::general_purpose, Engine}; use directory::core::config::ConfigDirectory; +use imap::core::{ImapSessionManager, IMAP}; use jmap::{ api::JmapSessionManager, services::{housekeeper::Event, IPC_CHANNEL_BUFFER}, @@ -77,6 +78,11 @@ protocol = "jmap" max-connections = 81920 tls.implicit = true +[server.listener.imap] +bind = ["127.0.0.1:9991"] +protocol = "imap" +max-connections = 81920 + [server.listener.lmtp-debug] bind = ['127.0.0.1:11200'] greeting = 'Test LMTP instance' @@ -91,6 +97,10 @@ enable = true implicit = false certificate = "default" +[server.security] +blocked-networks = {} +fail2ban = "101/5s" + [session.ehlo] reject-non-fqdn = false @@ -171,15 +181,13 @@ disable = true # Elastic is disabled by default cert = "file://{CERT}" private-key = "file://{PK}" -[jmap] -directory = "auth" - -[jmap.store] +[storage] data = "{STORE}" fts = "{STORE}" blob = "{STORE}" +directory = "auth" -[jmap.spam] +[storage.spam] header = "X-Spam-Status: Yes" [jmap.protocol.get] @@ -394,9 +402,14 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { let mut servers = config.parse_servers().unwrap(); let stores = config.parse_stores().await.failed("Invalid configuration"); let directory = config - .parse_directory(&stores, store_id.into()) + .parse_directory( + &stores, + &servers, + stores.stores.get(store_id).unwrap().clone(), + ) .await .unwrap(); + servers.blocked_ips.reload(&config).unwrap(); // Start JMAP and SMTP servers servers.bind(&config); @@ -408,12 +421,16 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { &config, &stores, &directory, - std::mem::take(&mut servers.certificates), + &mut servers, delivery_rx, smtp.clone(), ) .await .failed("Invalid configuration file"); + let imap: Arc = IMAP::init(&config) + .await + .failed("Invalid configuration file"); + let (shutdown_tx, _) = servers.spawn(|server, shutdown_rx| { match &server.protocol { ServerProtocol::Smtp | ServerProtocol::Lmtp => { @@ -422,6 +439,10 @@ async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { ServerProtocol::Jmap => { server.spawn(JmapSessionManager::new(jmap.clone()), shutdown_rx) } + ServerProtocol::Imap => server.spawn( + ImapSessionManager::new(jmap.clone(), imap.clone()), + shutdown_rx, + ), _ => unreachable!(), }; }); diff --git a/tests/src/smtp/config.rs b/tests/src/smtp/config.rs index aa2875f3..a2278c4b 100644 --- a/tests/src/smtp/config.rs +++ b/tests/src/smtp/config.rs @@ -458,6 +458,7 @@ fn parse_servers() { tls_implicit: false, max_connections: 8192, proxy_networks: vec![], + blocked_ips: Arc::new(Default::default()), }, Server { id: "smtps".to_string(), @@ -487,6 +488,7 @@ fn parse_servers() { tls_implicit: true, max_connections: 1024, proxy_networks: vec![], + blocked_ips: Arc::new(Default::default()), }, Server { id: "submission".to_string(), @@ -506,6 +508,7 @@ fn parse_servers() { tls_implicit: true, max_connections: 8192, proxy_networks: vec![], + blocked_ips: Arc::new(Default::default()), }, ]; @@ -622,7 +625,7 @@ async fn eval_dynvalue() { let envelope = TestEnvelope::from_config(&config); - for test_name in config.sub_keys("eval") { + for test_name in config.sub_keys("eval", "") { //println!("============= Testing {:?} ==================", key); let if_block = config .parse_if_block::>>( @@ -664,7 +667,7 @@ async fn eval_dynvalue() { .map(|(k, v)| (k.clone(), Arc::new(v.clone()))) .collect::>(); - for test_name in config.sub_keys("maybe-eval") { + for test_name in config.sub_keys("maybe-eval", "") { //println!("============= Testing {:?} ==================", key); let if_block = config .parse_if_block::>>( diff --git a/tests/src/smtp/inbound/auth.rs b/tests/src/smtp/inbound/auth.rs index 13d0eb1f..16c42ee5 100644 --- a/tests/src/smtp/inbound/auth.rs +++ b/tests/src/smtp/inbound/auth.rs @@ -23,8 +23,8 @@ use directory::core::config::ConfigDirectory; use smtp_proto::{AUTH_LOGIN, AUTH_PLAIN}; -use store::Stores; -use utils::config::{Config, DynValue}; +use store::{Store, Stores}; +use utils::config::{Config, DynValue, Servers}; use crate::smtp::{ session::{TestSession, VerifyResponse}, @@ -62,7 +62,7 @@ async fn auth() { let mut ctx = ConfigContext::new(&[]); ctx.directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); diff --git a/tests/src/smtp/inbound/data.rs b/tests/src/smtp/inbound/data.rs index a42f78ec..77ed1cad 100644 --- a/tests/src/smtp/inbound/data.rs +++ b/tests/src/smtp/inbound/data.rs @@ -22,8 +22,8 @@ */ use directory::core::config::ConfigDirectory; -use store::Stores; -use utils::config::Config; +use store::{Store, Stores}; +use utils::config::{Config, Servers}; use crate::smtp::{ inbound::{TestMessage, TestQueueEvent}, @@ -80,7 +80,7 @@ async fn data() { let mut qr = core.init_test_queue("smtp_data_test"); let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); let config = &mut core.session.config.rcpt; diff --git a/tests/src/smtp/inbound/dmarc.rs b/tests/src/smtp/inbound/dmarc.rs index 3a6a8bde..7469f97f 100644 --- a/tests/src/smtp/inbound/dmarc.rs +++ b/tests/src/smtp/inbound/dmarc.rs @@ -34,8 +34,8 @@ use mail_auth::{ report::DmarcResult, spf::Spf, }; -use store::Stores; -use utils::config::{Config, DynValue, Rate}; +use store::{Store, Stores}; +use utils::config::{Config, DynValue, Rate, Servers}; use crate::smtp::{ inbound::{sign::TextConfigContext, TestMessage, TestQueueEvent, TestReportingEvent}, @@ -135,7 +135,7 @@ async fn dmarc() { let mut rr = core.init_test_report(); let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); let config = &mut core.session.config.rcpt; diff --git a/tests/src/smtp/inbound/rcpt.rs b/tests/src/smtp/inbound/rcpt.rs index f9910835..e14c228c 100644 --- a/tests/src/smtp/inbound/rcpt.rs +++ b/tests/src/smtp/inbound/rcpt.rs @@ -25,8 +25,8 @@ use std::time::Duration; use directory::core::config::ConfigDirectory; use smtp_proto::{RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE, RCPT_NOTIFY_SUCCESS}; -use store::Stores; -use utils::config::Config; +use store::{Store, Stores}; +use utils::config::{Config, Servers}; use crate::smtp::{ session::{TestSession, VerifyResponse}, @@ -74,7 +74,7 @@ async fn rcpt() { let config_ext = &mut core.session.config.extensions; let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); let config = &mut core.session.config.rcpt; diff --git a/tests/src/smtp/inbound/rewrite.rs b/tests/src/smtp/inbound/rewrite.rs index 074b4260..c1cfe9a9 100644 --- a/tests/src/smtp/inbound/rewrite.rs +++ b/tests/src/smtp/inbound/rewrite.rs @@ -27,8 +27,8 @@ use smtp::{ config::{if_block::ConfigIf, scripts::ConfigSieve, ConfigContext, EnvelopeKey, IfBlock}, core::{Session, SMTP}, }; -use store::Stores; -use utils::config::{Config, DynValue}; +use store::{Store, Stores}; +use utils::config::{Config, DynValue, Servers}; const CONFIG: &str = r#" [session.mail] @@ -105,7 +105,7 @@ async fn address_rewrite() { let mut ctx = ConfigContext::new(&[]).parse_signatures(); let settings = Config::new(CONFIG).unwrap(); ctx.directory = settings - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); core.sieve = settings.parse_sieve(&mut ctx).unwrap(); diff --git a/tests/src/smtp/inbound/scripts.rs b/tests/src/smtp/inbound/scripts.rs index 5bb99c2a..af09e9bd 100644 --- a/tests/src/smtp/inbound/scripts.rs +++ b/tests/src/smtp/inbound/scripts.rs @@ -35,9 +35,9 @@ use smtp::{ core::{Session, SMTP}, scripts::ScriptResult, }; -use store::config::ConfigStore; +use store::{config::ConfigStore, Store}; use tokio::runtime::Handle; -use utils::config::Config; +use utils::config::{Config, Servers}; const CONFIG: &str = r#" [store."sql"] @@ -133,7 +133,10 @@ async fn sieve_scripts() { ) .unwrap(); ctx.stores = config.parse_stores().await.unwrap(); - ctx.directory = config.parse_directory(&ctx.stores, None).await.unwrap(); + ctx.directory = config + .parse_directory(&ctx.stores, &Servers::default(), Store::default()) + .await + .unwrap(); let pipes = config.parse_pipes(&ctx, &[EnvelopeKey::RemoteIp]).unwrap(); core.sieve = config.parse_sieve(&mut ctx).unwrap(); let config = &mut core.session.config; diff --git a/tests/src/smtp/inbound/sign.rs b/tests/src/smtp/inbound/sign.rs index 83adc566..81588751 100644 --- a/tests/src/smtp/inbound/sign.rs +++ b/tests/src/smtp/inbound/sign.rs @@ -28,8 +28,8 @@ use mail_auth::{ common::{parse::TxtRecordParser, verify::DomainKey}, spf::Spf, }; -use store::Stores; -use utils::config::{Config, DynValue}; +use store::{Store, Stores}; +use utils::config::{Config, DynValue, Servers}; use crate::smtp::{ inbound::{TestMessage, TestQueueEvent}, @@ -154,7 +154,7 @@ async fn sign_and_seal() { let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); let config = &mut core.session.config.rcpt; diff --git a/tests/src/smtp/inbound/vrfy.rs b/tests/src/smtp/inbound/vrfy.rs index 6249f0ef..5f0368a0 100644 --- a/tests/src/smtp/inbound/vrfy.rs +++ b/tests/src/smtp/inbound/vrfy.rs @@ -22,8 +22,8 @@ */ use directory::core::config::ConfigDirectory; -use store::Stores; -use utils::config::Config; +use store::{Store, Stores}; +use utils::config::{Config, Servers}; use crate::smtp::{ session::{TestSession, VerifyResponse}, @@ -68,7 +68,7 @@ async fn vrfy_expn() { let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); let config = &mut core.session.config.rcpt; diff --git a/tests/src/smtp/lookup/sql.rs b/tests/src/smtp/lookup/sql.rs index 5591ab0f..096ed17b 100644 --- a/tests/src/smtp/lookup/sql.rs +++ b/tests/src/smtp/lookup/sql.rs @@ -25,8 +25,8 @@ use std::time::Duration; use directory::core::config::ConfigDirectory; use smtp_proto::{AUTH_LOGIN, AUTH_PLAIN}; -use store::config::ConfigStore; -use utils::config::{Config, DynValue}; +use store::{config::ConfigStore, Store}; +use utils::config::{Config, DynValue, Servers}; use crate::{ directory::DirectoryStore, @@ -87,7 +87,10 @@ async fn lookup_sql() { let mut ctx = ConfigContext::new(&[]); let config = Config::new(&config_file).unwrap(); ctx.stores = config.parse_stores().await.unwrap(); - ctx.directory = config.parse_directory(&ctx.stores, None).await.unwrap(); + ctx.directory = config + .parse_directory(&ctx.stores, &Servers::default(), Store::default()) + .await + .unwrap(); // Obtain directory handle let handle = DirectoryStore { diff --git a/tests/src/smtp/management/queue.rs b/tests/src/smtp/management/queue.rs index 15526dbf..fb580507 100644 --- a/tests/src/smtp/management/queue.rs +++ b/tests/src/smtp/management/queue.rs @@ -31,8 +31,8 @@ use directory::core::config::ConfigDirectory; use mail_auth::MX; use mail_parser::DateTime; use reqwest::{header::AUTHORIZATION, StatusCode}; -use store::Stores; -use utils::config::{Config, ServerProtocol}; +use store::{Store, Stores}; +use utils::config::{Config, ServerProtocol, Servers}; use crate::smtp::{ inbound::TestQueueEvent, management::send_manage_request, outbound::start_test_server, @@ -96,10 +96,10 @@ async fn manage_queue() { // Start local management interface let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); - core.queue.config.management_lookup = directory.directories.get("local").unwrap().clone(); + core.queue.config.directory = directory.directories.get("local").unwrap().clone(); core.session.config.rcpt.relay = IfBlock::new(true); core.session.config.rcpt.max_recipients = IfBlock::new(100); core.session.config.extensions.future_release = IfBlock::new(Some(Duration::from_secs(86400))); diff --git a/tests/src/smtp/management/report.rs b/tests/src/smtp/management/report.rs index 0cd11d8b..a85b7eb1 100644 --- a/tests/src/smtp/management/report.rs +++ b/tests/src/smtp/management/report.rs @@ -34,9 +34,9 @@ use mail_auth::{ ActionDisposition, DmarcResult, Record, }, }; -use store::Stores; +use store::{Store, Stores}; use tokio::sync::mpsc; -use utils::config::{Config, ServerProtocol}; +use utils::config::{Config, ServerProtocol, Servers}; use crate::smtp::{ make_temp_dir, management::send_manage_request, outbound::start_test_server, TestConfig, @@ -83,10 +83,10 @@ async fn manage_reports() { config.tls.max_size = IfBlock::new(1024); let directory = Config::new(DIRECTORY) .unwrap() - .parse_directory(&Stores::default(), None) + .parse_directory(&Stores::default(), &Servers::default(), Store::default()) .await .unwrap(); - core.queue.config.management_lookup = directory.directories.get("local").unwrap().clone(); + core.queue.config.directory = directory.directories.get("local").unwrap().clone(); let (report_tx, report_rx) = mpsc::channel(1024); core.report.tx = report_tx; let core = Arc::new(core); diff --git a/tests/src/smtp/mod.rs b/tests/src/smtp/mod.rs index c15bf085..2600c891 100644 --- a/tests/src/smtp/mod.rs +++ b/tests/src/smtp/mod.rs @@ -25,7 +25,7 @@ use std::{path::PathBuf, sync::Arc, time::Duration}; use ahash::AHashMap; use dashmap::DashMap; -use directory::Directory; +use directory::{AddressMapping, Directory, DirectoryInner}; use mail_auth::{ common::lru::{DnsCache, LruCache}, hickory_resolver::config::{ResolverConfig, ResolverOpts}, @@ -34,6 +34,7 @@ use mail_auth::{ use mail_send::smtp::tls::build_tls_connector; use sieve::Runtime; use smtp_proto::{AUTH_LOGIN, AUTH_PLAIN}; +use store::{LookupStore, Store}; use tokio::sync::mpsc; use smtp::{ @@ -297,6 +298,7 @@ impl TestConfig for QueueCore { impl TestConfig for QueueConfig { fn test() -> Self { + let store = Store::default(); Self { path: Default::default(), hash: IfBlock::new(10), @@ -343,7 +345,15 @@ impl TestConfig for QueueConfig { rcpt: vec![], rcpt_domain: vec![], }, - management_lookup: Arc::new(Directory::default()), + directory: Arc::new(Directory { + store: DirectoryInner::Internal(store.clone()), + catch_all: AddressMapping::Disable, + subaddressing: AddressMapping::Disable, + cache: None, + blocked_ips: Arc::new(Default::default()), + }), + lookup_store: LookupStore::Store(store.clone()), + data_store: store, } } } @@ -440,8 +450,6 @@ impl TestConfig for SieveCore { sign: vec![], directories: Default::default(), lookup_stores: Default::default(), - default_lookup_store: None, - default_directory: None, } } } diff --git a/tests/src/smtp/session.rs b/tests/src/smtp/session.rs index d656492d..17a19a6c 100644 --- a/tests/src/smtp/session.rs +++ b/tests/src/smtp/session.rs @@ -372,6 +372,7 @@ impl TestServerInstance for ServerInstance { limiter: ConcurrencyLimiter::new(100), shutdown_rx, proxy_networks: vec![], + blocked_ips: Arc::new(Default::default()), } } }