fix: reject DeleteObjects requests with more than 1000 keys

* fix: reject DeleteObjects requests with more than 1000 keys

A DeleteObjects request may name at most 1000 keys; S3 rejects anything
larger with 400 InvalidRequest. The handler parsed the body and passed
every key through authorization and on to the backend, so a 1001-key
request was processed instead of being refused.

Check the count right after the body is parsed, before authorization and
before anything reaches the backend, so an over-limit batch can't be
applied partially. Exactly 1000 keys still succeed, and an empty delete
list keeps its current behavior (200 with an empty result).

Fixes #2196

* fix: return MalformedXML for an oversized DeleteObjects request

A DeleteObjects request naming more than 1000 keys is answered with
MalformedXML, not InvalidRequest. Update the check and the unit and
integration expectations.

---------

Co-authored-by: Tung Lam <lamphamabtung96@gmail.com>

s3api/controllers/base.go

@@ -54,6 +54,7 @@ const (
 	minPartNumber                = 1
 	maxPartNumber                = 10000
 	maxWebsiteConfigurationBytes = 131072
+	maxDeleteObjects             = 1000
 
 	defaultRegion      = "us-east-1"
 	defaultContentType = "binary/octet-stream"

s3api/controllers/bucket-post.go

@@ -56,6 +56,19 @@ func (c S3ApiController) DeleteObjects(ctx fiber.Ctx) (*Response, error) {
 		}, s3err.GetAPIError(s3err.ErrInvalidRequest)
 	}
 
+	// S3 caps a single DeleteObjects request at 1000 keys. Reject an
+	// over-limit request before authorization and before anything reaches
+	// the backend, so a too-large batch can't be applied partially.
+	if len(dObj.Objects) > maxDeleteObjects {
+		debuglogger.Logf("delete objects: %d keys exceeds the limit of %d",
+			len(dObj.Objects), maxDeleteObjects)
+		return &Response{
+			MetaOpts: &MetaOptions{
+				BucketOwner: parsedAcl.Owner,
+			},
+		}, s3err.GetAPIError(s3err.ErrMalformedXML)
+	}
+
 	// checkErrs holds one entry per requested object — nil where it may
 	// proceed to the backend, an AWS-shaped denial otherwise. DeleteObjects
 	// supports partial success, so a denial on one object (policy or object

s3api/controllers/bucket-post_test.go

@@ -19,6 +19,7 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"encoding/xml"
+	"fmt"
 	"io"
 	"net/http"
 	"strings"
@@ -58,6 +59,24 @@ func TestS3ApiController_DeleteObjects(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
+	// 1000 keys is the S3 limit for a single DeleteObjects request; 1001
+	// must be rejected before any of them reaches the backend. The counts
+	// are spelled out rather than derived from the limit constant so the
+	// test pins the documented boundary.
+	const keyLimit = 1000
+	keyLimitObjs := make([]types.ObjectIdentifier, keyLimit+1)
+	for i := range keyLimitObjs {
+		keyLimitObjs[i] = types.ObjectIdentifier{Key: utils.GetStringPtr(fmt.Sprintf("key-%d", i))}
+	}
+	atLimitBody, err := xml.Marshal(s3response.DeleteObjects{Objects: keyLimitObjs[:keyLimit]})
+	assert.NoError(t, err)
+
+	overLimitBody, err := xml.Marshal(s3response.DeleteObjects{Objects: keyLimitObjs})
+	assert.NoError(t, err)
+
+	emptyBody, err := xml.Marshal(s3response.DeleteObjects{Objects: []types.ObjectIdentifier{}})
+	assert.NoError(t, err)
+
 	lockConfig, err := json.Marshal(auth.BucketLockConfig{Enabled: true})
 	assert.NoError(t, err)
 
@@ -157,6 +176,70 @@ func TestS3ApiController_DeleteObjects(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "empty delete list",
+			input: testInput{
+				locals:       defaultLocals,
+				body:         emptyBody,
+				extraMockErr: s3err.GetAPIError(s3err.ErrObjectLockConfigurationNotFound),
+			},
+			output: testOutput{
+				response: &Response{
+					Data: s3response.DeleteResult{},
+					MetaOpts: &MetaOptions{
+						BucketOwner: "root",
+						EventName:   s3event.EventObjectRemovedDeleteObjects,
+						ObjectCount: 0,
+					},
+				},
+			},
+		},
+		{
+			name: "exactly at the 1000 key limit",
+			input: testInput{
+				locals:       defaultLocals,
+				body:         atLimitBody,
+				extraMockErr: s3err.GetAPIError(s3err.ErrObjectLockConfigurationNotFound),
+			},
+			output: testOutput{
+				response: &Response{
+					Data: s3response.DeleteResult{},
+					MetaOpts: &MetaOptions{
+						BucketOwner: "root",
+						EventName:   s3event.EventObjectRemovedDeleteObjects,
+						ObjectCount: 1000,
+					},
+				},
+			},
+			configureMock: func(be *BackendMock) {
+				be.DeleteObjectsFunc = func(contextMoqParam context.Context, deleteObjectsInput *s3.DeleteObjectsInput) (s3response.DeleteResult, error) {
+					assert.Len(t, deleteObjectsInput.Delete.Objects, 1000)
+					return s3response.DeleteResult{}, nil
+				}
+			},
+		},
+		{
+			name: "over the 1000 key limit",
+			input: testInput{
+				locals:       defaultLocals,
+				body:         overLimitBody,
+				extraMockErr: s3err.GetAPIError(s3err.ErrObjectLockConfigurationNotFound),
+			},
+			output: testOutput{
+				response: &Response{
+					MetaOpts: &MetaOptions{
+						BucketOwner: "root",
+					},
+				},
+				err: s3err.GetAPIError(s3err.ErrMalformedXML),
+			},
+			configureMock: func(be *BackendMock) {
+				be.DeleteObjectsFunc = func(contextMoqParam context.Context, deleteObjectsInput *s3.DeleteObjectsInput) (s3response.DeleteResult, error) {
+					t.Error("backend DeleteObjects called for an over-limit request")
+					return s3response.DeleteResult{}, nil
+				}
+			},
+		},
 		{
 			name: "partial success: one object locked, one succeeds",
 			input: testInput{