clawhdf5-remote: no overflow on lengths near u64::MAX

A server can claim any length in Content-Range. block_len computed
start + block_size, which overflowed in the last blocks of a file claimed
to be near u64::MAX (a panic in debug builds, a wrapped value in
release); insert() multiplied block indices unchecked. The cache's block
arithmetic is now saturating/checked, and a run that does not split into
whole blocks is an error instead of an endless loop or a slice panic.

The test server gains fake_total (claim a length, serve zeros past the
data); a test reads the last bytes of such files and opens a file whose
superblock EOF and root addresses sit near u64::MAX.

Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>

crates/clawhdf5-remote/src/cache.rs

@@ -305,9 +305,9 @@ impl<S: Storage> BlockCache<S> {
         let end = offset.saturating_add(bytes.len() as u64).min(self.len);
         let mut i = offset.div_ceil(bs);
         let mut st = lock(&self.state);
-        while i * bs < end {
-            let start = i * bs;
-            let block_end = (start + bs).min(self.len);
+        // Checked: a hostile server can claim a length near u64::MAX.
+        while let Some(start) = i.checked_mul(bs).filter(|&s| s < end) {
+            let block_end = start.saturating_add(bs).min(self.len);
             if block_end > end {
                 break;
             }
@@ -330,9 +330,14 @@ impl<S: Storage> BlockCache<S> {
         Some(offset / bs..(end - 1) / bs + 1)
     }
 
+    /// Length of block `i` (0 past the end of the file). Saturating: the
+    /// length may be anything a server claimed, up to `u64::MAX`.
     fn block_len(&self, i: u64) -> u64 {
-        let start = i * self.config.block_size;
-        (start + self.config.block_size).min(self.len) - start
+        let start = i.saturating_mul(self.config.block_size);
+        start
+            .saturating_add(self.config.block_size)
+            .min(self.len)
+            .saturating_sub(start)
     }
 
     fn keep(&self, st: &mut State, i: u64, data: Block) {
@@ -453,8 +458,17 @@ impl<S: Storage> BlockCache<S> {
                 let mut i = run.start / bs;
                 let mut pos = 0usize;
                 while pos < bytes.len() {
-                    let n = self.block_len(i) as usize;
-                    got.push((i, Arc::from(&bytes[pos..pos + n])));
+                    let n = usize::try_from(self.block_len(i)).unwrap_or(usize::MAX);
+                    let Some(block) = pos.checked_add(n).and_then(|e| bytes.get(pos..e)) else {
+                        return Err(guard.fail(FormatError::Storage(format!(
+                            "a run at offset {} does not split into whole blocks",
+                            run.start
+                        ))));
+                    };
+                    if n == 0 {
+                        break;
+                    }
+                    got.push((i, Arc::from(block)));
                     pos += n;
                     i += 1;
                 }
@@ -512,7 +526,9 @@ impl<S: Storage> BlockCache<S> {
             }
         }
         runs.into_iter()
-            .map(|(a, b)| a * bs..(b * bs + self.block_len(b)))
+            .map(|(a, b)| {
+                a.saturating_mul(bs)..b.saturating_mul(bs).saturating_add(self.block_len(b))
+            })
             .collect()
     }
 
@@ -863,4 +879,42 @@ mod tests {
         assert_eq!(n, blocks.len(), "a block was fetched twice: {fetched:?}");
         assert!(c.stats().waits > 0, "readers should have shared fetches");
     }
+
+    /// A backend claiming any length (a hostile server's `Content-Range`)
+    /// and serving zeros.
+    struct Zeros(u64);
+
+    impl Storage for Zeros {
+        fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, FormatError> {
+            let end = offset.saturating_add(len as u64).min(self.0);
+            Ok(Cow::Owned(vec![0; end.saturating_sub(offset) as usize]))
+        }
+        fn len(&self) -> u64 {
+            self.0
+        }
+    }
+
+    #[test]
+    fn lengths_near_u64_max_do_not_overflow() {
+        for len in [u64::MAX, u64::MAX - 1, u64::MAX - 1000, 1 << 63] {
+            let c = BlockCache::new(Zeros(len), small());
+            for (off, n) in [
+                (len - 100, 50),
+                (len - 10, 100),
+                (len - 1, 1),
+                (len - 3000, 3000),
+                (u64::MAX, 10),
+            ] {
+                let got = c.read_at(off, n).unwrap();
+                assert_eq!(got.len() as u64, len.saturating_sub(off).min(n as u64));
+            }
+            let got = c
+                .read_ranges(&[len - 5000..len, u64::MAX - 1..u64::MAX])
+                .unwrap();
+            assert_eq!(got[0].len(), 5000);
+            c.insert(len - 5, &[0; 5]);
+            c.insert(u64::MAX - 5, &[0; 5]);
+            c.prefetch(len - 1, 10).unwrap();
+        }
+    }
 }