SMTP server passing tests.

crates/utils/Cargo.toml

@@ -13,6 +13,15 @@ serde = { version = "1.0", features = ["derive"]}
 tracing = "0.1"
 mail-auth = { git = "https://github.com/stalwartlabs/mail-auth" }
 smtp-proto = { git = "https://github.com/stalwartlabs/smtp-proto" }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-appender = "0.2"
+tracing-opentelemetry = "0.18.0"
+opentelemetry = { version = "0.18.0", features = ["rt-tokio"] }
+opentelemetry-otlp = { version = "0.11.0", features = ["http-proto", "reqwest-client", "reqwest-rustls"] }
+opentelemetry-semantic-conventions = { version = "0.10.0" }
 
 [target.'cfg(unix)'.dependencies]
 privdrop = "0.5.3"
+
+[features]
+test_mode = []

crates/utils/src/config/listener.rs

@@ -277,14 +277,20 @@ impl Config {
                 .value_or_default(("server.listener", id, "hostname"), "server.hostname")
                 .ok_or("Hostname directive not found.")?
                 .to_string(),
-            data: if matches!(protocol, ServerProtocol::Smtp | ServerProtocol::Lmtp) {
-                self.value_or_default(("server.listener", id, "data"), "server.data")
+            data: match protocol {
+                ServerProtocol::Smtp | ServerProtocol::Lmtp => self
+                    .value_or_default(("server.listener", id, "greeting"), "server.greeting")
                     .unwrap_or("Stalwart SMTP at your service")
-                    .to_string()
-            } else {
-                self.value_or_default(("server.listener", id, "url"), "server.url")
+                    .to_string(),
+
+                ServerProtocol::Jmap => self
+                    .value_or_default(("server.listener", id, "url"), "server.url")
                     .failed(&format!("No 'url' directive found for listener {id:?}"))
-                    .to_string()
+                    .to_string(),
+                ServerProtocol::Imap | ServerProtocol::Http => self
+                    .value_or_default(("server.listener", id, "url"), "server.url")
+                    .unwrap_or_default()
+                    .to_string(),
             },
             max_connections: self
                 .property_or_default(

crates/utils/src/config/mod.rs

@@ -31,6 +31,8 @@ use std::{collections::BTreeMap, fmt::Display, net::SocketAddr, time::Duration};
 use rustls::ServerConfig;
 use tokio::net::TcpSocket;
 
+use crate::{failed, UnwrapFailure};
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Config {
     pub keys: BTreeMap<String, String>,
@@ -90,3 +92,36 @@ impl Display for ServerProtocol {
 }
 
 pub type Result<T> = std::result::Result<T, String>;
+
+impl Config {
+    pub fn init() -> Self {
+        let mut config_path = None;
+        let mut found_param = false;
+
+        for arg in std::env::args().skip(1) {
+            if let Some((key, value)) = arg.split_once('=') {
+                if key.starts_with("--config") {
+                    config_path = value.trim().to_string().into();
+                    break;
+                } else {
+                    failed(&format!("Invalid command line argument: {key}"));
+                }
+            } else if found_param {
+                config_path = arg.into();
+                break;
+            } else if arg.starts_with("--config") {
+                found_param = true;
+            } else {
+                failed(&format!("Invalid command line argument: {arg}"));
+            }
+        }
+
+        Config::parse(
+            &std::fs::read_to_string(
+                config_path.failed("Missing parameter --config=<path-to-config>."),
+            )
+            .failed("Could not read configuration file"),
+        )
+        .failed("Invalid configuration file")
+    }
+}

crates/utils/src/lib.rs

@@ -21,11 +21,27 @@
  * for more details.
 */
 
+use std::collections::HashMap;
+
+use config::Config;
+
 pub mod codec;
 pub mod config;
 pub mod listener;
 pub mod map;
 
+use opentelemetry::{
+    sdk::{
+        trace::{self, Sampler},
+        Resource,
+    },
+    KeyValue,
+};
+use opentelemetry_otlp::WithExportConfig;
+use opentelemetry_semantic_conventions::resource::{SERVICE_NAME, SERVICE_VERSION};
+use tracing_appender::non_blocking::WorkerGuard;
+use tracing_subscriber::{prelude::__tracing_subscriber_SubscriberExt, EnvFilter};
+
 pub trait UnwrapFailure<T> {
     fn failed(self, action: &str) -> T;
 }
@@ -47,8 +63,14 @@ impl<T, E: std::fmt::Display> UnwrapFailure<T> for Result<T, E> {
         match self {
             Ok(result) => result,
             Err(err) => {
-                eprintln!("{message}: {err}");
-                std::process::exit(1);
+                #[cfg(feature = "test_mode")]
+                panic!("{message}: {err}");
+
+                #[cfg(not(feature = "test_mode"))]
+                {
+                    eprintln!("{message}: {err}");
+                    std::process::exit(1);
+                }
             }
         }
     }
@@ -58,3 +80,127 @@ pub fn failed(message: &str) -> ! {
     eprintln!("{message}");
     std::process::exit(1);
 }
+
+pub fn enable_tracing(config: &Config) -> config::Result<Option<WorkerGuard>> {
+    let level = config.value("global.tracing.level").unwrap_or("info");
+    let env_filter = EnvFilter::builder()
+        .parse(format!("stalwart_smtp={}", level))
+        .failed("Failed to log level");
+    match config.value("global.tracing.method").unwrap_or_default() {
+        "log" => {
+            let path = config.value_require("global.tracing.path")?;
+            let prefix = config.value_require("global.tracing.prefix")?;
+            let file_appender = match config.value("global.tracing.rotate").unwrap_or("daily") {
+                "daily" => tracing_appender::rolling::daily(path, prefix),
+                "hourly" => tracing_appender::rolling::hourly(path, prefix),
+                "minutely" => tracing_appender::rolling::minutely(path, prefix),
+                "never" => tracing_appender::rolling::never(path, prefix),
+                rotate => {
+                    return Err(format!("Unsupported log rotation strategy {rotate:?}"));
+                }
+            };
+
+            let (non_blocking, guard) = tracing_appender::non_blocking(file_appender);
+            tracing::subscriber::set_global_default(
+                tracing_subscriber::FmtSubscriber::builder()
+                    .with_env_filter(env_filter)
+                    .with_writer(non_blocking)
+                    .finish(),
+            )
+            .failed("Failed to set subscriber");
+            Ok(guard.into())
+        }
+        "stdout" => {
+            tracing::subscriber::set_global_default(
+                tracing_subscriber::FmtSubscriber::builder()
+                    .with_env_filter(env_filter)
+                    .finish(),
+            )
+            .failed("Failed to set subscriber");
+
+            Ok(None)
+        }
+        "otel" | "open-telemetry" => {
+            let tracer = match config.value_require("global.tracing.transport")? {
+                "grpc" => {
+                    let mut exporter = opentelemetry_otlp::new_exporter().tonic();
+                    if let Some(endpoint) = config.value("global.tracing.endpoint") {
+                        exporter = exporter.with_endpoint(endpoint);
+                    }
+                    opentelemetry_otlp::new_pipeline()
+                        .tracing()
+                        .with_exporter(exporter)
+                }
+                "http" => {
+                    let mut headers = HashMap::new();
+                    for (_, value) in config.values("global.tracing.headers") {
+                        if let Some((key, value)) = value.split_once(':') {
+                            headers.insert(key.trim().to_string(), value.trim().to_string());
+                        } else {
+                            return Err(format!("Invalid open-telemetry header {value:?}"));
+                        }
+                    }
+                    let mut exporter = opentelemetry_otlp::new_exporter()
+                        .http()
+                        .with_endpoint(config.value_require("global.tracing.endpoint")?);
+                    if !headers.is_empty() {
+                        exporter = exporter.with_headers(headers);
+                    }
+                    opentelemetry_otlp::new_pipeline()
+                        .tracing()
+                        .with_exporter(exporter)
+                }
+                transport => {
+                    return Err(format!(
+                        "Unsupported open-telemetry transport {transport:?}"
+                    ));
+                }
+            }
+            .with_trace_config(
+                trace::config()
+                    .with_resource(Resource::new(vec![
+                        KeyValue::new(SERVICE_NAME, "stalwart-smtp".to_string()),
+                        KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION").to_string()),
+                    ]))
+                    .with_sampler(Sampler::AlwaysOn),
+            )
+            .install_batch(opentelemetry::runtime::Tokio)
+            .failed("Failed to create tracer");
+
+            tracing::subscriber::set_global_default(
+                tracing_subscriber::Registry::default()
+                    .with(tracing_opentelemetry::layer().with_tracer(tracer))
+                    .with(env_filter),
+            )
+            .failed("Failed to set subscriber");
+
+            Ok(None)
+        }
+        _ => Ok(None),
+    }
+}
+
+pub async fn wait_for_shutdown() {
+    #[cfg(not(target_env = "msvc"))]
+    {
+        use tokio::signal::unix::{signal, SignalKind};
+
+        let mut h_term = signal(SignalKind::terminate()).failed("start signal handler");
+        let mut h_int = signal(SignalKind::interrupt()).failed("start signal handler");
+
+        tokio::select! {
+            _ = h_term.recv() => tracing::debug!("Received SIGTERM."),
+            _ = h_int.recv() => tracing::debug!("Received SIGINT."),
+        };
+    }
+
+    #[cfg(target_env = "msvc")]
+    {
+        match tokio::signal::ctrl_c().await {
+            Ok(()) => {}
+            Err(err) => {
+                eprintln!("Unable to listen for shutdown signal: {}", err);
+            }
+        }
+    }
+}

crates/utils/src/listener/listen.rs

@@ -103,6 +103,7 @@ impl Server {
                                 instance = instance.id,
                                 protocol = ?instance.protocol,
                                 "Listener shutting down.");
+                            manager.shutdown();
                             break;
                         }
                     };
@@ -113,11 +114,7 @@ impl Server {
 }
 
 impl Servers {
-    pub fn spawn(
-        self,
-        config: &Config,
-        spawn: impl Fn(Server, watch::Receiver<bool>),
-    ) -> watch::Sender<bool> {
+    pub fn bind(&self, config: &Config) {
         // Bind as root
         for server in &self.inner {
             for listener in &server.listeners {
@@ -139,7 +136,9 @@ impl Servers {
                 pd.apply().failed("Failed to drop privileges");
             }
         }
+    }
 
+    pub fn spawn(self, spawn: impl Fn(Server, watch::Receiver<bool>)) -> watch::Sender<bool> {
         // Spawn listeners
         let (shutdown_tx, shutdown_rx) = watch::channel(false);
         for server in self.inner {

crates/utils/src/listener/mod.rs

@@ -37,4 +37,5 @@ pub struct SessionData<T: AsyncRead + AsyncWrite + Unpin + 'static> {
 
 pub trait SessionManager: Sync + Send + 'static + Clone {
     fn spawn(&self, session: SessionData<TcpStream>);
+    fn shutdown(&self);
 }