fix(kms): keep the immediate-deletion gate out of persisted config

The gate is per-server operator state, so it is read from server
configuration at every construction site and never serialized: a stored
config that claims it is not believed, and enabling it once on one node
cannot leak into the cluster-wide config every other node reloads.

Refs rustfs/backlog#1585

crates/kms/src/api_types.rs

@@ -16,8 +16,8 @@
 
 use crate::config::{
     BackendConfig, CacheConfig, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT, KmsBackend,
-    KmsConfig, LocalConfig, StaticConfig, TlsConfig, VaultAuthMethod, VaultConfig, VaultTransitConfig, redacted_secret,
-    redacted_secret_option,
+    KmsConfig, LocalConfig, StaticConfig, TlsConfig, VaultAuthMethod, VaultConfig, VaultTransitConfig,
+    allow_immediate_deletion_from_env, redacted_secret, redacted_secret_option,
 };
 use crate::service_manager::KmsServiceStatus;
 use crate::types::{KeyMetadata, KeyUsage};
@@ -495,7 +495,10 @@ impl ConfigureLocalKmsRequest {
                 file_permissions: self.file_permissions,
             }),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
-            allow_immediate_deletion: false,
+            // Read from server configuration, never from the request body: the
+            // gate must mean the same thing whether KMS was configured at
+            // startup or through this endpoint.
+            allow_immediate_deletion: allow_immediate_deletion_from_env(),
             timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
@@ -533,7 +536,10 @@ impl ConfigureVaultKmsRequest {
                 },
             })),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
-            allow_immediate_deletion: false,
+            // Read from server configuration, never from the request body: the
+            // gate must mean the same thing whether KMS was configured at
+            // startup or through this endpoint.
+            allow_immediate_deletion: allow_immediate_deletion_from_env(),
             timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
@@ -571,7 +577,10 @@ impl ConfigureVaultTransitKmsRequest {
                 },
             })),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
-            allow_immediate_deletion: false,
+            // Read from server configuration, never from the request body: the
+            // gate must mean the same thing whether KMS was configured at
+            // startup or through this endpoint.
+            allow_immediate_deletion: allow_immediate_deletion_from_env(),
             timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),
@@ -595,7 +604,10 @@ impl ConfigureStaticKmsRequest {
                 secret_key: self.secret_key.clone(),
             }),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
-            allow_immediate_deletion: false,
+            // Read from server configuration, never from the request body: the
+            // gate must mean the same thing whether KMS was configured at
+            // startup or through this endpoint.
+            allow_immediate_deletion: allow_immediate_deletion_from_env(),
             timeout: Duration::from_secs(self.timeout_seconds.unwrap_or(30)),
             retry_attempts: self.retry_attempts.unwrap_or(3),
             enable_cache: self.enable_cache.unwrap_or(true),

crates/kms/src/config.rs

@@ -165,11 +165,16 @@ pub struct KmsConfig {
     /// object encrypted under the key with it, so the waiting window (plus
     /// `CancelKeyDeletion`) is the only recovery path. Operators who genuinely
     /// need immediate deletion — throwaway test clusters, key material that was
-    /// never used — must turn this on in server configuration; the request must
-    /// still echo the key id back for confirmation. Deliberately not part of the
-    /// admin configure API: flipping the gate is an operator action, not
-    /// something a `kms:Configure` holder can do remotely.
-    #[serde(default)]
+    /// never used — must turn it on through server configuration
+    /// ([`ENV_KMS_ALLOW_IMMEDIATE_DELETION`]); the request must still echo the
+    /// key id back for confirmation.
+    ///
+    /// Not part of the serialized configuration, and not settable through the
+    /// admin configure API. It is per-server operator state that has to be
+    /// re-stated to survive a restart: persisting it would carry one operator's
+    /// one-time enablement into the cluster-wide config that every node reloads,
+    /// long after the deletion it was turned on for.
+    #[serde(skip)]
     pub allow_immediate_deletion: bool,
     /// Timeout for a single backend attempt.
     ///
@@ -1715,22 +1720,26 @@ mod tests {
         assert_eq!(refresh_safety_window_secs, None);
     }
 
-    /// The gate is off unless the operator turns it on, including for configs
-    /// persisted before the field existed: an absent value must never be read
-    /// as permission to skip the deletion waiting window.
+    /// The gate lives in server configuration only: it never rides along in a
+    /// serialized config, and a stored config that claims it must not be
+    /// believed. Otherwise one operator's one-time enablement would reach every
+    /// node that later reloads that config.
     #[test]
-    fn immediate_deletion_gate_defaults_off_and_reads_old_persisted_configs() {
-        let mut persisted =
+    fn immediate_deletion_gate_is_server_local_and_never_persisted() {
+        let persisted =
             serde_json::to_value(KmsConfig::default().with_immediate_deletion_allowed()).expect("kms config should serialize");
-        assert_eq!(persisted["allow_immediate_deletion"], serde_json::json!(true));
-        persisted
+        assert!(
+            persisted.get("allow_immediate_deletion").is_none(),
+            "the gate must not be written into a persisted config: {persisted}"
+        );
+
+        let mut forged = persisted;
+        forged
             .as_object_mut()
             .expect("a persisted config must be a JSON object")
-            .remove("allow_immediate_deletion")
-            .expect("current configs must carry the field");
-
-        let restored: KmsConfig = serde_json::from_value(persisted).expect("a config without the gate must still load");
-        assert!(!restored.allow_immediate_deletion, "an absent gate must fail closed");
+            .insert("allow_immediate_deletion".to_string(), serde_json::json!(true));
+        let restored: KmsConfig = serde_json::from_value(forged).expect("an unknown gate field must not break loading");
+        assert!(!restored.allow_immediate_deletion, "a stored gate must fail closed");
 
         with_vars(vec![(ENV_KMS_ALLOW_IMMEDIATE_DELETION, Some("true"))], || {
             assert!(

crates/kms/src/manager.rs

@@ -637,6 +637,7 @@ mod tests {
                         key_id: AUDITED_KEY_ID.to_string(),
                         pending_window_in_days: None,
                         force_immediate: None,
+                        confirm_key_id: None,
                     },
                     context,
                 )

rustfs/src/admin/handlers/kms_dynamic.rs

@@ -189,6 +189,9 @@ async fn save_kms_config(config: &KmsConfig) -> Result<(), String> {
 
 fn decode_persisted_kms_config(data: &[u8]) -> serde_json::Result<(KmsConfig, bool)> {
     let mut config: KmsConfig = serde_json::from_slice(data)?;
+    // The immediate-deletion gate is per-server operator state, never stored,
+    // so a config loaded from cluster storage still has to pick it up here.
+    config.allow_immediate_deletion = rustfs_kms::config::allow_immediate_deletion_from_env();
     let value: serde_json::Value = serde_json::from_slice(data)?;
     let is_missing_development_flag = value
         .as_object()