diff --git a/crates/kms/src/backends/local.rs b/crates/kms/src/backends/local.rs index 4cabef82c..3c48ff2c7 100644 --- a/crates/kms/src/backends/local.rs +++ b/crates/kms/src/backends/local.rs @@ -14,7 +14,9 @@ //! Local file-based KMS backend implementation -use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits}; +use crate::backends::{ + BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_status_permits, +}; use crate::config::KmsConfig; use crate::config::LocalConfig; use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material}; @@ -425,6 +427,10 @@ struct StoredMasterKey { #[serde(with = "crate::time_serde::option_zoned")] rotated_at: Option, created_by: Option, + /// Scheduled deletion deadline; absent on records written before deadline + /// persistence landed, so it must stay optional for backward compatibility. + #[serde(default, with = "crate::time_serde::option_zoned")] + deletion_date: Option, /// Encrypted key material (32 bytes encoded in base64 for AES-256) encrypted_key_material: String, /// Nonce used for encryption @@ -770,6 +776,7 @@ impl LocalKmsClient { created_at: stored_key.created_at, rotated_at: stored_key.rotated_at, created_by: stored_key.created_by, + deletion_date: stored_key.deletion_date, }) } @@ -843,6 +850,7 @@ impl LocalKmsClient { created_at: master_key.created_at.clone(), rotated_at: master_key.rotated_at.clone(), created_by: master_key.created_by.clone(), + deletion_date: master_key.deletion_date.clone(), encrypted_key_material, nonce, at_rest_protection, @@ -1141,7 +1149,7 @@ impl KmsClient for LocalKmsClient { async fn schedule_key_deletion( &self, key_id: &str, - _pending_window_days: u32, + pending_window_days: u32, _context: Option<&OperationContext>, ) -> Result<()> { debug!("Scheduling deletion for key: {}", key_id); @@ -1150,6 +1158,7 @@ impl KmsClient for LocalKmsClient { let mut master_key = self.load_master_key(key_id).await?; ensure_key_status_permits(key_id, &master_key.status, StateGatedOperation::ScheduleDeletion)?; master_key.status = KeyStatus::PendingDeletion; + master_key.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400)); // Preserve the existing key material (see enable_key): scheduling deletion must not // regenerate the master key, or cancelling the deletion later would recover a key that @@ -1170,6 +1179,7 @@ impl KmsClient for LocalKmsClient { return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion"))); } master_key.status = KeyStatus::Active; + master_key.deletion_date = None; // Preserve the existing key material (see enable_key): cancelling deletion must recover // the ORIGINAL key, not mint a new one that cannot decrypt existing data. @@ -1338,6 +1348,11 @@ impl KmsBackend for LocalKmsBackend { async fn describe_key(&self, request: DescribeKeyRequest) -> Result { let key_info = self.client.describe_key(&request.key_id, None).await?; + let deletion_date = if key_info.status == KeyStatus::PendingDeletion { + self.client.load_master_key(&request.key_id).await?.deletion_date + } else { + None + }; let metadata = KeyMetadata { key_id: key_info.key_id, @@ -1350,7 +1365,7 @@ impl KmsBackend for LocalKmsBackend { key_usage: key_info.usage, description: key_info.description, creation_date: key_info.created_at, - deletion_date: None, + deletion_date, origin: "KMS".to_string(), key_manager: "CUSTOMER".to_string(), tags: key_info.tags, @@ -1381,7 +1396,22 @@ impl KmsBackend for LocalKmsBackend { .map_err(|_| KmsError::key_not_found(format!("Key {key_id} not found")))?; let (deletion_date_str, deletion_date_dt) = if request.force_immediate.unwrap_or(false) { - // For immediate deletion, actually delete the key from filesystem + // Tombstone first: mark the record Deleted before removing the + // file, so a crash between the two steps leaves a key that is + // already unusable and whose removal can simply be re-run. + match self.client.decode_stored_key(key_id).await { + Ok((_stored, key_material)) => { + let mut tombstone = master_key.clone(); + tombstone.status = KeyStatus::Deleted; + tombstone.deletion_date = Some(Zoned::now()); + self.client.save_master_key(&tombstone, &key_material).await?; + } + Err(error) => { + // A record whose material can no longer be decoded cannot be + // re-encrypted into a tombstone; proceed with the removal. + warn!(key_id, %error, "skipping tombstone for undecodable key record"); + } + } let key_path = self.client.master_key_path(key_id)?; durable_file::remove_durably(key_path) .await @@ -1418,6 +1448,7 @@ impl KmsBackend for LocalKmsBackend { let deletion_date = Zoned::now() + Duration::from_secs(days as u64 * 86400); master_key.status = KeyStatus::PendingDeletion; + master_key.deletion_date = Some(deletion_date.clone()); (Some(deletion_date.to_string()), Some(deletion_date)) }; @@ -1471,6 +1502,7 @@ impl KmsBackend for LocalKmsBackend { // Cancel the deletion by resetting the state master_key.status = KeyStatus::Active; + master_key.deletion_date = None; // Save the updated key to disk - this is the missing critical step! // Preserve existing key material instead of generating new one @@ -1508,13 +1540,55 @@ impl KmsBackend for LocalKmsBackend { fn capabilities(&self) -> BackendCapabilities { // Rotation stays unadvertised until historical key versions can be // retained (see LocalKmsClient::rotate_key); without version history - // there is also no versioning capability. Deletion deadlines are not - // yet persisted across restarts, but scheduling itself is supported. + // there is also no versioning capability. BackendCapabilities::minimal() .with_enable_disable(true) .with_schedule_deletion(true) .with_physical_delete(true) } + + async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result { + // The per-key write lock serializes this against a concurrent + // cancellation, closing the check-then-remove race. + let _write_guard = self.client.lock_key_for_write(key_id).await; + + if !fs::try_exists(self.client.master_key_path(key_id)?).await? { + return Ok(ExpiredKeyRemoval::Removed); + } + let master_key = self.client.load_master_key(key_id).await?; + match master_key.status { + // Tombstone left by a crashed removal: complete it. + KeyStatus::Deleted => {} + KeyStatus::PendingDeletion => { + match &master_key.deletion_date { + Some(deadline) if deadline <= now => {} + // Not yet due, or a legacy record without a persisted + // deadline — never auto-remove those. + _ => return Ok(ExpiredKeyRemoval::NotExpired), + } + // Tombstone first (see delete_key): a crash between the state + // write and the file removal must leave an unusable record. + match self.client.decode_stored_key(key_id).await { + Ok((_stored, key_material)) => { + let mut tombstone = master_key.clone(); + tombstone.status = KeyStatus::Deleted; + tombstone.deletion_date = Some(now.clone()); + self.client.save_master_key(&tombstone, &key_material).await?; + } + Err(error) => { + warn!(key_id, %error, "skipping tombstone for undecodable key record"); + } + } + } + KeyStatus::Active | KeyStatus::Disabled => return Ok(ExpiredKeyRemoval::StateChanged), + } + + durable_file::remove_durably(self.client.master_key_path(key_id)?) + .await + .map_err(|e| KmsError::internal_error(format!("Failed to delete key file: {e}")))?; + debug!(key_id, "Local KMS expired key removed"); + Ok(ExpiredKeyRemoval::Removed) + } } #[cfg(test)] @@ -2653,4 +2727,67 @@ mod tests { "concurrent status updates must never lose or regenerate key material" ); } + + /// Records written before deadline persistence landed have no + /// deletion_date field and must keep deserializing (as None). + #[tokio::test] + async fn stored_master_key_without_deletion_date_still_deserializes() { + let (client, _temp_dir) = create_test_client().await; + client.create_key("legacy-key", "AES_256", None).await.expect("create key"); + + let path = client.master_key_path("legacy-key").expect("key path"); + let bytes = fs::read(&path).await.expect("read stored key"); + let mut value: serde_json::Value = serde_json::from_slice(&bytes).expect("stored key must be JSON"); + value + .as_object_mut() + .expect("stored key must be a JSON object") + .remove("deletion_date") + .expect("current records must carry the field"); + + let stored: StoredMasterKey = serde_json::from_value(value).expect("legacy record must deserialize"); + assert!(stored.deletion_date.is_none()); + } + + #[tokio::test] + async fn remove_expired_key_completes_a_tombstone_and_stays_idempotent() { + let temp_dir = TempDir::new().expect("temp dir"); + let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults(); + let backend = LocalKmsBackend::new(config).await.expect("backend"); + let created = backend + .create_key(CreateKeyRequest { + key_name: Some("tombstoned-key".to_string()), + key_usage: KeyUsage::EncryptDecrypt, + ..Default::default() + }) + .await + .expect("create key"); + let key_id = created.key_id; + + // Craft the state a removal crashed in: tombstone written, file not + // yet removed. + let client = backend.lifecycle_client(); + let (_stored, key_material) = client.decode_stored_key(&key_id).await.expect("decode stored key"); + let mut tombstone = client.load_master_key(&key_id).await.expect("load key"); + tombstone.status = KeyStatus::Deleted; + tombstone.deletion_date = Some(Zoned::now()); + client + .save_master_key(&tombstone, &key_material) + .await + .expect("write tombstone"); + + // The sweep primitive completes the crashed removal... + let outcome = backend + .remove_expired_key(&key_id, &Zoned::now()) + .await + .expect("tombstone completion"); + assert_eq!(outcome, crate::backends::ExpiredKeyRemoval::Removed); + assert!(!client.master_key_path(&key_id).expect("key path").exists()); + + // ...and stays idempotent once the key is gone. + let outcome = backend + .remove_expired_key(&key_id, &Zoned::now()) + .await + .expect("repeat removal"); + assert_eq!(outcome, crate::backends::ExpiredKeyRemoval::Removed); + } } diff --git a/crates/kms/src/backends/mod.rs b/crates/kms/src/backends/mod.rs index be78ab15c..ae2d531b9 100644 --- a/crates/kms/src/backends/mod.rs +++ b/crates/kms/src/backends/mod.rs @@ -17,6 +17,7 @@ use crate::error::{KmsError, Result}; use crate::types::*; use async_trait::async_trait; +use jiff::Zoned; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -265,6 +266,35 @@ pub trait KmsBackend: Send + Sync { fn capabilities(&self) -> BackendCapabilities { BackendCapabilities::minimal() } + + /// Remove a key whose scheduled deletion deadline has passed. + /// + /// Used by the background deletion worker. Implementations must re-check + /// state and deadline under their own write synchronization so that a + /// concurrent cancellation observed after the caller's inspection wins + /// ([`ExpiredKeyRemoval::StateChanged`]), must write a tombstone (a + /// `Deleted`/`Unavailable` record) before destroying material so a crashed + /// removal can simply be re-run, and must treat an already-removed key as + /// success so the operation stays idempotent across restarts and nodes. + /// + /// The default rejects the operation for backends without deletion + /// support. + async fn remove_expired_key(&self, _key_id: &str, _now: &Zoned) -> Result { + Err(KmsError::unsupported_capability("backend without deletion support", "remove_expired_key")) + } +} + +/// Outcome of [`KmsBackend::remove_expired_key`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ExpiredKeyRemoval { + /// The key's record and material were removed, or were already gone. + Removed, + /// The key is no longer pending deletion (for example the deletion was + /// cancelled after the caller inspected it); nothing was removed. + StateChanged, + /// The key is pending deletion but its deadline has not passed, or it has + /// no persisted deadline (legacy record) and is never auto-removed. + NotExpired, } /// Information about a KMS backend @@ -491,6 +521,18 @@ mod tests { assert!(!capabilities.physical_delete); } + #[tokio::test] + async fn default_remove_expired_key_is_unsupported() { + let error = MinimalBackend + .remove_expired_key("any-key", &jiff::Zoned::now()) + .await + .expect_err("backends without deletion support must reject expired-key removal"); + assert!( + matches!(error, KmsError::UnsupportedCapability { .. }), + "expected UnsupportedCapability, got {error:?}" + ); + } + #[tokio::test] async fn local_backend_capabilities_golden() { let temp_dir = tempfile::tempdir().expect("temp dir should be created"); diff --git a/crates/kms/src/backends/vault.rs b/crates/kms/src/backends/vault.rs index ac449d08e..d89dc5ed6 100644 --- a/crates/kms/src/backends/vault.rs +++ b/crates/kms/src/backends/vault.rs @@ -16,7 +16,7 @@ use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for}; use crate::backends::{ - BackendCapabilities, BackendInfo, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits, + BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits, ensure_key_status_permits, }; use crate::config::{KmsConfig, VaultConfig}; @@ -64,6 +64,10 @@ struct VaultKeyData { metadata: HashMap, /// Key tags tags: HashMap, + /// Scheduled deletion deadline; absent on records written before deadline + /// persistence landed, so it must stay optional for backward compatibility. + #[serde(default)] + deletion_date: Option, /// Encrypted key material (base64 encoded) encrypted_key_material: String, /// Version that pre-versioning envelopes (no `master_key_version`) resolve to. @@ -376,6 +380,7 @@ impl VaultKmsClient { description: request.description.clone(), metadata: existing_key_data.metadata.clone(), tags: request.tags.clone(), + deletion_date: existing_key_data.deletion_date.clone(), encrypted_key_material: existing_key_data.encrypted_key_material.clone(), // Preserve the key material baseline_version: existing_key_data.baseline_version, }; @@ -592,6 +597,7 @@ impl KmsClient for VaultKmsClient { description: None, metadata: HashMap::new(), tags: HashMap::new(), + deletion_date: None, encrypted_key_material: encrypted_material, baseline_version: None, }; @@ -610,6 +616,7 @@ impl KmsClient for VaultKmsClient { created_at: key_data.created_at, rotated_at: None, created_by: None, + deletion_date: None, }; debug!(key_id, "Vault KMS master key created"); @@ -700,7 +707,7 @@ impl KmsClient for VaultKmsClient { async fn schedule_key_deletion( &self, key_id: &str, - _pending_window_days: u32, + pending_window_days: u32, _context: Option<&OperationContext>, ) -> Result<()> { debug!("Scheduling key deletion: {}", key_id); @@ -708,6 +715,7 @@ impl KmsClient for VaultKmsClient { let mut key_data = self.get_key_data(key_id).await?; ensure_key_status_permits(key_id, &key_data.status, StateGatedOperation::ScheduleDeletion)?; key_data.status = KeyStatus::PendingDeletion; + key_data.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400)); self.store_key_data(key_id, &key_data).await?; debug!(key_id, "Vault KMS key deletion scheduled"); @@ -722,6 +730,7 @@ impl KmsClient for VaultKmsClient { return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion"))); } key_data.status = KeyStatus::Active; + key_data.deletion_date = None; self.store_key_data(key_id, &key_data).await?; debug!(key_id, "Vault KMS key deletion canceled"); @@ -823,6 +832,7 @@ impl KmsClient for VaultKmsClient { created_at: key_data.created_at.clone(), rotated_at: Some(Zoned::now()), created_by: None, + deletion_date: key_data.deletion_date.clone(), }) } @@ -909,6 +919,7 @@ impl VaultKmsBackend { KeyState::Unavailable => KeyStatus::Deleted, KeyState::PendingImport => KeyStatus::Disabled, // Treat as disabled until import completes }; + key_data.deletion_date = metadata.deletion_date.clone(); // Update the key data in Vault storage self.client.store_key_data(key_id, &key_data).await?; @@ -1008,7 +1019,7 @@ impl KmsBackend for VaultKmsBackend { key_usage: key_info.usage, description: key_info.description, creation_date: key_info.created_at, - deletion_date: None, + deletion_date: key_data.deletion_date.clone(), origin: "VAULT".to_string(), key_manager: "VAULT".to_string(), tags: key_data.tags, @@ -1037,8 +1048,17 @@ impl KmsBackend for VaultKmsBackend { }; let deletion_date = if request.force_immediate.unwrap_or(false) { - // Check if key is already in PendingDeletion state - if key_metadata.key_state == KeyState::PendingDeletion { + // Check if key is already in PendingDeletion state (or a tombstone + // left by a crashed removal, which may simply be completed) + if key_metadata.key_state == KeyState::PendingDeletion || key_metadata.key_state == KeyState::Unavailable { + // Tombstone first: mark the record Deleted before removing it, + // so a crash between the two steps leaves a key that is already + // unusable and whose removal can simply be re-run. + if key_metadata.key_state == KeyState::PendingDeletion { + let mut key_data = self.client.get_key_data(key_id).await?; + key_data.status = KeyStatus::Deleted; + self.client.store_key_data(key_id, &key_data).await?; + } // Force immediate deletion: physically delete the key from Vault storage self.client.delete_key(key_id).await?; @@ -1126,6 +1146,43 @@ impl KmsBackend for VaultKmsBackend { .with_schedule_deletion(true) .with_physical_delete(true) } + + async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result { + // Vault KV2 offers no compare-and-swap here, so a cancellation racing + // the read below can still lose; the window is a single read-write + // gap and the sweep re-reads on every pass. + let mut key_data = match self.client.get_key_data(key_id).await { + Ok(key_data) => key_data, + Err(KmsError::KeyNotFound { .. }) => return Ok(ExpiredKeyRemoval::Removed), + Err(error) => return Err(error), + }; + match key_data.status { + // Tombstone left by a crashed removal: complete it. + KeyStatus::Deleted => {} + KeyStatus::PendingDeletion => { + match &key_data.deletion_date { + Some(deadline) if deadline <= now => {} + // Not yet due, or a legacy record without a persisted + // deadline — never auto-remove those. + _ => return Ok(ExpiredKeyRemoval::NotExpired), + } + // Tombstone first: mark the record Deleted before removing it, + // so a crash between the two steps leaves a key that is + // already unusable and whose removal can simply be re-run. + key_data.status = KeyStatus::Deleted; + self.client.store_key_data(key_id, &key_data).await?; + } + KeyStatus::Active | KeyStatus::Disabled => return Ok(ExpiredKeyRemoval::StateChanged), + } + + match self.client.delete_key(key_id).await { + Ok(()) | Err(KmsError::KeyNotFound { .. }) => { + debug!(key_id, "Vault KV2 expired key removed"); + Ok(ExpiredKeyRemoval::Removed) + } + Err(error) => Err(error), + } + } } #[cfg(test)] @@ -1290,6 +1347,7 @@ mod tests { tags: HashMap::new(), encrypted_key_material: general_purpose::STANDARD.encode([0x42u8; 32]), baseline_version: Some(1), + deletion_date: None, }; let mut value = serde_json::to_value(&key_data).expect("serialize key data"); @@ -1648,4 +1706,41 @@ mod tests { "cancel_key_deletion must persist Active status to Vault, not only mutate the response" ); } + + /// The persisted KV2 record round-trips its deletion deadline, and records + /// written before the field existed keep deserializing (as None). A revert + /// of deadline persistence turns this test red. + #[test] + fn vault_key_data_deletion_date_round_trips_and_stays_backward_compatible() { + let deadline = Zoned::now() + Duration::from_secs(7 * 86400); + let key_data = VaultKeyData { + algorithm: "AES_256".to_string(), + usage: KeyUsage::EncryptDecrypt, + created_at: Zoned::now(), + status: KeyStatus::PendingDeletion, + version: 1, + description: None, + metadata: HashMap::new(), + tags: HashMap::new(), + deletion_date: Some(deadline.clone()), + encrypted_key_material: "material".to_string(), + baseline_version: None, + }; + + let mut value = serde_json::to_value(&key_data).expect("serialize"); + let restored: VaultKeyData = serde_json::from_value(value.clone()).expect("round trip"); + assert_eq!( + restored.deletion_date.as_ref().map(Zoned::timestamp), + Some(deadline.timestamp()), + "deletion deadline must survive the KV2 round trip" + ); + + value + .as_object_mut() + .expect("record must be a JSON object") + .remove("deletion_date") + .expect("current records must carry the field"); + let legacy: VaultKeyData = serde_json::from_value(value).expect("legacy record must deserialize"); + assert!(legacy.deletion_date.is_none()); + } } diff --git a/crates/kms/src/backends/vault_transit.rs b/crates/kms/src/backends/vault_transit.rs index de02667e1..b61a460e7 100644 --- a/crates/kms/src/backends/vault_transit.rs +++ b/crates/kms/src/backends/vault_transit.rs @@ -15,7 +15,9 @@ //! Vault Transit-based KMS backend. use crate::backends::vault_credentials::{VaultClientHandle, VaultConnectionSettings, VaultCredentialProvider, token_source_for}; -use crate::backends::{BackendCapabilities, BackendInfo, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits}; +use crate::backends::{ + BackendCapabilities, BackendInfo, ExpiredKeyRemoval, KmsBackend, KmsClient, StateGatedOperation, ensure_key_state_permits, +}; use crate::config::{KmsConfig, VaultTransitConfig}; use crate::encryption::{DataKeyEnvelope, generate_key_material}; use crate::error::{KmsError, Result}; @@ -479,6 +481,7 @@ impl KmsClient for VaultTransitKmsClient { created_at: metadata.created_at, rotated_at: None, created_by: metadata.created_by, + deletion_date: None, }) } @@ -585,6 +588,7 @@ impl KmsClient for VaultTransitKmsClient { created_at: metadata.created_at, rotated_at: Some(Zoned::now()), created_by: metadata.created_by, + deletion_date: None, }) } @@ -797,6 +801,60 @@ impl KmsBackend for VaultTransitKmsBackend { .with_versioning(true) .with_physical_delete(true) } + + async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result { + // The transit key's existence anchors "already removed": once it is + // gone only stale scheduling metadata can remain, so clean that up. + match self.client.read_transit_key(key_id).await { + Ok(_) => {} + Err(KmsError::KeyNotFound { .. }) => { + self.client.delete_key_metadata(key_id).await?; + return Ok(ExpiredKeyRemoval::Removed); + } + Err(error) => return Err(error), + } + + // A metadata read failure synthesizes an Enabled record (see + // TransitKeyMetadata::synthesized), which lands in StateChanged below: + // the worker never destroys material based on synthesized state. + let mut metadata = self.client.get_key_metadata(key_id).await?; + match metadata.key_state { + // Tombstone left by a crashed removal: complete it. + KeyState::Unavailable => {} + KeyState::PendingDeletion => { + match &metadata.deletion_date { + Some(deadline) if deadline <= now => {} + // Not yet due, or no persisted deadline — never auto-remove. + _ => return Ok(ExpiredKeyRemoval::NotExpired), + } + // Tombstone first: an Unavailable record is rejected by every + // state gate, and a crashed removal can simply be re-run. + metadata.key_state = KeyState::Unavailable; + self.client.store_key_metadata(key_id, &metadata).await?; + } + KeyState::Enabled | KeyState::Disabled | KeyState::PendingImport => { + return Ok(ExpiredKeyRemoval::StateChanged); + } + } + + if !self.client.read_transit_key(key_id).await?.deletion_allowed { + let mut update_builder = UpdateKeyConfigurationRequestBuilder::default(); + update_builder.deletion_allowed(true); + key::update( + &self.client.vault().client, + &self.client.config.mount_path, + key_id, + Some(&mut update_builder), + ) + .await + .map_err(|e| KmsError::backend_error(format!("Failed to allow deletion of Vault Transit key {key_id}: {e}")))?; + } + key::delete(&self.client.vault().client, &self.client.config.mount_path, key_id) + .await + .map_err(|e| KmsError::backend_error(format!("Failed to delete Vault Transit key {key_id}: {e}")))?; + self.client.delete_key_metadata(key_id).await?; + Ok(ExpiredKeyRemoval::Removed) + } } #[cfg(test)] diff --git a/crates/kms/src/deletion_worker.rs b/crates/kms/src/deletion_worker.rs new file mode 100644 index 000000000..fba80b121 --- /dev/null +++ b/crates/kms/src/deletion_worker.rs @@ -0,0 +1,399 @@ +// 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. + +//! Background worker that completes scheduled key deletions. +//! +//! Every sweep lists keys, picks the ones whose persisted deletion deadline +//! has passed (plus tombstones left by a crashed removal) and hands each to +//! [`KmsBackend::remove_expired_key`], which re-checks state under the +//! backend's own synchronization. The sweep is idempotent and keeps no state +//! of its own, so it is safe to re-run after a restart and safe to run on +//! every node of a deployment concurrently — a key is only ever removed while +//! its (re-read) record is an expired pending deletion or a tombstone. + +use crate::backends::{ExpiredKeyRemoval, KmsBackend}; +use crate::types::{KeyStatus, ListKeysRequest}; +use async_trait::async_trait; +use jiff::Zoned; +use std::sync::Arc; +use std::time::Duration; +use tokio_util::sync::CancellationToken; +use tracing::{debug, info, warn}; + +/// How often the worker looks for expired pending deletions. +pub const DEFAULT_SWEEP_INTERVAL: Duration = Duration::from_secs(60); + +/// Reports configuration that still references a KMS key. +/// +/// Consulted before any material is destroyed; a non-empty result blocks the +/// removal until the references disappear. Implementations live where the +/// referencing configuration lives (for example bucket encryption settings in +/// the server) and are injected via +/// [`crate::service_manager::KmsServiceManager::set_deletion_reference_checker`]. +#[async_trait] +pub trait DeletionReferenceChecker: Send + Sync { + /// Identifiers of configuration still referencing `key_id` (bucket names, + /// settings paths, ...). Errors must be reported as a reference so that + /// an unavailable checker never unblocks a deletion. + async fn references(&self, key_id: &str) -> Vec; +} + +/// Outcome of one sweep, for logging and tests. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct SweepReport { + /// Keys whose record and material were removed this sweep. + pub removed: Vec, + /// Keys left in place because configuration still references them. + pub blocked: Vec, + /// Keys that were pending but not yet due, without a persisted deadline, + /// or whose state changed between inspection and removal. + pub skipped: usize, + /// Keys whose removal attempt failed; retried on the next sweep. + pub failed: usize, +} + +pub(crate) struct DeletionWorker { + backend: Arc, + default_key_id: Option, + reference_checker: Option>, + interval: Duration, +} + +impl DeletionWorker { + pub(crate) fn new( + backend: Arc, + default_key_id: Option, + reference_checker: Option>, + ) -> Self { + Self { + backend, + default_key_id, + reference_checker, + interval: DEFAULT_SWEEP_INTERVAL, + } + } + + pub(crate) fn spawn(self, cancel: CancellationToken) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { self.run(cancel).await }) + } + + async fn run(self, cancel: CancellationToken) { + let mut ticker = tokio::time::interval(self.interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + _ = cancel.cancelled() => { + debug!("KMS deletion worker stopped"); + return; + } + _ = ticker.tick() => {} + } + let report = self.sweep(&Zoned::now()).await; + if !report.removed.is_empty() || !report.blocked.is_empty() || report.failed > 0 { + info!( + removed = ?report.removed, + blocked = ?report.blocked, + skipped = report.skipped, + failed = report.failed, + "KMS deletion sweep completed" + ); + } + } + } + + /// Run one sweep at the given time. Exposed separately so tests can drive + /// the expiry logic deterministically. + pub(crate) async fn sweep(&self, now: &Zoned) -> SweepReport { + let mut report = SweepReport::default(); + let mut marker: Option = None; + loop { + let request = ListKeysRequest { + limit: Some(100), + marker: marker.clone(), + usage_filter: None, + status_filter: None, + }; + let response = match self.backend.list_keys(request).await { + Ok(response) => response, + Err(error) => { + warn!(%error, "KMS deletion sweep could not list keys"); + report.failed += 1; + return report; + } + }; + for key in &response.keys { + if matches!(key.status, KeyStatus::PendingDeletion | KeyStatus::Deleted) { + self.process_key(&key.key_id, now, &mut report).await; + } + } + if !response.truncated { + break; + } + match response.next_marker { + Some(next_marker) => marker = Some(next_marker), + None => break, + } + } + report + } + + async fn process_key(&self, key_id: &str, now: &Zoned, report: &mut SweepReport) { + // Never remove a key that live configuration still points at. The + // default key check is built in; broader references (bucket + // encryption settings, ...) come from the injected checker. + if self.default_key_id.as_deref() == Some(key_id) { + warn!(key_id, "expired KMS key is still the default key; refusing removal"); + report.blocked.push(key_id.to_string()); + return; + } + if let Some(checker) = &self.reference_checker { + let references = checker.references(key_id).await; + if !references.is_empty() { + warn!(key_id, ?references, "expired KMS key is still referenced; refusing removal"); + report.blocked.push(key_id.to_string()); + return; + } + } + + // The backend re-checks state and deadline under its own write + // synchronization, so a cancellation racing this sweep wins there. + match self.backend.remove_expired_key(key_id, now).await { + Ok(ExpiredKeyRemoval::Removed) => report.removed.push(key_id.to_string()), + Ok(ExpiredKeyRemoval::StateChanged | ExpiredKeyRemoval::NotExpired) => report.skipped += 1, + Err(error) => { + warn!(key_id, %error, "failed to remove expired KMS key; will retry next sweep"); + report.failed += 1; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::backends::KmsClient as _; + use crate::backends::local::LocalKmsBackend; + use crate::config::KmsConfig; + use crate::error::KmsError; + use crate::types::{CreateKeyRequest, DeleteKeyRequest, DescribeKeyRequest, KeyState, KeyUsage}; + + async fn local_backend(temp_dir: &tempfile::TempDir) -> Arc { + let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults(); + Arc::new(LocalKmsBackend::new(config).await.expect("local backend should build")) + } + + async fn create_key(backend: &LocalKmsBackend, key_name: &str) -> String { + backend + .create_key(CreateKeyRequest { + key_name: Some(key_name.to_string()), + key_usage: KeyUsage::EncryptDecrypt, + ..Default::default() + }) + .await + .expect("key should be created") + .key_id + } + + async fn schedule(backend: &LocalKmsBackend, key_id: &str) { + backend + .delete_key(DeleteKeyRequest { + key_id: key_id.to_string(), + pending_window_in_days: Some(7), + force_immediate: None, + }) + .await + .expect("deletion should be scheduled"); + } + + fn worker(backend: Arc) -> DeletionWorker { + DeletionWorker::new(backend, None, None) + } + + fn after_window() -> Zoned { + Zoned::now() + Duration::from_secs(8 * 86400) + } + + async fn assert_key_gone(backend: &LocalKmsBackend, key_id: &str) { + let error = backend + .describe_key(DescribeKeyRequest { + key_id: key_id.to_string(), + }) + .await + .expect_err("removed key must not be describable"); + assert!(matches!(error, KmsError::KeyNotFound { .. }), "expected KeyNotFound, got {error:?}"); + } + + #[tokio::test] + async fn sweep_removes_expired_pending_key_and_is_idempotent() { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let key_id = create_key(&backend, "expired-key").await; + schedule(&backend, &key_id).await; + + let worker = worker(backend.clone()); + + // Not yet due: nothing happens. + let report = worker.sweep(&Zoned::now()).await; + assert!(report.removed.is_empty()); + assert_eq!(report.skipped, 1); + assert_eq!(report.failed, 0); + + // Past the deadline: the key is removed. + let report = worker.sweep(&after_window()).await; + assert_eq!(report.removed, vec![key_id.clone()]); + assert_eq!(report.failed, 0); + assert_key_gone(&backend, &key_id).await; + + // Re-running the sweep after the removal is a no-op. + let report = worker.sweep(&after_window()).await; + assert_eq!(report, SweepReport::default()); + } + + #[tokio::test] + async fn cancelled_deletion_always_beats_the_sweep() { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let cancelled = create_key(&backend, "cancelled-key").await; + let doomed = create_key(&backend, "doomed-key").await; + schedule(&backend, &cancelled).await; + schedule(&backend, &doomed).await; + + backend + .cancel_key_deletion(crate::types::CancelKeyDeletionRequest { + key_id: cancelled.clone(), + }) + .await + .expect("cancel should succeed"); + + let report = worker(backend.clone()).sweep(&after_window()).await; + assert_eq!(report.removed, vec![doomed.clone()]); + assert_eq!(report.failed, 0); + + // The cancelled key survives, enabled and usable. + let described = backend + .describe_key(DescribeKeyRequest { + key_id: cancelled.clone(), + }) + .await + .expect("cancelled key must still exist"); + assert_eq!(described.key_metadata.key_state, KeyState::Enabled); + assert_key_gone(&backend, &doomed).await; + } + + #[tokio::test] + async fn default_key_and_external_references_block_removal() { + struct StaticReferences(Vec); + + #[async_trait] + impl DeletionReferenceChecker for StaticReferences { + async fn references(&self, _key_id: &str) -> Vec { + self.0.clone() + } + } + + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let key_id = create_key(&backend, "referenced-key").await; + schedule(&backend, &key_id).await; + + // Blocked while it is the configured default key. + let as_default = DeletionWorker::new(backend.clone(), Some(key_id.clone()), None); + let report = as_default.sweep(&after_window()).await; + assert_eq!(report.blocked, vec![key_id.clone()]); + assert!(report.removed.is_empty()); + + // Blocked while external configuration still references it. + let with_references = DeletionWorker::new( + backend.clone(), + None, + Some(Arc::new(StaticReferences(vec!["bucket:sse-bucket".to_string()]))), + ); + let report = with_references.sweep(&after_window()).await; + assert_eq!(report.blocked, vec![key_id.clone()]); + assert!(report.removed.is_empty()); + backend + .describe_key(DescribeKeyRequest { key_id: key_id.clone() }) + .await + .expect("blocked key must still exist"); + + // Removed once nothing references it anymore. + let unreferenced = DeletionWorker::new(backend.clone(), None, Some(Arc::new(StaticReferences(Vec::new())))); + let report = unreferenced.sweep(&after_window()).await; + assert_eq!(report.removed, vec![key_id.clone()]); + } + + #[tokio::test] + async fn deadline_survives_backend_restart_and_sweep_completes_it() { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let key_id; + { + let backend = local_backend(&temp_dir).await; + key_id = create_key(&backend, "restart-key").await; + schedule(&backend, &key_id).await; + } + + // "Restart": a fresh backend over the same directory must still see + // the persisted deadline... + let backend = local_backend(&temp_dir).await; + let described = backend + .describe_key(DescribeKeyRequest { key_id: key_id.clone() }) + .await + .expect("key must survive the restart"); + assert_eq!(described.key_metadata.key_state, KeyState::PendingDeletion); + assert!( + described.key_metadata.deletion_date.is_some(), + "deletion deadline must survive a backend restart" + ); + + // ...and the worker completes the deletion without any new schedule call. + let report = worker(backend.clone()).sweep(&after_window()).await; + assert_eq!(report.removed, vec![key_id.clone()]); + assert_key_gone(&backend, &key_id).await; + } + + #[tokio::test(start_paused = true)] + async fn worker_loop_removes_due_keys_and_stops_on_cancel() { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let key_id = create_key(&backend, "loop-key").await; + // A zero-day window through the lifecycle client produces a deadline + // that is already due for the worker's wall-clock sweep. + backend + .lifecycle_client() + .schedule_key_deletion(&key_id, 0, None) + .await + .expect("schedule with zero window"); + + let cancel = CancellationToken::new(); + let task = worker(backend.clone()).spawn(cancel.clone()); + + // The paused clock auto-advances through the worker's interval ticks. + let mut removed = false; + for _ in 0..100 { + tokio::time::sleep(Duration::from_secs(1)).await; + if backend + .describe_key(DescribeKeyRequest { key_id: key_id.clone() }) + .await + .is_err() + { + removed = true; + break; + } + } + assert!(removed, "worker loop must remove the due key"); + + cancel.cancel(); + task.await.expect("worker task must stop after cancellation"); + } +} diff --git a/crates/kms/src/lib.rs b/crates/kms/src/lib.rs index f05d45e16..e88369669 100644 --- a/crates/kms/src/lib.rs +++ b/crates/kms/src/lib.rs @@ -69,6 +69,7 @@ pub mod backends; pub mod backup; mod cache; pub mod config; +pub mod deletion_worker; mod encryption; mod error; pub mod manager; @@ -89,6 +90,7 @@ pub use api_types::{ UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse, }; pub use config::*; +pub use deletion_worker::DeletionReferenceChecker; pub use encryption::is_data_key_envelope; pub use error::{KmsError, KmsUnavailableError, Result}; pub use manager::KmsManager; diff --git a/crates/kms/src/manager.rs b/crates/kms/src/manager.rs index c8bf9e115..8f670e1c7 100644 --- a/crates/kms/src/manager.rs +++ b/crates/kms/src/manager.rs @@ -164,6 +164,12 @@ impl KmsManager { pub fn backend_capabilities(&self) -> crate::backends::BackendCapabilities { self.backend.capabilities() } + + /// Direct handle to the configured backend, bypassing the metadata cache. + /// Used by background maintenance that must observe fresh state. + pub(crate) fn backend(&self) -> Arc { + self.backend.clone() + } } #[cfg(test)] diff --git a/crates/kms/src/service_manager.rs b/crates/kms/src/service_manager.rs index be463c42d..95547503c 100644 --- a/crates/kms/src/service_manager.rs +++ b/crates/kms/src/service_manager.rs @@ -16,6 +16,7 @@ use crate::backends::{KmsBackend, local::LocalKmsBackend}; use crate::config::{BackendConfig, KmsConfig}; +use crate::deletion_worker::{DeletionReferenceChecker, DeletionWorker}; use crate::error::{KmsError, Result}; use crate::manager::KmsManager; use crate::service::ObjectEncryptionService; @@ -28,6 +29,7 @@ use std::sync::{ }; use subtle::ConstantTimeEq; use tokio::sync::Mutex; +use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_KMS: &str = "kms"; @@ -110,6 +112,40 @@ struct ServiceVersion { service: Arc, /// The KMS manager instance manager: Arc, + /// Background deletion worker owned by this service version, if the + /// backend supports deletion scheduling + deletion_worker: Option>, +} + +impl ServiceVersion { + fn shutdown_deletion_worker(&self) { + if let Some(worker) = &self.deletion_worker { + worker.shutdown(); + } + } +} + +/// Cancellation handle for one service version's deletion worker. +struct DeletionWorkerHandle { + cancel: CancellationToken, + task: std::sync::Mutex>>, +} + +impl DeletionWorkerHandle { + fn shutdown(&self) { + self.cancel.cancel(); + if let Ok(mut task) = self.task.lock() { + // Detach: the task observes the cancelled token on its next poll. + drop(task.take()); + } + } +} + +impl Drop for DeletionWorkerHandle { + fn drop(&mut self) { + // Safety net for versions that are replaced without an explicit stop. + self.cancel.cancel(); + } } #[derive(Clone)] @@ -128,6 +164,8 @@ pub struct KmsServiceManager { /// Mutex to protect lifecycle operations (start, stop, reconfigure) /// This ensures only one lifecycle operation happens at a time lifecycle_mutex: Arc>, + /// External reference checker consulted before expired keys are removed + deletion_reference_checker: std::sync::RwLock>>, } impl KmsServiceManager { @@ -141,9 +179,23 @@ impl KmsServiceManager { }), version_counter: Arc::new(AtomicU64::new(0)), lifecycle_mutex: Arc::new(Mutex::new(())), + deletion_reference_checker: std::sync::RwLock::new(None), } } + /// Install the reference checker consulted before the deletion worker + /// removes an expired key. Takes effect for workers spawned by the next + /// start or reconfigure. + pub fn set_deletion_reference_checker(&self, checker: Arc) { + if let Ok(mut slot) = self.deletion_reference_checker.write() { + *slot = Some(checker); + } + } + + fn deletion_reference_checker(&self) -> Option> { + self.deletion_reference_checker.read().ok().and_then(|slot| slot.clone()) + } + /// Get current service status pub async fn get_status(&self) -> KmsServiceStatus { self.state.load().status.clone() @@ -321,6 +373,9 @@ impl KmsServiceManager { // Atomically clear current service version (lock-free, instant) // Note: Existing Arc references will keep the service alive until operations complete let state = self.state.load_full(); + if let Some(current) = state.current_service.as_ref() { + current.shutdown_deletion_worker(); + } self.state.store(Arc::new(RuntimeState { config: state.config.clone(), status: if state.config.is_some() { @@ -522,6 +577,7 @@ impl KmsServiceManager { version, service: encryption_service, manager: kms_manager, + deletion_worker: None, }) } @@ -533,7 +589,11 @@ impl KmsServiceManager { Ok(service_version) } - fn publish_running(&self, config: KmsConfig, service_version: ServiceVersion) { + fn publish_running(&self, config: KmsConfig, mut service_version: ServiceVersion) { + if let Some(previous) = self.state.load().current_service.as_ref() { + previous.shutdown_deletion_worker(); + } + service_version.deletion_worker = self.spawn_deletion_worker(&config, &service_version); self.state.store(Arc::new(RuntimeState { config: Some(config), status: KmsServiceStatus::Running, @@ -541,6 +601,24 @@ impl KmsServiceManager { })); } + /// Spawn the background deletion worker for a service version about to be + /// published, if its backend supports deletion scheduling. The worker is + /// only started at publish time so failed start/reconfigure candidates + /// never leak a running task. + fn spawn_deletion_worker(&self, config: &KmsConfig, service_version: &ServiceVersion) -> Option> { + let backend = service_version.manager.backend(); + if !backend.capabilities().schedule_deletion { + return None; + } + let cancel = CancellationToken::new(); + let worker = DeletionWorker::new(backend, config.default_key_id.clone(), self.deletion_reference_checker()); + let task = worker.spawn(cancel.clone()); + Some(Arc::new(DeletionWorkerHandle { + cancel, + task: std::sync::Mutex::new(Some(task)), + })) + } + fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) { let current = self.state.load_full(); if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) { @@ -824,6 +902,66 @@ mod tests { assert!(matches!(current.backend_config, BackendConfig::Local(_))); } + #[tokio::test] + async fn deletion_worker_follows_the_service_lifecycle() { + use tempfile::TempDir; + + let key_dir = TempDir::new().expect("create local KMS directory"); + let mut config = KmsConfig::local(key_dir.path().to_path_buf()); + config.allow_insecure_dev_defaults = true; + let manager = KmsServiceManager::new(); + manager.configure(config).await.expect("configure local KMS"); + manager.start().await.expect("start local KMS"); + + let first_worker = manager + .state + .load() + .current_service + .as_ref() + .expect("running service") + .deletion_worker + .clone() + .expect("local backend must run a deletion worker"); + assert!(!first_worker.cancel.is_cancelled()); + + // Replacing the service version replaces (and cancels) its worker. + manager.restart().await.expect("restart"); + assert!(first_worker.cancel.is_cancelled(), "replaced version's worker must be cancelled"); + let second_worker = manager + .state + .load() + .current_service + .as_ref() + .expect("running service") + .deletion_worker + .clone() + .expect("restarted service must run a fresh worker"); + assert!(!second_worker.cancel.is_cancelled()); + + // Stopping the service stops its worker. + manager.stop().await.expect("stop"); + assert!(second_worker.cancel.is_cancelled(), "stop must cancel the deletion worker"); + } + + #[tokio::test] + async fn static_backend_runs_no_deletion_worker() { + let manager = KmsServiceManager::new(); + manager.configure(static_config("key-a", 0x11)).await.expect("configure"); + manager.start().await.expect("start"); + + assert!( + manager + .state + .load() + .current_service + .as_ref() + .expect("running service") + .deletion_worker + .is_none(), + "a backend without deletion scheduling must not run a worker" + ); + } + #[tokio::test] async fn reconfigure_allows_safe_local_runtime_settings_only() { use tempfile::TempDir; diff --git a/crates/kms/src/types.rs b/crates/kms/src/types.rs index 39fda0ff4..f469b5936 100644 --- a/crates/kms/src/types.rs +++ b/crates/kms/src/types.rs @@ -117,6 +117,9 @@ pub struct MasterKeyInfo { pub rotated_at: Option, /// Key creator/owner pub created_by: Option, + /// Scheduled deletion deadline while the key is pending deletion + #[serde(default)] + pub deletion_date: Option, } impl MasterKeyInfo { @@ -142,6 +145,7 @@ impl MasterKeyInfo { created_at: Zoned::now(), rotated_at: None, created_by, + deletion_date: None, } } @@ -173,6 +177,7 @@ impl MasterKeyInfo { created_at: Zoned::now(), rotated_at: None, created_by, + deletion_date: None, } } }