From 6185d38911cd00a7f42d7de4c12712628dc2a941 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 12 May 2026 00:25:20 +0800 Subject: [PATCH] fix(protocols): add hot reload for WebDAV FTPS and SFTP (#2922) Signed-off-by: houseme Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- Cargo.lock | 1 + crates/config/src/constants/protocols.rs | 8 + crates/protocols/Cargo.toml | 1 + crates/protocols/src/ftps/server.rs | 24 +- crates/protocols/src/lib.rs | 3 + crates/protocols/src/sftp/config.rs | 30 ++- crates/protocols/src/sftp/mod.rs | 3 +- crates/protocols/src/sftp/server.rs | 259 ++++++++++++++++++++- crates/protocols/src/tls_hot_reload.rs | 279 +++++++++++++++++++++++ crates/protocols/src/webdav/server.rs | 24 +- 10 files changed, 586 insertions(+), 46 deletions(-) create mode 100644 crates/protocols/src/tls_hot_reload.rs diff --git a/Cargo.lock b/Cargo.lock index 6e4b38dc2..31f383e7e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9974,6 +9974,7 @@ dependencies = [ "percent-encoding", "proptest", "quick-xml 0.39.4", + "rcgen", "regex", "russh", "russh-sftp", diff --git a/crates/config/src/constants/protocols.rs b/crates/config/src/constants/protocols.rs index 1f3729531..a102b5235 100644 --- a/crates/config/src/constants/protocols.rs +++ b/crates/config/src/constants/protocols.rs @@ -69,6 +69,8 @@ pub const DEFAULT_SFTP_HOST_KEY_DIR: Option<&str> = None; pub const ENV_SFTP_ENABLE: &str = "RUSTFS_SFTP_ENABLE"; pub const ENV_SFTP_ADDRESS: &str = "RUSTFS_SFTP_ADDRESS"; pub const ENV_SFTP_HOST_KEY_DIR: &str = "RUSTFS_SFTP_HOST_KEY_DIR"; +pub const ENV_SFTP_HOST_KEY_RELOAD_ENABLE: &str = "RUSTFS_SFTP_HOST_KEY_RELOAD_ENABLE"; +pub const ENV_SFTP_HOST_KEY_RELOAD_INTERVAL: &str = "RUSTFS_SFTP_HOST_KEY_RELOAD_INTERVAL"; pub const ENV_SFTP_IDLE_TIMEOUT: &str = "RUSTFS_SFTP_IDLE_TIMEOUT"; /// S3 multipart part size in bytes. Default DEFAULT_SFTP_PART_SIZE (16 MiB). /// Valid range 5 MiB to 5 GiB (S3 protocol bounds), enforced at startup. @@ -146,6 +148,12 @@ pub const ENV_SFTP_READ_CACHE_TOTAL_MEM_BYTES: &str = "RUSTFS_SFTP_READ_CACHE_TO /// Default idle session timeout in seconds. pub const DEFAULT_SFTP_IDLE_TIMEOUT: u64 = 600; +/// Default SFTP host key hot reload enabled state. +pub const DEFAULT_SFTP_HOST_KEY_RELOAD_ENABLE: bool = false; + +/// Default SFTP host key hot reload interval in seconds. +pub const DEFAULT_SFTP_HOST_KEY_RELOAD_INTERVAL: u64 = 30; + /// Default S3 multipart upload part size in bytes (16 MiB). /// /// The per-upload size ceiling is part_size * 10_000 (the S3 parts cap), diff --git a/crates/protocols/Cargo.toml b/crates/protocols/Cargo.toml index 2879f27f9..61f52c554 100644 --- a/crates/protocols/Cargo.toml +++ b/crates/protocols/Cargo.toml @@ -128,6 +128,7 @@ socket2 = { workspace = true, optional = true } [dev-dependencies] tempfile = { workspace = true } proptest = "1" +rcgen = { workspace = true } tracing-subscriber = { workspace = true } [package.metadata.docs.rs] diff --git a/crates/protocols/src/ftps/server.rs b/crates/protocols/src/ftps/server.rs index 92d32b4f8..98f273d1f 100644 --- a/crates/protocols/src/ftps/server.rs +++ b/crates/protocols/src/ftps/server.rs @@ -17,13 +17,14 @@ use super::driver::FtpsDriver; use crate::common::client::s3::StorageBackend; use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext}; use crate::constants::{network::DEFAULT_SOURCE_IP, paths::ROOT_PATH}; +use crate::tls_hot_reload::{ReloadableCertResolver, spawn_cert_reload_loop}; use libunftp::options::FtpsRequired; -use rustfs_config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY}; use std::fmt::{Debug, Display, Formatter}; use std::net::IpAddr; use std::path::Path; use std::sync::Arc; use tokio::sync::broadcast; +use tokio::sync::watch; use tracing::{debug, error, info, warn}; use unftp_core::auth::{ AuthenticationError, Authenticator, Credentials, Principal, UserDetail, UserDetailError, UserDetailProvider, @@ -79,6 +80,7 @@ where /// then spawns the server loop in a background task. pub async fn start(&self, mut shutdown_rx: broadcast::Receiver<()>) -> Result<(), FtpsInitError> { info!("Initializing FTPS server on {}", self.config.bind_addr); + let (reload_shutdown_tx, reload_shutdown_rx) = watch::channel(false); let storage_clone = self.storage.clone(); let mut server_builder = libunftp::ServerBuilder::with_user_detail_provider( @@ -112,28 +114,16 @@ where if let Some(cert_dir) = &self.config.cert_dir { debug!("Enabling FTPS with multi-certificate support from directory: {}", cert_dir); - // Load all certificates from directory - let cert_key_pairs = rustfs_utils::load_all_certs_from_directory( - rustfs_utils::CertDirectoryLoadOptions::builder(cert_dir, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY).build(), - ) - .map_err(|e| FtpsInitError::InvalidConfig(format!("Failed to load certificates: {}", e)))?; - - if cert_key_pairs.is_empty() { - return Err(FtpsInitError::InvalidConfig("No valid certificates found in directory".into())); - } - - debug!("Loaded {} certificates for FTPS", cert_key_pairs.len()); - - // Create multi-certificate resolver with SNI support - let resolver = rustfs_utils::create_multi_cert_resolver(cert_key_pairs) + let resolver = ReloadableCertResolver::load_from_directory(cert_dir) .map_err(|e| FtpsInitError::InvalidConfig(format!("Failed to create certificate resolver: {}", e)))?; + let _reload_task = spawn_cert_reload_loop("ftps", cert_dir.clone(), resolver.clone(), reload_shutdown_rx.clone()); // Build ServerConfig with SNI support let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); let server_config = rustls::ServerConfig::builder() .with_no_client_auth() - .with_cert_resolver(Arc::new(resolver)); + .with_cert_resolver(resolver); server_builder = server_builder.ftps_manual::(Arc::new(server_config)); @@ -166,6 +156,7 @@ where // Wait for shutdown signal or server failure tokio::select! { result = server_handle => { + let _ = reload_shutdown_tx.send(true); match result { Ok(Ok(())) => { info!("FTPS server stopped normally"); @@ -183,6 +174,7 @@ where } _ = shutdown_rx.recv() => { info!("FTPS server received shutdown signal"); + let _ = reload_shutdown_tx.send(true); // libunftp listen() is not easily cancellable gracefully without dropping the future. // The select! dropping server_handle will close the listener. Ok(()) diff --git a/crates/protocols/src/lib.rs b/crates/protocols/src/lib.rs index 2133b614a..e6ac15bc6 100644 --- a/crates/protocols/src/lib.rs +++ b/crates/protocols/src/lib.rs @@ -17,6 +17,9 @@ pub mod common; pub mod constants; +#[cfg(any(feature = "ftps", feature = "webdav"))] +mod tls_hot_reload; + #[cfg(feature = "ftps")] pub mod ftps; diff --git a/crates/protocols/src/sftp/config.rs b/crates/protocols/src/sftp/config.rs index c4c334596..d0e934acb 100644 --- a/crates/protocols/src/sftp/config.rs +++ b/crates/protocols/src/sftp/config.rs @@ -28,6 +28,7 @@ use super::constants::limits::{ READ_CACHE_TOTAL_MEM_MIN, READ_CACHE_WINDOW_DEFAULT, READ_CACHE_WINDOW_MAX, READ_CACHE_WINDOW_MIN, S3_MAX_PART_SIZE, S3_MIN_PART_SIZE, }; +use russh::keys::PublicKeyBase64; use std::net::SocketAddr; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; @@ -421,15 +422,26 @@ impl SftpConfig { }); } - // Sort keys by algorithm preference: Ed25519 first, then ECDSA, - // then RSA. russh offers keys to clients in array order during - // key exchange. The ordering controls which algorithm the - // client attempts first. - keys.sort_by_key(|k| match k.algorithm() { - russh::keys::Algorithm::Ed25519 => 0, - russh::keys::Algorithm::Ecdsa { .. } => 1, - russh::keys::Algorithm::Rsa { .. } => 2, - _ => 3, + // Sort keys by algorithm preference, then by public key bytes + // for deterministic ordering within the same algorithm. + // russh offers keys to clients in array order during key exchange. + keys.sort_by(|left, right| { + let left_rank = match left.algorithm() { + russh::keys::Algorithm::Ed25519 => 0, + russh::keys::Algorithm::Ecdsa { .. } => 1, + russh::keys::Algorithm::Rsa { .. } => 2, + _ => 3, + }; + let right_rank = match right.algorithm() { + russh::keys::Algorithm::Ed25519 => 0, + russh::keys::Algorithm::Ecdsa { .. } => 1, + russh::keys::Algorithm::Rsa { .. } => 2, + _ => 3, + }; + + left_rank + .cmp(&right_rank) + .then_with(|| left.public_key_bytes().cmp(&right.public_key_bytes())) }); tracing::info!( diff --git a/crates/protocols/src/sftp/mod.rs b/crates/protocols/src/sftp/mod.rs index ce75db1d3..fcca85ccb 100644 --- a/crates/protocols/src/sftp/mod.rs +++ b/crates/protocols/src/sftp/mod.rs @@ -36,8 +36,9 @@ //! - read_cache: per-handle in-memory read-ahead cache with a process-wide //! memory ceiling. //! -//! Configuration contract. Eleven RUSTFS_SFTP_* environment variables drive +//! Configuration contract. Thirteen RUSTFS_SFTP_* environment variables drive //! the server: RUSTFS_SFTP_ENABLE, RUSTFS_SFTP_ADDRESS, RUSTFS_SFTP_HOST_KEY_DIR, +//! RUSTFS_SFTP_HOST_KEY_RELOAD_ENABLE, RUSTFS_SFTP_HOST_KEY_RELOAD_INTERVAL, //! RUSTFS_SFTP_IDLE_TIMEOUT, RUSTFS_SFTP_PART_SIZE, RUSTFS_SFTP_READ_ONLY, //! RUSTFS_SFTP_BANNER, RUSTFS_SFTP_HANDLES_PER_SESSION, //! RUSTFS_SFTP_BACKEND_OP_TIMEOUT_SECS, RUSTFS_SFTP_READ_CACHE_WINDOW_BYTES, diff --git a/crates/protocols/src/sftp/server.rs b/crates/protocols/src/sftp/server.rs index f51e828bf..92beb2f1d 100644 --- a/crates/protocols/src/sftp/server.rs +++ b/crates/protocols/src/sftp/server.rs @@ -32,19 +32,26 @@ use super::lifecycle::{SessionDiag, SessionRegistry, new_session_registry}; use super::wedge_watchdog; use crate::common::client::s3::StorageBackend; use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext}; -use russh::keys::{self, PrivateKey}; +use russh::keys::{self, PrivateKey, PublicKeyBase64}; use russh::server::{Auth, Msg, Session}; use russh::{Channel, ChannelId, MethodKind, MethodSet, Pty, Sig}; +use rustfs_config::{ + DEFAULT_SFTP_HOST_KEY_RELOAD_ENABLE, DEFAULT_SFTP_HOST_KEY_RELOAD_INTERVAL, ENV_SFTP_HOST_KEY_RELOAD_ENABLE, + ENV_SFTP_HOST_KEY_RELOAD_INTERVAL, +}; use std::borrow::Cow; use std::collections::HashMap; +use std::collections::hash_map::DefaultHasher; use std::fmt::Debug; +use std::hash::Hasher; use std::net::SocketAddr; use std::sync::Arc; +use std::sync::RwLock; use std::sync::atomic::AtomicU64; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::broadcast; use tokio::task::JoinSet; -use tokio::time::{Duration, timeout}; +use tokio::time::{Duration, MissedTickBehavior, timeout}; use tokio_util::sync::CancellationToken; use crate::sftp::constants::limits::SHUTDOWN_DRAIN_TIMEOUT_SECS; @@ -135,10 +142,143 @@ fn build_ssh_config(host_keys: Vec, idle_timeout_secs: u64, banner: }) } +#[derive(Debug)] +struct SshConfigHolder { + current: RwLock>, + fingerprint: RwLock, +} + +impl SshConfigHolder { + fn new(config: Arc) -> Self { + let fingerprint = fingerprint_host_keys(&config.keys); + Self { + current: RwLock::new(config), + fingerprint: RwLock::new(fingerprint), + } + } + + fn get(&self) -> Arc { + match self.current.read() { + Ok(guard) => Arc::clone(&guard), + Err(poisoned) => Arc::clone(&poisoned.into_inner()), + } + } + + async fn reload_from_config(&self, config: &SftpConfig) -> Result, SftpInitError> { + let host_keys = SftpConfig::load_host_keys(&config.host_key_dir).await?; + let host_key_count = host_keys.len(); + let fingerprint = fingerprint_host_keys(&host_keys); + let ssh_config = build_ssh_config(host_keys, config.idle_timeout_secs, &config.banner); + + let mut fingerprint_guard = match self.fingerprint.write() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + if *fingerprint_guard == fingerprint { + return Ok(None); + } + + match self.current.write() { + Ok(mut guard) => *guard = ssh_config, + Err(poisoned) => { + let mut guard = poisoned.into_inner(); + *guard = ssh_config; + } + } + *fingerprint_guard = fingerprint; + + Ok(Some(host_key_count)) + } +} + +fn fingerprint_host_keys(host_keys: &[PrivateKey]) -> u64 { + let mut hasher = DefaultHasher::new(); + let mut public_keys: Vec<(u8, String)> = host_keys + .iter() + .map(|key| (host_key_algorithm_rank(key.algorithm()), key.public_key_base64())) + .collect(); + public_keys.sort_unstable(); + + for (algorithm_rank, public_key_base64) in public_keys { + hasher.write_u8(algorithm_rank); + hasher.write_usize(public_key_base64.len()); + hasher.write(public_key_base64.as_bytes()); + } + + hasher.finish() +} + +fn host_key_algorithm_rank(algorithm: keys::Algorithm) -> u8 { + match algorithm { + keys::Algorithm::Ed25519 => 0, + keys::Algorithm::Ecdsa { .. } => 1, + keys::Algorithm::Rsa { .. } => 2, + _ => 3, + } +} + +fn spawn_host_key_reload_loop(config: SftpConfig, holder: Arc, shutdown_token: CancellationToken) { + let enabled = rustfs_utils::get_env_bool(ENV_SFTP_HOST_KEY_RELOAD_ENABLE, DEFAULT_SFTP_HOST_KEY_RELOAD_ENABLE); + if !enabled { + tracing::debug!( + "SFTP host key hot reload is disabled (set {}=1 to enable)", + ENV_SFTP_HOST_KEY_RELOAD_ENABLE + ); + return; + } + + let interval_secs = + rustfs_utils::get_env_u64(ENV_SFTP_HOST_KEY_RELOAD_INTERVAL, DEFAULT_SFTP_HOST_KEY_RELOAD_INTERVAL).max(5); + + tracing::info!( + host_key_dir = %config.host_key_dir.display(), + interval_secs, + "SFTP host key hot reload enabled" + ); + + tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(interval_secs)); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + interval.tick().await; + loop { + tokio::select! { + _ = shutdown_token.cancelled() => { + tracing::info!(host_key_dir = %config.host_key_dir.display(), "SFTP host key hot reload task stopped"); + break; + } + _ = interval.tick() => {} + } + + match holder.reload_from_config(&config).await { + Ok(Some(host_key_count)) => { + tracing::info!( + host_key_dir = %config.host_key_dir.display(), + host_key_count, + "SFTP host keys reloaded successfully" + ); + } + Ok(None) => { + tracing::debug!( + host_key_dir = %config.host_key_dir.display(), + "SFTP host key material unchanged; skipping reload" + ); + } + Err(err) => { + tracing::warn!( + host_key_dir = %config.host_key_dir.display(), + err = %err, + "SFTP host key reload failed; keeping previous keys" + ); + } + } + } + }); +} + /// SSH server hosting the SFTP subsystem. pub struct SftpServer { config: SftpConfig, - ssh_config: Arc, + ssh_config: Arc, storage: S, /// Weak refs to live per-session activity records. Walked by the /// per-session wedge watchdog and by external observers that @@ -166,7 +306,11 @@ where { /// Build a new server from validated configuration and loaded host keys. pub fn new(config: SftpConfig, storage: S, host_keys: Vec) -> Result { - let ssh_config = build_ssh_config(host_keys, config.idle_timeout_secs, &config.banner); + let ssh_config = Arc::new(SshConfigHolder::new(build_ssh_config( + host_keys, + config.idle_timeout_secs, + &config.banner, + ))); Ok(Self { config, ssh_config, @@ -221,6 +365,7 @@ where // the SHUTDOWN_DRAIN_TIMEOUT_SECS ceiling because no session // has to wait for the watchdog's natural tick to fire. let server_shutdown_token = CancellationToken::new(); + spawn_host_key_reload_loop(self.config.clone(), Arc::clone(&self.ssh_config), server_shutdown_token.child_token()); loop { self.drain_finished_tasks(&mut sessions); @@ -289,7 +434,7 @@ where } }; - let ssh_config = Arc::clone(&self.ssh_config); + let ssh_config = self.ssh_config.get(); // Capture local_addr for the wedge watchdog's TCP-state probe. // Failure here only happens if the kernel can no longer name // the accepted socket. Fall back to an unspecified address @@ -365,6 +510,110 @@ where } } +#[cfg(all(test, unix))] +mod hot_reload_tests { + use super::*; + use std::os::unix::fs::OpenOptionsExt; + use std::path::Path; + use tempfile::TempDir; + + const PEM_BOUNDARY_DASHES: &str = "-----"; + const PEM_OPENSSH_LABEL: &str = "OPENSSH PRIVATE KEY"; + + fn build_pem_block(body: &str) -> String { + format!("{d}BEGIN {l}{d}\n{body}\n{d}END {l}{d}\n", d = PEM_BOUNDARY_DASHES, l = PEM_OPENSSH_LABEL,) + } + + fn test_ed25519_pem() -> String { + build_pem_block( + "b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAMwAAAAtzc2gtZW\n\ + QyNTUxOQAAACCkeMEUpnJEbOMBXiQfjZcHZMEbHW3DlNRL+Jbi1cIqMgAAAKDviRiQ74kY\n\ + kAAAAAtzc2gtZWQyNTUxOQAAACCkeMEUpnJEbOMBXiQfjZcHZMEbHW3DlNRL+Jbi1cIqMg\n\ + AAAEBb5q0DpuL1Rbx4CHUEaRQRSVn1xS2SF+A+qES7OkhrOKR4wRSmckRs4wFeJB+Nlwdk\n\ + wRsdbcOU1Ev4luLVwioyAAAAGHNpbW9uc0B1YnVudHUtbGludXgtMjQwNAECAwQF", + ) + } + + fn test_ecdsa_pem() -> String { + build_pem_block( + "b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAaAAAABNlY2RzYS\n\ + 1zaGEyLW5pc3RwMjU2AAAACG5pc3RwMjU2AAAAQQSBp+cYoqTsQzIF+eQS23gIOBFkIqhi\n\ + M8u54NeDrEyxKSewEHP+5i6/+1HURUWDnW+YfS6nbfGb8GxBkJ2ghVvZAAAAqPpS97P6Uv\n\ + ezAAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBIGn5xiipOxDMgX5\n\ + 5BLbeAg4EWQiqGIzy7ng14OsTLEpJ7AQc/7mLr/7UdRFRYOdb5h9Lqdt8ZvwbEGQnaCFW9\n\ + kAAAAgBdQn3JuP2lSrY3082L+jmYvESyPu9bSmzUe8yMuILzIAAAALdGVzdC12ZWN0b3IB\n\ + AgMEBQ==", + ) + } + + fn write_file_with_mode(path: &Path, content: &str, mode: u32) { + let mut opts = std::fs::OpenOptions::new(); + opts.write(true).create(true).truncate(true).mode(mode); + let mut file = opts.open(path).expect("open file"); + std::io::Write::write_all(&mut file, content.as_bytes()).expect("write file"); + } + + fn test_config(host_key_dir: &Path) -> SftpConfig { + SftpConfig { + bind_addr: "0.0.0.0:2222".parse().unwrap(), + host_key_dir: host_key_dir.to_path_buf(), + idle_timeout_secs: 600, + part_size: 16 * 1024 * 1024, + handles_per_session: None, + backend_op_timeout_secs: None, + read_cache_window_bytes: None, + read_cache_total_mem_bytes: None, + read_only: false, + banner: "SSH-2.0-RustFS".to_string(), + } + } + + #[tokio::test] + async fn ssh_config_holder_reload_replaces_host_keys_for_new_sessions() { + let dir = TempDir::new().expect("tempdir"); + write_file_with_mode(&dir.path().join("ssh_host_ed25519_key"), &test_ed25519_pem(), 0o600); + + let config = test_config(dir.path()); + let initial_keys = SftpConfig::load_host_keys(dir.path()).await.expect("initial key load"); + let holder = SshConfigHolder::new(build_ssh_config(initial_keys, config.idle_timeout_secs, &config.banner)); + assert!(matches!(holder.get().keys[0].algorithm(), russh::keys::Algorithm::Ed25519)); + + std::fs::remove_file(dir.path().join("ssh_host_ed25519_key")).expect("remove old key"); + write_file_with_mode(&dir.path().join("ssh_host_ecdsa_key"), &test_ecdsa_pem(), 0o600); + + let reloaded = holder.reload_from_config(&config).await.expect("reload host keys"); + assert_eq!(reloaded, Some(1)); + assert!(matches!(holder.get().keys[0].algorithm(), russh::keys::Algorithm::Ecdsa { .. })); + } + + #[tokio::test] + async fn ssh_config_holder_reload_skips_when_host_keys_are_unchanged() { + let dir = TempDir::new().expect("tempdir"); + write_file_with_mode(&dir.path().join("ssh_host_ed25519_key"), &test_ed25519_pem(), 0o600); + + let config = test_config(dir.path()); + let initial_keys = SftpConfig::load_host_keys(dir.path()).await.expect("initial key load"); + let holder = SshConfigHolder::new(build_ssh_config(initial_keys, config.idle_timeout_secs, &config.banner)); + + let reloaded = holder.reload_from_config(&config).await.expect("reload host keys"); + assert_eq!(reloaded, None); + } + + #[tokio::test] + async fn fingerprint_host_keys_is_order_independent() { + let dir = TempDir::new().expect("tempdir"); + write_file_with_mode(&dir.path().join("ssh_host_ed25519_key"), &test_ed25519_pem(), 0o600); + write_file_with_mode(&dir.path().join("ssh_host_ecdsa_key"), &test_ecdsa_pem(), 0o600); + + let keys = SftpConfig::load_host_keys(dir.path()).await.expect("load keys"); + let forward = fingerprint_host_keys(&keys); + let reversed_keys: Vec<_> = keys.into_iter().rev().collect(); + let reversed = fingerprint_host_keys(&reversed_keys); + + assert_eq!(forward, reversed); + } +} + /// Drive one accepted SSH session through handshake, optional /// watchdog spawn, the post-handshake session loop, and cleanup. /// Free function (not a method) so the spawn closure on the JoinSet diff --git a/crates/protocols/src/tls_hot_reload.rs b/crates/protocols/src/tls_hot_reload.rs new file mode 100644 index 000000000..7209b44c7 --- /dev/null +++ b/crates/protocols/src/tls_hot_reload.rs @@ -0,0 +1,279 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use rustfs_config::{ + DEFAULT_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_INTERVAL, ENV_TLS_RELOAD_ENABLE, ENV_TLS_RELOAD_INTERVAL, RUSTFS_TLS_CERT, + RUSTFS_TLS_KEY, +}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer}; +use rustls::server::{ClientHello, ResolvesServerCert, ResolvesServerCertUsingSni}; +use rustls::sign::CertifiedKey; +use std::collections::HashMap; +use std::collections::hash_map::DefaultHasher; +use std::hash::Hasher; +use std::io::{self, Error}; +use std::sync::{Arc, RwLock}; +use std::time::Duration; +use tokio::sync::watch; +use tokio::task::JoinHandle; +use tokio::time::MissedTickBehavior; +use tracing::{debug, info, warn}; + +#[derive(Debug)] +struct ResolverState { + cert_resolver: ResolvesServerCertUsingSni, + default_cert: Option>, + cert_count: usize, + fingerprint: u64, +} + +impl ResolverState { + fn load_from_directory(cert_dir: &str) -> io::Result { + let cert_key_pairs = rustfs_utils::load_all_certs_from_directory( + rustfs_utils::CertDirectoryLoadOptions::builder(cert_dir, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY).build(), + )?; + if cert_key_pairs.is_empty() { + return Err(Error::other("No valid certificates found in directory")); + } + + Self::from_cert_key_pairs(cert_key_pairs) + } + + fn from_cert_key_pairs( + cert_key_pairs: HashMap>, PrivateKeyDer<'static>)>, + ) -> io::Result { + let cert_count = cert_key_pairs.len(); + let mut cert_resolver = ResolvesServerCertUsingSni::new(); + let mut default_cert = None; + let mut entries = cert_key_pairs.into_iter().collect::>(); + entries.sort_by(|(left_domain, _), (right_domain, _)| left_domain.cmp(right_domain)); + let fingerprint = fingerprint_tls_entries(&entries); + + for (domain, (certs, key)) in entries { + let signing_key = rustls::crypto::aws_lc_rs::sign::any_supported_type(&key) + .map_err(|e| Error::other(format!("unsupported private key type for {domain}: {e:?}")))?; + let certified_key = CertifiedKey::new(certs, signing_key); + + if domain.as_str() == "default" { + default_cert = Some(Arc::new(certified_key.clone())); + } else { + cert_resolver + .add(&domain, certified_key) + .map_err(|e| Error::other(format!("failed to add certificate for {domain}: {e:?}")))?; + } + } + + Ok(Self { + cert_resolver, + default_cert, + cert_count, + fingerprint, + }) + } +} + +fn fingerprint_tls_entries(entries: &[(String, (Vec>, PrivateKeyDer<'static>))]) -> u64 { + let mut hasher = DefaultHasher::new(); + + for (domain, (certs, key)) in entries { + hasher.write_usize(domain.len()); + hasher.write(domain.as_bytes()); + hasher.write_usize(certs.len()); + for cert in certs { + hasher.write_usize(cert.as_ref().len()); + hasher.write(cert.as_ref()); + } + hasher.write_usize(key.secret_der().len()); + hasher.write(key.secret_der()); + } + + hasher.finish() +} + +#[derive(Debug)] +pub(crate) struct ReloadableCertResolver { + current: RwLock, +} + +impl ReloadableCertResolver { + pub(crate) fn load_from_directory(cert_dir: &str) -> io::Result> { + let state = ResolverState::load_from_directory(cert_dir)?; + Ok(Arc::new(Self { + current: RwLock::new(state), + })) + } + + pub(crate) fn reload_from_directory(&self, cert_dir: &str) -> io::Result> { + let new_state = ResolverState::load_from_directory(cert_dir)?; + + match self.current.write() { + Ok(mut guard) => { + if guard.fingerprint == new_state.fingerprint { + return Ok(None); + } + let cert_count = new_state.cert_count; + *guard = new_state; + Ok(Some(cert_count)) + } + Err(poisoned) => { + let mut guard = poisoned.into_inner(); + if guard.fingerprint == new_state.fingerprint { + return Ok(None); + } + let cert_count = new_state.cert_count; + *guard = new_state; + Ok(Some(cert_count)) + } + } + } +} + +impl ResolvesServerCert for ReloadableCertResolver { + fn resolve(&self, client_hello: ClientHello) -> Option> { + let guard = match self.current.read() { + Ok(guard) => guard, + Err(poisoned) => poisoned.into_inner(), + }; + + guard + .cert_resolver + .resolve(client_hello) + .or_else(|| guard.default_cert.clone()) + } +} + +pub(crate) fn spawn_cert_reload_loop( + protocol: &'static str, + cert_dir: String, + resolver: Arc, + mut shutdown_rx: watch::Receiver, +) -> Option> { + let enabled = rustfs_utils::get_env_bool(ENV_TLS_RELOAD_ENABLE, DEFAULT_TLS_RELOAD_ENABLE); + if !enabled { + debug!( + protocol, + "TLS certificate hot reload is disabled (set {}=1 to enable)", ENV_TLS_RELOAD_ENABLE + ); + return None; + } + + let interval_secs = rustfs_utils::get_env_u64(ENV_TLS_RELOAD_INTERVAL, DEFAULT_TLS_RELOAD_INTERVAL).max(5); + info!( + protocol, + cert_dir = %cert_dir, + "TLS certificate hot reload enabled, checking every {}s", + interval_secs + ); + + Some(tokio::spawn(async move { + let mut interval = tokio::time::interval(Duration::from_secs(interval_secs)); + interval.set_missed_tick_behavior(MissedTickBehavior::Delay); + interval.tick().await; + + loop { + tokio::select! { + changed = shutdown_rx.changed() => { + match changed { + Ok(()) => { + if *shutdown_rx.borrow() { + info!(protocol, cert_dir = %cert_dir, "TLS certificate hot reload task stopped"); + break; + } + continue; + } + Err(_) => { + info!( + protocol, + cert_dir = %cert_dir, + "TLS certificate hot reload task stopped because the shutdown channel closed" + ); + break; + } + } + } + _ = interval.tick() => {} + } + + match resolver.reload_from_directory(&cert_dir) { + Ok(Some(cert_count)) => { + info!( + protocol, + cert_dir = %cert_dir, + cert_count, + "TLS certificates reloaded successfully" + ); + } + Ok(None) => { + debug!(protocol, cert_dir = %cert_dir, "TLS certificate material unchanged; skipping reload"); + } + Err(e) => { + warn!( + protocol, + cert_dir = %cert_dir, + "TLS certificate reload failed (will retry): {}", + e + ); + } + } + } + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use rcgen::generate_simple_self_signed; + use std::fs; + use tempfile::TempDir; + + fn write_default_cert(dir: &std::path::Path, san: &str) { + let cert = generate_simple_self_signed(vec![san.to_string()]).unwrap(); + fs::write(dir.join(RUSTFS_TLS_CERT), cert.cert.pem()).unwrap(); + fs::write(dir.join(RUSTFS_TLS_KEY), cert.signing_key.serialize_pem()).unwrap(); + } + + #[test] + fn reload_from_directory_replaces_default_certificate() { + let temp_dir = TempDir::new().unwrap(); + write_default_cert(temp_dir.path(), "localhost"); + + let resolver = ReloadableCertResolver::load_from_directory(temp_dir.path().to_str().unwrap()).unwrap(); + let before = { + let guard = resolver.current.read().unwrap(); + guard.default_cert.as_ref().unwrap().clone() + }; + + write_default_cert(temp_dir.path(), "rotated.local"); + + let cert_count = resolver.reload_from_directory(temp_dir.path().to_str().unwrap()).unwrap(); + assert_eq!(cert_count, Some(1)); + + let after = { + let guard = resolver.current.read().unwrap(); + guard.default_cert.as_ref().unwrap().clone() + }; + + assert_ne!(before.cert[0].as_ref(), after.cert[0].as_ref()); + } + + #[test] + fn reload_from_directory_skips_when_material_is_unchanged() { + let temp_dir = TempDir::new().unwrap(); + write_default_cert(temp_dir.path(), "localhost"); + + let resolver = ReloadableCertResolver::load_from_directory(temp_dir.path().to_str().unwrap()).unwrap(); + let outcome = resolver.reload_from_directory(temp_dir.path().to_str().unwrap()).unwrap(); + assert_eq!(outcome, None); + } +} diff --git a/crates/protocols/src/webdav/server.rs b/crates/protocols/src/webdav/server.rs index 0d925cb78..b788f9bba 100644 --- a/crates/protocols/src/webdav/server.rs +++ b/crates/protocols/src/webdav/server.rs @@ -16,6 +16,7 @@ use super::config::{WebDavConfig, WebDavInitError}; use super::driver::WebDavDriver; use crate::common::client::s3::StorageBackend; use crate::common::session::{Protocol, ProtocolPrincipal, SessionContext}; +use crate::tls_hot_reload::{ReloadableCertResolver, spawn_cert_reload_loop}; use bytes::Bytes; use dav_server::DavHandler; use dav_server::fakels::FakeLs; @@ -24,13 +25,12 @@ use hyper::server::conn::http1; use hyper::service::service_fn; use hyper::{Request, Response, StatusCode}; use hyper_util::rt::TokioIo; -use rustfs_config::{RUSTFS_TLS_CERT, RUSTFS_TLS_KEY}; use rustls::ServerConfig; use std::convert::Infallible; use std::net::IpAddr; use std::sync::Arc; use tokio::net::TcpListener; -use tokio::sync::broadcast; +use tokio::sync::{broadcast, watch}; use tokio_rustls::TlsAcceptor; use tracing::{debug, error, info, warn}; @@ -61,29 +61,21 @@ where let listener = TcpListener::bind(self.config.bind_addr).await?; info!("WebDAV server listening on {}", self.config.bind_addr); + let (reload_shutdown_tx, reload_shutdown_rx) = watch::channel(false); // Setup TLS if enabled let tls_acceptor = if self.config.tls_enabled { if let Some(cert_dir) = &self.config.cert_dir { debug!("Enabling WebDAV TLS with certificates from: {}", cert_dir); - let cert_key_pairs = rustfs_utils::load_all_certs_from_directory( - rustfs_utils::CertDirectoryLoadOptions::builder(cert_dir, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY).build(), - ) - .map_err(|e| WebDavInitError::Tls(format!("Failed to load certificates: {}", e)))?; - - if cert_key_pairs.is_empty() { - return Err(WebDavInitError::InvalidConfig("No valid certificates found".into())); - } - - let resolver = rustfs_utils::create_multi_cert_resolver(cert_key_pairs) + let resolver = ReloadableCertResolver::load_from_directory(cert_dir) .map_err(|e| WebDavInitError::Tls(format!("Failed to create certificate resolver: {}", e)))?; + let _reload_task = + spawn_cert_reload_loop("webdav", cert_dir.clone(), resolver.clone(), reload_shutdown_rx.clone()); let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); - let server_config = ServerConfig::builder() - .with_no_client_auth() - .with_cert_resolver(Arc::new(resolver)); + let server_config = ServerConfig::builder().with_no_client_auth().with_cert_resolver(resolver); Some(TlsAcceptor::from(Arc::new(server_config))) } else { @@ -134,11 +126,13 @@ where } _ = shutdown_rx.recv() => { info!("WebDAV server received shutdown signal"); + let _ = reload_shutdown_tx.send(true); break; } } } + let _ = reload_shutdown_tx.send(true); info!("WebDAV server stopped"); Ok(()) }