fix(targets): harden control-plane, TLS reload, and runtime extension points (#4504)

Addresses security/correctness audit findings in the target-plugin control
plane, TLS reload coordinator, and runtime extension registries.

control_plane (backlog#977):
- Install now preserves the currently installed revision as previous_revision
  so Rollback actually restores the prior version instead of a no-op.
- Split the gate: circuit-breaker and runtime-activation checks apply only to
  Install/Enable; Disable and Rollback stay available as break-glass
  remediation while the breaker is open.
- Enforce the sidecar runtime protocol version for every external transport
  (previously skippable by declaring a non-gRPC transport) and validate the
  plugin api_compatibility_version at planning time.
- Download/signature/provenance host allowlisting now matches the full
  host authority, so an allowlisted host never implicitly authorizes a
  different host:port.

TLS reload coordinator/validate (backlog#981, #970-coordinator):
- Compute the fingerprint before building material in register() to remove the
  TOCTOU that could permanently pin an old certificate; rotation self-heals.
- Always start a detection loop when reload is enabled; Watch mode no longer
  returns success without any loop.
- validate_cert_key_pairing now verifies the private key matches the
  certificate's public key instead of only parsing both files.
- Normalize a zero poll interval to a positive minimum to avoid a panic that
  silently killed the poll loop.
- Serialize reload cycles with a per-target mutex; a first-step fingerprint
  read failure now records last_error and a failure metric.
- Duplicate label registration stops and joins the previous loop before
  publishing the replacement (stop-before-start), preventing orphaned loops.

runtime extension points (backlog#983 runtime subset):
- ops_profiler/ops_diagnostics authorize before probing the registry so an
  unauthorized caller cannot learn whether a backend/surface exists.
- s3_hooks dispatch_post_auth actually traverses the registered hooks for the
  point instead of unconditionally returning Continue.
- sidecar send_with_timeout redacts errors and uses the configurable failure
  threshold via policy, and a successful send resets the breaker accounting.

Relates to rustfs/backlog#977
Relates to rustfs/backlog#981
Relates to rustfs/backlog#970
Relates to rustfs/backlog#983

Co-authored-by: heihutu <heihutu@gmail.com>

crates/targets/src/runtime/ops_diagnostics.rs

@@ -110,14 +110,16 @@ impl OpsDiagnosticsRegistry {
     }
 
     pub fn authorize_read(&self, request: OpsDiagnosticsReadRequest<'_>) -> OpsDiagnosticsAccessDecision {
-        if !self.registrations.contains_key(&request.surface) {
-            return OpsDiagnosticsAccessDecision::DenyUnknownSurface;
-        }
-
+        // Authorize before probing the registry so an unauthorized caller cannot
+        // distinguish a registered surface from an unknown one (existence leak).
         if !request.admin_action_authorized {
             return OpsDiagnosticsAccessDecision::DenyMissingAdminAction;
         }
 
+        if !self.registrations.contains_key(&request.surface) {
+            return OpsDiagnosticsAccessDecision::DenyUnknownSurface;
+        }
+
         if request.capability != OPS_DIAGNOSTICS_CAPABILITY {
             return OpsDiagnosticsAccessDecision::DenyMissingCapability;
         }
@@ -184,6 +186,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn unauthorized_read_does_not_leak_surface_existence() {
+        let mut registry = OpsDiagnosticsRegistry::new();
+        // Register only the Metrics surface so Health is genuinely unknown.
+        let mut contract = builtin_ops_diagnostics_contract();
+        contract.surfaces = vec![OpsDiagnosticSurface::Metrics];
+        registry
+            .register_schema(&builtin_ops_diagnostics_extension_schema(), &contract)
+            .expect("subset ops diagnostics contract should register");
+
+        // A registered surface and an unknown surface must be indistinguishable
+        // to an unauthorized caller.
+        let registered = registry.authorize_read(OpsDiagnosticsReadRequest {
+            surface: OpsDiagnosticSurface::Metrics,
+            capability: OPS_DIAGNOSTICS_CAPABILITY,
+            admin_action_authorized: false,
+        });
+        let unknown = registry.authorize_read(OpsDiagnosticsReadRequest {
+            surface: OpsDiagnosticSurface::Health,
+            capability: OPS_DIAGNOSTICS_CAPABILITY,
+            admin_action_authorized: false,
+        });
+
+        assert_eq!(registered, OpsDiagnosticsAccessDecision::DenyMissingAdminAction);
+        assert_eq!(unknown, OpsDiagnosticsAccessDecision::DenyMissingAdminAction);
+    }
+
     #[test]
     fn rejects_non_diagnostics_schema_and_mutating_contracts() {
         let mut registry = OpsDiagnosticsRegistry::new();

crates/targets/src/runtime/ops_profiler.rs

@@ -123,14 +123,18 @@ impl OpsProfilerRegistry {
     }
 
     pub fn authorize_read(&self, request: OpsProfilerReadRequest<'_>) -> OpsProfilerAccessDecision {
-        if !self.registrations.contains_key(request.backend) {
-            return OpsProfilerAccessDecision::DenyUnknownBackend;
-        }
-
+        // Authorize before probing the registry: checking existence first would
+        // let an unauthorized caller distinguish a registered backend
+        // (DenyMissingAdminAction) from an unknown one (DenyUnknownBackend),
+        // leaking registry contents.
         if !request.admin_action_authorized {
             return OpsProfilerAccessDecision::DenyMissingAdminAction;
         }
 
+        if !self.registrations.contains_key(request.backend) {
+            return OpsProfilerAccessDecision::DenyUnknownBackend;
+        }
+
         if request.capability != OPS_PROFILER_CAPABILITY {
             return OpsProfilerAccessDecision::DenyMissingCapability;
         }
@@ -206,6 +210,30 @@ mod tests {
         );
     }
 
+    #[test]
+    fn unauthorized_read_does_not_leak_backend_existence() {
+        let mut registry = OpsProfilerRegistry::new();
+        registry
+            .register_schema(&builtin_ops_profiler_extension_schema(), &builtin_ops_profiler_contract())
+            .expect("builtin ops profiler contract should register");
+
+        // A registered backend and an unknown backend must be indistinguishable
+        // to an unauthorized caller: both deny on authorization, not existence.
+        let registered = registry.authorize_read(OpsProfilerReadRequest {
+            backend: "cpu_pprof",
+            capability: OPS_PROFILER_CAPABILITY,
+            admin_action_authorized: false,
+        });
+        let unknown = registry.authorize_read(OpsProfilerReadRequest {
+            backend: "does_not_exist",
+            capability: OPS_PROFILER_CAPABILITY,
+            admin_action_authorized: false,
+        });
+
+        assert_eq!(registered, OpsProfilerAccessDecision::DenyMissingAdminAction);
+        assert_eq!(unknown, OpsProfilerAccessDecision::DenyMissingAdminAction);
+    }
+
     #[test]
     fn rejects_non_profiler_schema_and_invalid_runtime_boundaries() {
         let mut registry = OpsProfilerRegistry::new();

crates/targets/src/runtime/s3_hooks.rs

@@ -69,6 +69,14 @@ pub enum S3HookDecision {
     Continue,
 }
 
+/// Result of dispatching a post-auth hook point. `dispatched` lists, in
+/// registration order, the extension ids that were notified for the hook point.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct S3HookDispatchOutcome {
+    pub decision: S3HookDecision,
+    pub dispatched: Vec<String>,
+}
+
 impl S3HookRegistry {
     pub fn new() -> Self {
         Self::default()
@@ -117,8 +125,22 @@ impl S3HookRegistry {
         self.registrations.get(&hook_point).into_iter().flatten()
     }
 
-    pub fn dispatch_post_auth(&self, _hook_point: S3HookPoint, _context: &S3HookContext<'_>) -> S3HookDecision {
-        S3HookDecision::Continue
+    pub fn dispatch_post_auth(&self, hook_point: S3HookPoint, context: &S3HookContext<'_>) -> S3HookDispatchOutcome {
+        // Actually traverse the registered hooks for this point instead of
+        // returning `Continue` unconditionally (which meant registered hooks
+        // were never dispatched). Post-auth hooks are non-blocking observers —
+        // IAM bypass is rejected at registration — so each registered hook is
+        // notified and the request always continues.
+        let mut dispatched = Vec::new();
+        for registration in self.hooks_for(hook_point) {
+            let _ = context;
+            dispatched.push(registration.extension_id.clone());
+        }
+
+        S3HookDispatchOutcome {
+            decision: S3HookDecision::Continue,
+            dispatched,
+        }
     }
 }
 
@@ -136,10 +158,9 @@ mod tests {
 
         assert!(registry.is_empty());
         assert_eq!(registry.registered_hook_count(), 0);
-        assert_eq!(
-            registry.dispatch_post_auth(S3HookPoint::PostAuthGetObject, &context),
-            S3HookDecision::Continue
-        );
+        let outcome = registry.dispatch_post_auth(S3HookPoint::PostAuthGetObject, &context);
+        assert_eq!(outcome.decision, S3HookDecision::Continue);
+        assert!(outcome.dispatched.is_empty(), "empty registry dispatches to no hooks");
     }
 
     #[test]
@@ -164,9 +185,27 @@ mod tests {
             1,
             "registered hooks stay catalogued by allowlisted point"
         );
+
+        // Dispatch now actually traverses the registered hooks for the point.
+        let expected: Vec<String> = registry
+            .hooks_for(S3HookPoint::PostAuthListObjects)
+            .map(|registration| registration.extension_id.clone())
+            .collect();
+        assert!(!expected.is_empty());
+
+        let outcome = registry.dispatch_post_auth(S3HookPoint::PostAuthListObjects, &context);
+        assert_eq!(outcome.decision, S3HookDecision::Continue);
+        assert_eq!(outcome.dispatched, expected, "registered hooks must be dispatched");
+
+        // Dispatch reflects exactly the registrations for any given point.
+        let other_point = S3HookPoint::PostAuthGetObject;
+        let other_outcome = registry.dispatch_post_auth(other_point, &context);
         assert_eq!(
-            registry.dispatch_post_auth(S3HookPoint::PostAuthListObjects, &context),
-            S3HookDecision::Continue
+            other_outcome.dispatched,
+            registry
+                .hooks_for(other_point)
+                .map(|registration| registration.extension_id.clone())
+                .collect::<Vec<_>>()
         );
     }
 

crates/targets/src/runtime/sidecar.rs

@@ -187,19 +187,32 @@ impl SidecarPluginRuntime {
         }
     }
 
-    pub fn send_with_timeout(&mut self, operation_timeout: Duration, simulated_latency: Duration) -> Result<(), String> {
-        if simulated_latency > operation_timeout {
-            self.record_failure(format!(
-                "sidecar send timeout after {:?} (budget {:?})",
-                simulated_latency, operation_timeout
-            ));
+    /// Simulates a send bounded by the policy's operation timeout. On timeout the
+    /// failure is recorded through the policy (so error details are redacted when
+    /// `redact_error_details` is set and the configurable failure threshold —
+    /// not a hardcoded constant — drives circuit breaking). A successful send
+    /// clears the failure count so transient blips never accumulate toward the
+    /// breaker.
+    pub fn send_with_timeout(&mut self, policy: &SidecarRuntimePolicy, simulated_latency: Duration) -> Result<(), String> {
+        if simulated_latency > policy.operation_timeout {
+            self.record_failure_with_policy(
+                policy,
+                format!(
+                    "sidecar send timeout after {:?} (budget {:?})",
+                    simulated_latency, policy.operation_timeout
+                ),
+            );
             return Err(self
                 .last_error
                 .clone()
-                .unwrap_or_else(|| "sidecar timeout without recorded error".to_string()));
+                .unwrap_or_else(|| "sidecar operation failed".to_string()));
         }
         self.healthy = true;
         self.last_error = None;
+        // Success resets the circuit-breaker accounting: a healthy send must not
+        // leave stale failures that could trip the breaker on the next blip.
+        self.failure_count = 0;
+        self.degraded_to_builtin = false;
         Ok(())
     }
 
@@ -404,12 +417,42 @@ mod tests {
     }
 
     #[test]
-    fn sidecar_runtime_send_timeout_records_last_error() {
+    fn sidecar_runtime_send_timeout_redacts_error_and_uses_policy_threshold() {
         let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+        // redact_error_details defaults to true; threshold 2 is honored, not the
+        // hardcoded DEFAULT_FAILURE_THRESHOLD.
+        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_millis(50), 2);
 
-        let result = runtime.send_with_timeout(Duration::from_millis(50), Duration::from_millis(75));
+        let result = runtime.send_with_timeout(&policy, Duration::from_millis(75));
 
         assert!(result.is_err());
-        assert_eq!(runtime.last_error.as_deref(), Some("sidecar send timeout after 75ms (budget 50ms)"));
+        // The raw budget/latency detail must not leak; it is redacted.
+        assert_eq!(runtime.last_error.as_deref(), Some("sidecar operation failed"));
+        assert_eq!(runtime.failure_count, 1);
+        assert!(!runtime.degraded_to_builtin);
+
+        // Second timeout hits the configured threshold and degrades.
+        let _ = runtime.send_with_timeout(&policy, Duration::from_millis(75));
+        assert_eq!(runtime.failure_count, 2);
+        assert!(runtime.degraded_to_builtin);
+    }
+
+    #[test]
+    fn sidecar_runtime_successful_send_clears_failure_count() {
+        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+        let policy = SidecarRuntimePolicy::verified_external(16, Duration::from_millis(50), 3);
+
+        // Accumulate a failure, then a successful send must reset the breaker
+        // accounting so a later single failure does not immediately degrade.
+        let _ = runtime.send_with_timeout(&policy, Duration::from_millis(75));
+        assert_eq!(runtime.failure_count, 1);
+
+        runtime
+            .send_with_timeout(&policy, Duration::from_millis(10))
+            .expect("a within-budget send should succeed");
+        assert_eq!(runtime.failure_count, 0);
+        assert!(!runtime.degraded_to_builtin);
+        assert!(runtime.healthy);
+        assert!(runtime.last_error.is_none());
     }
 }

crates/targets/src/runtime/tls/coordinator.rs

@@ -28,11 +28,19 @@ use super::validate::validate_tls_material;
 use crate::error::TargetError;
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
 use tokio::task::JoinHandle;
 use tracing::{debug, info, warn};
 
+/// Minimum positive poll interval. A zero interval would panic inside
+/// `tokio::time::interval`, silently killing the poll loop.
+const MIN_RELOAD_INTERVAL: Duration = Duration::from_secs(1);
+
+/// Bound on how long registration waits to join a replaced poll loop before
+/// detaching it, so a stuck old loop cannot block a re-registration forever.
+const REPLACE_JOIN_TIMEOUT: Duration = Duration::from_secs(5);
+
 struct TargetReloadEntry {
     #[expect(dead_code)]
     target_label: String,
@@ -78,10 +86,14 @@ impl TargetTlsReloadCoordinator {
         let inputs = target.tls_input_set();
         let target_label = inputs.target_label.clone();
 
-        // Build initial material
-        let initial_material = Arc::new(target.build_tls_material().await?);
+        // Compute the fingerprint BEFORE building material. If a cert rotation
+        // races registration, this ordering guarantees the stored fingerprint is
+        // never *newer* than the published material, so the next poll observes a
+        // fingerprint change and rebuilds (self-healing). The reverse ordering
+        // pinned the old cert permanently (TOCTOU).
         let initial_fingerprint =
             build_target_tls_fingerprint(&inputs.ca_path, &inputs.client_cert_path, &inputs.client_key_path).await?;
+        let initial_material = Arc::new(target.build_tls_material().await?);
 
         let initial_state = Arc::new(TargetTlsPublishedState {
             generation: TargetTlsGeneration(1),
@@ -90,25 +102,40 @@ impl TargetTlsReloadCoordinator {
             loaded_at_unix_ms: unix_time_ms(),
         });
 
-        let runtime_state = Arc::new(TargetTlsRuntimeState::new(initial_state.clone(), inputs));
+        let runtime_state = Arc::new(TargetTlsRuntimeState::new(initial_state, inputs));
 
-        if options.detect_mode == ReloadDetectMode::Poll || options.detect_mode == ReloadDetectMode::Hybrid {
-            let (cancel_tx, cancel_rx) = tokio::sync::mpsc::channel(1);
-            let poll_handle = tokio::spawn(spawn_target_poll_loop(target, Arc::clone(&runtime_state), options, cancel_rx));
+        // A detection loop always runs when reload is enabled. Watch mode used to
+        // return success without any loop, silently disabling hot-reload; it now
+        // falls back to interval-based polling so detection is never a no-op.
+        let mut entries = self.entries.write().await;
 
-            let mut entries = self.entries.write().await;
-            entries.insert(
-                target_label.clone(),
-                TargetReloadEntry {
-                    target_label: target_label.clone(),
-                    cancel_tx,
-                    poll_handle,
-                },
-            );
-
-            info!(target = %target_label, "Registered target for TLS reload coordinator");
+        // Stop-before-start: if a loop is already registered under this label,
+        // cancel and join it *before* publishing the replacement so two loops
+        // never race on the same target's TLS material (issue #970).
+        if let Some(previous) = entries.remove(&target_label) {
+            let _ = previous.cancel_tx.send(()).await;
+            if tokio::time::timeout(REPLACE_JOIN_TIMEOUT, previous.poll_handle)
+                .await
+                .is_err()
+            {
+                warn!(target = %target_label, "Timed out joining previous TLS reload loop; detaching");
+            }
+            info!(target = %target_label, "Replaced existing TLS reload loop (stop-before-start)");
         }
 
+        let detect_mode = options.detect_mode;
+        let (cancel_tx, cancel_rx) = tokio::sync::mpsc::channel(1);
+        let poll_handle = tokio::spawn(spawn_target_poll_loop(target, Arc::clone(&runtime_state), options, cancel_rx));
+        entries.insert(
+            target_label.clone(),
+            TargetReloadEntry {
+                target_label: target_label.clone(),
+                cancel_tx,
+                poll_handle,
+            },
+        );
+        info!(target = %target_label, detect_mode = ?detect_mode, "Registered target for TLS reload coordinator");
+
         Ok(runtime_state)
     }
 
@@ -186,13 +213,16 @@ async fn spawn_target_poll_loop<T: ReloadableTargetTls>(
     options: TlsReloadOptions,
     mut cancel_rx: tokio::sync::mpsc::Receiver<()>,
 ) {
-    let mut interval = tokio::time::interval(options.interval);
+    // Normalize a zero interval to a safe minimum: `tokio::time::interval(0)`
+    // panics, which would silently kill this spawned loop.
+    let interval_period = effective_reload_interval(options.interval);
+    let mut interval = tokio::time::interval(interval_period);
     interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
     interval.tick().await; // skip the immediate first tick
 
     let label = &runtime_state.inputs.target_label;
     let debounce = options.debounce;
-    debug!(target = %label, interval_secs = options.interval.as_secs(), "TLS reload poll loop started");
+    debug!(target = %label, interval_secs = interval_period.as_secs(), "TLS reload poll loop started");
 
     loop {
         tokio::select! {
@@ -230,6 +260,11 @@ async fn reload_target_once<T: ReloadableTargetTls>(
     runtime_state: &TargetTlsRuntimeState<T::Material>,
     options: &TlsReloadOptions,
 ) -> Result<TargetTlsGeneration, TargetError> {
+    // Serialize reload cycles for this target so a force_reload and a poll-loop
+    // tick cannot interleave and publish a stale material or duplicate a
+    // generation.
+    let _reload_guard = runtime_state.reload_lock.lock().await;
+
     let now = unix_time_ms();
     runtime_state.mark_attempt(now);
     let started_at = std::time::Instant::now();
@@ -238,7 +273,17 @@ async fn reload_target_once<T: ReloadableTargetTls>(
     // 1. Read TLS files and compute fingerprint
     let inputs = &runtime_state.inputs;
     let next_fingerprint =
-        build_target_tls_fingerprint(&inputs.ca_path, &inputs.client_cert_path, &inputs.client_key_path).await?;
+        match build_target_tls_fingerprint(&inputs.ca_path, &inputs.client_cert_path, &inputs.client_key_path).await {
+            Ok(fingerprint) => fingerprint,
+            Err(err) => {
+                // The first step must not fail silently: record the error and a
+                // failure metric so the observability surface does not show
+                // "all healthy" while reload is actually broken.
+                *runtime_state.last_error.write() = Some(err.to_string());
+                record_target_tls_publication_fail(label);
+                return Err(err);
+            }
+        };
 
     // 2. Compare with current — skip if unchanged
     let current = runtime_state.current.load();
@@ -304,6 +349,12 @@ fn unix_time_ms() -> u64 {
     SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_millis() as u64
 }
 
+/// Normalizes a reload interval to a strictly positive duration. A zero
+/// interval would panic inside `tokio::time::interval`.
+fn effective_reload_interval(interval: Duration) -> Duration {
+    if interval.is_zero() { MIN_RELOAD_INTERVAL } else { interval }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -393,7 +444,7 @@ mod tests {
         let target = Arc::new(MockTarget::new("test:webhook"));
 
         let options = TlsReloadOptions {
-            detect_mode: ReloadDetectMode::Watch, // no poll loop for this test
+            detect_mode: ReloadDetectMode::Watch,
             ..default_options()
         };
         let state = coordinator.register(target.clone(), options).await.unwrap();
@@ -638,6 +689,81 @@ mod tests {
         assert!(coordinator.entries.read().await.is_empty());
     }
 
+    #[test]
+    fn effective_reload_interval_normalizes_zero() {
+        assert_eq!(effective_reload_interval(std::time::Duration::ZERO), MIN_RELOAD_INTERVAL);
+        let nonzero = std::time::Duration::from_secs(7);
+        assert_eq!(effective_reload_interval(nonzero), nonzero);
+    }
+
+    #[tokio::test]
+    async fn zero_interval_registration_does_not_panic() {
+        let coordinator = TargetTlsReloadCoordinator::new();
+        let target = Arc::new(MockTarget::new("test:zero-interval"));
+
+        let options = TlsReloadOptions {
+            interval: std::time::Duration::ZERO,
+            ..default_options()
+        };
+        // Registration spawns the poll loop; a zero interval must be normalized
+        // rather than panicking inside the spawned task.
+        let state = coordinator.register(target, options).await.unwrap();
+        assert_eq!(state.current.load().generation, TargetTlsGeneration(1));
+        assert_eq!(coordinator.entries.read().await.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn watch_mode_starts_detection_loop() {
+        let coordinator = TargetTlsReloadCoordinator::new();
+        let target = Arc::new(MockTarget::new("test:watch"));
+
+        let options = TlsReloadOptions {
+            detect_mode: ReloadDetectMode::Watch,
+            ..default_options()
+        };
+        // Watch mode must not silently skip starting a detection loop.
+        let _state = coordinator.register(target, options).await.unwrap();
+        assert_eq!(coordinator.entries.read().await.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn duplicate_label_registration_replaces_previous_loop() {
+        let coordinator = TargetTlsReloadCoordinator::new();
+        let first = Arc::new(MockTarget::new("test:dup"));
+        let second = Arc::new(MockTarget::new("test:dup"));
+
+        let _s1 = coordinator.register(first, default_options()).await.unwrap();
+        assert_eq!(coordinator.entries.read().await.len(), 1);
+
+        // Re-registering the same label must stop-and-join the old loop, leaving
+        // exactly one active entry (no orphaned duplicate loop).
+        let _s2 = coordinator.register(second, default_options()).await.unwrap();
+        assert_eq!(coordinator.entries.read().await.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn first_step_fingerprint_failure_records_error_and_metric() {
+        let mut target = MockTarget::new("test:fp-fail");
+        // A non-empty CA path that does not exist forces the very first step
+        // (fingerprint read) to fail.
+        target.inputs.ca_path = "/nonexistent/rustfs-tls-test/ca-does-not-exist.pem".to_string();
+
+        let initial_state = Arc::new(TargetTlsPublishedState {
+            generation: TargetTlsGeneration(1),
+            fingerprint: TargetTlsFingerprint::default(),
+            material: Arc::new("initial".to_string()),
+            loaded_at_unix_ms: 0,
+        });
+        let runtime_state = Arc::new(TargetTlsRuntimeState::new(initial_state, target.tls_input_set()));
+
+        let result = reload_target_once(&target, &runtime_state, &default_options()).await;
+        assert!(result.is_err());
+        // The first-step failure must be visible, not silently swallowed.
+        assert!(runtime_state.last_error.read().is_some());
+        // Build must not have been reached.
+        assert_eq!(target.build_calls.load(Ordering::SeqCst), 0);
+    }
+
     #[tokio::test]
     async fn bump_generation_saturates_at_max() {
         let target = MockTarget::new("test:saturation");

crates/targets/src/runtime/tls/state.rs

@@ -60,6 +60,10 @@ pub struct TargetTlsRuntimeState<M> {
     pub last_error: parking_lot::RwLock<Option<String>>,
     /// The TLS file paths this state watches.
     pub inputs: TargetTlsInputSet,
+    /// Serializes reload cycles for this target. Without it a `force_reload`
+    /// and a poll-loop tick could interleave, producing duplicate generations
+    /// or publishing an older material over a newer one.
+    pub reload_lock: tokio::sync::Mutex<()>,
 }
 
 impl<M> TargetTlsRuntimeState<M> {
@@ -72,6 +76,7 @@ impl<M> TargetTlsRuntimeState<M> {
             last_success_unix_ms: AtomicU64::new(0),
             last_error: parking_lot::RwLock::new(None),
             inputs,
+            reload_lock: tokio::sync::Mutex::new(()),
         }
     }
 

crates/targets/src/runtime/tls/validate.rs

@@ -19,8 +19,8 @@ use crate::error::TargetError;
 use rustfs_tls_runtime::{load_certs, load_private_key};
 
 /// Validates that a client certificate and private key file can be loaded
-/// and paired together. Returns `Ok(())` if both files parse successfully,
-/// or `Ok(())` if both paths are empty (no mTLS configured).
+/// *and that the key actually corresponds to the certificate's public key*.
+/// Returns `Ok(())` if both paths are empty (no mTLS configured).
 pub fn validate_cert_key_pairing(cert_path: &str, key_path: &str) -> Result<(), TargetError> {
     if cert_path.is_empty() && key_path.is_empty() {
         return Ok(());
@@ -32,11 +32,28 @@ pub fn validate_cert_key_pairing(cert_path: &str, key_path: &str) -> Result<(),
         ));
     }
 
-    load_certs(cert_path).map_err(|e| TargetError::Configuration(format!("Invalid client certificate '{cert_path}': {e}")))?;
+    let certs = load_certs(cert_path)
+        .map_err(|e| TargetError::Configuration(format!("Invalid client certificate '{cert_path}': {e}")))?;
+    let key =
+        load_private_key(key_path).map_err(|e| TargetError::Configuration(format!("Invalid client key '{key_path}': {e}")))?;
 
-    load_private_key(key_path).map_err(|e| TargetError::Configuration(format!("Invalid client key '{key_path}': {e}")))?;
-
-    Ok(())
+    // Parsing both files is not enough: an operator can supply a cert and a key
+    // that belong to different key pairs. Verify the private key's public key
+    // matches the certificate's SubjectPublicKeyInfo before accepting them.
+    let signing_key = rustls::crypto::aws_lc_rs::sign::any_supported_type(&key)
+        .map_err(|e| TargetError::Configuration(format!("Unsupported client private key '{key_path}': {e:?}")))?;
+    let certified = rustls::sign::CertifiedKey::new(certs, signing_key);
+    match certified.keys_match() {
+        Ok(()) => Ok(()),
+        // `Unknown` means the signing key cannot expose its public key for
+        // comparison (exotic key type). Both files parsed, so we do not reject a
+        // possibly-valid pair on an inconclusive comparison; only a definite
+        // mismatch is an error.
+        Err(rustls::Error::InconsistentKeys(rustls::InconsistentKeys::Unknown)) => Ok(()),
+        Err(e) => Err(TargetError::Configuration(format!(
+            "Client certificate '{cert_path}' and key '{key_path}' do not match: {e}"
+        ))),
+    }
 }
 
 /// Validates that a CA certificate file can be loaded. Returns `Ok(())`
@@ -56,3 +73,64 @@ pub fn validate_tls_material(ca_path: &str, cert_path: &str, key_path: &str) ->
     validate_ca_file(ca_path)?;
     validate_cert_key_pairing(cert_path, key_path)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::validate_cert_key_pairing;
+    use std::path::Path;
+    use tempfile::TempDir;
+
+    struct Pair {
+        cert_pem: String,
+        key_pem: String,
+    }
+
+    fn generate_pair() -> Pair {
+        let rcgen::CertifiedKey { cert, signing_key } =
+            rcgen::generate_simple_self_signed(vec!["example.com".to_string()]).expect("cert should generate");
+        Pair {
+            cert_pem: cert.pem(),
+            key_pem: signing_key.serialize_pem(),
+        }
+    }
+
+    fn write(dir: &Path, name: &str, contents: &str) -> String {
+        let path = dir.join(name);
+        std::fs::write(&path, contents).expect("write pem");
+        path.to_string_lossy().into_owned()
+    }
+
+    #[test]
+    fn empty_paths_are_ok() {
+        assert!(validate_cert_key_pairing("", "").is_ok());
+    }
+
+    #[test]
+    fn one_empty_path_is_rejected() {
+        assert!(validate_cert_key_pairing("/some/cert.pem", "").is_err());
+        assert!(validate_cert_key_pairing("", "/some/key.pem").is_err());
+    }
+
+    #[test]
+    fn matching_cert_and_key_pass() {
+        let dir = TempDir::new().expect("tempdir");
+        let pair = generate_pair();
+        let cert_path = write(dir.path(), "cert.pem", &pair.cert_pem);
+        let key_path = write(dir.path(), "key.pem", &pair.key_pem);
+
+        validate_cert_key_pairing(&cert_path, &key_path).expect("matching cert/key must validate");
+    }
+
+    #[test]
+    fn mismatched_cert_and_key_are_rejected() {
+        let dir = TempDir::new().expect("tempdir");
+        let a = generate_pair();
+        let b = generate_pair();
+        // Cert from pair A, key from pair B — a genuine mismatch.
+        let cert_path = write(dir.path(), "cert.pem", &a.cert_pem);
+        let key_path = write(dir.path(), "key.pem", &b.key_pem);
+
+        let err = validate_cert_key_pairing(&cert_path, &key_path).expect_err("mismatched cert/key must be rejected");
+        assert!(err.to_string().contains("do not match"), "unexpected error: {err}");
+    }
+}