fix(kms): enforce the deletion waiting window in the KMS service layer

Move the 7-30 day pending-deletion window check to KmsManager::delete_key so
every admin-facing deletion is validated once, before any backend runs, and
refuse `force_immediate` unless the server opts in through the new
`allow_immediate_deletion` config gate and the request echoes the key id back
in `confirm_key_id`.

The backends keep the same window bound as a defensive assertion for callers
that hold a backend handle directly, now expressed with the shared constants.

crates/kms/src/api_types.rs

@@ -495,6 +495,7 @@ impl ConfigureLocalKmsRequest {
                 file_permissions: self.file_permissions,
             }),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
+            allow_immediate_deletion: false,
             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),
@@ -532,6 +533,7 @@ impl ConfigureVaultKmsRequest {
                 },
             })),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
+            allow_immediate_deletion: false,
             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),
@@ -569,6 +571,7 @@ impl ConfigureVaultTransitKmsRequest {
                 },
             })),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
+            allow_immediate_deletion: false,
             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),
@@ -592,6 +595,7 @@ impl ConfigureStaticKmsRequest {
                 secret_key: self.secret_key.clone(),
             }),
             allow_insecure_dev_defaults: self.allow_insecure_dev_defaults.unwrap_or(false),
+            allow_immediate_deletion: false,
             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),
@@ -847,6 +851,7 @@ mod tests {
                 tls: None,
             })),
             allow_insecure_dev_defaults: true,
+            allow_immediate_deletion: false,
             timeout: Duration::from_secs(30),
             retry_attempts: 3,
             enable_cache: true,

crates/kms/src/backends/contract_tests.rs

@@ -108,6 +108,7 @@ fn schedule_request(key_id: &str) -> DeleteKeyRequest {
         key_id: key_id.to_string(),
         pending_window_in_days: Some(7),
         force_immediate: None,
+        confirm_key_id: None,
     }
 }
 

crates/kms/src/backends/local.rs

@@ -1584,9 +1584,15 @@ impl KmsBackend for LocalKmsBackend {
             // Schedule for deletion (default 30 days)
             ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::ScheduleDeletion)?;
 
-            let days = request.pending_window_in_days.unwrap_or(30);
-            if !(7..=30).contains(&days) {
-                return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30".to_string()));
+            // Defensive: KmsManager::delete_key is the enforcement point for the
+            // waiting window and rejects out-of-range requests before any
+            // backend runs. This repeats the bound for callers holding a backend
+            // handle directly (tests, maintenance tasks).
+            let days = request.pending_window_in_days.unwrap_or(DEFAULT_PENDING_DELETION_WINDOW_DAYS);
+            if !(MIN_PENDING_DELETION_WINDOW_DAYS..=MAX_PENDING_DELETION_WINDOW_DAYS).contains(&days) {
+                return Err(KmsError::invalid_parameter(format!(
+                    "pending_window_in_days must be between {MIN_PENDING_DELETION_WINDOW_DAYS} and {MAX_PENDING_DELETION_WINDOW_DAYS}"
+                )));
             }
 
             let deletion_date = Zoned::now() + Duration::from_secs(days as u64 * 86400);
@@ -2606,6 +2612,7 @@ mod tests {
                 key_id: "durable-key".to_string(),
                 pending_window_in_days: None,
                 force_immediate: Some(true),
+                confirm_key_id: None,
             })
             .await
             .expect("delete key");

crates/kms/src/backends/static_kms.rs

@@ -657,6 +657,7 @@ mod tests {
                 key_id: key_id.clone(),
                 pending_window_in_days: Some(7),
                 force_immediate: None,
+                confirm_key_id: None,
             },
         )
         .await;

crates/kms/src/backends/vault.rs

@@ -1364,11 +1364,15 @@ impl KmsBackend for VaultKmsBackend {
             // Schedule for deletion (default 30 days)
             ensure_key_state_permits(key_id, &key_metadata.key_state, StateGatedOperation::ScheduleDeletion)?;
 
-            let days = request.pending_window_in_days.unwrap_or(30);
-            if !(7..=30).contains(&days) {
-                return Err(crate::error::KmsError::invalid_parameter(
-                    "pending_window_in_days must be between 7 and 30".to_string(),
-                ));
+            // Defensive: KmsManager::delete_key is the enforcement point for the
+            // waiting window and rejects out-of-range requests before any
+            // backend runs. This repeats the bound for callers holding a backend
+            // handle directly (tests, maintenance tasks).
+            let days = request.pending_window_in_days.unwrap_or(DEFAULT_PENDING_DELETION_WINDOW_DAYS);
+            if !(MIN_PENDING_DELETION_WINDOW_DAYS..=MAX_PENDING_DELETION_WINDOW_DAYS).contains(&days) {
+                return Err(crate::error::KmsError::invalid_parameter(format!(
+                    "pending_window_in_days must be between {MIN_PENDING_DELETION_WINDOW_DAYS} and {MAX_PENDING_DELETION_WINDOW_DAYS}"
+                )));
             }
 
             let deletion_date = Zoned::now() + Duration::from_secs(days as u64 * 86400);
@@ -2198,6 +2202,7 @@ mod tests {
                 key_id: key_id.clone(),
                 pending_window_in_days: Some(7),
                 force_immediate: Some(false),
+                confirm_key_id: None,
             })
             .await
             .expect("schedule delete");
@@ -2701,6 +2706,7 @@ mod tests {
                 key_id: "wired-key".to_string(),
                 pending_window_in_days: Some(7),
                 force_immediate: Some(false),
+                confirm_key_id: None,
             })
             .await
             .expect("the schedule must retry past the lost race and commit");
@@ -2749,6 +2755,7 @@ mod tests {
                 key_id: "wired-key".to_string(),
                 pending_window_in_days: Some(7),
                 force_immediate: Some(false),
+                confirm_key_id: None,
             })
             .await
             .expect_err("the retry must re-run the state gate against the fresh record");

crates/kms/src/backends/vault_transit.rs

@@ -1174,9 +1174,15 @@ impl KmsBackend for VaultTransitKmsBackend {
         } else {
             ensure_key_state_permits(&key_id, &key_metadata.key_state, StateGatedOperation::ScheduleDeletion)?;
 
-            let days = request.pending_window_in_days.unwrap_or(30);
-            if !(7..=30).contains(&days) {
-                return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30"));
+            // Defensive: KmsManager::delete_key is the enforcement point for the
+            // waiting window and rejects out-of-range requests before any
+            // backend runs. This repeats the bound for callers holding a backend
+            // handle directly (tests, maintenance tasks).
+            let days = request.pending_window_in_days.unwrap_or(DEFAULT_PENDING_DELETION_WINDOW_DAYS);
+            if !(MIN_PENDING_DELETION_WINDOW_DAYS..=MAX_PENDING_DELETION_WINDOW_DAYS).contains(&days) {
+                return Err(KmsError::invalid_parameter(format!(
+                    "pending_window_in_days must be between {MIN_PENDING_DELETION_WINDOW_DAYS} and {MAX_PENDING_DELETION_WINDOW_DAYS}"
+                )));
             }
 
             let scheduled = Zoned::now() + Duration::from_secs(days as u64 * 86400);

crates/kms/src/config.rs

@@ -24,6 +24,7 @@ use std::time::Duration;
 use url::Url;
 
 pub const ENV_KMS_ALLOW_INSECURE_DEV_DEFAULTS: &str = "RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS";
+pub const ENV_KMS_ALLOW_IMMEDIATE_DELETION: &str = "RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION";
 pub const ENV_KMS_VAULT_SKIP_TLS_VERIFY: &str = "RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_PREFIX";
@@ -157,6 +158,19 @@ pub struct KmsConfig {
     /// Allow development-only insecure defaults such as plaintext local keys or HTTP Vault.
     #[serde(default)]
     pub allow_insecure_dev_defaults: bool,
+    /// Allow `DeleteKey` requests to skip the pending-deletion waiting window and
+    /// destroy key material right away.
+    ///
+    /// Off by default: an immediate deletion is unrecoverable and takes every
+    /// 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)]
+    pub allow_immediate_deletion: bool,
     /// Timeout for a single backend attempt.
     ///
     /// This bounds one outbound request, not the whole operation: the operation
@@ -180,6 +194,7 @@ impl Default for KmsConfig {
             default_key_id: None,
             backend_config: BackendConfig::default(),
             allow_insecure_dev_defaults: false,
+            allow_immediate_deletion: false,
             timeout: Duration::from_secs(30),
             retry_attempts: 3,
             enable_cache: true,
@@ -661,6 +676,12 @@ impl KmsConfig {
         self
     }
 
+    /// Explicitly allow deletions that bypass the pending-deletion waiting window.
+    pub fn with_immediate_deletion_allowed(mut self) -> Self {
+        self.allow_immediate_deletion = true;
+        self
+    }
+
     /// Set operation timeout
     pub fn with_timeout(mut self, timeout: Duration) -> Self {
         self.timeout = timeout;
@@ -854,6 +875,7 @@ impl KmsConfig {
         config.enable_cache = get_env_bool("RUSTFS_KMS_ENABLE_CACHE", config.enable_cache);
         config.allow_insecure_dev_defaults =
             get_env_bool(ENV_KMS_ALLOW_INSECURE_DEV_DEFAULTS, config.allow_insecure_dev_defaults);
+        config.allow_immediate_deletion = get_env_bool(ENV_KMS_ALLOW_IMMEDIATE_DELETION, config.allow_immediate_deletion);
 
         // Backend-specific configuration
         match config.backend {
@@ -961,6 +983,15 @@ impl KmsConfig {
     }
 }
 
+/// Read the immediate-deletion gate from the environment.
+///
+/// Callers that assemble a [`KmsConfig`] field by field instead of going
+/// through [`KmsConfig::from_env`] use this, so the gate keeps one name, one
+/// default, and one place to look it up.
+pub fn allow_immediate_deletion_from_env() -> bool {
+    get_env_bool(ENV_KMS_ALLOW_IMMEDIATE_DELETION, false)
+}
+
 fn vault_tls_config(skip_tls_verify: bool) -> Option<TlsConfig> {
     skip_tls_verify.then_some(TlsConfig {
         ca_cert_path: None,

crates/kms/src/deletion_worker.rs

@@ -376,6 +376,7 @@ mod tests {
                 key_id: key_id.to_string(),
                 pending_window_in_days: Some(7),
                 force_immediate: None,
+                confirm_key_id: None,
             })
             .await
             .expect("deletion should be scheduled");

crates/kms/src/manager.rs

@@ -17,16 +17,18 @@
 use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::KmsBackend;
 use crate::cache::{KmsCache, KmsCacheStats};
-use crate::config::KmsConfig;
-use crate::error::Result;
+use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, KmsConfig};
+use crate::error::{KmsError, Result};
 use crate::types::{
-    CancelKeyDeletionRequest, CancelKeyDeletionResponse, CreateKeyRequest, CreateKeyResponse, DecryptRequest, DecryptResponse,
-    DeleteKeyRequest, DeleteKeyResponse, DescribeKeyRequest, DescribeKeyResponse, EncryptRequest, EncryptResponse,
-    GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse, OperationContext,
+    CancelKeyDeletionRequest, CancelKeyDeletionResponse, CreateKeyRequest, CreateKeyResponse,
+    DEFAULT_PENDING_DELETION_WINDOW_DAYS, DecryptRequest, DecryptResponse, DeleteKeyRequest, DeleteKeyResponse,
+    DescribeKeyRequest, DescribeKeyResponse, EncryptRequest, EncryptResponse, GenerateDataKeyRequest, GenerateDataKeyResponse,
+    ListKeysRequest, ListKeysResponse, MAX_PENDING_DELETION_WINDOW_DAYS, MIN_PENDING_DELETION_WINDOW_DAYS, OperationContext,
 };
 use std::sync::Arc;
 use std::time::Instant;
 use tokio::sync::RwLock;
+use tracing::warn;
 
 /// KMS Manager coordinates operations between backends and caching
 #[derive(Clone)]
@@ -37,12 +39,18 @@ pub struct KmsManager {
     enable_cache: bool,
     backend_kind: &'static str,
     audit_sink: Option<Arc<dyn KmsAuditSink>>,
+    allow_immediate_deletion: bool,
 }
 
 impl KmsManager {
     /// Create a new KMS manager with the given backend and config
     pub fn new(backend: Arc<dyn KmsBackend>, config: KmsConfig) -> Self {
         let cache = Arc::new(RwLock::new(KmsCache::new(config.cache_config.max_keys as u64)));
+        if config.allow_immediate_deletion {
+            warn!(
+                "KMS immediate key deletion is enabled: a DeleteKey request may destroy key material without any waiting window, and every object encrypted under that key becomes permanently unreadable"
+            );
+        }
         Self {
             backend,
             cache,
@@ -50,6 +58,7 @@ impl KmsManager {
             enable_cache: config.enable_cache,
             backend_kind: config.backend.as_str(),
             audit_sink: None,
+            allow_immediate_deletion: config.allow_immediate_deletion,
         }
     }
 
@@ -224,7 +233,8 @@ impl KmsManager {
         Ok(())
     }
 
-    /// Delete a key
+    /// Delete a key, either scheduled behind the waiting window or — when the
+    /// server allows it — immediately.
     ///
     /// Audited as an internal operation; callers serving an authenticated
     /// request should use [`Self::delete_key_with_context`].
@@ -245,7 +255,13 @@ impl KmsManager {
         result
     }
 
+    /// This is the single enforcement point for the waiting window: every
+    /// admin-facing deletion goes through here, so the checks below run before
+    /// any backend sees the request. The backends repeat the window bound as a
+    /// defensive assertion for callers that hold a backend handle directly.
     async fn delete_key_inner(&self, request: DeleteKeyRequest) -> Result<DeleteKeyResponse> {
+        self.check_deletion_request(&request)?;
+
         let response = self.backend.delete_key(request).await?;
 
         // Remove from cache if enabled and key is being deleted
@@ -257,6 +273,44 @@ impl KmsManager {
         Ok(response)
     }
 
+    /// Gate a deletion request before it reaches the backend.
+    ///
+    /// Immediate deletion is unrecoverable, so it needs both a server-side
+    /// opt-in and a per-request confirmation that echoes the key id; without
+    /// either, the request is refused rather than downgraded to a scheduled
+    /// deletion, so a caller never believes a key is gone when it is not.
+    fn check_deletion_request(&self, request: &DeleteKeyRequest) -> Result<()> {
+        if !request.force_immediate.unwrap_or(false) {
+            let days = request.pending_window_in_days.unwrap_or(DEFAULT_PENDING_DELETION_WINDOW_DAYS);
+            if !(MIN_PENDING_DELETION_WINDOW_DAYS..=MAX_PENDING_DELETION_WINDOW_DAYS).contains(&days) {
+                return Err(KmsError::invalid_parameter(format!(
+                    "pending_window_in_days must be between {MIN_PENDING_DELETION_WINDOW_DAYS} and {MAX_PENDING_DELETION_WINDOW_DAYS}"
+                )));
+            }
+            return Ok(());
+        }
+
+        if !self.allow_immediate_deletion {
+            return Err(KmsError::invalid_operation(format!(
+                "immediate deletion of key {} is not allowed; schedule the deletion and wait out the pending window, or set {ENV_KMS_ALLOW_IMMEDIATE_DELETION}=true on the server",
+                request.key_id
+            )));
+        }
+
+        if request.confirm_key_id.as_deref() != Some(request.key_id.as_str()) {
+            return Err(KmsError::invalid_operation(format!(
+                "immediate deletion of key {} requires confirm_key_id to repeat the key id exactly",
+                request.key_id
+            )));
+        }
+
+        warn!(
+            key_id = %request.key_id,
+            "immediate KMS key deletion accepted; key material is destroyed without a waiting window and cannot be recovered"
+        );
+        Ok(())
+    }
+
     /// Cancel key deletion
     ///
     /// Audited as an internal operation; callers serving an authenticated
@@ -997,4 +1051,196 @@ mod tests {
             .await
             .expect("second data key should decrypt with its own context");
     }
+
+    /// Manager over a local backend, with the immediate-deletion gate set as
+    /// the server operator would set it.
+    async fn deletion_manager(temp_dir: &tempfile::TempDir, allow_immediate_deletion: bool) -> KmsManager {
+        let mut config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
+        config.allow_immediate_deletion = allow_immediate_deletion;
+        let backend = Arc::new(LocalKmsBackend::new(config.clone()).await.expect("Failed to create backend"));
+        KmsManager::new(backend, config)
+    }
+
+    async fn create_named_key(manager: &KmsManager, key_name: &str) -> String {
+        manager
+            .create_key(CreateKeyRequest {
+                key_name: Some(key_name.to_string()),
+                key_usage: KeyUsage::EncryptDecrypt,
+                ..Default::default()
+            })
+            .await
+            .expect("Failed to create key")
+            .key_id
+    }
+
+    /// Data key generated up front, decrypted again afterwards: a refused
+    /// deletion must leave the master key material byte-for-byte usable, not
+    /// merely leave a metadata record behind.
+    async fn data_key_probe(manager: &KmsManager, key_id: &str) -> (Vec<u8>, Vec<u8>) {
+        let generated = manager
+            .generate_data_key(GenerateDataKeyRequest {
+                key_id: key_id.to_string(),
+                key_spec: KeySpec::Aes256,
+                encryption_context: HashMap::new(),
+            })
+            .await
+            .expect("Failed to generate data key");
+        (generated.plaintext_key, generated.ciphertext_blob)
+    }
+
+    async fn assert_key_material_intact(manager: &KmsManager, key_id: &str, probe: &(Vec<u8>, Vec<u8>)) {
+        let state = manager
+            .describe_key(DescribeKeyRequest {
+                key_id: key_id.to_string(),
+            })
+            .await
+            .expect("a key that was not deleted must still be describable")
+            .key_metadata
+            .key_state;
+        assert_eq!(state, KeyState::Enabled, "a refused deletion must not change the key state");
+
+        let decrypted = manager
+            .decrypt(DecryptRequest {
+                ciphertext: probe.1.clone(),
+                encryption_context: HashMap::new(),
+                grant_tokens: Vec::new(),
+            })
+            .await
+            .expect("key material must still decrypt data keys issued before the refused deletion");
+        assert_eq!(decrypted.plaintext, probe.0, "decrypted data key must match the original plaintext");
+    }
+
+    #[tokio::test]
+    async fn immediate_deletion_is_refused_under_default_config() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let manager = deletion_manager(&temp_dir, false).await;
+        let key_id = create_named_key(&manager, "default-config-force-delete").await;
+        let probe = data_key_probe(&manager, &key_id).await;
+
+        // Confirmation present and correct: the server-side gate alone must
+        // refuse this, no matter how well-formed the request is.
+        let error = manager
+            .delete_key(DeleteKeyRequest {
+                key_id: key_id.clone(),
+                force_immediate: Some(true),
+                confirm_key_id: Some(key_id.clone()),
+                ..Default::default()
+            })
+            .await
+            .expect_err("immediate deletion must be refused unless the server allows it");
+        assert!(
+            matches!(error, KmsError::InvalidOperation { .. }),
+            "expected InvalidOperation, got {error:?}"
+        );
+
+        assert_key_material_intact(&manager, &key_id, &probe).await;
+    }
+
+    #[tokio::test]
+    async fn immediate_deletion_requires_a_matching_confirmation() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let manager = deletion_manager(&temp_dir, true).await;
+        let key_id = create_named_key(&manager, "confirmation-required").await;
+        let probe = data_key_probe(&manager, &key_id).await;
+
+        for confirmation in [None, Some(String::new()), Some(format!("{key_id}-typo"))] {
+            let result = manager
+                .delete_key(DeleteKeyRequest {
+                    key_id: key_id.clone(),
+                    force_immediate: Some(true),
+                    confirm_key_id: confirmation.clone(),
+                    ..Default::default()
+                })
+                .await;
+            assert!(
+                matches!(result, Err(KmsError::InvalidOperation { .. })),
+                "confirmation {confirmation:?} must be refused, got {result:?}"
+            );
+        }
+
+        assert_key_material_intact(&manager, &key_id, &probe).await;
+    }
+
+    #[tokio::test]
+    async fn immediate_deletion_succeeds_with_a_matching_confirmation() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let manager = deletion_manager(&temp_dir, true).await;
+        let key_id = create_named_key(&manager, "confirmed-force-delete").await;
+
+        manager
+            .delete_key(DeleteKeyRequest {
+                key_id: key_id.clone(),
+                force_immediate: Some(true),
+                confirm_key_id: Some(key_id.clone()),
+                ..Default::default()
+            })
+            .await
+            .expect("a confirmed immediate deletion must be allowed once the server enables it");
+
+        let error = manager
+            .describe_key(DescribeKeyRequest { key_id: key_id.clone() })
+            .await
+            .expect_err("an immediately deleted key must be gone");
+        assert!(matches!(error, KmsError::KeyNotFound { .. }), "expected KeyNotFound, got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn pending_window_outside_the_supported_range_is_refused() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let manager = deletion_manager(&temp_dir, false).await;
+        let key_id = create_named_key(&manager, "window-bounds").await;
+        let probe = data_key_probe(&manager, &key_id).await;
+
+        for days in [0, MIN_PENDING_DELETION_WINDOW_DAYS - 1, MAX_PENDING_DELETION_WINDOW_DAYS + 1] {
+            let result = manager
+                .delete_key(DeleteKeyRequest {
+                    key_id: key_id.clone(),
+                    pending_window_in_days: Some(days),
+                    ..Default::default()
+                })
+                .await;
+            assert!(
+                matches!(result, Err(KmsError::InvalidOperation { .. })),
+                "a {days}-day window must be refused, got {result:?}"
+            );
+        }
+
+        assert_key_material_intact(&manager, &key_id, &probe).await;
+    }
+
+    #[tokio::test]
+    async fn scheduled_deletion_keeps_its_existing_behaviour() {
+        let temp_dir = tempdir().expect("Failed to create temp dir");
+        let manager = deletion_manager(&temp_dir, false).await;
+
+        for (name, days) in [
+            ("schedule-default-window", None),
+            ("schedule-min-window", Some(MIN_PENDING_DELETION_WINDOW_DAYS)),
+            ("schedule-max-window", Some(MAX_PENDING_DELETION_WINDOW_DAYS)),
+        ] {
+            let key_id = create_named_key(&manager, name).await;
+            let response = manager
+                .delete_key(DeleteKeyRequest {
+                    key_id: key_id.clone(),
+                    pending_window_in_days: days,
+                    ..Default::default()
+                })
+                .await
+                .expect("scheduling a deletion inside the window must still succeed");
+            assert!(response.deletion_date.is_some(), "a scheduled deletion must report its deadline");
+            assert_eq!(response.key_metadata.key_state, KeyState::PendingDeletion);
+
+            manager
+                .cancel_key_deletion(CancelKeyDeletionRequest { key_id: key_id.clone() })
+                .await
+                .expect("a scheduled deletion must still be cancellable");
+            let state = manager
+                .describe_key(DescribeKeyRequest { key_id })
+                .await
+                .expect("describe should succeed")
+                .key_metadata
+                .key_state;
+            assert_eq!(state, KeyState::Enabled, "cancelling must restore the key");
+        }
+    }
 }

crates/kms/src/types.rs

@@ -893,15 +893,37 @@ impl std::str::FromStr for EncryptionAlgorithm {
     }
 }
 
+/// Shortest pending-deletion waiting window a caller may ask for, in days.
+///
+/// The window is enforced once, in [`crate::manager::KmsManager::delete_key`];
+/// backends keep the same bound as a defensive check for direct callers.
+pub const MIN_PENDING_DELETION_WINDOW_DAYS: u32 = 7;
+
+/// Longest pending-deletion waiting window a caller may ask for, in days.
+pub const MAX_PENDING_DELETION_WINDOW_DAYS: u32 = 30;
+
+/// Waiting window applied when a delete request does not name one.
+pub const DEFAULT_PENDING_DELETION_WINDOW_DAYS: u32 = MAX_PENDING_DELETION_WINDOW_DAYS;
+
 /// Request to delete a key
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct DeleteKeyRequest {
     /// Key ID to delete
     pub key_id: String,
-    /// Number of days to wait before deletion (7-30 days, optional)
+    /// Number of days to wait before deletion (7-30 days, optional, defaults to 30)
     pub pending_window_in_days: Option<u32>,
-    /// Force immediate deletion (for development/testing only)
+    /// Destroy the key material right away instead of scheduling it.
+    ///
+    /// Refused unless the server enables `allow_immediate_deletion`; see
+    /// [`crate::manager::KmsManager::delete_key`] for the gate.
     pub force_immediate: Option<bool>,
+    /// Key id echoed back by the caller to confirm an immediate deletion.
+    ///
+    /// Must equal `key_id` exactly whenever `force_immediate` is set. Optional
+    /// with a serde default because this type is part of the admin API
+    /// contract: clients that never ask for immediate deletion are unaffected.
+    #[serde(default)]
+    pub confirm_key_id: Option<String>,
 }
 
 /// Response from delete key operation

rustfs/src/admin/handlers/kms_keys.rs

@@ -1092,6 +1092,9 @@ impl Operation for DeleteKmsKeyHandler {
             key_id: request.key_id.clone(),
             pending_window_in_days: request.pending_window_in_days,
             force_immediate: request.force_immediate,
+            // The endpoint does not accept a confirmation yet, so an immediate
+            // deletion asked for here is always refused by the service gate.
+            confirm_key_id: None,
         };
 
         match manager.delete_key(kms_request).await {

rustfs/src/init.rs

@@ -291,6 +291,7 @@ fn build_local_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
             file_permissions: Some(0o600),
         }),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
+        allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
         timeout: std::time::Duration::from_secs(30),
         retry_attempts: 3,
@@ -328,6 +329,7 @@ fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
             tls: None,
         })),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
+        allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
         timeout: std::time::Duration::from_secs(30),
         retry_attempts: 3,
@@ -363,6 +365,7 @@ fn build_vault_transit_kms_config(cfg: &config::Config) -> std::io::Result<rustf
             ..rustfs_kms::config::VaultTransitConfig::default()
         })),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
+        allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
         timeout: std::time::Duration::from_secs(30),
         retry_attempts: 3,
@@ -417,6 +420,7 @@ fn build_static_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::
         default_key_id: cfg.kms_default_key_id.clone().or(Some(key_id)),
         backend_config: rustfs_kms::config::BackendConfig::Static(static_config),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
+        allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         ..Default::default()
     };