diff --git a/weed/storage/needle/needle_read.go b/weed/storage/needle/needle_read.go index c771aadb4..ac443a371 100644 --- a/weed/storage/needle/needle_read.go +++ b/weed/storage/needle/needle_read.go @@ -224,8 +224,15 @@ func (n *Needle) ReadNeedleBody(r backend.BackendStorageFile, version Version, o func (n *Needle) ReadNeedleBodyBytes(needleBody []byte, version Version) (err error) { - if len(needleBody) <= 0 { - return nil + // n.Size comes from the on-disk header, so a corrupted header can carry a + // negative size or one the body cannot hold along with its tail. + tailSize := NeedleChecksumSize + if version == Version3 { + tailSize += TimestampSize + } + if n.Size < 0 || int64(n.Size)+int64(tailSize) > int64(len(needleBody)) { + stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorIndexOutOfRange).Inc() + return fmt.Errorf("needle %v size %d out of range for body length %d: %w", n.Id, n.Size, len(needleBody), ErrorCorrupted) } switch version { case Version1: diff --git a/weed/storage/needle/needle_read_test.go b/weed/storage/needle/needle_read_test.go new file mode 100644 index 000000000..262f7ab05 --- /dev/null +++ b/weed/storage/needle/needle_read_test.go @@ -0,0 +1,87 @@ +package needle + +import ( + "bytes" + "errors" + "testing" + + . "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// readNeedleBodyBytes runs ReadNeedleBodyBytes and turns a panic into a test +// failure, so a regression reports which case broke instead of killing the run. +func readNeedleBodyBytes(t *testing.T, n *Needle, body []byte, version Version) (err error) { + t.Helper() + defer func() { + if r := recover(); r != nil { + t.Fatalf("ReadNeedleBodyBytes panicked for size %d, body length %d: %v", n.Size, len(body), r) + } + }() + return n.ReadNeedleBodyBytes(body, version) +} + +// 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) { + for _, version := range []Version{Version1, Version2, Version3} { + t.Run(versionString(version), func(t *testing.T) { + // A size of -1 is the case from #6763: its body length is still + // positive, so the scan reads a body and hands it over. + bodyLength := NeedleBodyLength(-1, version) + if bodyLength <= 0 { + t.Fatalf("expected a positive body length for size -1, got %d", bodyLength) + } + + cases := []struct { + name string + size Size + body int + }{ + {"size -1", -1, int(bodyLength)}, + {"size -12", -12, 32}, + {"size larger than body", 64, 32}, + {"no room for the tail", 32, 32}, + {"empty body", 0, 0}, + {"empty body with data size", 1, 0}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + n := &Needle{Size: c.size} + err := readNeedleBodyBytes(t, n, make([]byte, c.body), version) + if !errors.Is(err, ErrorCorrupted) { + t.Fatalf("expected an error wrapping ErrorCorrupted, got %v", err) + } + }) + } + }) + } +} + +// The size guard must still accept every record the writer produces, +// including the size-0 record a delete appends. +func TestReadNeedleBodyBytesWrittenNeedles(t *testing.T) { + for _, version := range []Version{Version1, Version2, Version3} { + t.Run(versionString(version), func(t *testing.T) { + for _, data := range [][]byte{nil, []byte("hello seaweed")} { + written := &Needle{Id: 7, Cookie: 9, Data: data, Checksum: NewCRC(data), AppendAtNs: 42} + buf := new(bytes.Buffer) + if _, _, err := writeNeedleByVersion(version, written, 0, buf); err != nil { + t.Fatalf("write needle: %v", err) + } + + n := new(Needle) + n.ParseNeedleHeader(buf.Bytes()) + body := buf.Bytes()[NeedleHeaderSize:] + if int64(len(body)) != NeedleBodyLength(n.Size, version) { + t.Fatalf("body length %d, want %d", len(body), NeedleBodyLength(n.Size, version)) + } + if err := readNeedleBodyBytes(t, n, body, version); err != nil { + t.Fatalf("read %d-byte needle: %v", len(data), err) + } + if !bytes.Equal(n.Data, data) { + t.Fatalf("data %q, want %q", n.Data, data) + } + } + }) + } +} diff --git a/weed/storage/volume_vacuum_test.go b/weed/storage/volume_vacuum_test.go index 03ffcaaad..6026b3a91 100644 --- a/weed/storage/volume_vacuum_test.go +++ b/weed/storage/volume_vacuum_test.go @@ -408,6 +408,68 @@ func TestCompactByIndex_ConcurrentWriteDoesNotFailIntegrityCheck(t *testing.T) { } } +// TestCompactByVolumeData_SkipsNegativeSizeHeader covers issue #6763: a .dat +// record whose header carries size -1 made compaction panic with "slice bounds +// out of range [:-1]". The corrupt record must be skipped and the needles on +// both sides of it kept. +func TestCompactByVolumeData_SkipsNegativeSizeHeader(t *testing.T) { + dir := t.TempDir() + + v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("volume creation: %v", err) + } + defer v.Close() + + if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil { + t.Fatalf("write needle 1: %v", err) + } + + // Size -1 still yields a positive body length, so the scan reads a body + // for this record and parses it. + datSize, _, err := v.DataBackend.GetStat() + if err != nil { + t.Fatalf("stat .dat: %v", err) + } + corrupt := make([]byte, needle.GetActualSize(-1, v.Version())) + types.NeedleIdToBytes(corrupt[types.CookieSize:types.CookieSize+types.NeedleIdSize], types.Uint64ToNeedleId(99)) + types.SizeToBytes(corrupt[types.CookieSize+types.NeedleIdSize:types.NeedleHeaderSize], -1) + if _, err := v.DataBackend.WriteAt(corrupt, datSize); err != nil { + t.Fatalf("append corrupt record: %v", err) + } + + if _, _, _, err := v.writeNeedle2(newRandomNeedle(2), true, false, false); err != nil { + t.Fatalf("write needle 2: %v", err) + } + + if err := v.CompactByVolumeData(nil); err != nil { + t.Fatalf("CompactByVolumeData: %v", err) + } + + cpx, err := os.Open(filepath.Join(dir, "1.cpx")) + if err != nil { + t.Fatalf("open .cpx: %v", err) + } + defer cpx.Close() + kept := map[types.NeedleId]bool{} + if err := idx.WalkIndexFile(cpx, 0, func(key types.NeedleId, _ types.Offset, size types.Size) error { + if size.IsValid() { + kept[key] = true + } + return nil + }); err != nil { + t.Fatalf("walk .cpx: %v", err) + } + for _, id := range []uint64{1, 2} { + if !kept[types.Uint64ToNeedleId(id)] { + t.Errorf("needle %d missing from the compacted index", id) + } + } + if kept[types.Uint64ToNeedleId(99)] { + t.Errorf("corrupt record 99 should not be in the compacted index") + } +} + // TestExceedsExpectedCompactedSize guards the copy-phase integrity check // against regressing into double-subtracting skipped bytes: expectedLiveBytes // already excludes needles dropped as unreadable (they return before being