Merge pull request #104 from pgsty/codex/issue-58-delete-version-authz

fix: authorize explicit version deletes with DeleteObjectVersion

buildscripts/rebrand-guard/compat-baseline.json

@@ -763,6 +763,7 @@
     "/debug/go",
     "/del-config-kv",
     "/delete-service-account",
+    "/deny/*",
     "/describe-job",
     "/dev/0",
     "/dev/1",

cmd/auth-handler.go

@@ -363,17 +363,6 @@ func checkRequestAuthTypeWithRequestTags(ctx context.Context, r *http.Request, a
 	return authorizeRequestWithTags(ctx, r, action, "", requestTags)
 }
 
-// checkRequestAuthTypeWithVID is similar to checkRequestAuthType
-// passes versionID additionally.
-func checkRequestAuthTypeWithVID(ctx context.Context, r *http.Request, action policy.Action, bucketName, objectName, versionID string) (s3Err APIErrorCode) {
-	logger.GetReqInfo(ctx).BucketName = bucketName
-	logger.GetReqInfo(ctx).ObjectName = objectName
-	logger.GetReqInfo(ctx).VersionID = versionID
-
-	_, _, s3Err = checkRequestAuthTypeCredential(ctx, r, action)
-	return s3Err
-}
-
 func authenticateRequest(ctx context.Context, r *http.Request, action policy.Action) (s3Err APIErrorCode) {
 	if logger.GetReqInfo(ctx) == nil {
 		bugLogIf(ctx, errors.New("unexpected context.Context does not have a logger.ReqInfo"), logger.ErrorKind)
@@ -439,6 +428,23 @@ func authorizeRequest(ctx context.Context, r *http.Request, action policy.Action
 	return authorizeRequestWithExistingTags(ctx, r, action, "")
 }
 
+func deleteObjectAction(versionID string) policy.Action {
+	if versionID != "" {
+		return policy.DeleteObjectVersionAction
+	}
+	return policy.DeleteObjectAction
+}
+
+func actionUsesObjectVersion(action policy.Action) bool {
+	switch action {
+	case policy.DeleteObjectAction, policy.DeleteObjectVersionAction,
+		policy.ReplicateDeleteAction, policy.BypassGovernanceRetentionAction:
+		return true
+	default:
+		return false
+	}
+}
+
 func authorizeRequestWithExistingTags(ctx context.Context, r *http.Request, action policy.Action, existingTags string) (s3Err APIErrorCode) {
 	return authorizeRequestWithTags(ctx, r, action, existingTags, nil)
 }
@@ -457,7 +463,7 @@ func authorizeRequestWithTags(ctx context.Context, r *http.Request, action polic
 	versionID := reqInfo.VersionID
 	conditionValuesForAuth := func(locationConstraint string, credentials auth.Credentials) map[string][]string {
 		values := getConditionValuesWithTags(r, locationConstraint, credentials, existingTags, requestTags)
-		if action == policy.DeleteObjectAction {
+		if actionUsesObjectVersion(action) {
 			// DeleteObjects carries the effective version ID in each XML object,
 			// not in the request query. Keep authorization scoped to that entry.
 			if versionID == "" {
@@ -503,21 +509,6 @@ func authorizeRequestWithTags(ctx context.Context, r *http.Request, action polic
 
 		return ErrAccessDenied
 	}
-	if action == policy.DeleteObjectAction && versionID != "" {
-		if !globalIAMSys.IsAllowed(policy.Args{
-			AccountName:     cred.AccessKey,
-			Groups:          cred.Groups,
-			Action:          policy.Action(policy.DeleteObjectVersionAction),
-			BucketName:      bucket,
-			ConditionValues: conditionValuesForAuth("", cred),
-			ObjectName:      object,
-			IsOwner:         owner,
-			Claims:          cred.Claims,
-			DenyOnly:        true,
-		}) { // Request is not allowed if Deny action on DeleteObjectVersionAction
-			return ErrAccessDenied
-		}
-	}
 	if globalIAMSys.IsAllowed(policy.Args{
 		AccountName:     cred.AccessKey,
 		Groups:          cred.Groups,
@@ -553,6 +544,40 @@ func authorizeRequestWithTags(ctx context.Context, r *http.Request, action polic
 	return ErrAccessDenied
 }
 
+// authorizeReplicationDelete preserves the established target-credential
+// contract for trusted replication: DeleteObject and ReplicateDelete must be
+// allowed, while an explicit DeleteObjectVersion deny still blocks a named
+// version. Ordinary S3 requests never use this compatibility path.
+func authorizeReplicationDelete(ctx context.Context, r *http.Request) APIErrorCode {
+	if s3Err := authorizeRequest(ctx, r, policy.DeleteObjectAction); s3Err != ErrNone {
+		return s3Err
+	}
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		return ErrAccessDenied
+	}
+	if reqInfo.VersionID == "" {
+		return ErrNone
+	}
+	cred := reqInfo.Cred
+	values := getConditionValuesWithTags(r, "", cred, "", nil)
+	values["versionid"] = []string{reqInfo.VersionID}
+	if !globalIAMSys.IsAllowed(policy.Args{
+		AccountName:     cred.AccessKey,
+		Groups:          cred.Groups,
+		Action:          policy.DeleteObjectVersionAction,
+		BucketName:      reqInfo.BucketName,
+		ConditionValues: values,
+		ObjectName:      reqInfo.ObjectName,
+		IsOwner:         reqInfo.Owner,
+		Claims:          cred.Claims,
+		DenyOnly:        true,
+	}) {
+		return ErrAccessDenied
+	}
+	return ErrNone
+}
+
 // Check request auth type verifies the incoming http request
 //   - validates the request signature
 //   - validates the policy action if anonymous tests bucket policies if any,

cmd/bucket-handlers.go

@@ -463,13 +463,20 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	// Make sure to update context to print ObjectNames for multi objects.
 	ctx = updateReqContext(ctx, objects...)
 
-	// Call checkRequestAuthType to populate ReqInfo.AccessKey before GetBucketInfo()
-	// Ignore errors here to preserve the S3 error behavior of GetBucketInfo()
-	checkRequestAuthType(ctx, r, policy.DeleteObjectAction, bucket, "")
-
 	deleteObjectsFn := objectAPI.DeleteObjects
 
-	// Return Malformed XML as S3 spec if the number of objects is empty
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+		return
+	}
+	reqInfo.BucketName = bucket
+	reqInfo.ObjectName = ""
+	if s3Err := authenticateRequest(ctx, r, policy.DeleteObjectAction); s3Err != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
+		return
+	}
+	// Return Malformed XML as S3 spec if the number of objects is empty.
 	if len(deleteObjectsReq.Objects) == 0 || len(deleteObjectsReq.Objects) > maxDeleteList {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMalformedXML), r.URL)
 		return
@@ -499,7 +506,9 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter,
 	vc, _ := globalBucketVersioningSys.Get(bucket)
 	oss := make([]*objSweeper, len(deleteObjectsReq.Objects))
 	for index, object := range deleteObjectsReq.Objects {
-		if apiErrCode := checkRequestAuthTypeWithVID(ctx, r, policy.DeleteObjectAction, bucket, object.ObjectName, object.VersionID); apiErrCode != ErrNone {
+		reqInfo.ObjectName = object.ObjectName
+		reqInfo.VersionID = object.VersionID
+		if apiErrCode := authorizeRequest(ctx, r, deleteObjectAction(object.VersionID)); apiErrCode != ErrNone {
 			if apiErrCode == ErrSignatureDoesNotMatch || apiErrCode == ErrInvalidAccessKeyID {
 				writeErrorResponse(ctx, w, errorCodes.ToAPIErr(apiErrCode), r.URL)
 				return

cmd/bucket-handlers_test.go

@@ -1018,14 +1018,23 @@ func testAPIDeleteMultipleObjectsVersionIDNullCondition(obj ObjectLayer, instanc
 
 	policyBytes := fmt.Appendf(nil, `{
 		"Version":"2012-10-17",
-		"Statement":[{
-			"Effect":"Allow",
-			"Principal":"*",
-			"Action":"s3:DeleteObject",
-			"Resource":"arn:aws:s3:::%s/*",
-			"Condition":{"Null":{"s3:versionid":"true"}}
-		}]
-	}`, bucketName)
+		"Statement":[
+			{
+				"Effect":"Allow",
+				"Principal":"*",
+				"Action":"s3:DeleteObject",
+				"Resource":"arn:aws:s3:::%s/*",
+				"Condition":{"Null":{"s3:versionid":"true"}}
+			},
+			{
+				"Effect":"Allow",
+				"Principal":"*",
+				"Action":"s3:DeleteObjectVersion",
+				"Resource":"arn:aws:s3:::%s/*",
+				"Condition":{"StringEquals":{"s3:versionid":"%s"}}
+			}
+		]
+	}`, bucketName, bucketName, versionIDs["with-version-id"])
 	policyReq, err := newTestSignedRequestV4(http.MethodPut, getPutPolicyURL("", bucketName), int64(len(policyBytes)),
 		bytes.NewReader(policyBytes), credentials.AccessKey, credentials.SecretKey, nil)
 	if err != nil {
@@ -1071,29 +1080,30 @@ func testAPIDeleteMultipleObjectsVersionIDNullCondition(obj ObjectLayer, instanc
 			t.Errorf("%s: %q was not a successful delete-marker creation: %+v", instanceType, objectName, response.DeletedObjects)
 		}
 	}
-	if len(deleted) != 2 {
+	if object, ok := deleted["with-version-id"]; !ok || object.VersionID != versionIDs["with-version-id"] {
+		t.Errorf("%s: matching explicit version was not deleted: %+v", instanceType, response.DeletedObjects)
+	}
+	if len(deleted) != 3 {
 		t.Errorf("%s: unexpected deleted objects: %+v", instanceType, response.DeletedObjects)
 	}
 	errorsByKey := make(map[string]DeleteError, len(response.Errors))
 	for _, deleteErr := range response.Errors {
 		errorsByKey[deleteErr.Key] = deleteErr
 	}
-	for objectName, versionID := range map[string]string{
-		"with-version-id":      versionIDs["with-version-id"],
-		"with-null-version-id": nullVersionID,
-	} {
+	for objectName, versionID := range map[string]string{"with-null-version-id": nullVersionID} {
 		deleteErr, ok := errorsByKey[objectName]
 		if !ok || deleteErr.VersionID != versionID || deleteErr.Code != errorCodes[ErrAccessDenied].Code {
 			t.Errorf("%s: %q did not return AccessDenied for version %q: %+v", instanceType, objectName, versionID, response.Errors)
 		}
 	}
-	if len(errorsByKey) != 2 {
+	if len(errorsByKey) != 1 {
 		t.Errorf("%s: unexpected delete errors: %+v", instanceType, response.Errors)
 	}
 
-	// A simple delete adds a marker and keeps the old version. The explicitly
-	// named version must also remain because its policy condition did not match.
-	for objectName, versionID := range versionIDs {
+	// A simple delete adds a marker and keeps the old version. The null-version
+	// delete remains denied because its per-entry condition does not match.
+	for _, objectName := range []string{"without-version-id-before", "without-version-id-after", "with-null-version-id"} {
+		versionID := versionIDs[objectName]
 		if _, err = obj.GetObjectInfo(t.Context(), bucketName, objectName, ObjectOptions{VersionID: versionID}); err != nil {
 			t.Errorf("%s: version %s of %q was not preserved: %v", instanceType, versionID, objectName, err)
 		}
@@ -1103,7 +1113,10 @@ func testAPIDeleteMultipleObjectsVersionIDNullCondition(obj ObjectLayer, instanc
 			t.Errorf("%s: simple delete of %q did not hide the latest object behind a delete marker: %v", instanceType, objectName, err)
 		}
 	}
-	for _, objectName := range []string{"with-version-id", "with-null-version-id"} {
+	if _, err = obj.GetObjectInfo(t.Context(), bucketName, "with-version-id", ObjectOptions{VersionID: versionIDs["with-version-id"]}); !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+		t.Errorf("%s: matching explicit version still exists: %v", instanceType, err)
+	}
+	for _, objectName := range []string{"with-null-version-id"} {
 		if info, err := obj.GetObjectInfo(t.Context(), bucketName, objectName, ObjectOptions{}); err != nil {
 			t.Errorf("%s: denied version delete removed latest %q: %v", instanceType, objectName, err)
 		} else if info.VersionID != versionIDs[objectName] {

cmd/delete-version-authz_test.go

@@ -0,0 +1,402 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+// Copyright (c) 2026 PGSTY
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+package cmd
+
+import (
+	"bytes"
+	"encoding/xml"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/auth"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/minio/pkg/v3/policy"
+)
+
+func TestDeleteObjectAction(t *testing.T) {
+	for _, test := range []struct {
+		versionID string
+		want      policy.Action
+	}{
+		{want: policy.DeleteObjectAction},
+		{versionID: nullVersionID, want: policy.DeleteObjectVersionAction},
+		{versionID: mustGetUUID(), want: policy.DeleteObjectVersionAction},
+		{versionID: " ", want: policy.DeleteObjectVersionAction},
+	} {
+		if got := deleteObjectAction(test.versionID); got != test.want {
+			t.Errorf("deleteObjectAction(%q) = %s, want %s", test.versionID, got, test.want)
+		}
+	}
+}
+
+func TestAPIDeleteObjectVersionAuthorization(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		objAPITest:        testAPIDeleteObjectVersionAuthorization,
+		endpoints:         []string{"DeleteObject"},
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+	})
+}
+
+func TestAPIDeleteMultipleObjectsVersionAuthorization(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		objAPITest:        testAPIDeleteMultipleObjectsVersionAuthorization,
+		endpoints:         []string{"DeleteMultipleObjects"},
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+	})
+}
+
+func testAPIDeleteMultipleObjectsVersionAuthorization(obj ObjectLayer, instanceType, bucket string,
+	apiRouter http.Handler, _ auth.Credentials, t *testing.T,
+) {
+	deleteOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:DeleteObject"`)
+	versionOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:DeleteObjectVersion"`)
+	payload := []byte("multi delete version authorization")
+
+	put := func(t *testing.T, object string, versioned bool) string {
+		t.Helper()
+		info, err := obj.PutObject(t.Context(), bucket, object,
+			mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), ObjectOptions{Versioned: versioned})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return info.VersionID
+	}
+	request := func(t *testing.T, prefix string, creds auth.Credentials) (DeleteObjectsResponse, map[string]string) {
+		t.Helper()
+		versions := map[string]string{
+			prefix + "simple":   put(t, prefix+"simple", true),
+			prefix + "explicit": put(t, prefix+"explicit", true),
+			prefix + "null":     put(t, prefix+"null", false),
+		}
+		body := encodeResponse(DeleteObjectsRequest{Objects: []ObjectToDelete{
+			{ObjectV: ObjectV{ObjectName: prefix + "simple"}},
+			{ObjectV: ObjectV{ObjectName: prefix + "explicit", VersionID: versions[prefix+"explicit"]}},
+			{ObjectV: ObjectV{ObjectName: prefix + "null", VersionID: nullVersionID}},
+			{ObjectV: ObjectV{ObjectName: prefix + "bad", VersionID: "not-a-uuid"}},
+		}})
+		target := getDeleteMultipleObjectsURL("", bucket) + "&versionId=query-level-decoy"
+		req, err := newTestSignedRequestV4(http.MethodPost, target, int64(len(body)), bytes.NewReader(body),
+			creds.AccessKey, creds.SecretKey, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		apiRouter.ServeHTTP(rec, req)
+		if rec.Code != http.StatusOK {
+			t.Fatalf("%s: multi-delete status %d: %s", instanceType, rec.Code, rec.Body.String())
+		}
+		var response DeleteObjectsResponse
+		if err = xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
+			t.Fatalf("decode response: %v: %s", err, rec.Body.String())
+		}
+		return response, versions
+	}
+	responseMap := func(response DeleteObjectsResponse) (map[string]DeletedObject, map[string]DeleteError) {
+		deleted := make(map[string]DeletedObject, len(response.DeletedObjects))
+		for _, object := range response.DeletedObjects {
+			deleted[object.ObjectName] = object
+		}
+		errs := make(map[string]DeleteError, len(response.Errors))
+		for _, deleteErr := range response.Errors {
+			errs[deleteErr.Key] = deleteErr
+		}
+		return deleted, errs
+	}
+
+	t.Run("DeleteObject only", func(t *testing.T) {
+		prefix := "multi-delete-only/"
+		response, versions := request(t, prefix, deleteOnly)
+		deleted, errs := responseMap(response)
+		if object, ok := deleted[prefix+"simple"]; !ok || !object.DeleteMarker {
+			t.Fatalf("simple delete did not create a marker: %+v", response)
+		}
+		for _, object := range []string{"explicit", "null", "bad"} {
+			if got := errs[prefix+object].Code; got != errorCodes[ErrAccessDenied].Code {
+				t.Errorf("%s error = %q, want AccessDenied", object, got)
+			}
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, prefix+"explicit", ObjectOptions{VersionID: versions[prefix+"explicit"]}); err != nil {
+			t.Fatalf("denied explicit delete removed its version: %v", err)
+		}
+	})
+
+	t.Run("DeleteObjectVersion only", func(t *testing.T) {
+		prefix := "multi-version-only/"
+		response, _ := request(t, prefix, versionOnly)
+		deleted, errs := responseMap(response)
+		for _, object := range []string{"explicit", "null"} {
+			if _, ok := deleted[prefix+object]; !ok {
+				t.Errorf("%s was not deleted: %+v", object, response)
+			}
+		}
+		if got := errs[prefix+"simple"].Code; got != errorCodes[ErrAccessDenied].Code {
+			t.Errorf("simple error = %q, want AccessDenied", got)
+		}
+		if got := errs[prefix+"bad"].Code; got != errorCodes[ErrNoSuchVersion].Code {
+			t.Errorf("bad UUID error = %q, want NoSuchVersion", got)
+		}
+	})
+}
+
+func testAPIDeleteObjectVersionAuthorization(obj ObjectLayer, instanceType, bucket string,
+	apiRouter http.Handler, _ auth.Credentials, t *testing.T,
+) {
+	deleteOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:DeleteObject"`)
+	versionOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:DeleteObjectVersion"`)
+	payload := []byte("delete version authorization")
+
+	put := func(t *testing.T, object string, versioned bool) string {
+		t.Helper()
+		info, err := obj.PutObject(t.Context(), bucket, object,
+			mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), ObjectOptions{Versioned: versioned})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if versioned && info.VersionID == "" {
+			t.Fatalf("%s: versioned PUT returned an empty version ID", instanceType)
+		}
+		return info.VersionID
+	}
+	remove := func(t *testing.T, object, versionID string, creds auth.Credentials) *httptest.ResponseRecorder {
+		t.Helper()
+		target := getDeleteObjectURL("", bucket, object)
+		if versionID != "" {
+			target += "?" + url.Values{xhttp.VersionID: {versionID}}.Encode()
+		}
+		req, err := newTestSignedRequestV4(http.MethodDelete, target, 0, nil, creds.AccessKey, creds.SecretKey, nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		apiRouter.ServeHTTP(rec, req)
+		return rec
+	}
+
+	t.Run("version permission deletes an explicit version", func(t *testing.T) {
+		object := "delete-authz/version-only-explicit"
+		versionID := put(t, object, true)
+		if rec := remove(t, object, versionID, versionOnly); rec.Code != http.StatusNoContent {
+			t.Fatalf("status %d, want 204: %s", rec.Code, rec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			t.Fatalf("explicit version still exists: %v", err)
+		}
+	})
+
+	t.Run("version permission cannot create a delete marker", func(t *testing.T) {
+		object := "delete-authz/version-only-simple"
+		versionID := put(t, object, true)
+		if rec := remove(t, object, "", versionOnly); rec.Code != http.StatusForbidden {
+			t.Fatalf("status %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID}); err != nil {
+			t.Fatalf("denied simple delete removed the version: %v", err)
+		}
+	})
+
+	t.Run("object permission cannot delete an explicit version", func(t *testing.T) {
+		object := "delete-authz/delete-only-explicit"
+		versionID := put(t, object, true)
+		if rec := remove(t, object, versionID, deleteOnly); rec.Code != http.StatusForbidden {
+			t.Fatalf("status %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID}); err != nil {
+			t.Fatalf("denied version delete removed the version: %v", err)
+		}
+	})
+
+	t.Run("object permission creates a delete marker", func(t *testing.T) {
+		object := "delete-authz/delete-only-simple"
+		versionID := put(t, object, true)
+		if rec := remove(t, object, "", deleteOnly); rec.Code != http.StatusNoContent {
+			t.Fatalf("status %d, want 204: %s", rec.Code, rec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID}); err != nil {
+			t.Fatalf("simple delete removed the old version: %v", err)
+		}
+	})
+
+	t.Run("null is an explicit version", func(t *testing.T) {
+		object := "delete-authz/null-version"
+		if versionID := put(t, object, false); versionID != "" {
+			t.Fatalf("unversioned PUT returned version ID %q", versionID)
+		}
+		if rec := remove(t, object, nullVersionID, versionOnly); rec.Code != http.StatusNoContent {
+			t.Fatalf("status %d, want 204: %s", rec.Code, rec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: nullVersionID}); !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			t.Fatalf("null version still exists: %v", err)
+		}
+	})
+
+	t.Run("authorization precedes invalid version parsing", func(t *testing.T) {
+		object := "delete-authz/invalid-version"
+		if rec := remove(t, object, "not-a-uuid", deleteOnly); rec.Code != http.StatusForbidden {
+			t.Fatalf("delete-only status %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+		if rec := remove(t, object, "not-a-uuid", versionOnly); rec.Code != http.StatusBadRequest {
+			t.Fatalf("version-only status %d, want 400: %s", rec.Code, rec.Body.String())
+		}
+	})
+
+	t.Run("padded version uses the effective ID", func(t *testing.T) {
+		object := "delete-authz/padded-version"
+		versionID := put(t, object, true)
+		if rec := remove(t, object, versionID+" ", versionOnly); rec.Code != http.StatusNoContent {
+			t.Fatalf("status %d, want 204: %s", rec.Code, rec.Body.String())
+		}
+	})
+}
+
+func TestAPIDeleteObjectVersionDenyAndReplicationCompatibility(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		objAPITest:        testAPIDeleteObjectVersionDenyAndReplicationCompatibility,
+		endpoints:         []string{"DeleteObject"},
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+	})
+}
+
+func testAPIDeleteObjectVersionDenyAndReplicationCompatibility(obj ObjectLayer, instanceType, bucket string,
+	apiRouter http.Handler, _ auth.Credentials, t *testing.T,
+) {
+	payload := []byte("delete version deny compatibility")
+	put := func(t *testing.T, object string) string {
+		t.Helper()
+		info, err := obj.PutObject(t.Context(), bucket, object,
+			mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), ObjectOptions{Versioned: true})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return info.VersionID
+	}
+	request := func(t *testing.T, object, versionID string, creds auth.Credentials, replicationRequest bool) *httptest.ResponseRecorder {
+		t.Helper()
+		target := getDeleteObjectURL("", bucket, object)
+		if versionID != "" {
+			target += "?" + url.Values{xhttp.VersionID: {versionID}}.Encode()
+		}
+		var headers map[string]string
+		if replicationRequest {
+			headers = map[string]string{
+				xhttp.MinIOSourceReplicationRequest: "true",
+				xhttp.AmzBucketReplicationStatus:    "REPLICA",
+				xhttp.MinIOSourceDeleteMarker:       "false",
+				xhttp.MinIOSourceMTime:              UTCNow().Format(time.RFC3339Nano),
+			}
+		}
+		req, err := newTestSignedRequestV4(http.MethodDelete, target, 0, nil, creds.AccessKey, creds.SecretKey, headers)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		apiRouter.ServeHTTP(rec, req)
+		return rec
+	}
+
+	minimal := newDeleteAuthzPolicyUser(t, instanceType, bucket, `[
+		{"Effect":"Allow","Action":["s3:DeleteObject","s3:ReplicateDelete"],"Resource":["arn:aws:s3:::`+bucket+`/*"]}
+	]`)
+	denied := newDeleteAuthzPolicyUser(t, instanceType, bucket, `[
+		{"Effect":"Allow","Action":["s3:DeleteObject","s3:DeleteObjectVersion","s3:ReplicateDelete"],"Resource":["arn:aws:s3:::`+bucket+`/*"]},
+		{"Effect":"Deny","Action":"s3:DeleteObjectVersion","Resource":"arn:aws:s3:::`+bucket+`/deny/*"}
+	]`)
+	deleteOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:DeleteObject"`)
+
+	t.Run("ordinary explicit deny wins", func(t *testing.T) {
+		object := "deny/ordinary"
+		versionID := put(t, object)
+		if rec := request(t, object, versionID, denied, false); rec.Code != http.StatusForbidden {
+			t.Fatalf("status %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+	})
+
+	t.Run("version deny does not block a simple delete", func(t *testing.T) {
+		object := "deny/simple"
+		put(t, object)
+		if rec := request(t, object, "", denied, false); rec.Code != http.StatusNoContent {
+			t.Fatalf("status %d, want 204: %s", rec.Code, rec.Body.String())
+		}
+	})
+
+	t.Run("replication keeps minimal target policy", func(t *testing.T) {
+		object := "replication/minimal"
+		versionID := put(t, object)
+		if rec := request(t, object, versionID, minimal, true); rec.Code != http.StatusNoContent {
+			t.Fatalf("status %d, want 204: %s", rec.Code, rec.Body.String())
+		}
+	})
+
+	t.Run("replication preserves explicit version deny", func(t *testing.T) {
+		object := "deny/replication"
+		versionID := put(t, object)
+		if rec := request(t, object, versionID, denied, true); rec.Code != http.StatusForbidden {
+			t.Fatalf("status %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+	})
+
+	t.Run("marker alone cannot enter the replication path", func(t *testing.T) {
+		object := "replication/fake-marker"
+		versionID := put(t, object)
+		target := getDeleteObjectURL("", bucket, object) + "?" + url.Values{xhttp.VersionID: {versionID}}.Encode()
+		req, err := newTestSignedRequestV4(http.MethodDelete, target, 0, nil, deleteOnly.AccessKey, deleteOnly.SecretKey,
+			map[string]string{xhttp.MinIOSourceReplicationRequest: "true"})
+		if err != nil {
+			t.Fatal(err)
+		}
+		rec := httptest.NewRecorder()
+		apiRouter.ServeHTTP(rec, req)
+		if rec.Code != http.StatusForbidden {
+			t.Fatalf("status %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+		if _, err = obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID}); err != nil {
+			t.Fatalf("fake marker removed the version: %v", err)
+		}
+	})
+}
+
+func newDeleteAuthzPolicyUser(t *testing.T, instanceType, bucket, statements string) auth.Credentials {
+	t.Helper()
+	accessKey, secretKey, err := auth.GenerateCredentials()
+	if err != nil {
+		t.Fatalf("%s: generate credentials: %v", instanceType, err)
+	}
+	creds := auth.Credentials{AccessKey: accessKey, SecretKey: secretKey}
+	if _, err = globalIAMSys.CreateUser(t.Context(), accessKey, madmin.AddOrUpdateUserReq{
+		SecretKey: secretKey,
+		Status:    madmin.AccountEnabled,
+	}); err != nil {
+		t.Fatalf("%s: create delete authz user: %v", instanceType, err)
+	}
+	policyJSON := `{"Version":"2012-10-17","Statement":` + statements + `}`
+	parsed, err := policy.ParseConfig(strings.NewReader(policyJSON))
+	if err != nil {
+		t.Fatalf("%s: parse delete authz policy: %v", instanceType, err)
+	}
+	policyName := "delete-version-authz-" + mustGetUUID()
+	if _, err = globalIAMSys.SetPolicy(t.Context(), policyName, *parsed); err != nil {
+		t.Fatalf("%s: install delete authz policy: %v", instanceType, err)
+	}
+	if _, err = globalIAMSys.PolicyDBSet(t.Context(), accessKey, policyName, regUser, false); err != nil {
+		t.Fatalf("%s: attach delete authz policy: %v", instanceType, err)
+	}
+	return creds
+}

cmd/object-handlers.go

@@ -2821,7 +2821,14 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		return
 	}
 
-	if s3Error := checkRequestAuthType(ctx, r, policy.DeleteObjectAction, bucket, object); s3Error != ErrNone {
+	reqInfo := logger.GetReqInfo(ctx)
+	if reqInfo == nil {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+		return
+	}
+	reqInfo.BucketName = bucket
+	reqInfo.ObjectName = object
+	if s3Error := authenticateRequest(ctx, r, policy.DeleteObjectAction); s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
 		return
 	}
@@ -2835,12 +2842,22 @@ func (api objectAPIHandlers) DeleteObjectHandler(w http.ResponseWriter, r *http.
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
 		return
 	}
+	trustedReplication := markerExact && replicationPermitted
+	var s3Error APIErrorCode
+	if trustedReplication {
+		s3Error = authorizeReplicationDelete(ctx, r)
+	} else {
+		s3Error = authorizeRequest(ctx, r, deleteObjectAction(reqInfo.VersionID))
+	}
+	if s3Error != ErrNone {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
+		return
+	}
 	if _, ok := r.Header[xhttp.MinIOSourceReplicationCheck]; ok {
 		// requests to just validate replication settings and permissions are not allowed to delete data
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrReplicationPermissionCheckError), r.URL)
 		return
 	}
-	trustedReplication := markerExact && replicationPermitted
 	replica := trustedReplication && rawReplica
 	if hasReplicationRequestHeaders(r.Header) {
 		ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replica)

docs/bucket/replication/README.md

@@ -96,6 +96,12 @@ The access key provided for the replication *target* cluster should have these m
 
 Please note that the permissions required by the admin user on the target cluster can be more fine grained to exclude permissions like "s3:ReplicateDelete", "s3:GetBucketObjectLockConfiguration" etc depending on whether delete replication rules are set up or if object locking is disabled on `destbucket`. The above policies assume that replication of objects, tags and delete marker replication are all enabled on object lock enabled buckets. A sample script to setup replication is provided [here](https://github.com/pgsty/silo/blob/main/docs/bucket/replication/setup_replication.sh)
 
+The target replication credential continues to authorize replicated deletes with
+`s3:DeleteObject` plus `s3:ReplicateDelete`; it does not need
+`s3:DeleteObjectVersion`. This internal receiver contract is deliberately
+separate from ordinary S3 requests: a client deleting an explicitly named
+version, including `versionId=null`, must have `s3:DeleteObjectVersion`.
+
 To set up replication from `srcbucket` on the `mysilo` cluster to `destbucket`
 on a target Silo cluster at `https://replica-endpoint:9000`, use:
 ```
@@ -200,6 +206,10 @@ To add a replication rule allowing both delete marker replication, versioned del
 
 Additional permission of "s3:ReplicateDelete" action would need to be specified on the access key configured for the target cluster if Delete Marker replication or versioned delete replication is enabled.
 
+An explicit deny on `s3:DeleteObjectVersion` still blocks the corresponding
+replicated version purge. An allow is not otherwise required for the target
+replication credential.
+
 ```
 mc replicate add mysilo/srcbucket/Tax --priority 1 --remote-bucket `remote-target` --tags "Year=2019&Company=AcmeCorp" --storage-class "STANDARD" --replicate "delete,delete-marker"
 Replication configuration applied successfully to mysilo/srcbucket.

docs/bucket/replication/delete-replication.sh

@@ -88,11 +88,7 @@ echo "=== mysilo2"
 
 versionId="$(./mc ls --json --versions mysilo1/testbucket/dir/ | tail -n1 | jq -r .versionId)"
 
-export AWS_ACCESS_KEY_ID=minioadmin
-export AWS_SECRET_ACCESS_KEY=minioadmin
-export AWS_REGION=us-east-1
-
-aws s3api --endpoint-url http://localhost:9001 delete-object --bucket testbucket --key dir/file --version-id "$versionId"
+./mc rm --version-id "$versionId" mysilo1/testbucket/dir/file
 
 ./mc ls -r --versions mysilo1/testbucket >/tmp/mysilo1.txt
 ./mc ls -r --versions mysilo2/testbucket >/tmp/mysilo2.txt
@@ -117,6 +113,76 @@ if [ $ret -ne 0 ]; then
 	exit 1
 fi
 
+# Verify the documented least-privilege target policy. Explicit version
+# deletion on the receiver is replication traffic, so the target credential
+# needs DeleteObject + ReplicateDelete but not DeleteObjectVersion.
+./mc mb mysilo1/leastpriv/ mysilo2/leastpriv/ --with-versioning
+./mc admin user add mysilo2 repluser repluser123
+cat >/tmp/xl/replpolicy.json <<'EOF'
+{
+ "Version": "2012-10-17",
+ "Statement": [
+  {
+   "Effect": "Allow",
+   "Action": [
+    "s3:GetReplicationConfiguration",
+    "s3:ListBucket",
+    "s3:ListBucketMultipartUploads",
+    "s3:GetBucketLocation",
+    "s3:GetBucketVersioning"
+   ],
+   "Resource": ["arn:aws:s3:::leastpriv"]
+  },
+  {
+   "Effect": "Allow",
+   "Action": [
+    "s3:GetReplicationConfiguration",
+    "s3:ReplicateTags",
+    "s3:AbortMultipartUpload",
+    "s3:GetObject",
+    "s3:GetObjectVersion",
+    "s3:GetObjectVersionTagging",
+    "s3:PutObject",
+    "s3:DeleteObject",
+    "s3:ReplicateObject",
+    "s3:ReplicateDelete"
+   ],
+   "Resource": ["arn:aws:s3:::leastpriv/*"]
+  }
+ ]
+}
+EOF
+./mc admin policy create mysilo2 replpolicy /tmp/xl/replpolicy.json
+./mc admin policy attach mysilo2 replpolicy --user repluser
+./mc replicate add mysilo1/leastpriv --remote-bucket http://repluser:repluser123@localhost:9002/leastpriv/ --priority 1 --replicate delete,delete-marker
+
+./mc cp README.md mysilo1/leastpriv/dir/file
+./mc cp README.md mysilo1/leastpriv/dir/file
+sleep 1s
+
+leastPrivVersionId="$(./mc ls --json --versions mysilo1/leastpriv/dir/ | tail -n1 | jq -r .versionId)"
+./mc rm --version-id "$leastPrivVersionId" mysilo1/leastpriv/dir/file
+sleep 1s
+./mc ls -r --versions mysilo1/leastpriv >/tmp/leastpriv1.txt
+./mc ls -r --versions mysilo2/leastpriv >/tmp/leastpriv2.txt
+out=$(diff -qpruN /tmp/leastpriv1.txt /tmp/leastpriv2.txt)
+ret=$?
+if [ $ret -ne 0 ]; then
+	echo "BUG: least-privilege version delete did not replicate: $out"
+	exit 1
+fi
+
+./mc rm mysilo1/leastpriv/dir/file
+sleep 1s
+./mc ls -r --versions mysilo1/leastpriv >/tmp/leastpriv1.txt
+./mc ls -r --versions mysilo2/leastpriv >/tmp/leastpriv2.txt
+out=$(diff -qpruN /tmp/leastpriv1.txt /tmp/leastpriv2.txt)
+ret=$?
+if [ $ret -ne 0 ]; then
+	echo "BUG: least-privilege delete marker did not replicate: $out"
+	exit 1
+fi
+
 # Test listing of non replicated permanent deletes
 
 set -x
@@ -129,7 +195,7 @@ versionId="$(./mc ls --json --versions mysilo1/foobucket/dir/ | jq -r .versionId
 
 kill ${pid2} && wait ${pid2} || true
 
-aws s3api --endpoint-url http://localhost:9001 delete-object --bucket foobucket --key dir/file --version-id "$versionId"
+./mc rm --version-id "$versionId" mysilo1/foobucket/dir/file
 
 out="$(./mc ls mysilo1/foobucket/dir/)"
 if [ "$out" != "" ]; then