fix(volume): validate sizes in ReadNeedleBlob and WriteNeedleBlob (#11399)

* fix(volume): reject negative sizes in ReadNeedleBlob and WriteNeedleBlob

A ReadNeedleBlob RPC with a size of -44 or below (-36 on v2 volumes)
panics in makeslice inside needle.ReadNeedleBlob. The volume gRPC server
has no recovery interceptor, so one request kills the process. Smaller
negative sizes return bytes that are not a record.

WriteNeedleBlob accepted a negative size whenever the blob header
carried the same value: it appended the blob to .dat and indexed the
needle with that size, which reads as deleted.

Reject size < 0 in both Volume methods. Size 0 still passes, since
delete records carry it. The Rust volume server got the same storage
guards in #11345.

* fix(volume): reject needle blobs whose length does not match their size

WriteNeedleBlob appends the blob as is. A blob that is not the length
its size implies leaves .dat off the 8-byte grid, and every later
ordinary write to the volume is indexed at a truncated offset and reads
back as EOF. A blob off by 8 bytes keeps the grid but leaves bytes that
a .dat scan reads as the next record.

The in-tree callers already send exact lengths. The one case this newly
refuses is a copy between volumes of different needle versions, and
that case already writes a broken record: a v3 record lands on a v2
volume with 8 extra bytes, and a v2 record on a v3 volume either fails
the timestamp check or lands 8 bytes short.

This is separate from the negative-size guards, whose Rust counterpart
is #11345. The Rust server does not check the length yet.

* fix(volume): guard the blob buffer allocation in needle.ReadNeedleBlob

Volume.ReadNeedleBlob rejected negative sizes, but needle.ReadNeedleBlob
still sized its buffer from the size and is called directly by vacuum and
other paths. Reject a deletion marker before make() there too, and use
size.IsDeleted() in the volume-level checks.

* fix(volume): mirror the blob length check in the rust volume server

write_needle_blob_and_index checked the size against the blob header but
appended the blob verbatim, so a blob that is not the length its size
implies still leaves .dat off the record grid. Match the Go check.

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>

seaweed-volume/src/storage/volume.rs

@@ -3484,6 +3484,21 @@ impl Volume {
                 ),
             )));
         }
+        // The blob is appended as is: its length must be the record this size takes in this volume's version.
+        let actual_size = get_actual_size(size, self.version());
+        if needle_blob.len() as i64 != actual_size {
+            return Err(VolumeError::Io(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "needle {} blob of {} bytes does not match the {} bytes size {} takes in a version {} volume",
+                    needle_id.0,
+                    needle_blob.len(),
+                    actual_size,
+                    size.0,
+                    self.version().0
+                ),
+            )));
+        }
 
         // Dedup check: if the same needle already exists with matching content, skip the write.
         // Matches Go's WriteNeedleBlob which reads existing needle and compares cookie+checksum+data.
@@ -6620,6 +6635,59 @@ mod tests {
             .unwrap();
     }
 
+    // A blob shorter or longer than its record size leaves .dat off the record
+    // grid: later writes index at truncated offsets, or a scan reads the leftover
+    // bytes as the next record.
+    #[test]
+    fn test_write_needle_blob_rejects_length_mismatch() {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+
+        let mut n = Needle {
+            id: NeedleId(1),
+            cookie: Cookie(0x12345678),
+            data: b"the merged payload".to_vec(),
+            data_size: 18,
+            ..Needle::default()
+        };
+        n.checksum = CRC::new(&n.data);
+        let (offset, _, _) = v.write_needle(&mut n, true, false).unwrap();
+        let blob = v.read_needle_blob(offset as i64, n.size).unwrap();
+
+        let dat_size_before = v.dat_file_size().unwrap();
+
+        let mut too_long = blob.clone();
+        too_long.push(0);
+        let too_short = blob[..blob.len() - 1].to_vec();
+        let mut eight_long = blob.clone();
+        eight_long.extend_from_slice(&[0u8; 8]);
+        for mutated in [&too_long, &too_short, &eight_long] {
+            let err = v
+                .write_needle_blob_and_index(NeedleId(2), mutated, n.size)
+                .unwrap_err();
+            assert!(matches!(err, VolumeError::Io(_)), "got {err:?}");
+            assert_eq!(v.dat_file_size().unwrap(), dat_size_before);
+        }
+
+        // Later ordinary writes still read back correctly.
+        let mut next = Needle {
+            id: NeedleId(3),
+            cookie: Cookie(2),
+            data: b"next".to_vec(),
+            data_size: 4,
+            ..Needle::default()
+        };
+        next.checksum = CRC::new(&next.data);
+        v.write_needle(&mut next, true, false).unwrap();
+        let mut got = Needle {
+            id: NeedleId(3),
+            ..Needle::default()
+        };
+        v.read_needle(&mut got).unwrap();
+        assert_eq!(got.data, b"next");
+    }
+
     #[test]
     fn test_read_blob_negative_does_not_panic() {
         let tmp = TempDir::new().unwrap();

weed/storage/needle/needle_read.go

@@ -39,6 +39,9 @@ func (n *Needle) DiskSize(version Version) int64 {
 
 func ReadNeedleBlob(r backend.BackendStorageFile, offset int64, size Size, version Version) (dataSlice []byte, err error) {
 
+	if size.IsDeleted() {
+		return nil, fmt.Errorf("invalid needle size %d: %w", size, ErrorSizeInvalid)
+	}
 	dataSize := GetActualSize(size, version)
 	dataSlice = make([]byte, int(dataSize))
 

weed/storage/needle/needle_read_test.go

@@ -21,6 +21,17 @@ func readNeedleBodyBytes(t *testing.T, n *Needle, body []byte, version Version)
 	return n.ReadNeedleBodyBytes(body, version)
 }
 
+// The size feeds the read buffer's length, so a negative one never reaches make().
+func TestReadNeedleBlobRejectsNegativeSize(t *testing.T) {
+	for _, version := range []Version{Version1, Version2, Version3} {
+		for _, size := range []Size{TombstoneFileSize, -100} {
+			if _, err := ReadNeedleBlob(nil, 0, size, version); !errors.Is(err, ErrorSizeInvalid) {
+				t.Fatalf("version %d size %d: expected ErrorSizeInvalid, got %v", version, size, err)
+			}
+		}
+	}
+}
+
 // A corrupted .dat header can carry a size that does not fit the body read for
 // it. Vacuum used to panic on it with "slice bounds out of range [:-1]" (#6763).
 func TestReadNeedleBodyBytesRejectsCorruptSize(t *testing.T) {

weed/storage/volume_read.go

@@ -235,6 +235,11 @@ func min(x, y int) int {
 
 // read fills in Needle content by looking up n.Id from NeedleMapper
 func (v *Volume) ReadNeedleBlob(offset int64, size Size) ([]byte, error) {
+	// A deletion marker is not a record length; reject it before taking the lock.
+	if size.IsDeleted() {
+		return nil, fmt.Errorf("invalid needle size %d", size)
+	}
+
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()
 

weed/storage/volume_write.go

@@ -457,6 +457,9 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 	if v.IsReadOnly() {
 		return fmt.Errorf("volume %d is read only", v.Id)
 	}
+	if size.IsDeleted() {
+		return fmt.Errorf("needle %d has invalid size %d", needleId, size)
+	}
 
 	// size indexes the needle and places the v3 append timestamp, so a caller using
 	// the payload-only DataSize corrupts both, silently until the needle is read back.
@@ -468,6 +471,10 @@ func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size
 	if blobHeader.Size != size {
 		return fmt.Errorf("needle %d size %d does not match its blob header size %d", needleId, size, blobHeader.Size)
 	}
+	// The blob is appended as is: its length must be the record this size takes in this volume's version.
+	if actualSize := needle.GetActualSize(size, v.Version()); int64(len(needleBlob)) != actualSize {
+		return fmt.Errorf("needle %d blob of %d bytes does not match the %d bytes size %d takes in a version %d volume", needleId, len(needleBlob), actualSize, size, v.Version())
+	}
 
 	if MaxPossibleVolumeSize < v.nm.ContentSize()+uint64(len(needleBlob)) {
 		return fmt.Errorf("volume size limit %d exceeded! current size is %d", MaxPossibleVolumeSize, v.nm.ContentSize())

weed/storage/volume_write_test.go

@@ -1,8 +1,10 @@
 package storage
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
+	"math"
 	"os"
 	"testing"
 	"time"
@@ -258,3 +260,190 @@ func TestWriteNeedleBlobRejectsSizeMismatch(t *testing.T) {
 		t.Fatalf("write needle blob with the header size: %v", err)
 	}
 }
+
+// A negative size reaches make() in needle.ReadNeedleBlob, and the blob RPCs
+// have no recover, so one request took down the volume server.
+func TestReadNeedleBlobRejectsNegativeSize(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	defer v.Close()
+
+	n := newRandomNeedle(1)
+	offset, _, _, err := v.writeNeedle2(n, true, false, false)
+	if err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+
+	for _, size := range []types.Size{types.TombstoneFileSize, -100, math.MinInt32} {
+		t.Run(fmt.Sprintf("size %d", size), func(t *testing.T) {
+			defer func() {
+				if r := recover(); r != nil {
+					t.Errorf("ReadNeedleBlob panicked: %v", r)
+				}
+			}()
+			if _, err := v.ReadNeedleBlob(int64(offset), size); err == nil {
+				t.Error("expected ReadNeedleBlob to reject a negative size")
+			}
+		})
+	}
+
+	// Only negative sizes are rejected: size 0 is what a delete record carries.
+	for _, size := range []types.Size{0, n.Size} {
+		if _, err := v.ReadNeedleBlob(int64(offset), size); err != nil {
+			t.Errorf("ReadNeedleBlob with size %d: %v", size, err)
+		}
+	}
+}
+
+// The blob header carries the same negative size, so only the sign is wrong.
+func TestWriteNeedleBlobRejectsNegativeSize(t *testing.T) {
+	for _, size := range []types.Size{types.TombstoneFileSize, -5} {
+		t.Run(fmt.Sprintf("size %d", size), func(t *testing.T) {
+			dir := t.TempDir()
+			v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+			if err != nil {
+				t.Fatalf("volume creation: %v", err)
+			}
+			defer v.Close()
+
+			n := newRandomNeedle(1)
+			offset, _, _, err := v.writeNeedle2(n, true, false, false)
+			if err != nil {
+				t.Fatalf("write needle: %v", err)
+			}
+			blob, err := v.ReadNeedleBlob(int64(offset), n.Size)
+			if err != nil {
+				t.Fatalf("read needle blob: %v", err)
+			}
+			// Make the header agree with the size, so only the sign is wrong.
+			types.SizeToBytes(blob[types.NeedleHeaderSize-types.SizeSize:types.NeedleHeaderSize], size)
+
+			datSizeBefore, _, _ := v.DataBackend.GetStat()
+
+			if err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, size); err == nil {
+				t.Error("expected WriteNeedleBlob to reject a negative size")
+			}
+
+			datSizeAfter, _, _ := v.DataBackend.GetStat()
+			if datSizeAfter != datSizeBefore {
+				t.Errorf(".dat grew from %d to %d on a rejected blob", datSizeBefore, datSizeAfter)
+			}
+			if nv, ok := v.nm.Get(types.Uint64ToNeedleId(2)); ok {
+				t.Errorf("needle 2 was indexed with size %d", nv.Size)
+			}
+		})
+	}
+}
+
+// A blob shorter or longer than its record size leaves .dat off the record grid:
+// later writes index at truncated offsets, or a scan reads the leftover bytes as
+// the next record.
+func TestWriteNeedleBlobRejectsLengthMismatch(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		mutate func(blob []byte) []byte
+	}{
+		{"one byte too long", func(blob []byte) []byte { return append(blob, 0) }},
+		{"one byte short", func(blob []byte) []byte { return blob[:len(blob)-1] }},
+		{"8 bytes too long", func(blob []byte) []byte { return append(blob, make([]byte, 8)...) }},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			dir := t.TempDir()
+			v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+			if err != nil {
+				t.Fatalf("volume creation: %v", err)
+			}
+			defer v.Close()
+
+			n := newRandomNeedle(1)
+			offset, _, _, err := v.writeNeedle2(n, true, false, false)
+			if err != nil {
+				t.Fatalf("write needle: %v", err)
+			}
+			blob, err := v.ReadNeedleBlob(int64(offset), n.Size)
+			if err != nil {
+				t.Fatalf("read needle blob: %v", err)
+			}
+			datSizeBefore, _, _ := v.DataBackend.GetStat()
+
+			if err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), tc.mutate(blob), n.Size); err == nil {
+				t.Error("expected WriteNeedleBlob to reject a blob whose length does not match its size")
+			}
+
+			datSizeAfter, _, _ := v.DataBackend.GetStat()
+			if datSizeAfter != datSizeBefore {
+				t.Errorf(".dat grew from %d to %d on a rejected blob", datSizeBefore, datSizeAfter)
+			}
+
+			next := newRandomNeedle(3)
+			if _, _, _, err = v.writeNeedle2(next, true, false, false); err != nil {
+				t.Fatalf("write needle 3: %v", err)
+			}
+			got := newEmptyNeedle(3)
+			if _, err = v.readNeedle(got, nil, nil); err != nil {
+				t.Fatalf("read back needle 3: %v", err)
+			}
+			if !bytes.Equal(got.Data, next.Data) {
+				t.Error("needle 3 read back with different data")
+			}
+		})
+	}
+}
+
+// The checks must pass what the real callers send: a needle's own record and the
+// size-0 record a delete leaves.
+func TestWriteNeedleBlobRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 7, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	defer v.Close()
+
+	n := newRandomNeedle(1)
+	offset, _, _, err := v.writeNeedle2(n, true, false, false)
+	if err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	blob, err := v.ReadNeedleBlob(int64(offset), n.Size)
+	if err != nil {
+		t.Fatalf("read needle blob: %v", err)
+	}
+	if err = v.WriteNeedleBlob(types.Uint64ToNeedleId(2), blob, n.Size); err != nil {
+		t.Fatalf("write needle blob: %v", err)
+	}
+	got := newEmptyNeedle(2)
+	if _, err = v.readNeedle(got, nil, nil); err != nil {
+		t.Fatalf("read back needle 2: %v", err)
+	}
+	if !bytes.Equal(got.Data, n.Data) {
+		t.Error("needle 2 read back with different data")
+	}
+
+	deleteOffset, _, _ := v.DataBackend.GetStat()
+	if _, err = v.doDeleteRequest(newEmptyNeedle(1)); err != nil {
+		t.Fatalf("delete needle 1: %v", err)
+	}
+	deleteRecord, err := v.ReadNeedleBlob(deleteOffset, 0)
+	if err != nil {
+		t.Fatalf("read delete record: %v", err)
+	}
+	if err = v.WriteNeedleBlob(types.Uint64ToNeedleId(3), deleteRecord, 0); err != nil {
+		t.Fatalf("write size-0 needle blob: %v", err)
+	}
+
+	next := newRandomNeedle(4)
+	if _, _, _, err = v.writeNeedle2(next, true, false, false); err != nil {
+		t.Fatalf("write needle 4: %v", err)
+	}
+	got = newEmptyNeedle(4)
+	if _, err = v.readNeedle(got, nil, nil); err != nil {
+		t.Fatalf("read back needle 4: %v", err)
+	}
+	if !bytes.Equal(got.Data, next.Data) {
+		t.Error("needle 4 read back with different data")
+	}
+}