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.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