feat(tls): reload and publish external certificate generations (#8467)

* feat(tls): reload external certificate snapshots coherently

Pin projected certificate generations, validate complete TLS material, coordinate server consumers, and invalidate outbound RPC caches coherently. Add externally managed TLS Secret support and rotation guidance.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* feat(tls): publish server certificates through a fenced admin API

Add opt-in server-leaf publication with dedicated IAM actions, actual TLS
transport checks, immutable generations, durable captured-token CAS and
idempotent receipts. Preserve independent trust and client identity sources
and distinguish visible publication state from the full serving chain.

Validate 64 TLS tests, 41 TLS-runtime tests, the focused IAM policy test,
and eight live local API scenarios including least privilege and restart.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(admin): route certificate handler types through S3 facade

Use the existing admin S3 facade to keep direct s3s imports within the footprint ratchet.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* test(admin): include TLS certificate routes in route matrix

Account for the registered certificate-state GET and certificate publication POST without changing route behavior or weakening matrix assertions.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(tls): reconcile startup trust before skipping reloads

Compare the initialized listener bundle with coherently published outbound material, retain startup trust and client identity when sources disappear, and cover independent startup rotations through real reload passes.

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* refactor(tls): reuse pinned generations and prepared listener state

Co-Authored-By: heihutu <heihutu@gmail.com>
Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

* fix(tls): preserve startup trust source requirements

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>

Cargo.lock

@@ -9686,6 +9686,7 @@ dependencies = [
 name = "rustfs-common"
 version = "1.0.1"
 dependencies = [
+ "arc-swap",
  "hotpath",
  "metrics",
  "smallvec",
@@ -10957,6 +10958,7 @@ dependencies = [
  "thiserror 2.0.21",
  "tokio",
  "tracing",
+ "x509-parser",
 ]
 
 [[package]]

crates/common/Cargo.toml

@@ -34,6 +34,7 @@ hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
 hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
 
 [dependencies]
+arc-swap.workspace = true
 hotpath.workspace = true
 tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }
 tonic = { workspace = true, features = ["gzip", "deflate"] }

crates/common/src/globals.rs

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use arc_swap::ArcSwap;
 use std::collections::HashMap;
-use std::sync::LazyLock;
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, LazyLock};
 use tokio::sync::RwLock;
 use tonic::transport::Channel;
 
@@ -27,6 +28,42 @@ pub static GLOBAL_ROOT_CERT: LazyLock<RwLock<Option<Vec<u8>>>> = LazyLock::new(|
 pub static GLOBAL_MTLS_IDENTITY: LazyLock<RwLock<Option<MtlsIdentityPem>>> = LazyLock::new(|| RwLock::new(None));
 pub static GLOBAL_OUTBOUND_TLS_GENERATION: LazyLock<AtomicU64> = LazyLock::new(|| AtomicU64::new(0));
 
+#[derive(Clone, Default)]
+pub struct OutboundTlsSnapshot {
+    pub generation: u64,
+    pub root_ca_pem: Option<Vec<u8>>,
+    pub mtls_identity: Option<MtlsIdentityPem>,
+}
+
+static GLOBAL_OUTBOUND_TLS_STATE: LazyLock<ArcSwap<OutboundTlsSnapshot>> =
+    LazyLock::new(|| ArcSwap::from_pointee(OutboundTlsSnapshot::default()));
+
+pub fn get_global_outbound_tls_snapshot() -> Arc<OutboundTlsSnapshot> {
+    GLOBAL_OUTBOUND_TLS_STATE.load_full()
+}
+
+/// Publish identity, trust and generation together; the individual globals are legacy views.
+pub async fn publish_global_outbound_tls_snapshot(snapshot: OutboundTlsSnapshot) -> u64 {
+    GLOBAL_OUTBOUND_TLS_STATE.rcu(|current| {
+        if current.generation != 0
+            && current.root_ca_pem == snapshot.root_ca_pem
+            && current.mtls_identity == snapshot.mtls_identity
+        {
+            return Arc::clone(current);
+        }
+        Arc::new(OutboundTlsSnapshot {
+            generation: snapshot.generation.max(current.generation.saturating_add(1)),
+            root_ca_pem: snapshot.root_ca_pem.clone(),
+            mtls_identity: snapshot.mtls_identity.clone(),
+        })
+    });
+    let accepted = get_global_outbound_tls_snapshot();
+    *GLOBAL_ROOT_CERT.write().await = accepted.root_ca_pem.clone();
+    *GLOBAL_MTLS_IDENTITY.write().await = accepted.mtls_identity.clone();
+    GLOBAL_OUTBOUND_TLS_GENERATION.store(accepted.generation, Ordering::Relaxed);
+    accepted.generation
+}
+
 /// Log level to use when reporting cached gRPC connection eviction.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum ConnectionEvictionLogLevel {
@@ -85,17 +122,27 @@ pub async fn get_global_addr() -> String {
 /// # Arguments
 /// * `cert` - A vector of bytes representing the PEM-encoded root CA certificate.
 pub async fn set_global_root_cert(cert: Vec<u8>) {
+    GLOBAL_OUTBOUND_TLS_STATE.rcu(|current| {
+        let mut next = (**current).clone();
+        next.root_ca_pem = Some(cert.clone());
+        Arc::new(next)
+    });
     *GLOBAL_ROOT_CERT.write().await = Some(cert);
 }
 
 /// Clear the global root CA certificate for outbound gRPC clients.
 pub async fn clear_global_root_cert() {
+    GLOBAL_OUTBOUND_TLS_STATE.rcu(|current| {
+        let mut next = (**current).clone();
+        next.root_ca_pem = None;
+        Arc::new(next)
+    });
     *GLOBAL_ROOT_CERT.write().await = None;
 }
 
 /// Get the global root CA certificate for outbound gRPC clients.
 pub async fn get_global_root_cert() -> Option<Vec<u8>> {
-    GLOBAL_ROOT_CERT.read().await.clone()
+    get_global_outbound_tls_snapshot().root_ca_pem.clone()
 }
 
 /// Set the global mTLS identity (cert+key PEM) for outbound gRPC clients.
@@ -105,22 +152,32 @@ pub async fn get_global_root_cert() -> Option<Vec<u8>> {
 /// # Arguments
 /// * `identity` - An optional MtlsIdentityPem struct containing the cert and key PEM.
 pub async fn set_global_mtls_identity(identity: Option<MtlsIdentityPem>) {
+    GLOBAL_OUTBOUND_TLS_STATE.rcu(|current| {
+        let mut next = (**current).clone();
+        next.mtls_identity = identity.clone();
+        Arc::new(next)
+    });
     *GLOBAL_MTLS_IDENTITY.write().await = identity;
 }
 
 /// Get the global mTLS identity for outbound gRPC clients.
 pub async fn get_global_mtls_identity() -> Option<MtlsIdentityPem> {
-    GLOBAL_MTLS_IDENTITY.read().await.clone()
+    get_global_outbound_tls_snapshot().mtls_identity.clone()
 }
 
 /// Set the global outbound TLS generation.
 pub fn set_global_outbound_tls_generation(generation: u64) {
+    GLOBAL_OUTBOUND_TLS_STATE.rcu(|current| {
+        let mut next = (**current).clone();
+        next.generation = generation;
+        Arc::new(next)
+    });
     GLOBAL_OUTBOUND_TLS_GENERATION.store(generation, Ordering::Relaxed);
 }
 
 /// Get the global outbound TLS generation.
 pub fn get_global_outbound_tls_generation() -> u64 {
-    GLOBAL_OUTBOUND_TLS_GENERATION.load(Ordering::Relaxed)
+    get_global_outbound_tls_snapshot().generation
 }
 
 /// Evict a stale/dead connection from the global connection cache.
@@ -194,8 +251,77 @@ pub async fn clear_all_connections() {
 ///
 /// When present, gRPC clients will present this identity to servers requesting/requiring mTLS.
 /// When absent, clients proceed with standard server-authenticated TLS.
-#[derive(Clone, Debug)]
+#[derive(Clone, PartialEq, Eq)]
 pub struct MtlsIdentityPem {
     pub cert_pem: Vec<u8>,
     pub key_pem: Vec<u8>,
 }
+
+impl std::fmt::Debug for MtlsIdentityPem {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("MtlsIdentityPem")
+            .field("cert_bytes", &self.cert_pem.len())
+            .field("key_pem", &"[REDACTED]")
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tls_snapshot_tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn outbound_snapshot_publication_is_coherent_and_skips_unchanged_material() {
+        let start = get_global_outbound_tls_generation().saturating_add(1);
+        let snapshot = OutboundTlsSnapshot {
+            generation: start,
+            root_ca_pem: Some(vec![1]),
+            mtls_identity: Some(MtlsIdentityPem {
+                cert_pem: vec![1],
+                key_pem: vec![1],
+            }),
+        };
+        let first = publish_global_outbound_tls_snapshot(snapshot.clone()).await;
+        assert_eq!(
+            publish_global_outbound_tls_snapshot(OutboundTlsSnapshot {
+                generation: first.saturating_add(1),
+                ..snapshot
+            })
+            .await,
+            first
+        );
+        let writer = tokio::spawn(async move {
+            for tag in 2_u8..=255 {
+                publish_global_outbound_tls_snapshot(OutboundTlsSnapshot {
+                    generation: start.saturating_add(u64::from(tag)),
+                    root_ca_pem: Some(vec![tag]),
+                    mtls_identity: Some(MtlsIdentityPem {
+                        cert_pem: vec![tag],
+                        key_pem: vec![tag],
+                    }),
+                })
+                .await;
+                tokio::task::yield_now().await;
+            }
+        });
+        while !writer.is_finished() {
+            let state = get_global_outbound_tls_snapshot();
+            let identity = state.mtls_identity.as_ref().expect("identity in complete snapshot");
+            assert_eq!(state.root_ca_pem.as_ref().expect("roots in complete snapshot"), &identity.cert_pem);
+            assert_eq!(identity.cert_pem, identity.key_pem);
+            tokio::task::yield_now().await;
+        }
+        writer.await.expect("writer completes");
+    }
+
+    #[test]
+    fn mtls_identity_debug_redacts_private_key() {
+        let identity = MtlsIdentityPem {
+            cert_pem: b"public certificate".to_vec(),
+            key_pem: b"PRIVATE-KEY-MARKER".to_vec(),
+        };
+        let debug = format!("{identity:?}");
+        assert!(!debug.contains("PRIVATE-KEY-MARKER"));
+        assert!(debug.contains("[REDACTED]"));
+    }
+}

crates/ecstore/src/cluster/rpc/client.rs

@@ -24,10 +24,11 @@ use crate::cluster::rpc::{gen_tonic_replay_scope_headers, gen_tonic_signature_he
 #[cfg(test)]
 use crate::cluster::rpc::{tonic_boot_epoch_challenge, tonic_boot_epoch_response_headers};
 use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError};
+#[cfg(test)]
 use crate::runtime::sources as runtime_sources;
 use http::{Request as HttpRequest, Response as HttpResponse, Uri};
 use rustfs_protos::{
-    ChannelClass, create_new_channel, get_channel_for_class,
+    ChannelClass, get_channel_for_class,
     proto_gen::node_service::{
         heal_control_service_client::HealControlServiceClient, node_service_client::NodeServiceClient,
         scanner_control_service_client::ScannerControlServiceClient,
@@ -66,12 +67,9 @@ pub(crate) async fn scanner_control_time_out_client(
     interceptor: TonicInterceptor,
 ) -> crate::error::Result<ScannerControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>> {
     let interceptor = interceptor.with_rpc_audience(addr)?;
-    let channel = match runtime_sources::cached_node_channel(addr).await {
-        Some(channel) => channel,
-        None => create_new_channel(addr)
-            .await
-            .map_err(|err| crate::error::Error::other(err.to_string()))?,
-    };
+    let channel = get_channel_for_class(addr, ChannelClass::Control)
+        .await
+        .map_err(|err| crate::error::Error::other(err.to_string()))?;
     let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
     let limit = rustfs_protos::scoped_dirty_usage::SCOPED_DIRTY_USAGE_MAX_REQUEST_BYTES as usize;
     Ok(ScannerControlServiceClient::with_interceptor(channel, interceptor)
@@ -84,10 +82,7 @@ pub async fn heal_control_time_out_client(
     interceptor: TonicInterceptor,
 ) -> Result<HealControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
     let interceptor = interceptor.with_rpc_audience(addr)?;
-    let channel = match runtime_sources::cached_node_channel(addr).await {
-        Some(channel) => channel,
-        None => create_new_channel(addr).await?,
-    };
+    let channel = get_channel_for_class(addr, ChannelClass::Control).await?;
     let max_message_size = rustfs_protos::HEAL_CONTROL_RPC_MAX_MESSAGE_SIZE;
     let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
     Ok(HealControlServiceClient::with_interceptor(channel, interceptor)
@@ -100,10 +95,7 @@ pub async fn tier_mutation_control_time_out_client(
     interceptor: TonicInterceptor,
 ) -> Result<TierMutationControlServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
     let interceptor = interceptor.with_rpc_audience(addr)?;
-    let channel = match runtime_sources::cached_node_channel(addr).await {
-        Some(channel) => channel,
-        None => create_new_channel(addr).await?,
-    };
+    let channel = get_channel_for_class(addr, ChannelClass::Control).await?;
     let max_message_size = rustfs_protos::TIER_MUTATION_RPC_MAX_MESSAGE_SIZE;
     let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
     Ok(TierMutationControlServiceClient::with_interceptor(channel, interceptor)
@@ -123,17 +115,7 @@ pub async fn node_service_time_out_client_for_class(
     class: ChannelClass,
 ) -> Result<NodeServiceClient<InterceptedService<AuthenticatedChannel, TonicInterceptor>>, Box<dyn Error>> {
     let interceptor = interceptor.with_rpc_audience(addr)?;
-    let channel = match class {
-        ChannelClass::Control => match runtime_sources::cached_node_channel(addr).await {
-            Some(channel) => {
-                debug!("Using cached gRPC channel for: {}", addr);
-                channel
-            }
-            // No cached connection, create new one.
-            None => create_new_channel(addr).await?,
-        },
-        ChannelClass::Bulk => get_channel_for_class(addr, ChannelClass::Bulk).await?,
-    };
+    let channel = get_channel_for_class(addr, class).await?;
 
     let max_message_size = rustfs_protos::internode_rpc_max_message_size();
     let channel = ReplayScopeChannel::new(channel, interceptor.replay_scope_audience());
@@ -546,6 +528,40 @@ mod tests {
     use tracing_opentelemetry::OpenTelemetrySpanExt;
     use tracing_subscriber::{Registry, layer::SubscriberExt};
 
+    #[tokio::test]
+    async fn rpc_client_factories_do_not_reuse_https_channels_without_tls_generation() {
+        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve local address");
+        let addr = format!("https://{}", listener.local_addr().expect("local address"));
+        drop(listener);
+        for factory in 0..4 {
+            let placeholder = tonic::transport::Endpoint::from_static("http://127.0.0.1:1").connect_lazy();
+            runtime_sources::cache_test_node_channel(addr.clone(), placeholder).await;
+            let interceptor = TonicInterceptor::NoOp(NoOpInterceptor);
+            let _result = match factory {
+                0 => scanner_control_time_out_client(&addr, interceptor)
+                    .await
+                    .map(|_| ())
+                    .map_err(|e| e.to_string()),
+                1 => heal_control_time_out_client(&addr, interceptor)
+                    .await
+                    .map(|_| ())
+                    .map_err(|e| e.to_string()),
+                2 => tier_mutation_control_time_out_client(&addr, interceptor)
+                    .await
+                    .map(|_| ())
+                    .map_err(|e| e.to_string()),
+                _ => node_service_time_out_client(&addr, interceptor)
+                    .await
+                    .map(|_| ())
+                    .map_err(|e| e.to_string()),
+            };
+            assert!(
+                !runtime_sources::test_node_channel_is_cached(&addr).await,
+                "factory {factory} must evict an unbound HTTPS cache entry before attempting a fresh dial"
+            );
+        }
+    }
+
     #[derive(Clone)]
     struct EpochProofService {
         audience: String,

crates/ecstore/src/runtime/sources.rs

@@ -50,6 +50,7 @@ use s3s::dto::BucketLifecycleConfiguration;
 use s3s::region::Region;
 use tokio::sync::{OwnedRwLockReadGuard, RwLock};
 use tokio_util::sync::CancellationToken;
+#[cfg(test)]
 use tonic::transport::Channel;
 use uuid::Uuid;
 
@@ -251,10 +252,6 @@ pub(crate) async fn root_disk_threshold_for_erasure_disk() -> Option<u64> {
     }
 }
 
-pub(crate) async fn cached_node_channel(addr: &str) -> Option<Channel> {
-    rustfs_common::cached_connection(addr).await
-}
-
 #[cfg(test)]
 pub(crate) async fn cache_test_node_channel(addr: String, channel: Channel) {
     rustfs_common::cache_connection(addr, channel).await;

crates/policy/src/policy/action.rs

@@ -447,6 +447,10 @@ pub enum AdminAction {
     ServiceFreezeAdminAction,
     #[strum(serialize = "admin:ConfigUpdate")]
     ConfigUpdateAdminAction,
+    #[strum(serialize = "admin:TLSCertificatePublish")]
+    TLSCertificatePublishAdminAction,
+    #[strum(serialize = "admin:TLSCertificateStatus")]
+    TLSCertificateStatusAdminAction,
     #[strum(serialize = "admin:CreateUser")]
     CreateUserAdminAction,
     #[strum(serialize = "admin:DeleteUser")]
@@ -646,6 +650,8 @@ impl AdminAction {
                 | AdminAction::ServiceStopAdminAction
                 | AdminAction::ServiceFreezeAdminAction
                 | AdminAction::ConfigUpdateAdminAction
+                | AdminAction::TLSCertificatePublishAdminAction
+                | AdminAction::TLSCertificateStatusAdminAction
                 | AdminAction::CreateUserAdminAction
                 | AdminAction::DeleteUserAdminAction
                 | AdminAction::ListUsersAdminAction
@@ -941,6 +947,34 @@ mod tests {
         assert!(!Action::AdminAction(get_action).is_match(&Action::AdminAction(set_action)));
     }
 
+    #[test]
+    fn tls_certificate_actions_preserve_least_privilege() {
+        let publish = Action::try_from("admin:TLSCertificatePublish").expect("parse certificate publish action");
+        let status = Action::try_from("admin:TLSCertificateStatus").expect("parse certificate status action");
+        for (action, name) in [
+            (&publish, "admin:TLSCertificatePublish"),
+            (&status, "admin:TLSCertificateStatus"),
+        ] {
+            let Action::AdminAction(admin_action) = action else {
+                panic!("certificate policy must use an admin action");
+            };
+            assert!(admin_action.is_valid());
+            assert_eq!(serde_json::to_string(action).expect("serialize action"), format!("\"{name}\""));
+        }
+        assert!(!status.is_match(&publish));
+        assert!(!publish.is_match(&status));
+        let config = Action::AdminAction(AdminAction::ConfigUpdateAdminAction);
+        let server_info = Action::AdminAction(AdminAction::ServerInfoAdminAction);
+        assert!(!config.is_match(&publish));
+        assert!(!server_info.is_match(&publish));
+        let admin = Action::AdminAction(AdminAction::AllAdminActions);
+        assert!(admin.is_match(&publish));
+        assert!(admin.is_match(&status));
+        let s3 = Action::try_from("s3:*").expect("parse S3 wildcard");
+        assert!(!s3.is_match(&publish));
+        assert!(!s3.is_match(&status));
+    }
+
     #[test]
     fn test_table_catalog_admin_action_is_valid() {
         let get_action = AdminAction::try_from("admin:GetTableCatalog").expect("Should parse GetTableCatalog action");

crates/protos/src/lib.rs

@@ -25,8 +25,8 @@ use rustfs_common::{cache_connection, cached_connection, evict_connection_with_l
 use std::{
     collections::HashMap,
     error::Error,
-    sync::LazyLock,
     sync::atomic::{AtomicU8, AtomicUsize, Ordering},
+    sync::{Arc, LazyLock, Mutex as StdMutex, Weak},
     time::{Duration, Instant},
 };
 use tokio::sync::Mutex;
@@ -59,6 +59,44 @@ const INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET: u8 = 0;
 const INTERNODE_RPC_MSGPACK_ONLY_CACHE_FALSE: u8 = 1;
 const INTERNODE_RPC_MSGPACK_ONLY_CACHE_TRUE: u8 = 2;
 static TLS_GENERATION_CACHE: LazyLock<Mutex<HashMap<String, u64>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+// Lock order: generation metadata, then the common connection cache. The dial
+// registry's standard mutex is released before waiting on any async lock.
+static DIAL_LOCK_REGISTRY: LazyLock<StdMutex<HashMap<String, Weak<Mutex<()>>>>> = LazyLock::new(|| StdMutex::new(HashMap::new()));
+
+struct DialLock {
+    key: String,
+    mutex: Arc<Mutex<()>>,
+}
+
+impl DialLock {
+    fn for_key(key: &str) -> Self {
+        let mut registry = DIAL_LOCK_REGISTRY.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        registry.retain(|_, lock| lock.strong_count() != 0);
+        let mutex = registry.get(key).and_then(Weak::upgrade).unwrap_or_else(|| {
+            let mutex = Arc::new(Mutex::new(()));
+            registry.insert(key.to_string(), Arc::downgrade(&mutex));
+            mutex
+        });
+        Self {
+            key: key.to_string(),
+            mutex,
+        }
+    }
+}
+
+impl Drop for DialLock {
+    fn drop(&mut self) {
+        let mut registry = DIAL_LOCK_REGISTRY.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        if Arc::strong_count(&self.mutex) == 1
+            && registry
+                .get(&self.key)
+                .is_some_and(|slot| Weak::ptr_eq(slot, &Arc::downgrade(&self.mutex)))
+        {
+            registry.remove(&self.key);
+        }
+    }
+}
+
 static INTERNODE_RPC_MSGPACK_ONLY_CACHE: AtomicU8 = AtomicU8::new(INTERNODE_RPC_MSGPACK_ONLY_CACHE_UNSET);
 
 fn enforce_tls_generation_cache_bound(generation_cache: &mut HashMap<String, u64>, generation: u64, addr: &str) {
@@ -2805,6 +2843,53 @@ fn bulk_cache_key(addr: &str, idx: usize) -> String {
     format!("{addr}\u{0}bulk\u{0}{idx}")
 }
 
+async fn cached_channel_for_tls_generation(dial_addr: &str, cache_key: &str) -> Option<Channel> {
+    if !dial_addr.starts_with(RUSTFS_HTTPS_PREFIX) {
+        return cached_connection(cache_key).await;
+    }
+    let mut generations = TLS_GENERATION_CACHE.lock().await;
+    let Some(channel) = cached_connection(cache_key).await else {
+        generations.remove(cache_key);
+        return None;
+    };
+    if generations.get(cache_key).copied() == Some(runtime_sources::outbound_tls_generation()) {
+        return Some(channel);
+    }
+    evict_connection_with_log_level(cache_key, ConnectionEvictionLogLevel::Debug).await;
+    generations.remove(cache_key);
+    runtime_sources::record_stale_grpc_channel_tls_generation();
+    None
+}
+
+async fn cache_dialed_channel(dial_addr: &str, cache_key: &str, channel: &Channel, generation: Option<u64>) -> bool {
+    if !dial_addr.starts_with(RUSTFS_HTTPS_PREFIX) {
+        cache_connection(cache_key.to_string(), channel.clone()).await;
+        return true;
+    }
+    let mut generations = TLS_GENERATION_CACHE.lock().await;
+    if generation != Some(runtime_sources::outbound_tls_generation()) {
+        runtime_sources::record_stale_grpc_channel_tls_generation();
+        return false;
+    }
+    cache_connection(cache_key.to_string(), channel.clone()).await;
+    // Publication can advance while the common cache's async lock is awaited.
+    if generation != Some(runtime_sources::outbound_tls_generation()) {
+        evict_connection_with_log_level(cache_key, ConnectionEvictionLogLevel::Debug).await;
+        generations.remove(cache_key);
+        runtime_sources::record_stale_grpc_channel_tls_generation();
+        return false;
+    }
+    if let Some(generation) = generation {
+        enforce_tls_generation_cache_bound(&mut generations, generation, cache_key);
+        generations.insert(cache_key.to_string(), generation);
+    }
+    true
+}
+
+fn channel_acquisition_timeout(_: tokio::time::error::Elapsed) -> std::io::Error {
+    std::io::Error::new(std::io::ErrorKind::TimedOut, "internode channel acquisition deadline elapsed")
+}
+
 /// Acquire a cached-or-newly-dialed channel for the given peer and channel class.
 ///
 /// For [`ChannelClass::Control`] (or whenever isolation is disabled) this is exactly the legacy
@@ -2813,22 +2898,27 @@ fn bulk_cache_key(addr: &str, idx: usize) -> String {
 /// per-peer bulk pool and dialed on demand, physically isolating large transfers from
 /// control-plane RPCs.
 pub async fn get_channel_for_class(addr: &str, class: ChannelClass) -> Result<Channel, Box<dyn Error>> {
-    if class == ChannelClass::Control || !channel_isolation_enabled() {
-        if let Some(channel) = cached_connection(addr).await {
-            debug!("Using cached control gRPC channel for: {}", addr);
+    let deadline = tokio::time::Instant::now() + internode_connect_timeout();
+    tokio::time::timeout_at(deadline, async {
+        if class == ChannelClass::Control || !channel_isolation_enabled() {
+            if let Some(channel) = cached_channel_for_tls_generation(addr, addr).await {
+                debug!("Using cached control gRPC channel for: {}", addr);
+                return Ok(channel);
+            }
+            return build_channel_until(addr, addr, deadline).await;
+        }
+
+        let pool_size = bulk_channel_pool_size();
+        let idx = BULK_CHANNEL_CURSOR.fetch_add(1, Ordering::Relaxed) % pool_size;
+        let cache_key = bulk_cache_key(addr, idx);
+        if let Some(channel) = cached_channel_for_tls_generation(addr, &cache_key).await {
+            debug!("Using cached bulk gRPC channel {} for: {}", idx, addr);
             return Ok(channel);
         }
-        return create_new_channel(addr).await;
-    }
-
-    let pool_size = bulk_channel_pool_size();
-    let idx = BULK_CHANNEL_CURSOR.fetch_add(1, Ordering::Relaxed) % pool_size;
-    let cache_key = bulk_cache_key(addr, idx);
-    if let Some(channel) = cached_connection(&cache_key).await {
-        debug!("Using cached bulk gRPC channel {} for: {}", idx, addr);
-        return Ok(channel);
-    }
-    build_channel(addr, &cache_key).await
+        build_channel_until(addr, &cache_key, deadline).await
+    })
+    .await
+    .map_err(channel_acquisition_timeout)?
 }
 
 /// Creates a new gRPC channel with optimized keepalive settings for cluster resilience.
@@ -2852,6 +2942,45 @@ pub async fn create_new_channel(addr: &str) -> Result<Channel, Box<dyn Error>> {
 /// for isolated bulk channels (see [`bulk_cache_key`]), letting several physically distinct
 /// channels to the same peer be cached independently.
 async fn build_channel(dial_addr: &str, cache_key: &str) -> Result<Channel, Box<dyn Error>> {
+    let deadline = tokio::time::Instant::now() + internode_connect_timeout();
+    build_channel_until(dial_addr, cache_key, deadline).await
+}
+
+async fn build_channel_until(
+    dial_addr: &str,
+    cache_key: &str,
+    deadline: tokio::time::Instant,
+) -> Result<Channel, Box<dyn Error>> {
+    tokio::time::timeout_at(deadline, async {
+        if let Some(channel) = cached_channel_for_tls_generation(dial_addr, cache_key).await {
+            return Ok(channel);
+        }
+        let slot = DialLock::for_key(cache_key);
+        let _guard = slot.mutex.lock().await;
+        loop {
+            // A successful concurrent miss has already populated this exact key.
+            if let Some(channel) = cached_channel_for_tls_generation(dial_addr, cache_key).await {
+                return Ok(channel);
+            }
+            let (channel, generation) = dial_channel(dial_addr, cache_key).await?;
+            if cache_dialed_channel(dial_addr, cache_key, &channel, generation).await
+                && let Some(channel) = cached_channel_for_tls_generation(dial_addr, cache_key).await
+            {
+                debug!(
+                    "Successfully created and cached gRPC channel to: {} (cache key: {})",
+                    dial_addr, cache_key
+                );
+                return Ok(channel);
+            }
+            // A generation published during dial cannot be returned; rebuilding
+            // shares the original deadline with lock wait and all earlier dials.
+        }
+    })
+    .await
+    .map_err(channel_acquisition_timeout)?
+}
+
+async fn dial_channel(dial_addr: &str, cache_key: &str) -> Result<(Channel, Option<u64>), Box<dyn Error>> {
     debug!("Creating new gRPC channel to: {} (cache key: {})", dial_addr, cache_key);
     let dial_started_at = Instant::now();
     let connect_timeout = internode_connect_timeout();
@@ -2887,18 +3016,10 @@ async fn build_channel(dial_addr: &str, cache_key: &str) -> Result<Channel, Box<
         connector = connector.initial_connection_window_size(conn_window);
     }
 
-    let outbound_tls = runtime_sources::outbound_tls_state().await;
-    let generation = outbound_tls.generation.0;
-    let mut stale_generation = false;
-    {
-        let generation_cache = TLS_GENERATION_CACHE.lock().await;
-        if let Some(cached_generation) = generation_cache.get(cache_key)
-            && *cached_generation != generation
-        {
-            stale_generation = true;
-        }
-    }
+    let mut generation = None;
     if dial_addr.starts_with(RUSTFS_HTTPS_PREFIX) {
+        let outbound_tls = runtime_sources::outbound_tls_state().await;
+        generation = Some(outbound_tls.generation.0);
         if let Some(cert_pem) = outbound_tls.root_ca_pem.as_ref() {
             let ca = Certificate::from_pem(cert_pem);
             // Derive the hostname from the HTTPS URL for TLS hostname verification.
@@ -2946,21 +3067,7 @@ async fn build_channel(dial_addr: &str, cache_key: &str) -> Result<Channel, Box<
         }
     };
 
-    cache_connection(cache_key.to_string(), channel.clone()).await;
-    {
-        let mut generation_cache = TLS_GENERATION_CACHE.lock().await;
-        enforce_tls_generation_cache_bound(&mut generation_cache, generation, cache_key);
-        generation_cache.insert(cache_key.to_string(), generation);
-    }
-    if stale_generation {
-        runtime_sources::record_stale_grpc_channel_tls_generation();
-    }
-
-    debug!(
-        "Successfully created and cached gRPC channel to: {} (cache key: {})",
-        dial_addr, cache_key
-    );
-    Ok(channel)
+    Ok((channel, generation))
 }
 
 /// Evict a connection from the cache after a failure.
@@ -3021,6 +3128,8 @@ mod tests {
     use prost::Message as _;
     use std::sync::Mutex;
 
+    static TLS02_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
     static INTERNODE_RPC_MSGPACK_ONLY_ENV_LOCK: Mutex<()> = Mutex::new(());
 
     #[derive(Clone, PartialEq, prost::Message)]
@@ -3144,6 +3253,243 @@ mod tests {
         assert_eq!(manifest_fields, proto_bin_json_fields("Response"));
     }
 
+    #[tokio::test]
+    async fn tls02_stale_https_control_and_bulk_cache_entries_are_evicted() {
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let addr = "https://tls02-stale-cache.invalid:9000";
+        let generation = rustfs_tls_runtime::load_global_outbound_tls_generation().0;
+        let stale_generation = generation.checked_add(1).unwrap_or(0);
+        for key in [addr.to_string(), bulk_cache_key(addr, 0)] {
+            cache_connection(key.clone(), Endpoint::from_static("http://127.0.0.1:1").connect_lazy()).await;
+            TLS_GENERATION_CACHE.lock().await.insert(key.clone(), stale_generation);
+            assert!(
+                cached_channel_for_tls_generation(addr, &key).await.is_none(),
+                "TLS02_INVARIANT stale HTTPS cache entry cannot be returned"
+            );
+            assert!(
+                !rustfs_common::has_cached_connection(&key).await,
+                "TLS02_INVARIANT stale entry is removed from the real connection cache"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn tls02_missing_https_generation_control_and_bulk_cache_entries_are_evicted() {
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let addr = "https://tls02-missing-generation.invalid:9000";
+        for key in [addr.to_string(), bulk_cache_key(addr, 0)] {
+            cache_connection(key.clone(), Endpoint::from_static("http://127.0.0.1:1").connect_lazy()).await;
+            TLS_GENERATION_CACHE.lock().await.remove(&key);
+            assert!(
+                cached_channel_for_tls_generation(addr, &key).await.is_none(),
+                "TLS02_INVARIANT HTTPS channel without a generation cannot be returned"
+            );
+            assert!(
+                !rustfs_common::has_cached_connection(&key).await,
+                "TLS02_INVARIANT unbound entry is removed from the real connection cache"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn tls02_current_https_control_and_bulk_cache_entries_are_reused() {
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let addr = "https://tls02-current-generation.invalid:9000";
+        let generation = rustfs_tls_runtime::load_global_outbound_tls_generation().0;
+        for key in [addr.to_string(), bulk_cache_key(addr, 0)] {
+            cache_connection(key.clone(), Endpoint::from_static("http://127.0.0.1:1").connect_lazy()).await;
+            TLS_GENERATION_CACHE.lock().await.insert(key.clone(), generation);
+            assert!(
+                cached_channel_for_tls_generation(addr, &key).await.is_some(),
+                "TLS02_INVARIANT HTTPS channel at the canonical generation is reused"
+            );
+            assert!(rustfs_common::has_cached_connection(&key).await);
+            evict_connection_with_log_level(&key, ConnectionEvictionLogLevel::Debug).await;
+            TLS_GENERATION_CACHE.lock().await.remove(&key);
+        }
+    }
+
+    #[tokio::test]
+    async fn tls02_http_control_and_bulk_cache_ignore_tls_generation() {
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let addr = "http://tls02-http-cache.invalid:9000";
+        let generation = rustfs_tls_runtime::load_global_outbound_tls_generation().0;
+        for key in [addr.to_string(), bulk_cache_key(addr, 0)] {
+            cache_connection(key.clone(), Endpoint::from_static("http://127.0.0.1:1").connect_lazy()).await;
+            TLS_GENERATION_CACHE
+                .lock()
+                .await
+                .insert(key.clone(), generation.checked_add(1).unwrap_or(0));
+            assert!(
+                cached_channel_for_tls_generation(addr, &key).await.is_some(),
+                "TLS02_INVARIANT HTTP cache is independent of TLS generations"
+            );
+            TLS_GENERATION_CACHE.lock().await.remove(&key);
+            assert!(
+                cached_channel_for_tls_generation(addr, &key).await.is_some(),
+                "TLS02_INVARIANT HTTP cache needs no TLS generation metadata"
+            );
+            evict_connection_with_log_level(&key, ConnectionEvictionLogLevel::Debug).await;
+        }
+    }
+
+    #[tokio::test]
+    async fn tls02_captured_identity_is_immutable_and_late_dial_keeps_its_actual_generation() {
+        use rustfs_common::MtlsIdentityPem;
+        use rustfs_tls_runtime::{OutboundTlsMaterial, TlsGeneration, publish_global_outbound_tls_state};
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let previous = runtime_sources::outbound_tls_state().await;
+        let first_generation = previous.generation.0.checked_add(1).expect("test generation capacity");
+        let first = OutboundTlsMaterial {
+            root_ca_pem: b"root-a".to_vec(),
+            mtls_identity: Some(MtlsIdentityPem {
+                cert_pem: b"certificate-a".to_vec(),
+                key_pem: b"test-key-a".to_vec(),
+            }),
+        };
+        publish_global_outbound_tls_state(TlsGeneration(first_generation), &first).await;
+        let captured = runtime_sources::outbound_tls_state().await;
+        let second = OutboundTlsMaterial {
+            root_ca_pem: b"root-b".to_vec(),
+            mtls_identity: Some(MtlsIdentityPem {
+                cert_pem: b"certificate-b".to_vec(),
+                key_pem: b"test-key-b".to_vec(),
+            }),
+        };
+        publish_global_outbound_tls_state(TlsGeneration(first_generation + 1), &second).await;
+        let current = runtime_sources::outbound_tls_state().await;
+        let addr = "https://tls02-late-dial.invalid:9000";
+        let channel = Endpoint::from_static("http://127.0.0.1:1").connect_lazy();
+        let cached = cache_dialed_channel(addr, addr, &channel, Some(captured.generation.0)).await;
+        let original_key = captured.mtls_identity.as_ref().expect("captured identity").key_pem.clone();
+        let current_key = current.mtls_identity.as_ref().expect("current identity").key_pem.clone();
+        let restore = OutboundTlsMaterial {
+            root_ca_pem: previous.root_ca_pem.unwrap_or_default(),
+            mtls_identity: previous.mtls_identity,
+        };
+        publish_global_outbound_tls_state(TlsGeneration(first_generation + 2), &restore).await;
+        assert_eq!(
+            original_key, b"test-key-a",
+            "TLS02_INVARIANT captured identity cannot change with publication"
+        );
+        assert_eq!(captured.root_ca_pem.as_deref(), Some(b"root-a".as_slice()));
+        assert_eq!(current_key, b"test-key-b");
+        assert!(!cached, "TLS02_INVARIANT late dial cannot be relabeled with the current generation");
+        assert!(
+            !rustfs_common::has_cached_connection(addr).await,
+            "TLS02_INVARIANT late stale dial is never cached"
+        );
+        assert!(!TLS_GENERATION_CACHE.lock().await.contains_key(addr));
+    }
+
+    #[tokio::test]
+    async fn tls02_concurrent_control_and_bulk_misses_merge_successful_real_dials() {
+        use tonic::codegen::tokio_stream::{StreamExt, wrappers::TcpListenerStream};
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind HTTP/2 test server");
+        let addr = format!("http://{}", listener.local_addr().expect("test server address"));
+        let accepted = Arc::new(AtomicUsize::new(0));
+        let counter = accepted.clone();
+        let incoming = TcpListenerStream::new(listener).map(move |stream| {
+            if stream.is_ok() {
+                counter.fetch_add(1, Ordering::SeqCst);
+            }
+            stream
+        });
+        let server = tokio::spawn(async move {
+            tonic::transport::Server::builder()
+                .add_routes(tonic::service::Routes::default())
+                .serve_with_incoming(incoming)
+                .await
+        });
+        for key in [addr.clone(), bulk_cache_key(&addr, 0)] {
+            let mut tasks = Vec::new();
+            for _ in 0..32 {
+                let dial_addr = addr.clone();
+                let cache_key = key.clone();
+                tasks.push(tokio::spawn(async move {
+                    build_channel(&dial_addr, &cache_key).await.map_err(|error| error.to_string())
+                }));
+            }
+            for task in tasks {
+                let channel = task.await.expect("channel acquisition task").expect("real HTTP/2 dial");
+                drop(channel);
+            }
+            assert!(rustfs_common::has_cached_connection(&key).await);
+            assert!(
+                !DIAL_LOCK_REGISTRY.lock().expect("dial registry").contains_key(&key),
+                "TLS02_INVARIANT completed miss groups leave no lock entries"
+            );
+        }
+        // Tonic returns its channel before the server task necessarily polls the
+        // already-established sockets out of the listener's accept queue.
+        tokio::time::sleep(Duration::from_millis(20)).await;
+        let connections = accepted.load(Ordering::SeqCst);
+        evict_connection_with_log_level(&addr, ConnectionEvictionLogLevel::Debug).await;
+        evict_connection_with_log_level(&bulk_cache_key(&addr, 0), ConnectionEvictionLogLevel::Debug).await;
+        server.abort();
+        let _ = server.await;
+        assert_eq!(connections, 2, "TLS02_INVARIANT 32 misses per control/bulk key dial once each");
+    }
+
+    #[tokio::test]
+    async fn tls02_concurrent_waiters_share_bounded_total_deadlines_and_release_registry() {
+        let _test_guard = TLS02_TEST_LOCK.lock().await;
+        let addr = "http://tls02-contended-dial.invalid:9000";
+        let owner = DialLock::for_key(addr);
+        let guard = owner.mutex.lock().await;
+        let mut tasks = Vec::new();
+        for _ in 0..8 {
+            tasks.push(tokio::spawn(async move {
+                let deadline = tokio::time::Instant::now() + Duration::from_millis(40);
+                let result = build_channel_until(addr, addr, deadline).await;
+                result.err().is_some_and(|error| {
+                    error
+                        .downcast_ref::<std::io::Error>()
+                        .is_some_and(|error| error.kind() == std::io::ErrorKind::TimedOut)
+                })
+            }));
+        }
+        let completed = tokio::time::timeout(Duration::from_millis(250), async {
+            for task in tasks {
+                assert!(
+                    task.await.expect("deadline waiter"),
+                    "TLS02_INVARIANT wait and rebuild share a typed timeout"
+                );
+            }
+        })
+        .await;
+        drop(guard);
+        drop(owner);
+        assert!(
+            completed.is_ok(),
+            "TLS02_INVARIANT eight waiters cannot accumulate eight connect timeouts"
+        );
+        assert!(
+            !DIAL_LOCK_REGISTRY.lock().expect("dial registry").contains_key(addr),
+            "TLS02_INVARIANT cancelled waiters release the lock registry"
+        );
+    }
+
+    #[test]
+    fn tls02_dial_registry_reclaims_unique_keys_and_preserves_live_lock_identity() {
+        for index in 0..1000 {
+            let key = format!("http://tls02-registry-{index}.invalid");
+            let first = DialLock::for_key(&key);
+            let second = DialLock::for_key(&key);
+            assert!(Arc::ptr_eq(&first.mutex, &second.mutex), "TLS02_INVARIANT one live lock per cache key");
+            drop(first);
+            assert!(DIAL_LOCK_REGISTRY.lock().expect("dial registry").contains_key(&key));
+            drop(second);
+            assert!(
+                !DIAL_LOCK_REGISTRY.lock().expect("dial registry").contains_key(&key),
+                "TLS02_INVARIANT last owner removes its own weak entry"
+            );
+        }
+    }
+
     #[test]
     fn enforce_tls_generation_cache_bound_evicts_when_retained_entries_still_full() {
         let mut cache = HashMap::new();

crates/protos/src/runtime_sources.rs

@@ -22,6 +22,10 @@ pub(crate) async fn outbound_tls_state() -> GlobalPublishedOutboundTlsState {
     load_global_outbound_tls_state().await
 }
 
+pub(crate) fn outbound_tls_generation() -> u64 {
+    rustfs_tls_runtime::load_global_outbound_tls_generation().0
+}
+
 pub(crate) fn record_stale_grpc_channel_tls_generation() {
     record_tls_consumer_stale_generation(PROTOS_GRPC_CHANNEL_TLS_CONSUMER);
 }

crates/tls-runtime/Cargo.toml

@@ -46,6 +46,7 @@ sha2.workspace = true
 thiserror.workspace = true
 tokio = { workspace = true, features = ["fs", "rt", "sync", "time"] }
 tracing.workspace = true
+x509-parser.workspace = true
 
 [dev-dependencies]
 rcgen.workspace = true

crates/tls-runtime/src/certs.rs

@@ -429,6 +429,209 @@ pub fn load_all_certs_from_directory(
     Ok(cert_key_pairs)
 }
 
+/// Strict runtime loader: a discovered identity must never disappear through
+/// best-effort filtering. The diagnostic inspection API remains independent.
+pub fn load_all_certs_from_directory_strict(
+    options: CertDirectoryLoadOptions,
+) -> Result<HashMap<String, (Vec<CertificateDer<'static>>, PrivateKeyDer<'static>)>, crate::error::TlsRuntimeError> {
+    use crate::error::TlsRuntimeError;
+    options.validate()?;
+    let dir = &options.dir_path;
+    let mut pairs = HashMap::new();
+    if let Some(pair) = read_cert_key_pair_strict(dir, &options.cert_filename, &options.key_filename, "default", true)? {
+        pairs.insert("default".to_string(), pair);
+    }
+    for entry in fs::read_dir(dir).map_err(|source| TlsRuntimeError::SourceIo {
+        path: dir.clone(),
+        source,
+    })? {
+        let entry = entry.map_err(|source| TlsRuntimeError::SourceIo {
+            path: dir.clone(),
+            source,
+        })?;
+        let path = entry.path();
+        let name = entry.file_name();
+        let name = name.to_str().ok_or_else(|| TlsRuntimeError::InvalidCertificate {
+            name: "SNI directory".to_string(),
+            reason: "directory name is not UTF-8".to_string(),
+        })?;
+        if !is_discoverable_cert_domain_dir(name) {
+            continue;
+        }
+        let metadata = fs::metadata(&path).map_err(|source| TlsRuntimeError::SourceIo {
+            path: path.clone(),
+            source,
+        })?;
+        if !metadata.is_dir() {
+            continue;
+        }
+        if let Some(pair) = read_cert_key_pair_strict(&path, &options.cert_filename, &options.key_filename, name, true)? {
+            pairs.insert(name.to_string(), pair);
+        }
+    }
+    Ok(pairs)
+}
+
+pub(crate) fn read_optional_tls_file(path: &Path) -> Result<Option<Vec<u8>>, crate::error::TlsRuntimeError> {
+    use crate::error::TlsRuntimeError;
+    match fs::symlink_metadata(path) {
+        Ok(_) => fs::read(path).map(Some).map_err(|source| TlsRuntimeError::SourceIo {
+            path: path.to_path_buf(),
+            source,
+        }),
+        Err(error) if error.kind() == io::ErrorKind::NotFound => {
+            // A disappearing pinned generation is an error, not optional absence.
+            if let Some(parent) = path.parent() {
+                fs::read_dir(parent).map_err(|source| TlsRuntimeError::SourceIo {
+                    path: parent.to_path_buf(),
+                    source,
+                })?;
+            }
+            Ok(None)
+        }
+        Err(source) => Err(TlsRuntimeError::SourceIo {
+            path: path.to_path_buf(),
+            source,
+        }),
+    }
+}
+
+pub(crate) fn parse_certificate_pem(
+    pem: &[u8],
+    name: &str,
+) -> Result<Vec<CertificateDer<'static>>, crate::error::TlsRuntimeError> {
+    use crate::error::TlsRuntimeError;
+    let certs = CertificateDer::pem_slice_iter(pem)
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|error| TlsRuntimeError::InvalidCertificate {
+            name: name.to_string(),
+            reason: error.to_string(),
+        })?;
+    if certs.is_empty() {
+        return Err(TlsRuntimeError::InvalidCertificate {
+            name: name.to_string(),
+            reason: "empty certificate PEM".to_string(),
+        });
+    }
+    for cert in &certs {
+        let (remaining, _) =
+            x509_parser::parse_x509_certificate(cert.as_ref()).map_err(|error| TlsRuntimeError::InvalidCertificate {
+                name: name.to_string(),
+                reason: error.to_string(),
+            })?;
+        if !remaining.is_empty() {
+            return Err(TlsRuntimeError::InvalidCertificate {
+                name: name.to_string(),
+                reason: "trailing certificate DER".to_string(),
+            });
+        }
+    }
+    Ok(certs)
+}
+
+pub(crate) fn parse_private_key_pem(pem: &[u8], name: &str) -> Result<PrivateKeyDer<'static>, crate::error::TlsRuntimeError> {
+    PrivateKeyDer::from_pem_slice(pem).map_err(|error| crate::error::TlsRuntimeError::InvalidPrivateKey {
+        name: name.to_string(),
+        reason: error.to_string(),
+    })
+}
+
+pub(crate) fn read_cert_key_pair_strict(
+    dir: &Path,
+    cert_filename: &str,
+    key_filename: &str,
+    name: &str,
+    server: bool,
+) -> Result<Option<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>)>, crate::error::TlsRuntimeError> {
+    let cert = read_optional_tls_file(&dir.join(cert_filename))?;
+    let key = read_optional_tls_file(&dir.join(key_filename))?;
+    match (cert, key) {
+        (None, None) => Ok(None),
+        (Some(cert), Some(key)) => {
+            let certs = parse_certificate_pem(&cert, name)?;
+            let key = parse_private_key_pem(&key, name)?;
+            validate_certified_key(&certs, &key, name, server)?;
+            Ok(Some((certs, key)))
+        }
+        _ => Err(crate::error::TlsRuntimeError::IncompleteIdentity { name: name.to_string() }),
+    }
+}
+
+/// Validate before publication, including the default certificate that bypasses
+/// rustls' SNI name checks.
+pub fn validate_certified_key(
+    certs: &[CertificateDer<'static>],
+    key: &PrivateKeyDer<'static>,
+    name: &str,
+    server: bool,
+) -> Result<CertifiedKey, crate::error::TlsRuntimeError> {
+    use crate::error::TlsRuntimeError;
+    if certs.is_empty() {
+        return Err(TlsRuntimeError::InvalidCertificate {
+            name: name.to_string(),
+            reason: "empty certificate chain".to_string(),
+        });
+    }
+    for (index, cert) in certs.iter().enumerate() {
+        let (remaining, parsed) =
+            x509_parser::parse_x509_certificate(cert.as_ref()).map_err(|error| TlsRuntimeError::InvalidCertificate {
+                name: name.to_string(),
+                reason: error.to_string(),
+            })?;
+        if !remaining.is_empty() {
+            return Err(TlsRuntimeError::InvalidCertificate {
+                name: name.to_string(),
+                reason: "trailing certificate DER".to_string(),
+            });
+        }
+        if index == 0 {
+            if !parsed.validity().is_valid() {
+                return Err(TlsRuntimeError::CertificateValidity { name: name.to_string() });
+            }
+            if parsed
+                .extended_key_usage()
+                .map_err(|error| TlsRuntimeError::InvalidCertificate {
+                    name: name.to_string(),
+                    reason: error.to_string(),
+                })?
+                .is_some_and(|eku| {
+                    if server {
+                        !eku.value.server_auth
+                    } else {
+                        !eku.value.client_auth
+                    }
+                })
+            {
+                return Err(if server {
+                    TlsRuntimeError::InvalidServerEku { name: name.to_string() }
+                } else {
+                    TlsRuntimeError::InvalidClientEku { name: name.to_string() }
+                });
+            }
+        }
+    }
+    let certified = CertifiedKey::from_der(certs.to_vec(), key.clone_key(), &rustls::crypto::aws_lc_rs::default_provider())
+        .map_err(|source| TlsRuntimeError::InvalidCertifiedKey {
+            name: name.to_string(),
+            source,
+        })?;
+    certified
+        .keys_match()
+        .map_err(|source| TlsRuntimeError::InvalidCertifiedKey {
+            name: name.to_string(),
+            source,
+        })?;
+    if server && name != "default" {
+        let mut sni = ResolvesServerCertUsingSni::new();
+        sni.add(name, certified.clone())
+            .map_err(|source| TlsRuntimeError::InvalidCertifiedKey {
+                name: name.to_string(),
+                source,
+            })?;
+    }
+    Ok(certified)
+}
+
 fn load_cert_key_pair(cert_path: &str, key_path: &str) -> io::Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>)> {
     let certs = load_certs(cert_path)?;
     let key = load_private_key(key_path)?;
@@ -494,10 +697,7 @@ pub fn create_multi_cert_resolver(
     let mut default_cert = None;
 
     for (domain, (certs, key)) in cert_key_pairs {
-        let signing_key = rustls::crypto::aws_lc_rs::sign::any_supported_type(&key)
-            .map_err(|e| certs_error(format!("unsupported private key types:{domain}, err:{e:?}")))?;
-
-        let certified_key = CertifiedKey::new(certs, signing_key);
+        let certified_key = validate_certified_key(&certs, &key, &domain, true).map_err(io::Error::other)?;
         if domain == "default" {
             default_cert = Some(Arc::new(certified_key.clone()));
         } else {

crates/tls-runtime/src/error.rs

@@ -23,6 +23,38 @@ pub enum TlsRuntimeError {
     DirectoryNotFound { path: PathBuf },
     #[error("TLS path is not a directory: {path}")]
     NotADirectory { path: PathBuf },
+    #[error("TLS source I/O error at {path}: {source}")]
+    SourceIo {
+        path: PathBuf,
+        #[source]
+        source: std::io::Error,
+    },
+    #[error("incomplete TLS certificate/private key pair for {name}")]
+    IncompleteIdentity { name: String },
+    #[error("invalid TLS certificate material for {name}: {reason}")]
+    InvalidCertificate { name: String, reason: String },
+    #[error("invalid TLS private key for {name}: {reason}")]
+    InvalidPrivateKey { name: String, reason: String },
+    #[error("TLS certificate/private key validation failed for {name}: {source}")]
+    InvalidCertifiedKey {
+        name: String,
+        #[source]
+        source: rustls::Error,
+    },
+    #[error("invalid TLS trust anchor in {name}: {source}")]
+    InvalidTrustAnchor {
+        name: String,
+        #[source]
+        source: rustls::Error,
+    },
+    #[error("TLS certificate for {name} is outside its validity interval")]
+    CertificateValidity { name: String },
+    #[error("TLS certificate for {name} does not permit server authentication")]
+    InvalidServerEku { name: String },
+    #[error("TLS certificate for {name} does not permit client authentication")]
+    InvalidClientEku { name: String },
+    #[error("TLS source loading task failed: {0}")]
+    SourceTaskJoin(#[from] tokio::task::JoinError),
     #[error("TLS material error: {0}")]
     Material(String),
     #[error("TLS publication error: {0}")]

crates/tls-runtime/src/lib.rs

@@ -28,7 +28,8 @@ pub mod state;
 pub use certs::{
     CertDirectoryLoadOptions, TlsCertPairInspection, TlsCertPairStatus, TlsDirectoryInspection, TlsDomainInspection,
     WebPkiClientVerifierOptions, build_webpki_client_verifier, create_multi_cert_resolver, inspect_cert_directory,
-    load_all_certs_from_directory, load_cert_bundle_der_bytes, load_certs, load_private_key,
+    load_all_certs_from_directory, load_all_certs_from_directory_strict, load_cert_bundle_der_bytes, load_certs,
+    load_private_key, validate_certified_key,
 };
 pub use config::{ReloadApplyHint, ReloadDetectMode, TlsReloadOptions};
 pub use coordinator::{TlsConsumer, TlsReloadCoordinator};
@@ -45,7 +46,7 @@ pub use outbound::{
     load_global_outbound_tls_state, publish_global_outbound_tls_state, summarize_global_outbound_tls_state,
 };
 pub use server::{ReloadableServerCertResolver, spawn_server_cert_reload_loop};
-pub use source::{TlsFileLayout, TlsSource, TlsSourceKind};
+pub use source::{TlsFileLayout, TlsSource, TlsSourceKind, TlsSourceSnapshot};
 pub use state::OutboundOnlySnapshotArgs;
 pub use state::{TlsGeneration, TlsPublishedState, TlsReloadRuntimeState, TlsRuntimeStatusSnapshot};
 pub use state::{TlsRuntimeConsumerSection, TlsRuntimeOutboundSection, TlsRuntimeRuntimeSection, TlsRuntimeServerSection};
@@ -111,6 +112,8 @@ mod tests {
         let coordinator = TlsReloadCoordinator::new(source.clone(), TlsReloadOptions::default());
         let snapshot = TlsMaterialSnapshot {
             source,
+            server_client_ca_pem: None,
+            material_files: std::collections::BTreeSet::new(),
             server: Some(ServerTlsMaterial::MultiCert {
                 cert_key_pairs: HashMap::new(),
             }),

crates/tls-runtime/src/material.rs

@@ -12,15 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::certs::{CertDirectoryLoadOptions, load_all_certs_from_directory, load_certs, load_private_key};
+use crate::certs::{
+    CertDirectoryLoadOptions, load_all_certs_from_directory_strict, parse_certificate_pem, parse_private_key_pem,
+    read_optional_tls_file, validate_certified_key,
+};
 use crate::error::TlsRuntimeError;
 use crate::fingerprint::TlsFingerprint;
-use crate::source::TlsSource;
+use crate::source::{TlsSource, TlsSourceSnapshot};
 use rustfs_common::MtlsIdentityPem;
-use rustls::pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject};
-use std::collections::HashMap;
-use std::io::Cursor;
-use std::io::ErrorKind;
+use rustls::pki_types::{CertificateDer, PrivateKeyDer};
+use std::collections::{BTreeSet, HashMap};
 use std::path::Path;
 
 #[derive(Debug, Clone)]
@@ -40,115 +41,196 @@ pub enum ServerTlsMaterial {
     },
 }
 
+impl ServerTlsMaterial {
+    /// Revalidate an accepted generation without reading the filesystem.
+    pub fn validate(&self) -> Result<(), TlsRuntimeError> {
+        match self {
+            Self::SingleCert { certs, key } => {
+                crate::certs::validate_certified_key(certs, key, "default", true)?;
+            }
+            Self::MultiCert { cert_key_pairs } => {
+                for (name, (certs, key)) in cert_key_pairs {
+                    crate::certs::validate_certified_key(certs, key, name, true)?;
+                }
+            }
+        }
+        Ok(())
+    }
+}
+
+impl Clone for ServerTlsMaterial {
+    fn clone(&self) -> Self {
+        match self {
+            Self::SingleCert { certs, key } => Self::SingleCert {
+                certs: certs.clone(),
+                key: key.clone_key(),
+            },
+            Self::MultiCert { cert_key_pairs } => Self::MultiCert {
+                cert_key_pairs: cert_key_pairs
+                    .iter()
+                    .map(|(name, (certs, key))| (name.clone(), (certs.clone(), key.clone_key())))
+                    .collect(),
+            },
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct TlsMaterialSnapshot {
+    /// The resolved directory generation used by every consumer of this snapshot.
     pub source: TlsSource,
     pub server: Option<ServerTlsMaterial>,
     pub outbound: OutboundTlsMaterial,
+    pub server_client_ca_pem: Option<Vec<u8>>,
+    pub material_files: BTreeSet<String>,
     pub fingerprint: TlsFingerprint,
 }
 
 impl TlsMaterialSnapshot {
     pub async fn load(source: &TlsSource) -> Result<Self, TlsRuntimeError> {
-        let base_dir = source.validate_directory()?.to_path_buf();
+        let source = source.clone();
+        tokio::task::spawn_blocking(move || Self::load_from_snapshot(&source.snapshot()?)).await?
+    }
 
-        let server = load_server_material(source, &base_dir)?;
+    /// Load all consumers' material from one previously pinned directory generation.
+    pub fn load_from_snapshot(snapshot: &TlsSourceSnapshot) -> Result<Self, TlsRuntimeError> {
+        Self::load_from_snapshot_with_server_directory(snapshot, None)
+    }
 
-        let public_ca_path = base_dir.join(&source.layout.public_ca_filename);
-        let client_ca_path = base_dir.join(&source.fallback_ca_filename);
-        let client_cert_path = base_dir.join(&source.layout.client_cert_filename);
-        let client_key_path = base_dir.join(&source.layout.client_key_filename);
+    /// Load a server-only publication without replacing the pinned trust or client identity source.
+    pub fn load_from_snapshot_with_server_directory(
+        snapshot: &TlsSourceSnapshot,
+        server_directory: Option<&Path>,
+    ) -> Result<Self, TlsRuntimeError> {
+        snapshot.validate_available()?;
+        let source = snapshot.resolved_source();
+        let base_dir = snapshot.base_dir();
+        let server = load_server_material(source, server_directory.unwrap_or(base_dir))?;
+        Self::load_from_snapshot_with_loaded_server(snapshot, server)
+    }
 
-        let public_ca_pem = tokio::fs::read(&public_ca_path).await.ok();
-        let client_ca_pem = tokio::fs::read(&client_ca_path).await.ok();
-        let root_ca_pem = combine_optional_pem(public_ca_pem.as_deref(), client_ca_pem.as_deref());
+    /// Validate an already decoded default server chain while reading trust and client identity from the pinned source.
+    /// Serving validity, EKU and key matching are checked here even when the publication ledger was verified earlier.
+    pub fn load_from_snapshot_with_server_identity(
+        snapshot: &TlsSourceSnapshot,
+        certificates: Vec<CertificateDer<'static>>,
+        private_key_pem: &[u8],
+    ) -> Result<Self, TlsRuntimeError> {
+        snapshot.validate_available()?;
+        let key = parse_private_key_pem(private_key_pem, "default")?;
+        validate_certified_key(&certificates, &key, "default", true)?;
+        Self::load_from_snapshot_with_loaded_server(
+            snapshot,
+            Some(ServerTlsMaterial::SingleCert {
+                certs: certificates,
+                key,
+            }),
+        )
+    }
 
-        let mtls_identity = match (
-            tokio::fs::read(&client_cert_path).await.ok(),
-            tokio::fs::read(&client_key_path).await.ok(),
-        ) {
-            (Some(cert_pem), Some(key_pem)) => {
-                let mut cert_reader = Cursor::new(&cert_pem);
-                if CertificateDer::pem_reader_iter(&mut cert_reader).next().is_none() {
-                    return Err(TlsRuntimeError::Material("no valid certificate in client cert PEM".to_string()));
+    fn load_from_snapshot_with_loaded_server(
+        snapshot: &TlsSourceSnapshot,
+        server: Option<ServerTlsMaterial>,
+    ) -> Result<Self, TlsRuntimeError> {
+        snapshot.validate_available()?;
+        let source = snapshot.resolved_source();
+        let base_dir = snapshot.base_dir();
+        let public_ca_pem = read_optional_tls_file(&base_dir.join(&source.layout.public_ca_filename))?;
+        let fallback_ca_pem = read_optional_tls_file(&base_dir.join(&source.fallback_ca_filename))?;
+        let client_ca_pem = read_optional_tls_file(&base_dir.join(&source.layout.client_ca_filename))?;
+        let mut material_files = BTreeSet::new();
+        for (name, present) in [
+            (&source.layout.public_ca_filename, public_ca_pem.is_some()),
+            (&source.fallback_ca_filename, fallback_ca_pem.is_some()),
+            (&source.layout.client_ca_filename, client_ca_pem.is_some()),
+        ] {
+            if present {
+                material_files.insert(name.clone());
+            }
+        }
+        for (name, pem) in [
+            (&source.layout.public_ca_filename, &public_ca_pem),
+            (&source.fallback_ca_filename, &fallback_ca_pem),
+            (&source.layout.client_ca_filename, &client_ca_pem),
+        ] {
+            if let Some(pem) = pem {
+                let certs = parse_certificate_pem(pem, name)?;
+                let mut roots = rustls::RootCertStore::empty();
+                for cert in certs {
+                    roots.add(cert).map_err(|source| TlsRuntimeError::InvalidTrustAnchor {
+                        name: name.clone(),
+                        source,
+                    })?;
                 }
-
-                let mut key_reader = Cursor::new(&key_pem);
-                PrivateKeyDer::from_pem_reader(&mut key_reader)
-                    .map_err(|e| TlsRuntimeError::Material(format!("invalid client key PEM: {e}")))?;
-
+            }
+        }
+        let server_client_ca_pem = client_ca_pem.or_else(|| fallback_ca_pem.clone());
+        let root_ca_pem = combine_optional_pem(public_ca_pem.as_deref(), fallback_ca_pem.as_deref());
+        let client_cert_pem = read_optional_tls_file(&base_dir.join(&source.layout.client_cert_filename))?;
+        let client_key_pem = read_optional_tls_file(&base_dir.join(&source.layout.client_key_filename))?;
+        if client_cert_pem.is_some() {
+            material_files.insert(source.layout.client_cert_filename.clone());
+        }
+        if client_key_pem.is_some() {
+            material_files.insert(source.layout.client_key_filename.clone());
+        }
+        let mtls_identity = match (client_cert_pem, client_key_pem) {
+            (Some(cert_pem), Some(key_pem)) => {
+                let certs = parse_certificate_pem(&cert_pem, "client identity")?;
+                let key = parse_private_key_pem(&key_pem, "client identity")?;
+                validate_certified_key(&certs, &key, "client identity", false)?;
                 Some(MtlsIdentityPem { cert_pem, key_pem })
             }
-            _ => None,
+            (None, None) => None,
+            _ => {
+                return Err(TlsRuntimeError::IncompleteIdentity {
+                    name: "client identity".to_string(),
+                });
+            }
         };
-
+        snapshot.validate_available()?;
         let outbound = OutboundTlsMaterial {
-            root_ca_pem: root_ca_pem.clone(),
+            root_ca_pem,
             mtls_identity: mtls_identity.clone(),
         };
-
-        let server_fingerprint_bytes = server.as_ref().map(serialize_server_material_for_fingerprint);
+        let mut server_fingerprint_bytes = server.as_ref().map(serialize_server_material_for_fingerprint);
+        if let Some(pem) = &server_client_ca_pem {
+            let bytes = server_fingerprint_bytes.get_or_insert_with(Vec::new);
+            bytes.extend_from_slice(b"\0server-client-ca\0");
+            bytes.extend_from_slice(pem);
+        }
         let fingerprint = TlsFingerprint::from_optional_bytes(
             server_fingerprint_bytes.as_deref(),
             public_ca_pem.as_deref(),
-            client_ca_pem.as_deref(),
+            fallback_ca_pem.as_deref(),
             mtls_identity.as_ref().map(|identity| identity.cert_pem.as_slice()),
             mtls_identity.as_ref().map(|identity| identity.key_pem.as_slice()),
         );
-
         Ok(Self {
-            source: source.clone(),
+            source: snapshot.resolved_source().clone(),
             server,
             outbound,
+            server_client_ca_pem,
+            material_files,
             fingerprint,
         })
     }
 }
 
 fn load_server_material(source: &TlsSource, base_dir: &Path) -> Result<Option<ServerTlsMaterial>, TlsRuntimeError> {
-    let root_cert = base_dir.join(&source.layout.server_cert_filename);
-    let root_key = base_dir.join(&source.layout.server_key_filename);
-
-    let has_root_pair = root_cert.exists() && root_key.exists();
-    let cert_key_pairs = load_all_certs_from_directory(
+    let mut cert_key_pairs = load_all_certs_from_directory_strict(
         CertDirectoryLoadOptions::builder(base_dir, &source.layout.server_cert_filename, &source.layout.server_key_filename)
             .build(),
-    );
-
-    match cert_key_pairs {
-        Ok(cert_key_pairs) if cert_key_pairs.len() > 1 || cert_key_pairs.keys().any(|key| key != "default") => {
-            Ok(Some(ServerTlsMaterial::MultiCert { cert_key_pairs }))
-        }
-        Ok(cert_key_pairs) if !cert_key_pairs.is_empty() => {
-            if let Some((certs, key)) = cert_key_pairs.get("default") {
-                return Ok(Some(ServerTlsMaterial::SingleCert {
-                    certs: certs.clone(),
-                    key: key.clone_key(),
-                }));
-            }
-            Ok(Some(ServerTlsMaterial::MultiCert { cert_key_pairs }))
-        }
-        Ok(_) => Ok(None),
-        Err(_err) if has_root_pair => {
-            let root_cert_path = path_to_utf8_str(&root_cert, "root TLS certificate")?;
-            let root_key_path = path_to_utf8_str(&root_key, "root TLS private key")?;
-            let certs = load_certs(root_cert_path)
-                .map_err(|e| TlsRuntimeError::Material(format!("load root TLS certificate {}: {e}", root_cert.display())))?;
-            let key = load_private_key(root_key_path)
-                .map_err(|e| TlsRuntimeError::Material(format!("load root TLS private key {}: {e}", root_key.display())))?;
-            Ok(Some(ServerTlsMaterial::SingleCert { certs, key }))
-        }
-        Err(err) if err.kind() == ErrorKind::NotFound => Ok(None),
-        Err(err) => Err(TlsRuntimeError::Material(format!(
-            "discover server TLS certificates under '{}': {err}",
-            base_dir.display()
-        ))),
+    )?;
+    if cert_key_pairs.is_empty() {
+        return Ok(None);
     }
-}
-
-fn path_to_utf8_str<'a>(path: &'a Path, description: &str) -> Result<&'a str, TlsRuntimeError> {
-    path.to_str()
-        .ok_or_else(|| TlsRuntimeError::Material(format!("{description} path '{}' is not valid UTF-8", path.display())))
+    if cert_key_pairs.len() == 1 {
+        if let Some((certs, key)) = cert_key_pairs.remove("default") {
+            return Ok(Some(ServerTlsMaterial::SingleCert { certs, key }));
+        }
+    }
+    Ok(Some(ServerTlsMaterial::MultiCert { cert_key_pairs }))
 }
 
 fn combine_optional_pem(primary: Option<&[u8]>, fallback: Option<&[u8]>) -> Vec<u8> {
@@ -198,3 +280,425 @@ pub(crate) fn server_material_fingerprint(material: &ServerTlsMaterial) -> TlsFi
     let bytes = serialize_server_material_for_fingerprint(material);
     TlsFingerprint::from_optional_bytes(Some(&bytes), None, None, None, None)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rcgen::{CertificateParams, ExtendedKeyUsagePurpose, KeyPair};
+    use std::fs;
+    use tempfile::TempDir;
+
+    fn write_pair(dir: &Path, domain: &str) {
+        let pair = rcgen::generate_simple_self_signed(vec![domain.to_string()]).expect("generate certificate");
+        fs::write(dir.join("rustfs_cert.pem"), pair.cert.pem()).expect("write certificate");
+        fs::write(dir.join("rustfs_key.pem"), pair.signing_key.serialize_pem()).expect("write private key");
+    }
+
+    #[test]
+    fn provided_server_identity_does_not_reread_server_files_and_preserves_trust() {
+        let base = TempDir::new().expect("base source");
+        let serving = rcgen::generate_simple_self_signed(vec!["example.com".into()]).expect("server identity");
+        let trust = rcgen::generate_simple_self_signed(vec!["trust.example".into()]).expect("trust source");
+        fs::write(base.path().join("rustfs_cert.pem"), "unusable server certificate").expect("poison base certificate");
+        fs::write(base.path().join("rustfs_key.pem"), "unusable server key").expect("poison base key");
+        fs::write(base.path().join(rustfs_config::RUSTFS_CA_CERT), trust.cert.pem()).expect("trust file");
+        let source = TlsSource::from_directory(base.path()).snapshot().expect("source");
+        let loaded = TlsMaterialSnapshot::load_from_snapshot_with_server_identity(
+            &source,
+            vec![serving.cert.der().clone()],
+            serving.signing_key.serialize_pem().as_bytes(),
+        )
+        .expect("provided identity");
+        assert_eq!(loaded.outbound.root_ca_pem, trust.cert.pem().as_bytes());
+        assert!(
+            matches!(loaded.server, Some(ServerTlsMaterial::SingleCert { certs, .. }) if certs == vec![serving.cert.der().clone()])
+        );
+        fs::write(base.path().join(rustfs_config::RUSTFS_CA_CERT), "corrupt trust").expect("corrupt independent trust");
+        assert!(
+            TlsMaterialSnapshot::load_from_snapshot_with_server_identity(
+                &source,
+                vec![serving.cert.der().clone()],
+                serving.signing_key.serialize_pem().as_bytes()
+            )
+            .is_err(),
+            "providing server material must not bypass trust validation"
+        );
+    }
+
+    #[test]
+    fn provided_server_identity_preserves_key_time_and_eku_validation() {
+        let base = TempDir::new().expect("base source");
+        let source = TlsSource::from_directory(base.path()).snapshot().expect("source");
+        let serving = rcgen::generate_simple_self_signed(vec!["example.com".into()]).expect("server identity");
+        let wrong_key = KeyPair::generate().expect("wrong key");
+        assert!(
+            TlsMaterialSnapshot::load_from_snapshot_with_server_identity(
+                &source,
+                vec![serving.cert.der().clone()],
+                wrong_key.serialize_pem().as_bytes()
+            )
+            .is_err(),
+            "key matching remains mandatory"
+        );
+        let mut expired = CertificateParams::new(vec!["example.com".into()]).expect("parameters");
+        expired.not_before = rcgen::date_time_ymd(2000, 1, 1);
+        expired.not_after = rcgen::date_time_ymd(2001, 1, 1);
+        let cert = expired.self_signed(&wrong_key).expect("expired certificate");
+        assert!(
+            TlsMaterialSnapshot::load_from_snapshot_with_server_identity(
+                &source,
+                vec![cert.der().clone()],
+                wrong_key.serialize_pem().as_bytes()
+            )
+            .is_err(),
+            "serving validity remains mandatory"
+        );
+        let mut client = CertificateParams::new(vec!["example.com".into()]).expect("parameters");
+        client.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
+        let cert = client.self_signed(&wrong_key).expect("client certificate");
+        assert!(
+            TlsMaterialSnapshot::load_from_snapshot_with_server_identity(
+                &source,
+                vec![cert.der().clone()],
+                wrong_key.serialize_pem().as_bytes()
+            )
+            .is_err(),
+            "server EKU remains mandatory"
+        );
+    }
+
+    #[tokio::test]
+    async fn server_overlay_survives_expired_bootstrap_without_replacing_trust() {
+        let base = TempDir::new().expect("base source");
+        let overlay = TempDir::new().expect("server overlay");
+        let key = KeyPair::generate().expect("bootstrap key");
+        let mut params = CertificateParams::new(vec!["example.com".into()]).expect("bootstrap params");
+        params.not_before = rcgen::date_time_ymd(2000, 1, 1);
+        params.not_after = rcgen::date_time_ymd(2001, 1, 1);
+        let expired = params.self_signed(&key).expect("expired bootstrap");
+        fs::write(base.path().join("rustfs_cert.pem"), expired.pem()).expect("bootstrap certificate");
+        fs::write(base.path().join("rustfs_key.pem"), key.serialize_pem()).expect("bootstrap key");
+        let ca = rcgen::generate_simple_self_signed(vec!["trust.example".into()]).expect("trust source");
+        fs::write(base.path().join(rustfs_config::RUSTFS_CA_CERT), ca.cert.pem()).expect("trust file");
+        write_pair(overlay.path(), "example.com");
+        let source = TlsSource::from_directory(base.path());
+        assert!(
+            TlsMaterialSnapshot::load(&source).await.is_err(),
+            "the expired base alone must remain invalid"
+        );
+        let loaded = TlsMaterialSnapshot::load_from_snapshot_with_server_directory(
+            &source.snapshot().expect("pinned base"),
+            Some(overlay.path()),
+        )
+        .expect("valid overlay");
+        assert_eq!(loaded.outbound.root_ca_pem, ca.cert.pem().as_bytes());
+        assert!(matches!(loaded.server, Some(ServerTlsMaterial::SingleCert { .. })));
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_mismatched_default_private_key() {
+        let dir = TempDir::new().expect("create temp directory");
+        write_pair(dir.path(), "example.com");
+        let other = KeyPair::generate().expect("generate other key");
+        fs::write(dir.path().join("rustfs_key.pem"), other.serialize_pem()).expect("replace key");
+        assert!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path()))
+                .await
+                .is_err()
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_partial_sni_pair_even_with_valid_default() {
+        let dir = TempDir::new().expect("create temp directory");
+        write_pair(dir.path(), "example.com");
+        let domain = dir.path().join("sni.example.com");
+        fs::create_dir(&domain).expect("create SNI directory");
+        fs::write(domain.join("rustfs_cert.pem"), "invalid certificate").expect("write partial SNI pair");
+        assert!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path()))
+                .await
+                .is_err()
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_invalid_existing_ca() {
+        let dir = TempDir::new().expect("create temp directory");
+        write_pair(dir.path(), "example.com");
+        let source = TlsSource::from_directory(dir.path());
+        fs::write(dir.path().join(&source.layout.public_ca_filename), "invalid CA").expect("write invalid CA");
+        assert!(TlsMaterialSnapshot::load(&source).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_incomplete_client_identity() {
+        let dir = TempDir::new().expect("create temp directory");
+        write_pair(dir.path(), "example.com");
+        let source = TlsSource::from_directory(dir.path());
+        fs::write(dir.path().join(&source.layout.client_cert_filename), "invalid client certificate").expect("write client cert");
+        assert!(TlsMaterialSnapshot::load(&source).await.is_err());
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_expired_server_certificate() {
+        let dir = TempDir::new().expect("create temp directory");
+        let key = KeyPair::generate().expect("generate key");
+        let mut params = CertificateParams::new(vec!["example.com".to_string()]).expect("certificate params");
+        params.not_before = rcgen::date_time_ymd(2000, 1, 1);
+        params.not_after = rcgen::date_time_ymd(2001, 1, 1);
+        let cert = params.self_signed(&key).expect("generate expired certificate");
+        fs::write(dir.path().join("rustfs_cert.pem"), cert.pem()).expect("write certificate");
+        fs::write(dir.path().join("rustfs_key.pem"), key.serialize_pem()).expect("write key");
+        assert!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path()))
+                .await
+                .is_err()
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_client_only_server_eku() {
+        let dir = TempDir::new().expect("create temp directory");
+        let key = KeyPair::generate().expect("generate key");
+        let mut params = CertificateParams::new(vec!["example.com".to_string()]).expect("certificate params");
+        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
+        let cert = params.self_signed(&key).expect("generate client-only certificate");
+        fs::write(dir.path().join("rustfs_cert.pem"), cert.pem()).expect("write certificate");
+        fs::write(dir.path().join("rustfs_key.pem"), key.serialize_pem()).expect("write key");
+        assert!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path()))
+                .await
+                .is_err()
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_server_only_client_eku() {
+        let dir = TempDir::new().expect("client identity directory");
+        write_pair(dir.path(), "server.example");
+        let key = KeyPair::generate().expect("client key");
+        let mut params = CertificateParams::new(vec!["client.example".to_string()]).expect("client params");
+        params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ServerAuth];
+        let cert = params.self_signed(&key).expect("server-only certificate");
+        fs::write(dir.path().join("client_cert.pem"), cert.pem()).expect("client certificate");
+        fs::write(dir.path().join("client_key.pem"), key.serialize_pem()).expect("client key");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path())).await,
+            Err(TlsRuntimeError::InvalidClientEku { .. })
+        ));
+    }
+    #[tokio::test]
+    async fn snapshot_rejects_future_server_certificate() {
+        let dir = TempDir::new().expect("create directory");
+        let key = KeyPair::generate().expect("generate key");
+        let mut params = CertificateParams::new(vec!["example.com".to_string()]).expect("certificate params");
+        params.not_before = rcgen::date_time_ymd(2100, 1, 1);
+        params.not_after = rcgen::date_time_ymd(2101, 1, 1);
+        let cert = params.self_signed(&key).expect("generate future certificate");
+        fs::write(dir.path().join("rustfs_cert.pem"), cert.pem()).expect("write certificate");
+        fs::write(dir.path().join("rustfs_key.pem"), key.serialize_pem()).expect("write key");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path())).await,
+            Err(TlsRuntimeError::CertificateValidity { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_wrong_sni_certificate_name() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        let domain = dir.path().join("other.example.com");
+        fs::create_dir(&domain).expect("create domain directory");
+        write_pair(&domain, "wrong.example.com");
+        assert!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path()))
+                .await
+                .is_err()
+        );
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_malformed_complete_sni_pair() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        let domain = dir.path().join("sni.example.com");
+        fs::create_dir(&domain).expect("create domain directory");
+        write_pair(&domain, "sni.example.com");
+        fs::write(domain.join("rustfs_cert.pem"), "invalid certificate").expect("corrupt domain certificate");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path())).await,
+            Err(TlsRuntimeError::InvalidCertificate { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_cert_only_root() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        fs::remove_file(dir.path().join("rustfs_key.pem")).expect("remove key");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&TlsSource::from_directory(dir.path())).await,
+            Err(TlsRuntimeError::IncompleteIdentity { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn snapshot_fingerprint_is_stable_and_certificate_only_renewal_changes_it() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        let source = TlsSource::from_directory(dir.path());
+        let initial = TlsMaterialSnapshot::load(&source)
+            .await
+            .expect("load valid initial generation");
+        let unchanged = TlsMaterialSnapshot::load(&source).await.expect("load unchanged generation");
+        assert_eq!(initial.fingerprint, unchanged.fingerprint);
+        let key = KeyPair::from_pem(&fs::read_to_string(dir.path().join("rustfs_key.pem")).expect("read key"))
+            .expect("parse existing key");
+        let mut params = CertificateParams::new(vec!["example.com".to_string()]).expect("certificate params");
+        params.serial_number = Some(123_u64.into());
+        let renewed = params.self_signed(&key).expect("renew certificate with existing key");
+        fs::write(dir.path().join("rustfs_cert.pem"), renewed.pem()).expect("write renewed certificate");
+        let changed = TlsMaterialSnapshot::load(&source).await.expect("load renewed generation");
+        assert_ne!(initial.fingerprint, changed.fingerprint);
+    }
+
+    #[tokio::test]
+    async fn snapshot_server_client_ca_rotation_changes_fingerprint() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        let source = TlsSource::from_directory(dir.path());
+        let first_ca = rcgen::generate_simple_self_signed(vec!["first-ca".to_string()]).expect("first CA");
+        fs::write(dir.path().join(&source.layout.client_ca_filename), first_ca.cert.pem()).expect("write first CA");
+        let initial = TlsMaterialSnapshot::load(&source).await.expect("load first CA");
+        let second_ca = rcgen::generate_simple_self_signed(vec!["second-ca".to_string()]).expect("second CA");
+        fs::write(dir.path().join(&source.layout.client_ca_filename), second_ca.cert.pem()).expect("write second CA");
+        let changed = TlsMaterialSnapshot::load(&source).await.expect("load second CA");
+        assert_ne!(initial.fingerprint, changed.fingerprint);
+        assert_eq!(changed.server_client_ca_pem.as_deref(), Some(second_ca.cert.pem().as_bytes()));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn snapshot_kubernetes_switch_loads_all_material_from_pinned_generation() {
+        use std::os::unix::fs::symlink;
+        let dir = TempDir::new().expect("create projection");
+        let source = TlsSource::from_directory(dir.path());
+        for generation in ["..a", "..b"] {
+            let path = dir.path().join(generation);
+            fs::create_dir(&path).expect("create generation");
+            write_pair(&path, "example.com");
+            let cert = fs::read(path.join(&source.layout.server_cert_filename)).expect("read generated certificate");
+            let key = fs::read(path.join(&source.layout.server_key_filename)).expect("read generated key");
+            for filename in [
+                &source.layout.public_ca_filename,
+                &source.fallback_ca_filename,
+                &source.layout.client_ca_filename,
+                &source.layout.client_cert_filename,
+            ] {
+                fs::write(path.join(filename), &cert).expect("write generation CA/client certificate");
+            }
+            fs::write(path.join(&source.layout.client_key_filename), key).expect("write generation client key");
+        }
+        symlink("..a", dir.path().join("..data")).expect("link initial generation");
+        for filename in [
+            &source.layout.server_cert_filename,
+            &source.layout.server_key_filename,
+            &source.layout.public_ca_filename,
+            &source.fallback_ca_filename,
+            &source.layout.client_ca_filename,
+            &source.layout.client_cert_filename,
+            &source.layout.client_key_filename,
+        ] {
+            symlink(format!("..data/{filename}"), dir.path().join(filename)).expect("link projected file");
+        }
+        let pinned = source.snapshot().expect("pin initial generation");
+        let expected = TlsMaterialSnapshot::load_from_snapshot(&pinned).expect("load initial generation");
+        symlink("..b", dir.path().join("..next")).expect("link next generation");
+        fs::rename(dir.path().join("..next"), dir.path().join("..data")).expect("switch projection");
+        let actual = TlsMaterialSnapshot::load_from_snapshot(&pinned).expect("load pinned generation after switch");
+        assert_eq!(actual.fingerprint, expected.fingerprint);
+        assert_eq!(actual.source.base_dir, pinned.base_dir());
+        assert_eq!(actual.outbound.root_ca_pem, expected.outbound.root_ca_pem);
+        assert_eq!(actual.server_client_ca_pem, expected.server_client_ca_pem);
+        assert_eq!(
+            actual.outbound.mtls_identity.expect("client identity").key_pem,
+            expected.outbound.mtls_identity.expect("expected identity").key_pem
+        );
+        let current = TlsMaterialSnapshot::load_from_snapshot(&source.snapshot().expect("pin new generation"))
+            .expect("load new generation");
+        assert_ne!(actual.fingerprint, current.fingerprint);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn snapshot_rejects_disappeared_projection_source_and_dangling_optional_ca() {
+        use std::os::unix::fs::symlink;
+        let dir = TempDir::new().expect("create directory");
+        let source = TlsSource::from_directory(dir.path());
+        write_pair(dir.path(), "example.com");
+        symlink("missing-ca", dir.path().join(&source.layout.public_ca_filename)).expect("link missing CA");
+        let pinned = source.snapshot().expect("pin directory");
+        assert!(matches!(
+            TlsMaterialSnapshot::load_from_snapshot(&pinned),
+            Err(TlsRuntimeError::SourceIo { .. })
+        ));
+        fs::remove_file(dir.path().join(&source.layout.public_ca_filename)).expect("remove CA link");
+        fs::remove_dir_all(dir.path()).expect("remove pinned source");
+        assert!(matches!(
+            TlsMaterialSnapshot::load_from_snapshot(&pinned),
+            Err(TlsRuntimeError::SourceIo { .. })
+        ));
+    }
+    #[tokio::test]
+    async fn snapshot_rejects_mismatched_sni_key_and_malformed_chain_member() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        let domain = dir.path().join("sni.example.com");
+        fs::create_dir(&domain).expect("create SNI directory");
+        write_pair(&domain, "sni.example.com");
+        let source = TlsSource::from_directory(dir.path());
+        let valid = TlsMaterialSnapshot::load(&source)
+            .await
+            .expect("load valid multi certificate material");
+        assert!(matches!(valid.server, Some(ServerTlsMaterial::MultiCert { ref cert_key_pairs }) if cert_key_pairs.len() == 2));
+        fs::write(
+            domain.join("rustfs_key.pem"),
+            KeyPair::generate().expect("generate different key").serialize_pem(),
+        )
+        .expect("replace SNI key");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&source).await,
+            Err(TlsRuntimeError::InvalidCertifiedKey { .. })
+        ));
+        write_pair(&domain, "sni.example.com");
+        let cert_path = dir.path().join("rustfs_cert.pem");
+        let mut cert = fs::read_to_string(&cert_path).expect("read leaf certificate");
+        cert.push_str("-----BEGIN CERTIFICATE-----\nAQID\n-----END CERTIFICATE-----\n");
+        fs::write(cert_path, cert).expect("append malformed chain member");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&source).await,
+            Err(TlsRuntimeError::InvalidCertificate { .. })
+        ));
+    }
+
+    #[tokio::test]
+    async fn snapshot_rejects_invalid_private_key_pem_and_invalid_primary_client_ca() {
+        let dir = TempDir::new().expect("create directory");
+        write_pair(dir.path(), "example.com");
+        let source = TlsSource::from_directory(dir.path());
+        fs::write(dir.path().join("rustfs_key.pem"), "invalid key").expect("corrupt key");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&source).await,
+            Err(TlsRuntimeError::InvalidPrivateKey { .. })
+        ));
+        write_pair(dir.path(), "example.com");
+        let valid_ca = fs::read(dir.path().join("rustfs_cert.pem")).expect("read valid certificate");
+        fs::write(dir.path().join(&source.fallback_ca_filename), valid_ca).expect("write valid fallback CA");
+        fs::write(dir.path().join(&source.layout.client_ca_filename), "invalid primary CA").expect("write invalid primary CA");
+        assert!(matches!(
+            TlsMaterialSnapshot::load(&source).await,
+            Err(TlsRuntimeError::InvalidCertificate { .. })
+        ));
+    }
+}

crates/tls-runtime/src/outbound.rs

@@ -16,8 +16,8 @@ use crate::material::OutboundTlsMaterial;
 use crate::metrics::record_outbound_tls_publication;
 use crate::state::TlsGeneration;
 use rustfs_common::{
-    MtlsIdentityPem, clear_global_root_cert, get_global_mtls_identity, get_global_outbound_tls_generation, get_global_root_cert,
-    set_global_mtls_identity, set_global_outbound_tls_generation, set_global_root_cert,
+    MtlsIdentityPem, OutboundTlsSnapshot, get_global_outbound_tls_generation, get_global_outbound_tls_snapshot,
+    publish_global_outbound_tls_snapshot,
 };
 
 #[derive(Debug, Clone)]
@@ -35,21 +35,21 @@ pub struct GlobalOutboundTlsStateSummary {
 }
 
 pub async fn publish_global_outbound_tls_state(generation: TlsGeneration, material: &OutboundTlsMaterial) {
-    if !material.root_ca_pem.is_empty() {
-        set_global_root_cert(material.root_ca_pem.clone()).await;
-    } else {
-        clear_global_root_cert().await;
-    }
-    set_global_mtls_identity(material.mtls_identity.clone()).await;
-    set_global_outbound_tls_generation(generation.0);
-    record_outbound_tls_publication(generation.0, !material.root_ca_pem.is_empty(), material.mtls_identity.is_some());
+    let accepted = publish_global_outbound_tls_snapshot(OutboundTlsSnapshot {
+        generation: generation.0,
+        root_ca_pem: (!material.root_ca_pem.is_empty()).then(|| material.root_ca_pem.clone()),
+        mtls_identity: material.mtls_identity.clone(),
+    })
+    .await;
+    record_outbound_tls_publication(accepted, !material.root_ca_pem.is_empty(), material.mtls_identity.is_some());
 }
 
 pub async fn load_global_outbound_tls_state() -> GlobalPublishedOutboundTlsState {
+    let snapshot = get_global_outbound_tls_snapshot();
     GlobalPublishedOutboundTlsState {
-        generation: TlsGeneration(get_global_outbound_tls_generation()),
-        root_ca_pem: get_global_root_cert().await,
-        mtls_identity: get_global_mtls_identity().await,
+        generation: TlsGeneration(snapshot.generation),
+        root_ca_pem: snapshot.root_ca_pem.clone(),
+        mtls_identity: snapshot.mtls_identity.clone(),
     }
 }
 

crates/tls-runtime/src/source.rs

@@ -92,3 +92,125 @@ impl TlsSource {
         Ok(self.base_dir.as_path())
     }
 }
+
+/// A directory generation resolved once before loading any TLS material.
+///
+/// Projected Kubernetes volumes are read through their physical `..data`
+/// directory. Ordinary directory and certificate-file symlinks remain supported
+/// for layouts such as Certbot. Writers must keep pinned generations immutable.
+#[derive(Debug, Clone)]
+pub struct TlsSourceSnapshot {
+    source: TlsSource,
+    resolved_source: TlsSource,
+}
+
+impl TlsSource {
+    pub fn snapshot(&self) -> Result<TlsSourceSnapshot, TlsRuntimeError> {
+        self.validate_directory()?;
+        let mut resolved_source = self.clone();
+        let root = std::fs::canonicalize(&self.base_dir).map_err(|source| TlsRuntimeError::SourceIo {
+            path: self.base_dir.clone(),
+            source,
+        })?;
+        let projection = root.join("..data");
+        resolved_source.base_dir = match std::fs::symlink_metadata(&projection) {
+            Ok(metadata) if metadata.file_type().is_symlink() => {
+                std::fs::canonicalize(&projection).map_err(|source| TlsRuntimeError::SourceIo {
+                    path: projection,
+                    source,
+                })?
+            }
+            Ok(_) => root,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => root,
+            Err(source) => {
+                return Err(TlsRuntimeError::SourceIo {
+                    path: projection,
+                    source,
+                });
+            }
+        };
+        resolved_source.validate_directory()?;
+        Ok(TlsSourceSnapshot {
+            source: self.clone(),
+            resolved_source,
+        })
+    }
+}
+
+impl TlsSourceSnapshot {
+    pub fn source(&self) -> &TlsSource {
+        &self.source
+    }
+
+    pub fn resolved_source(&self) -> &TlsSource {
+        &self.resolved_source
+    }
+
+    pub fn base_dir(&self) -> &Path {
+        &self.resolved_source.base_dir
+    }
+
+    /// Missing optional files are allowed only while the pinned source exists.
+    pub(crate) fn validate_available(&self) -> Result<(), TlsRuntimeError> {
+        std::fs::read_dir(self.base_dir()).map_err(|source| TlsRuntimeError::SourceIo {
+            path: self.base_dir().to_path_buf(),
+            source,
+        })?;
+        Ok(())
+    }
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use std::fs;
+    use std::os::unix::fs::symlink;
+    use tempfile::TempDir;
+
+    #[test]
+    fn source_snapshot_pins_kubernetes_projection_generation() {
+        let dir = TempDir::new().expect("create projection directory");
+        for generation in ["..generation-a", "..generation-b"] {
+            fs::create_dir(dir.path().join(generation)).expect("create generation");
+            fs::write(dir.path().join(generation).join("marker"), generation).expect("write generation marker");
+        }
+        symlink("..generation-a", dir.path().join("..data")).expect("link initial generation");
+        symlink("..data/marker", dir.path().join("marker")).expect("link projected file");
+        let pinned = TlsSource::from_directory(dir.path()).snapshot().expect("pin projection");
+        symlink("..generation-b", dir.path().join("..next-data")).expect("link next generation");
+        fs::rename(dir.path().join("..next-data"), dir.path().join("..data")).expect("swap projection");
+        assert_eq!(
+            fs::read_to_string(pinned.base_dir().join("marker")).expect("read pinned file"),
+            "..generation-a"
+        );
+        assert_eq!(
+            fs::read_to_string(dir.path().join("marker")).expect("read current projection"),
+            "..generation-b"
+        );
+    }
+
+    #[test]
+    fn source_snapshot_pins_tls_path_directory_symlink() {
+        let dir = TempDir::new().expect("create TLS directory");
+        fs::create_dir(dir.path().join("a")).expect("create generation A");
+        fs::create_dir(dir.path().join("b")).expect("create generation B");
+        symlink("a", dir.path().join("live")).expect("link generation A");
+        let pinned = TlsSource::from_directory(dir.path().join("live"))
+            .snapshot()
+            .expect("pin TLS source");
+        fs::remove_file(dir.path().join("live")).expect("unlink source");
+        symlink("b", dir.path().join("live")).expect("link generation B");
+        assert_eq!(pinned.base_dir(), fs::canonicalize(dir.path().join("a")).expect("resolve generation A"));
+    }
+
+    #[test]
+    fn source_snapshot_rejects_disappeared_generation() {
+        let dir = TempDir::new().expect("create TLS directory");
+        let generation = dir.path().join("..generation");
+        fs::create_dir(&generation).expect("create generation");
+        symlink("..generation", dir.path().join("..data")).expect("link generation");
+        let pinned = TlsSource::from_directory(dir.path()).snapshot().expect("pin TLS source");
+        fs::remove_dir(&generation).expect("remove pinned generation");
+        assert!(matches!(pinned.validate_available(), Err(TlsRuntimeError::SourceIo { .. })));
+    }
+}

docs/README.md

@@ -8,6 +8,8 @@ collection:
 
 ## Operations
 
+- [External certificate issuance and TLS rotation](operations/tls-certificate-rotation.md) — publication, native reload, trust separation and failure handling.
+
 - [Architecture guard troubleshooting](operations/architecture-guard-troubleshooting.md) — diagnose repository guard failures.
 - [Logging governance](operations/logging-governance.md) — broad event audits, migrations, and guardrail changes.
 

docs/operations/tls-certificate-rotation.md

@@ -0,0 +1,204 @@
+# External certificate issuance and TLS rotation
+
+Use an independent certificate controller to obtain and renew certificates.
+RustFS consumes certificate material; it does not register ACME accounts or
+change DNS records. The independent
+[rustfs-cert-controller](https://github.com/houseme/rustfs-cert-controller)
+owns issuance, renewal scheduling and publication. `plugin-runner` starts its
+container; it does not own certificate schedules or CA credentials. Progress
+and deployment acceptance are tracked in
+[backlog #2806](https://github.com/rustfs/backlog/issues/2806).
+
+## Deployment boundaries
+
+For a distributed Kubernetes deployment, prefer public TLS at an
+Ingress/Gateway and private TLS/mTLS between the ingress and RustFS. Public
+CA certificates cannot cover internal names such as `*.svc.cluster.local`;
+DNS-01 can cover names under a controlled public domain even when the service
+is private. A listener
+that requires mTLS also requires client certificates from browsers and S3
+clients; selecting a different SNI certificate does not change that policy.
+
+Native server TLS can consume an externally managed `kubernetes.io/tls`
+Secret with only `tls.crt` and `tls.key`:
+
+```yaml
+mode:
+  distributed:
+    enabled: false
+  standalone:
+    enabled: true
+ingress:
+  enabled: false
+tls:
+  enabled: true
+  existingSecret: rustfs-serving-tls
+  reloadInterval: 30
+```
+
+Create and renew that Secret using cert-manager or an independently managed
+publisher. Configure the issuing resource with the actual server DNS/IP SANs
+and explicit private-key rotation. The chart does not install cert-manager,
+create an ACME account or grant DNS permissions. For distributed native TLS,
+the certificate must also cover the configured peer hostnames; explicitly
+configure independent outbound trust when a private server CA is used.
+
+`tls.trustSecret` optionally names a separate Secret with an explicitly chosen
+`ca.crt`, mounted as outbound `public.crt`. It does not enable mTLS or add a
+public issuing CA to client authentication trust. Do not write a fabricated
+`ca.crt` into the Secret managed by an ACME issuer.
+
+`tls.enabled` and `mtls.enabled` are mutually exclusive. Existing mTLS mode
+keeps its separate server/client certificate Secrets and projects their
+contents into one read-only directory. The chart enables certificate reload
+for either mode and disables TLS key logging. Explicit container environment
+variables can override these settings.
+
+## Material layout and publication
+
+`RUSTFS_TLS_PATH` contains `rustfs_cert.pem` and `rustfs_key.pem`. SNI
+identities can reside in visible domain subdirectories with the same two
+filenames. Outbound trust uses `public.crt` and `ca.crt`; inbound mTLS client
+trust prefers `client_ca.crt`, falling back to `ca.crt` only at initial load.
+The default outbound client identity is `client_cert.pem`/`client_key.pem`.
+
+Runtime loading validates existing PEM/DER, key pairing, leaf validity and
+explicit authentication EKUs. Each named SNI identity must match its domain.
+Chain structure validation does not replace issuer/trust-policy verification
+by the issuing controller and client PKI verifier.
+
+Kubernetes mounts must project the entire directory; single-file `subPath`
+mounts do not receive Secret updates. RustFS pins one physical `..data`
+generation before loading its material. Separate controllers may update the
+serving identity and trust independently, so private CA migration still needs
+an old-plus-new trust overlap.
+
+For VMs, publish immutable version directories and atomically change the
+directory pointer after syncing the complete candidate and its containing
+directories. Keep current/accepted/in-use versions during cleanup. Certbot
+file symlinks remain compatible, but independent file replacements do not
+provide an atomic directory generation; key mismatch is rejected, and a
+publisher should use a complete immutable bundle for coherent publication.
+
+`RUSTFS_MTLS_CLIENT_CERT`/`RUSTFS_MTLS_CLIENT_KEY` overrides are validated as
+one identity. Explicit missing, malformed, mismatched, expired or unsuitable
+override material rejects the update; it does not silently select another
+identity. Keep default identity files valid when they are present as well.
+
+## Controller publication API
+
+For a VM or Docker installation with one default serving certificate, enable
+reload and configure `RUSTFS_TLS_PUBLISH_DIR` to a dedicated persistent directory
+owned by the RustFS process with mode `0700`. Keep `RUSTFS_TLS_PATH` configured
+for the initial serving identity and independent trust/client material. The
+publication directory must be separate from the original TLS source. A valid
+initial certificate and trusted management credentials must be provisioned
+before a controller can connect to the HTTPS API. That bootstrap identity must
+cover the names the configured issuer will continue to renew; publication
+preserves existing SAN coverage.
+
+The controller signs requests using the existing S3 SigV4 authentication:
+
+- `GET /rustfs/admin/v3/tls/certificate-state` requires
+  `admin:TLSCertificateStatus`.
+- `POST /rustfs/admin/v3/tls/certificate` requires
+  `admin:TLSCertificatePublish` and a connection terminated by RustFS TLS.
+
+Grant only these two actions to the controller identity. Read permissions,
+general configuration permissions and forwarded HTTPS headers do not authorize
+private-key publication. Keep API requests and credentials out of access-body
+logs. The API does not add MinIO-compatible aliases.
+
+The `rustfs.tls-publish/v1alpha1` request fixes `resource_id` to `server-default`
+and carries an `issuance_id`, the captured `expected_target_revision`,
+`certificate_pem` and `private_key_pem`. Read the target token before issuance;
+retain it with that immutable candidate. Repeating the same issuance and payload
+recovers a lost receipt. A conflict stops that candidate; fetching a new token
+and replaying the old candidate would defeat this concurrency protection.
+An HTTP 503 or uncertain network result is retryable with the same issuance,
+payload and token; HTTP 409 indicates a revision or payload conflict.
+
+RustFS validates the complete candidate before publishing an immutable leaf
+generation and syncing its atomic pointer. It does not copy or overwrite CA
+trust or client credentials. This mode rejects projected Kubernetes sources,
+multiple SNI identities and configurations that trust the serving leaf as a CA.
+Use the existing external-file/cert-manager integration for those deployments.
+
+The state reports the visible target revision and public leaf fingerprint. A
+successful POST receipt proves the publication was synced; a GET during an
+uncertain write does not replace that receipt. Retry the identical POST to
+confirm acceptance. `observed_certificate_fingerprint` is present only when
+this process's serving reload group has the exact current certificate chain;
+matching the leaf alone is insufficient. A durable receipt can precede that
+observation. Check convergence and a fresh TLS handshake on every relevant
+endpoint. One process's response does not prove another node has converged.
+
+Use a stable local publication volume with one owner. Captured-token CAS and
+local process locking do not implement HA leadership, restore ordering or PITR
+protection. Do not remove active generation directories; automated generation
+cleanup is not part of this interface.
+
+## Activation and failure handling
+
+Native `RUSTFS_TLS_RELOAD_ENABLE` defaults to false. Set it to `1` outside the
+chart. `RUSTFS_TLS_RELOAD_INTERVAL` defaults to 30 seconds and is clamped to a
+minimum of 5 seconds.
+
+Listeners sharing the same configured TLS directory use one reload group.
+The group prepares TCP and enabled HTTP/3 configurations from the same
+snapshot before swapping them. A late listener first joins the accepted
+configuration, even if the current files have become invalid. TCP, QUIC and
+separate processes still have a short activation window; there is no global
+atomic certificate switch.
+TCP and QUIC derive their protocol settings from one prepared certificate
+resolver and client verifier for that generation. Publication state and retries
+load each immutable generation once per operation; the next operation validates
+its disk contents again. Reload uses the verified serving identity with the
+independently pinned trust/client source, without reopening its certificate and
+key files. Candidate staging still performs its separate disk validation.
+
+Unchanged material is skipped. A serving certificate renewal does not advance
+the outbound generation unless outbound trust or identity actually changes.
+Outbound state is published as one immutable trust/identity/generation
+snapshot. Control and bulk RPC cache acquisition checks that generation;
+existing in-flight calls retain their channels.
+
+Reload rejects the disappearance of accepted CA/identity source components,
+SNI entries or accepted DNS/IP SAN coverage. Removing or renaming an identity
+or trust source requires an explicit configuration change and restart;
+ordinary renewal must preserve those requirements. Rejected updates retain
+the last accepted in-memory configuration. New handshakes reject certificates
+outside their validity interval. Normal renewal does not terminate existing
+uploads, downloads or sessions.
+
+The chart uses TCP listener liveness for TLS deployments so certificate expiry
+does not trigger a restart loop. Readiness continues to check the TLS service.
+This liveness probe detects listener availability, not complete application
+health. Keep ingress traffic admission separate from internal peer routing
+and storage quorum. Do not restart every node because a CA is unavailable.
+
+There is no automatic persisted private-key fallback in Kubernetes. On
+restart, invalid explicitly configured TLS material prevents startup. Key
+compromise, explicit certificate revocation, session invalidation, per-purpose
+profile coordination and deployment-boundary fencing require the separate
+operator/control-plane workflow; ordinary reload does not implement them.
+
+## Verification
+
+Check a fresh TLS connection to each endpoint and compare the leaf serial,
+public certificate fingerprint, SANs and expiry. Include HTTP/2, enabled
+HTTP/3, Console and internal RPC endpoints. Do not infer served certificates
+from Secret contents or an outbound generation gauge alone.
+
+Focused local checks:
+
+```sh
+RUSTUP_TOOLCHAIN=stable cargo test -p rustfs-tls-runtime
+RUSTUP_TOOLCHAIN=stable cargo test -p rustfs-protos --lib
+bash scripts/test_helm_tls_reload.sh
+```
+
+Production acceptance also requires real Kubernetes projection, CA/DNS
+issuance and renewal, outage/expiry, long-transfer continuity, private CA
+overlap and distributed convergence tests. These are separate from protocol
+and local loader tests.

helm/rustfs/templates/_helpers.tpl

@@ -154,6 +154,10 @@ Merges (in order of increasing precedence):
 }}
 {{- $annotations = merge $annotations $mtls }}
 {{- end }}
+{{- $tls := .Values.tls | default dict -}}
+{{- if and ($tls.enabled | default false) (eq .Values.ingress.className "traefik") -}}
+{{- $_ := set $annotations "traefik.ingress.kubernetes.io/service.serversscheme" "https" -}}
+{{- end -}}
 {{- if $annotations }}
 {{- toYaml $annotations }}
 {{- end }}
@@ -188,6 +192,10 @@ Merges (in order of increasing precedence):
 {{- end }}
 {{- $annotations = merge $annotations (default (dict) .Values.ingress.customAnnotations) }}
 {{- $annotations = merge $annotations (default (dict) .Values.ingress.annotations) }}
+{{- $tls := .Values.tls | default dict -}}
+{{- if and ($tls.enabled | default false) (eq .Values.ingress.className "nginx") -}}
+{{- $_ := set $annotations "nginx.ingress.kubernetes.io/backend-protocol" "HTTPS" -}}
+{{- end -}}
 {{- if $annotations }}
 {{- toYaml $annotations }}
 {{- end }}
@@ -285,7 +293,8 @@ RUSTFS_VOLUMES on spaces, one pool per expression).
 */}}
 {{- define "rustfs.volumes" -}}
 {{- $protocol := "http" -}}
-{{- if .Values.mtls.enabled -}}
+{{- $tls := .Values.tls | default dict -}}
+{{- if or .Values.mtls.enabled ($tls.enabled | default false) -}}
   {{- $protocol = "https" -}}
 {{- end -}}
 {{- $headless := printf "%s-headless" (include "rustfs.fullname" .) -}}
@@ -347,17 +356,17 @@ Render liveness and readiness probe for http and https
 */}}
 
 {{- define "rustfs.probes" -}}
+{{- $tls := .Values.tls | default dict -}}
 {{- if .Values.livenessProbe.enabled }}
 livenessProbe:
-  {{- if .Values.mtls.enabled }}
-  exec:
-    command:
-{{ include "rustfs.probeCommand" (dict "root" . "endpointPath" "/health") | nindent 6 }}
+  {{- if or .Values.mtls.enabled ($tls.enabled | default false) }}
+  tcpSocket:
+    port: {{ .Values.service.endpoint.port | default 9000 }}
   {{- else }}
   httpGet:
     path: /health
     port: {{ .Values.service.endpoint.port | default 9000 }}
-    scheme: {{ if .Values.mtls.enabled }}HTTPS{{ else }}HTTP{{ end }}
+    scheme: {{ if or .Values.mtls.enabled ($tls.enabled | default false) }}HTTPS{{ else }}HTTP{{ end }}
   {{- end }}
   initialDelaySeconds: {{ .Values.livenessProbe.initialDelaySeconds | default 60 }}
   periodSeconds: {{ .Values.livenessProbe.periodSeconds | default 5 }}
@@ -376,7 +385,7 @@ readinessProbe:
   httpGet:
     path: /health/ready
     port: {{ .Values.service.endpoint.port | default 9000 }}
-    scheme: {{ if .Values.mtls.enabled }}HTTPS{{ else }}HTTP{{ end }}
+    scheme: {{ if or .Values.mtls.enabled ($tls.enabled | default false) }}HTTPS{{ else }}HTTP{{ end }}
   {{- end }}
   initialDelaySeconds: {{ .Values.readinessProbe.initialDelaySeconds | default 60 }}
   periodSeconds: {{ .Values.readinessProbe.periodSeconds | default 5 }}

helm/rustfs/templates/configmap.yaml

@@ -1,3 +1,8 @@
+{{- $tls := .Values.tls | default dict -}}
+{{- $tlsEnabled := $tls.enabled | default false -}}
+{{- if and $tlsEnabled .Values.mtls.enabled -}}
+{{- fail "tls.enabled and mtls.enabled are mutually exclusive; use ingress TLS for public access to an mTLS backend" -}}
+{{- end -}}
 apiVersion: v1
 kind: ConfigMap
 metadata:
@@ -41,8 +46,12 @@ data:
   {{- end }}
   {{- if .Values.mtls.enabled }}
   RUSTFS_SERVER_MTLS_ENABLE: "1"
-  RUSTFS_TLS_KEYLOG: "1"
+  {{- end }}
+  {{- if or .Values.mtls.enabled $tlsEnabled }}
+  RUSTFS_TLS_KEYLOG: "0"
   RUSTFS_TLS_PATH: "/opt/tls"
+  RUSTFS_TLS_RELOAD_ENABLE: "1"
+  RUSTFS_TLS_RELOAD_INTERVAL: {{ $tls.reloadInterval | default 30 | quote }}
   {{- end }}
   {{- if .Values.config.rustfs.obs_endpoint.enabled }}
   RUSTFS_OBS_ENDPOINT: {{ .Values.config.rustfs.obs_endpoint.base_endpoint | quote }}

helm/rustfs/templates/deployment.yaml

@@ -1,3 +1,5 @@
+{{- $tls := .Values.tls | default dict -}}
+{{- $tlsEnabled := $tls.enabled | default false -}}
 {{- $logDir := .Values.config.rustfs.obs_log_directory }}
 {{- $logDirEnabled := ne $logDir "" }}
 
@@ -123,25 +125,10 @@ spec:
             {{- toYaml .Values.resources | nindent 12 }}
           {{- include "rustfs.probes" . | nindent 10 }}
           volumeMounts:
-            {{- if .Values.mtls.enabled }}
-            - name: server-cert
-              mountPath: /opt/tls/rustfs_cert.pem
-              subPath: rustfs_cert.pem
-            - name: server-cert
-              mountPath: /opt/tls/rustfs_key.pem
-              subPath: rustfs_key.pem
-            - name: server-cert
-              mountPath: /opt/tls/ca.crt
-              subPath: ca.crt
-            - name: client-cert
-              mountPath: /opt/tls/client_cert.pem
-              subPath: client_cert.pem
-            - name: client-cert
-              mountPath: /opt/tls/client_key.pem
-              subPath: client_key.pem
-            - name: client-cert
-              mountPath: /opt/tls/client_ca.crt
-              subPath: client_ca.crt
+            {{- if or .Values.mtls.enabled $tlsEnabled }}
+            - name: tls-material
+              mountPath: /opt/tls
+              readOnly: true
             {{- end }}
             {{- with .Values.extraVolumeMounts }}
             {{- toYaml . | nindent 12 }}
@@ -157,27 +144,40 @@ spec:
       {{- with .Values.extraVolumes }}
       {{- toYaml . | nindent 8 }}
       {{- end }}
-      {{- if .Values.mtls.enabled }}
-        - name: server-cert
-          secret:
-            secretName: {{ include "rustfs.fullname" . }}-server-tls
-            items:
-              - key: tls.crt
-                path: rustfs_cert.pem
-              - key: tls.key
-                path: rustfs_key.pem
-              - key: ca.crt
-                path: ca.crt
-        - name: client-cert
-          secret:
-            secretName: {{ include "rustfs.fullname" . }}-client-tls
-            items:
-              - key: tls.crt
-                path: client_cert.pem
-              - key: tls.key
-                path: client_key.pem
-              - key: ca.crt
-                path: client_ca.crt
+      {{- if or .Values.mtls.enabled $tlsEnabled }}
+        - name: tls-material
+          projected:
+            defaultMode: 0440
+            sources:
+              - secret:
+                  name: {{ if .Values.mtls.enabled }}{{ include "rustfs.fullname" . }}-server-tls{{ else }}{{ required "tls.existingSecret is required when tls.enabled=true" $tls.existingSecret }}{{ end }}
+                  items:
+                    - key: tls.crt
+                      path: rustfs_cert.pem
+                    - key: tls.key
+                      path: rustfs_key.pem
+                    {{- if .Values.mtls.enabled }}
+                    - key: ca.crt
+                      path: ca.crt
+                    {{- end }}
+              {{- if .Values.mtls.enabled }}
+              - secret:
+                  name: {{ include "rustfs.fullname" . }}-client-tls
+                  items:
+                    - key: tls.crt
+                      path: client_cert.pem
+                    - key: tls.key
+                      path: client_key.pem
+                    - key: ca.crt
+                      path: client_ca.crt
+              {{- end }}
+              {{- if and $tlsEnabled $tls.trustSecret }}
+              - secret:
+                  name: {{ $tls.trustSecret }}
+                  items:
+                    - key: ca.crt
+                      path: public.crt
+              {{- end }}
       {{- end }}
       {{- if $logDirEnabled }}
         - name: logs

helm/rustfs/templates/statefulset.yaml

@@ -1,3 +1,5 @@
+{{- $tls := .Values.tls | default dict -}}
+{{- $tlsEnabled := $tls.enabled | default false -}}
 {{- $logDir := .Values.config.rustfs.obs_log_directory }}
 {{- $logDirEnabled := ne $logDir "" }}
 {{- $poolsEnabled := and .Values.pools .Values.pools.enabled }}
@@ -217,25 +219,10 @@ spec:
             {{- toYaml $.Values.resources | nindent 12 }}
           {{- include "rustfs.probes" $ | nindent 10 }}
           volumeMounts:
-            {{- if $.Values.mtls.enabled }}
-            - name: client-cert
-              mountPath: /opt/tls/client_cert.pem
-              subPath: client_cert.pem
-            - name: client-cert
-              mountPath: /opt/tls/client_key.pem
-              subPath: client_key.pem
-            - name: client-cert
-              mountPath: /opt/tls/client_ca.crt
-              subPath: client_ca.crt
-            - name: server-cert
-              mountPath: /opt/tls/rustfs_cert.pem
-              subPath: rustfs_cert.pem
-            - name: server-cert
-              mountPath: /opt/tls/rustfs_key.pem
-              subPath: rustfs_key.pem
-            - name: server-cert
-              mountPath: /opt/tls/ca.crt
-              subPath: ca.crt
+            {{- if or $.Values.mtls.enabled $tlsEnabled }}
+            - name: tls-material
+              mountPath: /opt/tls
+              readOnly: true
             {{- end }}
             {{- with $.Values.extraVolumeMounts }}
             {{- toYaml . | nindent 12 }}
@@ -254,29 +241,42 @@ spec:
             - name: data
               mountPath: /data
             {{- end }}
-      {{- if or $.Values.mtls.enabled $.Values.extraVolumes }}
+      {{- if or $.Values.mtls.enabled $tlsEnabled $.Values.extraVolumes }}
       volumes:
-        {{- if $.Values.mtls.enabled }}
-        - name: server-cert
-          secret:
-            secretName: {{ include "rustfs.fullname" $ }}-server-tls
-            items:
-              - key: tls.crt
-                path: rustfs_cert.pem
-              - key: tls.key
-                path: rustfs_key.pem
-              - key: ca.crt
-                path: ca.crt
-        - name: client-cert
-          secret:
-            secretName: {{ include "rustfs.fullname" $ }}-client-tls
-            items:
-              - key: tls.crt
-                path: client_cert.pem
-              - key: tls.key
-                path: client_key.pem
-              - key: ca.crt
-                path: client_ca.crt
+        {{- if or $.Values.mtls.enabled $tlsEnabled }}
+        - name: tls-material
+          projected:
+            defaultMode: 0440
+            sources:
+              - secret:
+                  name: {{ if $.Values.mtls.enabled }}{{ include "rustfs.fullname" $ }}-server-tls{{ else }}{{ required "tls.existingSecret is required when tls.enabled=true" $tls.existingSecret }}{{ end }}
+                  items:
+                    - key: tls.crt
+                      path: rustfs_cert.pem
+                    - key: tls.key
+                      path: rustfs_key.pem
+                    {{- if $.Values.mtls.enabled }}
+                    - key: ca.crt
+                      path: ca.crt
+                    {{- end }}
+              {{- if $.Values.mtls.enabled }}
+              - secret:
+                  name: {{ include "rustfs.fullname" $ }}-client-tls
+                  items:
+                    - key: tls.crt
+                      path: client_cert.pem
+                    - key: tls.key
+                      path: client_key.pem
+                    - key: ca.crt
+                      path: client_ca.crt
+              {{- end }}
+              {{- if and $tlsEnabled $tls.trustSecret }}
+              - secret:
+                  name: {{ $tls.trustSecret }}
+                  items:
+                    - key: ca.crt
+                      path: public.crt
+              {{- end }}
         {{- end }}
         {{- with $.Values.extraVolumes }}
         {{- toYaml . | nindent 8 }}

helm/rustfs/values.yaml

@@ -388,6 +388,14 @@ gatewayApi:
     name: ""
     namespace: ""
 
+# Native TLS identity managed by an external certificate controller.
+# Public TLS and an mTLS backend use separate listeners or an ingress terminator.
+tls:
+  enabled: false
+  existingSecret: "" # kubernetes.io/tls: tls.crt and tls.key only
+  trustSecret: "" # Optional independent outbound trust bundle: ca.crt
+  reloadInterval: 30
+
 mtls:
   enabled: false
   clientCertPath: "/opt/tls/client_cert.pem"

rustfs/src/admin/handlers/mod.rs

@@ -70,6 +70,7 @@ pub mod system;
 pub mod table_catalog;
 mod target_descriptor;
 pub mod tier;
+pub mod tls_certificate;
 pub mod tls_debug;
 pub mod trace;
 pub mod usage_prefix;

rustfs/src/admin/handlers/tls_certificate.rs

@@ -0,0 +1,177 @@
+// 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 crate::admin::{
+    auth::authorize_admin_request,
+    router::{AdminOperation, Operation, S3Router},
+    storage_api::s3::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result},
+};
+use crate::server::{
+    ADMIN_PREFIX, NativeTlsTransport,
+    tls_publish::{self, PublishError, PublishRequest},
+};
+use http::{HeaderMap, HeaderValue, StatusCode};
+use hyper::Method;
+use matchit::Params;
+use rustfs_policy::policy::{action::Action, action::AdminAction};
+
+pub fn register_tls_certificate_routes(router: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    router.insert(
+        Method::GET,
+        &format!("{ADMIN_PREFIX}/v3/tls/certificate-state"),
+        AdminOperation(&CertificateStateHandler),
+    )?;
+    router.insert(
+        Method::POST,
+        &format!("{ADMIN_PREFIX}/v3/tls/certificate"),
+        AdminOperation(&PublishCertificateHandler),
+    )?;
+    Ok(())
+}
+
+pub struct CertificateStateHandler;
+pub struct PublishCertificateHandler;
+
+fn require_native_tls(req: &S3Request<Body>) -> S3Result<()> {
+    if !req
+        .extensions
+        .get::<NativeTlsTransport>()
+        .is_some_and(|transport| transport.0)
+    {
+        return Err(S3Error::with_message(
+            S3ErrorCode::AccessDenied,
+            "certificate management requires TLS on the RustFS listener",
+        ));
+    }
+    Ok(())
+}
+
+async fn authorize_certificate_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
+    if req.credentials.is_none() {
+        return Err(S3Error::with_message(
+            S3ErrorCode::AccessDenied,
+            "certificate management requires credentials",
+        ));
+    }
+    authorize_admin_request(req, vec![Action::AdminAction(action)]).await?;
+    require_native_tls(req)
+}
+
+fn tls_path() -> S3Result<String> {
+    crate::server::tls_material::active_server_tls_path()
+        .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "certificate publication is not configured"))
+}
+
+fn publication_error(error: PublishError) -> S3Error {
+    let (code, status) = match error {
+        PublishError::Conflict => (S3ErrorCode::InvalidRequest, StatusCode::CONFLICT),
+        PublishError::Disabled => (S3ErrorCode::InvalidRequest, StatusCode::SERVICE_UNAVAILABLE),
+        PublishError::Invalid => (S3ErrorCode::InvalidRequest, StatusCode::BAD_REQUEST),
+        PublishError::Storage | PublishError::Busy => (S3ErrorCode::InternalError, StatusCode::SERVICE_UNAVAILABLE),
+    };
+    let mut result = S3Error::with_message(code, error.to_string());
+    result.set_status_code(status);
+    result
+}
+
+fn json_response(value: &impl serde::Serialize) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let bytes = serde_json::to_vec(value)
+        .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "certificate status serialization failed"))?;
+    let mut headers = HeaderMap::new();
+    headers.insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
+    headers.insert(http::header::CACHE_CONTROL, HeaderValue::from_static("no-store"));
+    Ok(S3Response::with_headers((StatusCode::OK, Body::from(bytes)), headers))
+}
+
+#[async_trait::async_trait]
+impl Operation for CertificateStateHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_certificate_request(&req, AdminAction::TLSCertificateStatusAdminAction).await?;
+        json_response(&tls_publish::certificate_state(tls_path()?).await.map_err(publication_error)?)
+    }
+}
+
+#[async_trait::async_trait]
+impl Operation for PublishCertificateHandler {
+    async fn call(&self, mut req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_certificate_request(&req, AdminAction::TLSCertificatePublishAdminAction).await?;
+        let body = req.input.store_all_limited(128 * 1024).await.map_err(|_| {
+            S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                "certificate publication body exceeds the limit or is unreadable",
+            )
+        })?;
+        let request: PublishRequest = serde_json::from_slice(&body)
+            .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidRequest, "invalid certificate publication JSON"))?;
+        json_response(&tls_publish::publish(tls_path()?, request).await.map_err(publication_error)?)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn request(method: Method) -> S3Request<Body> {
+        S3Request {
+            input: Body::empty(),
+            method,
+            uri: "/rustfs/admin/v3/tls/certificate".parse().expect("URI"),
+            headers: HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: None,
+            region: None,
+            service: None,
+            trailing_headers: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn certificate_handlers_reject_missing_credentials() {
+        assert_eq!(
+            CertificateStateHandler
+                .call(request(Method::GET), Params::new())
+                .await
+                .expect_err("unauthenticated GET")
+                .code(),
+            &S3ErrorCode::AccessDenied
+        );
+        assert_eq!(
+            PublishCertificateHandler
+                .call(request(Method::POST), Params::new())
+                .await
+                .expect_err("unauthenticated POST")
+                .code(),
+            &S3ErrorCode::AccessDenied
+        );
+    }
+
+    #[test]
+    fn forwarded_protocol_never_authorizes_private_key_upload() {
+        let mut request = request(Method::POST);
+        request.headers.insert("x-forwarded-proto", HeaderValue::from_static("https"));
+        assert!(require_native_tls(&request).is_err());
+        request.extensions.insert(NativeTlsTransport(false));
+        assert!(require_native_tls(&request).is_err());
+        request.extensions.insert(NativeTlsTransport(true));
+        assert!(require_native_tls(&request).is_ok());
+    }
+
+    #[test]
+    fn unknown_json_fields_are_rejected_without_echoing_material() {
+        let error = serde_json::from_str::<PublishRequest>(
+            r#"{"schema_version":"x","resource_id":"x","expected_target_revision":"x","issuance_id":"00000000-0000-0000-0000-000000000001","certificate_pem":"cert","private_key_pem":"secret","extra":true}"#,
+        );
+        assert!(error.is_err());
+    }
+}

rustfs/src/admin/mod.rs

@@ -103,6 +103,7 @@ fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()
     inspect_archive::register_inspect_archive_route(r)?;
     gateway_key_inventory::register_gateway_key_inventory_route(r)?;
     tls_debug::register_tls_debug_route(r)?;
+    handlers::tls_certificate::register_tls_certificate_routes(r)?;
     kms::register_kms_route(r)?;
     oidc::register_oidc_route(r)?;
     idp_compat::register_idp_compat_route(r)?;

rustfs/src/admin/route_policy.rs

@@ -21,6 +21,8 @@ const ALL_ADMIN: AdminActionRef = AdminActionRef::new("AllAdminActions");
 const ADD_USER_TO_GROUP: AdminActionRef = AdminActionRef::new("AddUserToGroupAdminAction");
 const ATTACH_POLICY: AdminActionRef = AdminActionRef::new("AttachPolicyAdminAction");
 const CONFIG_UPDATE: AdminActionRef = AdminActionRef::new("ConfigUpdateAdminAction");
+const TLS_CERTIFICATE_PUBLISH: AdminActionRef = AdminActionRef::new("TLSCertificatePublishAdminAction");
+const TLS_CERTIFICATE_STATUS: AdminActionRef = AdminActionRef::new("TLSCertificateStatusAdminAction");
 const CONSOLE_LOG: AdminActionRef = AdminActionRef::new("ConsoleLogAdminAction");
 const COMMIT_TABLE: AdminActionRef = AdminActionRef::new("CommitTableAction");
 const CREATE_POLICY: AdminActionRef = AdminActionRef::new("CreatePolicyAdminAction");
@@ -501,6 +503,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
     ),
     admin(HttpMethod::Get, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
     admin(HttpMethod::Put, "/rustfs/admin/v3/config", CONFIG_UPDATE, RouteRiskLevel::High),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/tls/certificate-state",
+        TLS_CERTIFICATE_STATUS,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/tls/certificate",
+        TLS_CERTIFICATE_PUBLISH,
+        RouteRiskLevel::High,
+    ),
     admin(HttpMethod::Get, "/rustfs/admin/v3/scanner/status", SERVER_INFO, RouteRiskLevel::Sensitive),
     admin(
         HttpMethod::Post,
@@ -2233,6 +2247,14 @@ mod tests {
         }
     }
 
+    #[test]
+    fn route_policy_records_dedicated_tls_actions() {
+        assert_action(HttpMethod::Get, "/rustfs/admin/v3/tls/certificate-state", TLS_CERTIFICATE_STATUS);
+        assert_action(HttpMethod::Post, "/rustfs/admin/v3/tls/certificate", TLS_CERTIFICATE_PUBLISH);
+        assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/tls/certificate", CONFIG_UPDATE);
+        assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/tls/certificate-state", TLS_CERTIFICATE_PUBLISH);
+    }
+
     /// `Critical` marks the routes whose worst case is permanent loss of user
     /// data, so the inventory is pinned in both directions: the KMS key
     /// deletion route may not be quietly downgraded, and no other route may be

rustfs/src/admin/route_registration_test.rs

@@ -302,6 +302,8 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         admin_route(Method::PUT, "/v3/restore-config-history-kv"),
         admin_route(Method::GET, "/v3/config"),
         admin_route(Method::PUT, "/v3/config"),
+        admin_route(Method::GET, "/v3/tls/certificate-state"),
+        admin_route(Method::POST, "/v3/tls/certificate"),
         admin_route(Method::GET, "/v3/scanner/status"),
         admin_route(Method::POST, "/v3/scanner/cycle-state/reset"),
         admin_route(Method::POST, "/v3/scanner/usage-state/reset"),

rustfs/src/server/http.rs

@@ -32,12 +32,12 @@ use crate::server::{
     ssec_transport::SsecTransportLayer,
     strip_valid_port_suffix,
     tls_material::{
-        TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline, build_acceptor_from_loaded,
+        PreparedServerTls, TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline,
         load_tls_material, spawn_reload_loop,
     },
 };
 #[cfg(feature = "http3")]
-use crate::server::{http3, hybrid::HybridBody, tls_material::build_http3_server_config_from_loaded};
+use crate::server::{http3, hybrid::HybridBody};
 use crate::storage_api::server::http as storage;
 use crate::storage_api::server::http::rpc::InternodeRpcService;
 #[cfg(test)]
@@ -1448,6 +1448,7 @@ pub async fn start_http_server(
     // Load TLS materials and build server acceptor in a single pass.
     // Outbound material (root CAs, mTLS identity) was already published in main.rs;
     // this load is needed for the server-side TLS acceptor and reload loop.
+    let mut tls_initial_snapshot = None;
     let tls_acceptor = if tls_path_configured {
         let snapshot = load_tls_material(tls_path).await.map_err(|e| {
             Error::other(format!(
@@ -1456,27 +1457,30 @@ pub async fn start_http_server(
             ))
         })?;
 
-        #[cfg(feature = "http3")]
-        if http3_enabled {
-            http3_server_config = build_http3_server_config_from_loaded(snapshot.server.as_ref(), std::path::Path::new(tls_path))
-                .await
-                .map_err(|e| Error::other(format!("HTTP/3 TLS configuration failed: {e}")))?;
-        }
-
-        let acceptor = build_acceptor_from_loaded(snapshot.server, std::path::Path::new(tls_path))
-            .await
-            .map_err(|e| Error::other(e.to_string()))?;
+        let prepared = PreparedServerTls::from_snapshot(&snapshot).map_err(|e| Error::other(e.to_string()))?;
 
         // Fail closed: if TLS was explicitly configured but no server certificates
         // were found, refuse to start rather than silently falling back to plain HTTP.
-        match acceptor {
+        match prepared {
             None => {
                 return Err(Error::other(format!(
                     "TLS is explicitly configured via RUSTFS_TLS_PATH/tls_path='{}' but no server certificates were found",
                     tls_path
                 )));
             }
-            Some(a) => Some(a),
+            Some(prepared) => {
+                #[cfg(feature = "http3")]
+                if http3_enabled {
+                    http3_server_config = Some(
+                        prepared
+                            .http3_config()
+                            .map_err(|e| Error::other(format!("HTTP/3 TLS configuration failed: {e}")))?,
+                    );
+                }
+                let acceptor = prepared.tcp_acceptor().map_err(|e| Error::other(e.to_string()))?;
+                tls_initial_snapshot = Some((snapshot, prepared));
+                Some(acceptor)
+            }
         }
     } else {
         None
@@ -1755,14 +1759,20 @@ pub async fn start_http_server(
     } else {
         None
     };
-    let tls_reload_task = tls_acceptor.as_ref().and_then(|holder| {
+    let tls_reload_task = if let (Some(holder), Some((initial, prepared))) = (tls_acceptor.as_ref(), tls_initial_snapshot.take())
+    {
         spawn_reload_loop(
             tls_path.to_string(),
             holder.clone(),
+            initial,
+            prepared,
             #[cfg(feature = "http3")]
             http3_server.as_ref().map(|(endpoint, _)| endpoint.clone()),
         )
-    });
+        .map_err(|e| Error::other(e.to_string()))?
+    } else {
+        None
+    };
 
     let is_console = config.console_enable;
     let server_domains_configured = !config.server_domains.is_empty();

rustfs/src/server/mod.rs

@@ -29,6 +29,7 @@ pub(crate) mod runtime_sources;
 mod service_state;
 mod ssec_transport;
 pub mod tls_material;
+pub(crate) mod tls_publish;
 
 #[cfg(feature = "http3")]
 mod http3;
@@ -85,6 +86,7 @@ pub(crate) use readiness::collect_node_readiness_report;
 pub use readiness::publish_ready_when_runtime_ready;
 pub(crate) use readiness::snapshot_dependency_readiness_report;
 pub(crate) use readiness::{collect_cluster_read_health_report, collect_cluster_write_health_report};
+pub(crate) use ssec_transport::NativeTlsTransport;
 
 pub use crate::shared_types::RemoteAddr;
 

rustfs/src/server/ssec_transport.rs

@@ -63,6 +63,10 @@ pub(crate) const METRIC_SSEC_PLAINTEXT_REQUESTS_TOTAL: &str = "rustfs_ssec_plain
 type BoxError = Box<dyn std::error::Error + Send + Sync>;
 type BoxBody = http_body_util::combinators::UnsyncBoxBody<Bytes, BoxError>;
 
+/// Set only by the listener stack; proxy protocol assertions do not establish this marker.
+#[derive(Clone, Copy)]
+pub(crate) struct NativeTlsTransport(pub(crate) bool);
+
 /// Every header that carries, or announces, a customer-provided key: the
 /// object's own triple and the copy-source triple that CopyObject and
 /// UploadPartCopy use to read an SSE-C source.
@@ -191,7 +195,8 @@ where
         self.inner.poll_ready(cx)
     }
 
-    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
+    fn call(&mut self, mut req: Request<ReqBody>) -> Self::Future {
+        req.extensions_mut().insert(NativeTlsTransport(self.connection_is_tls));
         if !carries_ssec_headers(req.headers())
             || is_secure_transport(self.connection_is_tls, req.extensions().get::<ClientInfo>())
         {

rustfs/src/startup_runtime_sources.rs

@@ -17,6 +17,7 @@ use crate::runtime_sources::current_outbound_tls_generation as runtime_current_o
 use crate::storage_api::startup::runtime_sources::{InstanceContext, set_global_rustfs_port};
 use rustfs_kms::KmsServiceManager;
 use rustfs_obs::{GlobalError as ObservabilityError, OtelGuard};
+pub(crate) use rustfs_tls_runtime::load_global_outbound_tls_state as current_outbound_tls_state;
 use rustfs_tls_runtime::{OutboundTlsMaterial, TlsGeneration};
 use std::sync::Arc;
 use tokio_util::sync::CancellationToken;

rustfs/src/startup_tls_material.rs

@@ -14,7 +14,9 @@
 
 use crate::config::Config;
 use crate::startup_runtime_sources;
+use std::collections::BTreeSet;
 use std::io::{Error, Result};
+use std::sync::Mutex;
 use tracing::{error, info};
 
 const LOG_COMPONENT_MAIN: &str = "main";
@@ -23,6 +25,25 @@ const EVENT_TLS_OUTBOUND_INITIALIZED: &str = "tls_outbound_initialized";
 const EVENT_TLS_OUTBOUND_INITIALIZATION_FAILED: &str = "tls_outbound_initialization_failed";
 const TLS_STARTUP_GENERATION_CONSUMER: &str = "rustfs_server_startup";
 
+// Startup completes before listeners register; retain source names independently
+// of the later listener snapshot, without pinning renewable certificate bytes.
+static STARTUP_MATERIAL_FILES: Mutex<Option<(String, BTreeSet<String>)>> = Mutex::new(None);
+
+pub(crate) fn required_startup_material_files(tls_path: &str) -> BTreeSet<String> {
+    STARTUP_MATERIAL_FILES
+        .lock()
+        .unwrap_or_else(|error| error.into_inner())
+        .as_ref()
+        .filter(|(path, _)| path == tls_path.trim())
+        .map(|(_, files)| files.clone())
+        .unwrap_or_default()
+}
+
+#[cfg(test)]
+pub(crate) fn replace_startup_material_files(files: Option<(String, BTreeSet<String>)>) -> Option<(String, BTreeSet<String>)> {
+    std::mem::replace(&mut *STARTUP_MATERIAL_FILES.lock().unwrap_or_else(|error| error.into_inner()), files)
+}
+
 pub(crate) async fn init_outbound_tls_material(config: &Config) -> Result<()> {
     crate::server::tls_material::validate_configured_oidc_extra_ca_cert()
         .await
@@ -32,6 +53,8 @@ pub(crate) async fn init_outbound_tls_material(config: &Config) -> Result<()> {
         match crate::server::tls_material::load_tls_material(tls_path).await {
             Ok(snapshot) => {
                 publish_outbound_tls_material(&snapshot.outbound, Some(tls_path)).await;
+                *STARTUP_MATERIAL_FILES.lock().unwrap_or_else(|error| error.into_inner()) =
+                    Some((tls_path.to_string(), snapshot.material_files));
             }
             Err(err) => {
                 error!(
@@ -49,6 +72,11 @@ pub(crate) async fn init_outbound_tls_material(config: &Config) -> Result<()> {
         if startup_runtime_sources::observability_metric_enabled() {
             startup_runtime_sources::init_tls_metrics();
         }
+    } else {
+        STARTUP_MATERIAL_FILES
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .take();
     }
 
     Ok(())

scripts/test_helm_templates.sh

@@ -568,7 +568,7 @@ nondefault_statefulset=$(awk '
 nondefault_local_host=$(statefulset_env_value RUSTFS_LOCAL_ENDPOINT_HOST <<<"$nondefault_statefulset")
 nondefault_namespace=$(yq eval '.metadata.namespace' - <<<"$nondefault_statefulset")
 nondefault_endpoint_port=$(yq eval '.spec.template.spec.containers[0].ports[] | select(.name == "endpoint") | .containerPort' - <<<"$nondefault_statefulset")
-mtls_liveness_command=$(yq eval '.spec.template.spec.containers[0].livenessProbe.exec.command[-1]' - <<<"$nondefault_statefulset")
+mtls_liveness_port=$(yq eval '.spec.template.spec.containers[0].livenessProbe.tcpSocket.port' - <<<"$nondefault_statefulset")
 mtls_readiness_command=$(yq eval '.spec.template.spec.containers[0].readinessProbe.exec.command[-1]' - <<<"$nondefault_statefulset")
 if [[ "$nondefault_local_host" != '$(RUSTFS_CHART_POD_NAME).tenant-rustfs-headless.object-storage.svc.cluster.corp' ]]; then
   echo "Combined generated topology produced an unexpected local endpoint host: $nondefault_local_host" >&2
@@ -578,9 +578,8 @@ if [[ "$nondefault_namespace" != "object-storage" || "$nondefault_endpoint_port"
   echo "Combined generated topology must retain its namespace and custom endpoint port" >&2
   exit 1
 fi
-if ! grep -Fq 'https://127.0.0.1:9443/health' <<<"$mtls_liveness_command" ||
-   grep -Fq '/health/ready' <<<"$mtls_liveness_command"; then
-  echo "mTLS liveness must use the process health endpoint, not readiness" >&2
+if [[ "$mtls_liveness_port" != 9443 ]]; then
+  echo "mTLS liveness must check the listener independently of certificate validity" >&2
   exit 1
 fi
 if ! grep -Fq 'https://127.0.0.1:9443/health/ready' <<<"$mtls_readiness_command"; then

scripts/test_helm_tls_reload.sh

@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)
+CHART_DIR="$ROOT_DIR/helm/rustfs"
+
+for distributed in false true; do
+  rendered=$(helm template tls-reload "$CHART_DIR" \
+    --set secret.allowInsecureDefaults=true \
+    --set mode.distributed.enabled="$distributed" \
+    --set mode.standalone.enabled="$([[ "$distributed" == false ]] && echo true || echo false)" \
+    --set tls.enabled=true --set tls.existingSecret=issued-tls)
+  workload=$(yq eval 'select(.kind == "Deployment" or .kind == "StatefulSet")' - <<<"$rendered")
+  mount=$(yq eval '.spec.template.spec.containers[0].volumeMounts[] | select(.name == "tls-material")' - <<<"$workload")
+  [[ $(yq eval '.mountPath' - <<<"$mount") == /opt/tls ]]
+  [[ $(yq eval '.readOnly' - <<<"$mount") == true ]]
+  [[ $(yq eval 'has("subPath")' - <<<"$mount") == false ]]
+  keys=$(yq eval '.spec.template.spec.volumes[] | select(.name == "tls-material") | .projected.sources[].secret.items[].key' - <<<"$workload")
+  [[ "$keys" == $'tls.crt\ntls.key' ]]
+  [[ $(yq eval '.spec.template.spec.containers[0].readinessProbe.httpGet.scheme' - <<<"$workload") == HTTPS ]]
+  [[ $(yq eval '.spec.template.spec.containers[0].livenessProbe.tcpSocket.port' - <<<"$workload") == 9000 ]]
+  config=$(yq eval 'select(.kind == "ConfigMap") | .data' - <<<"$rendered")
+  [[ $(yq eval '.RUSTFS_TLS_RELOAD_ENABLE' - <<<"$config") == 1 ]]
+  [[ $(yq eval '.RUSTFS_TLS_KEYLOG' - <<<"$config") == 0 ]]
+  if [[ "$distributed" == true ]]; then
+    [[ $(yq eval '.RUSTFS_VOLUMES' - <<<"$config") == https://* ]]
+  fi
+
+  mtls=$(helm template tls-reload "$CHART_DIR" --set secret.allowInsecureDefaults=true \
+    --set mode.distributed.enabled="$distributed" \
+    --set mode.standalone.enabled="$([[ "$distributed" == false ]] && echo true || echo false)" \
+    --set mtls.enabled=true)
+  workload=$(yq eval 'select(.kind == "Deployment" or .kind == "StatefulSet")' - <<<"$mtls")
+  [[ $(yq eval '.spec.template.spec.volumes[] | select(.name == "tls-material") | .projected.sources | length' - <<<"$workload") == 2 ]]
+  [[ $(yq eval '.spec.template.spec.containers[0].volumeMounts[] | select(.name == "tls-material") | has("subPath")' - <<<"$workload") == false ]]
+  [[ $(yq eval '.spec.template.spec.containers[0].livenessProbe.tcpSocket.port' - <<<"$workload") == 9000 ]]
+done
+
+if helm template tls-reload "$CHART_DIR" --set secret.allowInsecureDefaults=true \
+  --set tls.enabled=true --set tls.existingSecret=issued-tls --set mtls.enabled=true >/dev/null 2>&1; then
+  echo "Public TLS must not silently enable global client authentication" >&2
+  exit 1
+fi
+if helm template tls-reload "$CHART_DIR" --set secret.allowInsecureDefaults=true \
+  --set tls.enabled=true >/dev/null 2>&1; then
+  echo "Native TLS requires an explicitly configured serving Secret" >&2
+  exit 1
+fi
+echo "TLS Secret projection, rotation settings, topology and probes passed"