fix(ecstore): validate erasure distribution values to avoid shuffle index panic (#4427)

fix(ecstore): validate erasure distribution values to avoid shuffle index panic (backlog#949)

The element values of `erasure.distribution` read from `xl.meta` were never
range-checked. `FileInfo::is_valid()` and `MetaObjectV1::valid()` only verified
`distribution.len()` and the `erasure.index` bound, not that each distribution
value is a valid 1-based slot in `[1, N]`. The metadata shuffle helpers then use
these values directly as `distribution[k] - 1` indices, so a corrupt or
adversarial `xl.meta` carrying a `0` (usize underflow) or a value greater than N
(out-of-bounds) triggers a panic in the shuffle path, turning bad-disk metadata
that erasure coding is meant to tolerate into a request/task crash.

Fix, two layers:
- Validate distribution values at metadata acceptance: `is_valid_distribution`
  now requires the distribution to be a permutation of `1..=N` (correct length,
  every value in range, no duplicates). `FileInfo::is_valid()` and
  `MetaObjectV1::valid()` use it, so `find_file_info_in_quorum` rejects corrupt
  metadata and it surfaces as a clean `ErasureReadQuorum` error instead of an
  index path.
- Defensive indexing in the shuffle helpers (`shuffle_disks_and_parts_metadata`,
  `_by_index`, `_by_index_owned`, `shuffle_parts_metadata`, `shuffle_disks`,
  `shuffle_check_parts`): out-of-range distribution values are skipped via
  `checked_sub(1)` + bounds-checked slot access instead of a bare `idx - 1`
  index, matching the existing pattern in
  `collect_inline_data_shard_fileinfos_by_index`.

Regression tests: `is_valid_distribution`/`is_valid`/`valid` reject distributions
containing `0`, values greater than N, duplicates, and wrong length while
accepting valid permutations; the shuffle helpers no longer panic on corrupt
distributions and preserve output length.

Refs: https://github.com/rustfs/backlog/issues/949

crates/ecstore/src/set_disk/metadata.rs

@@ -885,9 +885,17 @@ impl SetDisks {
                 continue;
             }
 
-            let block_idx = distribution[k];
-            shuffled_parts_metadata[block_idx - 1] = std::mem::take(&mut parts_metadata[k]);
-            shuffled_disks[block_idx - 1] = disks[k].take();
+            // Defensive: a corrupt/adversarial `distribution` value of `0` would
+            // underflow `block_idx - 1`, and a value `> N` would index out of
+            // bounds. Skip such entries instead of panicking (backlog#949).
+            let Some(slot) = distribution[k]
+                .checked_sub(1)
+                .filter(|slot| *slot < shuffled_parts_metadata.len() && *slot < shuffled_disks.len())
+            else {
+                continue;
+            };
+            shuffled_parts_metadata[slot] = std::mem::take(&mut parts_metadata[k]);
+            shuffled_disks[slot] = disks[k].take();
         }
 
         (shuffled_disks, shuffled_parts_metadata)
@@ -919,9 +927,17 @@ impl SetDisks {
                 continue;
             }
 
-            let block_idx = distribution[k];
-            shuffled_parts_metadata[block_idx - 1] = parts_metadata[k].clone();
-            shuffled_disks[block_idx - 1].clone_from(&disks[k]);
+            // Defensive: reject out-of-range distribution values instead of
+            // underflowing/indexing out of bounds (backlog#949).
+            let Some(slot) = distribution[k]
+                .checked_sub(1)
+                .filter(|slot| *slot < shuffled_parts_metadata.len() && *slot < shuffled_disks.len())
+            else {
+                inconsistent += 1;
+                continue;
+            };
+            shuffled_parts_metadata[slot] = parts_metadata[k].clone();
+            shuffled_disks[slot].clone_from(&disks[k]);
         }
 
         if inconsistent < fi.erasure.parity_blocks {
@@ -955,9 +971,16 @@ impl SetDisks {
             //     continue;
             // }
 
-            let block_idx = distribution[k];
-            shuffled_parts_metadata[block_idx - 1] = parts_metadata[k].clone();
-            shuffled_disks[block_idx - 1].clone_from(&disks[k]);
+            // Defensive: reject out-of-range distribution values instead of
+            // underflowing/indexing out of bounds (backlog#949).
+            let Some(slot) = distribution[k]
+                .checked_sub(1)
+                .filter(|slot| *slot < shuffled_parts_metadata.len() && *slot < shuffled_disks.len())
+            else {
+                continue;
+            };
+            shuffled_parts_metadata[slot] = parts_metadata[k].clone();
+            shuffled_disks[slot].clone_from(&disks[k]);
         }
 
         (shuffled_disks, shuffled_parts_metadata)
@@ -969,9 +992,17 @@ impl SetDisks {
         }
         let mut shuffled_parts_metadata = vec![FileInfo::default(); parts_metadata.len()];
         // Shuffle slice xl metadata for expected distribution.
-        for index in 0..parts_metadata.len() {
-            let block_index = distribution[index];
-            shuffled_parts_metadata[block_index - 1] = parts_metadata[index].clone();
+        for (index, part) in parts_metadata.iter().enumerate() {
+            // Defensive: skip missing or out-of-range distribution values
+            // instead of underflowing/indexing out of bounds (backlog#949).
+            let Some(slot) = distribution
+                .get(index)
+                .and_then(|block_index| block_index.checked_sub(1))
+                .filter(|slot| *slot < shuffled_parts_metadata.len())
+            else {
+                continue;
+            };
+            shuffled_parts_metadata[slot] = part.clone();
         }
         shuffled_parts_metadata
     }
@@ -984,8 +1015,16 @@ impl SetDisks {
         let mut shuffled_disks = vec![None; disks.len()];
 
         for (i, v) in disks.iter().enumerate() {
-            let idx = distribution[i];
-            shuffled_disks[idx - 1].clone_from(v);
+            // Defensive: skip missing or out-of-range distribution values
+            // instead of underflowing/indexing out of bounds (backlog#949).
+            let Some(slot) = distribution
+                .get(i)
+                .and_then(|idx| idx.checked_sub(1))
+                .filter(|slot| *slot < shuffled_disks.len())
+            else {
+                continue;
+            };
+            shuffled_disks[slot].clone_from(v);
         }
 
         shuffled_disks
@@ -997,8 +1036,16 @@ impl SetDisks {
         }
         let mut shuffled_parts_errs = vec![0; parts_errs.len()];
         for (i, v) in parts_errs.iter().enumerate() {
-            let idx = distribution[i];
-            shuffled_parts_errs[idx - 1] = *v;
+            // Defensive: skip missing or out-of-range distribution values
+            // instead of underflowing/indexing out of bounds (backlog#949).
+            let Some(slot) = distribution
+                .get(i)
+                .and_then(|idx| idx.checked_sub(1))
+                .filter(|slot| *slot < shuffled_parts_errs.len())
+            else {
+                continue;
+            };
+            shuffled_parts_errs[slot] = *v;
         }
         shuffled_parts_errs
     }
@@ -1175,4 +1222,56 @@ mod tests {
         let owned_slots: Vec<bool> = owned_disks.iter().map(Option::is_some).collect();
         assert_eq!(owned_slots, expected_slots, "fallback disk slots must match the borrowing variant");
     }
+
+    // backlog#949: corrupt/adversarial distribution values (0 or > N) must not
+    // trigger a `usize` underflow / out-of-bounds panic in the shuffle helpers.
+    #[test]
+    fn shuffle_parts_metadata_survives_corrupt_distribution() {
+        let parts = vec![FileInfo::default(); 4];
+        // distribution[0] = 0 underflows; distribution[1] = 9 is out of range.
+        let result = SetDisks::shuffle_parts_metadata(&parts, &[0, 9, 3, 4]);
+        assert_eq!(result.len(), parts.len(), "output length must be preserved");
+    }
+
+    #[test]
+    fn shuffle_check_parts_survives_corrupt_distribution() {
+        let errs = vec![10usize, 20, 30, 40];
+        let result = SetDisks::shuffle_check_parts(&errs, &[0, 9, 3, 4]);
+        assert_eq!(result.len(), errs.len(), "output length must be preserved");
+        // In-range entries are still placed; corrupt ones are safely skipped.
+        assert_eq!(result[2], 30, "distribution[2]=3 places errs[2] into slot 2");
+        assert_eq!(result[3], 40, "distribution[3]=4 places errs[3] into slot 3");
+    }
+
+    #[tokio::test]
+    async fn shuffle_disks_survives_corrupt_distribution() {
+        let tempdir = tempfile::tempdir().expect("tempdir should be created");
+        let disks = shuffle_test_disks(&tempdir, 4).await;
+        // distribution[0] = 0 underflows; distribution[1] = 9 is out of range.
+        let result = SetDisks::shuffle_disks(&disks, &[0, 9, 3, 4]);
+        assert_eq!(result.len(), disks.len(), "output length must be preserved");
+    }
+
+    #[tokio::test]
+    async fn shuffle_disks_and_parts_metadata_survives_corrupt_distribution() {
+        let tempdir = tempfile::tempdir().expect("tempdir should be created");
+        let (mut fi, parts) = shuffle_fixture(true);
+        let slots = parts.len();
+
+        // Corrupt the selected FileInfo's distribution: a `0` (underflow) and an
+        // out-of-range value. Length and `erasure.index` stay well-formed so the
+        // corruption is only in the distribution values.
+        fi.erasure.distribution = vec![0; slots];
+        fi.erasure.distribution[0] = slots + 5;
+
+        let disks = shuffle_test_disks(&tempdir, slots).await;
+
+        // None of these must panic on the corrupt distribution.
+        let (d1, _) = SetDisks::shuffle_disks_and_parts_metadata(&disks, &parts, &fi);
+        assert_eq!(d1.len(), disks.len());
+        let (d2, _) = SetDisks::shuffle_disks_and_parts_metadata_by_index(&disks, &parts, &fi);
+        assert_eq!(d2.len(), disks.len());
+        let (d3, _) = SetDisks::shuffle_disks_and_parts_metadata_by_index_owned(disks, parts, &fi);
+        assert_eq!(d3.len(), slots);
+    }
 }

crates/filemeta/src/fileinfo.rs

@@ -243,6 +243,36 @@ pub struct FileInfo {
     pub uses_legacy_checksum: bool,
 }
 
+/// Validates that an erasure `distribution` is a permutation of `1..=n`.
+///
+/// A well-formed distribution has exactly `n` entries and each 1-based slot
+/// index in `1..=n` appears exactly once. Corrupt or adversarial `xl.meta`
+/// can carry values of `0` or greater than `n`, which are later used as
+/// `distribution[k] - 1` indices into fixed-size vectors and would trigger a
+/// `usize` underflow / out-of-bounds panic. Rejecting such distributions here
+/// lets the metadata surface as a clean quorum/corruption error instead.
+pub(crate) fn is_valid_distribution(distribution: &[usize], n: usize) -> bool {
+    if n == 0 || distribution.len() != n {
+        return false;
+    }
+
+    let mut seen = vec![false; n];
+    for &block_idx in distribution {
+        // Valid 1-based slots are `1..=n`; anything else (including `0`) is invalid.
+        if block_idx < 1 || block_idx > n {
+            return false;
+        }
+        let slot = block_idx - 1;
+        if seen[slot] {
+            // Duplicate slot: not a permutation.
+            return false;
+        }
+        seen[slot] = true;
+    }
+
+    true
+}
+
 impl FileInfo {
     pub fn new(object: &str, data_blocks: usize, parity_blocks: usize) -> Self {
         let indices = {
@@ -286,7 +316,7 @@ impl FileInfo {
             && (data_blocks > 0)
             && (self.erasure.index > 0
                 && self.erasure.index <= data_blocks + parity_blocks
-                && self.erasure.distribution.len() == (data_blocks + parity_blocks))
+                && is_valid_distribution(&self.erasure.distribution, data_blocks + parity_blocks))
     }
 
     pub fn get_etag(&self) -> Option<String> {
@@ -685,6 +715,76 @@ mod tests {
     use proptest::collection::{hash_map, vec};
     use proptest::prelude::*;
 
+    // backlog#949: distribution range/permutation validation.
+    #[test]
+    fn is_valid_distribution_accepts_permutation() {
+        assert!(is_valid_distribution(&[1, 2, 3, 4], 4));
+        assert!(is_valid_distribution(&[4, 2, 1, 3], 4));
+        assert!(is_valid_distribution(&[1], 1));
+    }
+
+    #[test]
+    fn is_valid_distribution_rejects_zero_value() {
+        // A `0` would underflow `block_idx - 1` in the shuffle helpers.
+        assert!(!is_valid_distribution(&[0, 2, 3, 4], 4));
+    }
+
+    #[test]
+    fn is_valid_distribution_rejects_out_of_range_value() {
+        // A value greater than N would index out of bounds in the shuffle helpers.
+        assert!(!is_valid_distribution(&[1, 2, 3, 5], 4));
+        assert!(!is_valid_distribution(&[usize::MAX, 2, 3, 4], 4));
+    }
+
+    #[test]
+    fn is_valid_distribution_rejects_duplicates() {
+        // In-range but not a permutation.
+        assert!(!is_valid_distribution(&[1, 1, 3, 4], 4));
+    }
+
+    #[test]
+    fn is_valid_distribution_rejects_wrong_length_and_empty() {
+        assert!(!is_valid_distribution(&[1, 2, 3], 4));
+        assert!(!is_valid_distribution(&[1, 2, 3, 4, 4], 4));
+        assert!(!is_valid_distribution(&[], 0));
+        assert!(!is_valid_distribution(&[], 4));
+    }
+
+    fn distribution_test_fileinfo() -> FileInfo {
+        // data=2, parity=2 => N=4, distribution is a valid permutation of 1..=4.
+        let mut fi = FileInfo::new("bucket/object", 2, 2);
+        fi.erasure.index = 1;
+        fi
+    }
+
+    #[test]
+    fn is_valid_accepts_well_formed_distribution() {
+        assert!(distribution_test_fileinfo().is_valid());
+    }
+
+    #[test]
+    fn is_valid_rejects_corrupt_distribution_values() {
+        // Zero value.
+        let mut fi = distribution_test_fileinfo();
+        fi.erasure.distribution = vec![0, 2, 3, 4];
+        assert!(!fi.is_valid());
+
+        // Out-of-range value.
+        let mut fi = distribution_test_fileinfo();
+        fi.erasure.distribution = vec![1, 2, 3, 9];
+        assert!(!fi.is_valid());
+
+        // Duplicate value (not a permutation).
+        let mut fi = distribution_test_fileinfo();
+        fi.erasure.distribution = vec![1, 1, 3, 4];
+        assert!(!fi.is_valid());
+
+        // Wrong length.
+        let mut fi = distribution_test_fileinfo();
+        fi.erasure.distribution = vec![1, 2, 3];
+        assert!(!fi.is_valid());
+    }
+
     fn small_string_strategy() -> impl Strategy<Value = String> {
         proptest::string::string_regex("[A-Za-z0-9._/-]{0,16}").expect("small string regex should compile")
     }

crates/filemeta/src/filemeta/version.rs

@@ -1328,7 +1328,8 @@ impl MetaObjectV1 {
         data_blocks > 0
             && data_blocks >= parity_blocks
             && self.erasure.index > 0
-            && self.erasure.distribution.len() == data_blocks + parity_blocks
+            && self.erasure.index <= data_blocks + parity_blocks
+            && crate::fileinfo::is_valid_distribution(&self.erasure.distribution, data_blocks + parity_blocks)
     }
 
     fn decode_from<R: std::io::Read>(&mut self, rd: &mut R) -> Result<()> {