gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-01 11:02:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(kms): persist deletion deadlines and run a restartable deletion worker

Browse Source 
Persist the scheduled deletion deadline in the Local (StoredMasterKey /
MasterKeyInfo) and Vault KV2 (VaultKeyData) records as an optional,
serde(default) field so old records keep deserializing, surface it from
describe_key, and clear it on cancel. Previously both backends lost the
deadline on restart and KV2 describe always reported none.

Add KmsBackend::remove_expired_key: a race-safe removal primitive that
re-checks state and deadline under the backend's own synchronization
(per-key write lock on Local), writes a tombstone (Deleted/Unavailable
record) before destroying material so a crashed removal can simply be
re-run, treats an already-removed key as success, and never auto-removes
pending records without a persisted deadline. The user-facing force
delete paths now write the same tombstone first.

Add a background deletion worker owned by each KmsServiceManager
service version: spawned at publish time when the backend supports
deletion scheduling, cancelled on stop/restart/reconfigure via a
CancellationToken, and sweeping every 60s for expired pending deletions
and leftover tombstones. Sweeps are stateless and idempotent, so they
are restart-safe and safe to run on every node concurrently without
leader election. Before removing anything the worker refuses keys that
are still the configured default key or that an injectable
DeletionReferenceChecker reports as referenced.

Refs rustfs/backlog#1571 (part of rustfs/backlog#1562)
This commit is contained in:
overtrue
2026-07-31 01:45:17 +08:00
parent fb867cd9f0
commit 4ecb2e47e7
9 changed files with 895 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/kms/src/backends/local.rs
+143 -6
View File
@@ -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<Zoned>,
created_by: Option<String>,
/// 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<Zoned>,
/// 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<DescribeKeyResponse> {
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<ExpiredKeyRemoval> {
// 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);
}
}
crates/kms/src/backends/mod.rs
+42
View File
@@ -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<ExpiredKeyRemoval> {
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");
crates/kms/src/backends/vault.rs
+100 -5
View File
@@ -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<String, String>,
/// Key tags
tags: HashMap<String, String>,
/// Scheduled deletion deadline; absent on records written before deadline
/// persistence landed, so it must stay optional for backward compatibility.
#[serde(default)]
deletion_date: Option<Zoned>,
/// 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<ExpiredKeyRemoval> {
// 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());
}
}
crates/kms/src/backends/vault_transit.rs
+59 -1
View File
@@ -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<ExpiredKeyRemoval> {
// 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)]
crates/kms/src/deletion_worker.rs
+399
View File
@@ -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<String>;
}
/// 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<String>,
/// Keys left in place because configuration still references them.
pub blocked: Vec<String>,
/// 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<dyn KmsBackend>,
default_key_id: Option<String>,
reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
interval: Duration,
}
impl DeletionWorker {
pub(crate) fn new(
backend: Arc<dyn KmsBackend>,
default_key_id: Option<String>,
reference_checker: Option<Arc<dyn DeletionReferenceChecker>>,
) -> 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<String> = 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<LocalKmsBackend> {
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<LocalKmsBackend>) -> 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<String>);
#[async_trait]
impl DeletionReferenceChecker for StaticReferences {
async fn references(&self, _key_id: &str) -> Vec<String> {
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");
}
}
crates/kms/src/lib.rs
+2
View File
@@ -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;
crates/kms/src/manager.rs
+6
View File
@@ -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<dyn KmsBackend> {
self.backend.clone()
}
}
#[cfg(test)]
crates/kms/src/service_manager.rs
+139 -1
View File
@@ -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<ObjectEncryptionService>,
/// The KMS manager instance
manager: Arc<KmsManager>,
/// Background deletion worker owned by this service version, if the
/// backend supports deletion scheduling
deletion_worker: Option<Arc<DeletionWorkerHandle>>,
}
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<Option<tokio::task::JoinHandle<()>>>,
}
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<Mutex<()>>,
/// External reference checker consulted before expired keys are removed
deletion_reference_checker: std::sync::RwLock<Option<Arc<dyn DeletionReferenceChecker>>>,
}
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<dyn DeletionReferenceChecker>) {
if let Ok(mut slot) = self.deletion_reference_checker.write() {
*slot = Some(checker);
}
}
fn deletion_reference_checker(&self) -> Option<Arc<dyn DeletionReferenceChecker>> {
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<Arc<DeletionWorkerHandle>> {
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;
crates/kms/src/types.rs
+5
View File
@@ -117,6 +117,9 @@ pub struct MasterKeyInfo {
pub rotated_at: Option<Zoned>,
/// Key creator/owner
pub created_by: Option<String>,
/// Scheduled deletion deadline while the key is pending deletion
#[serde(default)]
pub deletion_date: Option<Zoned>,
}
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,
}
}
}