diff --git a/crates/kms/src/backends/aws.rs b/crates/kms/src/backends/aws.rs index 54321b067..464f6dbab 100644 --- a/crates/kms/src/backends/aws.rs +++ b/crates/kms/src/backends/aws.rs @@ -67,7 +67,7 @@ use aws_smithy_types::Blob; use jiff::Zoned; use tokio_util::sync::CancellationToken; -use super::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend}; +use super::{BackendCapabilities, ExpiredKeyRemoval, KmsBackend, empty_key_page, list_keys_page_size}; use crate::config::{BackendConfig, KmsConfig}; use crate::error::{KmsError, Result}; use crate::policy::{self, AttemptError, ErrorClass, OpClass, RetryPolicy, classify_status}; @@ -577,6 +577,13 @@ impl KmsBackend for AwsKmsBackend { /// and creation date every caller relies on costs one `DescribeKey` per /// listed key. The page size is bounded by the caller's `limit`. async fn list_keys(&self, request: ListKeysRequest) -> Result { + // AWS rejects a `Limit` of zero, and clamping it up to one would return + // a key to a caller that asked for none; the empty page is answered + // here instead. + if list_keys_page_size(request.limit).is_none() { + return Ok(empty_key_page()); + } + let limit = request .limit .map(|limit| i32::try_from(limit).unwrap_or(i32::MAX).clamp(1, 1000)); @@ -1113,6 +1120,26 @@ mod tests { assert_eq!(http_client.actual_requests().count(), 0, "no key may be created in AWS"); } + /// AWS rejects `Limit: 0` outright, so the request cannot be forwarded as + /// written; clamping it up to one would answer a caller that asked for no + /// keys with a key. The empty page is served locally instead. + #[tokio::test] + async fn zero_limit_list_returns_an_empty_page_without_calling_aws() { + let (http_client, backend) = scripted_backend(Vec::new()); + let response = backend + .list_keys(ListKeysRequest { + limit: Some(0), + ..Default::default() + }) + .await + .expect("a zero-limit list must succeed"); + + assert!(response.keys.is_empty()); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + assert_eq!(http_client.actual_requests().count(), 0, "no request should reach AWS"); + } + /// Signing keys are outside the envelope-encryption surface this backend /// serves; creating one is refused rather than silently downgraded. #[tokio::test] diff --git a/crates/kms/src/backends/local.rs b/crates/kms/src/backends/local.rs index f472f1f4c..9a9269d3a 100644 --- a/crates/kms/src/backends/local.rs +++ b/crates/kms/src/backends/local.rs @@ -16,7 +16,7 @@ use crate::backends::{ BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_status_permits, - ensure_tag_keys_are_mutable, + ensure_tag_keys_are_mutable, paginate_keys, }; use crate::config::KmsConfig; use crate::config::LocalConfig; @@ -1214,6 +1214,32 @@ impl LocalKmsClient { Ok(master_key.into()) } + /// Every key identifier in the key directory, sorted. + /// + /// `read_dir` order is arbitrary and may differ between calls over the same + /// directory, so it cannot carry a pagination cursor. Sorting gives the key + /// set a stable total order that a marker can point into. + async fn sorted_key_ids(&self) -> Result> { + let mut key_ids = Vec::new(); + let mut entries = fs::read_dir(&self.config.key_dir).await?; + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + if path.extension().is_some_and(|ext| ext == "key") + && let Some(key_id) = path.file_stem().and_then(|stem| stem.to_str()) + { + key_ids.push(key_id.to_string()); + } + } + key_ids.sort_unstable(); + Ok(key_ids) + } + + /// One page of the key set, ordered by key identifier. + /// + /// Paging is real here rather than a first-page-only approximation: + /// callers that must see every key — the deletion sweep above all, which + /// only destroys expired material for keys it actually lists — depend on + /// `truncated` and `next_marker` to reach past the first page. pub(crate) async fn list_keys( &self, request: &ListKeysRequest, @@ -1221,44 +1247,43 @@ impl LocalKmsClient { ) -> Result { debug!("Listing keys"); - let mut keys = Vec::new(); - let limit = request.limit.unwrap_or(100) as usize; - let mut count = 0; + let key_ids = self.sorted_key_ids().await?; + let page = paginate_keys(&key_ids, request, String::as_str); - let mut entries = fs::read_dir(&self.config.key_dir).await?; + // Only the page is read from disk, so the cost of a list stays bounded + // by the requested limit rather than by the size of the key set. + let mut keys = Vec::with_capacity(page.items.len()); + for key_id in page.items { + // A key that vanished or cannot be decoded is dropped from the page + // instead of failing it: concurrent removal is normal, and the + // cursor is derived from the identifier list, so the listing still + // advances past it. + let key_info = match self.describe_key(key_id, None).await { + Ok(key_info) => key_info, + Err(error) => { + debug!(key_id, %error, "skipping unreadable key while listing"); + continue; + } + }; - while let Some(entry) = entries.next_entry().await? { - if count >= limit { - break; - } - - let path = entry.path(); - if path.extension().is_some_and(|ext| ext == "key") - && let Some(stem) = path.file_stem() - && let Some(key_id) = stem.to_str() - && let Ok(key_info) = self.describe_key(key_id, None).await + if let Some(ref status_filter) = request.status_filter + && &key_info.status != status_filter { - // Apply filters - if let Some(ref status_filter) = request.status_filter - && &key_info.status != status_filter - { - continue; - } - if let Some(ref usage_filter) = request.usage_filter - && &key_info.usage != usage_filter - { - continue; - } - - keys.push(key_info); - count += 1; + continue; } + if let Some(ref usage_filter) = request.usage_filter + && &key_info.usage != usage_filter + { + continue; + } + + keys.push(key_info); } Ok(ListKeysResponse { keys, - next_marker: None, // Simple implementation without pagination - truncated: false, + next_marker: page.next_marker, + truncated: page.truncated, }) } @@ -3087,4 +3112,131 @@ mod tests { .expect("repeat removal"); assert_eq!(outcome, crate::backends::ExpiredKeyRemoval::Removed); } + + // -- Listing and pagination --------------------------------------------- + + fn page_request(limit: u32, marker: Option<&str>) -> ListKeysRequest { + ListKeysRequest { + limit: Some(limit), + marker: marker.map(str::to_string), + usage_filter: None, + status_filter: None, + } + } + + async fn create_keys(client: &LocalKmsClient, key_ids: &[String]) { + for key_id in key_ids { + client + .create_key(key_id, "AES_256", None) + .await + .expect("key should be created"); + } + } + + /// Paging must reach every key exactly once. A backend that reports the + /// first page as the whole key set strands everything behind it — the + /// deletion sweep would never destroy expired material past page one. + #[tokio::test] + async fn list_keys_pages_through_the_whole_key_set() { + let (client, _temp_dir) = create_test_client().await; + let expected: Vec = (0..7).map(|index| format!("page-key-{index:02}")).collect(); + create_keys(&client, &expected).await; + + let mut seen = Vec::new(); + let mut marker: Option = None; + // Bounded so a listing that cannot advance fails the assertion below + // instead of hanging the test run. + for _ in 0..expected.len() + 1 { + let response = client + .list_keys(&page_request(3, marker.as_deref()), None) + .await + .expect("list should succeed"); + assert!(response.keys.len() <= 3, "a page must not exceed the requested limit"); + seen.extend(response.keys.iter().map(|key| key.key_id.clone())); + if !response.truncated { + assert!(response.next_marker.is_none(), "a final page must not offer a cursor"); + break; + } + marker = Some(response.next_marker.expect("a truncated page must offer a cursor")); + } + + assert_eq!(seen, expected, "paging must visit every key exactly once, in identifier order"); + } + + /// Exact-limit boundary: a page that ends on the last key is complete. + #[tokio::test] + async fn list_keys_page_ending_on_the_last_key_is_not_truncated() { + let (client, _temp_dir) = create_test_client().await; + let key_ids: Vec = (0..3).map(|index| format!("exact-key-{index}")).collect(); + create_keys(&client, &key_ids).await; + + let response = client + .list_keys(&page_request(3, None), None) + .await + .expect("list should succeed"); + assert_eq!(response.keys.len(), 3); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + + // One key short of the set, the same listing is truncated. + let response = client + .list_keys(&page_request(2, None), None) + .await + .expect("list should succeed"); + assert!(response.truncated); + assert_eq!(response.next_marker.as_deref(), Some("exact-key-1")); + } + + /// The cursor is an identifier, not an index, so it keeps working after the + /// key it names is destroyed — which is exactly what the deletion sweep + /// does to the keys it retires between pages. + #[tokio::test] + async fn list_keys_resumes_after_a_deleted_marker_key() { + let (client, _temp_dir) = create_test_client().await; + let key_ids: Vec = ["marker-a", "marker-b", "marker-c"].iter().map(|id| id.to_string()).collect(); + create_keys(&client, &key_ids).await; + + fs::remove_file(client.master_key_path("marker-b").expect("key path")) + .await + .expect("key file should be removable"); + + let response = client + .list_keys(&page_request(10, Some("marker-b")), None) + .await + .expect("list should succeed"); + let listed: Vec<&str> = response.keys.iter().map(|key| key.key_id.as_str()).collect(); + assert_eq!(listed, vec!["marker-c"], "a vanished marker must not restart the listing"); + assert!(!response.truncated); + } + + /// A zero limit means zero keys, and no cursor to loop on. + #[tokio::test] + async fn list_keys_with_zero_limit_returns_an_empty_page() { + let (client, _temp_dir) = create_test_client().await; + create_keys(&client, &["zero-limit-key".to_string()]).await; + + let response = client + .list_keys(&page_request(0, None), None) + .await + .expect("a zero-limit list must succeed"); + assert!(response.keys.is_empty()); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + } + + /// A limit past the end of the key set returns everything, once. + #[tokio::test] + async fn list_keys_limit_beyond_the_key_set_returns_one_complete_page() { + let (client, _temp_dir) = create_test_client().await; + let key_ids: Vec = (0..3).map(|index| format!("huge-limit-key-{index}")).collect(); + create_keys(&client, &key_ids).await; + + let response = client + .list_keys(&page_request(u32::MAX, None), None) + .await + .expect("list should succeed"); + assert_eq!(response.keys.len(), 3); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + } } diff --git a/crates/kms/src/backends/mod.rs b/crates/kms/src/backends/mod.rs index acef3c7e8..d0742de04 100644 --- a/crates/kms/src/backends/mod.rs +++ b/crates/kms/src/backends/mod.rs @@ -126,6 +126,96 @@ pub(crate) fn ensure_tag_keys_are_mutable<'a>(tag_keys: impl IntoIterator { + /// The identifiers this page covers, in listing order. + pub(crate) items: &'a [T], + /// Where the next page resumes; `None` when this page is the last one. + pub(crate) next_marker: Option, + /// Whether keys remain beyond this page. + pub(crate) truncated: bool, +} + +/// Cut the page `request` asks for out of `sorted`. +/// +/// `sorted` must be ordered by the identifier `key_id_of` returns, and the +/// marker is an *exclusive lower bound* on that identifier rather than an index +/// into the sequence. That is what makes paging survive concurrent mutation: +/// the next page resumes at the first identifier greater than the marker, so a +/// key added or removed elsewhere in the ordering — including the marker key +/// itself, which the deletion sweep routinely destroys — cannot make the +/// listing skip keys or restart from the beginning. +/// +/// A `limit` of zero is honoured as written: the caller asked for no keys and +/// gets an empty, non-truncated page (see [`list_keys_page_size`]). +/// +/// Filters are applied by the caller to `items` after this slice, so a filtered +/// page can be shorter than `limit` — and even empty — while more keys remain. +/// Callers must page until `truncated` is false rather than until a page comes +/// back short. +pub(crate) fn paginate_keys<'a, T>(sorted: &'a [T], request: &ListKeysRequest, key_id_of: impl Fn(&T) -> &str) -> KeyPage<'a, T> { + let Some(limit) = list_keys_page_size(request.limit) else { + return KeyPage { + items: &[], + next_marker: None, + truncated: false, + }; + }; + + let start = match request.marker.as_deref() { + Some(marker) => sorted.partition_point(|item| key_id_of(item) <= marker), + None => 0, + }; + // `partition_point` never exceeds the length, so both bounds stay in range + // however large `limit` is. + let end = start.saturating_add(limit).min(sorted.len()); + let items = &sorted[start..end]; + let truncated = end < sorted.len(); + + KeyPage { + items, + // Resuming from the last identifier on the page, not from an index, + // keeps the cursor meaningful after the key it names disappears. + next_marker: if truncated { + items.last().map(|item| key_id_of(item).to_string()) + } else { + None + }, + truncated, + } +} + +/// Resolve the page size of a [`ListKeysRequest`]; `None` means the caller +/// asked for no keys at all. +/// +/// `Some(0)` is a well-formed request for an empty page — the reading rustfs +/// already gives `max-keys=0` on the S3 listing path — not a malformed one and +/// not an omitted value. Rounding it up to a default would hand back a full +/// page of keys to a caller that explicitly asked for none. +pub(crate) fn list_keys_page_size(limit: Option) -> Option { + match limit.unwrap_or(DEFAULT_LIST_KEYS_PAGE_SIZE) { + 0 => None, + size => Some(size as usize), + } +} + +/// The response to a request for zero keys. +/// +/// `truncated` is false even when keys exist: a zero-length page carries no +/// identifier to resume from, so claiming more results would hand back a cursor +/// the caller can never advance and turn a `while truncated` loop into a +/// non-terminating one. +pub(crate) fn empty_key_page() -> ListKeysResponse { + ListKeysResponse { + keys: Vec::new(), + next_marker: None, + truncated: false, + } +} + /// Simplified KMS backend interface for manager #[async_trait] pub trait KmsBackend: Send + Sync { @@ -546,4 +636,84 @@ mod tests { insta::assert_json_snapshot!("static_backend_capabilities", capabilities_snapshot(backend.capabilities())); } + + // -- Pagination boundaries ---------------------------------------------- + + fn key_ids(count: usize) -> Vec { + (0..count).map(|index| format!("key-{index:02}")).collect() + } + + fn page_request(limit: Option, marker: Option<&str>) -> ListKeysRequest { + ListKeysRequest { + limit, + marker: marker.map(str::to_string), + usage_filter: None, + status_filter: None, + } + } + + fn page_of(keys: &[String], limit: Option, marker: Option<&str>) -> (Vec, Option, bool) { + let page = paginate_keys(keys, &page_request(limit, marker), String::as_str); + (page.items.to_vec(), page.next_marker, page.truncated) + } + + /// Zero keys requested, zero keys returned — and no cursor, so a caller + /// looping on `truncated` terminates instead of asking forever. Slicing a + /// zero-length page out of a non-empty key set must not reach for the + /// element before the page either. + #[test] + fn zero_limit_returns_an_empty_untruncated_page() { + let keys = key_ids(3); + assert_eq!(page_of(&keys, Some(0), None), (Vec::new(), None, false)); + assert_eq!(page_of(&keys, Some(0), Some("key-01")), (Vec::new(), None, false)); + // Also at the ends of the key set, where a page has no predecessor. + assert_eq!(page_of(&[], Some(0), None), (Vec::new(), None, false)); + assert_eq!(page_of(&keys, Some(0), Some("key-02")), (Vec::new(), None, false)); + + assert_eq!(list_keys_page_size(Some(0)), None); + assert_eq!(list_keys_page_size(None), Some(DEFAULT_LIST_KEYS_PAGE_SIZE as usize)); + assert_eq!(list_keys_page_size(Some(7)), Some(7)); + } + + /// A limit past the end of the key set is not an overflow. + #[test] + fn oversized_limit_returns_the_whole_key_set_once() { + let keys = key_ids(3); + assert_eq!(page_of(&keys, Some(u32::MAX), None), (keys.clone(), None, false)); + assert_eq!(page_of(&keys, Some(u32::MAX), Some("key-01")), (vec![keys[2].clone()], None, false)); + } + + /// The cursor is an identifier, so a marker naming a key that no longer + /// exists resumes after where it would have been instead of restarting. + #[test] + fn marker_for_a_removed_key_resumes_after_it() { + let keys = vec!["key-00".to_string(), "key-02".to_string()]; + assert_eq!(page_of(&keys, Some(10), Some("key-01")), (vec!["key-02".to_string()], None, false)); + // A marker past every key ends the listing rather than wrapping. + assert_eq!(page_of(&keys, Some(10), Some("key-99")), (Vec::new(), None, false)); + // A marker before every key yields the whole set. + assert_eq!(page_of(&keys, Some(10), Some("key")), (keys.clone(), None, false)); + } + + /// Truncation flips exactly at the page boundary, and paging covers the + /// key set once end to end. + #[test] + fn pages_tile_the_key_set_exactly_at_the_limit_boundary() { + let keys = key_ids(4); + assert_eq!(page_of(&keys, Some(4), None), (keys.clone(), None, false)); + assert_eq!(page_of(&keys, Some(3), None), (keys[..3].to_vec(), Some("key-02".to_string()), true)); + + let mut seen = Vec::new(); + let mut marker = None; + loop { + let (items, next_marker, truncated) = page_of(&keys, Some(2), marker.as_deref()); + seen.extend(items); + if !truncated { + assert!(next_marker.is_none(), "a final page must not offer a cursor"); + break; + } + marker = Some(next_marker.expect("a truncated page must offer a cursor")); + } + assert_eq!(seen, keys, "paging must tile the key set exactly once"); + } } diff --git a/crates/kms/src/backends/static_kms.rs b/crates/kms/src/backends/static_kms.rs index 6f048d41a..20574b5bb 100644 --- a/crates/kms/src/backends/static_kms.rs +++ b/crates/kms/src/backends/static_kms.rs @@ -21,7 +21,7 @@ //! //! encrypted_data(plaintext_len+16) || nonce (12 bytes) -use crate::backends::{BackendCapabilities, KmsBackend}; +use crate::backends::{BackendCapabilities, KmsBackend, empty_key_page, list_keys_page_size}; use crate::config::{BackendConfig, KmsConfig}; use crate::encryption::DataKeyEnvelope; use crate::error::{KmsError, Result}; @@ -262,6 +262,12 @@ impl StaticKmsBackend { /// List the single configured key, honouring the pagination marker. pub(crate) fn list_configured_key(&self, request: &ListKeysRequest) -> Result { + // A caller asking for no keys gets none, even from a backend whose + // whole key set is one key. + if list_keys_page_size(request.limit).is_none() { + return Ok(empty_key_page()); + } + let key_info = KeyInfo { key_id: self.key_id.clone(), description: Some("Static single-key KMS backend".to_string()), @@ -657,6 +663,23 @@ mod tests { assert!(!response.truncated); } + /// A zero limit means zero keys, even for a backend whose whole key set is + /// a single configured key. + #[tokio::test] + async fn zero_limit_list_returns_an_empty_page() { + let (backend, _key_id, _key) = create_test_backend().await; + + let response = backend + .list_configured_key(&ListKeysRequest { + limit: Some(0), + ..Default::default() + }) + .expect("a zero-limit list must succeed"); + assert!(response.keys.is_empty()); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + } + #[tokio::test] async fn lifecycle_mutations_are_unsupported_at_the_product_surface() { let (backend, key_id, _key) = create_test_backend().await; diff --git a/crates/kms/src/backends/vault.rs b/crates/kms/src/backends/vault.rs index e7c12a32a..3800ccd61 100644 --- a/crates/kms/src/backends/vault.rs +++ b/crates/kms/src/backends/vault.rs @@ -19,8 +19,8 @@ use crate::backends::vault_credentials::{ token_source_for, }; use crate::backends::{ - BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits, ensure_key_status_permits, - ensure_tag_keys_are_mutable, + BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, empty_key_page, ensure_key_state_permits, + ensure_key_status_permits, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, }; use crate::config::{KmsConfig, VaultConfig}; use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material}; @@ -1051,37 +1051,42 @@ impl VaultKmsClient { ) -> Result { debug!("Listing keys with limit: {:?}", request.limit); - let all_keys = self.list_vault_keys().await?; - let limit = request.limit.unwrap_or(100) as usize; - - // Simple pagination implementation - let start_idx = request - .marker - .as_ref() - .and_then(|m| all_keys.iter().position(|k| k == m)) - .map(|idx| idx + 1) - .unwrap_or(0); - - let end_idx = std::cmp::min(start_idx + limit, all_keys.len()); - let keys_page = &all_keys[start_idx..end_idx]; - - let mut key_infos = Vec::new(); - for key_id in keys_page { - if let Ok(key_info) = self.describe_key(key_id, None).await { - key_infos.push(key_info); - } + // A caller asking for no keys is answered without reaching Vault. + if list_keys_page_size(request.limit).is_none() { + return Ok(empty_key_page()); } - let next_marker = if end_idx < all_keys.len() { - Some(all_keys[end_idx - 1].clone()) - } else { - None - }; + let mut all_keys = self.list_vault_keys().await?; + // Vault's own LIST ordering is not part of its contract, so the sort is + // what makes the marker a stable cursor across calls. + all_keys.sort_unstable(); + let page = paginate_keys(&all_keys, request, String::as_str); + + let mut key_infos = Vec::with_capacity(page.items.len()); + for key_id in page.items { + // A key that disappeared between the listing and the read is + // dropped from the page rather than failing it; the cursor comes + // from the identifier list, so the listing still advances past it. + let Ok(key_info) = self.describe_key(key_id, None).await else { + continue; + }; + if request + .status_filter + .as_ref() + .is_some_and(|status| status != &key_info.status) + { + continue; + } + if request.usage_filter.as_ref().is_some_and(|usage| usage != &key_info.usage) { + continue; + } + key_infos.push(key_info); + } Ok(ListKeysResponse { keys: key_infos, - next_marker, - truncated: end_idx < all_keys.len(), + next_marker: page.next_marker, + truncated: page.truncated, }) } @@ -1751,6 +1756,35 @@ mod tests { }) } + /// A caller asking for no keys gets an empty page, and the page arithmetic + /// never reaches for the element before an empty page. The scripted key + /// listing stays unused: a request for zero keys has nothing to ask Vault. + #[tokio::test] + async fn zero_limit_list_returns_an_empty_page_without_calling_vault() { + let (vault, client) = + scripted_client(vec![ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a", "key-b"] }))]).await; + + let response = client + .list_keys( + &ListKeysRequest { + limit: Some(0), + ..Default::default() + }, + None, + ) + .await + .expect("a zero-limit list must succeed"); + + assert!(response.keys.is_empty()); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + assert!( + vault.requests().is_empty(), + "a request for no keys must not reach Vault: {:?}", + vault.requests() + ); + } + #[tokio::test] async fn wired_read_retries_transient_status_then_succeeds() { let (vault, client) = scripted_client(vec![ diff --git a/crates/kms/src/backends/vault_transit.rs b/crates/kms/src/backends/vault_transit.rs index 096913400..47854d9d7 100644 --- a/crates/kms/src/backends/vault_transit.rs +++ b/crates/kms/src/backends/vault_transit.rs @@ -19,8 +19,8 @@ use crate::backends::vault_credentials::{ token_source_for, }; use crate::backends::{ - BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, ensure_key_state_permits, - ensure_tag_keys_are_mutable, + BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, empty_key_page, ensure_key_state_permits, + ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, }; use crate::config::{KmsConfig, VaultTransitConfig}; use crate::encryption::{DataKeyEnvelope, generate_key_material}; @@ -852,7 +852,12 @@ impl VaultTransitKmsClient { request: &ListKeysRequest, _context: Option<&OperationContext>, ) -> Result { - let all_keys = self + // A caller asking for no keys is answered without reaching Vault. + if list_keys_page_size(request.limit).is_none() { + return Ok(empty_key_page()); + } + + let mut all_keys = self .run("vault_transit_list_keys", OpClass::ReadIdempotent, move || async move { let vault = self.vault().map_err(AttemptError::fatal)?; key::list(&vault.client, &self.config.mount_path).await.map_err(|e| { @@ -861,36 +866,27 @@ impl VaultTransitKmsClient { }) .await? .keys; + // Vault's own LIST ordering is not part of its contract, so the sort is + // what makes the marker a stable cursor across calls. + all_keys.sort_unstable(); + let page = paginate_keys(&all_keys, request, String::as_str); - let mut filtered = Vec::new(); - for key_id in all_keys { - let key_info = self.key_info(&key_id).await?; + // Reading metadata only for the page keeps a list bounded by the + // requested limit instead of by the size of the transit mount. + let mut keys = Vec::with_capacity(page.items.len()); + for key_id in page.items { + let key_info = self.key_info(key_id).await?; let usage_matches = request.usage_filter.as_ref().is_none_or(|usage| usage == &key_info.usage); let status_matches = request.status_filter.as_ref().is_none_or(|status| status == &key_info.status); if usage_matches && status_matches { - filtered.push(key_info); + keys.push(key_info); } } - let start_idx = request - .marker - .as_ref() - .and_then(|marker| filtered.iter().position(|info| &info.key_id == marker)) - .map(|idx| idx + 1) - .unwrap_or(0); - let limit = request.limit.unwrap_or(100) as usize; - let end_idx = std::cmp::min(start_idx + limit, filtered.len()); - let keys = filtered[start_idx..end_idx].to_vec(); - let next_marker = if end_idx < filtered.len() { - Some(filtered[end_idx - 1].key_id.clone()) - } else { - None - }; - Ok(ListKeysResponse { keys, - next_marker, - truncated: end_idx < filtered.len(), + next_marker: page.next_marker, + truncated: page.truncated, }) } @@ -1453,6 +1449,35 @@ mod tests { serde_json::to_value(&response).expect("serialize transit key read response") } + /// A caller asking for no keys gets an empty page, and the page arithmetic + /// never reaches for the element before an empty page. The scripted key + /// listing stays unused: a request for zero keys has nothing to ask Vault. + #[tokio::test] + async fn zero_limit_list_returns_an_empty_page_without_calling_vault() { + let (vault, client) = + scripted_client(vec![ScriptedResponse::ok(serde_json::json!({ "keys": ["key-a", "key-b"] }))]).await; + + let response = client + .list_keys( + &ListKeysRequest { + limit: Some(0), + ..Default::default() + }, + None, + ) + .await + .expect("a zero-limit list must succeed"); + + assert!(response.keys.is_empty()); + assert!(!response.truncated); + assert!(response.next_marker.is_none()); + assert!( + vault.requests().is_empty(), + "a request for no keys must not reach Vault: {:?}", + vault.requests() + ); + } + #[tokio::test] async fn wired_transit_encrypt_retries_transient_status() { let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default()); diff --git a/crates/kms/src/deletion_worker.rs b/crates/kms/src/deletion_worker.rs index b39748afa..c03a17b6e 100644 --- a/crates/kms/src/deletion_worker.rs +++ b/crates/kms/src/deletion_worker.rs @@ -36,6 +36,11 @@ use tracing::{debug, info, warn}; /// How often the worker looks for expired pending deletions. pub const DEFAULT_SWEEP_INTERVAL: Duration = Duration::from_secs(60); +/// Keys requested per `ListKeys` call while sweeping. The sweep follows +/// `truncated`/`next_marker` until the key set is exhausted, so this bounds the +/// working set of one call, not how many keys a sweep can reach. +const SWEEP_PAGE_SIZE: u32 = 100; + // --------------------------------------------------------------------------- // Metrics // @@ -240,7 +245,7 @@ impl DeletionWorker { let mut marker: Option = None; let listed_everything = loop { let request = ListKeysRequest { - limit: Some(100), + limit: Some(SWEEP_PAGE_SIZE), marker: marker.clone(), usage_filter: None, status_filter: None, @@ -270,6 +275,13 @@ impl DeletionWorker { break true; } match response.next_marker { + // A backend that hands back the cursor it was just given cannot + // advance. Following it would re-list the same page forever, so + // the sweep stops and reports the key set as partially seen. + Some(ref next_marker) if marker.as_ref() == Some(next_marker) => { + warn!(marker = %next_marker, "KMS deletion sweep listing did not advance"); + break false; + } Some(next_marker) => marker = Some(next_marker), // Truncated without a marker: the rest of the key set is out // of reach, so the census is incomplete. @@ -426,6 +438,41 @@ mod tests { assert_eq!(report, SweepReport::default()); } + /// Schedule `count` keys for deletion and report their identifiers. + async fn schedule_keys(backend: &LocalKmsBackend, count: usize) -> Vec { + let mut key_ids = Vec::with_capacity(count); + for index in 0..count { + let key_id = create_key(backend, &format!("expiring-{index:04}")).await; + schedule(backend, &key_id).await; + key_ids.push(key_id); + } + key_ids + } + + /// A deployment with more keys than fit in one listing page must still have + /// every expired key destroyed: the sweep has to follow the cursor instead + /// of stopping at the first page. + #[tokio::test] + async fn sweep_removes_expired_keys_beyond_the_first_page() { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + let total = SWEEP_PAGE_SIZE as usize + 5; + let key_ids = schedule_keys(&backend, total).await; + + let report = worker(backend.clone()).sweep(&after_window()).await; + assert_eq!(report.failed, 0); + assert_eq!( + report.removed.len(), + total, + "every expired key must be swept, not just the ones on the first page" + ); + // The keys sorting last are the ones a first-page-only sweep leaves + // behind for good. + for key_id in key_ids.iter().rev().take(5) { + assert_key_gone(&backend, key_id).await; + } + } + #[tokio::test] async fn cancelled_deletion_always_beats_the_sweep() { let temp_dir = tempfile::tempdir().expect("temp dir"); @@ -757,4 +804,27 @@ mod tests { } } } + + /// The census counts the whole key set, not the first page of it. A sweep + /// that stops after one page would publish a gauge that understates every + /// large deployment by exactly the keys it never listed. + #[test] + fn lifecycle_gauges_count_keys_beyond_the_first_page() { + let total = SWEEP_PAGE_SIZE as usize + 5; + let (snapshot, ()) = record_metrics(|| { + Box::pin(async move { + let temp_dir = tempfile::tempdir().expect("temp dir"); + let backend = local_backend(&temp_dir).await; + schedule_keys(&backend, total).await; + + // Well inside the seven-day window: every key is observed as + // pending rather than removed. + let report = worker(backend.clone()).sweep(&Zoned::now()).await; + assert_eq!(report.skipped, total, "every scheduled key must be inspected"); + assert!(report.removed.is_empty()); + }) + }); + + assert_eq!(gauge_value(&snapshot, METRIC_PENDING_DELETION_KEYS), Some(total as f64)); + } }