From c1da3e02597acda4764906da241dde85ea0e13ff Mon Sep 17 00:00:00 2001 From: Jeremy HERGAULT Date: Thu, 6 Aug 2026 16:55:33 +0200 Subject: [PATCH 1/3] feat: improve processor reloading and allow it for SSL Signed-off-by: Jeremy HERGAULT --- prosa/Cargo.toml | 2 +- prosa/examples/proc.rs | 34 ++--- prosa/src/core/adaptor.rs | 11 +- prosa/src/core/proc.rs | 8 +- prosa/src/core/settings.rs | 149 ++++++++++++++++++- prosa/src/inj/proc.rs | 24 +--- prosa/src/io.rs | 11 +- prosa/src/io/listener.rs | 216 ++++++++++++++++++---------- prosa/src/io/stream.rs | 173 ++++++++++++++++------ prosa/src/stub/proc.rs | 69 ++++----- prosa_book/src/ch01-02-03-stream.md | 24 ++++ prosa_book/src/ch03-01-settings.md | 2 +- prosa_book/src/ch03-02-creation.md | 11 +- prosa_book/src/ch03-05-service.md | 20 +-- prosa_book/src/ch03-06-events.md | 15 +- 15 files changed, 541 insertions(+), 228 deletions(-) diff --git a/prosa/Cargo.toml b/prosa/Cargo.toml index a4a7e35..fc29d02 100644 --- a/prosa/Cargo.toml +++ b/prosa/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "prosa" -version = "0.5.0" +version = "0.5.1" authors.workspace = true description = "ProSA core" homepage.workspace = true diff --git a/prosa/examples/proc.rs b/prosa/examples/proc.rs index 708a167..b9677ba 100644 --- a/prosa/examples/proc.rs +++ b/prosa/examples/proc.rs @@ -79,23 +79,25 @@ where info!("Proc {} received an error: {:?}", self.get_proc_id(), err); }, InternalMsg::Config(config) => { - let settings = config.get_proc::(self.proc.as_ref())?; - - if self.settings.service_name != settings.service_name { - self.proc - .remove_service_proc(vec![self.settings.service_name.clone()]) - .await?; - self.proc - .add_service_proc(vec![settings.service_name.clone()]) - .await?; + if let Some(settings) = config + .reload_proc::(self.proc.as_ref(), &adaptor) + { + if self.settings.service_name != settings.service_name { + self.proc + .remove_service_proc(vec![self.settings.service_name.clone()]) + .await?; + self.proc + .add_service_proc(vec![settings.service_name.clone()]) + .await?; + } + + if self.settings.tick_secs != settings.tick_secs { + interval = settings.interval(); + } + + info!("Proc {} reloaded settings: {:?}", self.get_proc_id(), settings); + self.settings = settings; } - - if self.settings.tick_secs != settings.tick_secs { - interval = settings.interval(); - } - - info!("Proc {} reloaded settings: {:?}", self.get_proc_id(), settings); - self.settings = settings; }, InternalMsg::Service(table) => { debug!("New service table received:\n{}\n", table); diff --git a/prosa/src/core/adaptor.rs b/prosa/src/core/adaptor.rs index 28cc399..5a0d2e4 100644 --- a/prosa/src/core/adaptor.rs +++ b/prosa/src/core/adaptor.rs @@ -62,12 +62,17 @@ pub trait Adaptor { /// existing adaptors compatible by ignoring the configuration. /// /// Processor implementations should call this from their loop when they - /// receive [`InternalMsg::Config`](crate::core::msg::InternalMsg::Config): + /// receive [`InternalMsg::Config`](crate::core::msg::InternalMsg::Config), which + /// [`ProsaConfig::reload_proc`](crate::core::settings::ProsaConfig::reload_proc) does along with + /// the processor settings: /// /// ```rust,ignore /// InternalMsg::Config(config) => { - /// self.settings = config.get_proc(self.proc.as_ref())?; - /// adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref()))?; + /// if let Some(settings) = + /// config.reload_proc::(self.proc.as_ref(), &adaptor) + /// { + /// self.settings = settings; + /// } /// } /// ``` fn reload_config(&self, _config: Option<&config::Config>) -> Result<(), config::ConfigError> { diff --git a/prosa/src/core/proc.rs b/prosa/src/core/proc.rs index b1bc55f..778865b 100644 --- a/prosa/src/core/proc.rs +++ b/prosa/src/core/proc.rs @@ -116,8 +116,12 @@ //! // TODO process the error //! }, //! InternalMsg::Config(config) => { -//! self.settings = config.get_proc(self.proc.as_ref())?; -//! adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref()))?; +//! if let Some(settings) = +//! config.reload_proc::(self.proc.as_ref(), &adaptor) +//! { +//! // TODO apply the difference between `settings` and `self.settings` +//! self.settings = settings; +//! } //! }, //! InternalMsg::Service(table) => self.service = table, //! InternalMsg::Shutdown => { diff --git a/prosa/src/core/settings.rs b/prosa/src/core/settings.rs index 4c0e2a6..d9084eb 100644 --- a/prosa/src/core/settings.rs +++ b/prosa/src/core/settings.rs @@ -20,6 +20,7 @@ use serde::Serialize; use serde::de::DeserializeOwned; use tokio::sync::mpsc; +use super::adaptor::Adaptor; use super::proc::ProcBusParam; /// Re-export of prosa_utils for observability config @@ -312,7 +313,7 @@ impl ProsaConfig { } /// Deserialize a processor configuration from its processor name. - pub fn get_proc(&self, proc: &impl ProcBusParam) -> Result + pub fn get_proc(&self, proc: &(impl ProcBusParam + ?Sized)) -> Result where C: DeserializeOwned, { @@ -320,10 +321,53 @@ impl ProsaConfig { } /// Access a processor adaptor configuration from its processor name. - pub fn get_adaptor_config(&self, proc: &impl ProcBusParam) -> Option<&Config> { + pub fn get_adaptor_config(&self, proc: &(impl ProcBusParam + ?Sized)) -> Option<&Config> { self.adaptor_configs.get(&proc.get_proc_config_key()) } + /// Reload a processor settings and its adaptor configuration in one step. + /// + /// Return `None` if either of them can't be reloaded, so the processor can keep running on its + /// current configuration. A processor that has no configuration section keeps the settings it + /// was created with, which is reported at debug level. Anything else is logged as a warning: + /// + /// ```rust,ignore + /// InternalMsg::Config(config) => { + /// if let Some(settings) = + /// config.reload_proc::(self.proc.as_ref(), &adaptor) + /// { + /// // ... apply the difference between `settings` and `self.settings` + /// self.settings = settings; + /// } + /// } + /// ``` + pub fn reload_proc(&self, proc: &dyn ProcBusParam, adaptor: &dyn Adaptor) -> Option + where + S: DeserializeOwned, + { + let settings = match self.get_proc::(proc) { + Ok(settings) => settings, + Err(config::ConfigError::NotFound(_)) => { + log::debug!("No configuration section for processor {}", proc.name()); + return None; + } + Err(err) => { + log::warn!("Can't reload settings for processor {}: {err}", proc.name()); + return None; + } + }; + + if let Err(err) = adaptor.reload_config(self.get_adaptor_config(proc)) { + log::warn!( + "Can't reload adaptor configuration for processor {}: {err}", + proc.name() + ); + return None; + } + + Some(settings) + } + /// Return every configuration path watched to maintain this configuration. pub fn watch_paths(&self, config_path: &str) -> Vec { let mut watch_paths = config_watch_paths(Path::new(config_path)) @@ -662,6 +706,107 @@ mod tests { Ok(()) } + #[test] + fn test_reload_proc() -> Result<(), config::ConfigError> { + struct TestProc(&'static str); + impl ProcBusParam for TestProc { + fn get_proc_id(&self) -> u32 { + 1 + } + + fn name(&self) -> &str { + self.0 + } + } + + struct TestAdaptor { + fail: bool, + } + impl Adaptor for TestAdaptor { + fn reload_config(&self, _config: Option<&Config>) -> Result<(), config::ConfigError> { + if self.fail { + Err(config::ConfigError::Message("adaptor failure".into())) + } else { + Ok(()) + } + } + + fn terminate(&self) {} + } + + #[derive(serde::Deserialize)] + struct TestProcSettings { + service_name: String, + } + + let config = ProsaConfig::from_config( + Config::builder() + .set_override("proc_1.service_name", "PROC_TEST")? + .build()?, + )?; + + let settings = config + .reload_proc::(&TestProc("proc-1"), &TestAdaptor { fail: false }) + .expect("Processor settings should be reloaded"); + assert_eq!("PROC_TEST", settings.service_name); + + // The adaptor keeps the processor on its current configuration + assert!( + config + .reload_proc::(&TestProc("proc-1"), &TestAdaptor { fail: true }) + .is_none() + ); + + // A processor without a configuration section keeps its settings without warning + assert!(matches!( + config.get_proc::(&TestProc("proc-unknown")), + Err(config::ConfigError::NotFound(_)) + )); + assert!( + config + .reload_proc::( + &TestProc("proc-unknown"), + &TestAdaptor { fail: false } + ) + .is_none() + ); + + // But an invalid section is a reload failure + let invalid_config = ProsaConfig::from_config( + Config::builder() + .set_override("proc_1.service_name", vec!["not", "a", "string"])? + .build()?, + )?; + assert!(matches!( + invalid_config.get_proc::(&TestProc("proc-1")), + Err(config::ConfigError::Type { .. }) + )); + assert!( + invalid_config + .reload_proc::(&TestProc("proc-1"), &TestAdaptor { fail: false }) + .is_none() + ); + + // So is a section that exists but misses a mandatory setting, which `config` reports as + // `At` rather than the `NotFound` of an absent section + let incomplete_config = ProsaConfig::from_config( + Config::builder() + .set_override("proc_1.unrelated", "value")? + .build()?, + )?; + assert!(matches!( + incomplete_config.get_proc::(&TestProc("proc-1")), + Err(config::ConfigError::At { .. }) + )); + assert!( + incomplete_config + .reload_proc::(&TestProc("proc-1"), &TestAdaptor { fail: false }) + .is_none() + ); + + Ok(()) + } + fn unique_test_dir(prefix: &str) -> PathBuf { let timestamp = SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/prosa/src/inj/proc.rs b/prosa/src/inj/proc.rs index 6160324..fe19209 100644 --- a/prosa/src/inj/proc.rs +++ b/prosa/src/inj/proc.rs @@ -1,7 +1,7 @@ use std::time::Duration; use crate::otel::{KeyValue, metrics::Histogram}; -use crate::tracing::{debug, warn}; +use crate::tracing::debug; use prosa_macros::{proc, proc_settings}; use serde::{Deserialize, Serialize}; @@ -198,26 +198,12 @@ impl InjProc { let _ = next_transaction.get_or_insert_with(|| adaptor.build_transaction()); } InternalMsg::Config(config) => { - let settings = match config.get_proc::(self.proc.as_ref()) { - Ok(settings) => settings, - Err(err) => { - warn!("Can't reload settings for processor {}: {err}", self.name()); - return Ok(()); - } - }; - - if let Err(err) = - adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref())) + if let Some(settings) = + config.reload_proc::(self.proc.as_ref(), adaptor) { - warn!( - "Can't reload adaptor configuration for processor {}: {err}", - self.name() - ); - return Ok(()); + *regulator = settings.get_regulator(); + self.settings = settings; } - - *regulator = settings.get_regulator(); - self.settings = settings; } InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { diff --git a/prosa/src/io.rs b/prosa/src/io.rs index 49ffbcc..0dc44bb 100644 --- a/prosa/src/io.rs +++ b/prosa/src/io.rs @@ -554,11 +554,10 @@ mod tests { .expect("Certificate path should exist") .to_string(); - let mut server_ssl_config = SslConfig::new_self_cert(cert_path.clone()); - server_ssl_config.set_alpn(vec!["prosa/1".into(), "h2".into()]); - - let listener_settings = + let server_ssl_config = SslConfig::new_self_cert(cert_path.clone()); + let mut listener_settings = listener::ListenerSetting::new(addr.clone(), Some(server_ssl_config)); + listener_settings.set_alpn(vec!["prosa/1".into(), "h2".into()]); assert!( format!("{listener_settings:?}").contains("tls") && format!("{listener_settings:?}").contains("localhost") @@ -599,9 +598,9 @@ mod tests { }; let mut client_ssl_config = SslConfig::default(); - client_ssl_config.set_alpn(vec!["http/1.1".into(), "prosa/1".into()]); client_ssl_config.set_store(Store::File { path: cert_path }); - let target_settings = stream::TargetSetting::new(addr, Some(client_ssl_config), None); + let mut target_settings = stream::TargetSetting::new(addr, Some(client_ssl_config), None); + target_settings.set_alpn(vec!["http/1.1".into(), "prosa/1".into()]); assert_eq!(addr_str, target_settings.to_string()); let client = async { diff --git a/prosa/src/io/listener.rs b/prosa/src/io/listener.rs index a896fc9..48f409c 100644 --- a/prosa/src/io/listener.rs +++ b/prosa/src/io/listener.rs @@ -364,16 +364,12 @@ impl From for StreamListener { /// Ok(()) /// } /// ``` -#[derive(Deserialize, Serialize, Clone)] +#[derive(Deserialize, Serialize, Clone, PartialEq)] pub struct ListenerSetting { /// Url of the listening pub url: Url, - /// SSL configuration for target destination - pub ssl: Option, - #[cfg(feature = "openssl")] - #[serde(skip)] - /// OpenSSL configuration for target destination - openssl_context: Option<::openssl::ssl::SslAcceptor>, + /// SSL configuration of the listener + ssl: Option, #[serde(skip_serializing)] #[serde(default = "ListenerSetting::default_max_socket")] /// Maximum number of socket @@ -400,19 +396,57 @@ impl ListenerSetting { /// Method to create manually a target pub fn new(url: Url, ssl: Option) -> ListenerSetting { - #[allow(unused_mut)] - let mut target = ListenerSetting { - url: url.clone(), + ListenerSetting { + url, ssl, - #[cfg(feature = "openssl")] - openssl_context: None, max_socket: Self::default_max_socket(), - }; + } + } - #[cfg(feature = "openssl")] - target.init_ssl_context(url.host_str()); + /// Method to know if the listener will accept SSL connections + /// + /// ``` + /// use url::Url; + /// use prosa::io::listener::ListenerSetting; + /// + /// assert!(ListenerSetting::from(Url::parse("https://[::]:4443").unwrap()).is_ssl()); + /// assert!(!ListenerSetting::from(Url::parse("tcp://[::]:8080").unwrap()).is_ssl()); + /// ``` + pub fn is_ssl(&self) -> bool { + self.ssl.is_some() || url_is_ssl(&self.url) + } + + /// Getter of the SSL configuration of the listener + pub fn ssl(&self) -> Option<&SslConfig> { + self.ssl.as_ref() + } - target + /// Setter of the SSL configuration of the listener + pub fn set_ssl(&mut self, ssl: Option) { + self.ssl = ssl; + } + + /// Method to set the ALPN protocols to negotiate. + /// A default SSL configuration is created if the listener accepts SSL connections but none was configured. + /// + /// Nothing is done for a plain listener. Idempotent, so it can be called on every configuration reload. + /// + /// ``` + /// use url::Url; + /// use prosa::io::listener::ListenerSetting; + /// + /// let mut ssl_listener = ListenerSetting::from(Url::parse("https://[::]:4443").unwrap()); + /// ssl_listener.set_alpn(vec!["h2".into()]); + /// assert!(ssl_listener.ssl().is_some()); + /// + /// let mut plain_listener = ListenerSetting::from(Url::parse("tcp://[::]:8080").unwrap()); + /// plain_listener.set_alpn(vec!["h2".into()]); + /// assert!(plain_listener.ssl().is_none()); + /// ``` + pub fn set_alpn(&mut self, alpn: Vec) { + if self.is_ssl() { + self.ssl.get_or_insert_default().set_alpn(alpn); + } } /// Return a borrowed safe view of the listener URL. @@ -424,17 +458,6 @@ impl ListenerSetting { get_safe_url(&self.url) } - #[cfg(feature = "openssl")] - /// Method to init the ssl context out of the ssl target configuration. - /// Must be call when the configuration is retrieved - pub fn init_ssl_context(&mut self, domain: Option<&str>) { - if let Some(ssl_config) = self.ssl.as_ref() { - let ssl_acceptor_builder: Option<::openssl::ssl::SslAcceptorBuilder> = - ssl_config.init_tls_server_context(domain).ok(); - self.openssl_context = ssl_acceptor_builder.map(|a| a.build()); - } - } - /// Method to connect a ProSA stream to the remote target using the configuration pub async fn bind(&self) -> Result { #[cfg(target_family = "unix")] @@ -447,59 +470,34 @@ impl ListenerSetting { #[allow(unused_mut)] let mut stream_listener = StreamListener::bind(&*addrs).await?; - #[cfg(feature = "openssl")] - if let Some(ssl_acceptor) = &self.openssl_context { - stream_listener = stream_listener.ssl_acceptor( - ssl_acceptor.clone(), - self.ssl.as_ref().map(|c| c.get_ssl_timeout()), - ); - return Ok(stream_listener); - } - - if let Some(_ssl_config) = self.ssl.as_ref() { - #[cfg(feature = "openssl")] - { - let ssl_acceptor_builder_result: Result< - ::openssl::ssl::SslAcceptorBuilder, - prosa_utils::config::ConfigError, - > = _ssl_config.init_tls_server_context(self.url.host_str()); - if let Ok(ssl_acceptor_builder) = ssl_acceptor_builder_result { - stream_listener = stream_listener.ssl_acceptor( - ssl_acceptor_builder.build(), - Some(_ssl_config.get_ssl_timeout()), - ); - return Ok(stream_listener); - } - } - - Err(io::Error::new( - io::ErrorKind::Unsupported, - "No SSL engine available", - )) - } else if url_is_ssl(&self.url) { + if self.is_ssl() { #[cfg(feature = "openssl")] { - let ssl_config = SslConfig::default(); - let ssl_acceptor_builder_result: Result< - ::openssl::ssl::SslAcceptorBuilder, - prosa_utils::config::ConfigError, - > = ssl_config.init_tls_server_context(self.url.host_str()); - if let Ok(ssl_acceptor_builder) = ssl_acceptor_builder_result { - stream_listener = stream_listener.ssl_acceptor( - ssl_acceptor_builder.build(), - Some(ssl_config.get_ssl_timeout()), - ); - return Ok(stream_listener); - } + let ssl_config = self.ssl.clone().unwrap_or_default(); + let ssl_timeout = ssl_config.get_ssl_timeout(); + let host = self.url.host_str().map(String::from); + + // Reading the certificates blocks, so it's done on the blocking pool + let ssl_acceptor = tokio::task::spawn_blocking(move || { + ssl_config + .init_tls_server_context(host.as_deref()) + .map(|ssl_context_builder| ssl_context_builder.build()) + }) + .await + .map_err(io::Error::other)? + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + stream_listener = stream_listener.ssl_acceptor(ssl_acceptor, Some(ssl_timeout)); + return Ok(stream_listener); } - Err(io::Error::new( + #[cfg(not(feature = "openssl"))] + return Err(io::Error::new( io::ErrorKind::Unsupported, "No SSL engine available", - )) - } else { - Ok(stream_listener) + )); } + + Ok(stream_listener) } } @@ -508,8 +506,6 @@ impl From for ListenerSetting { ListenerSetting { url, ssl: None, - #[cfg(feature = "openssl")] - openssl_context: None, max_socket: Self::default_max_socket(), } } @@ -570,4 +566,74 @@ mod tests { setting.to_string() ); } + + #[test] + fn listener_setting_partial_eq() { + let config = "url = \"https://localhost:4443\"\nssl = { alpn = [\"h2\"] }\n"; + let setting: ListenerSetting = + toml::from_str(config).expect("Listener settings should deserialize"); + + assert_eq!( + setting, + toml::from_str::(config) + .expect("Listener settings should deserialize") + ); + assert_ne!( + setting, + toml::from_str::( + "url = \"https://localhost:4444\"\nssl = { alpn = [\"h2\"] }\n" + ) + .expect("Listener settings should deserialize") + ); + assert_ne!( + setting, + toml::from_str::( + "url = \"https://localhost:4443\"\nssl = { alpn = [\"http/1.1\"] }\n" + ) + .expect("Listener settings should deserialize") + ); + + // A programmatically built listener matches its configured counterpart + let mut built = ListenerSetting::new( + Url::parse("https://localhost:4443").expect("Listener url is invalid"), + None, + ); + built.set_alpn(vec!["h2".into()]); + assert_eq!(setting, built); + } + + #[test] + fn listener_setting_set_alpn() { + let mut expected_ssl = SslConfig::default(); + expected_ssl.set_alpn(vec!["h2".into()]); + + // An SSL url without SSL configuration gets a default one + let mut ssl_url = ListenerSetting::from( + Url::parse("https://[::]:4443").expect("Listener url is invalid"), + ); + ssl_url.set_alpn(vec!["h2".into()]); + assert_eq!(Some(&expected_ssl), ssl_url.ssl()); + + // Idempotent, so a configuration reload doesn't rebind + let reloaded = { + let mut reloaded = ssl_url.clone(); + reloaded.set_alpn(vec!["h2".into()]); + reloaded + }; + assert_eq!(ssl_url, reloaded); + + // A plain listener has nothing to negotiate + let mut plain_url = + ListenerSetting::from(Url::parse("tcp://[::]:8080").expect("Listener url is invalid")); + plain_url.set_alpn(vec!["h2".into()]); + assert_eq!(None, plain_url.ssl()); + + // But a plain url with an explicit SSL configuration does use SSL + let mut plain_url_with_ssl = ListenerSetting::new( + Url::parse("tcp://[::]:8080").expect("Listener url is invalid"), + Some(SslConfig::default()), + ); + plain_url_with_ssl.set_alpn(vec!["h2".into()]); + assert_eq!(Some(&expected_ssl), plain_url_with_ssl.ssl()); + } } diff --git a/prosa/src/io/stream.rs b/prosa/src/io/stream.rs index a3a819e..c2239f1 100644 --- a/prosa/src/io/stream.rs +++ b/prosa/src/io/stream.rs @@ -704,18 +704,14 @@ impl From> for Stream { /// Ok(()) /// } /// ``` -#[derive(Deserialize, Serialize, Clone)] +#[derive(Deserialize, Serialize, Clone, PartialEq)] pub struct TargetSetting { /// Url of the target destination pub url: Url, /// SSL configuration for target destination - pub ssl: Option, + ssl: Option, /// Optional proxy use to reach the target pub proxy: Option, - #[cfg(feature = "openssl")] - #[serde(skip)] - /// OpenSSL configuration for target destination - openssl_context: Option<::openssl::ssl::SslConnector>, #[serde(skip_serializing)] #[serde(default = "TargetSetting::get_default_connect_timeout")] /// Timeout for socket connection in milliseconds @@ -729,29 +725,52 @@ impl TargetSetting { /// Method to create manually a target pub fn new(url: Url, ssl: Option, proxy: Option) -> TargetSetting { - let mut target = TargetSetting { + TargetSetting { url, ssl, proxy, - #[cfg(feature = "openssl")] - openssl_context: None, connect_timeout: Self::get_default_connect_timeout(), - }; - - target.init_ssl_context(); - target + } } /// Method to know if the target will be connected with SSL pub fn is_ssl(&self) -> bool { - #[cfg(feature = "openssl")] - if self.openssl_context.is_some() { - return true; - } - self.ssl.is_some() || url_is_ssl(&self.url) } + /// Getter of the SSL configuration of the target + pub fn ssl(&self) -> Option<&SslConfig> { + self.ssl.as_ref() + } + + /// Setter of the SSL configuration of the target + pub fn set_ssl(&mut self, ssl: Option) { + self.ssl = ssl; + } + + /// Method to set the ALPN protocols to negotiate. + /// A default SSL configuration is created if the target is reached over SSL but none was configured. + /// + /// Nothing is done for a plain target. Idempotent, so it can be called on every configuration reload. + /// + /// ``` + /// use url::Url; + /// use prosa::io::stream::TargetSetting; + /// + /// let mut ssl_target = TargetSetting::from(Url::parse("https://worldline.com").unwrap()); + /// ssl_target.set_alpn(vec!["h2".into()]); + /// assert!(ssl_target.ssl().is_some()); + /// + /// let mut plain_target = TargetSetting::from(Url::parse("tcp://worldline.com:80").unwrap()); + /// plain_target.set_alpn(vec!["h2".into()]); + /// assert!(plain_target.ssl().is_none()); + /// ``` + pub fn set_alpn(&mut self, alpn: Vec) { + if self.is_ssl() { + self.ssl.get_or_insert_default().set_alpn(alpn); + } + } + /// Return a borrowed safe view of the target URL. /// /// Formatting the [`SafeUrl`] masks credentials and omits the query and fragment without @@ -779,18 +798,6 @@ impl TargetSetting { url_authentication(&self.url) } - /// Method to init the ssl context out of the ssl target configuration. - /// Must be call when the configuration is retrieved - pub fn init_ssl_context(&mut self) { - #[cfg(feature = "openssl")] - if let Some(ssl_config) = self.ssl.as_ref() { - // Init OpenSSL context by default - let ssl_context_builder: Option = - SslConfigContext::init_tls_client_context(ssl_config).ok(); - self.openssl_context = ssl_context_builder.map(|c| c.build()); - } - } - /// Method to connect a ProSA stream to the remote target using the configuration pub async fn connect(&self) -> Result { #[cfg(target_family = "unix")] @@ -808,22 +815,32 @@ impl TargetSetting { })?; } + // Built for every connection so a certificate rotated on disk applies to the next one. + // Reading the certificates blocks, so it's done on the blocking pool. #[cfg(feature = "openssl")] - let openssl_context = if self.openssl_context.is_some() { - self.openssl_context.clone() - } else if let Some(ssl_config) = &self.ssl { - let ssl_context_builder: openssl::ssl::SslConnectorBuilder = - SslConfigContext::init_tls_client_context(ssl_config)?; - Some(ssl_context_builder.build()) - } else if url_is_ssl(&self.url) { - let ssl_config = SslConfig::default(); - let ssl_context_builder: openssl::ssl::SslConnectorBuilder = - SslConfigContext::init_tls_client_context(&ssl_config)?; - Some(ssl_context_builder.build()) + let openssl_context = if self.is_ssl() { + let ssl_config = self.ssl.clone().unwrap_or_default(); + Some( + tokio::task::spawn_blocking(move || { + let ssl_context_builder: openssl::ssl::SslConnectorBuilder = + SslConfigContext::init_tls_client_context(&ssl_config)?; + Ok::<_, io::Error>(ssl_context_builder.build()) + }) + .await + .map_err(io::Error::other)??, + ) } else { None }; + #[cfg(not(feature = "openssl"))] + if self.is_ssl() { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "No SSL engine available", + )); + } + if let Some(proxy_url) = &self.proxy { if proxy_url.scheme() == "http" { #[cfg(feature = "http-proxy")] @@ -921,8 +938,6 @@ impl From for TargetSetting { url, ssl: None, proxy: None, - #[cfg(feature = "openssl")] - openssl_context: None, connect_timeout: Self::get_default_connect_timeout(), } } @@ -1041,4 +1056,76 @@ mod tests { format!("{target_with_token:?}") ); } + + #[test] + fn target_setting_partial_eq() { + let config = "url = \"https://localhost:4443/v1\"\nssl = { alpn = [\"h2\"] }\n"; + let setting: TargetSetting = + toml::from_str(config).expect("Target settings should deserialize"); + + assert_eq!( + setting, + toml::from_str::(config).expect("Target settings should deserialize") + ); + assert_ne!( + setting, + toml::from_str::( + "url = \"https://localhost:4444/v1\"\nssl = { alpn = [\"h2\"] }\n" + ) + .expect("Target settings should deserialize") + ); + assert_ne!( + setting, + toml::from_str::( + "url = \"https://localhost:4443/v1\"\nssl = { alpn = [\"http/1.1\"] }\n" + ) + .expect("Target settings should deserialize") + ); + + // A programmatically built target matches its configured counterpart + let mut built = TargetSetting::new( + Url::parse("https://localhost:4443/v1").expect("Target url is invalid"), + None, + None, + ); + built.set_alpn(vec!["h2".into()]); + assert_eq!(setting, built); + } + + #[test] + fn target_setting_set_alpn() { + let mut expected_ssl = SslConfig::default(); + expected_ssl.set_alpn(vec!["h2".into()]); + + // An SSL url without SSL configuration gets a default one + let mut ssl_url = TargetSetting::from( + Url::parse("https://worldline.com").expect("Target url is invalid"), + ); + ssl_url.set_alpn(vec!["h2".into()]); + assert_eq!(Some(&expected_ssl), ssl_url.ssl()); + + // Idempotent, so a configuration reload doesn't reconnect + let reloaded = { + let mut reloaded = ssl_url.clone(); + reloaded.set_alpn(vec!["h2".into()]); + reloaded + }; + assert_eq!(ssl_url, reloaded); + + // A plain target has nothing to negotiate + let mut plain_url = TargetSetting::from( + Url::parse("tcp://worldline.com:80").expect("Target url is invalid"), + ); + plain_url.set_alpn(vec!["h2".into()]); + assert_eq!(None, plain_url.ssl()); + + // But a plain url with an explicit SSL configuration does use SSL + let mut plain_url_with_ssl = TargetSetting::new( + Url::parse("tcp://worldline.com:80").expect("Target url is invalid"), + Some(SslConfig::default()), + None, + ); + plain_url_with_ssl.set_alpn(vec!["h2".into()]); + assert_eq!(Some(&expected_ssl), plain_url_with_ssl.ssl()); + } } diff --git a/prosa/src/stub/proc.rs b/prosa/src/stub/proc.rs index 8e09832..7d760d5 100644 --- a/prosa/src/stub/proc.rs +++ b/prosa/src/stub/proc.rs @@ -1,6 +1,6 @@ use std::{collections::HashSet, sync::Arc}; -use crate::tracing::{debug, info, warn}; +use crate::tracing::{debug, info}; use prosa_macros::proc_settings; use serde::{Deserialize, Serialize}; @@ -118,50 +118,37 @@ where err ), InternalMsg::Config(config) => { - let settings = match config.get_proc::(self.proc.as_ref()) { - Ok(settings) => settings, - Err(err) => { - warn!("Can't reload settings for processor {}: {err}", self.name()); - continue; - } - }; - - if let Err(err) = - adaptor.reload_config(config.get_adaptor_config(self.proc.as_ref())) + if let Some(settings) = + config.reload_proc::(self.proc.as_ref(), adaptor.as_ref()) { - warn!( - "Can't reload adaptor configuration for processor {}: {err}", - self.name() - ); - continue; - } - - let current_services = - self.settings.service_names.iter().collect::>(); - let new_services = settings.service_names.iter().collect::>(); + let current_services = + self.settings.service_names.iter().collect::>(); + let new_services = + settings.service_names.iter().collect::>(); + + let services_to_remove = current_services + .difference(&new_services) + .map(|service| (*service).clone()) + .collect::>(); + if !services_to_remove.is_empty() { + self.proc.remove_service_proc(services_to_remove).await?; + } - let services_to_remove = current_services - .difference(&new_services) - .map(|service| (*service).clone()) - .collect::>(); - if !services_to_remove.is_empty() { - self.proc.remove_service_proc(services_to_remove).await?; - } + let services_to_add = new_services + .difference(¤t_services) + .map(|service| (*service).clone()) + .collect::>(); + if !services_to_add.is_empty() { + self.proc.add_service_proc(services_to_add).await?; + } - let services_to_add = new_services - .difference(¤t_services) - .map(|service| (*service).clone()) - .collect::>(); - if !services_to_add.is_empty() { - self.proc.add_service_proc(services_to_add).await?; + info!( + "{} reloaded settings for services: {}", + self.name(), + settings.service_names.join(", ") + ); + self.settings = settings; } - - info!( - "{} reloaded settings for services: {}", - self.name(), - settings.service_names.join(", ") - ); - self.settings = settings; } InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { diff --git a/prosa_book/src/ch01-02-03-stream.md b/prosa_book/src/ch01-02-03-stream.md index eec4e9d..083c7df 100644 --- a/prosa_book/src/ch01-02-03-stream.md +++ b/prosa_book/src/ch01-02-03-stream.md @@ -49,3 +49,27 @@ Target and proxy URLs are redacted when settings or connection errors are format are masked, while query parameters and fragments are omitted. URL paths are preserved and should not contain secrets. `TargetSetting::get_safe_url()` returns the same borrowed `SafeUrl` view, so the caller chooses whether to format it or convert it into either form of owned sanitized URL. + +## SSL configuration of a listener or a target + +Read the SSL configuration with `ssl()`, and change it with `set_ssl()` or `set_alpn()`. + +`set_alpn()` creates a default SSL configuration when the listener or the target uses SSL but has +none, and does nothing on a plain one. `is_ssl()` tells whether SSL applies: an SSL configuration +**or** an SSL URL scheme is enough, so a plain `tcp://` URL with an explicit `ssl` block does +negotiate ALPN. + +The OpenSSL context is built from that configuration every time a listener binds or a target +connects, so a certificate or a CA rotated on disk applies to the next bind or connection without +needing a configuration change. + +Both settings implement `PartialEq`. On a configuration reload, compare the new settings with the +current ones and rebind or reconnect only when they differ. `set_alpn()` is idempotent, so +normalise before comparing: + +```rust,ignore +listener_setting.set_alpn(vec!["h2".into()]); +if listener_setting != self.settings.listener { + // ... rebind +} +``` diff --git a/prosa_book/src/ch03-01-settings.md b/prosa_book/src/ch03-01-settings.md index 8084207..9b04980 100644 --- a/prosa_book/src/ch03-01-settings.md +++ b/prosa_book/src/ch03-01-settings.md @@ -18,7 +18,7 @@ let settings = config.try_deserialize::()?; `ProsaConfig::from_path()` uses the same configuration loading rules as ProSA itself: the path can be a single configuration file or a directory recursively containing `yml`, `yaml`, or `toml` files, and `PROSA_*` environment variables are applied on top of the file sources. -When the main task notifies a processor about a configuration change, the message contains the same `ProsaConfig` wrapper. Processors should reload their own section with `config.get_proc(self.proc.as_ref())?`. +When the main task notifies a processor about a configuration change, the message contains the same `ProsaConfig` wrapper. Processors should reload their own section with `config.reload_proc::(self.proc.as_ref(), &adaptor)`, which deserializes the processor settings and reloads the adaptor configuration in one step. If the processor section has `adaptor_config_path`, `ProsaConfig` also loads that adaptor configuration and watches it as part of the global configuration reload flow. ## Creation diff --git a/prosa_book/src/ch03-02-creation.md b/prosa_book/src/ch03-02-creation.md index 91dfb5c..6fe74fd 100644 --- a/prosa_book/src/ch03-02-creation.md +++ b/prosa_book/src/ch03-02-creation.md @@ -58,7 +58,12 @@ where // TODO: process the error } InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + if let Some(settings) = + config.reload_proc::(self.proc.as_ref(), &adaptor) + { + // TODO: apply the difference between `settings` and `self.settings` + self.settings = settings; + } }, InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { @@ -73,7 +78,9 @@ where } ``` -When receiving `InternalMsg::Config(config)`, call `config.get_proc(self.proc.as_ref())?` to deserialize the section matching the processor configuration key (the processor name with `-` replaced by `_`) into the processor settings type. The main task only sends this message to processors whose own configuration section changed. +When receiving `InternalMsg::Config(config)`, call `config.reload_proc::(self.proc.as_ref(), &adaptor)` to deserialize the section matching the processor configuration key (the processor name with `-` replaced by `_`) into the processor settings type, and reload the adaptor configuration in the same step. It returns `None` if either fails, so the processor keeps running on its current configuration. A processor that has no configuration section keeps the settings it was created with, which is reported at debug level; anything else is logged as a warning. + +The main task sends this message once when the processor registers, then on every reload that changes its own configuration section. So compare the new settings with the ones the processor already holds, and apply only the difference. The generic parameter `A` represents the adaptor type your processor uses. Specify in the _where_ clause which traits your adaptor must implement (commonly, [`Adaptor`](https://docs.rs/prosa/latest/prosa/core/adaptor/trait.Adaptor.html) plus `Send` and `Sync`) diff --git a/prosa_book/src/ch03-05-service.md b/prosa_book/src/ch03-05-service.md index b1d1282..262fbfe 100644 --- a/prosa_book/src/ch03-05-service.md +++ b/prosa_book/src/ch03-05-service.md @@ -41,8 +41,8 @@ To start listening to a specific service, call [`add_service_proc()`](https://do InternalMsg::Error(err) => { // Handle errors as if they were responses }, - InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + InternalMsg::Config(_config) => { + // Reload the settings, see the processor creation chapter }, InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { @@ -108,8 +108,8 @@ In this case, you can declare multiple listener subtasks, each of which subscrib InternalMsg::Error(err) => { // Handle errors as if they were responses }, - InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + InternalMsg::Config(_config) => { + // Reload the settings, see the processor creation chapter }, InternalMsg::Service(table) => service = table, InternalMsg::Shutdown => { @@ -133,8 +133,8 @@ In this case, you can declare multiple listener subtasks, each of which subscrib InternalMsg::Error(err) => { // Handle errors as if they were responses }, - InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + InternalMsg::Config(_config) => { + // Reload the settings, see the processor creation chapter }, InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { @@ -181,8 +181,8 @@ After that, you are free to call any services. InternalMsg::Error(err) => { // Handle errors }, - InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + InternalMsg::Config(_config) => { + // Reload the settings, see the processor creation chapter }, InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { @@ -248,8 +248,8 @@ The logic is similar to single senders, but you specify the queue when sending m InternalMsg::Error(err) => { // Handle errors for this subtask }, - InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + InternalMsg::Config(_config) => { + // Reload the settings, see the processor creation chapter }, InternalMsg::Service(table) => self.service = table, InternalMsg::Shutdown => { diff --git a/prosa_book/src/ch03-06-events.md b/prosa_book/src/ch03-06-events.md index fe3e7b3..e9f5141 100644 --- a/prosa_book/src/ch03-06-events.md +++ b/prosa_book/src/ch03-06-events.md @@ -15,7 +15,7 @@ There are three important methods you need to use for this object: - [`pull()`](https://docs.rs/prosa/latest/prosa/event/pending/struct.PendingMsgs.html#method.pull) Async method to retrieve all messages that have expired (timed out). ```rust,ignore -# #[proc] +# #[proc(settings = MyProcSettings)] # struct MyProc {} # # #[proc] @@ -23,10 +23,7 @@ There are three important methods you need to use for this object: # where # A: Default + Adaptor + std::marker::Send + std::marker::Sync, # { - async fn internal_run( - &mut self, - _name: String, - ) -> Result<(), Box> { + async fn internal_run(&mut self) -> Result<(), Box> { let mut adaptor = A::default(); self.proc.add_proc().await?; self.proc @@ -55,7 +52,11 @@ There are three important methods you need to use for this object: info!("Proc {} receive an error: {:?}", self.get_proc_id(), err); }, InternalMsg::Config(config) => { - self.settings = config.get_proc(self.proc.as_ref())?; + if let Some(settings) = config + .reload_proc::(self.proc.as_ref(), &adaptor) + { + self.settings = settings; + } }, InternalMsg::Service(table) => { debug!("New service table received:\n{}\n", table); @@ -77,7 +78,7 @@ There are three important methods you need to use for this object: } } } -} +# } ``` ## Regulator - `Regulator` From 630a54bf91f95fa3db16c6f1387947dc06c1fd7d Mon Sep 17 00:00:00 2001 From: Jeremy HERGAULT Date: Tue, 18 Aug 2026 18:04:51 +0200 Subject: [PATCH 2/3] feat: listener don't rebind if ssl change Signed-off-by: Jeremy HERGAULT --- prosa/src/io/listener.rs | 788 ++++++++++++++++++++++++---- prosa_book/src/ch01-02-03-stream.md | 37 +- prosa_book/src/ch03-07-io.md | 48 ++ 3 files changed, 761 insertions(+), 112 deletions(-) diff --git a/prosa/src/io/listener.rs b/prosa/src/io/listener.rs index 48f409c..9173a65 100644 --- a/prosa/src/io/listener.rs +++ b/prosa/src/io/listener.rs @@ -18,6 +18,104 @@ use url::Url; use super::{SafeUrl, SocketAddr, get_safe_url, stream::Stream, url_is_ssl}; +#[cfg(feature = "openssl")] +/// SSL parameters a listener serves to the clients it accepts. +/// +/// Cheap to clone, because an OpenSSL context is reference counted, so an accept loop can hand one +/// to the task that handshakes a client and go straight back to accepting. A client is served the +/// parameters the listener held when it was accepted. +/// +/// ``` +/// use tokio::io; +/// use prosa::io::listener::{ListenerSetting, StreamListener}; +/// +/// async fn accepting(setting: &ListenerSetting) -> Result<(), io::Error> { +/// let stream_listener: StreamListener = setting.bind().await?; +/// +/// loop { +/// let (stream, addr) = stream_listener.accept_raw().await?; +/// +/// // Owned snapshot of the SSL parameters, so the accept loop never carries the handshake +/// let handshaker = stream_listener.handshaker().cloned(); +/// tokio::spawn(async move { +/// let stream = match handshaker { +/// Some(handshaker) => handshaker.handshake(stream).await?, +/// None => stream, +/// }; +/// +/// // Handle the stream like any tokio stream +/// Ok::<_, io::Error>(()) +/// }); +/// } +/// } +/// ``` +#[derive(Clone)] +pub struct SslHandshaker { + /// Acceptor holding the certificate served to the clients + acceptor: ::openssl::ssl::SslAcceptor, + /// Timeout of the SSL handshake with a client + timeout: Duration, +} + +#[cfg(feature = "openssl")] +impl SslHandshaker { + /// Method to create the SSL parameters served by a listener. + /// By default, the SSL handshake timeout is 3 seconds + pub fn new(acceptor: ::openssl::ssl::SslAcceptor, timeout: Option) -> SslHandshaker { + SslHandshaker { + acceptor, + timeout: timeout.unwrap_or(StreamListener::DEFAULT_SSL_TIMEOUT), + } + } + + /// Getter of the timeout of the SSL handshake with a client + pub fn ssl_timeout(&self) -> Duration { + self.timeout + } + + /// Method to negotiate SSL with a client that has just been accepted. + /// A stream that is not a plain TCP one is returned as is + pub async fn handshake(&self, stream: Stream) -> Result { + let Stream::Tcp(tcp_stream) = stream else { + return Ok(stream); + }; + + let ssl = openssl::ssl::Ssl::new(self.acceptor.context()) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + let mut stream = tokio_openssl::SslStream::new(ssl, tcp_stream) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + if let Err(e) = tokio::time::timeout(self.timeout, std::pin::Pin::new(&mut stream).accept()) + .await + .map_err(|_| { + io::Error::new( + io::ErrorKind::TimedOut, + format!( + "SSL timeout[{} ms] for {stream:?}", + self.timeout.as_millis() + ), + ) + })? + { + return Err(io::Error::other(format!("Can't accept the client: {e}"))); + } + + Ok(Stream::OpenSsl(stream)) + } +} + +#[cfg(feature = "openssl")] +impl fmt::Debug for SslHandshaker { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SslHandshaker") + .field("ssl_timeout", &self.timeout) + .field( + "certificate", + &self.acceptor.context().certificate().map(|c| c.to_text()), + ) + .finish() + } +} + /// ProSA socket object to handle TCP/SSL server socket pub enum StreamListener { #[cfg(target_family = "unix")] @@ -26,8 +124,11 @@ pub enum StreamListener { /// TCP server socket Tcp(TcpListener), #[cfg(feature = "openssl")] - /// OpenSSL server socket - OpenSsl(TcpListener, ::openssl::ssl::SslAcceptor, Duration), + /// OpenSSL server socket. + /// + /// The SSL parameters are held apart from the socket so + /// [`StreamListener::set_handshaker`] can serve new ones on the socket it is already bound to + OpenSsl(TcpListener, SslHandshaker), } impl fmt::Debug for StreamListener { @@ -37,14 +138,10 @@ impl fmt::Debug for StreamListener { StreamListener::Unix(l) => f.debug_struct("Unix").field("listener", &l).finish(), StreamListener::Tcp(l) => f.debug_struct("Tcp").field("listener", &l).finish(), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, a, t) => f + StreamListener::OpenSsl(l, ssl) => f .debug_struct("Ssl") .field("listener", &l) - .field("ssl_timeout", &t) - .field( - "certificate", - &a.context().certificate().map(|c| c.to_text()), - ) + .field("ssl", &ssl) .finish(), } } @@ -80,9 +177,7 @@ impl StreamListener { StreamListener::Unix(listener) => listener.local_addr().map(|addr| addr.into()), StreamListener::Tcp(listener) => listener.local_addr().map(|addr| addr.into()), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(listener, _, _) => { - listener.local_addr().map(|addr| addr.into()) - } + StreamListener::OpenSsl(listener, _) => listener.local_addr().map(|addr| addr.into()), } } @@ -143,18 +238,71 @@ impl StreamListener { ssl_acceptor: ::openssl::ssl::SslAcceptor, ssl_timeout: Option, ) -> StreamListener { + self.set_handshaker(Some(SslHandshaker::new(ssl_acceptor, ssl_timeout))) + } + + #[cfg(feature = "openssl")] + /// Getter of the SSL parameters served to a client accepted now, [`None`] on a plain listener. + /// + /// Clone it into the task that handshakes the client, so the accept loop can go straight back + /// to accepting. + /// + /// These are the parameters the listener holds, and the ones [`StreamListener::accept`] and + /// [`StreamListener::handshake`] serve. Rotating a copy taken from here doesn't rotate them, so + /// a listener that can't be replaced by [`StreamListener::set_handshaker`] keeps serving the + /// superseded certificate. Bind such a listener with [`ListenerSetting::bind_raw`] instead, so + /// the handshaker held beside it is the only one + pub fn handshaker(&self) -> Option<&SslHandshaker> { match self { - StreamListener::Tcp(listener) => StreamListener::OpenSsl( - listener, - ssl_acceptor, - ssl_timeout.unwrap_or(Self::DEFAULT_SSL_TIMEOUT), - ), - StreamListener::OpenSsl(listener, _, _) => StreamListener::OpenSsl( - listener, - ssl_acceptor, - ssl_timeout.unwrap_or(Self::DEFAULT_SSL_TIMEOUT), - ), - _ => self, + StreamListener::OpenSsl(_l, handshaker) => Some(handshaker), + _ => None, + } + } + + #[cfg(feature = "openssl")] + /// Method to serve new SSL parameters on the socket that is already bound. + /// + /// The socket is moved into the returned listener rather than bound again, so the port is + /// never released and no client is refused: rotating a certificate, turning SSL on and turning + /// it off all keep the same socket. The connections that are established, and the ones in the + /// middle of their handshake, keep the parameters they started with; only the clients accepted + /// afterwards are served the new ones. + /// + /// A Unix socket never serves SSL and is returned as is. + /// + /// The listener has to be owned to be replaced, so this is for a listener a single task holds. + /// One that is shared should be bound with [`ListenerSetting::bind_raw`] and rotated through + /// the handshaker held beside it, otherwise the copy the listener keeps is the one it serves. + /// + /// ``` + /// use tokio::io; + /// use prosa::io::listener::{ListenerSetting, StreamListener}; + /// + /// async fn rotating(setting: &ListenerSetting) -> Result<(), io::Error> { + /// let mut stream_listener: StreamListener = setting.bind().await?; + /// let addr = stream_listener.local_addr()?; + /// + /// // The certificate the configuration points at has been renewed. Built before the + /// // listener is touched, so a broken certificate leaves it serving the one it has + /// let handshaker = setting.build_handshaker().await?; + /// stream_listener = stream_listener.set_handshaker(handshaker); + /// + /// // Still the very same socket + /// assert_eq!(addr, stream_listener.local_addr()?); + /// + /// Ok(()) + /// } + /// ``` + pub fn set_handshaker(self, handshaker: Option) -> StreamListener { + let listener = match self { + StreamListener::Tcp(listener) | StreamListener::OpenSsl(listener, _) => listener, + #[cfg(target_family = "unix")] + unix_listener => return unix_listener, + }; + + match handshaker { + Some(handshaker) => StreamListener::OpenSsl(listener, handshaker), + None => StreamListener::Tcp(listener), } } @@ -189,30 +337,15 @@ impl StreamListener { StreamListener::Unix(l) => l.accept().await.map(|s| (Stream::Unix(s.0), s.1.into())), StreamListener::Tcp(l) => l.accept().await.map(|s| (Stream::Tcp(s.0), s.1.into())), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, ssl_acceptor, ssl_timeout) => { - let ssl = openssl::ssl::Ssl::new(ssl_acceptor.context()) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + StreamListener::OpenSsl(l, handshaker) => { let (stream, addr) = l.accept().await?; - let mut stream = tokio_openssl::SslStream::new(ssl, stream) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - if let Err(e) = - tokio::time::timeout(*ssl_timeout, std::pin::Pin::new(&mut stream).accept()) - .await - .map_err(|_| { - io::Error::new( - io::ErrorKind::TimedOut, - format!( - "SSL timeout[{} ms] for {:?}", - ssl_timeout.as_millis(), - stream - ), - ) - })? - { - return Err(io::Error::other(format!("Can't accept the client: {e}"))); - } - Ok((Stream::OpenSsl(stream), addr.into())) + // Read after the accept, so a client that connects once the certificate has been + // rotated is served the new one + handshaker + .handshake(Stream::Tcp(stream)) + .await + .map(|stream| (stream, addr.into())) } } } @@ -250,7 +383,7 @@ impl StreamListener { StreamListener::Unix(l) => l.accept().await.map(|s| (Stream::Unix(s.0), s.1.into())), StreamListener::Tcp(l) => l.accept().await.map(|s| (Stream::Tcp(s.0), s.1.into())), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, _ssl_acceptor, _ssl_timeout) => { + StreamListener::OpenSsl(l, _handshaker) => { l.accept().await.map(|s| (Stream::Tcp(s.0), s.1.into())) } } @@ -258,37 +391,12 @@ impl StreamListener { /// Method to do an handshake with a client after an accept (Do nothing if the handshake is already done) pub async fn handshake(&self, stream: Stream) -> Result { - match stream { - Stream::Tcp(tcp_stream) => match self { - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(_l, ssl_acceptor, ssl_timeout) => { - let ssl = openssl::ssl::Ssl::new(ssl_acceptor.context()) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let mut stream = tokio_openssl::SslStream::new(ssl, tcp_stream) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - if let Err(e) = - tokio::time::timeout(*ssl_timeout, std::pin::Pin::new(&mut stream).accept()) - .await - .map_err(|_| { - io::Error::new( - io::ErrorKind::TimedOut, - format!( - "SSL timeout[{} ms] for {:?}", - ssl_timeout.as_millis(), - stream - ), - ) - })? - { - return Err(io::Error::other(format!("Can't accept the client: {e}"))); - } - - Ok(Stream::OpenSsl(stream)) - } - _ => Ok(Stream::Tcp(tcp_stream)), - }, - s => Ok(s), + #[cfg(feature = "openssl")] + if let StreamListener::OpenSsl(_l, handshaker) = self { + return handshaker.handshake(stream).await; } + + Ok(stream) } } @@ -299,7 +407,7 @@ impl AsFd for StreamListener { StreamListener::Unix(l) => l.as_fd(), StreamListener::Tcp(l) => l.as_fd(), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, _, _) => l.as_fd(), + StreamListener::OpenSsl(l, _) => l.as_fd(), } } } @@ -311,7 +419,7 @@ impl AsRawFd for StreamListener { StreamListener::Unix(l) => l.as_raw_fd(), StreamListener::Tcp(l) => l.as_raw_fd(), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, _, _) => l.as_raw_fd(), + StreamListener::OpenSsl(l, _) => l.as_raw_fd(), } } } @@ -329,7 +437,7 @@ impl fmt::Display for StreamListener { StreamListener::Unix(_) => write!(f, "unix://{addr}"), StreamListener::Tcp(_) => write!(f, "tcp://{addr}"), #[cfg(feature = "openssl")] - StreamListener::OpenSsl(_, _, _) => write!(f, "ssl://{addr}"), + StreamListener::OpenSsl(_, _) => write!(f, "ssl://{addr}"), } } } @@ -458,8 +566,84 @@ impl ListenerSetting { get_safe_url(&self.url) } - /// Method to connect a ProSA stream to the remote target using the configuration - pub async fn bind(&self) -> Result { + /// Method to know if serving `other` needs a new socket, because it doesn't listen on the same + /// address. + /// + /// Only the host and the port are compared, so everything else is served on the socket that is + /// already bound, whether that rotates a certificate or turns SSL on or off. A scheme carries + /// SSL rather than an address, and `max_socket` is a cap the processor enforces rather than a + /// property of the socket, so neither of them ever needs a new one. + /// + /// ``` + /// use url::Url; + /// use prosa::io::listener::ListenerSetting; + /// + /// let plain = ListenerSetting::from(Url::parse("tcp://[::]:8080").unwrap()); + /// + /// // Adding SSL under the same URL is served on the socket that is already bound + /// let mut with_ssl = plain.clone(); + /// with_ssl.set_alpn(vec!["h2".into()]); + /// assert!(!plain.needs_rebind(&with_ssl)); + /// + /// // And so is turning SSL on with the scheme + /// let ssl_scheme = ListenerSetting::from(Url::parse("ssl://[::]:8080").unwrap()); + /// assert!(!plain.needs_rebind(&ssl_scheme)); + /// + /// // Listening somewhere else needs a new socket + /// let moved = ListenerSetting::from(Url::parse("tcp://[::]:8081").unwrap()); + /// assert!(plain.needs_rebind(&moved)); + /// ``` + pub fn needs_rebind(&self, other: &ListenerSetting) -> bool { + self.url.host() != other.url.host() + || self.url.port_or_known_default() != other.url.port_or_known_default() + } + + #[cfg(feature = "openssl")] + /// Method to build the SSL parameters this configuration serves, [`None`] when it listens + /// without SSL. + /// + /// The certificates are read again on every call, on the blocking pool, because [`SslConfig`] + /// holds their *paths*: a rotation that rewrites a file in place leaves the configuration equal + /// to what it was, so a caller has nothing to compare and should call this on every + /// configuration reload. + /// + /// Hand the result to [`StreamListener::set_handshaker`] to serve it without rebinding. It is + /// built before the listener is touched, so a broken certificate leaves the listener serving + /// the one it already has. + /// + /// A listener that is SSL through its URL scheme alone is served a default [`SslConfig`], which + /// signs a certificate of its own rather than reading one. That certificate is signed again on + /// every call, so such a listener serves a new identity on every configuration reload and a + /// client that pins it stops trusting it. Configure a certificate to serve a stable one. + pub async fn build_handshaker(&self) -> Result, io::Error> { + // A Unix socket never serves SSL, `bind` ignores the SSL configuration for it + #[cfg(target_family = "unix")] + if self.url.scheme() == "unix" || self.url.scheme() == "file" { + return Ok(None); + } + + if !self.is_ssl() { + return Ok(None); + } + + let ssl_config = self.ssl.clone().unwrap_or_default(); + let timeout = ssl_config.get_ssl_timeout(); + let host = self.url.host_str().map(String::from); + + let acceptor = tokio::task::spawn_blocking(move || { + ssl_config + .init_tls_server_context(host.as_deref()) + .map(|ssl_context_builder| ssl_context_builder.build()) + }) + .await + .map_err(io::Error::other)? + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + + Ok(Some(SslHandshaker::new(acceptor, Some(timeout)))) + } + + /// Bind the socket of the configuration, without SSL + async fn bind_socket(&self) -> Result { #[cfg(target_family = "unix")] if self.url.scheme() == "unix" || self.url.scheme() == "file" { return Ok(StreamListener::Unix(UnixListener::bind(self.url.path())?)); @@ -467,37 +651,77 @@ impl ListenerSetting { let addrs = self.url.socket_addrs(|| self.url.port_or_known_default())?; - #[allow(unused_mut)] - let mut stream_listener = StreamListener::bind(&*addrs).await?; + StreamListener::bind(&*addrs).await + } - if self.is_ssl() { - #[cfg(feature = "openssl")] - { - let ssl_config = self.ssl.clone().unwrap_or_default(); - let ssl_timeout = ssl_config.get_ssl_timeout(); - let host = self.url.host_str().map(String::from); - - // Reading the certificates blocks, so it's done on the blocking pool - let ssl_acceptor = tokio::task::spawn_blocking(move || { - ssl_config - .init_tls_server_context(host.as_deref()) - .map(|ssl_context_builder| ssl_context_builder.build()) - }) - .await - .map_err(io::Error::other)? - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - stream_listener = stream_listener.ssl_acceptor(ssl_acceptor, Some(ssl_timeout)); - return Ok(stream_listener); - } + #[cfg(feature = "openssl")] + /// Method to bind the socket of the configuration and build the SSL parameters to serve on it, + /// without attaching them to the listener. + /// + /// Use this when the listener has to be shared, held in an [`Arc`](std::sync::Arc) by an accept + /// loop and by the tasks that handshake its clients, so [`StreamListener::set_handshaker`] + /// can't replace it. Rotating then means replacing the [`SslHandshaker`] returned here, and + /// because the listener never holds one there is no second copy of it to go stale. + /// + /// The listener describes the socket, so it formats as `tcp://` and its + /// [`StreamListener::handshaker`] is [`None`] even though the clients accepted on it are handed + /// a certificate. Hand the handshaker to [`StreamListener::set_handshaker`] instead when the + /// listener is owned by a single task and should serve it itself. + /// + /// ``` + /// use std::sync::Arc; + /// use tokio::io; + /// use prosa::io::listener::{ListenerSetting, StreamListener}; + /// + /// async fn accepting(setting: &ListenerSetting) -> Result<(), io::Error> { + /// let (listener, handshaker) = setting.bind_raw().await?; + /// let listener = Arc::new(listener); + /// + /// // The socket carries no SSL parameters, the accept loop holds the only copy + /// assert!(listener.handshaker().is_none()); + /// + /// loop { + /// let (stream, addr) = listener.accept_raw().await?; + /// + /// let handshaker = handshaker.clone(); + /// tokio::spawn(async move { + /// let stream = match handshaker { + /// Some(handshaker) => handshaker.handshake(stream).await?, + /// None => stream, + /// }; + /// + /// // Handle the stream like any tokio stream + /// Ok::<_, io::Error>(()) + /// }); + /// } + /// } + /// ``` + pub async fn bind_raw(&self) -> Result<(StreamListener, Option), io::Error> { + // Built first, so a broken certificate doesn't take the port on its way out + let handshaker = self.build_handshaker().await?; - #[cfg(not(feature = "openssl"))] - return Err(io::Error::new( - io::ErrorKind::Unsupported, - "No SSL engine available", - )); + Ok((self.bind_socket().await?, handshaker)) + } + + /// Method to connect a ProSA stream to the remote target using the configuration + pub async fn bind(&self) -> Result { + #[cfg(feature = "openssl")] + { + let (stream_listener, handshaker) = self.bind_raw().await?; + Ok(stream_listener.set_handshaker(handshaker)) } - Ok(stream_listener) + #[cfg(not(feature = "openssl"))] + { + if self.is_ssl() { + return Err(io::Error::new( + io::ErrorKind::Unsupported, + "No SSL engine available", + )); + } + + self.bind_socket().await + } } } @@ -545,6 +769,15 @@ impl fmt::Display for ListenerSetting { mod tests { use super::*; + /// Path of a test file no other test, and no other test run, writes to + fn unique_test_path(name: &str) -> std::path::PathBuf { + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos(); + std::env::temp_dir().join(format!("{}-{timestamp}-{name}", std::process::id())) + } + #[test] fn listener_setting_display_redacts_url_secrets() { let mut setting = ListenerSetting::from( @@ -602,6 +835,347 @@ mod tests { assert_eq!(setting, built); } + #[cfg(feature = "openssl")] + fn served_certificate(listener: &StreamListener) -> Vec { + listener + .handshaker() + .expect("The listener should accept SSL connections") + .acceptor + .context() + .certificate() + .expect("The acceptor should serve a certificate") + .to_pem() + .expect("The certificate should be readable") + } + + #[cfg(feature = "openssl")] + #[tokio::test] + async fn listener_rotate_certificate_keeps_the_socket() { + // Port 0 so the test doesn't depend on one being free, and a default SSL configuration + // because it generates a self signed certificate, a new one on every build + let setting = ListenerSetting::from( + Url::parse("https://127.0.0.1:0").expect("Listener url is valid"), + ); + let mut listener = setting.bind().await.expect("The listener should bind"); + + let addr = listener.local_addr().expect("The listener should be bound"); + let certificate = served_certificate(&listener); + + let handshaker = setting + .build_handshaker() + .await + .expect("The certificate should be read"); + listener = listener.set_handshaker(handshaker); + + // The very same socket, so no client was refused and nothing could steal the port + assert_eq!( + addr, + listener.local_addr().expect("The listener should be bound") + ); + // And a different certificate, so the rotation really re-read it instead of comparing the + // configuration it was handed with the one it already had + assert_ne!(certificate, served_certificate(&listener)); + } + + #[cfg(feature = "openssl")] + #[tokio::test] + async fn listener_turns_ssl_on_and_off_on_the_same_socket() { + let plain = + ListenerSetting::from(Url::parse("tcp://127.0.0.1:0").expect("Listener url is valid")); + let ssl = ListenerSetting::from( + Url::parse("https://127.0.0.1:0").expect("Listener url is valid"), + ); + + let mut listener = plain.bind().await.expect("The listener should bind"); + let addr = listener.local_addr().expect("The listener should be bound"); + assert!(listener.handshaker().is_none()); + + // Turning SSL on reuses the socket that is already bound + listener = listener.set_handshaker( + ssl.build_handshaker() + .await + .expect("The certificate should be read"), + ); + assert!(!served_certificate(&listener).is_empty()); + assert_eq!( + addr, + listener.local_addr().expect("The listener should be bound") + ); + + // And so does turning it off + listener = listener.set_handshaker( + plain + .build_handshaker() + .await + .expect("A plain listener has no certificate to read"), + ); + assert!(listener.handshaker().is_none()); + assert_eq!( + addr, + listener.local_addr().expect("The listener should be bound") + ); + } + + #[cfg(feature = "openssl")] + #[tokio::test] + async fn listener_setting_build_handshaker() { + // A plain listener has no SSL parameters to serve + assert!( + ListenerSetting::from(Url::parse("tcp://127.0.0.1:0").expect("Listener url is valid")) + .build_handshaker() + .await + .expect("A plain listener has no certificate to read") + .is_none() + ); + + // And neither does a Unix socket, even with an explicit SSL configuration, because `bind` + // ignores it there + assert!( + ListenerSetting::new( + Url::parse("unix:///tmp/prosa_build_handshaker.sock") + .expect("Listener url is valid"), + Some(SslConfig::default()), + ) + .build_handshaker() + .await + .expect("A Unix listener has no certificate to read") + .is_none() + ); + + let handshaker = ListenerSetting::from( + Url::parse("https://127.0.0.1:0").expect("Listener url is valid"), + ) + .build_handshaker() + .await + .expect("The certificate should be read") + .expect("An SSL listener serves SSL parameters"); + assert_eq!( + SslConfig::default().get_ssl_timeout(), + handshaker.ssl_timeout() + ); + } + + #[cfg(feature = "openssl")] + fn peer_certificate(stream: &Stream) -> Vec { + let Stream::OpenSsl(ssl_stream) = stream else { + panic!("The client should be connected over SSL"); + }; + + ssl_stream + .ssl() + .peer_certificate() + .expect("The listener should serve a certificate") + .to_pem() + .expect("The certificate should be readable") + } + + #[cfg(feature = "openssl")] + #[tokio::test] + async fn listener_rotate_certificate_serves_the_new_one() -> io::Result<()> { + use prosa_utils::config::ssl::Store; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + let cert_path = unique_test_path("test_listener_rotate_certificate.pem") + .to_str() + .expect("Certificate path should exist") + .to_string(); + + // Regenerates a self signed certificate on every build and writes it there, so the client + // trusts whichever one the listener currently serves. Port 0 so the test doesn't depend on + // one being free + let setting = ListenerSetting::new( + Url::parse("https://localhost:0").expect("Listener url is valid"), + Some(SslConfig::new_self_cert(cert_path.clone())), + ); + let mut listener = setting.bind().await?; + let addr = listener.local_addr()?; + let url = + Url::parse(&format!("tls://localhost:{}", addr.port())).expect("Target url is valid"); + + let connect = || async { + let mut client_config = SslConfig::default(); + client_config.set_store(Store::File { + path: cert_path.clone(), + }); + let connector: ::openssl::ssl::SslConnectorBuilder = + client_config.init_tls_client_context()?; + + Stream::connect_openssl(&url, &connector.build()).await + }; + + let (served, connected) = futures_util::future::join(listener.accept(), connect()).await; + let (mut served, _) = served?; + let mut connected = connected?; + let certificate = peer_certificate(&connected); + + // The certificate the configuration points at is renewed while the listener accepts + let handshaker = setting.build_handshaker().await?; + listener = listener.set_handshaker(handshaker); + + // The very same socket, so no client was refused and nothing could steal the port + assert_eq!(addr, listener.local_addr()?); + + // The connection established before the rotation keeps working + connected.write_all(b"ProSA").await?; + let mut buf = [0; 5]; + served.read_exact(&mut buf).await?; + assert_eq!(&buf, b"ProSA"); + + // And a client connecting afterwards is served the new certificate + let (served, connected) = futures_util::future::join(listener.accept(), connect()).await; + served?; + assert_ne!(certificate, peer_certificate(&connected?)); + + Ok(()) + } + + #[cfg(feature = "openssl")] + #[tokio::test] + async fn listener_setting_bind_raw_leaves_the_ssl_parameters_out() { + // An SSL listener bound raw serves the socket only, so the handshaker handed back is the + // only copy of the SSL parameters + let (listener, handshaker) = ListenerSetting::from( + Url::parse("https://localhost:0").expect("Listener url is valid"), + ) + .bind_raw() + .await + .expect("The listener should bind"); + assert!(handshaker.is_some()); + assert!(listener.handshaker().is_none()); + assert!(matches!(listener, StreamListener::Tcp(_))); + assert!(listener.to_string().starts_with("tcp://")); + + // A plain listener has none to hand back + let (listener, handshaker) = + ListenerSetting::from(Url::parse("tcp://localhost:0").expect("Listener url is valid")) + .bind_raw() + .await + .expect("The listener should bind"); + assert!(handshaker.is_none()); + assert!(listener.handshaker().is_none()); + + // And neither does a Unix socket, even with an explicit SSL configuration + let socket_path = unique_test_path("test_listener_bind_raw.sock"); + let (listener, handshaker) = ListenerSetting::new( + Url::parse(&format!( + "unix://{}", + socket_path.to_str().expect("Socket path should be string") + )) + .expect("Listener url is valid"), + Some(SslConfig::default()), + ) + .bind_raw() + .await + .expect("The listener should bind"); + assert!(handshaker.is_none()); + assert!(matches!(listener, StreamListener::Unix(_))); + } + + #[cfg(feature = "openssl")] + #[tokio::test] + async fn listener_bound_raw_rotates_through_its_own_handshaker() -> io::Result<()> { + use prosa_utils::config::ssl::Store; + use std::sync::Arc; + + let cert_path = unique_test_path("test_listener_bind_raw_rotate.pem") + .to_str() + .expect("Certificate path should exist") + .to_string(); + + let setting = ListenerSetting::new( + Url::parse("https://localhost:0").expect("Listener url is valid"), + Some(SslConfig::new_self_cert(cert_path.clone())), + ); + + // The shape of a listener that is shared with the tasks handshaking its clients, so it + // can't be replaced to rotate + let (listener, mut handshaker) = setting.bind_raw().await?; + let listener = Arc::new(listener); + let addr = listener.local_addr()?; + let url = + Url::parse(&format!("tls://localhost:{}", addr.port())).expect("Target url is valid"); + + let connect = || async { + let mut client_config = SslConfig::default(); + client_config.set_store(Store::File { + path: cert_path.clone(), + }); + let connector: ::openssl::ssl::SslConnectorBuilder = + client_config.init_tls_client_context()?; + + Stream::connect_openssl(&url, &connector.build()).await + }; + + let serve = |handshaker: Option| { + let listener = listener.clone(); + async move { + let (stream, _addr) = listener.accept_raw().await?; + match handshaker { + Some(handshaker) => handshaker.handshake(stream).await, + None => Ok(stream), + } + } + }; + + let (served, connected) = + futures_util::future::join(serve(handshaker.clone()), connect()).await; + served?; + let certificate = peer_certificate(&connected?); + + // Rotating replaces the only copy there is + handshaker = setting.build_handshaker().await?; + + // The very same socket, and the listener still carries nothing that could go stale + assert_eq!(addr, listener.local_addr()?); + assert!(listener.handshaker().is_none()); + + // And the next client is served the new certificate + let (served, connected) = + futures_util::future::join(serve(handshaker.clone()), connect()).await; + served?; + assert_ne!(certificate, peer_certificate(&connected?)); + + Ok(()) + } + + #[test] + fn listener_setting_needs_rebind() { + let plain = ListenerSetting::from( + Url::parse("tcp://localhost:8080").expect("Listener url is invalid"), + ); + + // Turning SSL on with the scheme keeps the socket, like turning it on with a configuration + assert!(!plain.needs_rebind(&ListenerSetting::from( + Url::parse("ssl://localhost:8080").expect("Listener url is invalid") + ))); + assert!(!plain.needs_rebind(&ListenerSetting::new( + plain.url.clone(), + Some(SslConfig::default()) + ))); + + // And so does anything else that doesn't name an address + let mut path_changed = plain.clone(); + path_changed.url.set_path("/v2"); + path_changed.max_socket = plain.max_socket / 2; + assert!(!plain.needs_rebind(&path_changed)); + + // Listening on another port or another host needs a new socket + assert!(plain.needs_rebind(&ListenerSetting::from( + Url::parse("tcp://localhost:8081").expect("Listener url is invalid") + ))); + assert!(plain.needs_rebind(&ListenerSetting::from( + Url::parse("tcp://127.0.0.1:8080").expect("Listener url is invalid") + ))); + + // A scheme that implies another port does too + assert!( + ListenerSetting::from(Url::parse("http://localhost").expect("Listener url is invalid")) + .needs_rebind(&ListenerSetting::from( + Url::parse("https://localhost").expect("Listener url is invalid") + )) + ); + } + #[test] fn listener_setting_set_alpn() { let mut expected_ssl = SslConfig::default(); diff --git a/prosa_book/src/ch01-02-03-stream.md b/prosa_book/src/ch01-02-03-stream.md index 083c7df..c52c6bc 100644 --- a/prosa_book/src/ch01-02-03-stream.md +++ b/prosa_book/src/ch01-02-03-stream.md @@ -63,13 +63,40 @@ The OpenSSL context is built from that configuration every time a listener binds connects, so a certificate or a CA rotated on disk applies to the next bind or connection without needing a configuration change. -Both settings implement `PartialEq`. On a configuration reload, compare the new settings with the -current ones and rebind or reconnect only when they differ. `set_alpn()` is idempotent, so -normalise before comparing: +Both settings implement `PartialEq`, and `set_alpn()` is idempotent, so normalise before comparing. + +A target reconnects, so on a configuration reload compare the new settings with the current ones +and reconnect only when they differ. + +A listener owns a bound socket, and rebinding it releases the port: another process can take it, +and every client is refused until the new socket is bound. So only change the socket when the +listener has to listen somewhere else, which is what `needs_rebind()` answers by comparing the host +and the port. Everything else is served on the socket that is already bound: build the new SSL +parameters with `build_handshaker()`, then hand them to `set_handshaker()`, which moves the socket +into the returned listener. That covers rotating a certificate, turning SSL on and turning SSL off, +whether SSL is declared by the `ssl` block or by the URL scheme. ```rust,ignore listener_setting.set_alpn(vec!["h2".into()]); -if listener_setting != self.settings.listener { - // ... rebind +if self.settings.listener.needs_rebind(&listener_setting) { + listener = listener_setting.bind().await?; +} else { + // Built before the listener is touched, so a broken certificate leaves it serving the one it + // already has + let handshaker = listener_setting.build_handshaker().await?; + listener = listener.set_handshaker(handshaker); } ``` + +Call it on every configuration reload of a listener, not only when the settings differ: `SslConfig` +holds the *paths* of the certificates, so a rotation that rewrites a file in place leaves the +configuration equal to what it was and there is nothing to compare. `build_handshaker()` reads them +again on every call, on the blocking pool. + +A listener that is SSL through its URL scheme alone is served a default SSL configuration, which +signs a certificate of its own rather than reading one. That certificate is signed again on every +call, so such a listener serves a new identity on every configuration reload and a client that pins +it stops trusting it. Configure a certificate to serve a stable one. + +The clients that are already connected, and the ones in the middle of their handshake, keep the +parameters they started with; only the clients accepted afterwards are served the new ones. diff --git a/prosa_book/src/ch03-07-io.md b/prosa_book/src/ch03-07-io.md index d73f4c9..aac3fce 100644 --- a/prosa_book/src/ch03-07-io.md +++ b/prosa_book/src/ch03-07-io.md @@ -17,6 +17,54 @@ It supports three types of server sockets: Once the object is created, you must call the `accept` method in a loop to accept client connections. Each accepted connection will create a `Stream` socket, which can be managed just like a client socket. +`accept` negotiates SSL before it returns, so the loop carries the handshake of every client. A +handshake lasts up to the configured `ssl_timeout`, which is long enough for a slow client to keep +the loop from accepting anyone else. An SSL server should instead accept with `accept_raw`, take an +owned snapshot of the SSL parameters with `handshaker()`, and handshake in a spawned task: + +```rust,ignore +let (stream, addr) = listener.accept_raw().await?; + +// An OpenSSL context is reference counted, so the snapshot costs an atomic increment +let handshaker = listener.handshaker().cloned(); +tokio::spawn(async move { + let stream = match handshaker { + Some(handshaker) => handshaker.handshake(stream).await?, + None => stream, + }; + + // ... serve the client +}); +``` + +The snapshot is taken when the client is accepted, so a certificate rotated meanwhile applies to +the clients accepted afterwards and leaves the handshakes in flight alone. + +### Sharing the listener + +`set_handshaker` takes the listener by value, so a listener that is shared — held in an `Arc` by the +accept loop and by whatever else drives it — cannot be replaced to rotate its certificate. Bind it +with `bind_raw` instead: it binds the socket and hands the SSL parameters back beside it rather than +inside it, so the handshaker you hold is the only copy and rotating it is an assignment. + +```rust,ignore +let (listener, mut handshaker) = listener_setting.bind_raw().await?; +let listener = Arc::new(listener); + +// Accept loop, same shape, from the handshaker held next to the listener +let (stream, addr) = listener.accept_raw().await?; +let client_handshaker = handshaker.clone(); +tokio::spawn(async move { /* ... */ }); + +// Configuration reload +handshaker = listener_setting.build_handshaker().await?; +``` + +A listener bound this way carries no SSL parameters of its own: it formats as `tcp://`, its +`handshaker()` is `None`, and `accept` returns a plain TCP stream. That is what keeps a rotation +honest — there is no copy left inside the listener for `accept`, `handshake` or `Debug` to serve +after you have replaced yours. + ## Stream A `Stream` represents a client socket or a socket created by a `StreamListener` when a client connects. From 648d3803c337ce415bc3b8bf3ffbededc3768c44 Mon Sep 17 00:00:00 2001 From: Jeremy HERGAULT Date: Wed, 19 Aug 2026 14:09:59 +0200 Subject: [PATCH 3/3] fix: certificate reload edge cases Signed-off-by: Jeremy HERGAULT --- prosa/src/io/listener.rs | 228 ++++++++++++++++------------ prosa_book/src/ch01-02-03-stream.md | 32 ++-- prosa_book/src/ch03-07-io.md | 2 +- 3 files changed, 154 insertions(+), 108 deletions(-) diff --git a/prosa/src/io/listener.rs b/prosa/src/io/listener.rs index 9173a65..b2f33d4 100644 --- a/prosa/src/io/listener.rs +++ b/prosa/src/io/listener.rs @@ -18,7 +18,6 @@ use url::Url; use super::{SafeUrl, SocketAddr, get_safe_url, stream::Stream, url_is_ssl}; -#[cfg(feature = "openssl")] /// SSL parameters a listener serves to the clients it accepts. /// /// Cheap to clone, because an OpenSSL context is reference counted, so an accept loop can hand one @@ -51,14 +50,15 @@ use super::{SafeUrl, SocketAddr, get_safe_url, stream::Stream, url_is_ssl}; /// ``` #[derive(Clone)] pub struct SslHandshaker { + #[cfg(feature = "openssl")] /// Acceptor holding the certificate served to the clients acceptor: ::openssl::ssl::SslAcceptor, /// Timeout of the SSL handshake with a client timeout: Duration, } -#[cfg(feature = "openssl")] impl SslHandshaker { + #[cfg(feature = "openssl")] /// Method to create the SSL parameters served by a listener. /// By default, the SSL handshake timeout is 3 seconds pub fn new(acceptor: ::openssl::ssl::SslAcceptor, timeout: Option) -> SslHandshaker { @@ -76,43 +76,56 @@ impl SslHandshaker { /// Method to negotiate SSL with a client that has just been accepted. /// A stream that is not a plain TCP one is returned as is pub async fn handshake(&self, stream: Stream) -> Result { - let Stream::Tcp(tcp_stream) = stream else { - return Ok(stream); - }; - - let ssl = openssl::ssl::Ssl::new(self.acceptor.context()) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let mut stream = tokio_openssl::SslStream::new(ssl, tcp_stream) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - if let Err(e) = tokio::time::timeout(self.timeout, std::pin::Pin::new(&mut stream).accept()) - .await - .map_err(|_| { - io::Error::new( - io::ErrorKind::TimedOut, - format!( - "SSL timeout[{} ms] for {stream:?}", - self.timeout.as_millis() - ), - ) - })? + #[cfg(feature = "openssl")] { - return Err(io::Error::other(format!("Can't accept the client: {e}"))); + let Stream::Tcp(tcp_stream) = stream else { + return Ok(stream); + }; + + let ssl = openssl::ssl::Ssl::new(self.acceptor.context()) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + let mut stream = tokio_openssl::SslStream::new(ssl, tcp_stream) + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + if let Err(e) = + tokio::time::timeout(self.timeout, std::pin::Pin::new(&mut stream).accept()) + .await + .map_err(|_| { + io::Error::new( + io::ErrorKind::TimedOut, + format!( + "SSL timeout[{} ms] for {stream:?}", + self.timeout.as_millis() + ), + ) + })? + { + return Err(io::Error::other(format!("Can't accept the client: {e}"))); + } + + Ok(Stream::OpenSsl(stream)) } - Ok(Stream::OpenSsl(stream)) + #[cfg(not(feature = "openssl"))] + { + let _ = stream; + Err(io::Error::new( + io::ErrorKind::Unsupported, + "No SSL engine available", + )) + } } } -#[cfg(feature = "openssl")] impl fmt::Debug for SslHandshaker { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.debug_struct("SslHandshaker") - .field("ssl_timeout", &self.timeout) - .field( - "certificate", - &self.acceptor.context().certificate().map(|c| c.to_text()), - ) - .finish() + let mut debug = f.debug_struct("SslHandshaker"); + debug.field("ssl_timeout", &self.timeout); + #[cfg(feature = "openssl")] + debug.field( + "certificate", + &self.acceptor.context().certificate().map(|c| c.to_text()), + ); + debug.finish() } } @@ -123,12 +136,11 @@ pub enum StreamListener { Unix(tokio::net::UnixListener), /// TCP server socket Tcp(TcpListener), - #[cfg(feature = "openssl")] - /// OpenSSL server socket. + /// SSL server socket. /// /// The SSL parameters are held apart from the socket so /// [`StreamListener::set_handshaker`] can serve new ones on the socket it is already bound to - OpenSsl(TcpListener, SslHandshaker), + Ssl(TcpListener, SslHandshaker), } impl fmt::Debug for StreamListener { @@ -137,8 +149,7 @@ impl fmt::Debug for StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(l) => f.debug_struct("Unix").field("listener", &l).finish(), StreamListener::Tcp(l) => f.debug_struct("Tcp").field("listener", &l).finish(), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, ssl) => f + StreamListener::Ssl(l, ssl) => f .debug_struct("Ssl") .field("listener", &l) .field("ssl", &ssl) @@ -176,8 +187,7 @@ impl StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(listener) => listener.local_addr().map(|addr| addr.into()), StreamListener::Tcp(listener) => listener.local_addr().map(|addr| addr.into()), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(listener, _) => listener.local_addr().map(|addr| addr.into()), + StreamListener::Ssl(listener, _) => listener.local_addr().map(|addr| addr.into()), } } @@ -241,7 +251,6 @@ impl StreamListener { self.set_handshaker(Some(SslHandshaker::new(ssl_acceptor, ssl_timeout))) } - #[cfg(feature = "openssl")] /// Getter of the SSL parameters served to a client accepted now, [`None`] on a plain listener. /// /// Clone it into the task that handshakes the client, so the accept loop can go straight back @@ -254,12 +263,11 @@ impl StreamListener { /// the handshaker held beside it is the only one pub fn handshaker(&self) -> Option<&SslHandshaker> { match self { - StreamListener::OpenSsl(_l, handshaker) => Some(handshaker), + StreamListener::Ssl(_l, handshaker) => Some(handshaker), _ => None, } } - #[cfg(feature = "openssl")] /// Method to serve new SSL parameters on the socket that is already bound. /// /// The socket is moved into the returned listener rather than bound again, so the port is @@ -295,13 +303,13 @@ impl StreamListener { /// ``` pub fn set_handshaker(self, handshaker: Option) -> StreamListener { let listener = match self { - StreamListener::Tcp(listener) | StreamListener::OpenSsl(listener, _) => listener, + StreamListener::Tcp(listener) | StreamListener::Ssl(listener, _) => listener, #[cfg(target_family = "unix")] unix_listener => return unix_listener, }; match handshaker { - Some(handshaker) => StreamListener::OpenSsl(listener, handshaker), + Some(handshaker) => StreamListener::Ssl(listener, handshaker), None => StreamListener::Tcp(listener), } } @@ -336,8 +344,7 @@ impl StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(l) => l.accept().await.map(|s| (Stream::Unix(s.0), s.1.into())), StreamListener::Tcp(l) => l.accept().await.map(|s| (Stream::Tcp(s.0), s.1.into())), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, handshaker) => { + StreamListener::Ssl(l, handshaker) => { let (stream, addr) = l.accept().await?; // Read after the accept, so a client that connects once the certificate has been @@ -382,8 +389,7 @@ impl StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(l) => l.accept().await.map(|s| (Stream::Unix(s.0), s.1.into())), StreamListener::Tcp(l) => l.accept().await.map(|s| (Stream::Tcp(s.0), s.1.into())), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, _handshaker) => { + StreamListener::Ssl(l, _handshaker) => { l.accept().await.map(|s| (Stream::Tcp(s.0), s.1.into())) } } @@ -391,8 +397,7 @@ impl StreamListener { /// Method to do an handshake with a client after an accept (Do nothing if the handshake is already done) pub async fn handshake(&self, stream: Stream) -> Result { - #[cfg(feature = "openssl")] - if let StreamListener::OpenSsl(_l, handshaker) = self { + if let StreamListener::Ssl(_l, handshaker) = self { return handshaker.handshake(stream).await; } @@ -406,8 +411,7 @@ impl AsFd for StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(l) => l.as_fd(), StreamListener::Tcp(l) => l.as_fd(), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, _) => l.as_fd(), + StreamListener::Ssl(l, _) => l.as_fd(), } } } @@ -418,8 +422,7 @@ impl AsRawFd for StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(l) => l.as_raw_fd(), StreamListener::Tcp(l) => l.as_raw_fd(), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(l, _) => l.as_raw_fd(), + StreamListener::Ssl(l, _) => l.as_raw_fd(), } } } @@ -436,8 +439,7 @@ impl fmt::Display for StreamListener { #[cfg(target_family = "unix")] StreamListener::Unix(_) => write!(f, "unix://{addr}"), StreamListener::Tcp(_) => write!(f, "tcp://{addr}"), - #[cfg(feature = "openssl")] - StreamListener::OpenSsl(_, _) => write!(f, "ssl://{addr}"), + StreamListener::Ssl(_, _) => write!(f, "ssl://{addr}"), } } } @@ -569,10 +571,11 @@ impl ListenerSetting { /// Method to know if serving `other` needs a new socket, because it doesn't listen on the same /// address. /// - /// Only the host and the port are compared, so everything else is served on the socket that is - /// already bound, whether that rotates a certificate or turns SSL on or off. A scheme carries - /// SSL rather than an address, and `max_socket` is a cap the processor enforces rather than a - /// property of the socket, so neither of them ever needs a new one. + /// TCP listeners are compared by host and port. Unix listeners are compared by path, with + /// `unix` and `file` treated as equivalent socket schemes. Everything else is served on the + /// socket that is already bound, whether that rotates a certificate or turns SSL on or off. + /// Other scheme changes carry protocol information rather than a bind address, and + /// `max_socket` is a cap the processor enforces rather than a property of the socket. /// /// ``` /// use url::Url; @@ -594,18 +597,24 @@ impl ListenerSetting { /// assert!(plain.needs_rebind(&moved)); /// ``` pub fn needs_rebind(&self, other: &ListenerSetting) -> bool { + let self_is_unix = matches!(self.url.scheme(), "unix" | "file"); + let other_is_unix = matches!(other.url.scheme(), "unix" | "file"); + if self_is_unix || other_is_unix { + return self_is_unix != other_is_unix || self.url.path() != other.url.path(); + } + self.url.host() != other.url.host() || self.url.port_or_known_default() != other.url.port_or_known_default() } - #[cfg(feature = "openssl")] /// Method to build the SSL parameters this configuration serves, [`None`] when it listens /// without SSL. /// /// The certificates are read again on every call, on the blocking pool, because [`SslConfig`] - /// holds their *paths*: a rotation that rewrites a file in place leaves the configuration equal - /// to what it was, so a caller has nothing to compare and should call this on every - /// configuration reload. + /// holds their *paths*. Calling this is explicit: rebuild when the configuration changes or + /// when the certificate source reports a new version. The ProSA configuration watcher does not + /// watch certificate files, and a file rewritten at the same path does not change the parsed + /// configuration. /// /// Hand the result to [`StreamListener::set_handshaker`] to serve it without rebinding. It is /// built before the listener is touched, so a broken certificate leaves the listener serving @@ -613,8 +622,8 @@ impl ListenerSetting { /// /// A listener that is SSL through its URL scheme alone is served a default [`SslConfig`], which /// signs a certificate of its own rather than reading one. That certificate is signed again on - /// every call, so such a listener serves a new identity on every configuration reload and a - /// client that pins it stops trusting it. Configure a certificate to serve a stable one. + /// every call, so each rebuild serves a new identity and a client that pins it stops trusting + /// it. Configure a certificate to serve a stable one. pub async fn build_handshaker(&self) -> Result, io::Error> { // A Unix socket never serves SSL, `bind` ignores the SSL configuration for it #[cfg(target_family = "unix")] @@ -626,20 +635,29 @@ impl ListenerSetting { return Ok(None); } - let ssl_config = self.ssl.clone().unwrap_or_default(); - let timeout = ssl_config.get_ssl_timeout(); - let host = self.url.host_str().map(String::from); + #[cfg(feature = "openssl")] + { + let ssl_config = self.ssl.clone().unwrap_or_default(); + let timeout = ssl_config.get_ssl_timeout(); + let host = self.url.host_str().map(String::from); + + let acceptor = tokio::task::spawn_blocking(move || { + ssl_config + .init_tls_server_context(host.as_deref()) + .map(|ssl_context_builder| ssl_context_builder.build()) + }) + .await + .map_err(io::Error::other)? + .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; - let acceptor = tokio::task::spawn_blocking(move || { - ssl_config - .init_tls_server_context(host.as_deref()) - .map(|ssl_context_builder| ssl_context_builder.build()) - }) - .await - .map_err(io::Error::other)? - .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?; + Ok(Some(SslHandshaker::new(acceptor, Some(timeout)))) + } - Ok(Some(SslHandshaker::new(acceptor, Some(timeout)))) + #[cfg(not(feature = "openssl"))] + Err(io::Error::new( + io::ErrorKind::Unsupported, + "No SSL engine available", + )) } /// Bind the socket of the configuration, without SSL @@ -654,7 +672,6 @@ impl ListenerSetting { StreamListener::bind(&*addrs).await } - #[cfg(feature = "openssl")] /// Method to bind the socket of the configuration and build the SSL parameters to serve on it, /// without attaching them to the listener. /// @@ -705,23 +722,8 @@ impl ListenerSetting { /// Method to connect a ProSA stream to the remote target using the configuration pub async fn bind(&self) -> Result { - #[cfg(feature = "openssl")] - { - let (stream_listener, handshaker) = self.bind_raw().await?; - Ok(stream_listener.set_handshaker(handshaker)) - } - - #[cfg(not(feature = "openssl"))] - { - if self.is_ssl() { - return Err(io::Error::new( - io::ErrorKind::Unsupported, - "No SSL engine available", - )); - } - - self.bind_socket().await - } + let (stream_listener, handshaker) = self.bind_raw().await?; + Ok(stream_listener.set_handshaker(handshaker)) } } @@ -769,6 +771,7 @@ impl fmt::Display for ListenerSetting { mod tests { use super::*; + #[cfg(feature = "openssl")] /// Path of a test file no other test, and no other test run, writes to fn unique_test_path(name: &str) -> std::path::PathBuf { let timestamp = std::time::SystemTime::now() @@ -835,6 +838,30 @@ mod tests { assert_eq!(setting, built); } + #[cfg(not(feature = "openssl"))] + #[tokio::test] + async fn listener_setting_build_handshaker_without_ssl_engine() { + let plain = ListenerSetting::from( + Url::parse("tcp://127.0.0.1:0").expect("Listener URL should be valid"), + ); + assert!( + plain + .build_handshaker() + .await + .expect("A plain listener should not need an SSL engine") + .is_none() + ); + + let ssl = ListenerSetting::from( + Url::parse("https://127.0.0.1:0").expect("Listener URL should be valid"), + ); + let error = ssl + .build_handshaker() + .await + .expect_err("An SSL listener should require an SSL engine"); + assert_eq!(io::ErrorKind::Unsupported, error.kind()); + } + #[cfg(feature = "openssl")] fn served_certificate(listener: &StreamListener) -> Vec { listener @@ -1174,6 +1201,17 @@ mod tests { Url::parse("https://localhost").expect("Listener url is invalid") )) ); + + let unix = ListenerSetting::from( + Url::parse("unix:///tmp/prosa-listener.sock").expect("Listener url is invalid"), + ); + assert!(unix.needs_rebind(&ListenerSetting::from( + Url::parse("unix:///tmp/prosa-listener-new.sock").expect("Listener url is invalid") + ))); + assert!(!unix.needs_rebind(&ListenerSetting::from( + Url::parse("file:///tmp/prosa-listener.sock").expect("Listener url is invalid") + ))); + assert!(unix.needs_rebind(&plain)); } #[test] diff --git a/prosa_book/src/ch01-02-03-stream.md b/prosa_book/src/ch01-02-03-stream.md index c52c6bc..f36dd64 100644 --- a/prosa_book/src/ch01-02-03-stream.md +++ b/prosa_book/src/ch01-02-03-stream.md @@ -60,8 +60,10 @@ none, and does nothing on a plain one. `is_ssl()` tells whether SSL applies: an negotiate ALPN. The OpenSSL context is built from that configuration every time a listener binds or a target -connects, so a certificate or a CA rotated on disk applies to the next bind or connection without -needing a configuration change. +connects. This makes the convenience API pick up certificate changes, but parsing certificates for +every target connection has a cost. Performance-sensitive clients can build an `SslConnector` +snapshot with `SslConfigContext`, use the lower-level `Stream::connect_openssl` API, and replace the +snapshot only when their certificate source reports a change. Both settings implement `PartialEq`, and `set_alpn()` is idempotent, so normalise before comparing. @@ -71,10 +73,11 @@ and reconnect only when they differ. A listener owns a bound socket, and rebinding it releases the port: another process can take it, and every client is refused until the new socket is bound. So only change the socket when the listener has to listen somewhere else, which is what `needs_rebind()` answers by comparing the host -and the port. Everything else is served on the socket that is already bound: build the new SSL -parameters with `build_handshaker()`, then hand them to `set_handshaker()`, which moves the socket -into the returned listener. That covers rotating a certificate, turning SSL on and turning SSL off, -whether SSL is declared by the `ssl` block or by the URL scheme. +and port of TCP listeners or the path of Unix listeners. Everything else is served on the socket +that is already bound: build the new SSL parameters with `build_handshaker()`, then hand them to +`set_handshaker()`, which moves the socket into the returned listener. That covers rotating a +certificate, turning SSL on and turning SSL off, whether SSL is declared by the `ssl` block or by +the URL scheme. ```rust,ignore listener_setting.set_alpn(vec!["h2".into()]); @@ -88,15 +91,20 @@ if self.settings.listener.needs_rebind(&listener_setting) { } ``` -Call it on every configuration reload of a listener, not only when the settings differ: `SslConfig` -holds the *paths* of the certificates, so a rotation that rewrites a file in place leaves the -configuration equal to what it was and there is nothing to compare. `build_handshaker()` reads them -again on every call, on the blocking pool. +`build_handshaker()` reads certificate paths again on every call, on the blocking pool, but deciding +when to call it belongs to the certificate source. A processor can react to a configuration change, +a filesystem watcher can react to a replaced certificate, and a remote secret provider can react to +a new version or lease. ProSA's configuration watcher does not watch certificate files and does not +notify processors when only a file referenced by an unchanged configuration is replaced. + +The handshaker is the common runtime snapshot for those sources. File-backed configurations can use +`build_handshaker()`; another provider can build an OpenSSL `SslAcceptor`, wrap it with +`SslHandshaker::new`, and install it through the same `set_handshaker()` operation. A listener that is SSL through its URL scheme alone is served a default SSL configuration, which signs a certificate of its own rather than reading one. That certificate is signed again on every -call, so such a listener serves a new identity on every configuration reload and a client that pins -it stops trusting it. Configure a certificate to serve a stable one. +rebuild, so such a listener serves a new identity and a client that pins it stops trusting it. +Configure a certificate to serve a stable one. The clients that are already connected, and the ones in the middle of their handshake, keep the parameters they started with; only the clients accepted afterwards are served the new ones. diff --git a/prosa_book/src/ch03-07-io.md b/prosa_book/src/ch03-07-io.md index aac3fce..c5bb648 100644 --- a/prosa_book/src/ch03-07-io.md +++ b/prosa_book/src/ch03-07-io.md @@ -56,7 +56,7 @@ let (stream, addr) = listener.accept_raw().await?; let client_handshaker = handshaker.clone(); tokio::spawn(async move { /* ... */ }); -// Configuration reload +// Configuration reload or certificate-source change notification handshaker = listener_setting.build_handshaker().await?; ```