diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index fb3c3eb57..ed82c2769 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/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(); diff --git a/weed/storage/needle/needle_read.go b/weed/storage/needle/needle_read.go index 545f85baa..860eaee3e 100644 --- a/weed/storage/needle/needle_read.go +++ b/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)) diff --git a/weed/storage/needle/needle_read_test.go b/weed/storage/needle/needle_read_test.go index 1c57c02ca..213b4ba6b 100644 --- a/weed/storage/needle/needle_read_test.go +++ b/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) { diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go index 61fa767fe..fa96693de 100644 --- a/weed/storage/volume_read.go +++ b/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() diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index b1d8818cd..a4964832c 100644 --- a/weed/storage/volume_write.go +++ b/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()) diff --git a/weed/storage/volume_write_test.go b/weed/storage/volume_write_test.go index 9073de325..24beac8c2 100644 --- a/weed/storage/volume_write_test.go +++ b/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") + } +}