From 732816ba9d67fdfaa2033e46e8f00f0d9d4dcf2b Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 1 Aug 2026 00:56:13 +0800 Subject: [PATCH] fix(kms): CAS Transit metadata writes and bound the metadata cache Transit KV metadata writes were whole-record overwrites with no precondition, so two nodes mutating the same key could silently clobber each other's lifecycle state, and the process-local metadata cache had neither a TTL nor a capacity bound, so a key disabled or scheduled for deletion on one node stayed usable on every other node until restart. - Replace write_metadata_to_kv with a versioned read (read_metadata_from_kv_versioned) plus a check-and-set write (cas_write_metadata_to_kv); mutate_key_metadata re-reads the authoritative record and re-runs the state gate on every attempt, and a lost CAS race retries with a fresh snapshot (bounded budget) instead of replaying the stale one. - Migrate every read-modify-write caller: enable, disable, schedule and cancel deletion, rotate version bump, the expired-key tombstone, and both create paths (create-only CAS that read-confirms the winner on a lost race). - Bound the metadata cache with moka (300s TTL, 1024 entries) and drop a key's entry when a transit data call reports it gone server-side. - Fail closed when the synthesized-metadata fallback cannot be read or persisted: the fabricated Enabled record is only served after a durable create-only CAS write, closing the gate weakening documented as a KNOWN RISK; the persistence fallback for pre-metadata keys is kept. Refs rustfs/backlog#1581 (part of rustfs/backlog#1562) --- crates/kms/src/backends/vault_transit.rs | 1107 +++++++++++++++++----- 1 file changed, 878 insertions(+), 229 deletions(-) diff --git a/crates/kms/src/backends/vault_transit.rs b/crates/kms/src/backends/vault_transit.rs index 2b01e8478..2c8db47d7 100644 --- a/crates/kms/src/backends/vault_transit.rs +++ b/crates/kms/src/backends/vault_transit.rs @@ -27,25 +27,53 @@ use crate::types::*; use async_trait::async_trait; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use jiff::Zoned; +use moka::future::Cache; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap}; use std::future::Future; use std::sync::Arc; use std::time::Duration; -use tokio::sync::RwLock; use tokio_util::sync::CancellationToken; use tracing::info; use vaultrs::{ + api::kv2::requests::SetSecretRequestOptions, api::transit::{ KeyType, requests::{ CreateKeyRequestBuilder, DecryptDataRequestBuilder, EncryptDataRequestBuilder, UpdateKeyConfigurationRequestBuilder, }, }, + error::ClientError, kv2, transit::{data, key}, }; +/// Attempt budget for metadata read-modify-write cycles: every check-and-set +/// conflict triggers a fresh read plus state-gate re-validation, never a blind +/// replay of the stale snapshot. +const METADATA_CAS_ATTEMPTS: usize = 3; + +/// TTL bound on cached metadata records. This caps how long one node can keep +/// acting on lifecycle state another node has since changed (disable, +/// schedule-deletion): the divergence window is one TTL instead of "until +/// process restart". Matches the manager-level `KmsCache` TTL. +const METADATA_CACHE_TTL: Duration = Duration::from_secs(300); + +/// Capacity bound on the metadata cache so an unbounded key namespace cannot +/// grow process memory without limit. +const METADATA_CACHE_CAPACITY: u64 = 1024; + +/// Whether a KV2 write failed its check-and-set precondition. +/// +/// Mirrors the helper of the same name in `vault.rs`; the two backends keep +/// separate copies because they share no private module. +fn is_cas_conflict(error: &ClientError) -> bool { + matches!( + error, + ClientError::APIError { code: 400, errors } if errors.iter().any(|message| message.contains("check-and-set")) + ) +} + #[derive(Debug, Clone)] struct TransitKeyMetadata { key_usage: KeyUsage, @@ -88,12 +116,16 @@ impl TransitKeyMetadata { } } - // KNOWN RISK (rustfs/backlog#1571, residual of rustfs/backlog#808): this - // fallback defaults to Enabled, so a key whose KV metadata read fails is - // treated as usable — a disabled or pending-deletion key can transiently - // "revive" on that path. State gates therefore only hold as strongly as - // metadata reads do. Changing the fallback is out of scope here; the - // synthesized_metadata_defaults_to_enabled test pins the current behavior. + // Fallback record for transit keys created before metadata persistence + // existed (rustfs#4256 / rustfs#4262): those keys have no KV record at + // all, and failing closed on the missing record would brick every one of + // them, so the record defaults to Enabled to match their pre-persistence + // behavior. The historical fail-open around it (rustfs/backlog#808, + // rustfs/backlog#1571: any metadata read failure yielded a usable Enabled + // key) is resolved for rustfs/backlog#1581: `get_key_metadata` only serves + // this record after durably persisting it with a create-only + // check-and-set, and any read or persist failure on that path fails + // closed. fn synthesized() -> Self { Self { key_usage: KeyUsage::EncryptDecrypt, @@ -148,7 +180,10 @@ pub struct VaultTransitKmsClient { metadata_kv_mount: String, /// Path prefix under metadata_kv_mount for storing transit key metadata records metadata_key_prefix: String, - metadata_cache: RwLock>, + /// Process-local metadata cache, TTL- and capacity-bounded (see + /// [`METADATA_CACHE_TTL`]): a lifecycle change made by another node + /// becomes visible here within one TTL window at the latest. + metadata_cache: Cache, /// Budgets wrapping every outbound Vault call (see `crate::policy`). retry: RetryPolicy, /// Cancellation point for the operation executor: aborts in-flight @@ -179,7 +214,10 @@ impl VaultTransitKmsClient { metadata_kv_mount: config.metadata_kv_mount.clone(), metadata_key_prefix: config.metadata_key_prefix.clone(), config, - metadata_cache: RwLock::new(HashMap::new()), + metadata_cache: Cache::builder() + .max_capacity(METADATA_CACHE_CAPACITY) + .time_to_live(METADATA_CACHE_TTL) + .build(), retry: RetryPolicy::from_config(kms_config), cancel: CancellationToken::new(), }) @@ -276,11 +314,7 @@ impl VaultTransitKmsClient { } data::encrypt(&vault.client, &self.config.mount_path, key_id, plaintext_b64, Some(&mut builder)) .await - .map_err(|e| { - AttemptError::from_vaultrs(e, |e| { - KmsError::backend_error(format!("Failed to encrypt data with Vault Transit key {key_id}: {e}")) - }) - }) + .map_err(|e| AttemptError::from_vaultrs(e, |e| Self::map_vault_error(key_id, e, "encrypt"))) }) .await?; @@ -305,11 +339,7 @@ impl VaultTransitKmsClient { } data::decrypt(&vault.client, &self.config.mount_path, key_id, ciphertext, Some(&mut builder)) .await - .map_err(|e| { - AttemptError::from_vaultrs(e, |e| { - KmsError::backend_error(format!("Failed to decrypt data with Vault Transit key {key_id}: {e}")) - }) - }) + .map_err(|e| AttemptError::from_vaultrs(e, |e| Self::map_vault_error(key_id, e, "decrypt"))) }) .await?; @@ -339,28 +369,87 @@ impl VaultTransitKmsClient { .await } - async fn write_metadata_to_kv(&self, key_id: &str, metadata: &TransitKeyMetadata) -> Result<()> { + /// Read the persisted metadata record together with the KV2 secret version + /// holding it, so a later write can be check-and-set against exactly this + /// snapshot. `None` means no record exists (a pre-persistence key). + async fn read_metadata_from_kv_versioned(&self, key_id: &str) -> Result> { + let path = self.metadata_key_path(key_id); + let path = path.as_str(); + + let kv_metadata = self + .run("vault_transit_read_metadata_version", OpClass::ReadIdempotent, move || async move { + let vault = self.vault().map_err(AttemptError::fatal)?; + match kv2::read_metadata(&vault.client, &self.metadata_kv_mount, path).await { + Ok(metadata) => Ok(Some(metadata)), + Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None), + Err(e) => Err(AttemptError::from_vaultrs(e, |e| { + KmsError::backend_error(format!("Failed to read transit key metadata version from Vault KV: {e}")) + })), + } + }) + .await?; + let Some(kv_metadata) = kv_metadata else { + return Ok(None); + }; + let cas = u32::try_from(kv_metadata.current_version) + .map_err(|_| KmsError::backend_error(format!("KV2 secret version for transit key {key_id} metadata exceeds u32")))?; + + // Read the exact secret version named by the metadata so the + // (cas, record) pair stays consistent even if another writer lands in + // between the two reads. + let secret_version = kv_metadata.current_version; + let record: Option = self + .run("vault_transit_read_metadata_at_version", OpClass::ReadIdempotent, move || async move { + let vault = self.vault().map_err(AttemptError::fatal)?; + match kv2::read_version(&vault.client, &self.metadata_kv_mount, path, secret_version).await { + Ok(persisted) => Ok(Some(persisted)), + Err(ClientError::ResponseWrapError) | Err(ClientError::APIError { code: 404, .. }) => Ok(None), + Err(e) => Err(AttemptError::from_vaultrs(e, |e| { + KmsError::backend_error(format!("Failed to read transit key metadata from Vault KV: {e}")) + })), + } + }) + .await?; + + Ok(record.map(|persisted| (cas, persisted.into()))) + } + + /// Check-and-set write of the metadata record. + /// + /// `cas` must match the KV2 secret version currently holding the record + /// (0 = create-only). Returns `Ok(false)` when the precondition failed — a + /// concurrent writer landed first — so the caller re-reads instead of + /// clobbering. Single attempt: replaying a lost-response write would + /// double-apply the mutation, and a CAS conflict is a normal concurrency + /// signal, not a backend failure. + async fn cas_write_metadata_to_kv(&self, key_id: &str, metadata: &TransitKeyMetadata, cas: u32) -> Result { let path = self.metadata_key_path(key_id); let path = path.as_str(); let persisted: TransitKeyMetadataPersisted = metadata.clone().into(); let persisted = &persisted; - // Single attempt: this is a whole-record overwrite without a CAS - // precondition, so a replay after a lost response could clobber a - // concurrent writer. - self.run("vault_transit_write_metadata", OpClass::MutatingNonIdempotent, move || async move { + self.run("vault_transit_cas_write_metadata", OpClass::MutatingNonIdempotent, move || async move { let vault = self.vault().map_err(AttemptError::fatal)?; - kv2::set(&vault.client, &self.metadata_kv_mount, path, persisted) + match kv2::set_with_options(&vault.client, &self.metadata_kv_mount, path, persisted, SetSecretRequestOptions { cas }) .await - .map(|_| ()) - .map_err(|e| { - AttemptError::from_vaultrs(e, |e| { - KmsError::backend_error(format!("Failed to write transit key metadata to Vault KV: {e}")) - }) - }) + { + Ok(_) => Ok(true), + Err(e) if is_cas_conflict(&e) => Ok(false), + Err(e) => Err(AttemptError::from_vaultrs(e, |e| { + KmsError::backend_error(format!("Failed to write transit key metadata to Vault KV: {e}")) + })), + } }) .await } + /// The error surfaced when a metadata read-modify-write exhausts its + /// [`METADATA_CAS_ATTEMPTS`] budget without winning a check-and-set write. + fn metadata_cas_conflict(key_id: &str) -> KmsError { + KmsError::invalid_operation(format!( + "Concurrent modification of transit key {key_id} metadata detected; retry the operation" + )) + } + async fn delete_metadata_from_kv(&self, key_id: &str) -> Result<()> { let path = self.metadata_key_path(key_id); let path = path.as_str(); @@ -414,45 +503,103 @@ impl VaultTransitKmsClient { } async fn get_key_metadata(&self, key_id: &str) -> Result { - // Check in-memory cache first. - if let Some(metadata) = self.metadata_cache.read().await.get(key_id).cloned() { + // Check in-memory cache first (TTL-bounded, so a stale entry can only + // survive one TTL window). + if let Some(metadata) = self.metadata_cache.get(key_id).await { return Ok(metadata); } - // On cache miss, try reading from the persistent KV store. - if let Some(persisted) = self.read_metadata_from_kv(key_id).await? { - self.metadata_cache - .write() - .await - .insert(key_id.to_string(), persisted.clone()); - return Ok(persisted); - } + for _ in 0..METADATA_CAS_ATTEMPTS { + // On cache miss, try reading from the persistent KV store. + if let Some(persisted) = self.read_metadata_from_kv(key_id).await? { + self.metadata_cache.insert(key_id.to_string(), persisted.clone()).await; + return Ok(persisted); + } - // Deliberate exemption from the "read paths never write" rule (rustfs#4256 / - // rustfs#4262): transit keys created before metadata persistence existed have no - // KV record at all, so failing closed here would brick every pre-existing transit - // key. The synthesised record only describes metadata — key material lives solely - // inside Vault's transit engine and is never generated or written by this path. - // - // Verify the transit key actually exists in Vault before synthesising. - self.read_transit_key(key_id).await?; - let metadata = TransitKeyMetadata::synthesized(); - // Persist the synthesised metadata so future cache misses pick it up (best - // effort: the KV write failing must not fail the read). - let _ = self.write_metadata_to_kv(key_id, &metadata).await; - self.metadata_cache.write().await.insert(key_id.to_string(), metadata.clone()); - Ok(metadata) + // Deliberate exemption from the "read paths never write" rule (rustfs#4256 / + // rustfs#4262): transit keys created before metadata persistence existed have no + // KV record at all, so failing closed here would brick every pre-existing transit + // key. The synthesised record only describes metadata — key material lives solely + // inside Vault's transit engine and is never generated or written by this path. + // + // Verify the transit key actually exists in Vault before synthesising. + self.read_transit_key(key_id).await?; + let metadata = TransitKeyMetadata::synthesized(); + // Fail closed on the persist (rustfs/backlog#1581): the synthesised record is + // only served once it is durable, so every node gates on the same stored + // state; a failed KV write must fail the read instead of minting a usable + // Enabled record out of thin air. The create-only check-and-set keeps two + // nodes from fabricating divergent records — losing that race loops back to + // re-read the winner's record. + if self.cas_write_metadata_to_kv(key_id, &metadata, 0).await? { + self.metadata_cache.insert(key_id.to_string(), metadata.clone()).await; + return Ok(metadata); + } + } + Err(Self::metadata_cas_conflict(key_id)) } - async fn store_key_metadata(&self, key_id: &str, metadata: &TransitKeyMetadata) -> Result<()> { - self.write_metadata_to_kv(key_id, metadata).await?; - self.metadata_cache.write().await.insert(key_id.to_string(), metadata.clone()); - Ok(()) + /// Create-only write of the metadata record (check-and-set of 0). + /// + /// Returns `Ok(false)` when a record already exists — a concurrent creator + /// won the race — and never overwrites it; the caller reconciles by + /// reading the stored record back. + async fn create_key_metadata(&self, key_id: &str, metadata: &TransitKeyMetadata) -> Result { + if self.cas_write_metadata_to_kv(key_id, metadata, 0).await? { + self.metadata_cache.insert(key_id.to_string(), metadata.clone()).await; + return Ok(true); + } + Ok(false) + } + + /// Read-modify-write of the persisted metadata record under KV2 + /// check-and-set. + /// + /// Every attempt re-reads the authoritative record, re-runs `apply` — + /// which owns state-gate validation — against that fresh snapshot, and + /// writes back with the snapshot's KV2 secret version as the check-and-set + /// precondition, so a concurrent writer is never clobbered blind. Losing + /// the race drops the (now stale) cache entry and retries with a fresh + /// read; exhausting the budget surfaces the conflict to the caller. + async fn mutate_key_metadata(&self, key_id: &str, mut apply: F) -> Result + where + F: FnMut(&mut TransitKeyMetadata) -> Result<()>, + { + for _ in 0..METADATA_CAS_ATTEMPTS { + let (cas, mut metadata) = match self.read_metadata_from_kv_versioned(key_id).await? { + Some(snapshot) => snapshot, + None => { + // Pre-persistence key without a KV record (see + // get_key_metadata): mutate the synthesised record and + // create it with a create-only check-and-set so two nodes + // cannot fabricate divergent records. + self.read_transit_key(key_id).await?; + (0, TransitKeyMetadata::synthesized()) + } + }; + apply(&mut metadata)?; + if self.cas_write_metadata_to_kv(key_id, &metadata, cas).await? { + self.metadata_cache.insert(key_id.to_string(), metadata.clone()).await; + return Ok(metadata); + } + self.metadata_cache.invalidate(key_id).await; + } + Err(Self::metadata_cas_conflict(key_id)) + } + + /// Drop the cached metadata record when a transit data-path call failed in + /// a way that signals the cached lifecycle state diverged from Vault (the + /// key is gone server-side), so the next state gate re-reads the + /// authoritative record instead of trusting the stale entry until its TTL. + async fn invalidate_metadata_on_state_error(&self, key_id: &str, error: &KmsError) { + if matches!(error, KmsError::KeyNotFound { .. }) { + self.metadata_cache.invalidate(key_id).await; + } } async fn delete_key_metadata(&self, key_id: &str) -> Result<()> { self.delete_metadata_from_kv(key_id).await?; - self.metadata_cache.write().await.remove(key_id); + self.metadata_cache.invalidate(key_id).await; Ok(()) } @@ -514,9 +661,16 @@ impl VaultTransitKmsClient { .await?; let plaintext_key = generate_key_material(&request.key_spec)?; - let encrypted_key = self + let encrypted_key = match self .transit_encrypt(&request.master_key_id, &plaintext_key, &request.encryption_context) - .await?; + .await + { + Ok(encrypted_key) => encrypted_key, + Err(error) => { + self.invalidate_metadata_on_state_error(&request.master_key_id, &error).await; + return Err(error); + } + }; let envelope = DataKeyEnvelope { key_id: uuid::Uuid::new_v4().to_string(), @@ -545,9 +699,16 @@ impl VaultTransitKmsClient { let metadata = self .ensure_key_state_allows(&request.key_id, StateGatedOperation::Encrypt) .await?; - let ciphertext = self + let ciphertext = match self .transit_encrypt(&request.key_id, &request.plaintext, &request.encryption_context) - .await?; + .await + { + Ok(ciphertext) => ciphertext, + Err(error) => { + self.invalidate_metadata_on_state_error(&request.key_id, &error).await; + return Err(error); + } + }; Ok(EncryptResponse { ciphertext: ciphertext.into_bytes(), @@ -575,8 +736,16 @@ impl VaultTransitKmsClient { let encrypted_key = std::str::from_utf8(&envelope.encrypted_key) .map_err(|e| KmsError::cryptographic_error("utf8", format!("Invalid Transit ciphertext: {e}")))?; - self.transit_decrypt(&envelope.master_key_id, encrypted_key, &envelope.encryption_context) + match self + .transit_decrypt(&envelope.master_key_id, encrypted_key, &envelope.encryption_context) .await + { + Ok(plaintext) => Ok(plaintext), + Err(error) => { + self.invalidate_metadata_on_state_error(&envelope.master_key_id, &error).await; + Err(error) + } + } } /// Test-only lifecycle driver: the product path goes through [`KmsBackend`]. @@ -598,59 +767,68 @@ impl VaultTransitKmsClient { // this create would have produced; report it as the create result. // Anything else keeps failing. A failed pre-check read must fail the // create rather than fall through to re-creating over an unknown key. - match self.read_transit_key(key_id).await { - Ok(_) => { - let existing = self.get_key_metadata(key_id).await?; - return if existing.key_state == KeyState::Enabled && existing.key_usage == KeyUsage::EncryptDecrypt { - info!( - key_id, - "Vault Transit create found an identical enabled key; treating it as a recovered create" - ); - Ok(MasterKeyInfo { - key_id: key_id.to_string(), - version: existing.current_version, - algorithm: algorithm.to_string(), - usage: existing.key_usage, - status: KeyStatus::Active, - description: existing.description, - metadata: existing.tags.clone(), - created_at: existing.created_at, - rotated_at: None, - created_by: existing.created_by, - deletion_date: None, - }) - } else { - Err(KmsError::key_already_exists(key_id)) - }; + // + // Two passes: losing the create-only metadata check-and-set race loops + // back here so the pre-check read-confirms the winning record. + for _ in 0..2 { + match self.read_transit_key(key_id).await { + Ok(_) => { + let existing = self.get_key_metadata(key_id).await?; + return if existing.key_state == KeyState::Enabled && existing.key_usage == KeyUsage::EncryptDecrypt { + info!( + key_id, + "Vault Transit create found an identical enabled key; treating it as a recovered create" + ); + Ok(MasterKeyInfo { + key_id: key_id.to_string(), + version: existing.current_version, + algorithm: algorithm.to_string(), + usage: existing.key_usage, + status: KeyStatus::Active, + description: existing.description, + metadata: existing.tags.clone(), + created_at: existing.created_at, + rotated_at: None, + created_by: existing.created_by, + deletion_date: None, + }) + } else { + Err(KmsError::key_already_exists(key_id)) + }; + } + Err(KmsError::KeyNotFound { .. }) => {} + Err(error) => return Err(error), } - Err(KmsError::KeyNotFound { .. }) => {} - Err(error) => return Err(error), + + self.create_transit_key(key_id).await?; + + let metadata = TransitKeyMetadata { + created_by: Some("vault-transit".to_string()), + ..TransitKeyMetadata::from_create_request(&CreateKeyRequest { + key_name: Some(key_id.to_string()), + ..Default::default() + }) + }; + if self.create_key_metadata(key_id, &metadata).await? { + return Ok(MasterKeyInfo { + key_id: key_id.to_string(), + version: metadata.current_version, + algorithm: algorithm.to_string(), + usage: metadata.key_usage, + status: KeyStatus::Active, + description: metadata.description, + metadata: metadata.tags, + created_at: metadata.created_at, + rotated_at: None, + created_by: metadata.created_by, + deletion_date: None, + }); + } + // A concurrent creator persisted metadata first; make sure the + // pre-check reads their record, not a stale cache entry. + self.metadata_cache.invalidate(key_id).await; } - - self.create_transit_key(key_id).await?; - - let metadata = TransitKeyMetadata { - created_by: Some("vault-transit".to_string()), - ..TransitKeyMetadata::from_create_request(&CreateKeyRequest { - key_name: Some(key_id.to_string()), - ..Default::default() - }) - }; - self.store_key_metadata(key_id, &metadata).await?; - - Ok(MasterKeyInfo { - key_id: key_id.to_string(), - version: metadata.current_version, - algorithm: algorithm.to_string(), - usage: metadata.key_usage, - status: KeyStatus::Active, - description: metadata.description, - metadata: metadata.tags, - created_at: metadata.created_at, - rotated_at: None, - created_by: metadata.created_by, - deletion_date: None, - }) + Err(KmsError::key_already_exists(key_id)) } /// Test-only lifecycle driver: the product path goes through [`KmsBackend`]. @@ -708,17 +886,26 @@ impl VaultTransitKmsClient { pub(crate) async fn enable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> { // A pending deletion must be reverted through cancel_key_deletion, not - // silently by enabling, so the gate rejects PendingDeletion here. - let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Enable).await?; - metadata.key_state = KeyState::Enabled; - metadata.deletion_date = None; - self.store_key_metadata(key_id, &metadata).await + // silently by enabling, so the gate rejects PendingDeletion here. The + // gate runs inside the check-and-set loop against every fresh snapshot. + self.mutate_key_metadata(key_id, |metadata| { + ensure_key_state_permits(key_id, &metadata.key_state, StateGatedOperation::Enable)?; + metadata.key_state = KeyState::Enabled; + metadata.deletion_date = None; + Ok(()) + }) + .await + .map(|_| ()) } pub(crate) async fn disable_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<()> { - let mut metadata = self.ensure_key_state_allows(key_id, StateGatedOperation::Disable).await?; - metadata.key_state = KeyState::Disabled; - self.store_key_metadata(key_id, &metadata).await + self.mutate_key_metadata(key_id, |metadata| { + ensure_key_state_permits(key_id, &metadata.key_state, StateGatedOperation::Disable)?; + metadata.key_state = KeyState::Disabled; + Ok(()) + }) + .await + .map(|_| ()) } /// Test-only lifecycle driver: the product path goes through [`KmsBackend`]. @@ -729,12 +916,15 @@ impl VaultTransitKmsClient { pending_window_days: u32, _context: Option<&OperationContext>, ) -> Result<()> { - let mut metadata = self - .ensure_key_state_allows(key_id, StateGatedOperation::ScheduleDeletion) - .await?; - metadata.key_state = KeyState::PendingDeletion; - metadata.deletion_date = Some(Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400)); - self.store_key_metadata(key_id, &metadata).await + let deletion_date = Zoned::now() + Duration::from_secs(pending_window_days as u64 * 86400); + self.mutate_key_metadata(key_id, |metadata| { + ensure_key_state_permits(key_id, &metadata.key_state, StateGatedOperation::ScheduleDeletion)?; + metadata.key_state = KeyState::PendingDeletion; + metadata.deletion_date = Some(deletion_date.clone()); + Ok(()) + }) + .await + .map(|_| ()) } pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result { @@ -755,9 +945,15 @@ impl VaultTransitKmsClient { }) .await?; - let mut metadata = self.get_key_metadata(key_id).await?; - metadata.current_version += 1; - self.store_key_metadata(key_id, &metadata).await?; + let metadata = self + .mutate_key_metadata(key_id, |metadata| { + // The transit rotation above has already happened; recording + // the version bump must not be blocked by a concurrent + // lifecycle transition, so no state gate here. + metadata.current_version += 1; + Ok(()) + }) + .await?; Ok(MasterKeyInfo { key_id: key_id.to_string(), @@ -788,6 +984,15 @@ impl VaultTransitKmsClient { } } +#[cfg(test)] +impl VaultTransitKmsClient { + /// Rebuild the metadata cache with test-controlled bounds so TTL and + /// capacity behavior can be exercised without real sleeps. + fn rebuild_metadata_cache_for_tests(&mut self, capacity: u64, ttl: Duration) { + self.metadata_cache = Cache::builder().max_capacity(capacity).time_to_live(ttl).build(); + } +} + pub struct VaultTransitKmsBackend { client: VaultTransitKmsClient, } @@ -835,59 +1040,68 @@ impl KmsBackend for VaultTransitKmsBackend { // what this request would have written, report it as the create // result; any divergence keeps failing so a create can never adopt or // reshape a key it would not have produced. - match self.client.read_transit_key(&key_id).await { - Ok(_) => { - let existing = self.client.get_key_metadata(&key_id).await?; - let requested = TransitKeyMetadata::from_create_request(&request); - return if existing.key_state == KeyState::Enabled - && existing.key_usage == requested.key_usage - && existing.description == requested.description - && existing.tags == requested.tags - { - info!( - key_id, - "Vault Transit create found an identical enabled key; treating it as a recovered create" - ); - Ok(CreateKeyResponse { - key_id: key_id.clone(), - key_metadata: KeyMetadata { + // + // Two passes: losing the create-only metadata check-and-set race loops + // back here so the pre-check read-confirms the winning record. + for _ in 0..2 { + match self.client.read_transit_key(&key_id).await { + Ok(_) => { + let existing = self.client.get_key_metadata(&key_id).await?; + let requested = TransitKeyMetadata::from_create_request(&request); + return if existing.key_state == KeyState::Enabled + && existing.key_usage == requested.key_usage + && existing.description == requested.description + && existing.tags == requested.tags + { + info!( key_id, - key_state: existing.key_state, - key_usage: existing.key_usage, - description: existing.description, - creation_date: existing.created_at, - deletion_date: existing.deletion_date, - origin: existing.origin, - key_manager: "VAULT_TRANSIT".to_string(), - tags: existing.tags, - }, - }) - } else { - Err(KmsError::key_already_exists(&key_id)) - }; + "Vault Transit create found an identical enabled key; treating it as a recovered create" + ); + Ok(CreateKeyResponse { + key_id: key_id.clone(), + key_metadata: KeyMetadata { + key_id, + key_state: existing.key_state, + key_usage: existing.key_usage, + description: existing.description, + creation_date: existing.created_at, + deletion_date: existing.deletion_date, + origin: existing.origin, + key_manager: "VAULT_TRANSIT".to_string(), + tags: existing.tags, + }, + }) + } else { + Err(KmsError::key_already_exists(&key_id)) + }; + } + Err(KmsError::KeyNotFound { .. }) => {} + Err(error) => return Err(error), } - Err(KmsError::KeyNotFound { .. }) => {} - Err(error) => return Err(error), + + self.client.create_transit_key(&key_id).await?; + let metadata = TransitKeyMetadata::from_create_request(&request); + if self.client.create_key_metadata(&key_id, &metadata).await? { + return Ok(CreateKeyResponse { + key_id: key_id.clone(), + key_metadata: KeyMetadata { + key_id, + key_state: metadata.key_state, + key_usage: metadata.key_usage, + description: metadata.description, + creation_date: metadata.created_at, + deletion_date: metadata.deletion_date, + origin: metadata.origin, + key_manager: "VAULT_TRANSIT".to_string(), + tags: metadata.tags, + }, + }); + } + // A concurrent creator persisted metadata first; make sure the + // pre-check reads their record, not a stale cache entry. + self.client.metadata_cache.invalidate(&key_id).await; } - - self.client.create_transit_key(&key_id).await?; - let metadata = TransitKeyMetadata::from_create_request(&request); - self.client.store_key_metadata(&key_id, &metadata).await?; - - Ok(CreateKeyResponse { - key_id: key_id.clone(), - key_metadata: KeyMetadata { - key_id, - key_state: metadata.key_state, - key_usage: metadata.key_usage, - description: metadata.description, - creation_date: metadata.created_at, - deletion_date: metadata.deletion_date, - origin: metadata.origin, - key_manager: "VAULT_TRANSIT".to_string(), - tags: metadata.tags, - }, - }) + Err(KmsError::key_already_exists(&key_id)) } async fn encrypt(&self, request: EncryptRequest) -> Result { @@ -946,10 +1160,14 @@ impl KmsBackend for VaultTransitKmsBackend { self.client.delete_key_metadata(&key_id).await?; None } else { - let mut metadata = self.client.get_key_metadata(&key_id).await?; - metadata.key_state = KeyState::PendingDeletion; - metadata.deletion_date = Some(Zoned::now()); - self.client.store_key_metadata(&key_id, &metadata).await?; + let now = Zoned::now(); + self.client + .mutate_key_metadata(&key_id, |metadata| { + metadata.key_state = KeyState::PendingDeletion; + metadata.deletion_date = Some(now.clone()); + Ok(()) + }) + .await?; key_metadata = self.client.key_metadata_response(&key_id).await?; None } @@ -961,11 +1179,17 @@ impl KmsBackend for VaultTransitKmsBackend { return Err(KmsError::invalid_parameter("pending_window_in_days must be between 7 and 30")); } - let mut metadata = self.client.get_key_metadata(&key_id).await?; let scheduled = Zoned::now() + Duration::from_secs(days as u64 * 86400); - metadata.key_state = KeyState::PendingDeletion; - metadata.deletion_date = Some(scheduled.clone()); - self.client.store_key_metadata(&key_id, &metadata).await?; + self.client + .mutate_key_metadata(&key_id, |metadata| { + // Re-run the gate against every fresh snapshot: the check + // above used a possibly cached record. + ensure_key_state_permits(&key_id, &metadata.key_state, StateGatedOperation::ScheduleDeletion)?; + metadata.key_state = KeyState::PendingDeletion; + metadata.deletion_date = Some(scheduled.clone()); + Ok(()) + }) + .await?; key_metadata = self.client.key_metadata_response(&key_id).await?; Some(scheduled.to_string()) }; @@ -978,14 +1202,20 @@ impl KmsBackend for VaultTransitKmsBackend { } async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result { - let mut metadata = self.client.get_key_metadata(&request.key_id).await?; - if metadata.key_state != KeyState::PendingDeletion { - return Err(KmsError::invalid_key_state(format!("Key {} is not pending deletion", request.key_id))); - } - - metadata.key_state = KeyState::Enabled; - metadata.deletion_date = None; - self.client.store_key_metadata(&request.key_id, &metadata).await?; + let key_id = request.key_id.as_str(); + self.client + .mutate_key_metadata(key_id, |metadata| { + // Re-checked against every fresh snapshot: a concurrent sweep + // that tombstoned the key must fail this cancel, not be + // overwritten blind. + if metadata.key_state != KeyState::PendingDeletion { + return Err(KmsError::invalid_key_state(format!("Key {key_id} is not pending deletion"))); + } + metadata.key_state = KeyState::Enabled; + metadata.deletion_date = None; + Ok(()) + }) + .await?; Ok(CancelKeyDeletionResponse { key_id: request.key_id.clone(), @@ -1033,27 +1263,51 @@ impl KmsBackend for VaultTransitKmsBackend { 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 => { + // Tombstone under check-and-set: every attempt re-reads the record and + // re-validates state and due-ness, so a cancel_key_deletion racing the + // sweep either lands before the tombstone (the re-read sees Enabled + // and the sweep backs off) or after it (the cancel's own + // check-and-set write fails). + let mut tombstoned = false; + for _ in 0..METADATA_CAS_ATTEMPTS { + let Some((cas, mut metadata)) = self.client.read_metadata_from_kv_versioned(key_id).await? else { + // No persisted lifecycle record (pre-persistence key): the + // worker never destroys material whose scheduling state was + // never recorded. return Ok(ExpiredKeyRemoval::StateChanged); + }; + match metadata.key_state { + // Tombstone left by a crashed removal: complete it. + KeyState::Unavailable => { + tombstoned = true; + } + 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; + if self.client.cas_write_metadata_to_kv(key_id, &metadata, cas).await? { + self.client.metadata_cache.insert(key_id.to_string(), metadata.clone()).await; + tombstoned = true; + } else { + // Lost the check-and-set race — most likely a + // concurrent cancel; re-read and re-decide. + self.client.metadata_cache.invalidate(key_id).await; + continue; + } + } + KeyState::Enabled | KeyState::Disabled | KeyState::PendingImport => { + return Ok(ExpiredKeyRemoval::StateChanged); + } } + break; + } + if !tombstoned { + return Err(VaultTransitKmsClient::metadata_cas_conflict(key_id)); } if !self.client.read_transit_key(key_id).await?.deletion_allowed { @@ -1432,10 +1686,14 @@ mod tests { let _ = client.schedule_key_deletion(&key_id, 7, None).await; } - /// Pins the known-risk fallback documented on `TransitKeyMetadata::synthesized`: - /// when KV metadata cannot be read, the synthesized record defaults to Enabled, - /// which weakens every state gate on that path. If this test turns red the - /// fallback semantics changed on purpose — update the comment there as well. + /// The persistence fallback for pre-metadata keys deliberately fabricates + /// an Enabled record (rustfs#4256 / rustfs#4262): those keys were usable + /// before metadata persistence existed and must stay usable once the + /// record is durably persisted. The old fail-open this test used to pin — + /// a failed metadata read or persist still yielded a usable Enabled key — + /// was flipped to fail closed for rustfs/backlog#1581; that side is + /// covered by `wired_encrypt_fails_closed_when_the_metadata_read_fails` + /// and `wired_synthesized_metadata_is_not_served_when_the_persist_fails`. #[test] fn synthesized_metadata_defaults_to_enabled() { let metadata = TransitKeyMetadata::synthesized(); @@ -1457,15 +1715,21 @@ mod tests { #[tokio::test] async fn wired_backend_lifecycle_overrides_reach_the_client() { let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let mut disabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + disabled.key_state = KeyState::Disabled; let vault = ScriptedVault::serve(vec![ - // disable: metadata cache miss reads KV, then persists Disabled. + // disable: versioned read (secret metadata + pinned version), then + // the check-and-set write persisting Disabled. + ScriptedResponse::ok(kv2_metadata_read_data(1)), ScriptedResponse::ok(metadata_read_data(&metadata)), ScriptedResponse::ok(kv2_write_ack()), - // enable: the state gate hits the metadata cache, so only the - // persisting write goes out. + // enable: another versioned read against the Disabled record, then + // the check-and-set write persisting Enabled. + ScriptedResponse::ok(kv2_metadata_read_data(2)), + ScriptedResponse::ok(metadata_read_data(&disabled)), ScriptedResponse::ok(kv2_write_ack()), - // rotate: the gate hits the cache again; the single rotate - // attempt fails and must not be retried. + // rotate: the state gate hits the metadata cache; the single + // rotate attempt fails and must not be retried. ScriptedResponse::error(503, "standby"), ]) .await; @@ -1493,7 +1757,392 @@ mod tests { assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}"); let requests = vault.requests(); - assert_eq!(requests.len(), 4, "gated reads, two writes and one rotate attempt: {requests:?}"); - assert_eq!(requests[3], "POST /v1/transit/keys/wired-key/rotate", "{requests:?}"); + assert_eq!(requests.len(), 7, "two versioned read+write cycles plus one rotate attempt: {requests:?}"); + assert_eq!(requests[6], "POST /v1/transit/keys/wired-key/rotate", "{requests:?}"); + } + + /// KV2 secret-metadata read payload (`kv2::read_metadata`) pinning the + /// current secret version used as the check-and-set base. + fn kv2_metadata_read_data(current_version: u64) -> serde_json::Value { + serde_json::json!({ + "cas_required": false, + "created_time": "2026-01-01T00:00:00Z", + "current_version": current_version, + "delete_version_after": "0s", + "max_versions": 0, + "oldest_version": 0, + "updated_time": "2026-01-01T00:00:00Z", + "custom_metadata": null, + "versions": {}, + }) + } + + const CAS_CONFLICT_MESSAGE: &str = "check-and-set parameter did not match the current version"; + + const METADATA_PATH: &str = "/v1/secret/data/rustfs/kms/transit-metadata/wired-key"; + const METADATA_VERSION_PATH: &str = "/v1/secret/metadata/rustfs/kms/transit-metadata/wired-key"; + + #[tokio::test] + async fn wired_disable_retries_past_a_cas_conflict_with_a_fresh_read() { + let enabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let (vault, client) = scripted_client(vec![ + ScriptedResponse::ok(kv2_metadata_read_data(1)), + ScriptedResponse::ok(metadata_read_data(&enabled)), + ScriptedResponse::error(400, CAS_CONFLICT_MESSAGE), + // The conflict must trigger a fresh versioned read, then the + // write is check-and-set against the new snapshot. + ScriptedResponse::ok(kv2_metadata_read_data(2)), + ScriptedResponse::ok(metadata_read_data(&enabled)), + ScriptedResponse::ok(kv2_write_ack()), + ]) + .await; + + client + .disable_key("wired-key", None) + .await + .expect("a single check-and-set conflict must be absorbed by a re-read"); + + let requests = vault.requests(); + assert_eq!(requests.len(), 6, "two read+read+write cycles: {requests:?}"); + assert_eq!(requests[0], format!("GET {METADATA_VERSION_PATH}")); + assert_eq!(requests[1], format!("GET {METADATA_PATH}?version=1")); + assert_eq!(requests[2], format!("POST {METADATA_PATH}")); + assert_eq!(requests[3], format!("GET {METADATA_VERSION_PATH}"), "conflict must re-read: {requests:?}"); + assert_eq!(requests[4], format!("GET {METADATA_PATH}?version=2")); + assert_eq!(requests[5], format!("POST {METADATA_PATH}")); + } + + #[tokio::test] + async fn wired_disable_cas_conflict_budget_is_bounded() { + let enabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let mut responses = Vec::new(); + for cycle in 0..3u64 { + responses.push(ScriptedResponse::ok(kv2_metadata_read_data(cycle + 1))); + responses.push(ScriptedResponse::ok(metadata_read_data(&enabled))); + responses.push(ScriptedResponse::error(400, CAS_CONFLICT_MESSAGE)); + } + let (vault, client) = scripted_client(responses).await; + + let error = client + .disable_key("wired-key", None) + .await + .expect_err("exhausting the check-and-set budget must surface the conflict"); + assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}"); + assert!( + error.to_string().contains("Concurrent modification"), + "the error must name the conflict: {error}" + ); + + let requests = vault.requests(); + assert_eq!(requests.len(), 9, "exactly three read+read+write cycles, no blind replays: {requests:?}"); + } + + #[tokio::test] + async fn wired_cas_conflict_reread_revalidates_the_state_gate() { + let enabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let mut pending = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + pending.key_state = KeyState::PendingDeletion; + let (vault, client) = scripted_client(vec![ + ScriptedResponse::ok(kv2_metadata_read_data(1)), + ScriptedResponse::ok(metadata_read_data(&enabled)), + ScriptedResponse::error(400, CAS_CONFLICT_MESSAGE), + // The concurrent writer scheduled the key for deletion; the + // re-read must re-run the state gate and reject the disable. + ScriptedResponse::ok(kv2_metadata_read_data(2)), + ScriptedResponse::ok(metadata_read_data(&pending)), + ]) + .await; + + let error = client + .disable_key("wired-key", None) + .await + .expect_err("the re-read state gate must reject a pending-deletion key"); + assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}"); + assert!(error.to_string().contains("pending deletion"), "got {error}"); + + let requests = vault.requests(); + assert_eq!(requests.len(), 5, "the gate rejection must not issue another write: {requests:?}"); + assert!(requests[4].starts_with("GET "), "{requests:?}"); + } + + #[tokio::test] + async fn wired_encrypt_key_not_found_invalidates_the_cached_metadata() { + let enabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let mut disabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + disabled.key_state = KeyState::Disabled; + let (vault, client) = scripted_client(vec![ + // First encrypt: gate reads Enabled and caches it, then the + // transit call reports the key gone server-side. + ScriptedResponse::ok(metadata_read_data(&enabled)), + ScriptedResponse::error(404, "encryption key not found"), + // Second encrypt: the state error must have dropped the cache + // entry, so the gate re-reads and sees the Disabled record. + ScriptedResponse::ok(metadata_read_data(&disabled)), + ]) + .await; + + let request = EncryptRequest { + key_id: "wired-key".to_string(), + plaintext: b"plaintext".to_vec(), + encryption_context: HashMap::new(), + grant_tokens: Vec::new(), + }; + let error = client + .encrypt(&request, None) + .await + .expect_err("the scripted 404 must fail the encrypt"); + assert!(matches!(error, KmsError::KeyNotFound { .. }), "got {error:?}"); + + let error = client + .encrypt(&request, None) + .await + .expect_err("the re-read Disabled record must reject the encrypt"); + assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}"); + + let requests = vault.requests(); + assert_eq!( + requests.len(), + 3, + "the second gate must re-read instead of trusting the stale Enabled entry, \ + and must not reach the encrypt endpoint: {requests:?}" + ); + assert_eq!(requests[2], format!("GET {METADATA_PATH}"), "{requests:?}"); + } + + #[tokio::test] + async fn wired_metadata_cache_ttl_expiry_forces_a_fresh_read() { + let enabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let mut disabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + disabled.key_state = KeyState::Disabled; + let (vault, mut client) = scripted_client(vec![ + ScriptedResponse::ok(metadata_read_data(&enabled)), + ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:scripted" })), + // Post-expiry gate read observes the disable another node + // persisted in the meantime. + ScriptedResponse::ok(metadata_read_data(&disabled)), + ]) + .await; + // A 1ns TTL expires between any two awaits, standing in for the real + // 300s bound without a wall-clock sleep. + client.rebuild_metadata_cache_for_tests(METADATA_CACHE_CAPACITY, Duration::from_nanos(1)); + + let request = EncryptRequest { + key_id: "wired-key".to_string(), + plaintext: b"plaintext".to_vec(), + encryption_context: HashMap::new(), + grant_tokens: Vec::new(), + }; + client + .encrypt(&request, None) + .await + .expect("the first encrypt must pass the Enabled gate"); + + let error = client + .encrypt(&request, None) + .await + .expect_err("after TTL expiry the gate must see the remote disable"); + assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}"); + + let requests = vault.requests(); + assert_eq!(requests.len(), 3, "the expired entry must force a fresh KV read: {requests:?}"); + assert_eq!(requests[2], format!("GET {METADATA_PATH}"), "{requests:?}"); + } + + #[tokio::test] + async fn metadata_cache_capacity_is_bounded() { + let records: Vec<_> = (0..3) + .map(|_| { + ScriptedResponse::ok(metadata_read_data(&TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()))) + }) + .collect(); + let (_vault, mut client) = scripted_client(records).await; + client.rebuild_metadata_cache_for_tests(2, METADATA_CACHE_TTL); + + for key_id in ["key-a", "key-b", "key-c"] { + client + .get_key_metadata(key_id) + .await + .expect("each scripted metadata read must succeed"); + } + + client.metadata_cache.run_pending_tasks().await; + assert!( + client.metadata_cache.entry_count() <= 2, + "the cache must not hold more entries than its capacity, got {}", + client.metadata_cache.entry_count() + ); + } + + #[tokio::test] + async fn wired_encrypt_fails_closed_when_the_metadata_read_fails() { + let (vault, client) = scripted_client(vec![ScriptedResponse::error(403, "permission denied")]).await; + + let error = client + .encrypt( + &EncryptRequest { + key_id: "wired-key".to_string(), + plaintext: b"plaintext".to_vec(), + encryption_context: HashMap::new(), + grant_tokens: Vec::new(), + }, + None, + ) + .await + .expect_err("a failed metadata read must fail the encrypt, not synthesize Enabled"); + assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}"); + + let requests = vault.requests(); + assert_eq!(requests.len(), 1, "the gate failure must never reach the encrypt endpoint: {requests:?}"); + } + + #[tokio::test] + async fn wired_synthesized_metadata_is_not_served_when_the_persist_fails() { + // Regression for the rustfs/backlog#1581 fail-open flip: a missing + // metadata record used to synthesize a usable Enabled record even when + // persisting it failed, letting encrypt proceed on state no other node + // could observe. The persist failure must now fail the read. + let (vault, client) = scripted_client(vec![ + ScriptedResponse::error(404, "no value found"), + ScriptedResponse::ok(transit_key_read_data("wired-key")), + ScriptedResponse::error(500, "kv write failed"), + ]) + .await; + + let error = client + .encrypt( + &EncryptRequest { + key_id: "wired-key".to_string(), + plaintext: b"plaintext".to_vec(), + encryption_context: HashMap::new(), + grant_tokens: Vec::new(), + }, + None, + ) + .await + .expect_err("an unpersisted synthesized record must never gate an encrypt open"); + assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}"); + + let requests = vault.requests(); + assert_eq!(requests.len(), 3, "read, existence check, failed persist — and no encrypt: {requests:?}"); + assert_eq!(requests[2], format!("POST {METADATA_PATH}"), "{requests:?}"); + } + + #[tokio::test] + async fn wired_synthesized_metadata_create_race_adopts_the_winning_record() { + let mut disabled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + disabled.key_state = KeyState::Disabled; + let (vault, client) = scripted_client(vec![ + ScriptedResponse::error(404, "no value found"), + ScriptedResponse::ok(transit_key_read_data("wired-key")), + // Another node persisted a record first; the create-only + // check-and-set loses and the re-read adopts the winner. + ScriptedResponse::error(400, CAS_CONFLICT_MESSAGE), + ScriptedResponse::ok(metadata_read_data(&disabled)), + ]) + .await; + + let error = client + .encrypt( + &EncryptRequest { + key_id: "wired-key".to_string(), + plaintext: b"plaintext".to_vec(), + encryption_context: HashMap::new(), + grant_tokens: Vec::new(), + }, + None, + ) + .await + .expect_err("the winner's Disabled record must gate the encrypt, not the loser's Enabled one"); + assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}"); + + let requests = vault.requests(); + assert_eq!(requests.len(), 4, "the lost create race must re-read, never overwrite: {requests:?}"); + assert_eq!(requests[3], format!("GET {METADATA_PATH}"), "{requests:?}"); + } + + #[tokio::test] + async fn wired_backend_create_loses_the_metadata_create_race_and_read_confirms() { + let winner = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let vault = ScriptedVault::serve(vec![ + // Pre-check: the transit key does not exist yet. + ScriptedResponse::error(404, "not found"), + // Transit create succeeds, but a concurrent creator persists the + // metadata record first. + ScriptedResponse::ok(serde_json::json!({})), + ScriptedResponse::error(400, CAS_CONFLICT_MESSAGE), + // Second pass: the pre-check now read-confirms the winner. + ScriptedResponse::ok(transit_key_read_data("wired-key")), + ScriptedResponse::ok(metadata_read_data(&winner)), + ]) + .await; + let config = KmsConfig::vault_transit( + url::Url::parse(&vault.address).expect("scripted vault address should parse"), + "scripted-token".to_string(), + ) + .with_insecure_development_defaults(); + let backend = VaultTransitKmsBackend::new(config) + .await + .expect("vault transit backend should build"); + + let response = backend + .create_key(CreateKeyRequest { + key_name: Some("wired-key".to_string()), + ..Default::default() + }) + .await + .expect("losing the metadata create race to an identical record must recover the create"); + assert_eq!(response.key_metadata.key_state, KeyState::Enabled); + + let requests = vault.requests(); + assert_eq!(requests.len(), 5, "one lost create pass plus one read-confirm pass: {requests:?}"); + assert!( + requests[3].starts_with("GET ") && requests[4].starts_with("GET "), + "the recovery pass must be reads only: {requests:?}" + ); + } + + #[tokio::test] + async fn wired_expired_sweep_backs_off_when_cancel_wins_the_cas_race() { + let mut pending = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + pending.key_state = KeyState::PendingDeletion; + pending.deletion_date = Some(Zoned::now()); + let cancelled = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); + let vault = ScriptedVault::serve(vec![ + // The transit key still exists. + ScriptedResponse::ok(transit_key_read_data("wired-key")), + // Versioned read finds a due pending-deletion record, but the + // tombstone write loses the check-and-set race to a cancel. + ScriptedResponse::ok(kv2_metadata_read_data(1)), + ScriptedResponse::ok(metadata_read_data(&pending)), + ScriptedResponse::error(400, CAS_CONFLICT_MESSAGE), + // The re-read sees the cancelled (Enabled) record: back off. + ScriptedResponse::ok(kv2_metadata_read_data(2)), + ScriptedResponse::ok(metadata_read_data(&cancelled)), + ]) + .await; + let config = KmsConfig::vault_transit( + url::Url::parse(&vault.address).expect("scripted vault address should parse"), + "scripted-token".to_string(), + ) + .with_insecure_development_defaults(); + let backend = VaultTransitKmsBackend::new(config) + .await + .expect("vault transit backend should build"); + + let now = Zoned::now() + Duration::from_secs(3600); + let outcome = backend + .remove_expired_key("wired-key", &now) + .await + .expect("losing the tombstone race to a cancel must back off cleanly"); + assert_eq!(outcome, ExpiredKeyRemoval::StateChanged); + + let requests = vault.requests(); + assert_eq!(requests.len(), 6, "no delete may follow a lost tombstone race: {requests:?}"); + assert!( + !requests + .iter() + .any(|line| line.contains("/transit/keys/wired-key/config") || line.starts_with("DELETE ")), + "the sweep must not touch the transit key after backing off: {requests:?}" + ); } }