diff --git a/crates/common/src/network/control.rs b/crates/common/src/network/control.rs index 4e36983..cb97633 100644 --- a/crates/common/src/network/control.rs +++ b/crates/common/src/network/control.rs @@ -23,11 +23,18 @@ //! process stops answering; anything that still routes the port is the //! operator's to reconcile, and is deliberately left alone. -use crate::config::server::ServerProtocol; +use crate::config::server::{Listener, ServerProtocol}; +use crate::network::TcpAcceptor; use ahash::AHashMap; use parking_lot::RwLock; +use std::sync::OnceLock; use tokio::sync::watch; +/// How a listener is spawned. Only `main` knows how to build the session +/// manager for a protocol, so it leaves this behind at startup and the policy +/// uses it to put a listener back without a restart (LP-5). +pub type SpawnListener = Box) + Send + Sync>; + /// A listener that is currently accepting, and the switch that stops it. struct Running { protocol: ServerProtocol, @@ -47,6 +54,7 @@ pub struct ListenerInfo { #[derive(Default)] pub struct ListenerControl { running: RwLock>, + spawner: OnceLock, } impl ListenerControl { @@ -70,6 +78,33 @@ impl ListenerControl { shutdown_rx } + /// Remembers how to spawn a listener, once, at startup. Later calls are + /// ignored, so nothing can swap the spawner out from under a running + /// server. + pub fn set_spawner(&self, spawner: SpawnListener) { + let _ = self.spawner.set(spawner); + } + + /// Whether a spawner has been left behind. Without one, a listener can be + /// stopped but not started, and the caller has to say so rather than + /// promise a port that will not open until a restart. + pub fn can_spawn(&self) -> bool { + self.spawner.get().is_some() + } + + /// Starts a listener and registers it, so it can be stopped again. + /// Returns false when no spawner was left behind. + pub fn spawn(&self, listener: Listener, acceptor: TcpAcceptor) -> bool { + let Some(spawner) = self.spawner.get() else { + return false; + }; + + let ports = listener.listeners.iter().map(|l| l.addr.port()).collect(); + let shutdown_rx = self.register(listener.id.clone(), listener.protocol, ports); + spawner(listener, acceptor, shutdown_rx); + true + } + /// Stops one listener by id. Returns what was stopped, or `None` when no /// listener of that id is running. pub fn stop(&self, id: &str) -> Option { diff --git a/crates/common/src/network/legacy.rs b/crates/common/src/network/legacy.rs new file mode 100644 index 0000000..5517a4f --- /dev/null +++ b/crates/common/src/network/legacy.rs @@ -0,0 +1,188 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Turning the legacy-protocols switch, and making it true of the running +//! server (legacy-protocols spec, LP-1, LP-2 and LP-5). +//! +//! Two halves meet here. `inbuxa_features::security` decides what the policy +//! means and owns the listener **objects**; [`ListenerControl`] owns the +//! running **sockets**. Neither can do the job alone, and only `Server` has +//! both, so the join lives here. +//! +//! Order matters in both directions. Closing removes the object first and then +//! stops the socket: a socket stopped before its object is gone would come +//! back on the next restart. Opening puts the object back first and then +//! spawns, for the same reason in reverse. +//! +//! Nothing here touches the host's firewall, NAT port-forwards or any proxy +//! (LP-20). The server stops answering; what still routes the port is the +//! operator's to reconcile. + +use crate::{Server, config::server::Listeners, network::TcpAcceptor}; +use inbuxa_features::security::{ + listeners, + protocol_policy::{self, ProtocolPolicy, SavedListener}, +}; +use store::registry::bootstrap::Bootstrap; + +/// What turning the switch actually did. +#[derive(Debug, Default)] +pub struct PolicyChange { + /// Listeners removed and stopped (LP-1). + pub closed: Vec, + /// Listeners put back and started again (LP-5). + pub reopened: Vec, + /// Listeners that could not be put back, with the reason. Each stays + /// saved for another try (LP-5). + pub failed: Vec<(SavedListener, String)>, + /// Properties the locks overruled (LP-21). + pub overruled: Vec<&'static str>, + /// Listeners whose object is right but whose socket needs a restart, + /// because no spawner was left behind. Empty on a normally booted server. + pub pending_restart: Vec, +} + +impl PolicyChange { + /// Whether anything at all happened, for the caller deciding to emit + /// `security.legacy-protocols-changed` (LP-8). + pub fn is_empty(&self) -> bool { + self.closed.is_empty() + && self.reopened.is_empty() + && self.failed.is_empty() + && self.overruled.is_empty() + } +} + +impl Server { + /// The policy in force. + pub async fn protocol_policy(&self) -> trc::Result { + protocol_policy::get(&self.core.storage.data).await + } + + /// Turns the switch, and makes it true of the running server. + /// + /// `requested` is what the client asked for; the locks are applied to it + /// first (LP-21), so what gets stored is what the server allows, not what + /// was asked. Returns what actually happened, for the response and the + /// event. + pub async fn set_protocol_policy( + &self, + requested: ProtocolPolicy, + changed_by: Option, + ) -> trc::Result { + let mut policy = requested; + let mut change = PolicyChange { + overruled: policy.apply_locks(), + ..Default::default() + }; + + // Carry forward what earlier changes saved: the client never sets + // this, and a /set that omitted it must not lose the listeners still + // waiting to come back. + let previous = self.protocol_policy().await?; + policy.saved_listeners = previous.saved_listeners; + policy.changed_at = Some(store::write::now() * 1000); + policy.changed_by = changed_by; + + if policy.legacy_protocols.is_disabled() { + self.close_legacy_listeners(&mut policy, &mut change).await?; + } else { + self.reopen_legacy_listeners(&mut policy, &mut change) + .await?; + } + + protocol_policy::set(&self.core.storage.data, &policy).await?; + + Ok(change) + } + + /// Removes the listener objects the policy closes, then stops their + /// sockets (LP-1, LP-2). + async fn close_legacy_listeners( + &self, + policy: &mut ProtocolPolicy, + change: &mut PolicyChange, + ) -> trc::Result<()> { + let removed = listeners::close(self.registry(), policy).await?; + + for saved in &removed { + // The runtime registry is keyed by the listener's name, which is + // what `close` returns as the saved listener's id. + self.inner.data.listener_control.stop(&saved.id); + } + + policy.saved_listeners.extend(removed.iter().cloned()); + change.closed = removed; + + Ok(()) + } + + /// Puts back every saved listener and starts it again (LP-5). + async fn reopen_legacy_listeners( + &self, + policy: &mut ProtocolPolicy, + change: &mut PolicyChange, + ) -> trc::Result<()> { + if policy.saved_listeners.is_empty() { + return Ok(()); + } + + let saved = std::mem::take(&mut policy.saved_listeners); + let (restored, failed) = listeners::reopen(self.registry(), &saved).await?; + + // A listener that could not be put back stays saved for another try. + policy.saved_listeners = failed.iter().map(|(listener, _)| listener.clone()).collect(); + change.failed = failed; + + if !restored.is_empty() { + change.pending_restart = self.spawn_restored_listeners(&restored).await?; + } + change.reopened = restored; + + Ok(()) + } + + /// Binds and spawns the listeners just put back, so a port opens without a + /// restart. Returns the names that still need one. + async fn spawn_restored_listeners(&self, restored: &[SavedListener]) -> trc::Result> { + let control = &self.inner.data.listener_control; + if !control.can_spawn() { + return Ok(restored.iter().map(|listener| listener.id.clone()).collect()); + } + + // Re-parse from the registry rather than from the saved object: the + // socket has to be created and bound afresh, and the parser is what + // knows how. The objects are already back, so this sees them. + let mut bootstrap = Bootstrap::new(self.registry().clone()).await; + let mut parsed = Listeners::parse(&mut bootstrap).await; + parsed + .parse_tcp_acceptors(&mut bootstrap, self.inner.clone()) + .await; + + let wanted: Vec<&str> = restored.iter().map(|l| l.id.as_str()).collect(); + let mut spawned = Vec::new(); + + let mut acceptors = std::mem::take(&mut parsed.tcp_acceptors); + for listener in parsed.servers { + if !wanted.contains(&listener.id.as_str()) || control.is_running(&listener.id) { + continue; + } + let acceptor = acceptors + .remove(&listener.id) + .unwrap_or(TcpAcceptor::Plain); + let id = listener.id.clone(); + if control.spawn(listener, acceptor) { + spawned.push(id); + } + } + + Ok(restored + .iter() + .map(|listener| listener.id.clone()) + .filter(|id| !spawned.contains(id)) + .collect()) + } +} diff --git a/crates/common/src/network/mod.rs b/crates/common/src/network/mod.rs index 46d2e32..b4db05c 100644 --- a/crates/common/src/network/mod.rs +++ b/crates/common/src/network/mod.rs @@ -36,6 +36,7 @@ pub mod autoconfig; pub mod control; pub mod dkim; pub mod dns; +pub mod legacy; pub mod limiter; pub mod listen; pub mod mta; diff --git a/crates/main/src/main.rs b/crates/main/src/main.rs index 724cd41..f44e44f 100644 --- a/crates/main/src/main.rs +++ b/crates/main/src/main.rs @@ -9,14 +9,21 @@ #![warn(clippy::cast_possible_wrap)] #![warn(clippy::cast_sign_loss)] -use common::{BuildServer, config::server::ServerProtocol, manager::boot::BootManager}; +use common::{ + BuildServer, Inner, + config::server::{Listener, ServerProtocol}, + manager::boot::BootManager, + network::TcpAcceptor, +}; use http::HttpSessionManager; use imap::core::ImapSessionManager; use managesieve::core::ManageSieveSessionManager; use pop3::Pop3SessionManager; use services::{StartServices, broadcast::subscriber::spawn_broadcast_subscriber}; use smtp::{StartQueueManager, core::SmtpSessionManager}; +use std::sync::Arc; use std::time::Duration; +use tokio::sync::watch; use trc::Collector; use utils::wait_for_shutdown; @@ -75,43 +82,24 @@ async fn main() -> std::io::Result<()> { // rest accepting (legacy-protocols LP-2). The registry lives in `Data` and // so outlives the listeners, which it must: it owns the sending ends. let listener_control = &init.inner.data.listener_control; + let spawn_inner = init.inner.clone(); let (shutdown_tx, shutdown_rx) = init.servers .spawn_with_control(listener_control, |server, acceptor, shutdown_rx| { - match &server.protocol { - ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( - SmtpSessionManager::new(init.inner.clone()), - init.inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Http => server.spawn( - HttpSessionManager::new(init.inner.clone()), - init.inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Imap => server.spawn( - ImapSessionManager::new(init.inner.clone()), - init.inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::Pop3 => server.spawn( - Pop3SessionManager::new(init.inner.clone()), - init.inner.clone(), - acceptor, - shutdown_rx, - ), - ServerProtocol::ManageSieve => server.spawn( - ManageSieveSessionManager::new(init.inner.clone()), - init.inner.clone(), - acceptor, - shutdown_rx, - ), - }; + spawn_listener(&spawn_inner, server, acceptor, shutdown_rx); }); + // Leave behind how to spawn a listener, so putting one back opens its port + // without a restart (LP-5). Only this file knows the session manager for a + // protocol, so only this file can say. + let spawn_inner = init.inner.clone(); + init.inner + .data + .listener_control + .set_spawner(Box::new(move |server, acceptor, shutdown_rx| { + spawn_listener(&spawn_inner, server, acceptor, shutdown_rx); + })); + // Start broadcast subscriber let inner = init.inner.clone(); spawn_broadcast_subscriber(init.inner, shutdown_rx); @@ -132,3 +120,47 @@ async fn main() -> std::io::Result<()> { Ok(()) } + +/// Starts one listener under the session manager its protocol calls for. +/// +/// Used twice: once for every listener at startup, and again whenever the +/// legacy-protocols switch puts a listener back (LP-5). +fn spawn_listener( + inner: &Arc, + server: Listener, + acceptor: TcpAcceptor, + shutdown_rx: watch::Receiver, +) { + match &server.protocol { + ServerProtocol::Smtp | ServerProtocol::Lmtp => server.spawn( + SmtpSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Http => server.spawn( + HttpSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Imap => server.spawn( + ImapSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::Pop3 => server.spawn( + Pop3SessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + ServerProtocol::ManageSieve => server.spawn( + ManageSieveSessionManager::new(inner.clone()), + inner.clone(), + acceptor, + shutdown_rx, + ), + } +}