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

Settings hot reloading - Part 1

Browse Source
This commit is contained in:
mdecimus
2024-03-21 20:07:21 +01:00
parent 7e1a95c1ee
commit 333a0d5a1b
115 changed files with 8814 additions and 1651 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
crates/utils/Cargo.toml
View File

@@ -31,7 +31,7 @@ ring = { version = "0.17" }
base64 = "0.21"
serde_json = "1.0"
rcgen = "0.12"
reqwest = { version = "0.11", default-features = false, features = ["rustls-tls-webpki-roots"]}
reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-webpki-roots"]}
x509-parser = "0.16.0"
pem = "3.0"
parking_lot = "0.12"

18
crates/utils/src/config/ipmask.rs
View File

@@ -23,7 +23,7 @@
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use super::utils::{AsKey, ParseValue};
use super::utils::{AsKey, ParseKey, ParseValue};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IpAddrMask {
@@ -31,6 +31,12 @@ pub enum IpAddrMask {
V6 { addr: Ipv6Addr, mask: u128 },
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IpAddrOrMask {
Ip(IpAddr),
Mask(IpAddrMask),
}
impl IpAddrMask {
pub fn matches(&self, remote: &IpAddr) -> bool {
match self {
@@ -130,6 +136,16 @@ impl ParseValue for IpAddrMask {
}
}
impl ParseValue for IpAddrOrMask {
fn parse_value(key: impl AsKey, ip: &str) -> super::Result<Self> {
if ip.contains('/') {
ip.parse_key(key).map(IpAddrOrMask::Mask)
} else {
ip.parse_key(key).map(IpAddrOrMask::Ip)
}
}
}
#[cfg(test)]
mod tests {
use super::*;

45
crates/utils/src/config/listener.rs
View File

@@ -135,7 +135,7 @@ impl Config {
"recv-buffer-size",
"tos",
] {
if let Some(value) = self.value_or_default(
if let Some(value) = self.value_or_else(
("server.listener", id, "socket", option),
("server.socket", option),
) {
@@ -158,20 +158,18 @@ impl Config {
listeners.push(Listener {
socket,
addr,
ttl: self.property_or_default(
("server.listener", id, "socket.ttl"),
"server.socket.ttl",
)?,
backlog: self.property_or_default(
ttl: self
.property_or_else(("server.listener", id, "socket.ttl"), "server.socket.ttl")?,
backlog: self.property_or_else(
("server.listener", id, "socket.backlog"),
"server.socket.backlog",
)?,
linger: self.property_or_default(
linger: self.property_or_else(
("server.listener", id, "socket.linger"),
"server.socket.linger",
)?,
nodelay: self
.property_or_default(
.property_or_else(
("server.listener", id, "socket.nodelay"),
"server.socket.nodelay",
)?
@@ -185,13 +183,13 @@ impl Config {
// Build TLS config
let (acceptor, tls_implicit) = if self
.property_or_default(("server.listener", id, "tls.enable"), "server.tls.enable")?
.property_or_else(("server.listener", id, "tls.enable"), "server.tls.enable")?
.unwrap_or(false)
{
// Parse protocol versions
let mut tls_v2 = false;
let mut tls_v3 = false;
for (key, protocol) in self.values_or_default(
for (key, protocol) in self.values_or_else(
("server.listener", id, "tls.protocols"),
"server.tls.protocols",
) {
@@ -209,7 +207,7 @@ impl Config {
// Parse cipher suites
let mut ciphers: Vec<SupportedCipherSuite> = Vec::new();
for (key, protocol) in
self.values_or_default(("server.listener", id, "tls.ciphers"), "server.tls.ciphers")
self.values_or_else(("server.listener", id, "tls.ciphers"), "server.tls.ciphers")
{
ciphers.push(protocol.parse_key(key)?);
}
@@ -217,14 +215,15 @@ impl Config {
// Build resolver
let mut acme_acceptor = None;
let resolver: Arc<dyn ResolvesServerCert> = if let Some(acme_id) =
self.value_or_default(("server.listener", id, "tls.acme"), "server.tls.acme")
self.value_or_else(("server.listener", id, "tls.acme"), "server.tls.acme")
{
let acme = acmes.get(acme_id).ok_or_else(|| {
format!("Undefined ACME id {acme_id:?} for listener {id:?}.",)
})?;
// Check if this port is used to receive ACME challenges
let acme_port = self.property_or_static::<u16>(("acme", acme_id, "port"), "443")?;
let acme_port =
self.property_or_default::<u16>(("acme", acme_id, "port"), "443")?;
if listeners.iter().any(|l| l.addr.port() == acme_port) {
acme_acceptor = Some(acme.clone());
}
@@ -232,7 +231,7 @@ impl Config {
acme.clone()
} else {
let cert_id = self
.value_or_default(
.value_or_else(
("server.listener", id, "tls.certificate"),
"server.tls.certificate",
)
@@ -249,7 +248,7 @@ impl Config {
// Add SNI certificates
for (key, value) in
self.values_or_default(("server.listener", id, "tls.sni"), "server.tls.sni")
self.values_or_else(("server.listener", id, "tls.sni"), "server.tls.sni")
{
if let Some(prefix) = key.strip_suffix(".subject") {
resolver
@@ -294,7 +293,7 @@ impl Config {
.with_no_client_auth()
.with_cert_resolver(resolver.clone());
config.ignore_client_order = self
.property_or_default(
.property_or_else(
("server.listener", id, "tls.ignore-client-order"),
"server.tls.ignore-client-order",
)?
@@ -317,7 +316,7 @@ impl Config {
(
acceptor,
self.property_or_default(
self.property_or_else(
("server.listener", id, "tls.implicit"),
"server.tls.implicit",
)?
@@ -331,7 +330,7 @@ impl Config {
// Parse proxy networks
let mut proxy_networks = Vec::new();
for network in self.set_values_or_default(
for network in self.set_values_or_else(
("server.listener", id, "proxy.trusted-networks"),
"server.proxy.trusted-networks",
) {
@@ -342,12 +341,12 @@ impl Config {
id: id.to_string(),
internal_id: 0,
hostname: self
.value_or_default(("server.listener", id, "hostname"), "server.hostname")
.value_or_else(("server.listener", id, "hostname"), "server.hostname")
.ok_or("Hostname directive not found.")?
.to_string(),
data: match protocol {
ServerProtocol::Smtp | ServerProtocol::Lmtp => self
.value_or_default(("server.listener", id, "greeting"), "server.greeting")
.value_or_else(("server.listener", id, "greeting"), "server.greeting")
.unwrap_or(concat!(
"Stalwart SMTP v",
env!("CARGO_PKG_VERSION"),
@@ -356,16 +355,16 @@ impl Config {
.to_string(),
ServerProtocol::Jmap => self
.value_or_default(("server.listener", id, "url"), "server.url")
.value_or_else(("server.listener", id, "url"), "server.url")
.failed(&format!("No 'url' directive found for listener {id:?}"))
.to_string(),
ServerProtocol::Imap | ServerProtocol::Http | ServerProtocol::ManageSieve => self
.value_or_default(("server.listener", id, "url"), "server.url")
.value_or_else(("server.listener", id, "url"), "server.url")
.unwrap_or_default()
.to_string(),
},
max_connections: self
.property_or_default(
.property_or_else(
("server.listener", id, "max-connections"),
"server.max-connections",
)?

170
crates/utils/src/config/mod.rs
View File

@@ -31,7 +31,7 @@ pub mod utils;
use std::{collections::BTreeMap, fmt::Display, net::SocketAddr, sync::Arc, time::Duration};
use ahash::{AHashMap, AHashSet};
use ahash::AHashMap;
use tokio::net::TcpSocket;
use crate::{
@@ -46,6 +46,15 @@ use self::ipmask::IpAddrMask;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct Config {
pub keys: BTreeMap<String, String>,
pub missing: AHashMap<String, Option<String>>,
pub errors: AHashMap<String, ConfigError>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConfigError {
Parse(String),
Build(String),
Macro(String),
}
#[derive(Debug, Default, PartialEq, Eq)]
@@ -153,79 +162,108 @@ impl Config {
)
.failed("Invalid configuration file");
// Extract macros and includes
let mut keys = BTreeMap::new();
let mut includes = AHashSet::new();
let mut macros = AHashMap::new();
config
}
for (key, value) in config.keys {
if let Some(macro_name) = key.strip_prefix("macros.") {
macros.insert(macro_name.to_ascii_lowercase(), value);
} else if key.starts_with("include.files.") {
includes.insert(value);
} else {
keys.insert(key, value);
pub async fn resolve_macros(&mut self) {
let mut replacements = AHashMap::new();
'outer: for (key, value) in &self.keys {
if value.contains("%{") && value.contains("}%") {
let mut result = String::with_capacity(value.len());
let mut snippet: &str = value.as_str();
loop {
if let Some((suffix, macro_name)) = snippet.split_once("%{") {
if !suffix.is_empty() {
result.push_str(suffix);
}
if let Some((class, location, rest)) =
macro_name.split_once("}%").and_then(|(name, rest)| {
name.split_once(':')
.map(|(class, location)| (class, location, rest))
})
{
match class {
"cfg" => {
if let Some(value) = replacements
.get(location)
.or_else(|| self.keys.get(location))
{
result.push_str(value);
} else {
self.errors.insert(
key.clone(),
ConfigError::Macro(format!("Unknown key {location:?}")),
);
}
}
"env" => match std::env::var(location) {
Ok(value) => {
result.push_str(&value);
}
Err(_) => {
self.errors.insert(
key.clone(),
ConfigError::Macro(format!(
"Failed to obtain environment variable {location:?}"
)),
);
}
},
"file" => {
let file_name = location.strip_prefix("//").unwrap_or(location);
match tokio::fs::read(file_name).await {
Ok(value) => match String::from_utf8(value) {
Ok(value) => {
result.push_str(&value);
}
Err(err) => {
self.errors.insert(
key.clone(),
ConfigError::Macro(format!(
"Failed to read file {file_name:?}: {err}"
)),
);
continue 'outer;
}
},
Err(err) => {
self.errors.insert(
key.clone(),
ConfigError::Macro(format!(
"Failed to read file {file_name:?}: {err}"
)),
);
continue 'outer;
}
}
}
"http" | "https" => {}
_ => {
continue 'outer;
}
};
snippet = rest;
}
} else {
result.push_str(snippet);
break;
}
}
replacements.insert(key.clone(), result);
}
}
// Include files
config.keys = keys;
for mut include in includes {
include.replace_macros("include.files", &macros);
config
.parse(&std::fs::read_to_string(&include).failed(&format!(
"Could not read included configuration file {include:?}"
)))
.failed(&format!("Invalid included configuration file {include:?}"));
if !replacements.is_empty() {
for (key, value) in replacements {
self.keys.insert(key, value);
}
}
// Replace macros
for (key, value) in &mut config.keys {
value.replace_macros(key, &macros);
}
config
}
pub fn update(&mut self, settings: Vec<(String, String)>) {
self.keys.extend(settings);
}
}
trait ReplaceMacros: Sized {
fn replace_macros(&mut self, key: &str, macros: &AHashMap<String, String>);
}
impl ReplaceMacros for String {
fn replace_macros(&mut self, key: &str, macros: &AHashMap<String, String>) {
if self.contains("%{") {
let mut result = String::with_capacity(self.len());
let mut value = self.as_str();
loop {
if let Some((suffix, macro_name)) = value.split_once("%{") {
if !suffix.is_empty() {
result.push_str(suffix);
}
if let Some((macro_name, rest)) = macro_name.split_once("}%") {
if let Some(macro_value) = macros.get(&macro_name.to_ascii_lowercase()) {
result.push_str(macro_value);
value = rest;
} else {
failed(&format!("Unknown macro {macro_name:?} for key {key:?}"));
}
} else {
failed(&format!(
"Unterminated macro name {value:?} for key {key:?}"
));
}
} else {
result.push_str(value);
break;
}
}
*self = result;
}
}
}

19
crates/utils/src/config/parser.rs
View File

@@ -33,7 +33,6 @@ use std::fmt::Write;
const MAX_NEST_LEVEL: usize = 10;
// Simple TOML parser for Stalwart Mail Server configuration files.
impl Config {
pub fn new(toml: &str) -> Result<Self> {
let mut config = Config::default();
@@ -399,24 +398,6 @@ impl<'x, 'y> TomlParser<'x, 'y> {
}
self.insert_key(key, value)?;
}
'!' => {
let mut value = String::with_capacity(4);
while let Some(ch) = self.iter.peek() {
if ch.is_alphanumeric() || ['_', '-'].contains(ch) {
value.push(self.next_char(true, false)?);
} else {
break;
}
}
let value = match std::env::var(value.as_str()) {
Ok(value) => value,
Err(_) => {
tracing::warn!("Failed to get environment variable {value:?}");
String::new()
}
};
self.insert_key(key, value)?;
}
ch => {
return if stop_chars.contains(&ch) {
Ok(ch)

10
crates/utils/src/config/tls.rs
View File

@@ -54,8 +54,8 @@ impl Config {
let mut cert = Certificate {
cert: ArcSwap::from(Arc::new(build_certified_key(
self.file_contents(key_cert)?,
self.file_contents(key_pk)?,
self.value_require(key_cert)?.as_bytes().to_vec(),
self.value_require(key_pk)?.as_bytes().to_vec(),
&format!("certificate.{cert_id}"),
)?)),
path: Vec::with_capacity(2),
@@ -94,7 +94,7 @@ impl Config {
.collect::<Vec<_>>();
let cache = PathBuf::from(self.value_require(("acme", acme_id, "cache"))?);
let renew_before: Duration =
self.property_or_static(("acme", acme_id, "renew-before"), "30d")?;
self.property_or_default(("acme", acme_id, "renew-before"), "30d")?;
if directory.is_empty() {
return Err(format!("Missing directory for acme.{acme_id}."));
@@ -108,8 +108,8 @@ impl Config {
let mut domains = Vec::new();
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"),
self.value_or_else(("server.listener", id, "tls.acme"), "server.tls.acme"),
self.value_or_else(("server.listener", id, "hostname"), "server.hostname"),
) {
(Some(listener_acme), Some(hostname)) if listener_acme == acme_id => {
let hostname = hostname.trim().to_lowercase();

218
crates/utils/src/config/utils.rs
View File

@@ -36,7 +36,7 @@ use smtp_proto::MtPriority;
use crate::expr::{Constant, Variable};
use super::{Config, Rate};
use super::{Config, ConfigError, Rate};
impl Config {
pub fn property<T: ParseValue>(&self, key: impl AsKey) -> super::Result<Option<T>> {
@@ -48,7 +48,22 @@ impl Config {
}
}
pub fn property_or_static<T: ParseValue>(
pub fn property_<T: ParseValue>(&mut self, key: impl AsKey) -> Option<T> {
let key = key.as_key();
if let Some(value) = self.keys.get(&key) {
match T::parse_value(key.as_str(), value) {
Ok(value) => Some(value),
Err(err) => {
self.new_parse_error(key, err);
None
}
}
} else {
None
}
}
pub fn property_or_default<T: ParseValue>(
&self,
key: impl AsKey,
default: &str,
@@ -58,7 +73,29 @@ impl Config {
T::parse_value(key, value)
}
pub fn property_or_default<T: ParseValue>(
pub fn property_or_default_<T: ParseValue>(
&mut self,
key: impl AsKey,
default: &str,
) -> Option<T> {
let key = key.as_key();
let value = match self.keys.get(&key) {
Some(value) => value.as_str(),
None => {
self.missing.insert(key.clone(), default.to_string().into());
default
}
};
match T::parse_value(key.as_str(), value) {
Ok(value) => Some(value),
Err(err) => {
self.new_parse_error(key, err);
None
}
}
}
pub fn property_or_else<T: ParseValue>(
&self,
key: impl AsKey,
default: impl AsKey,
@@ -69,6 +106,29 @@ impl Config {
}
}
pub fn property_or_else_<T: ParseValue>(
&mut self,
key: impl AsKey,
default: impl AsKey,
) -> Option<T> {
let key = key.as_key();
let value = match self.value_or_else(key.as_str(), default.clone()) {
Some(value) => value,
None => {
self.missing.insert(default.as_key(), None);
return None;
}
};
match T::parse_value(key.as_str(), value) {
Ok(value) => Some(value),
Err(err) => {
self.new_parse_error(key, err);
None
}
}
}
pub fn property_require<T: ParseValue>(&self, key: impl AsKey) -> super::Result<T> {
match self.property(key.clone()) {
Ok(Some(result)) => Ok(result),
@@ -77,6 +137,22 @@ impl Config {
}
}
pub fn property_require_<T: ParseValue>(&mut self, key: impl AsKey) -> Option<T> {
let key = key.as_key();
if let Some(value) = self.keys.get(&key) {
match T::parse_value(key.as_str(), value) {
Ok(value) => Some(value),
Err(err) => {
self.new_parse_error(key, err);
None
}
}
} else {
self.new_parse_error(key, "Missing property");
None
}
}
pub fn sub_keys<'x, 'y: 'x>(
&'y self,
prefix: impl AsKey,
@@ -111,7 +187,7 @@ impl Config {
.filter_map(move |key| key.strip_prefix(&prefix))
}
pub fn set_values_or_default(
pub fn set_values_or_else(
&self,
prefix: impl AsKey,
default: impl AsKey,
@@ -144,6 +220,27 @@ impl Config {
})
}
pub fn properties_<T: ParseValue>(&mut self, prefix: impl AsKey) -> Vec<(String, T)> {
let full_prefix = prefix.as_key();
let prefix = prefix.as_prefix();
let mut results = Vec::new();
for (key, value) in &self.keys {
if key.starts_with(&prefix) || key == &full_prefix {
match T::parse_value(key.as_str(), value) {
Ok(value) => {
results.push((key.to_string(), value));
}
Err(err) => {
self.errors.insert(key.to_string(), ConfigError::Parse(err));
}
}
}
}
results
}
pub fn value(&self, key: impl AsKey) -> Option<&str> {
self.keys.get(&key.as_key()).map(|s| s.as_str())
}
@@ -159,7 +256,28 @@ impl Config {
.ok_or_else(|| format!("Missing property {:?}.", key.as_key()))
}
pub fn value_or_default(&self, key: impl AsKey, default: impl AsKey) -> Option<&str> {
pub fn value_require_(&mut self, key: impl AsKey) -> Option<&str> {
let key = key.as_key();
if let Some(value) = self.keys.get(&key) {
Some(value.as_str())
} else {
self.errors
.insert(key, ConfigError::Parse("Missing property".to_string()));
None
}
}
pub fn try_parse_value<T: ParseValue>(&mut self, key: impl AsKey, value: &str) -> Option<T> {
match T::parse_value(key.clone(), value) {
Ok(value) => Some(value),
Err(err) => {
self.errors.insert(key.as_key(), ConfigError::Parse(err));
None
}
}
}
pub fn value_or_else(&self, key: impl AsKey, default: impl AsKey) -> Option<&str> {
self.keys
.get(&key.as_key())
.or_else(|| self.keys.get(&default.as_key()))
@@ -179,7 +297,7 @@ impl Config {
})
}
pub fn values_or_default(
pub fn values_or_else(
&self,
prefix: impl AsKey,
default: impl AsKey,
@@ -194,40 +312,38 @@ impl Config {
})
}
pub fn has_prefix(&self, prefix: impl AsKey) -> bool {
let prefix = prefix.as_prefix();
self.keys.keys().any(|k| k.starts_with(&prefix))
}
pub fn take_value(&mut self, key: &str) -> Option<String> {
self.keys.remove(key)
}
pub fn file_contents(&self, key: impl AsKey) -> super::Result<Vec<u8>> {
pub fn value_or_warn(&mut self, key: impl AsKey) -> Option<&str> {
let key = key.as_key();
if let Some(value) = self.keys.get(&key) {
if let Some(value) = value.strip_prefix("file://") {
std::fs::read(value).map_err(|err| {
format!("Failed to read file {value:?} for property {key:?}: {err}")
})
} else {
Ok(value.to_string().into_bytes())
match self.keys.get(&key) {
Some(value) => Some(value.as_str()),
None => {
self.missing.insert(key, None);
None
}
} else {
Err(format!("Property {key:?} not found in configuration file."))
}
}
pub fn text_file_contents(&self, key: impl AsKey) -> super::Result<Option<String>> {
let key = key.as_key();
if let Some(value) = self.keys.get(&key) {
if let Some(value) = value.strip_prefix("file://") {
std::fs::read_to_string(value)
.map_err(|err| {
format!("Failed to read file {value:?} for property {key:?}: {err}")
})
.map(Some)
} else {
Ok(Some(value.to_string()))
}
} else {
Ok(None)
}
pub fn new_parse_error(&mut self, key: impl AsKey, details: impl Into<String>) {
self.errors
.insert(key.as_key(), ConfigError::Parse(details.into()));
}
pub fn new_build_error(&mut self, key: impl AsKey, details: impl Into<String>) {
self.errors
.insert(key.as_key(), ConfigError::Build(details.into()));
}
pub fn new_missing_property(&mut self, key: impl AsKey) {
self.missing.insert(key.as_key(), None);
}
}
@@ -559,21 +675,33 @@ impl ParseValue for HashAlgorithm {
impl ParseValue for Duration {
fn parse_value(key: impl AsKey, value: &str) -> super::Result<Self> {
let duration = value.trim_end().to_ascii_lowercase();
let (num, multiplier) = if let Some(num) = duration.strip_suffix('d') {
(num, 24 * 60 * 60 * 1000)
} else if let Some(num) = duration.strip_suffix('h') {
(num, 60 * 60 * 1000)
} else if let Some(num) = duration.strip_suffix('m') {
(num, 60 * 1000)
} else if let Some(num) = duration.strip_suffix("ms") {
(num, 1)
} else if let Some(num) = duration.strip_suffix('s') {
(num, 1000)
} else {
(duration.as_str(), 1)
let mut digits = String::new();
let mut multiplier = String::new();
for ch in value.chars() {
if ch.is_ascii_digit() {
digits.push(ch);
} else if !ch.is_ascii_whitespace() {
multiplier.push(ch.to_ascii_lowercase());
}
}
let multiplier = match multiplier.as_str() {
"d" => 24 * 60 * 60 * 1000,
"h" => 60 * 60 * 1000,
"m" => 60 * 1000,
"s" => 1000,
"ms" | "" => 1,
_ => {
return Err(format!(
"Invalid duration value {:?} for property {:?}.",
value,
key.as_key()
))
}
};
num.trim()
digits
.parse::<u64>()
.ok()
.and_then(|num| {

4
crates/utils/src/lib.rs
View File

@@ -182,7 +182,7 @@ pub fn enable_tracing(config: &Config, message: &str) -> config::Result<Option<W
tracing_subscriber::FmtSubscriber::builder()
.with_env_filter(env_filter)
.with_writer(non_blocking)
.with_ansi(config.property_or_static("global.tracing.ansi", "true")?)
.with_ansi(config.property_or_default("global.tracing.ansi", "true")?)
.finish(),
)
.failed("Failed to set subscriber");
@@ -192,7 +192,7 @@ pub fn enable_tracing(config: &Config, message: &str) -> config::Result<Option<W
tracing::subscriber::set_global_default(
tracing_subscriber::FmtSubscriber::builder()
.with_env_filter(env_filter)
.with_ansi(config.property_or_static("global.tracing.ansi", "true")?)
.with_ansi(config.property_or_default("global.tracing.ansi", "true")?)
.finish(),
)
.failed("Failed to set subscriber");