diff --git a/crates/ecstore/src/client/api_put_object_streaming.rs b/crates/ecstore/src/client/api_put_object_streaming.rs index 9fdcb9b8a..67df7e42b 100644 --- a/crates/ecstore/src/client/api_put_object_streaming.rs +++ b/crates/ecstore/src/client/api_put_object_streaming.rs @@ -208,20 +208,20 @@ impl TransitionClient { let mut compl_multipart_upload = CompleteMultipartUpload::default(); - let mut all_parts = Vec::::with_capacity(parts_info.len()); - let part_number = total_parts_count; - for i in 1..part_number { - let part = parts_info[&i].clone(); - - all_parts.push(part.clone()); + // Parts are keyed 1..=total_parts_count during upload; every one — including the last — + // must be collected. The previous exclusive `1..total_parts_count` bound dropped the final + // part, silently truncating the completed object (and produced zero parts for a single-part + // upload). + let mut all_parts = collect_complete_parts(&parts_info, total_parts_count)?; + for part in &all_parts { compl_multipart_upload.parts.push(CompletePart { - etag: part.etag, + etag: part.etag.clone(), part_num: part.part_num, - checksum_crc32: part.checksum_crc32, - checksum_crc32c: part.checksum_crc32c, - checksum_sha1: part.checksum_sha1, - checksum_sha256: part.checksum_sha256, - checksum_crc64nvme: part.checksum_crc64nvme, + checksum_crc32: part.checksum_crc32.clone(), + checksum_crc32c: part.checksum_crc32c.clone(), + checksum_sha1: part.checksum_sha1.clone(), + checksum_sha256: part.checksum_sha256.clone(), + checksum_crc64nvme: part.checksum_crc64nvme.clone(), }); } @@ -397,20 +397,20 @@ impl TransitionClient { let mut compl_multipart_upload = CompleteMultipartUpload::default(); - let part_number: i64 = total_parts_count; - let mut all_parts = Vec::::with_capacity(parts_info.read().unwrap().len()); - for i in 1..part_number { - let part = parts_info.read().unwrap()[&i].clone(); - - all_parts.push(part.clone()); + // Same inclusive collection as the serial path: parts are keyed 1..=total_parts_count, so + // the exclusive `1..total_parts_count` bound dropped the final part (and produced zero + // parts for a single-part upload), silently truncating the object. + let parts_snapshot = parts_info.read().unwrap().clone(); + let mut all_parts = collect_complete_parts(&parts_snapshot, total_parts_count)?; + for part in &all_parts { compl_multipart_upload.parts.push(CompletePart { - etag: part.etag, + etag: part.etag.clone(), part_num: part.part_num, - checksum_crc32: part.checksum_crc32, - checksum_crc32c: part.checksum_crc32c, - checksum_sha1: part.checksum_sha1, - checksum_sha256: part.checksum_sha256, - checksum_crc64nvme: part.checksum_crc64nvme, + checksum_crc32: part.checksum_crc32.clone(), + checksum_crc32c: part.checksum_crc32c.clone(), + checksum_sha1: part.checksum_sha1.clone(), + checksum_sha256: part.checksum_sha256.clone(), + checksum_crc64nvme: part.checksum_crc64nvme.clone(), ..Default::default() }); } @@ -573,3 +573,67 @@ impl TransitionClient { }) } } + +/// Collect the uploaded parts for CompleteMultipartUpload in ascending part order. +/// +/// Parts are keyed `1..=total_parts_count` during upload (see the upload loop that inserts each +/// part), so every one — including the final part — must be collected. The previous exclusive +/// `1..total_parts_count` bound dropped the last part, silently truncating the completed object, +/// and collected zero parts for a single-part upload. +fn collect_complete_parts(parts_info: &HashMap, total_parts_count: i64) -> Result, Error> { + let mut all_parts = Vec::with_capacity(parts_info.len()); + for i in 1..=total_parts_count { + let part = parts_info + .get(&i) + .ok_or_else(|| Error::other(format!("missing uploaded part {i} of {total_parts_count}")))?; + all_parts.push(part.clone()); + } + Ok(all_parts) +} + +#[cfg(test)] +mod tests { + use super::{ObjectPart, collect_complete_parts}; + use std::collections::HashMap; + + fn parts_map(n: i64) -> HashMap { + let mut m = HashMap::new(); + for i in 1..=n { + m.insert( + i, + ObjectPart { + part_num: i, + ..Default::default() + }, + ); + } + m + } + + #[test] + fn collects_every_part_including_the_last() { + let collected: Vec = collect_complete_parts(&parts_map(3), 3) + .expect("all parts present") + .iter() + .map(|p| p.part_num) + .collect(); + assert_eq!(collected, vec![1, 2, 3], "CompleteMultipartUpload must include the final part"); + } + + #[test] + fn single_part_upload_submits_one_part() { + let collected = collect_complete_parts(&parts_map(1), 1).expect("single part present"); + assert_eq!(collected.len(), 1, "a single-part object must submit exactly one part, not zero"); + assert_eq!(collected[0].part_num, 1); + } + + #[test] + fn missing_part_is_an_error_not_a_panic() { + let mut m = parts_map(3); + m.remove(&2); + assert!( + collect_complete_parts(&m, 3).is_err(), + "a gap in the parts map must be an error, not a panic" + ); + } +} diff --git a/crates/ecstore/src/services/tier/warm_backend_gcs.rs b/crates/ecstore/src/services/tier/warm_backend_gcs.rs index 425e2507f..fccec129d 100644 --- a/crates/ecstore/src/services/tier/warm_backend_gcs.rs +++ b/crates/ecstore/src/services/tier/warm_backend_gcs.rs @@ -26,6 +26,7 @@ use google_cloud_auth::credentials::Credentials; use google_cloud_auth::credentials::user_account::Builder; use google_cloud_storage as gcs; use google_cloud_storage::client::Storage; +use google_cloud_storage::client::StorageControl; use std::convert::TryFrom; use crate::client::{ @@ -46,6 +47,7 @@ const MIN_PART_SIZE: i64 = 1024 * 1024 * 128; pub struct WarmBackendGCS { pub client: Arc, + pub control: Arc, pub bucket: String, pub prefix: String, pub storage_class: String, @@ -70,15 +72,22 @@ impl WarmBackendGCS { let Ok(client) = Storage::builder() .with_endpoint(conf.endpoint.clone()) - .with_credentials(credentials) + .with_credentials(credentials.clone()) .build() .await else { return Err(std::io::Error::other("Storage::builder error")); }; let client = Arc::new(client); + // Control-plane client: the data-plane `Storage` client cannot delete or list objects; + // delete_object/list_objects live on StorageControl. + let Ok(control) = StorageControl::builder().with_credentials(credentials).build().await else { + return Err(std::io::Error::other("StorageControl::builder error")); + }; + let control = Arc::new(control); Ok(Self { client, + control, bucket: conf.bucket.clone(), prefix: conf.prefix.strip_suffix("/").unwrap_or(&conf.prefix).to_owned(), storage_class: "".to_string(), @@ -125,7 +134,23 @@ impl WarmBackend for WarmBackendGCS { } async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result { - let Ok(mut reader) = self.client.read_object(&self.bucket, &self.get_dest(object)).send().await else { + let mut req = self.client.read_object(&self.bucket, &self.get_dest(object)); + + // Honor the requested byte range so Range GETs on tiered objects return the exact + // interval instead of the whole object (matches the s3/s3sdk/rustfs warm backends). + if opts.start_offset >= 0 && opts.length > 0 { + let offset: u64 = opts + .start_offset + .try_into() + .map_err(|_| std::io::Error::other("invalid range: negative start_offset"))?; + let count: u64 = opts + .length + .try_into() + .map_err(|_| std::io::Error::other("invalid range: negative length"))?; + req = req.set_read_range(google_cloud_storage::model_ext::ReadRange::segment(offset, count)); + } + + let Ok(mut reader) = req.send().await else { return Err(std::io::Error::other("read_object error")); }; let mut contents = Vec::new(); @@ -136,23 +161,33 @@ impl WarmBackend for WarmBackendGCS { } async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error> { - /*self.client - .delete_object() - .set_bucket(&self.bucket) - .set_object(&self.get_dest(object)) - //.set_generation(object.generation) - .send() - .await?;*/ + // gRPC v2 DeleteObject requires the bucket in resource-name form. Without this the + // deleted tiered object was never removed from GCS (empty impl returned Ok), leaking + // remote data forever. + self.control + .delete_object() + .set_bucket(format!("projects/_/buckets/{}", self.bucket)) + .set_object(self.get_dest(object)) + .send() + .await + .map_err(|e| std::io::Error::other(e.to_string()))?; Ok(()) } async fn in_use(&self) -> Result { - /*let result = self.client - .list_objects_v2(&self.bucket, &self.prefix, "", "", SLASH_SEPARATOR, 1) - .await?; + // Scope the listing to this tier's prefix (matching the other warm backends) and only + // need to know whether a single object exists. + let resp = self + .control + .list_objects() + .set_parent(format!("projects/_/buckets/{}", self.bucket)) + .set_prefix(self.prefix.clone()) + .set_page_size(1) + .send() + .await + .map_err(|e| std::io::Error::other(e.to_string()))?; - Ok(result.common_prefixes.len() > 0 || result.contents.len() > 0)*/ - Ok(false) + Ok(!resp.objects.is_empty()) } } diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index cbbdf5094..98f0c6a08 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -666,8 +666,10 @@ impl SetDisks { ..Default::default() }; - result.before.drives = vec![HealDriveInfo::default(); disks.len()]; - result.after.drives = vec![HealDriveInfo::default(); disks.len()]; + // Filled below by pushing one entry per disk while zipping the (index-aligned) `errs`. + // Pre-filling here would double the reported drive list once the push loop runs. + result.before.drives = Vec::with_capacity(disks.len()); + result.after.drives = Vec::with_capacity(disks.len()); let errs = stat_all_dirs(&disks, bucket, object).await; let dangling_object = is_object_dir_dangling(&errs); diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index e002919e4..82f8f9d04 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -6862,13 +6862,20 @@ async fn get_storage_info(disks: &[Option], eps: &[Endpoint]) -> rust } } pub async fn stat_all_dirs(disks: &[Option], bucket: &str, prefix: &str) -> Vec> { - let mut errs = Vec::with_capacity(disks.len()); let mut futures = Vec::with_capacity(disks.len()); - for disk in disks.iter().flatten() { + // Spawn one future per disk slot so the returned vector stays index-aligned with `disks` + // (and therefore with `set_endpoints`). Offline/None disks must yield DiskNotFound in-place + // rather than being skipped, otherwise callers that zip `errs` against the full disks array + // (heal_object_dir) would pair every error with the wrong disk/endpoint whenever any disk is + // offline — and could `make_volume` on the wrong disk. + for disk in disks.iter() { let disk = disk.clone(); let bucket = bucket.to_string(); let prefix = prefix.to_string(); futures.push(tokio::spawn(async move { + let Some(disk) = disk else { + return Some(DiskError::DiskNotFound); + }; match disk.list_dir("", &bucket, &prefix, 1).await { Ok(entries) => { if !entries.is_empty() { @@ -6883,8 +6890,14 @@ pub async fn stat_all_dirs(disks: &[Option], bucket: &str, prefix: &s let results = join_all(futures).await; - for err in results.into_iter().flatten() { - errs.push(err); + // Preserve length/index alignment: a panicked probe becomes a corrupt-state error instead of + // a silently-dropped slot that would re-shift every subsequent index. + let mut errs = Vec::with_capacity(disks.len()); + for res in results.into_iter() { + match res { + Ok(err) => errs.push(err), + Err(join_err) => errs.push(Some(DiskError::other(join_err.to_string()))), + } } errs } @@ -10615,4 +10628,28 @@ mod tests { .expect_err("abandoned-parts check should stay in the upper reconciliation layer"); assert!(matches!(abandoned_err, StorageError::NotImplemented)); } + + #[tokio::test] + async fn stat_all_dirs_returns_index_aligned_vector_for_offline_disks() { + // All-offline set: no real disk I/O needed. Isolates the length/index-alignment contract + // that heal_object_dir depends on when it zips `errs` against the full `disks` array. + let disks: Vec> = vec![None, None, None, None]; + + let errs = stat_all_dirs(&disks, "bucket", "object").await; + + // Before the fix, offline disks contributed no future and the collected vector had length + // 0, so any zip against `disks` paired errors with the wrong disk. After the fix each slot + // is DiskNotFound, index-aligned with `disks`. + assert_eq!( + errs.len(), + disks.len(), + "stat_all_dirs must return one entry per disk slot to stay index-aligned" + ); + for err in &errs { + assert!( + matches!(err, Some(DiskError::DiskNotFound)), + "offline (None) disk slot must map to DiskNotFound in-place, got {err:?}" + ); + } + } } diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 311174706..7f27b5baa 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -290,7 +290,12 @@ impl PriorityHealQueue { }); if displaced.is_some() { - debug_assert_eq!(self.push(request), QueuePushOutcome::Accepted); + // The enqueue side effect must run in ALL builds. Do NOT fold `self.push(request)` + // into `debug_assert_eq!` — in release builds (`debug_assertions` off) the whole + // macro, including its argument expression, is compiled out, which would silently + // drop the new high-priority request after having already evicted a queued item. + let outcome = self.push(request); + debug_assert_eq!(outcome, QueuePushOutcome::Accepted); } displaced @@ -3064,6 +3069,38 @@ mod tests { request } + #[test] + fn test_push_displacing_lower_priority_actually_enqueues_new_request() { + // Regression for the release-build defect where the enqueue side effect lived inside + // `debug_assert_eq!(self.push(request), ...)` and was compiled out under + // `cargo test --release` (debug_assertions off), silently dropping the displacing + // high-priority request while still having evicted a queued item. + // + // Must run with --release to expose the original bug. + let mut queue = PriorityHealQueue::new(); + + let low = bucket_request("victim-bucket", HealPriority::Low, HealRequestSource::Scanner); + assert_eq!(queue.push(low), QueuePushOutcome::Accepted); + assert_eq!(queue.len(), 1); + + let high = bucket_request("admin-bucket", HealPriority::High, HealRequestSource::Admin); + let high_id = high.id.clone(); + assert!(queue.can_displace_lower_priority(high.priority)); + + let displaced = queue + .push_displacing_lower_priority(high) + .expect("a lower-priority item should have been displaced"); + assert_eq!(displaced.priority, HealPriority::Low); + + // The displacing high-priority request must actually be enqueued (pre-fix under + // --release, len() is 0 because self.push(request) was elided with debug_assert_eq!). + assert_eq!(queue.len(), 1, "displacing request must remain enqueued"); + let admitted = queue.pop_next().expect("displacing high-priority request must be enqueued"); + assert_eq!(admitted.priority, HealPriority::High); + assert_eq!(admitted.id, high_id); + assert_eq!(queue.len(), 0); + } + #[test] fn test_priority_queue_ordering() { let mut queue = PriorityHealQueue::new(); diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index d5200f067..e84d6b1d2 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -514,12 +514,16 @@ where } if let Err(err) = self.api.delete_policy_doc(name).await { + // A real backend failure (disk IO, insufficient quorum, etc.) means the on-disk + // policy was NOT removed: propagate the error so callers do not report a phantom + // success and evict a policy that is still persisted (it would reappear on the + // next full IAM reload). if !is_err_no_such_policy(&err) { - self.cache.delete_policy_doc(name, OffsetDateTime::now_utc()); - return Ok(()); + return Err(err); } - return Err(err); + // NoSuchPolicy means the doc is already gone on the backend; treat the delete as + // idempotently successful and fall through to evict any stale cache entry below. } } @@ -2926,4 +2930,39 @@ mod tests { assert_eq!(desc.members, vec!["alice".to_string()]); assert!(desc.updated_at.is_some()); } + + #[tokio::test] + async fn delete_policy_propagates_backend_error_and_keeps_cache() { + // Regression: on the admin delete path (is_from_notify = true), a real backend delete + // failure (here Error::InvalidArgument, standing in for disk IO / insufficient quorum) + // must propagate — NOT be swallowed as Ok(()) while evicting the still-persisted policy. + let cache = build_test_iam_cache(FailingInitialLoadStore); + + let policy = Policy { + id: Default::default(), + version: "2012-10-17".to_string(), + statements: vec![], + }; + let policy_doc = PolicyDoc { + version: 1, + policy, + create_date: Some(OffsetDateTime::now_utc()), + update_date: Some(OffsetDateTime::now_utc()), + }; + cache + .cache + .add_or_update_policy_doc("permissive-policy", &policy_doc, OffsetDateTime::now_utc()); + + let result = cache.delete_policy("permissive-policy", true).await; + + // Pre-fix this returned Ok(()) (phantom success) and evicted the cache entry. + assert!( + result.is_err(), + "delete_policy must surface a real backend delete failure instead of reporting success" + ); + assert!( + cache.cache.snapshot().policy_docs.contains_key("permissive-policy"), + "cache must not evict a policy whose backend delete failed" + ); + } } diff --git a/crates/kms/src/backends/local.rs b/crates/kms/src/backends/local.rs index d2a48c2c8..a82cac7cc 100644 --- a/crates/kms/src/backends/local.rs +++ b/crates/kms/src/backends/local.rs @@ -524,9 +524,10 @@ impl KmsClient for LocalKmsClient { let mut master_key = self.load_master_key(key_id).await?; master_key.status = KeyStatus::Active; - // For simplicity, we'll regenerate key material - // In a real implementation, we'd preserve the original key material - let key_material = generate_key_material(&master_key.algorithm)?; + // Preserve the existing key material. Regenerating it on a pure status change would + // destroy the original master key and make every DEK ever wrapped by it permanently + // undecryptable (silent data loss). + let key_material = self.get_key_material(key_id).await?; self.save_master_key(&master_key, &key_material).await?; // Update cache @@ -543,7 +544,9 @@ impl KmsClient for LocalKmsClient { let mut master_key = self.load_master_key(key_id).await?; master_key.status = KeyStatus::Disabled; - let key_material = generate_key_material(&master_key.algorithm)?; + // Preserve the existing key material (see enable_key): a status change must never + // regenerate the master key, or every DEK wrapped by it becomes undecryptable. + let key_material = self.get_key_material(key_id).await?; self.save_master_key(&master_key, &key_material).await?; // Update cache @@ -565,7 +568,10 @@ impl KmsClient for LocalKmsClient { let mut master_key = self.load_master_key(key_id).await?; master_key.status = KeyStatus::PendingDeletion; - let key_material = generate_key_material(&master_key.algorithm)?; + // 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 + // can no longer decrypt existing data. + let key_material = self.get_key_material(key_id).await?; self.save_master_key(&master_key, &key_material).await?; // Update cache @@ -582,7 +588,9 @@ impl KmsClient for LocalKmsClient { let mut master_key = self.load_master_key(key_id).await?; master_key.status = KeyStatus::Active; - let key_material = generate_key_material(&master_key.algorithm)?; + // 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. + let key_material = self.get_key_material(key_id).await?; self.save_master_key(&master_key, &key_material).await?; // Update cache @@ -1020,6 +1028,41 @@ mod tests { assert_eq!(decrypted, data_key.plaintext.clone().expect("No plaintext")); } + #[tokio::test] + async fn key_state_transitions_preserve_master_key_material() { + // Regression: enable/disable/schedule_deletion/cancel_deletion previously regenerated the + // master key material on a pure status change, permanently destroying the ability to + // decrypt any DEK wrapped by that key. A status cycle must preserve the material. + let (client, _temp_dir) = create_test_client().await; + + let key_id = "state-cycle-key"; + client.create_key(key_id, "AES_256", None).await.expect("create"); + + let request = GenerateKeyRequest::new(key_id.to_string(), "AES_256".to_string()) + .with_context("bucket".to_string(), "b".to_string()); + let data_key = client.generate_data_key(&request, None).await.expect("generate data key"); + let ciphertext = data_key.ciphertext.clone(); + let plaintext = data_key.plaintext.clone().expect("no plaintext"); + + // Cycle through every status-changing method the fix touches. + client.disable_key(key_id, None).await.expect("disable"); + client.enable_key(key_id, None).await.expect("enable"); + client + .schedule_key_deletion(key_id, 7, None) + .await + .expect("schedule deletion"); + client.cancel_key_deletion(key_id, None).await.expect("cancel deletion"); + + // Pre-fix, each of those regenerated the master key, so this unwrap fails with an AEAD + // error. Post-fix, the original material is preserved and the DEK still decrypts. + let decrypt_request = DecryptRequest::new(ciphertext).with_context("bucket".to_string(), "b".to_string()); + let decrypted = client + .decrypt(&decrypt_request, None) + .await + .expect("DEK must still decrypt after status transitions"); + assert_eq!(decrypted, plaintext, "master key material must survive status transitions"); + } + #[tokio::test] async fn test_encryption_operations() { let (client, _temp_dir) = create_test_client().await; diff --git a/crates/kms/src/backends/vault.rs b/crates/kms/src/backends/vault.rs index 3983c304c..ebdfc2eb6 100644 --- a/crates/kms/src/backends/vault.rs +++ b/crates/kms/src/backends/vault.rs @@ -147,28 +147,29 @@ impl VaultKmsClient { let key_material = match self.decrypt_key_material(&key_data.encrypted_key_material).await { Ok(km) => km, Err(e) => { - warn!(key_id, error = %e, "Vault KMS key material decrypt failed; regenerating"); - let new_key_material = generate_key_material(&key_data.algorithm)?; - key_data.encrypted_key_material = self.encrypt_key_material(&new_key_material).await?; - // Store the updated key data back to Vault - self.store_key_data(key_id, &key_data).await?; - return Ok(new_key_material); + // Never regenerate/overwrite the master key on a decrypt failure: that would + // destroy the original material and make every DEK wrapped by this key + // permanently undecryptable. Surface the error so the read fails recoverably + // instead of causing silent data loss. + warn!(key_id, error = %e, "Vault KMS key material could not be decoded"); + return Err(KmsError::cryptographic_error( + "decrypt", + format!("Stored key material for {key_id} is corrupted: {e}"), + )); } }; - // Validate key material length (should be 32 bytes for AES-256) + // Validate key material length (should be 32 bytes for AES-256). if key_material.len() != 32 { - // Try to fix: generate new key material if length is wrong - warn!( - "Key {} has invalid key material length ({} bytes), generating new key material", - key_id, - key_material.len() - ); - let new_key_material = generate_key_material(&key_data.algorithm)?; - key_data.encrypted_key_material = self.encrypt_key_material(&new_key_material).await?; - // Store the updated key data back to Vault - self.store_key_data(key_id, &key_data).await?; - return Ok(new_key_material); + // As above: do not overwrite the stored key. Report the fault instead. + warn!(key_id, len = key_material.len(), "Vault KMS key material has invalid length"); + return Err(KmsError::cryptographic_error( + "decrypt", + format!( + "Stored key material for {key_id} has invalid length ({} bytes, expected 32)", + key_material.len() + ), + )); } Ok(key_material) @@ -812,6 +813,11 @@ impl KmsBackend for VaultKmsBackend { key_metadata.key_state = KeyState::Enabled; key_metadata.deletion_date = None; + // Persist the reset state back to Vault. Without this the key stays PendingDeletion in + // storage and would still be reaped, so we must fail the request if the write fails + // rather than report a false success. + self.update_key_metadata_in_storage(key_id, &key_metadata).await?; + Ok(CancelKeyDeletionResponse { key_id: key_id.clone(), key_metadata, @@ -877,4 +883,94 @@ mod tests { // Test health check client.health_check().await.expect("Health check failed"); } + + fn integration_vault_config() -> VaultConfig { + VaultConfig { + address: "http://127.0.0.1:8200".to_string(), + auth_method: VaultAuthMethod::Token { + token: "dev-only-token".to_string(), + }, + kv_mount: "secret".to_string(), + key_path_prefix: "rustfs/kms/keys".to_string(), + mount_path: "transit".to_string(), + namespace: None, + tls: None, + } + } + + #[tokio::test] + #[ignore] // Requires a running Vault instance (dev mode) + async fn test_corrupted_key_material_does_not_regenerate() { + // Regression: get_key_material previously "self-healed" a decrypt/length failure by + // minting a fresh random master key and overwriting the stored value — destroying the + // original key and making every DEK wrapped by it permanently undecryptable. + let client = VaultKmsClient::new(integration_vault_config()).await.expect("client"); + + let key_id = format!("corrupt-{}", uuid::Uuid::new_v4()); + client.create_key(&key_id, "AES_256", None).await.expect("create"); + + // Corrupt the stored material to an invalid base64 string. + let mut key_data = client.get_key_data(&key_id).await.expect("read"); + key_data.encrypted_key_material = "!!!not-base64!!!".to_string(); + client.store_key_data(&key_id, &key_data).await.expect("store corrupt"); + + // Reading the material must now ERROR, not silently regenerate + overwrite. + assert!( + client.get_key_material(&key_id).await.is_err(), + "corrupted key material must yield an error, not a fresh key" + ); + + // And the stored (corrupted) material must be UNCHANGED. + let after = client.get_key_data(&key_id).await.expect("reread"); + assert_eq!( + after.encrypted_key_material, "!!!not-base64!!!", + "get_key_material must not overwrite stored master key material on failure" + ); + } + + #[tokio::test] + #[ignore] // Requires a running Vault instance (dev mode) + async fn test_vault_cancel_key_deletion_persists_state() { + use crate::config::{BackendConfig, KmsConfig}; + use crate::types::{CancelKeyDeletionRequest, CreateKeyRequest, DeleteKeyRequest, KeyStatus, KeyUsage}; + + let kms_config = KmsConfig { + backend_config: BackendConfig::VaultKv2(Box::new(integration_vault_config())), + ..Default::default() + }; + let backend = VaultKmsBackend::new(kms_config).await.expect("backend"); + + let key_id = format!("cancel-persist-{}", uuid::Uuid::new_v4()); + backend + .create_key(CreateKeyRequest { + key_name: Some(key_id.clone()), + key_usage: KeyUsage::EncryptDecrypt, + ..Default::default() + }) + .await + .expect("create"); + + backend + .delete_key(DeleteKeyRequest { + key_id: key_id.clone(), + pending_window_in_days: Some(7), + force_immediate: Some(false), + }) + .await + .expect("schedule delete"); + + backend + .cancel_key_deletion(CancelKeyDeletionRequest { key_id: key_id.clone() }) + .await + .expect("cancel"); + + // Re-read the PERSISTED state from Vault. Before the fix, storage still held + // PendingDeletion because cancel only mutated the response, never wrote back. + let persisted = backend.client.get_key_data(&key_id).await.expect("reread"); + assert_eq!( + persisted.status, + KeyStatus::Active, + "cancel_key_deletion must persist Active status to Vault, not only mutate the response" + ); + } } diff --git a/crates/rio/src/compress_reader.rs b/crates/rio/src/compress_reader.rs index 418373a89..1c5f05b86 100644 --- a/crates/rio/src/compress_reader.rs +++ b/crates/rio/src/compress_reader.rs @@ -304,7 +304,20 @@ where } } let compressed_buf = &this.compressed_buf[..*this.compressed_len]; - let (uncompress_len, uvarint) = uvarint(&compressed_buf[0..16]); + // `compressed_buf`'s length comes from the untrusted 24-bit header length field, so it + // can be shorter than 16 bytes. `uvarint` is safe on any slice length (reads at most 10 + // bytes and stops at the terminator), so pass the whole slice instead of a fixed + // `[0..16]` index that panics on corrupted/truncated blocks shorter than 16 bytes. + let (uncompress_len, uvarint) = uvarint(compressed_buf); + // Reject a length prefix that could not be decoded: `uvarint <= 0` means the varint was + // empty/unterminated (0) or overflowed (negative — as usize it would index far past the + // buffer and panic the slice below). The `> len` bound is belt-and-suspenders (uvarint's + // positive return is always <= buf.len()) but keeps the slice panic-free regardless. + if uvarint <= 0 || uvarint as usize > compressed_buf.len() { + *this.compressed_read = 0; + *this.compressed_len = 0; + return Poll::Ready(Err(io::Error::new(io::ErrorKind::InvalidData, "Invalid compressed block length prefix"))); + } let compressed_data = &compressed_buf[uvarint as usize..]; let decompressed = if typ == COMPRESS_TYPE_COMPRESSED { match decompress_block(compressed_data, *this.compression_algorithm) { @@ -479,4 +492,53 @@ mod tests { assert_eq!(&decompressed, &data); } + + // Regression: a corrupted block whose 24-bit length field is < 16 must not panic. + // Header layout (HEADER_LEN = 8): [type, len_lo, len_mid, len_hi, crc0..crc3], then `len` + // bytes of block body. Pre-fix, poll_read sliced `compressed_buf[0..16]` unconditionally, + // panicking with "range end index 16 out of range for slice of length N" when N < 16. + #[tokio::test] + async fn test_decompress_reader_short_block_no_panic() { + let len: usize = 3; + let mut input = vec![ + COMPRESS_TYPE_COMPRESSED, + (len & 0xFF) as u8, + ((len >> 8) & 0xFF) as u8, + ((len >> 16) & 0xFF) as u8, + ]; + input.extend_from_slice(&[0u8; 4]); // bogus CRC + // Body: a uvarint claiming uncompressed length = 127, followed by 2 bytes that are not + // a valid compressed stream — post-fix this must surface as a clean InvalidData error. + input.extend_from_slice(&[0x7f, 0xAB, 0xCD]); + + let mut decompress_reader = DecompressReader::new(Cursor::new(input), CompressionAlgorithm::default()); + let mut out = Vec::new(); + let res = decompress_reader.read_to_end(&mut out).await; + assert!(res.is_err(), "corrupted short block must return an error, not panic or succeed"); + assert_eq!(res.unwrap_err().kind(), std::io::ErrorKind::InvalidData); + } + + // Directly exercises the length-prefix guard: an unterminated varint (all continuation bytes) + // makes `uvarint` return 0, which must be rejected as an invalid length prefix. + #[tokio::test] + async fn test_decompress_reader_unterminated_length_prefix_is_rejected() { + let len: usize = 3; + let mut input = vec![ + COMPRESS_TYPE_COMPRESSED, + (len & 0xFF) as u8, + ((len >> 8) & 0xFF) as u8, + ((len >> 16) & 0xFF) as u8, + ]; + input.extend_from_slice(&[0u8; 4]); // bogus CRC + input.extend_from_slice(&[0x80, 0x80, 0x80]); // 3 continuation bytes, no terminator + + let mut decompress_reader = DecompressReader::new(Cursor::new(input), CompressionAlgorithm::default()); + let mut out = Vec::new(); + let err = decompress_reader + .read_to_end(&mut out) + .await + .expect_err("unterminated length prefix must error"); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidData); + assert!(err.to_string().contains("length prefix"), "got: {err}"); + } } diff --git a/crates/rio/src/encrypt_reader.rs b/crates/rio/src/encrypt_reader.rs index 1197ecfda..03ef64650 100644 --- a/crates/rio/src/encrypt_reader.rs +++ b/crates/rio/src/encrypt_reader.rs @@ -391,7 +391,19 @@ where } let ciphertext_buf = &this.ciphertext_buf[..*this.ciphertext_len]; - let (plaintext_len, uvarint_len) = rustfs_utils::uvarint(&ciphertext_buf[0..16]); + // `ciphertext_buf`'s length derives from the untrusted 24-bit header length field, so + // it can be shorter than 16 bytes. `uvarint` is safe on any slice length, so pass the + // whole slice instead of a fixed `[0..16]` index that panics on corrupted/truncated + // blocks shorter than 16 bytes. + // `uvarint_len <= 0` means the length varint was empty/unterminated (0) or overflowed + // (negative — as usize it would index far past the buffer). The `> len` bound is + // belt-and-suspenders (a positive return is always <= buf.len()). + let (plaintext_len, uvarint_len) = rustfs_utils::uvarint(ciphertext_buf); + if uvarint_len <= 0 || uvarint_len as usize > ciphertext_buf.len() { + *this.ciphertext_read = 0; + *this.ciphertext_len = 0; + return Poll::Ready(Err(Error::new(std::io::ErrorKind::InvalidData, "Invalid encrypted block length prefix"))); + } let ciphertext = &ciphertext_buf[uvarint_len as usize..]; let block_nonce = derive_block_nonce(this.current_nonce_base, *this.block_index); let nonce = Nonce::try_from(block_nonce.as_slice()).map_err(|_| Error::other("invalid nonce length"))?; @@ -1007,4 +1019,30 @@ mod tests { assert_eq!(decrypted, expected); } + + // Regression: a corrupted block header whose length yields a payload shorter than 16 bytes + // must not panic. Header (8 bytes): [typ, len_lo, len_mid, len_hi, crc0..crc3]; payload is + // `len - 4` bytes. Pre-fix, poll_read sliced `ciphertext_buf[0..16]` unconditionally, + // panicking with "range end index 16 out of range for slice of length N" when N < 16. + #[tokio::test] + async fn test_decrypt_reader_short_block_no_panic() { + let key = [0u8; 32]; + let nonce = [0u8; 12]; + + // len = 8 -> payload_len = 4 (< 16). Provide exactly 4 payload bytes. + let len: usize = 8; + let mut input = vec![ + 0x00u8, // typ (regular block) + (len & 0xFF) as u8, + ((len >> 8) & 0xFF) as u8, + ((len >> 16) & 0xFF) as u8, + ]; + input.extend_from_slice(&[0u8; 4]); // crc (unused before the panic site) + input.extend_from_slice(&[0x01u8, 0x02, 0x03, 0x04]); // 4-byte payload + + let mut decrypt_reader = DecryptReader::new(Cursor::new(input), key, nonce); + let mut out = Vec::new(); + let res = decrypt_reader.read_to_end(&mut out).await; + assert!(res.is_err(), "corrupted short encrypted block must return an error, not panic"); + } } diff --git a/crates/utils/src/egress.rs b/crates/utils/src/egress.rs index 495de1797..a04adbf36 100644 --- a/crates/utils/src/egress.rs +++ b/crates/utils/src/egress.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::fmt; -use std::net::{IpAddr, Ipv4Addr}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use url::Url; #[derive(Debug, Clone, PartialEq, Eq)] @@ -59,6 +59,36 @@ pub fn validate_outbound_url(url: &Url) -> Result<(), OutboundUrlError> { } fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> { + // Reject pure-IPv6 special forms first, before any IPv4 normalization. ::1 (loopback) and :: + // (unspecified) are technically IPv4-compatible forms too, so normalizing first would map + // them to a harmless-looking 0.0.0.1 / 0.0.0.0 and let them through. + if let IpAddr::V6(v6) = ip { + if v6.is_loopback() { + return Err("loopback address"); + } + if v6.is_unspecified() { + return Err("unspecified address"); + } + if v6.is_unicast_link_local() { + return Err("link-local address"); + } + if v6.is_unique_local() { + return Err("private address"); + } + } + + // Normalize IPv4-mapped (::ffff:a.b.c.d) AND IPv4-compatible (::a.b.c.d) IPv6 addresses to + // their embedded IPv4 so the IPv4 rules below apply. The std is_* checks on the IPv6 variant + // never inspect the embedded IPv4, so without this an attacker bypasses the guard with e.g. + // ::ffff:127.0.0.1, ::127.0.0.1 (loopback) or ::169.254.169.254 (cloud metadata service). + let ip = match ip { + IpAddr::V6(v6) => match embedded_ipv4(v6) { + Some(v4) => IpAddr::V4(v4), + None => IpAddr::V6(v6), + }, + other => other, + }; + if ip.is_unspecified() { return Err("unspecified address"); } @@ -79,22 +109,34 @@ fn validate_outbound_ip(ip: IpAddr) -> Result<(), &'static str> { return Err("private address"); } } - IpAddr::V6(ipv6) => { - if ipv6.is_loopback() { - return Err("loopback address"); - } - if ipv6.is_unicast_link_local() { - return Err("link-local address"); - } - if ipv6.is_unique_local() { - return Err("private address"); - } - } + // Genuine IPv6 (no embedded IPv4) was already classified above. + IpAddr::V6(_) => {} } Ok(()) } +/// Extract the embedded IPv4 from an IPv4-mapped (`::ffff:a.b.c.d`) or IPv4-compatible +/// (`::a.b.c.d`) IPv6 address. The pure-IPv6 specials `::` and `::1` are rejected by the caller +/// before this runs, so returning `None` here means a genuine IPv6 host. +fn embedded_ipv4(v6: Ipv6Addr) -> Option { + if let Some(v4) = v6.to_ipv4_mapped() { + return Some(v4); + } + // IPv4-compatible: the top 96 bits are zero and the low 32 bits carry the IPv4. + let segs = v6.segments(); + if segs[0..6] == [0, 0, 0, 0, 0, 0] { + let hi = segs[6].to_be_bytes(); + let lo = segs[7].to_be_bytes(); + let v4 = Ipv4Addr::new(hi[0], hi[1], lo[0], lo[1]); + // `::` and `::1` are already handled by the caller; anything else is a real embedded v4. + if !v4.is_unspecified() && v4 != Ipv4Addr::new(0, 0, 0, 1) { + return Some(v4); + } + } + None +} + #[cfg(test)] mod tests { use super::{OutboundUrlError, validate_outbound_url}; @@ -170,4 +212,98 @@ mod tests { } )); } + + #[test] + fn validate_outbound_url_rejects_ipv4_mapped_loopback() { + let url = Url::parse("http://[::ffff:127.0.0.1]/webhook").expect("mapped loopback URL should parse"); + let err = validate_outbound_url(&url).expect_err("IPv4-mapped loopback should be rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "loopback address", + .. + } + )); + } + + #[test] + fn validate_outbound_url_rejects_ipv4_mapped_private() { + let url = Url::parse("http://[::ffff:10.0.0.5]/webhook").expect("mapped private URL should parse"); + let err = validate_outbound_url(&url).expect_err("IPv4-mapped private should be rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "private address", + .. + } + )); + } + + #[test] + fn validate_outbound_url_rejects_ipv4_mapped_metadata_endpoint() { + let url = Url::parse("http://[::ffff:169.254.169.254]/latest/meta-data").expect("mapped metadata URL should parse"); + let err = validate_outbound_url(&url).expect_err("IPv4-mapped metadata endpoint should be rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "metadata endpoint", + .. + } + )); + } + + #[test] + fn validate_outbound_url_still_allows_public_ipv6() { + // Pure public IPv6 (Google DNS) must remain allowed after normalization. + let url = Url::parse("https://[2001:4860:4860::8888]/webhook").expect("public IPv6 URL should parse"); + assert!(validate_outbound_url(&url).is_ok()); + } + + #[test] + fn validate_outbound_url_rejects_ipv4_compatible_loopback() { + // IPv4-compatible form ::a.b.c.d (deprecated but still routable) must also be caught. + let url = Url::parse("http://[::127.0.0.1]/webhook").expect("compatible loopback URL should parse"); + let err = validate_outbound_url(&url).expect_err("IPv4-compatible loopback should be rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "loopback address", + .. + } + )); + } + + #[test] + fn validate_outbound_url_rejects_ipv4_compatible_metadata_endpoint() { + let url = Url::parse("http://[::169.254.169.254]/latest/meta-data").expect("compatible metadata URL should parse"); + let err = validate_outbound_url(&url).expect_err("IPv4-compatible metadata endpoint should be rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "metadata endpoint", + .. + } + )); + } + + #[test] + fn validate_outbound_url_rejects_ipv6_loopback_and_unspecified() { + // ::1 / :: must stay rejected even though they look like IPv4-compatible forms. + let err = validate_outbound_url(&Url::parse("http://[::1]/x").unwrap()).expect_err("::1 rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "loopback address", + .. + } + )); + let err = validate_outbound_url(&Url::parse("http://[::]/x").unwrap()).expect_err(":: rejected"); + assert!(matches!( + err, + OutboundUrlError::ForbiddenHost { + reason: "unspecified address", + .. + } + )); + } } diff --git a/rustfs/src/admin/handlers/bucket_meta.rs b/rustfs/src/admin/handlers/bucket_meta.rs index fac7179b2..3cf50a50b 100644 --- a/rustfs/src/admin/handlers/bucket_meta.rs +++ b/rustfs/src/admin/handlers/bucket_meta.rs @@ -827,6 +827,32 @@ impl Operation for ImportBucketMetadata { } } + // Persist the assembled metadata to disk. Prior to this, the import only mutated the + // in-memory `bucket_metadatas` map and returned 200, silently dropping every imported + // config. `metadata_sys::update` loads the on-disk metadata, overwrites the given config + // field and saves it, preserving any configs not present in the import archive. + for (bucket_name, metadata) in &bucket_metadatas { + for (config_file, data) in imported_configs_to_persist(metadata) { + if let Err(e) = metadata_sys::update(bucket_name, config_file, data).await { + warn!( + event = EVENT_ADMIN_BUCKET_META_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_BUCKET_META, + action = "import_bucket_metadata", + result = "config_persist_failed", + bucket = %bucket_name, + config_name = %config_file, + error = %e, + "admin bucket meta state" + ); + return Err(s3_error!( + InternalError, + "failed to persist imported bucket metadata for {bucket_name}/{config_file}: {e}" + )); + } + } + } + // TODO: site replication notify let mut header = HeaderMap::new(); @@ -835,3 +861,62 @@ impl Operation for ImportBucketMetadata { Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header)) } } + +/// The `(config_file, data)` pairs to persist for an imported bucket's metadata: every non-empty +/// config field keyed by its on-disk config-file name, as owned data ready for +/// `metadata_sys::update`. Empty fields are skipped so an import never overwrites an existing +/// on-disk config with an empty payload. Shared by [`import_bucket_metadata`] and its tests so both +/// exercise the same mapping. +fn imported_configs_to_persist(metadata: &BucketMetadata) -> Vec<(&'static str, Vec)> { + let configs: [(&'static str, &Vec); 10] = [ + (BUCKET_POLICY_CONFIG, &metadata.policy_config_json), + (BUCKET_NOTIFICATION_CONFIG, &metadata.notification_config_xml), + (BUCKET_LIFECYCLE_CONFIG, &metadata.lifecycle_config_xml), + (BUCKET_SSECONFIG, &metadata.encryption_config_xml), + (BUCKET_TAGGING_CONFIG, &metadata.tagging_config_xml), + (BUCKET_QUOTA_CONFIG_FILE, &metadata.quota_config_json), + (OBJECT_LOCK_CONFIG, &metadata.object_lock_config_xml), + (BUCKET_VERSIONING_CONFIG, &metadata.versioning_config_xml), + (BUCKET_REPLICATION_CONFIG, &metadata.replication_config_xml), + (BUCKET_TARGETS_FILE, &metadata.bucket_targets_config_json), + ]; + configs + .into_iter() + .filter(|(_, d)| !d.is_empty()) + .map(|(name, d)| (name, d.clone())) + .collect() +} + +#[cfg(test)] +mod import_persist_tests { + use super::*; + + #[test] + fn imported_versioning_and_policy_are_scheduled_for_persistence() { + // State the second pass builds in memory after importing a versioning + policy config. + let mut metadata = BucketMetadata::new("restored-bucket"); + metadata.versioning_config_xml = b"Enabled".to_vec(); + metadata.policy_config_json = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(); + + let plan = imported_configs_to_persist(&metadata); + + // The bug: the old handler produced zero persistence calls (mutated memory, returned 200). + assert_eq!(plan.len(), 2, "both imported configs must be persisted, got {plan:?}"); + assert!( + plan.iter() + .any(|(n, d)| *n == BUCKET_VERSIONING_CONFIG && d == &metadata.versioning_config_xml) + ); + assert!( + plan.iter() + .any(|(n, d)| *n == BUCKET_POLICY_CONFIG && d == &metadata.policy_config_json) + ); + } + + #[test] + fn empty_configs_are_not_persisted() { + // A freshly-created metadata with no imported configs must schedule nothing, so import + // never overwrites existing on-disk configs with empty payloads. + let metadata = BucketMetadata::new("untouched-bucket"); + assert!(imported_configs_to_persist(&metadata).is_empty()); + } +} diff --git a/rustfs/src/admin/handlers/sts.rs b/rustfs/src/admin/handlers/sts.rs index b8ef631ba..e4bda352b 100644 --- a/rustfs/src/admin/handlers/sts.rs +++ b/rustfs/src/admin/handlers/sts.rs @@ -54,6 +54,27 @@ const ASSUME_ROLE_ACTION: &str = "AssumeRole"; const ASSUME_ROLE_WITH_WEB_IDENTITY_ACTION: &str = "AssumeRoleWithWebIdentity"; const ASSUME_ROLE_VERSION: &str = "2011-06-15"; +/// Default STS temporary credential lifetime (seconds) when the client omits DurationSeconds. +const STS_DEFAULT_DURATION_SECS: usize = 3600; +/// Minimum STS temporary credential lifetime (seconds), matching AWS/MinIO (15 minutes). +const STS_MIN_DURATION_SECS: usize = 900; +/// Maximum STS temporary credential lifetime (seconds), matching AWS/MinIO AssumeRole (12 hours). +const STS_MAX_DURATION_SECS: usize = 43200; + +/// Clamp the client-supplied DurationSeconds into the allowed STS window. +/// +/// A value of 0 (unset) falls back to the default; any other value is clamped into +/// `[STS_MIN_DURATION_SECS, STS_MAX_DURATION_SECS]`. This prevents callers from minting +/// near-permanent temporary credentials and keeps the standard AssumeRole path consistent +/// with the AssumeRoleWithWebIdentity path. +fn clamp_assume_role_duration(duration_seconds: usize) -> usize { + if duration_seconds == 0 { + STS_DEFAULT_DURATION_SECS + } else { + duration_seconds.clamp(STS_MIN_DURATION_SECS, STS_MAX_DURATION_SECS) + } +} + fn has_identity_authorization_context(policies: &[String], groups: &[String]) -> bool { !policies.is_empty() || !groups.is_empty() } @@ -216,17 +237,13 @@ async fn handle_assume_role( populate_session_policy(&mut claims, &body.policy)?; - let exp = { - if body.duration_seconds > 0 { - body.duration_seconds - } else { - 3600 - } - }; + let exp = clamp_assume_role_duration(body.duration_seconds); claims.insert( "exp".to_string(), - Value::Number(serde_json::Number::from(OffsetDateTime::now_utc().unix_timestamp() + exp as i64)), + Value::Number(serde_json::Number::from( + OffsetDateTime::now_utc().unix_timestamp().saturating_add(exp as i64), + )), ); claims.insert("parent".to_string(), Value::String(cred.access_key.clone())); @@ -543,6 +560,24 @@ mod tests { assert_eq!(clamp(999999), 43200); // clamped to max } + #[test] + fn test_assume_role_duration_is_clamped_to_max() { + // Regression: the standard AssumeRole path previously used the raw client-supplied + // DurationSeconds with no upper bound, allowing near-permanent temporary credentials. + let ten_years_secs: usize = 315_360_000; + assert_eq!(clamp_assume_role_duration(ten_years_secs), STS_MAX_DURATION_SECS); + assert_eq!(STS_MAX_DURATION_SECS, 43200); + assert_eq!(clamp_assume_role_duration(0), STS_DEFAULT_DURATION_SECS); + assert_eq!(clamp_assume_role_duration(60), STS_MIN_DURATION_SECS); + assert_eq!(clamp_assume_role_duration(3600), 3600); + assert_eq!(clamp_assume_role_duration(43200), 43200); + + // The exp timestamp derived from a huge duration must not exceed now + 12h. + let now = OffsetDateTime::now_utc().unix_timestamp(); + let exp = now.saturating_add(clamp_assume_role_duration(ten_years_secs) as i64); + assert!(exp - now <= STS_MAX_DURATION_SECS as i64); + } + #[test] fn test_has_identity_authorization_context() { let empty: Vec = vec![];