test(kms): pin the deletion window and immediate-deletion gate

Cover the service gate over a local backend (refused force_immediate leaves
key material decryptable, confirmation mismatches are refused, a confirmed
immediate deletion still works once enabled, and the scheduled path is
unchanged), the backends' defensive window bound on Local, Vault KV2 and Vault
Transit, and the config gate defaulting off for configs persisted before the
field existed.

crates/kms/src/backends/local.rs

@@ -2620,6 +2620,41 @@ mod tests {
         assert!(fsync_recorder::dir_sync_count(dir) > dirs_before, "delete must fsync the key directory");
     }
 
+    /// KmsManager::delete_key is the enforcement point for the waiting window;
+    /// this pins the backend's defensive copy of the same bound, which is all
+    /// that stands between a direct backend caller and a one-day window.
+    #[tokio::test]
+    async fn delete_key_refuses_a_window_outside_the_supported_range() {
+        let (client, _temp_dir) = create_test_client().await;
+        let key_id = "window-bounds-key";
+        client.create_key(key_id, "AES_256", None).await.expect("create key");
+        let backend = LocalKmsBackend { client };
+
+        for days in [MIN_PENDING_DELETION_WINDOW_DAYS - 1, MAX_PENDING_DELETION_WINDOW_DAYS + 1] {
+            let result = backend
+                .delete_key(DeleteKeyRequest {
+                    key_id: key_id.to_string(),
+                    pending_window_in_days: Some(days),
+                    ..Default::default()
+                })
+                .await;
+            assert!(
+                matches!(result, Err(KmsError::InvalidOperation { .. })),
+                "a {days}-day window must be refused, got {result:?}"
+            );
+        }
+
+        let state = backend
+            .describe_key(DescribeKeyRequest {
+                key_id: key_id.to_string(),
+            })
+            .await
+            .expect("describe should succeed")
+            .key_metadata
+            .key_state;
+        assert_eq!(state, KeyState::Enabled, "a refused window must not schedule the key");
+    }
+
     #[tokio::test]
     async fn interrupted_update_commit_recovers_to_complete_old_or_new_state() {
         use durable_file::{CommitStep, failpoint};

crates/kms/src/backends/vault.rs

@@ -2696,6 +2696,45 @@ mod tests {
         );
     }
 
+    /// KmsManager::delete_key is the enforcement point for the waiting window;
+    /// this pins the backend's defensive copy of the same bound, which is all
+    /// that stands between a direct backend caller and a one-day window.
+    #[tokio::test]
+    async fn wired_schedule_deletion_refuses_a_window_outside_the_supported_range() {
+        for days in [MIN_PENDING_DELETION_WINDOW_DAYS - 1, MAX_PENDING_DELETION_WINDOW_DAYS + 1] {
+            let vault = ScriptedVault::serve(vec![
+                // describe_key: key info plus stored metadata.
+                ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+                ScriptedResponse::ok(kv2_read_data(&healthy_key_data())),
+            ])
+            .await;
+            let config = KmsConfig::vault(
+                url::Url::parse(&vault.address).expect("scripted vault address should parse"),
+                "scripted-token".to_string(),
+            )
+            .with_insecure_development_defaults();
+            let backend = VaultKmsBackend::new(config).await.expect("vault kv2 backend should build");
+
+            let result = backend
+                .delete_key(DeleteKeyRequest {
+                    key_id: "wired-key".to_string(),
+                    pending_window_in_days: Some(days),
+                    ..Default::default()
+                })
+                .await;
+            assert!(
+                matches!(result, Err(KmsError::InvalidOperation { .. })),
+                "a {days}-day window must be refused, got {result:?}"
+            );
+
+            let requests = vault.requests();
+            assert!(
+                !requests.iter().any(|line| line.starts_with("POST ")),
+                "a refused window must not write anything: {requests:?}"
+            );
+        }
+    }
+
     /// Conflict semantics are re-read *and* re-gate: when the re-read after a
     /// lost race shows the key was concurrently scheduled for deletion, the
     /// state gate rejects the retry instead of blindly re-applying it.

crates/kms/src/backends/vault_transit.rs

@@ -1767,6 +1767,48 @@ mod tests {
         assert_eq!(requests[6], "POST /v1/transit/keys/wired-key/rotate", "{requests:?}");
     }
 
+    /// KmsManager::delete_key is the enforcement point for the waiting window;
+    /// this pins the backend's defensive copy of the same bound, which is all
+    /// that stands between a direct backend caller and a one-day window.
+    #[tokio::test]
+    async fn wired_backend_delete_refuses_a_window_outside_the_supported_range() {
+        for days in [MIN_PENDING_DELETION_WINDOW_DAYS - 1, MAX_PENDING_DELETION_WINDOW_DAYS + 1] {
+            let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+            let vault = ScriptedVault::serve(vec![
+                // The state gate reads the transit key, then its metadata record.
+                ScriptedResponse::ok(transit_key_read_data("wired-key")),
+                ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ])
+            .await;
+            let config = KmsConfig::vault_transit(
+                url::Url::parse(&vault.address).expect("scripted vault address should parse"),
+                "scripted-token".to_string(),
+            )
+            .with_insecure_development_defaults();
+            let backend = VaultTransitKmsBackend::new(config)
+                .await
+                .expect("vault transit backend should build");
+
+            let result = backend
+                .delete_key(DeleteKeyRequest {
+                    key_id: "wired-key".to_string(),
+                    pending_window_in_days: Some(days),
+                    ..Default::default()
+                })
+                .await;
+            assert!(
+                matches!(result, Err(KmsError::InvalidOperation { .. })),
+                "a {days}-day window must be refused, got {result:?}"
+            );
+
+            let requests = vault.requests();
+            assert!(
+                !requests.iter().any(|line| line.starts_with("POST ")),
+                "a refused window must not write anything: {requests:?}"
+            );
+        }
+    }
+
     /// KV2 secret-metadata read payload (`kv2::read_metadata`) pinning the
     /// current secret version used as the check-and-set base.
     fn kv2_metadata_read_data(current_version: u64) -> serde_json::Value {

crates/kms/src/config.rs

@@ -1700,6 +1700,34 @@ 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.
+    #[test]
+    fn immediate_deletion_gate_defaults_off_and_reads_old_persisted_configs() {
+        let mut 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
+            .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");
+
+        with_vars(vec![(ENV_KMS_ALLOW_IMMEDIATE_DELETION, Some("true"))], || {
+            assert!(
+                allow_immediate_deletion_from_env(),
+                "the gate must be reachable from server configuration"
+            );
+        });
+        with_vars(vec![(ENV_KMS_ALLOW_IMMEDIATE_DELETION, None::<&str>)], || {
+            assert!(!allow_immediate_deletion_from_env());
+        });
+    }
+
     #[test]
     fn test_validate_rejects_incomplete_approle() {
         let mut config = KmsConfig::vault_approle(