fix(pools): keep rejected names out of a single-pool bulk delete

DeleteObjects checks every name with checkDelObjArgs, but a single pool
still sent all entries to the drives and cleared the error of an entry
that a drive reported as found. Local drives join the name to the bucket
directory, so a name with a ".." segment reached another bucket or the
meta bucket. An S3 DeleteObjects caller allowed to delete in one bucket
could remove a version elsewhere, including one under Object Lock
retention, and from a versioned bucket could write a delete marker on an
object of another bucket or on stored bucket metadata. The bulk lock was
taken on the normalized name as well, that is on the other object.

Entries that fail the check now keep their error, and the rest of the
request runs as a request of its own, before any lock or lookup. A
request whose names are all valid does not enter the new code. More than
one pool already skipped such entries.

Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

CHANGELOG.md

@@ -20,6 +20,17 @@
   `svcacct info` reports with `Policy: implied`, and set the policy again on
   those that were meant to be restricted.
 
+- S3 `DeleteObjects` on a single pool stops at an entry whose name fails the
+  object name check: the entry returns `XMinioInvalidObjectName` and is not
+  locked or sent to a drive, as with a single-object DELETE and with more than
+  one pool. Until now such an entry still reached the drives of the node, where
+  a name with a `..` segment was joined to the request bucket. A caller allowed
+  to delete in one bucket could remove a version in another bucket, including
+  one under Object Lock retention, and from a versioned bucket could put a
+  delete marker on an object of another bucket or on stored system metadata
+  such as a bucket's configuration. The other entries of a request are deleted
+  as before.
+
 ## RELEASE.2026-11-09T00-00-00Z
 
 Package version: `20261109000000.0.0`. Publication status and artifacts are

cmd/bulk-delete-invalid-name_test.go

@@ -0,0 +1,156 @@
+// Copyright (c) 2026 PGSTY
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cmd
+
+import (
+	"bytes"
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"testing"
+
+	"github.com/minio/minio/internal/auth"
+)
+
+// A bulk delete joins each name to the request bucket on the drives, so a name
+// with a ".." segment names a path in another bucket or in the meta bucket.
+// Such a name is the error of its own entry and reaches no drive, and the
+// other entries of the request are deleted as usual. This holds for the one
+// bulk lock of a single node, for a pool that locks per set, and for more than
+// one pool.
+func TestPoolsBulkDeleteInvalidNames(t *testing.T) {
+	for _, topology := range []struct {
+		name string
+		new  func(t *testing.T) (*erasureServerPools, string)
+	}{
+		{"single node", func(t *testing.T) (*erasureServerPools, string) { return bulkDeletePools(t, 1) }},
+		{"set locks", func(t *testing.T) (*erasureServerPools, string) { return ldPool(t, ldLocal) }},
+		{"two pools", func(t *testing.T) (*erasureServerPools, string) { return bulkDeletePools(t, 2) }},
+	} {
+		t.Run(topology.name, func(t *testing.T) {
+			z, bucket := topology.new(t)
+			ctx := t.Context()
+			other := "bulk-delete-other"
+			if err := z.MakeBucket(ctx, other, MakeBucketOptions{VersioningEnabled: true}); err != nil {
+				t.Fatal(err)
+			}
+			old := batchPutVersion(t, z, other, "kept", nil)
+			latest := batchPutVersion(t, z, other, "kept", nil)
+			batchPutVersion(t, z, other, "visible", nil)
+			metaKey := pathJoin(bucketMetaPrefix, other, bucketMetadataFile)
+			if !batchVersionExists(t, z, minioMetaBucket, metaKey, "") {
+				t.Fatal("the bucket metadata is not an object of the meta bucket")
+			}
+			for _, key := range []string{"own", "dir/"} {
+				if _, err := z.PutObject(ctx, bucket, key, mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{}); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			names := []ObjectToDelete{
+				{ObjectV: ObjectV{ObjectName: "../" + other + "/kept", VersionID: old.VersionID}}, // purge in another bucket
+				{ObjectV: ObjectV{ObjectName: "own"}},
+				{ObjectV: ObjectV{ObjectName: "a/../../" + other + "/visible"}},         // marker in another bucket
+				{ObjectV: ObjectV{ObjectName: "../" + minioMetaBucket + "/" + metaKey}}, // marker on bucket metadata
+				{ObjectV: ObjectV{ObjectName: "dir/"}},
+				{ObjectV: ObjectV{ObjectName: ""}},
+			}
+			invalid := map[int]bool{0: true, 2: true, 3: true, 5: true}
+			// Without a version ID these options ask for a delete marker, the
+			// one write a bulk delete makes.
+			deleted, errs := z.DeleteObjects(ctx, bucket, slices.Clone(names), ObjectOptions{Versioned: true})
+			for i, name := range names {
+				_, rejected := errs[i].(ObjectNameInvalid)
+				if rejected != invalid[i] || !invalid[i] && errs[i] != nil {
+					t.Errorf("%d %q: error = %v, rejected as invalid = %v, want %v", i, name.ObjectName, errs[i], rejected, invalid[i])
+				}
+				if deleted[i].ObjectName != name.ObjectName || deleted[i].found == invalid[i] {
+					t.Errorf("%d %q: result = %+v", i, name.ObjectName, deleted[i])
+				}
+			}
+
+			for _, version := range []string{old.VersionID, latest.VersionID} {
+				if !batchVersionExists(t, z, other, "kept", version) {
+					t.Errorf("version %q of the object in the other bucket was removed", version)
+				}
+			}
+			if !batchVersionExists(t, z, other, "visible", "") {
+				t.Error("the object of the other bucket is hidden by a marker")
+			}
+			if !batchVersionExists(t, z, minioMetaBucket, metaKey, "") {
+				t.Error("the metadata of the other bucket is hidden by a marker")
+			}
+			for _, key := range []string{"own", "dir/"} {
+				if batchVersionExists(t, z, bucket, key, "") {
+					t.Errorf("%q of the request bucket was not deleted", key)
+				}
+			}
+		})
+	}
+}
+
+// The names of an S3 DeleteObjects request come from its body, which the path
+// checks of the request URL do not see. A user allowed to delete in one bucket
+// reaches nothing outside it through such a name: a version under COMPLIANCE
+// retention in another bucket stays, and no delete marker lands on that object
+// or on the stored metadata of its bucket.
+func TestBulkDeleteObjectLockNameOutsideBucket(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t: t, endpoints: []string{"DeleteMultipleObjects"},
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, _ auth.Credentials, t *testing.T) {
+			ctx := t.Context()
+			user := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:DeleteObject","s3:DeleteObjectVersion"`)
+			locked := getRandomBucketName()
+			if err := obj.MakeBucket(ctx, locked, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}); err != nil {
+				t.Fatal(err)
+			}
+			enableBucketObjectLock(t, locked)
+			worm := batchPutVersion(t, obj, locked, "secret", batchLockMeta("COMPLIANCE"))
+			batchPutVersion(t, obj, bucket, "own", nil)
+			metaKey := pathJoin(bucketMetaPrefix, locked, bucketMetadataFile)
+
+			outside := []ObjectToDelete{
+				{ObjectV: ObjectV{ObjectName: "../" + locked + "/secret", VersionID: worm.VersionID}},
+				{ObjectV: ObjectV{ObjectName: "../" + locked + "/secret"}},
+				{ObjectV: ObjectV{ObjectName: "../" + minioMetaBucket + "/" + metaKey}},
+			}
+			body := encodeResponse(DeleteObjectsRequest{Objects: append(slices.Clone(outside), ObjectToDelete{ObjectV: ObjectV{ObjectName: "own"}})})
+			req, err := newTestSignedRequestV4(http.MethodPost, getDeleteMultipleObjectsURL("", bucket), int64(len(body)), bytes.NewReader(body),
+				user.AccessKey, user.SecretKey, nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec := httptest.NewRecorder()
+			router.ServeHTTP(rec, req)
+			var resp DeleteObjectsResponse
+			if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &resp) != nil {
+				t.Fatalf("%s: status %d: %s", instance, rec.Code, rec.Body.String())
+			}
+
+			if len(resp.DeletedObjects) != 1 || resp.DeletedObjects[0].ObjectName != "own" || len(resp.Errors) != len(outside) {
+				t.Errorf("%s: want the three names outside the bucket refused and \"own\" deleted: %s", instance, rec.Body.String())
+			}
+			for _, e := range resp.Errors {
+				if e.Code != errorCodes[ErrInvalidObjectName].Code {
+					t.Errorf("%s: %q: code %s, want %s", instance, e.Key, e.Code, errorCodes[ErrInvalidObjectName].Code)
+				}
+			}
+			if !batchVersionExists(t, obj, locked, "secret", worm.VersionID) {
+				t.Errorf("%s: the version under COMPLIANCE retention in the other bucket was removed", instance)
+			}
+			if !batchVersionExists(t, obj, locked, "secret", "") {
+				t.Errorf("%s: a delete marker hides the object of the other bucket", instance)
+			}
+			if !batchVersionExists(t, obj, minioMetaBucket, metaKey, "") {
+				t.Errorf("%s: a delete marker hides the stored metadata of the other bucket", instance)
+			}
+			if batchVersionExists(t, obj, bucket, "own", "") {
+				t.Errorf("%s: the object of the request bucket was not deleted", instance)
+			}
+		},
+	})
+}

cmd/erasure-server-pool-lifecycle-authority_test.go

@@ -20,6 +20,7 @@ package cmd
 import (
 	"context"
 	"crypto/sha256"
+	"encoding/binary"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -29,6 +30,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cespare/xxhash/v2"
 	"github.com/google/uuid"
 	"github.com/minio/madmin-go/v3"
 	"github.com/minio/minio/internal/bucket/lifecycle"
@@ -632,6 +634,176 @@ func TestPoolsLifecycleLegacyHistory(t *testing.T) {
 	}
 }
 
+// authoritySignature2021 is the header signature of o, a version whose part
+// ETags are all empty, as the releases from RELEASE.2021-11-24 until
+// RELEASE.2022-01-25 computed it: with these ETags.
+func authoritySignature2021(o xlMetaV2Object) (sig [4]byte) {
+	o.ErasureIndex = 0
+	crc := hashDeterministicString(o.MetaUser) ^ hashDeterministicBytes(o.MetaSys)
+	o.MetaUser, o.MetaSys = nil, nil
+	bts, _ := o.MarshalMsg(nil)
+	crc ^= xxhash.Sum64(bts)
+	binary.LittleEndian.PutUint32(sig[:], uint32(crc^(crc>>32)))
+	return sig
+}
+
+// A drive holds the header that the release which wrote a version there
+// stored. Before RELEASE.2024-04-28 it has no erasure coding, and until
+// RELEASE.2022-01-25 its signature counted the empty part ETags of versions
+// written before RELEASE.2021-11-24. A heal writes the header of its own
+// release and leaves the other drives alone, so the drives of an old version
+// differ there for good once some were replaced. That is one revision, as
+// reads and heal take it, on however many of the drives.
+func TestPoolsLifecycleHeaderEras(t *testing.T) {
+	id := [16]byte{1}
+	current := func(drive int) xlMetaV2ShallowVersion { return authorityRevision(t, id, 1, drive+1, "") }
+	// stored returns the version as drive holds it after change, under the
+	// header that header derives from the one a current release writes.
+	stored := func(drive int, change func(*xlMetaV2Object), header func(xlMetaV2VersionHeader, xlMetaV2Object) xlMetaV2VersionHeader) xlMetaV2ShallowVersion {
+		ver := current(drive)
+		var v xlMetaV2Version
+		if _, err := v.UnmarshalMsg(ver.meta); err != nil {
+			t.Fatal(err)
+		}
+		change(v.ObjectV2)
+		meta, err := v.MarshalMsg(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return xlMetaV2ShallowVersion{header: header(ver.header, *v.ObjectV2), meta: meta}
+	}
+	eras := map[string]func(drive int) xlMetaV2ShallowVersion{
+		"no-erasure-coding": func(drive int) xlMetaV2ShallowVersion {
+			return stored(drive, func(*xlMetaV2Object) {}, func(h xlMetaV2VersionHeader, _ xlMetaV2Object) xlMetaV2VersionHeader {
+				h.EcN, h.EcM = 0, 0
+				return h
+			})
+		},
+		"empty-part-etags-signed": func(drive int) xlMetaV2ShallowVersion {
+			return stored(drive, func(o *xlMetaV2Object) { o.PartETags = []string{""} }, func(h xlMetaV2VersionHeader, o xlMetaV2Object) xlMetaV2VersionHeader {
+				h.EcN, h.EcM, h.Signature = 0, 0, authoritySignature2021(o)
+				return h
+			})
+		},
+	}
+	metaVs := slices.Repeat([]uint8{xlMetaVersion}, 16)
+	readQuorum := func(xlMetaV2ShallowVersion, uint8) int { return 12 }
+	for name, old := range eras {
+		t.Run(name, func(t *testing.T) {
+			if old(0).header == current(0).header {
+				t.Fatal("fixture: the old header is the current one")
+			}
+			if versionRevision(old(0), xlMetaVersion) != versionRevision(current(7), xlMetaVersion) {
+				t.Error("one version under the headers of two releases has two keys")
+			}
+			// The header of the key orders the merge: it is the current one.
+			if versionRevision(old(0), xlMetaVersion).header != current(0).header {
+				t.Error("the key does not carry the header of the current release")
+			}
+			for healed := range 17 {
+				metas := make([][]xlMetaV2ShallowVersion, 16)
+				for i := range metas {
+					if i < healed {
+						metas[i] = []xlMetaV2ShallowVersion{current(i)}
+					} else {
+						metas[i] = []xlMetaV2ShallowVersion{old(i)}
+					}
+				}
+				merged, err := authoritativeHistory(t.Context(), metas, metaVs, xlMetaVersion, 9, readQuorum)
+				if err != nil || len(merged) != 1 {
+					t.Errorf("%d healed drives: %d versions, err=%v", healed, len(merged), err)
+				}
+			}
+		})
+	}
+	// Control: the erasure coding still tells revisions apart, by the
+	// metadata. Another one under the same header is another revision.
+	other := stored(0, func(o *xlMetaV2Object) { o.ErasureM, o.ErasureN = 8, 8 }, func(h xlMetaV2VersionHeader, _ xlMetaV2Object) xlMetaV2VersionHeader { return h })
+	if versionRevision(other, xlMetaVersion) == versionRevision(current(0), xlMetaVersion) {
+		t.Error("another erasure coding under one header keeps the revision")
+	}
+}
+
+// The same on drives, through a heal: two versions with headers from before
+// RELEASE.2024-04-28 on a set where half of the drives were replaced. GET
+// reads them and heal has nothing left to do. Their expiry has to run as
+// well: of a noncurrent version with several pools, and of all versions of a
+// key with one pool too, which reads the history like the pools do. A
+// restore takes the drives for one generation, not for a failed status write.
+func TestPoolsLifecycleHealedOldHeaders(t *testing.T) {
+	for _, single := range []bool{false, true} {
+		name, key := "noncurrent-version", "ncv-old-headers"
+		if single {
+			name, key = "all-versions-one-pool", "eav-old-headers"
+		}
+		t.Run(name, func(t *testing.T) {
+			s := newSplitILM(t, versioning.Enabled, splitILMRules)
+			ctx := t.Context()
+			s.write(key, splitAt("old", 5, 0))
+			s.write(key, splitAt("new", 3, 0))
+			ids := []string{s.ids[key+" old"], s.ids[key+" new"]}
+			if single {
+				prior := s.z.serverPools
+				t.Cleanup(func() { s.z.serverPools = prior })
+				s.z.serverPools = s.z.serverPools[:1]
+			}
+			set := s.z.serverPools[0].getHashedSet(key)
+			disks := set.getDisks()
+			for _, disk := range disks {
+				rewriteTierTestMeta(t, disk, s.bucket, key, func(xl *xlMetaV2) {
+					for i := range xl.versions {
+						// Control: the key of a version this release wrote
+						// carries its header as stored.
+						if versionRevision(xl.versions[i], xl.metaV).header != xl.versions[i].header {
+							t.Fatal("the key of a current version carries another header")
+						}
+						xl.versions[i].header.EcN, xl.versions[i].header.EcM = 0, 0
+					}
+				})
+			}
+			replaced := len(disks) / 2
+			for _, disk := range disks[:replaced] {
+				if err := disk.Delete(ctx, s.bucket, key, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
+					t.Fatal(err)
+				}
+				for _, id := range ids {
+					if _, err := s.z.HealObject(ctx, s.bucket, key, id, madmin.HealOpts{ScanMode: madmin.HealNormalScan}); err != nil {
+						t.Fatalf("heal %s: %v", id, err)
+					}
+				}
+			}
+			for i, disk := range disks {
+				if authorityHeader(t, disk, s.bucket, key, ids[0]).hasEC() != (i < replaced) {
+					t.Fatalf("fixture: drive %d of %d, %d replaced, holds the other header", i, len(disks), replaced)
+				}
+			}
+			authorityPresent(t, s, 0, key, ids[0])
+			res, err := s.z.HealObject(ctx, s.bucket, key, ids[0], madmin.HealOpts{DryRun: true, ScanMode: madmin.HealNormalScan})
+			if err != nil || slices.ContainsFunc(res.Before.Drives, func(d madmin.HealDriveInfo) bool { return d.State != madmin.DriveStateOk }) {
+				t.Fatalf("fixture: heal has work left: %v", err)
+			}
+
+			if gen, err := set.readRestoreGeneration(ctx, s.bucket, key, ids[0]); err != nil || gen.split {
+				t.Errorf("restore generation: split=%v, err=%v", gen.split, err)
+			}
+			if single {
+				_, err = s.z.DeleteObject(ctx, s.bucket, key, authorityDeleteAll)
+			} else {
+				err = authorityExpireVersion(t, s, key, ids[0])
+			}
+			if err != nil {
+				t.Fatalf("expiry: %v", err)
+			}
+			if _, err := s.z.GetObjectInfo(ctx, s.bucket, key, ObjectOptions{VersionID: ids[0]}); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+				t.Errorf("the expired version: %v", err)
+			}
+			if !single {
+				authorityPresent(t, s, 0, key, ids[1])
+			}
+		})
+	}
+}
+
 // authorityDeleteAll is the expiry of all versions of a key as the expiry
 // worker sends it: a purge of all versions, approved by the lifecycle.
 var authorityDeleteAll = ObjectOptions{DeleteAllVersions: true, Expiration: ExpirationOptions{Expire: true}}

cmd/erasure-server-pool-lifecycle.go

@@ -187,9 +187,17 @@ type revisionKey struct {
 // input is canonical: each map, including the part checksums of a legacy
 // version, is its own section with its entry count and sorted,
 // length-prefixed entries, since msgp encodes maps in iteration order and an
-// entry must not move between maps unnoticed. The header of a
-// legacy version carries a signature over its per-shard checksums, so the
-// key replaces it with one from the digest.
+// entry must not move between maps unnoticed.
+//
+// A drive holds the header that the release which wrote the version there
+// stored, and a heal writes the one of its own release: headers from before
+// RELEASE.2024-04-28 carry no erasure coding, and the signature counted
+// empty part ETags until RELEASE.2022-01-25. Reads and heal take such drives
+// for one version (matchesNotStrict), and nothing rewrites them. The key has
+// both as the current release computes them from the metadata, so that the
+// header of a version it wrote stays as stored: the signature also orders
+// versions of one time in the merge. The signature of a legacy version is
+// over its per-shard checksums; the key has one from the digest.
 func versionRevision(ver xlMetaV2ShallowVersion, metaV uint8) revisionKey {
 	h := sha256.New()
 	writeUint := func(n int) {
@@ -267,8 +275,12 @@ func versionRevision(ver xlMetaV2ShallowVersion, metaV uint8) revisionKey {
 			section("fields", 1)
 			writeField(bts)
 			h.Sum(key.digest[:0])
-			if v.Type == LegacyType {
+			switch v.Type {
+			case LegacyType:
 				copy(key.header.Signature[:], key.digest[:])
+			case ObjectType:
+				key.header.Signature = v.ObjectV2.Signature()
+				key.header.EcN, key.header.EcM = uint8(v.ObjectV2.ErasureN), uint8(v.ObjectV2.ErasureM)
 			}
 			return key
 		}
@@ -350,7 +362,7 @@ func authoritativeHistory(ctx context.Context, metas [][]xlMetaV2ShallowVersion,
 	// its count, and one that shares the header could stand in for the
 	// accepted one, so the merge only sees the accepted revision of each
 	// version and free versions as they are. It sees the headers of the
-	// keys, which do not differ between the drives of a legacy version.
+	// keys, which do not differ between the drives of a revision.
 	filtered := make([][]xlMetaV2ShallowVersion, len(metas))
 	for i, versions := range metas {
 		for j, ver := range versions {

cmd/erasure-server-pool.go

@@ -26,6 +26,7 @@ import (
 	"math/rand"
 	"net/http"
 	"path"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -1543,6 +1544,27 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
 		objSets.Add(objects[i].ObjectName)
 	}
 
+	// A name that fails the check keeps its error and goes no further. Joined
+	// to the bucket it can name a path in another bucket or in the meta
+	// bucket, so it is not locked, looked up or sent to a drive. The other
+	// names are deleted as a request of their own: everything below sees only
+	// names that passed.
+	if slices.ContainsFunc(derrs, func(err error) bool { return err != nil }) {
+		valid, at := make([]ObjectToDelete, 0, len(objects)), make([]int, 0, len(objects))
+		for i, obj := range objects {
+			if derrs[i] == nil {
+				valid, at = append(valid, obj), append(at, i)
+			}
+		}
+		if len(valid) > 0 {
+			deleted, errs := z.DeleteObjects(ctx, bucket, valid, opts)
+			for j, i := range at {
+				dobjects[i], derrs[i] = deleted[j], errs[j]
+			}
+		}
+		return dobjects, derrs
+	}
+
 	// A single pool locks each name in the set its writers lock it in, one
 	// set at a time. A single node keeps one lock table for all sets, so it
 	// keeps the one bulk lock below: its local lock takes the names one by