mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-03 11:57:43 +00:00
fix: 12 P1 reliability/security defects from the full-repo audit (backlog#806) (#4256)
* fix(rio): reject corrupted short compressed/encrypted blocks instead of panicking DecompressReader::poll_read and DecryptReader::poll_read sliced the block body with a fixed `[0..16]` index to read the length varint. The body length comes from an untrusted 24-bit header field, so a corrupted/truncated block shorter than 16 bytes made the slice panic and crash the request task — a read-path DoS on GET of tiered/corrupted data. Pass the whole (arbitrary-length-safe) slice to uvarint and reject a non-positive or out-of-range length prefix with InvalidData. Adds a repro test for each reader; all existing round-trip tests still pass. Refs rustfs/backlog#812 * fix(utils): close SSRF bypass via IPv4-mapped IPv6 addresses validate_outbound_ip branched on the IpAddr variant, and the V6 branch's is_loopback/is_unicast_link_local/is_unique_local checks never inspect the embedded IPv4 of an IPv4-mapped address (::ffff:a.b.c.d). The metadata guard also only matched the plain V4 169.254.169.254. So ::ffff:127.0.0.1, ::ffff:10.0.0.5 and ::ffff:169.254.169.254 all passed the outbound guard, letting an attacker reach loopback/private/metadata endpoints. Normalize IPv4-mapped IPv6 to its embedded IPv4 (via to_ipv4_mapped, which matches only the true mapped form) before classification. Adds reject tests for mapped loopback/private/metadata and an allow test for public IPv6. Refs rustfs/backlog#813 * fix(ecstore): streaming last-part loss, GCS tier Range/remove, stat_all_dirs alignment Four confirmed data-reliability defects: - put_object_multipart_stream: the CompleteMultipartUpload part-collection loop used exclusive `1..total_parts_count`, dropping the final part (and collecting zero parts for a single-part object) — silently truncating the completed object. Extracted collect_complete_parts (1..=total_parts_count) with unit tests. - GCS warm backend get() ignored the requested byte range, returning the whole object for a Range GET; now applies ReadRange::segment like the other backends. - GCS warm backend remove() was an empty stub, so deleting a tiered object left it on GCS forever; now deletes via StorageControl (added a control-plane client), and in_use() actually lists (prefix-scoped) instead of always returning false. - stat_all_dirs skipped None disk slots and dropped JoinErrors, returning a compressed, misaligned error vector; heal_object_dir then zipped it against the full disks array and could make_volume on the WRONG disk. Now returns one index-aligned entry per slot (None -> DiskNotFound), and heal no longer pre-fills the drive report (which would double it). Added an alignment test. Refs rustfs/backlog#807 * fix(kms): stop Vault backend from destroying/reviving keys on failure Two confirmed key-safety defects in the Vault KV2 backend: - get_key_material() 'self-healed' a decrypt or wrong-length failure by minting a fresh random master key and overwriting the stored value. That destroys the original key material, making every DEK ever wrapped by it permanently undecryptable. Decryption must never mutate the stored key: both branches now return a cryptographic_error instead. (The empty-material bootstrap path, which only fills a never-initialized key, is intentionally left intact.) - cancel_key_deletion() reset key_state to Enabled only in the returned response and never persisted it, so the key stayed PendingDeletion in storage and would still be reaped. It now writes the state back via update_key_metadata_in_storage and fails the request if the write fails. Adds ignored (Vault-requiring) integration tests documenting both behaviours. The third item (VaultTransit key state only in memory -> revived as Enabled after restart) is deferred: a fail-closed guard would break restart availability for all transit keys; the correct fix needs a persistent metadata store + Vault integration testing. Tracked in rustfs/backlog#808. Refs rustfs/backlog#808 * fix(admin): clamp STS AssumeRole duration; persist ImportBucketMetadata to disk Two confirmed admin-API defects: - Standard AssumeRole used the raw client-supplied DurationSeconds with no upper bound, so a caller could mint near-permanent temporary credentials. Clamp it to the AWS/MinIO STS window [900, 43200] (with 0 -> default 3600) via a shared clamp_assume_role_duration helper, and build the exp claim with saturating_add. This matches the existing AssumeRoleWithWebIdentity path. - ImportBucketMetadata only mutated an in-memory map and returned 200, silently dropping every imported config. It now persists each non-empty config via metadata_sys::update (which merges onto existing on-disk metadata) and returns InternalError if a write fails. Mapping extracted to imported_configs_to_persist with unit tests. Refs rustfs/backlog#809 * fix(heal): enqueue displacing request in release builds push_displacing_lower_priority folded the real enqueue call into debug_assert_eq!(self.push(request), Accepted). In release builds (debug_assertions off) the whole macro — including its argument — is compiled out, so after evicting a lower-priority queued item the new high-priority request was silently dropped and never healed. Hoist self.push(request) out of the assertion so the side effect runs in all builds. Adds a --release regression test. Refs rustfs/backlog#811 * fix(iam): propagate real delete_policy backend errors instead of swallowing them delete_policy's is_from_notify path had its error handling inverted: a real backend failure (disk IO / insufficient quorum) evicted the cache and returned Ok(()), reporting a phantom success while policy.json survived on disk (to be reloaded on the next full IAM reload); NoSuchPolicy — which should be idempotent success — returned Err. Propagate real errors and let NoSuchPolicy fall through to the idempotent cache-evict + Ok, matching delete_user / the notification handler in the same file. Adds a backend-error-injection regression test. Refs rustfs/backlog#810 * fix(utils): also normalize IPv4-compatible IPv6 in the SSRF guard The initial fix only unwrapped IPv4-mapped (::ffff:a.b.c.d) addresses; the deprecated IPv4-compatible form (::a.b.c.d, e.g. ::127.0.0.1 / ::169.254.169.254) still bypassed the guard. Reject pure-IPv6 specials (::, ::1, fe80::, fc00::) first, then normalize BOTH embedded-IPv4 forms before the IPv4 rules. Adds tests for compatible-form loopback/metadata and confirms ::1 / :: stay rejected. Found by adversarial review of the initial fix. Refs rustfs/backlog#813 * fix(ecstore): fix the same last-part loss in the parallel streaming path put_object_multipart_stream_parallel had the identical off-by-one (1..total_parts_count) that truncated the last part / produced zero parts for a single-part upload — reachable when concurrent stream parts are enabled. Reuse collect_complete_parts, which now returns an error instead of panicking on a gap in the parts map. Adds a missing-part error test. Found by adversarial review of the initial fix. Refs rustfs/backlog#807 * fix(kms): local backend must preserve key material on status change LocalKmsClient (the default KMS backend) regenerated the master key material on enable_key/disable_key/schedule_key_deletion/cancel_key_deletion — a pure status change. A single disable+enable cycle therefore destroyed the original key, making every DEK ever wrapped by it permanently undecryptable (silent data loss, no network needed). Preserve the existing material via get_key_material and re-save with only the status changed. Adds a hermetic regression test that wraps a DEK, cycles all four status methods, and asserts the DEK still decrypts. Found by adversarial review of the Vault fix. Refs rustfs/backlog#808 * test(rio): cover the length-prefix guard; correct its comment Add a DecompressReader test that feeds an unterminated length varint so uvarint returns 0 and the new guard (not the downstream codec) produces the InvalidData error, and reword the guard comment which overclaimed that the > len bound prevents a reachable panic (it is belt-and-suspenders). No behavior change. Found by adversarial review. Refs rustfs/backlog#812 * test(rio): build test block headers via vec! to satisfy clippy The new corrupted-block tests built the header with Vec::new() + repeated push, tripping clippy::vec_init_then_push (-D warnings in CI). Construct the fixed header bytes with vec![] instead. No behavior change. --------- Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Zhengchao An
@@ -208,20 +208,20 @@ impl TransitionClient {
|
||||
|
||||
let mut compl_multipart_upload = CompleteMultipartUpload::default();
|
||||
|
||||
let mut all_parts = Vec::<ObjectPart>::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::<ObjectPart>::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<i64, ObjectPart>, total_parts_count: i64) -> Result<Vec<ObjectPart>, 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<i64, ObjectPart> {
|
||||
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<i64> = 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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Storage>,
|
||||
pub control: Arc<StorageControl>,
|
||||
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<ReadCloser, std::io::Error> {
|
||||
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<bool, std::io::Error> {
|
||||
/*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())
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -6862,13 +6862,20 @@ async fn get_storage_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> rust
|
||||
}
|
||||
}
|
||||
pub async fn stat_all_dirs(disks: &[Option<DiskStore>], bucket: &str, prefix: &str) -> Vec<Option<DiskError>> {
|
||||
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<DiskStore>], 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<Option<DiskStore>> = 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:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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}");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
}
|
||||
|
||||
+148
-12
@@ -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<Ipv4Addr> {
|
||||
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",
|
||||
..
|
||||
}
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<u8>)> {
|
||||
let configs: [(&'static str, &Vec<u8>); 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"<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>".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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String> = vec![];
|
||||
|
||||
Reference in New Issue
Block a user