fix: reject object lock preflight read failures

Return storage read failures before accepting an object overwrite. Keep the null-version and multipart target checks under the destination write lock, and add slow-body, read-fault, legacy-upload and recovery-collision regressions to the Object Lock race gate.

Signed-off-by: Feng Ruohang <rh@vonng.com>

.github/workflows/go.yml

@@ -102,6 +102,11 @@ jobs:
       - name: Run tag replication tests under race detector
         run: go test -race ./cmd -run '^TestAPITagging' -count=1 -timeout=5m
 
+      - name: Run Object Lock authorization and overwrite tests under race detector
+        env:
+          MINIO_API_REQUESTS_MAX: "10000"
+        run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m
+
   crosscompile:
     name: Cross Compile
     runs-on: ubuntu-latest

Dockerfile.goreleaser

@@ -1,4 +1,4 @@
-FROM golang:1.27.1-alpine@sha256:cf6fca6641884b8433441b2b0652976f975e1d0fdd26d177eaaf8596087f3125 AS curl-build
+FROM golang:1.27.2-alpine@sha256:f92b6ef800e499660581efdabdf25d9d817a9d124eaf900924f0504e7e27e12d AS curl-build
 ARG TARGETARCH
 COPY dockerscripts/build-static-curl.sh /build/build-static-curl
 RUN /bin/sh /build/build-static-curl
@@ -9,7 +9,7 @@ COPY --from=curl-build /go/bin/curl /curl
 COPY --from=curl-build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 ENTRYPOINT ["/curl"]
 
-FROM golang:1.27.1-alpine@sha256:cf6fca6641884b8433441b2b0652976f975e1d0fdd26d177eaaf8596087f3125 AS build
+FROM golang:1.27.2-alpine@sha256:f92b6ef800e499660581efdabdf25d9d817a9d124eaf900924f0504e7e27e12d AS build
 
 ARG TARGETARCH
 

Makefile

@@ -4,7 +4,7 @@ LDFLAGS := $(shell go run buildscripts/gen-ldflags.go)
 
 GOOS ?= $(shell go env GOOS)
 GOARCH ?= $(shell go env GOARCH)
-GOLANGCI_VERSION ?= v2.13.1
+GOLANGCI_VERSION ?= v2.14.0
 
 VERSION ?= $(shell git describe --tags)
 REPO ?= docker.io/pgsty

buildscripts/checkdeps.sh

@@ -7,7 +7,7 @@ _init() {
 
 	## Minimum required versions for build dependencies
 	GIT_VERSION="1.0"
-	GO_VERSION="1.27.1"
+	GO_VERSION="1.27.2"
 	OSX_VERSION="10.8"
 	KNAME=$(uname -s)
 	ARCH=$(uname -m)

cmd/api-errors.go

@@ -550,7 +550,7 @@ var errorCodes = errorCodeMap{
 	},
 	ErrBadDigest: {
 		Code:           "BadDigest",
-		Description:    "The Content-Md5 you specified did not match what we received.",
+		Description:    "The Content-MD5 you specified did not match what we received.",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
 	ErrEntityTooSmall: {

cmd/auth-handler.go

@@ -28,13 +28,10 @@ import (
 	"mime"
 	"net/http"
 	"net/url"
-	"strconv"
 	"strings"
 	"sync/atomic"
-	"time"
 
 	"github.com/minio/minio/internal/auth"
-	objectlock "github.com/minio/minio/internal/bucket/object/lock"
 	"github.com/minio/minio/internal/etag"
 	"github.com/minio/minio/internal/hash"
 	xhttp "github.com/minio/minio/internal/http"
@@ -733,48 +730,6 @@ func setAuthMiddleware(h http.Handler) http.Handler {
 	})
 }
 
-func isPutRetentionAllowed(bucketName, objectName string, retDays int, retDate time.Time, retMode objectlock.RetMode, byPassSet bool, r *http.Request, cred auth.Credentials, owner bool) (s3Err APIErrorCode) {
-	var retSet bool
-	if cred.AccessKey == "" {
-		return ErrAccessDenied
-	}
-
-	conditions := getConditionValues(r, "", cred)
-	conditions["object-lock-mode"] = []string{string(retMode)}
-	conditions["object-lock-retain-until-date"] = []string{retDate.UTC().Format(time.RFC3339)}
-	if retDays > 0 {
-		conditions["object-lock-remaining-retention-days"] = []string{strconv.Itoa(retDays)}
-	}
-	if retMode == objectlock.RetGovernance && byPassSet {
-		byPassSet = globalIAMSys.IsAllowed(policy.Args{
-			AccountName:     cred.AccessKey,
-			Groups:          cred.Groups,
-			Action:          policy.BypassGovernanceRetentionAction,
-			BucketName:      bucketName,
-			ObjectName:      objectName,
-			ConditionValues: conditions,
-			IsOwner:         owner,
-			Claims:          cred.Claims,
-		})
-	}
-	if globalIAMSys.IsAllowed(policy.Args{
-		AccountName:     cred.AccessKey,
-		Groups:          cred.Groups,
-		Action:          policy.PutObjectRetentionAction,
-		BucketName:      bucketName,
-		ConditionValues: conditions,
-		ObjectName:      objectName,
-		IsOwner:         owner,
-		Claims:          cred.Claims,
-	}) {
-		retSet = true
-	}
-	if byPassSet || retSet {
-		return ErrNone
-	}
-	return ErrAccessDenied
-}
-
 // isPutActionAllowed - check if PUT operation is allowed on the resource, this
 // call verifies bucket policies and IAM policies, supports multi user
 // checks etc.

cmd/batch-handlers.go

@@ -583,7 +583,11 @@ func toObjectInfo(bucket, object string, objInfo minio.ObjectInfo) ObjectInfo {
 func (r BatchJobReplicateV1) writeAsArchive(ctx context.Context, objAPI ObjectLayer, remoteClnt *minio.Client, entries []ObjectInfo, prefix string) error {
 	input := make(chan minio.SnowballObject, 1)
 	opts := minio.SnowballOptions{
-		Opts:     minio.PutObjectOptions{},
+		Opts: minio.PutObjectOptions{
+			Internal: minio.AdvancedPutOptions{
+				ReplicationRequest: true,
+			},
+		},
 		InMemory: *r.Source.Snowball.InMemory,
 		Compress: *r.Source.Snowball.Compress,
 		SkipErrs: *r.Source.Snowball.SkipErrs,

cmd/bucket-handlers.go

@@ -1273,6 +1273,12 @@ func (api objectAPIHandlers) PostPolicyBucketHandler(w http.ResponseWriter, r *h
 		writeErrorResponseHeadersOnly(w, toAPIError(ctx, err))
 		return
 	}
+	if isDirObject(object) {
+		if _, _, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, ErrNone, ErrNone, false); s3Err != ErrNone {
+			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
+			return
+		}
+	}
 	opts.WantChecksum = checksum
 
 	fanOutOpts := fanOutOptions{Checksum: checksum}

cmd/bucket-object-lock.go

@@ -20,8 +20,8 @@ package cmd
 import (
 	"context"
 	"errors"
-	"math"
 	"net/http"
+	"strconv"
 	"strings"
 	"time"
 
@@ -164,76 +164,129 @@ func enforceRetentionBypassForDelete(ctx context.Context, r *http.Request, bucke
 	return nil
 }
 
-// enforceRetentionBypassForPut enforces whether an existing object under governance can be overwritten
-// with governance bypass headers set in the request.
-// Objects under site wide WORM cannot be overwritten.
-// For objects in "Governance" mode, overwrite is allowed if a) object retention date is past OR
-// governance bypass headers are set and user has governance bypass permissions.
-// Objects in compliance mode can be overwritten only if retention date is being extended. No mode change is permitted.
+// isRetentionActionAllowed evaluates action for a PutObjectRetention request
+// with the object-lock condition keys derived from the requested retention.
+// The keys come from the request, not from the stored version, so the
+// handler can decide s3:PutObjectRetention before reading any object state.
+func isRetentionActionAllowed(action policy.Action, bucket, object string, objRetention *objectlock.ObjectRetention, r *http.Request, cred auth.Credentials, owner bool) bool {
+	if cred.AccessKey == "" {
+		return false
+	}
+	t, err := objectlock.UTCNowNTP()
+	if err != nil {
+		return false
+	}
+	conditions := getConditionValues(r, "", cred)
+	conditions["object-lock-mode"] = []string{string(objRetention.Mode)}
+	conditions["object-lock-retain-until-date"] = []string{objRetention.RetainUntilDate.UTC().Format(time.RFC3339)}
+	// Relative days from now, for "object-lock-remaining-retention-days".
+	// Clearing retention (<Retention/>) has a zero RetainUntilDate, which must
+	// evaluate as 0 remaining days rather than a large positive math.Abs delta
+	// from year 0001.
+	days := 0
+	if !objRetention.RetainUntilDate.IsZero() && objRetention.RetainUntilDate.After(t) {
+		// Time.Sub saturates after about 292 years. Count seconds instead so
+		// maximum-retention policies also apply to distant RFC3339 dates.
+		seconds := objRetention.RetainUntilDate.Unix() - t.Unix()
+		days = int(seconds / 86400)
+		if seconds%86400 != 0 || objRetention.RetainUntilDate.Nanosecond() > t.Nanosecond() {
+			days++
+		}
+	}
+	conditions["object-lock-remaining-retention-days"] = []string{strconv.Itoa(days)}
+	return globalIAMSys.IsAllowed(policy.Args{
+		AccountName:     cred.AccessKey,
+		Groups:          cred.Groups,
+		Action:          action,
+		BucketName:      bucket,
+		ConditionValues: conditions,
+		ObjectName:      object,
+		IsOwner:         owner,
+		Claims:          cred.Claims,
+	})
+}
+
+// enforceRetentionBypassForPut enforces the stored retention of oi against a
+// PutObjectRetention request. The caller must already hold
+// s3:PutObjectRetention; this only decides changes that depend on the stored
+// state. Expired or absent retention may be changed freely. Unexpired
+// GOVERNANCE retention may be shortened, removed or have its mode changed only
+// with the bypass header and s3:BypassGovernanceRetention. Unexpired
+// COMPLIANCE retention may only be extended.
 func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, oi ObjectInfo, objRetention *objectlock.ObjectRetention, cred auth.Credentials, owner bool) error {
-	byPassSet := objectlock.IsObjectLockGovernanceBypassSet(r.Header)
+	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
+	if !ret.Mode.Valid() {
+		return nil
+	}
 
 	t, err := objectlock.UTCNowNTP()
 	if err != nil {
 		internalLogIf(ctx, err, logger.WarningKind)
 		return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
 	}
+	if ret.RetainUntilDate.Before(t) {
+		return nil
+	}
 
-	// Pass in relative days from current time, to additionally
-	// to verify "object-lock-remaining-retention-days" policy if any.
-	days := int(math.Ceil(math.Abs(objRetention.RetainUntilDate.Sub(t).Hours()) / 24))
-
-	ret := objectlock.GetObjectRetentionMeta(oi.UserDefined)
-	if ret.Mode.Valid() {
-		// Retention has expired you may change whatever you like.
-		if ret.RetainUntilDate.Before(t) {
-			apiErr := isPutRetentionAllowed(oi.Bucket, oi.Name,
-				days, objRetention.RetainUntilDate.Time,
-				objRetention.Mode, byPassSet, r, cred,
-				owner)
-			if apiErr == ErrAccessDenied {
-				return errAuthentication
-			}
-			return nil
+	weakens := objRetention.Mode != ret.Mode || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time)
+	if !weakens {
+		return nil
+	}
+	switch ret.Mode {
+	case objectlock.RetGovernance:
+		// https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lock.html#object-lock-bypass-governance
+		if !objectlock.IsObjectLockGovernanceBypassSet(r.Header) {
+			return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
 		}
-
-		switch ret.Mode {
-		case objectlock.RetGovernance:
-			govPerm := isPutRetentionAllowed(oi.Bucket, oi.Name, days,
-				objRetention.RetainUntilDate.Time, objRetention.Mode,
-				byPassSet, r, cred, owner)
-			// Governance mode retention period cannot be shortened, if x-amz-bypass-governance is not set.
-			if !byPassSet {
-				if objRetention.Mode != objectlock.RetGovernance || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time) {
-					return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
-				}
-			}
-			if govPerm == ErrAccessDenied {
-				return errAuthentication
-			}
-			return nil
-		case objectlock.RetCompliance:
-			// Compliance retention mode cannot be changed or shortened.
-			// https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html#object-lock-retention-modes
-			if objRetention.Mode != objectlock.RetCompliance || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time) {
-				return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
-			}
-			apiErr := isPutRetentionAllowed(oi.Bucket, oi.Name,
-				days, objRetention.RetainUntilDate.Time, objRetention.Mode,
-				false, r, cred, owner)
-			if apiErr == ErrAccessDenied {
-				return errAuthentication
-			}
-			return nil
+		if !isRetentionActionAllowed(policy.BypassGovernanceRetentionAction, oi.Bucket, oi.Name, objRetention, r, cred, owner) {
+			return errAuthentication
 		}
 		return nil
-	} // No pre-existing retention metadata present.
+	default:
+		// Compliance retention mode cannot be changed or shortened.
+		// https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html#object-lock-retention-modes
+		return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
+	}
+}
 
-	apiErr := isPutRetentionAllowed(oi.Bucket, oi.Name,
-		days, objRetention.RetainUntilDate.Time,
-		objRetention.Mode, byPassSet, r, cred, owner)
-	if apiErr == ErrAccessDenied {
-		return errAuthentication
+// checkObjectLockedForOverwrite re-checks under the object write lock whether an
+// in-place write to the null version (such as a directory object) would
+// overwrite a locked version in an Object Lock bucket.
+func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, getObjectInfoFn GetObjectInfoFn) error {
+	if isMinioMetaBucketName(bucket) {
+		return nil
+	}
+	retentionCfg, err := globalBucketObjectLockSys.Get(bucket)
+	if err != nil {
+		return err
+	}
+	if !retentionCfg.LockEnabled {
+		return nil
+	}
+	objInfo, err := getObjectInfoFn(ctx, bucket, object, ObjectOptions{
+		VersionID:  nullVersionID,
+		NoLock:     true,
+		NoAuditLog: true,
+	})
+	if err != nil {
+		if isErrObjectNotFound(err) || isErrVersionNotFound(err) || errors.As(err, &MethodNotAllowed{}) {
+			return nil
+		}
+		return err
+	}
+	if objectlock.GetObjectLegalHoldMeta(objInfo.UserDefined).Status == objectlock.LegalHoldOn {
+		return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: nullVersionID}
+	}
+	r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
+	if r.Mode == objectlock.RetCompliance || r.Mode == objectlock.RetGovernance {
+		t, err := objectlock.UTCNowNTP()
+		if err != nil {
+			internalLogIf(ctx, err, logger.WarningKind)
+			return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: nullVersionID}
+		}
+		if r.RetainUntilDate.After(t) {
+			return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: nullVersionID}
+		}
 	}
 	return nil
 }
@@ -274,16 +327,24 @@ func checkPutObjectLockAllowed(ctx context.Context, rq *http.Request, bucket, ob
 	if err != nil {
 		return mode, retainDate, legalHold, toAPIErrorCode(ctx, err)
 	}
+	if isDirObject(object) && opts.VersionID == "" {
+		opts.VersionID = nullVersionID
+	}
 
 	if opts.VersionID != "" && !replicaTrusted {
-		if objInfo, err := getObjectInfoFn(ctx, bucket, object, opts); err == nil {
+		objInfo, err := getObjectInfoFn(ctx, bucket, object, opts)
+		if err != nil {
+			if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) && !errors.As(err, &MethodNotAllowed{}) {
+				return mode, retainDate, legalHold, toAPIErrorCode(ctx, err)
+			}
+		} else {
 			r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
 			t, err := objectlock.UTCNowNTP()
 			if err != nil {
 				internalLogIf(ctx, err, logger.WarningKind)
 				return mode, retainDate, legalHold, ErrObjectLocked
 			}
-			if r.Mode == objectlock.RetCompliance && r.RetainUntilDate.After(t) {
+			if (r.Mode == objectlock.RetCompliance || r.Mode == objectlock.RetGovernance) && r.RetainUntilDate.After(t) {
 				return mode, retainDate, legalHold, ErrObjectLocked
 			}
 			mode = r.Mode

cmd/erasure-multipart-fullobject_test.go

@@ -253,7 +253,7 @@ func testAPICompleteMultipartFullObjectChecksumMismatch(obj ObjectLayer, instanc
 	if apiErr.Code != "BadDigest" {
 		t.Fatalf("%s: expected BadDigest, got %q", instanceType, apiErr.Code)
 	}
-	if want := "The CRC32 checksum you specified did not match the calculated checksum."; apiErr.Message != want {
+	if want := "The CRC32 you specified did not match the calculated checksum."; apiErr.Message != want {
 		t.Fatalf("%s: expected message %q, got %q", instanceType, want, apiErr.Message)
 	}
 
@@ -361,7 +361,7 @@ func testAPICompleteMultipartCompositeChecksumMismatch(obj ObjectLayer, instance
 	if apiErr.Code != "BadDigest" {
 		t.Fatalf("%s: expected BadDigest, got %q", instanceType, apiErr.Code)
 	}
-	if want := "The CRC32 checksum you specified did not match the calculated checksum."; apiErr.Message != want {
+	if want := "The CRC32 you specified did not match the calculated checksum."; apiErr.Message != want {
 		t.Fatalf("%s: expected message %q, got %q", instanceType, want, apiErr.Message)
 	}
 	if _, err := obj.GetObjectInfo(t.Context(), bucketName, objectName, ObjectOptions{}); err == nil {

cmd/erasure-multipart.go

@@ -1696,6 +1696,32 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		defer lk.Unlock(lkctx)
 	}
 
+	if !opts.ReplicationRequest && !opts.ReplicaLockReconcile && !opts.DataMovement {
+		getObjectInfo := er.getObjectInfo
+		if opts.replicaObjectInfo != nil {
+			getObjectInfo = opts.replicaObjectInfo
+		}
+		if fi.VersionID == "" || fi.VersionID == nullVersionID {
+			if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo); err != nil {
+				return oi, toObjectErr(err, bucket, object)
+			}
+		} else {
+			_, gerr := getObjectInfo(ctx, bucket, object, ObjectOptions{
+				VersionID:        fi.VersionID,
+				Versioned:        opts.Versioned,
+				VersionSuspended: opts.VersionSuspended,
+				NoLock:           true,
+			})
+			switch {
+			case gerr == nil, errors.As(gerr, &MethodNotAllowed{}):
+				return oi, toObjectErr(errAuthentication, bucket, object)
+			case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr):
+			default:
+				return oi, toObjectErr(gerr, bucket, object)
+			}
+		}
+	}
+
 	er.cleanupMultipartPath(ctx, paths...) // cleanup all part.N.meta, and skipped part.N's before final rename().
 
 	defer func() {

cmd/erasure-object.go

@@ -150,6 +150,17 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 		}
 	}
 
+	if !dstOpts.ReplicationRequest && !dstOpts.ReplicaLockReconcile && !dstOpts.DataMovement &&
+		(versionID == "" || versionID == nullVersionID) {
+		getObjectInfo := er.getObjectInfo
+		if dstOpts.replicaObjectInfo != nil {
+			getObjectInfo = dstOpts.replicaObjectInfo
+		}
+		if err := checkObjectLockedForOverwrite(ctx, dstBucket, dstObject, getObjectInfo); err != nil {
+			return oi, toObjectErr(err, dstBucket, dstObject)
+		}
+	}
+
 	modTime = UTCNow() // We only preserve modTime if dstOpts.MTime is true.
 	// in all other cases mtime is latest.
 
@@ -1594,6 +1605,17 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 		defer lk.Unlock(lkctx)
 	}
 
+	if !opts.ReplicationRequest && !opts.ReplicaLockReconcile && !opts.DataMovement &&
+		(fi.VersionID == "" || fi.VersionID == nullVersionID) {
+		getObjectInfo := er.getObjectInfo
+		if opts.replicaObjectInfo != nil {
+			getObjectInfo = opts.replicaObjectInfo
+		}
+		if err := checkObjectLockedForOverwrite(ctx, bucket, object, getObjectInfo); err != nil {
+			return ObjectInfo{}, toObjectErr(err, bucket, object)
+		}
+	}
+
 	// Rename the successfully written temporary object to final location.
 	onlineDisks, versions, oldDataDir, err := renameData(ctx, onlineDisks, minioMetaTmpBucket, tempObj, partsMetadata, bucket, object, writeQuorum)
 	if err != nil {

cmd/erasure-server-pool-multipart-http_test.go

@@ -265,6 +265,7 @@ func TestPoolsMultipartConditionalLatestVersionAndCallbackOnce(t *testing.T) {
 			if explicit {
 				current = putConsistencyObject(t, z, bucket, object, 1, "latest-other-version", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Minute)})
 			}
+			opts.VersionID = ""
 			id, parts := multipartConditionUpload(t, z, bucket, object, 1, opts)
 			called := 0
 			opts.CheckPrecondFn = func(oi ObjectInfo) bool { called++; return oi.ETag != current.ETag }

cmd/erasure-server-pool-put-conditional_test.go

@@ -633,6 +633,8 @@ func TestPoolsConditionalPutReplicaAvailability(t *testing.T) {
 	}
 }
 
+// A public client cannot address a destination version, with or without a
+// precondition; only trusted replication may (SN-2026-016).
 func TestPoolsConditionalPutDestinationVersionHTTP(t *testing.T) {
 	z, _ := consistencyPools(t)
 	bucket, router := conditionalPutBucket(t, z, "versioned")
@@ -641,21 +643,14 @@ func TestPoolsConditionalPutDestinationVersionHTTP(t *testing.T) {
 	current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{Versioned: true, MTime: UTCNow().Add(-time.Minute)})
 	defer conditionalPutPool(t, z, object, 0)()
 	url := getPutObjectURL("", bucket, object) + "?versionId=" + old.VersionID
-	for _, stale := range []bool{true, false} {
-		tag, status := current.ETag, http.StatusOK
-		if stale {
-			tag, status = old.ETag, http.StatusPreconditionFailed
-		}
+	for _, tag := range []string{old.ETag, current.ETag} {
 		put := multipartConditionRequest(t, router, http.MethodPut, url, "replacement", map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", tag)})
-		if put.Code != status {
-			t.Fatalf("version-addressed public PUT: %d want %d: %s", put.Code, status, put.Body.String())
-		}
-		if got := put.Header()[xhttp.AmzVersionID]; !stale && (len(got) != 1 || got[0] != old.VersionID) {
-			t.Fatalf("write version changed: %v", put.Header())
+		if put.Code != http.StatusForbidden {
+			t.Fatalf("version-addressed public PUT: %d want 403: %s", put.Code, put.Body.String())
 		}
 	}
 	get := multipartConditionRequest(t, router, http.MethodGet, url, "", nil)
-	if get.Code != http.StatusOK || get.Body.String() != "replacement" {
+	if get.Code != http.StatusOK || get.Body.String() != "old" {
 		t.Fatalf("addressed GET: %d %q", get.Code, get.Body.String())
 	}
 }

cmd/erasure-server-pool.go

@@ -1199,6 +1199,10 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 
 	if opts.ReplicaLockReconcile || opts.DataMovement {
 		opts.ReplicaLockReconcile = true
+	}
+	// Public overwrites need all pools for Object Lock. Unaddressed replica
+	// callbacks retain their set-local reads unless reconciliation is needed.
+	if opts.ReplicaLockReconcile || !opts.ReplicationRequest {
 		opts.replicaObjectInfo = z.replicaObjectInfo
 	}
 	oi, err := z.serverPools[idx].PutObject(ctx, bucket, object, data, opts)
@@ -1510,9 +1514,9 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		return objInfo, err
 	}
 
-	if !z.SinglePool() && dstOpts.ReplicaLockReconcile {
+	if !z.SinglePool() && (dstOpts.ReplicaLockReconcile || !dstOpts.ReplicationRequest) {
 		dstOpts.replicaObjectInfo = z.replicaObjectInfo
-		if !dstOpts.DataMovement {
+		if dstOpts.ReplicaLockReconcile && !dstOpts.DataMovement {
 			defer func() {
 				if err == nil {
 					err = z.retireReplicaCopies(ctx, dstBucket, dstObject, poolIdx, objInfo)
@@ -1549,13 +1553,16 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		ServerSideEncryption:       dstOpts.ServerSideEncryption,
 		UserDefined:                srcInfo.UserDefined,
 		Versioned:                  dstOpts.Versioned,
+		VersionSuspended:           dstOpts.VersionSuspended,
 		VersionID:                  dstOpts.VersionID,
 		MTime:                      dstOpts.MTime,
 		NoLock:                     true,
 		EncryptFn:                  dstOpts.EncryptFn,
 		WantChecksum:               dstOpts.WantChecksum,
 		WantServerSideChecksumType: dstOpts.WantServerSideChecksumType,
+		ReplicationRequest:         dstOpts.ReplicationRequest,
 		ReplicaLockReconcile:       dstOpts.ReplicaLockReconcile,
+		DataMovement:               dstOpts.DataMovement,
 		replicaObjectInfo:          dstOpts.replicaObjectInfo,
 	}
 
@@ -2305,6 +2312,8 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 
 	if opts.ReplicaLockReconcile || opts.DataMovement {
 		opts.ReplicaLockReconcile = true
+	}
+	if opts.ReplicaLockReconcile || !opts.ReplicationRequest {
 		opts.replicaObjectInfo = z.replicaObjectInfo
 	}
 

cmd/erasure-sets.go

@@ -869,11 +869,17 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
 		ServerSideEncryption:       dstOpts.ServerSideEncryption,
 		UserDefined:                srcInfo.UserDefined,
 		Versioned:                  dstOpts.Versioned,
+		VersionSuspended:           dstOpts.VersionSuspended,
 		VersionID:                  dstOpts.VersionID,
 		MTime:                      dstOpts.MTime,
+		NoLock:                     dstOpts.NoLock,
 		EncryptFn:                  dstOpts.EncryptFn,
 		WantChecksum:               dstOpts.WantChecksum,
 		WantServerSideChecksumType: dstOpts.WantServerSideChecksumType,
+		ReplicationRequest:         dstOpts.ReplicationRequest,
+		ReplicaLockReconcile:       dstOpts.ReplicaLockReconcile,
+		DataMovement:               dstOpts.DataMovement,
+		replicaObjectInfo:          dstOpts.replicaObjectInfo,
 	}
 
 	return dstSet.putObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, putOpts)

cmd/object-api-errors.go

@@ -668,7 +668,7 @@ var (
 func completeMultipartChecksumMismatch(algorithm string) error {
 	description := "The checksum you specified did not match the calculated checksum."
 	if algorithm != "" {
-		description = fmt.Sprintf("The %s checksum you specified did not match the calculated checksum.", algorithm)
+		description = fmt.Sprintf("The %s you specified did not match the calculated checksum.", algorithm)
 	}
 	return fmt.Errorf("%w: %s", errCompleteMultipartChecksumMismatch, description)
 }

cmd/object-api-options.go

@@ -347,6 +347,11 @@ func putOpts(ctx context.Context, bucket, object, vid string, hdrs http.Header,
 	versionSuspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
 
 	vid = strings.TrimSpace(vid)
+	// A caller-chosen version ID writes that version in place. Only trusted
+	// replication may recreate a source version under its original ID.
+	if vid != "" && !trustedReplication {
+		return opts, errAuthentication
+	}
 	if vid != "" && vid != nullVersionID {
 		_, err := uuid.Parse(vid)
 		if err != nil {

cmd/object-handlers.go

@@ -3435,27 +3435,16 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 		return
 	}
 
-	// Check permissions to perform this object retention operation
 	if s3Error := authenticateRequest(ctx, r, policy.PutObjectRetentionAction); s3Error != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
 		return
 	}
 
-	if _, err := objectAPI.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
-		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
-		return
-	}
-
 	if !validateLengthAndChecksum(r) {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentMD5), r.URL)
 		return
 	}
 
-	if rcfg, _ := globalBucketObjectLockSys.Get(bucket); !rcfg.LockEnabled {
-		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidBucketObjectLockConfiguration), r.URL)
-		return
-	}
-
 	objRetention, err := objectlock.ParseObjectRetention(r.Body)
 	if err != nil {
 		apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
@@ -3467,6 +3456,24 @@ func (api objectAPIHandlers) PutObjectRetentionHandler(w http.ResponseWriter, r
 	reqInfo := logger.GetReqInfo(ctx)
 	reqInfo.SetTags("retention", objRetention.String())
 
+	// Check permissions to perform this object retention operation. The
+	// object-lock condition keys come from the request body, so this is
+	// decided before any bucket or object state is read.
+	if !isRetentionActionAllowed(policy.PutObjectRetentionAction, bucket, object, objRetention, r, reqInfo.Cred, reqInfo.Owner) {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
+		return
+	}
+
+	if _, err := objectAPI.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
+		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+		return
+	}
+
+	if rcfg, _ := globalBucketObjectLockSys.Get(bucket); !rcfg.LockEnabled {
+		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidBucketObjectLockConfiguration), r.URL)
+		return
+	}
+
 	opts, err := getOpts(ctx, r, bucket, object)
 	if err != nil {
 		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)

cmd/object-multipart-handlers.go

@@ -1166,7 +1166,7 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
 		return
 	}
 
-	if _, _, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, ErrNone, ErrNone, false); s3Err != ErrNone {
+	if _, _, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, ErrNone, ErrNone, trustedReplication); s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
 		return
 	}

cmd/object-retention-authz_test.go

@@ -0,0 +1,679 @@
+// 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 (
+	"archive/tar"
+	"bytes"
+	"context"
+	"crypto/md5"
+	"encoding/base64"
+	"encoding/xml"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/amztime"
+	"github.com/minio/minio/internal/auth"
+	objectlock "github.com/minio/minio/internal/bucket/object/lock"
+	xhttp "github.com/minio/minio/internal/http"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+func retentionAuthzRequest(t *testing.T, router http.Handler, creds auth.Credentials, method, target string, body []byte, headers map[string]string) *httptest.ResponseRecorder {
+	t.Helper()
+	var rd io.ReadSeeker
+	if body != nil {
+		rd = bytes.NewReader(body)
+	}
+	req, err := newTestSignedRequestV4(method, target, int64(len(body)), rd, creds.AccessKey, creds.SecretKey, headers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, req)
+	return rec
+}
+
+func retentionBody(mode string, until time.Time) []byte {
+	if mode == "" {
+		return []byte(`<Retention xmlns="http://s3.amazonaws.com/doc/2006-03-01/"></Retention>`)
+	}
+	return fmt.Appendf(nil, `<Retention xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Mode>%s</Mode><RetainUntilDate>%s</RetainUntilDate></Retention>`,
+		mode, until.UTC().Format(time.RFC3339))
+}
+
+func putRetention(t *testing.T, router http.Handler, cred auth.Credentials, bucket, key, vid, mode string, until time.Time, bypass bool) int {
+	t.Helper()
+	body := retentionBody(mode, until)
+	sum := md5.Sum(body)
+	headers := map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])}
+	if bypass {
+		headers[objectlock.AmzObjectLockBypassRetGovernance] = "true"
+	}
+	rec := retentionAuthzRequest(t, router, cred, http.MethodPut,
+		makeTestTargetURL("", bucket, key, url.Values{"retention": {""}, "versionId": {vid}}), body, headers)
+	return rec.Code
+}
+
+func newRetentionPolicyUser(t *testing.T, instanceType, policyJSON string) auth.Credentials {
+	t.Helper()
+	ctx := t.Context()
+	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(ctx, creds.AccessKey, madmin.AddOrUpdateUserReq{
+		SecretKey: creds.SecretKey,
+		Status:    madmin.AccountEnabled,
+	}); err != nil {
+		t.Fatalf("%s: create user: %v", instanceType, err)
+	}
+	parsed, err := policy.ParseConfig(strings.NewReader(policyJSON))
+	if err != nil {
+		t.Fatalf("%s: parse policy: %v", instanceType, err)
+	}
+	policyName := "retention-authz-" + mustGetUUID()
+	if _, err = globalIAMSys.SetPolicy(ctx, policyName, *parsed); err != nil {
+		t.Fatalf("%s: install policy: %v", instanceType, err)
+	}
+	if _, err = globalIAMSys.PolicyDBSet(ctx, creds.AccessKey, policyName, regUser, false); err != nil {
+		t.Fatalf("%s: attach policy: %v", instanceType, err)
+	}
+	return creds
+}
+
+// TestAPIPutObjectRetentionAuthorization covers SN-2026-015: the governance
+// bypass header is intent only; s3:PutObjectRetention is always required and
+// weakening unexpired GOVERNANCE also requires s3:BypassGovernanceRetention.
+func TestAPIPutObjectRetentionAuthorization(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:          t,
+		objAPITest: testAPIPutObjectRetentionAuthorization,
+	})
+}
+
+func testAPIPutObjectRetentionAuthorization(obj ObjectLayer, instanceType, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+	enableBucketObjectLock(t, bucket)
+
+	noRet := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:GetObject"`)
+	retOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:PutObjectRetention"`)
+	retBypass := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:PutObjectRetention","s3:BypassGovernanceRetention"`)
+	bypassOnly := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:BypassGovernanceRetention"`)
+	condDenyShort := newRetentionPolicyUser(t, instanceType, `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Action": ["s3:PutObjectRetention", "s3:BypassGovernanceRetention"],
+				"Resource": ["arn:aws:s3:::`+bucket+`/*"]
+			},
+			{
+				"Effect": "Deny",
+				"Action": ["s3:PutObjectRetention"],
+				"Resource": ["arn:aws:s3:::`+bucket+`/*"],
+				"Condition": {"NumericLessThan": {"s3:object-lock-remaining-retention-days": "60"}}
+			}
+		]
+	}`)
+	explicitDenyBypass := newRetentionPolicyUser(t, instanceType, `{
+		"Version": "2012-10-17",
+		"Statement": [
+			{
+				"Effect": "Allow",
+				"Action": ["s3:*"],
+				"Resource": ["arn:aws:s3:::`+bucket+`/*"]
+			},
+			{
+				"Effect": "Deny",
+				"Action": ["s3:BypassGovernanceRetention"],
+				"Resource": ["arn:aws:s3:::`+bucket+`/*"]
+			}
+		]
+	}`)
+
+	far := UTCNow().Add(365 * 24 * time.Hour).Truncate(time.Second)
+	nearer := UTCNow().Add(30 * 24 * time.Hour).Truncate(time.Second)
+	farther := far.Add(24 * time.Hour)
+
+	put := func(t *testing.T, object, mode string) string {
+		t.Helper()
+		payload := []byte("retention authorization")
+		opts := ObjectOptions{Versioned: true, UserDefined: map[string]string{}}
+		if mode != "" {
+			opts.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = mode
+			opts.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(far)
+		}
+		info, err := obj.PutObject(t.Context(), bucket, object,
+			mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), opts)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return info.VersionID
+	}
+	stored := func(t *testing.T, object, vid string) objectlock.ObjectRetention {
+		t.Helper()
+		info, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: vid})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return objectlock.GetObjectRetentionMeta(info.UserDefined)
+	}
+
+	for _, test := range []struct {
+		name     string
+		creds    auth.Credentials
+		existing string // stored retention mode on the version, until far
+		mode     string // requested mode, "" removes retention
+		until    time.Time
+		bypass   bool
+		want     int
+		changed  bool
+	}{
+		{name: "no-permission removal", creds: noRet, existing: "GOVERNANCE", bypass: true, want: http.StatusForbidden},
+		{name: "no-permission mode change", creds: noRet, existing: "GOVERNANCE", mode: "COMPLIANCE", until: nearer, bypass: true, want: http.StatusForbidden},
+		{name: "no-permission new compliance", creds: noRet, mode: "COMPLIANCE", until: nearer, bypass: true, want: http.StatusForbidden},
+		{name: "retention-only shorten with header", creds: retOnly, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, bypass: true, want: http.StatusForbidden},
+		{name: "retention-only removal with header", creds: retOnly, existing: "GOVERNANCE", bypass: true, want: http.StatusForbidden},
+		{name: "retention-only mode change with header", creds: retOnly, existing: "GOVERNANCE", mode: "COMPLIANCE", until: nearer, bypass: true, want: http.StatusForbidden},
+		{name: "retention-only shorten without header", creds: retOnly, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, want: http.StatusBadRequest},
+		{name: "retention-only extend with header", creds: retOnly, existing: "GOVERNANCE", mode: "GOVERNANCE", until: farther, bypass: true, want: http.StatusOK, changed: true},
+		{name: "retention-only new retention", creds: retOnly, mode: "GOVERNANCE", until: nearer, want: http.StatusOK, changed: true},
+		{name: "bypass shorten with header", creds: retBypass, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, bypass: true, want: http.StatusOK, changed: true},
+		{name: "bypass removal with header", creds: retBypass, existing: "GOVERNANCE", bypass: true, want: http.StatusOK, changed: true},
+		{name: "bypass shorten without header", creds: retBypass, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, want: http.StatusBadRequest},
+		{name: "bypass shorten compliance", creds: retBypass, existing: "COMPLIANCE", mode: "COMPLIANCE", until: nearer, bypass: true, want: http.StatusBadRequest},
+		{name: "bypass-only extend", creds: bypassOnly, existing: "GOVERNANCE", mode: "GOVERNANCE", until: farther, bypass: true, want: http.StatusForbidden},
+		{name: "conditional-deny-put-retention with bypass header", creds: condDenyShort, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, bypass: true, want: http.StatusForbidden},
+		{name: "conditional-allow-put-retention satisfied", creds: condDenyShort, existing: "GOVERNANCE", mode: "GOVERNANCE", until: farther, bypass: true, want: http.StatusOK, changed: true},
+		{name: "explicit-deny-bypass shorten with header", creds: explicitDenyBypass, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, bypass: true, want: http.StatusForbidden},
+		{name: "owner shorten with header", creds: root, existing: "GOVERNANCE", mode: "GOVERNANCE", until: nearer, bypass: true, want: http.StatusOK, changed: true},
+	} {
+		t.Run(instanceType+"/"+test.name, func(t *testing.T) {
+			object := "retention-authz/" + strings.ReplaceAll(test.name, " ", "-")
+			vid := put(t, object, test.existing)
+			before := stored(t, object, vid)
+
+			body := retentionBody(test.mode, test.until)
+			sum := md5.Sum(body)
+			headers := map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])}
+			if test.bypass {
+				headers[objectlock.AmzObjectLockBypassRetGovernance] = "true"
+			}
+			target := makeTestTargetURL("", bucket, object, url.Values{"retention": {""}, "versionId": {vid}})
+			rec := retentionAuthzRequest(t, router, test.creds, http.MethodPut, target, body, headers)
+			if rec.Code != test.want {
+				t.Fatalf("status = %d, want %d: %s", rec.Code, test.want, rec.Body.String())
+			}
+			after := stored(t, object, vid)
+			if changed := after.Mode != before.Mode || !after.RetainUntilDate.Equal(before.RetainUntilDate.Time); changed != test.changed {
+				t.Fatalf("retention changed = %v, want %v (before %s, after %s)", changed, test.changed, before.String(), after.String())
+			}
+		})
+	}
+
+	t.Run(instanceType+"/no-permission missing bucket", func(t *testing.T) {
+		body := retentionBody("", time.Time{})
+		sum := md5.Sum(body)
+		target := makeTestTargetURL("", "retention-authz-missing", "object", url.Values{"retention": {""}})
+		rec := retentionAuthzRequest(t, router, noRet, http.MethodPut, target, body,
+			map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])})
+		if rec.Code != http.StatusForbidden {
+			t.Fatalf("status = %d, want 403 before bucket lookup: %s", rec.Code, rec.Body.String())
+		}
+	})
+}
+
+// TestAPIVersionAddressedWriteRequiresReplicationTrust covers SN-2026-016: a
+// caller-chosen destination versionId is honored only for trusted replication,
+// and write paths cannot overwrite locked versions via directory keys or
+// version-addressed multipart completions.
+func TestAPIVersionAddressedWriteRequiresReplicationTrust(t *testing.T) {
+	defer DetectTestLeak(t)()
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		objAPITest:        testAPIVersionAddressedWriteRequiresReplicationTrust,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+	})
+}
+
+func testAPIVersionAddressedWriteRequiresReplicationTrust(obj ObjectLayer, instanceType, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+	writer := newObjectAttributesAuthzUser(t, instanceType, bucket, `"s3:GetObject","s3:PutObject"`)
+	original := []byte("original version content")
+	info, err := obj.PutObject(t.Context(), bucket, "addressed",
+		mustGetPutObjReader(t, bytes.NewReader(original), int64(len(original)), "", ""), ObjectOptions{Versioned: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	vid := info.VersionID
+
+	unchanged := func(t *testing.T) {
+		t.Helper()
+		got, err := obj.GetObjectInfo(t.Context(), bucket, "addressed", ObjectOptions{VersionID: vid})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got.Size != int64(len(original)) || got.ETag != info.ETag {
+			t.Fatalf("version %s was rewritten: size %d etag %s", vid, got.Size, got.ETag)
+		}
+	}
+
+	for _, creds := range []struct {
+		name  string
+		creds auth.Credentials
+	}{{"writer", writer}, {"owner", root}} {
+		t.Run(instanceType+"/"+creds.name+"/put", func(t *testing.T) {
+			body := []byte("replacement")
+			target := makeTestTargetURL("", bucket, "addressed", url.Values{"versionId": {vid}})
+			rec := retentionAuthzRequest(t, router, creds.creds, http.MethodPut, target, body, nil)
+			if rec.Code != http.StatusForbidden {
+				t.Fatalf("status = %d, want 403: %s", rec.Code, rec.Body.String())
+			}
+			unchanged(t)
+		})
+		t.Run(instanceType+"/"+creds.name+"/null-version", func(t *testing.T) {
+			body := []byte("replacement")
+			target := makeTestTargetURL("", bucket, "addressed", url.Values{"versionId": {nullVersionID}})
+			rec := retentionAuthzRequest(t, router, creds.creds, http.MethodPut, target, body, nil)
+			if rec.Code != http.StatusForbidden {
+				t.Fatalf("status = %d, want 403: %s", rec.Code, rec.Body.String())
+			}
+		})
+		t.Run(instanceType+"/"+creds.name+"/copy", func(t *testing.T) {
+			target := makeTestTargetURL("", bucket, "addressed", url.Values{"versionId": {vid}})
+			req, err := newTestSignedRequestV4(http.MethodPut, target, 0, nil, creds.creds.AccessKey, creds.creds.SecretKey,
+				map[string]string{xhttp.AmzCopySource: "/" + bucket + "/addressed?versionId=" + vid, xhttp.AmzMetadataDirective: "REPLACE"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec := httptest.NewRecorder()
+			router.ServeHTTP(rec, req)
+			if rec.Code != http.StatusForbidden {
+				t.Fatalf("status = %d, want 403: %s", rec.Code, rec.Body.String())
+			}
+			unchanged(t)
+		})
+		t.Run(instanceType+"/"+creds.name+"/multipart", func(t *testing.T) {
+			target := makeTestTargetURL("", bucket, "addressed", url.Values{"uploads": {""}, "versionId": {vid}})
+			rec := retentionAuthzRequest(t, router, creds.creds, http.MethodPost, target, nil, nil)
+			if rec.Code != http.StatusForbidden {
+				t.Fatalf("status = %d, want 403: %s", rec.Code, rec.Body.String())
+			}
+		})
+		t.Run(instanceType+"/"+creds.name+"/complete-multipart-existing-version", func(t *testing.T) {
+			// Simulate a pending upload persisted with an existing version ID
+			// (e.g. initiated pre-upgrade or by a replication peer).
+			up, err := obj.NewMultipartUpload(t.Context(), bucket, "addressed", ObjectOptions{
+				Versioned: true,
+				VersionID: vid,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer obj.AbortMultipartUpload(t.Context(), bucket, "addressed", up.UploadID, ObjectOptions{})
+
+			partPayload := []byte("replacement-part")
+			pi, err := obj.PutObjectPart(t.Context(), bucket, "addressed", up.UploadID, 1,
+				mustGetPutObjReader(t, bytes.NewReader(partPayload), int64(len(partPayload)), "", ""), ObjectOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			completeBody := fmt.Appendf(nil, `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>"%s"</ETag></Part></CompleteMultipartUpload>`, pi.ETag)
+			target := makeTestTargetURL("", bucket, "addressed", url.Values{"uploadId": {up.UploadID}})
+			rec := retentionAuthzRequest(t, router, creds.creds, http.MethodPost, target, completeBody, nil)
+			if rec.Code != http.StatusForbidden {
+				t.Fatalf("status = %d, want 403: %s", rec.Code, rec.Body.String())
+			}
+			unchanged(t)
+		})
+	}
+
+	t.Run(instanceType+"/locked-directory-object", func(t *testing.T) {
+		far := UTCNow().Add(365 * 24 * time.Hour).Truncate(time.Second)
+		for _, lockCase := range []struct {
+			name string
+			meta map[string]string
+		}{
+			{
+				name: "governance",
+				meta: map[string]string{
+					strings.ToLower(xhttp.AmzObjectLockMode):            "GOVERNANCE",
+					strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(far),
+				},
+			},
+			{
+				name: "compliance",
+				meta: map[string]string{
+					strings.ToLower(xhttp.AmzObjectLockMode):            "COMPLIANCE",
+					strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(far),
+				},
+			},
+			{
+				name: "legal-hold",
+				meta: map[string]string{
+					strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+				},
+			},
+		} {
+			t.Run(lockCase.name, func(t *testing.T) {
+				dirKey := "locked-dir-" + lockCase.name + "/"
+				if _, err := obj.PutObject(t.Context(), bucket, dirKey,
+					mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""),
+					ObjectOptions{Versioned: true, VersionID: nullVersionID, UserDefined: lockCase.meta}); err != nil {
+					t.Fatal(err)
+				}
+				target := makeTestTargetURL("", bucket, dirKey, nil)
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPut, target, nil, nil)
+				if rec.Code != http.StatusBadRequest {
+					t.Fatalf("PUT locked dir status = %d, want 400: %s", rec.Code, rec.Body.String())
+				}
+				got, err := obj.GetObjectInfo(t.Context(), bucket, dirKey, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				for k, wantVal := range lockCase.meta {
+					if got.UserDefined[k] != wantVal {
+						t.Fatalf("metadata %s = %q, want %q", k, got.UserDefined[k], wantVal)
+					}
+				}
+			})
+		}
+	})
+
+	t.Run(instanceType+"/snowball-archive-version-id", func(t *testing.T) {
+		var tarBuf bytes.Buffer
+		tw := tar.NewWriter(&tarBuf)
+		entryData := []byte("snowball-overwrite")
+		if err := tw.WriteHeader(&tar.Header{
+			Typeflag:   tar.TypeReg,
+			Name:       "addressed",
+			Size:       int64(len(entryData)),
+			Format:     tar.FormatPAX,
+			PAXRecords: map[string]string{"minio.versionId": vid},
+		}); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := tw.Write(entryData); err != nil {
+			t.Fatal(err)
+		}
+		if err := tw.Close(); err != nil {
+			t.Fatal(err)
+		}
+
+		// Untrusted writer cannot overwrite an existing version via PAX minio.versionId.
+		target := makeTestTargetURL("", bucket, "archive.tar", nil)
+		rec := retentionAuthzRequest(t, router, writer, http.MethodPut, target, tarBuf.Bytes(),
+			map[string]string{"x-amz-meta-snowball-auto-extract": "true"})
+		if rec.Code != http.StatusForbidden {
+			t.Fatalf("untrusted snowball status = %d, want 403: %s", rec.Code, rec.Body.String())
+		}
+		unchanged(t)
+
+		// Trusted batch replication (with outer MinIOSourceReplicationRequest) preserves PAX minio.versionId.
+		replicaVID := mustGetUUID()
+		var repBuf bytes.Buffer
+		repTW := tar.NewWriter(&repBuf)
+		repData := []byte("replicated-via-snowball")
+		if err := repTW.WriteHeader(&tar.Header{
+			Typeflag:   tar.TypeReg,
+			Name:       "snowball-replicated",
+			Size:       int64(len(repData)),
+			Format:     tar.FormatPAX,
+			PAXRecords: map[string]string{"minio.versionId": replicaVID},
+		}); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := repTW.Write(repData); err != nil {
+			t.Fatal(err)
+		}
+		if err := repTW.Close(); err != nil {
+			t.Fatal(err)
+		}
+		repRec := retentionAuthzRequest(t, router, root, http.MethodPut, target, repBuf.Bytes(),
+			map[string]string{
+				"x-amz-meta-snowball-auto-extract":  "true",
+				xhttp.MinIOSourceReplicationRequest: "true",
+			})
+		if repRec.Code != http.StatusOK {
+			t.Fatalf("trusted snowball status = %d, want 200: %s", repRec.Code, repRec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, "snowball-replicated", ObjectOptions{VersionID: replicaVID}); err != nil {
+			t.Fatalf("trusted snowball replication did not preserve version ID: %v", err)
+		}
+	})
+
+	t.Run(instanceType+"/trusted replication", func(t *testing.T) {
+		replicaVID := mustGetUUID()
+		body := []byte("replicated version")
+		target := makeTestTargetURL("", bucket, "replicated", url.Values{"versionId": {replicaVID}})
+		rec := retentionAuthzRequest(t, router, root, http.MethodPut, target, body,
+			map[string]string{xhttp.MinIOSourceReplicationRequest: "true"})
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body.String())
+		}
+		if _, err := obj.GetObjectInfo(t.Context(), bucket, "replicated", ObjectOptions{VersionID: replicaVID}); err != nil {
+			t.Fatalf("trusted replication did not keep its version ID: %v", err)
+		}
+	})
+}
+
+func TestPutObjectRetentionClearMinimumDaysPolicy(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			for _, conditionalAllow := range []bool{false, true} {
+				t.Run(fmt.Sprintf("%s/conditional-allow-%v", instance, conditionalAllow), func(t *testing.T) {
+					statement := `{"Effect":"Allow","Action":["s3:PutObjectRetention","s3:BypassGovernanceRetention"],"Resource":["arn:aws:s3:::` + bucket + `/*"]}, {"Effect":"Deny","Action":["s3:PutObjectRetention"],"Resource":["arn:aws:s3:::` + bucket + `/*"],"Condition":{"NumericLessThan":{"s3:object-lock-remaining-retention-days":"60"}}}`
+					if conditionalAllow {
+						statement = `{"Effect":"Allow","Action":["s3:BypassGovernanceRetention"],"Resource":["arn:aws:s3:::` + bucket + `/*"]}, {"Effect":"Allow","Action":["s3:PutObjectRetention"],"Resource":["arn:aws:s3:::` + bucket + `/*"],"Condition":{"NumericGreaterThanEquals":{"s3:object-lock-remaining-retention-days":"60"}}}`
+					}
+					cred := newRetentionPolicyUser(t, instance, `{"Version":"2012-10-17","Statement":[`+statement+`]}`)
+					key := fmt.Sprintf("review-minimum-%v", conditionalAllow)
+					far := UTCNow().Add(365 * 24 * time.Hour).Truncate(time.Second)
+					info, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{
+						Versioned: true,
+						UserDefined: map[string]string{
+							strings.ToLower(xhttp.AmzObjectLockMode):            "GOVERNANCE",
+							strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(far),
+						},
+					})
+					if err != nil {
+						t.Fatal(err)
+					}
+					if status := putRetention(t, router, cred, bucket, key, info.VersionID, "GOVERNANCE", UTCNow().Add(30*24*time.Hour), true); status != http.StatusForbidden {
+						t.Fatalf("shortening to 30 days status = %d, want 403", status)
+					}
+					status := putRetention(t, router, cred, bucket, key, info.VersionID, "", time.Time{}, true)
+					after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: info.VersionID})
+					if err != nil {
+						t.Fatal(err)
+					}
+					ret := objectlock.GetObjectRetentionMeta(after.UserDefined)
+					if status != http.StatusForbidden || ret.Mode != objectlock.RetGovernance {
+						t.Fatalf("clearing retention status = %d, stored retention = %s; want 403 and original GOVERNANCE", status, ret.String())
+					}
+				})
+			}
+		},
+	})
+}
+
+type hookBeforeWrite struct {
+	ObjectLayer
+	key    string
+	before func()
+}
+
+func (p *hookBeforeWrite) fire(key string) {
+	if key == p.key && p.before != nil {
+		f := p.before
+		p.before = nil
+		f()
+	}
+}
+
+func (p *hookBeforeWrite) PutObject(ctx context.Context, bucket, key string, r *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+	p.fire(key)
+	return p.ObjectLayer.PutObject(ctx, bucket, key, r, opts)
+}
+
+func (p *hookBeforeWrite) CopyObject(ctx context.Context, srcBucket, srcObject, dstBucket, dstObject string, srcInfo ObjectInfo, srcOpts, dstOpts ObjectOptions) (ObjectInfo, error) {
+	p.fire(dstObject)
+	return p.ObjectLayer.CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
+}
+
+func (p *hookBeforeWrite) CompleteMultipartUpload(ctx context.Context, bucket, key, uploadID string, parts []CompletePart, opts ObjectOptions) (ObjectInfo, error) {
+	p.fire(key)
+	return p.ObjectLayer.CompleteMultipartUpload(ctx, bucket, key, uploadID, parts, opts)
+}
+
+func TestDirectoryObjectLockAfterPreflight(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:GetObject","s3:PutObject"`)
+			srcDir := "other-dir/"
+			if rec := retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, srcDir, nil), nil, nil); rec.Code != http.StatusOK {
+				t.Fatalf("srcDir PUT: %d %s", rec.Code, rec.Body.String())
+			}
+
+			for _, mode := range []string{"GOVERNANCE", "COMPLIANCE", "LEGAL-HOLD"} {
+				for _, op := range []string{"PutObject", "CopyObject", "SelfCopy"} {
+					t.Run(instance+"/"+mode+"/"+op, func(t *testing.T) {
+						key := fmt.Sprintf("review-race-%s-%s/", mode, op)
+						if rec := retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), nil, nil); rec.Code != http.StatusOK {
+							t.Fatalf("initial PUT: %d %s", rec.Code, rec.Body.String())
+						}
+
+						var locked ObjectInfo
+						hook := &hookBeforeWrite{ObjectLayer: obj, key: key}
+						hook.before = func() {
+							if mode == "LEGAL-HOLD" {
+								body := []byte(`<LegalHold><Status>ON</Status></LegalHold>`)
+								sum := md5.Sum(body)
+								rec := retentionAuthzRequest(t, router, root, http.MethodPut, makeTestTargetURL("", bucket, key, url.Values{"legal-hold": {""}, "versionId": {nullVersionID}}), body, map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])})
+								if rec.Code != http.StatusOK {
+									t.Fatalf("set legal hold: %d %s", rec.Code, rec.Body.String())
+								}
+							} else if status := putRetention(t, router, root, bucket, key, nullVersionID, mode, UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
+								t.Fatalf("set retention: %d", status)
+							}
+							var err error
+							locked, err = obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+							if err != nil {
+								t.Fatal(err)
+							}
+						}
+						setObjectLayer(hook)
+						defer setObjectLayer(obj)
+
+						var rec *httptest.ResponseRecorder
+						switch op {
+						case "PutObject":
+							rec = retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), nil, nil)
+						case "CopyObject":
+							rec = retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), nil, map[string]string{
+								xhttp.AmzCopySource: pathJoin(bucket, srcDir),
+							})
+						case "SelfCopy":
+							rec = retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), nil, map[string]string{
+								xhttp.AmzCopySource: pathJoin(bucket, key), xhttp.AmzMetadataDirective: "REPLACE", "x-amz-meta-updated": "true",
+							})
+						}
+						if hook.before != nil {
+							t.Fatalf("%s never reached backend: %d %s", op, rec.Code, rec.Body.String())
+						}
+						after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+						if err != nil {
+							t.Fatal(err)
+						}
+						if rec.Code != http.StatusBadRequest {
+							t.Fatalf("locked null write via %s: %d %s", op, rec.Code, rec.Body.String())
+						}
+						multipartConditionError(t, rec, "InvalidRequest")
+						retentionOverwriteUnchanged(t, locked, after)
+					})
+				}
+			}
+		},
+	})
+}
+
+func TestMultipartVersioningTransitions(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t: t,
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			t.Run(instance+"/unversioned-to-enabled", func(t *testing.T) {
+				key := "review-upload-before-versioning"
+				initRec := retentionAuthzRequest(t, router, root, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploads": {""}}), nil, nil)
+				if initRec.Code != http.StatusOK {
+					t.Fatalf("init: %d %s", initRec.Code, initRec.Body.String())
+				}
+				var mp InitiateMultipartUploadResponse
+				if err := xml.Unmarshal(initRec.Body.Bytes(), &mp); err != nil {
+					t.Fatal(err)
+				}
+				part := retentionAuthzRequest(t, router, root, http.MethodPut, getPutObjectPartURL("", bucket, key, mp.UploadID, "1"), []byte("ordinary pending upload"), nil)
+				if part.Code != http.StatusOK {
+					t.Fatalf("part: %d %s", part.Code, part.Body.String())
+				}
+				if _, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, []byte(`<VersioningConfiguration><Status>Enabled</Status></VersioningConfiguration>`)); err != nil {
+					t.Fatal(err)
+				}
+				body := fmt.Appendf(nil, `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>%s</ETag></Part></CompleteMultipartUpload>`, multipartConditionResponseETag(part))
+				rec := retentionAuthzRequest(t, router, root, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploadId": {mp.UploadID}}), body, nil)
+				if rec.Code != http.StatusOK {
+					t.Fatalf("ordinary upload initiated before versioning fails completion: %d %s", rec.Code, rec.Body.String())
+				}
+			})
+			t.Run(instance+"/addressed-upload-after-suspension", func(t *testing.T) {
+				key := "review-addressed-before-suspension"
+				original := []byte("original")
+				info, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(original), int64(len(original)), "", ""), ObjectOptions{Versioned: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true, VersionID: info.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				part := []byte("replaced existing version")
+				pi, err := obj.PutObjectPart(t.Context(), bucket, key, up.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(part), int64(len(part)), "", ""), ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if _, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig, []byte(`<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>`)); err != nil {
+					t.Fatal(err)
+				}
+				writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+				body := fmt.Appendf(nil, `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>"%s"</ETag></Part></CompleteMultipartUpload>`, pi.ETag)
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: info.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusForbidden || after.ETag != info.ETag {
+					t.Fatalf("suspended-bucket existing UUID overwritten: status=%d old-etag=%s new-etag=%s body=%s", rec.Code, info.ETag, after.ETag, rec.Body.String())
+				}
+			})
+		},
+	})
+}

cmd/object-retention-overwrite_test.go

@@ -0,0 +1,546 @@
+// Copyright (c) 2026 Feng Ruohang
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//
+// This file is part of Silo 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.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"crypto/md5"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/minio/minio/internal/amztime"
+	"github.com/minio/minio/internal/auth"
+	objectlock "github.com/minio/minio/internal/bucket/object/lock"
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+func retentionOverwriteUnchanged(t *testing.T, before, after ObjectInfo) {
+	t.Helper()
+	if after.ETag != before.ETag || after.Size != before.Size || after.VersionID != before.VersionID || !after.ModTime.Equal(before.ModTime) {
+		t.Error("rejected write changed the stored object")
+	}
+	if objectlock.GetObjectRetentionMeta(after.UserDefined).String() != objectlock.GetObjectRetentionMeta(before.UserDefined).String() || objectlock.GetObjectLegalHoldMeta(after.UserDefined).Status != objectlock.GetObjectLegalHoldMeta(before.UserDefined).Status {
+		t.Error("rejected write changed the stored lock")
+	}
+}
+
+func TestObjectLockOverwriteConfigUnavailable(t *testing.T) {
+	previous := newObjectLayerFn()
+	setObjectLayer(nil)
+	defer setObjectLayer(previous)
+	err := checkObjectLockedForOverwrite(t.Context(), "lock-config-unavailable", "object", func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) {
+		t.Fatal("object must not be read without a bucket configuration")
+		return ObjectInfo{}, nil
+	})
+	if !errors.Is(err, errServerNotInitialized) {
+		t.Fatalf("unavailable lock configuration must fail closed, got %v", err)
+	}
+}
+
+func TestPutObjectRetentionMaximumDaysPolicy(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, _ auth.Credentials, t *testing.T) {
+			for _, conditionalAllow := range []bool{false, true} {
+				t.Run(fmt.Sprintf("%s/conditional-allow=%t", instance, conditionalAllow), func(t *testing.T) {
+					statement := fmt.Sprintf(`{"Effect":"Allow","Action":["s3:PutObjectRetention"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"NumericLessThanEquals":{"s3:object-lock-remaining-retention-days":"200000"}}}`, bucket)
+					if !conditionalAllow {
+						statement = fmt.Sprintf(`{"Effect":"Allow","Action":["s3:PutObjectRetention"],"Resource":["arn:aws:s3:::%s/*"]},{"Effect":"Deny","Action":["s3:PutObjectRetention"],"Resource":["arn:aws:s3:::%s/*"],"Condition":{"NumericGreaterThan":{"s3:object-lock-remaining-retention-days":"200000"}}}`, bucket, bucket)
+					}
+					cred := newRetentionPolicyUser(t, instance, `{"Version":"2012-10-17","Statement":[`+statement+`]}`)
+					key := fmt.Sprintf("maximum-retention-%t", conditionalAllow)
+					initial, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{Versioned: true})
+					if err != nil {
+						t.Fatal(err)
+					}
+					// Both dates exceed time.Duration's range (~292 years). A saturated
+					// Time.Sub would report 106752 days and incorrectly allow 1000 years.
+					if status := putRetention(t, router, cred, bucket, key, initial.VersionID, "GOVERNANCE", UTCNow().AddDate(1000, 0, 0), false); status != http.StatusForbidden {
+						t.Fatalf("1000-year retention: want 403, got %d", status)
+					}
+					after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
+					if err != nil {
+						t.Fatal(err)
+					}
+					retentionOverwriteUnchanged(t, initial, after)
+					if status := putRetention(t, router, cred, bucket, key, initial.VersionID, "GOVERNANCE", UTCNow().AddDate(500, 0, 0), false); status != http.StatusOK {
+						t.Fatalf("500-year retention: want 200, got %d", status)
+					}
+				})
+			}
+		},
+	})
+}
+
+func TestAPICopyObjectPreservesVersionAfterSuspension(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			t.Run(instance, func(t *testing.T) {
+				key := "self-copy-after-suspension"
+				original, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("payload"), 7, "", ""), ObjectOptions{
+					Versioned: true, UserDefined: map[string]string{"x-amz-meta-state": "original"},
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if _, err := globalBucketMetadataSys.Update(t.Context(), bucket, bucketVersioningConfig,
+					[]byte(`<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>`)); err != nil {
+					t.Fatal(err)
+				}
+				rec := retentionAuthzRequest(t, router, root, http.MethodPut, makeTestTargetURL("", bucket, key, nil), nil, map[string]string{
+					xhttp.AmzCopySource: "/" + bucket + "/" + key, xhttp.AmzMetadataDirective: "REPLACE", "x-amz-meta-state": "replacement",
+				})
+				if rec.Code != http.StatusOK {
+					t.Fatalf("copy: %d %s", rec.Code, rec.Body.String())
+				}
+				old, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: original.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if old.UserDefined["x-amz-meta-state"] != "original" {
+					t.Errorf("copy changed metadata of existing UUID %s: %v", original.VersionID, old.UserDefined)
+				}
+				current, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatalf("copy must create the null version: %v", err)
+				}
+				if current.UserDefined["X-Amz-Meta-State"] != "replacement" {
+					t.Errorf("new null version metadata: %v", current.UserDefined)
+				}
+			})
+		},
+	})
+}
+
+// A persisted null upload can survive an upgrade or a versioning transition.
+// The completion must see locks committed after the handler's preflight.
+func TestMultipartNullObjectLockAfterPreflight(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			for _, mode := range []string{"GOVERNANCE", "COMPLIANCE", "LEGAL-HOLD"} {
+				t.Run(instance+"/"+mode, func(t *testing.T) {
+					ctx := t.Context()
+					key := "multipart-null-" + mode
+					opts := ObjectOptions{Versioned: true, VersionID: nullVersionID, ReplicationRequest: true}
+					if _, err := obj.PutObject(ctx, bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), opts); err != nil {
+						t.Fatal(err)
+					}
+					mp, err := obj.NewMultipartUpload(ctx, bucket, key, opts)
+					if err != nil {
+						t.Fatal(err)
+					}
+					part, err := obj.PutObjectPart(ctx, bucket, key, mp.UploadID, 1, mustGetPutObjReader(t, bytes.NewBufferString("replacement"), 11, "", ""), ObjectOptions{})
+					if err != nil {
+						t.Fatal(err)
+					}
+					var locked ObjectInfo
+					hook := &hookBeforeWrite{ObjectLayer: obj, key: key, before: func() {
+						if mode == "LEGAL-HOLD" {
+							body := []byte(`<LegalHold><Status>ON</Status></LegalHold>`)
+							sum := md5.Sum(body)
+							rec := retentionAuthzRequest(t, router, root, http.MethodPut, makeTestTargetURL("", bucket, key, url.Values{"legal-hold": {""}, "versionId": {nullVersionID}}), body, map[string]string{xhttp.ContentMD5: base64.StdEncoding.EncodeToString(sum[:])})
+							if rec.Code != http.StatusOK {
+								t.Fatalf("set legal hold: %d %s", rec.Code, rec.Body.String())
+							}
+						} else if status := putRetention(t, router, root, bucket, key, nullVersionID, mode, UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
+							t.Fatalf("set retention: %d", status)
+						}
+						var err error
+						locked, err = obj.GetObjectInfo(ctx, bucket, key, ObjectOptions{VersionID: nullVersionID})
+						if err != nil {
+							t.Fatal(err)
+						}
+					}}
+					setObjectLayer(hook)
+					defer setObjectLayer(obj)
+					body := []byte(multipartConditionCompleteBody([]CompletePart{{PartNumber: 1, ETag: part.ETag}}))
+					rec := retentionAuthzRequest(t, router, writer, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploadId": {mp.UploadID}}), body, nil)
+					if hook.before != nil || rec.Code != http.StatusBadRequest {
+						t.Fatalf("completion after locking: %d %s", rec.Code, rec.Body.String())
+					}
+					multipartConditionError(t, rec, "InvalidRequest")
+					after, err := obj.GetObjectInfo(ctx, bucket, key, ObjectOptions{VersionID: nullVersionID})
+					if err != nil {
+						t.Fatal(err)
+					}
+					retentionOverwriteUnchanged(t, locked, after)
+					parts, err := obj.ListObjectParts(ctx, bucket, key, mp.UploadID, 0, 100, ObjectOptions{})
+					if err != nil || len(parts.Parts) != 1 || parts.Parts[0].ETag != part.ETag {
+						t.Fatalf("rejected completion removed upload parts: %v %+v", err, parts.Parts)
+					}
+				})
+			}
+		},
+	})
+}
+
+// The newer, unlocked copy is in the write pool; the lock exists only in the
+// other pool. A set-local check would miss it even while holding a write lock.
+func TestPoolsObjectLockOverwrite(t *testing.T) {
+	z, _ := consistencyPools(t)
+	bucket, _, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, state := range []string{"GOVERNANCE", "COMPLIANCE", "LEGAL-HOLD", "expired", "unlocked", "unreadable"} {
+		for _, op := range []string{"put", "copy", "self-copy", "multipart"} {
+			t.Run(state+"/"+op, func(t *testing.T) {
+				ctx := t.Context()
+				key := encodeDirObject("null-" + state + "-" + op + "/")
+				meta := map[string]string{}
+				switch state {
+				case "GOVERNANCE", "COMPLIANCE":
+					meta[strings.ToLower(xhttp.AmzObjectLockMode)] = state
+					meta[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(UTCNow().Add(365 * 24 * time.Hour))
+				case "LEGAL-HOLD":
+					meta[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = "ON"
+				case "expired":
+					meta[strings.ToLower(xhttp.AmzObjectLockMode)] = "GOVERNANCE"
+					meta[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(UTCNow().Add(-24 * time.Hour))
+				}
+				seed := ObjectOptions{Versioned: true, VersionID: nullVersionID, ReplicationRequest: true, MTime: UTCNow().Add(-time.Hour), UserDefined: meta}
+				putConsistencyObject(t, z, bucket, key, 1, "original", seed)
+				seed.UserDefined = map[string]string{}
+				seed.MTime = UTCNow()
+				putConsistencyObject(t, z, bucket, key, 0, "original", seed)
+				before := make([]ObjectInfo, len(z.serverPools))
+				for i, pool := range z.serverPools {
+					var err error
+					before[i], err = pool.GetObjectInfo(ctx, bucket, key, ObjectOptions{VersionID: nullVersionID})
+					if err != nil {
+						t.Fatal(err)
+					}
+				}
+				defer conditionalPutPool(t, z, key, 0)()
+				var restore func()
+				if state == "unreadable" {
+					restore = conditionalPutSwapDisks(z.serverPools[1], key, func(d StorageAPI) StorageAPI {
+						return consistencyReadFaultDisk{StorageAPI: d, bucket: bucket, object: key}
+					})
+					defer restore()
+				}
+				opts := ObjectOptions{Versioned: true, VersionID: nullVersionID, UserDefined: map[string]string{"x-amz-meta-updated": "true"}}
+				var writeErr error
+				switch op {
+				case "put":
+					_, writeErr = z.PutObject(ctx, bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("replacement"), 11, "", ""), opts)
+				case "copy", "self-copy":
+					source := key
+					if op == "copy" {
+						source = "source-" + state
+						putConsistencyObject(t, z, bucket, source, 0, "replacement", ObjectOptions{Versioned: true})
+					}
+					gr, gerr := z.GetObjectNInfo(ctx, bucket, source, nil, nil, ObjectOptions{})
+					if gerr != nil {
+						t.Fatal(gerr)
+					}
+					defer gr.Close()
+					info := gr.ObjInfo
+					info.UserDefined = opts.UserDefined
+					info.metadataOnly = op == "self-copy"
+					info.PutObjReader = mustGetPutObjReader(t, gr, info.Size, "", "")
+					info.Reader = info.PutObjReader.Reader
+					_, writeErr = z.CopyObject(ctx, bucket, source, bucket, key, info, ObjectOptions{VersionID: nullVersionID}, opts)
+				case "multipart":
+					id, parts := multipartConditionUpload(t, z, bucket, key, 0, opts)
+					_, writeErr = z.CompleteMultipartUpload(ctx, bucket, key, id, parts, ObjectOptions{Versioned: true})
+				}
+				allowed := state == "expired" || state == "unlocked"
+				if allowed {
+					if writeErr != nil {
+						t.Fatalf("unlocked write: %v", writeErr)
+					}
+					return
+				}
+				if state == "unreadable" {
+					if !isErrReadQuorum(writeErr) {
+						t.Fatalf("unreadable pool: want quorum error, got %v", writeErr)
+					}
+					restore()
+				} else if !errors.As(writeErr, &ObjectLocked{}) {
+					t.Fatalf("lock in other pool: want ObjectLocked, got %v", writeErr)
+				}
+				for i, original := range before {
+					after, gerr := z.serverPools[i].GetObjectInfo(ctx, bucket, key, ObjectOptions{VersionID: nullVersionID})
+					if gerr != nil {
+						t.Fatal(gerr)
+					}
+					retentionOverwriteUnchanged(t, original, after)
+				}
+			})
+		}
+	}
+}
+
+type planReviewGatedBody struct {
+	io.ReadCloser
+	first   sync.Once
+	ready   chan struct{}
+	release chan struct{}
+}
+
+func (g *planReviewGatedBody) Read(p []byte) (int, error) {
+	g.first.Do(func() { close(g.ready); <-g.release })
+	return g.ReadCloser.Read(p)
+}
+
+func TestPlanReviewSlowDirectoryPut(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				key := "slow-body-directory/"
+				before, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				payload := []byte("replacement-body")
+				req, err := newTestSignedRequestV4(http.MethodPut, makeTestTargetURL("", bucket, key, nil), int64(len(payload)), bytes.NewReader(payload), writer.AccessKey, writer.SecretKey, nil)
+				if err != nil {
+					t.Fatal(err)
+				}
+				gate := &planReviewGatedBody{ReadCloser: req.Body, ready: make(chan struct{}), release: make(chan struct{})}
+				req.Body = gate
+				var once sync.Once
+				release := func() { once.Do(func() { close(gate.release) }) }
+				defer release()
+				rec := httptest.NewRecorder()
+				done := make(chan struct{})
+				go func() { router.ServeHTTP(rec, req); close(done) }()
+				select {
+				case <-gate.ready:
+				case <-done:
+					t.Fatalf("PUT finished before body: %d %s", rec.Code, rec.Body.String())
+				case <-time.After(5 * time.Second):
+					t.Fatal("PUT did not reach body read")
+				}
+				locked := make(chan int, 1)
+				go func() {
+					locked <- putRetention(t, router, root, bucket, key, nullVersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false)
+				}()
+				select {
+				case status := <-locked:
+					if status != http.StatusOK {
+						t.Fatalf("set lock while body paused: %d", status)
+					}
+				case <-time.After(5 * time.Second):
+					t.Fatal("retention update blocked behind paused body")
+				}
+				release()
+				select {
+				case <-done:
+				case <-time.After(5 * time.Second):
+					t.Fatal("PUT did not finish")
+				}
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusBadRequest || after.ETag != before.ETag || objectlock.GetObjectRetentionMeta(after.UserDefined).Mode != objectlock.RetCompliance {
+					t.Fatal("paused ordinary PUT overwrote the COMPLIANCE null version after the lock was acknowledged")
+				}
+			})
+		},
+	})
+}
+
+type planReviewReadFault struct {
+	ObjectLayer
+	key       string
+	fired     bool
+	putCalled bool
+}
+
+func (p *planReviewReadFault) GetObjectInfo(ctx context.Context, bucket, key string, opts ObjectOptions) (ObjectInfo, error) {
+	if key == p.key && opts.VersionID == nullVersionID && !p.fired {
+		p.fired = true
+		return ObjectInfo{}, InsufficientReadQuorum{}
+	}
+	return p.ObjectLayer.GetObjectInfo(ctx, bucket, key, opts)
+}
+
+func (p *planReviewReadFault) PutObject(ctx context.Context, bucket, key string, data *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+	if key == p.key {
+		p.putCalled = true
+	}
+	return p.ObjectLayer.PutObject(ctx, bucket, key, data, opts)
+}
+
+func TestPlanReviewPreflightReadFailure(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, _ auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				key := "locked-directory-read-fault/"
+				_, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{
+					Versioned: true, VersionID: nullVersionID, ReplicationRequest: true,
+					UserDefined: map[string]string{
+						strings.ToLower(xhttp.AmzObjectLockMode):            "COMPLIANCE",
+						strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): amztime.ISO8601Format(UTCNow().Add(365 * 24 * time.Hour)),
+					},
+				})
+				if err != nil {
+					t.Fatal(err)
+				}
+				initial, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				fault := &planReviewReadFault{ObjectLayer: obj, key: key}
+				setObjectLayer(fault)
+				defer setObjectLayer(obj)
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPut, makeTestTargetURL("", bucket, key, nil), []byte("replacement"), nil)
+				if !fault.fired {
+					t.Fatal("preflight fault was not injected")
+				}
+				if fault.putCalled {
+					t.Fatal("preflight read error allowed PutObject to proceed to storage layer")
+				}
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: nullVersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code < 400 {
+					t.Errorf("metadata read quorum error was treated as absence: status=%d", rec.Code)
+				}
+				retentionOverwriteUnchanged(t, initial, after)
+			})
+		},
+	})
+}
+
+func TestPlanReviewLegacyMPUCompliance(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				key := "legacy-mpu-compliance"
+				initial, err := obj.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("original"), 8, "", ""), ObjectOptions{Versioned: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true, VersionID: initial.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				payload := []byte("replacement-part")
+				part, err := obj.PutObjectPart(t.Context(), bucket, key, up.UploadID, 1, mustGetPutObjReader(t, bytes.NewReader(payload), int64(len(payload)), "", ""), ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if status := putRetention(t, router, root, bucket, key, initial.VersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
+					t.Fatalf("set compliance: %d", status)
+				}
+				before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				body := fmt.Appendf(nil, `<CompleteMultipartUpload><Part><PartNumber>1</PartNumber><ETag>"%s"</ETag></Part></CompleteMultipartUpload>`, part.ETag)
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPost, makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: initial.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusForbidden {
+					t.Errorf("legacy upload replaced locked UUID: status=%d", rec.Code)
+				}
+				retentionOverwriteUnchanged(t, before, after)
+			})
+		},
+	})
+}
+
+// A recovered upload retains an authentic server-generated version ID while a
+// surviving replica reintroduces its already-completed version and lock.
+// Only the fixture uses replication privileges; completion uses PutObject only.
+func TestRecoveredMultipartUploadAfterReplica(t *testing.T) {
+	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+		t:                 t,
+		makeBucketOptions: MakeBucketOptions{VersioningEnabled: true, LockEnabled: true},
+		objAPITest: func(obj ObjectLayer, instance, bucket string, router http.Handler, root auth.Credentials, t *testing.T) {
+			writer := newObjectAttributesAuthzUser(t, instance, bucket, `"s3:PutObject"`)
+			t.Run(instance, func(t *testing.T) {
+				z := obj.(*erasureServerPools)
+				key := "recovered-generated-upload"
+				up, err := obj.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{Versioned: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				er := z.serverPools[0].getHashedSet(key)
+				fi, _, err := er.checkUploadIDExists(t.Context(), bucket, key, up.UploadID, true)
+				if err != nil {
+					t.Fatal(err)
+				}
+				_, err = obj.PutObject(t.Context(), bucket, key,
+					mustGetPutObjReader(t, bytes.NewBufferString("protected-original"), 18, "", ""),
+					ObjectOptions{Versioned: true, VersionID: fi.VersionID, ReplicationRequest: true})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if status := putRetention(t, router, root, bucket, key, fi.VersionID, "COMPLIANCE", UTCNow().Add(365*24*time.Hour), false); status != http.StatusOK {
+					t.Fatalf("setting compliance lock: %d", status)
+				}
+				before, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: fi.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				partRec := retentionAuthzRequest(t, router, writer, http.MethodPut,
+					makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}, "partNumber": {"1"}}),
+					[]byte("attacker-replacement"), nil)
+				if partRec.Code != http.StatusOK {
+					t.Fatalf("upload part: %d %s", partRec.Code, partRec.Body.String())
+				}
+				etag := multipartConditionResponseETag(partRec)
+				body := []byte(multipartConditionCompleteBody([]CompletePart{{PartNumber: 1, ETag: etag}}))
+				rec := retentionAuthzRequest(t, router, writer, http.MethodPost,
+					makeTestTargetURL("", bucket, key, url.Values{"uploadId": {up.UploadID}}), body, nil)
+				after, err := obj.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: fi.VersionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if rec.Code != http.StatusForbidden {
+					t.Errorf("recovered upload bypassed existing UUID guard: status=%d", rec.Code)
+				}
+				retentionOverwriteUnchanged(t, before, after)
+			})
+		},
+	})
+}

cmd/replication-tagging-order_test.go

@@ -238,7 +238,7 @@ func r5APIOrdering(t *testing.T, encrypted bool) {
 						if err != nil {
 							t.Fatal(err)
 						}
-						original, err = obj.CompleteMultipartUpload(t.Context(), bucket, name, mp.UploadID, []CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{Versioned: true})
+						original, err = obj.CompleteMultipartUpload(t.Context(), bucket, name, mp.UploadID, []CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{Versioned: true, ReplicationRequest: true})
 						if err != nil {
 							t.Fatal(err)
 						}

docs/distributed/README.md

@@ -10,13 +10,13 @@ Silo in distributed mode can help you setup a highly-available storage system wi
 
 Distributed Silo provides protection against multiple node/drive failures and [bit rot](https://github.com/pgsty/silo/blob/main/docs/erasure/README.md#what-is-bit-rot-protection) using [erasure code](https://silo.pgsty.com/operations/concepts/erasure-coding/). As the minimum drives required for distributed Silo is 2 (same as minimum drives required for erasure coding), erasure code automatically kicks in as you launch distributed Silo.
 
-If one or more drives are offline at the start of a PutObject or NewMultipartUpload operation the object will have additional data protection bits added automatically to provide additional safety for these objects.
+With the default `availability` storage-class optimization, Silo can increase parity for new objects when drives are offline, up to half the drives in the erasure set. Writes must still meet the quorum for the resulting shard layout. This does not guarantee unchanged availability or keep a two-drive `EC:1` set writable after one drive fails. See the [storage-class reference](https://silo.pgsty.com/reference/minio-server/settings/storage-class/) for details.
 
 ### High availability
 
-A stand-alone Silo server would go down if the server hosting the drives goes offline. In contrast, a distributed Silo setup with _m_ servers and _n_ drives will have your data safe as long as _m/2_ servers or _m*n_/2 or more drives are online.
+A stand-alone Silo server becomes unavailable if its host goes offline. In a distributed deployment, determine node-failure tolerance from how many drives each failed node removes from every erasure set and from the parity recorded for the affected objects. Aggregate online node or drive counts alone do not establish read or write quorum.
 
-For example, an 16-server distributed setup with 200 drives per node would continue serving files, up to 4 servers can be offline in default configuration i.e around 800 drives down Silo would continue to read and write objects.
+For example, if a 16-node pool has 16-drive erasure sets with one drive per node in each set, a fixed `EC:4` layout has a read and write quorum of 12. Losing four nodes leaves 12 drives in every set. This result depends on that drive placement and parity; it cannot be generalized to arbitrary failures of the same total number of drives. In a two-node, two-drive `EC:1` set, losing one node leaves only one drive, so existing intact objects can remain readable but writes cannot continue.
 
 Refer to sizing guide for more understanding on default values chosen depending on your erasure stripe size [here](https://github.com/pgsty/silo/blob/main/docs/distributed/SIZING.md). Parity settings can be changed using [storage classes](https://github.com/pgsty/silo/tree/main/docs/erasure/storage-class).
 

docs/erasure/README.md

@@ -1,18 +1,18 @@
 # Silo Erasure Code Quickstart Guide
 
-Silo protects data against hardware failures and silent data corruption using erasure code and checksums. With the highest level of redundancy, you may lose up to half (N/2) of the total drives and still be able to recover the data.
+Silo protects data against hardware failures and silent data corruption using erasure coding and checksums. Recovery depends on the healthy shards and metadata remaining in each object's erasure set. With maximum parity, an object can tolerate the loss of up to `floor(N/2)` shards in its `N`-drive set; the deployment's total online drive count is not sufficient to determine recoverability.
 
 ## What is Erasure Code?
 
-Erasure code is a mathematical algorithm to reconstruct missing or corrupted data. Silo uses Reed-Solomon code to shard objects into variable data and parity blocks. For example, in a 12 drive setup, an object can be sharded to a variable number of data and parity blocks across all the drives - ranging from six data and six parity blocks to ten data and two parity blocks.
+Erasure coding uses mathematical algorithms to reconstruct missing or corrupted data. Silo uses Reed-Solomon codes to split objects into data and parity shards. For a 12-drive erasure set, parity can range from `EC:0` (12 data shards and no parity) to `EC:6` (six data shards and six parity shards).
 
-By default, Silo shards the objects across N/2 data and N/2 parity drives. Though, you can use [storage classes](https://github.com/pgsty/silo/tree/main/docs/erasure/storage-class) to use a custom configuration. We recommend N/2 data and parity blocks, as it ensures the best protection from drive failures.
+The default parity depends on the erasure set size: `EC:0` for 1 drive, `EC:1` for 2–3 drives, `EC:2` for 4–5 drives, `EC:3` for 6–7 drives, and `EC:4` for 8–16 drives. See the [Silo storage-class reference](https://silo.pgsty.com/reference/minio-server/settings/storage-class/) for configuration details.
 
-In 12 drive example above, with Silo server running in the default configuration, you can lose any of the six drives and still reconstruct the data reliably from the remaining drives.
+In the 12-drive example, the default `EC:4` layout uses eight data shards and four parity shards. An existing object can tolerate four unavailable shards if the remaining shards and metadata are intact. Explicit `EC:6` instead uses six data and six parity shards, with a read quorum of six and a write quorum of seven.
 
 ## Why is Erasure Code useful?
 
-Erasure code protects data from multiple drives failure, unlike RAID or replication. For example, RAID6 can protect against two drive failure whereas in Silo erasure code you can lose as many as half of drives and still the data remains safe. Further, Silo's erasure code is at the object level and can heal one object at a time. For RAID, healing can be done only at the volume level which translates into high downtime. As Silo encodes each object individually, it can heal objects incrementally. Storage servers once deployed should not require drive replacement or healing for the lifetime of the server. Silo's erasure coded backend is designed for operational efficiency and takes full advantage of hardware acceleration whenever available.
+Erasure coding protects objects against drive failures within their erasure set, up to the parity recorded for each object. Silo encodes objects individually and can heal them incrementally when enough healthy shards and consistent metadata remain. Failed drives still need repair or replacement to restore redundancy; erasure coding does not eliminate that maintenance.
 
 ![Erasure](https://github.com/pgsty/silo/blob/main/docs/screenshots/erasure-code.jpg?raw=true)
 
@@ -64,4 +64,4 @@ podman run \
 
 ### 3. Test your setup
 
-You may unplug drives randomly and continue to perform I/O on the system.
+In a test deployment, take drives offline and verify reads and writes against the quorum required by each erasure set and object layout. Restore the drives after each test; continued I/O depends on the remaining healthy shards and metadata.

docs/erasure/storage-class/README.md

@@ -38,31 +38,23 @@ You can calculate _approximate_ storage usage ratio using the formula - total dr
 
 ### Allowed values for STANDARD storage class
 
-`STANDARD` storage class implies more parity than `REDUCED_REDUNDANCY` class. So, `STANDARD` parity drives should be
+`STANDARD` supports `EC:0` without erasure-code redundancy and nonzero parity values up to `floor(N/2)`, where `N` is the number of drives in the erasure set. When both `STANDARD` and `REDUCED_REDUNDANCY` parity are nonzero, `STANDARD` must be greater than or equal to `REDUCED_REDUNDANCY`; equal parity is allowed.
 
-- Greater than or equal to 2, if `REDUCED_REDUNDANCY` parity is not set.
-- Greater than `REDUCED_REDUNDANCY` parity, if it is set.
+The default `STANDARD` parity is:
 
-Parity blocks can not be higher than data blocks, so `STANDARD` storage class parity can not be higher than N/2. (N being total number of drives)
-
-The default value for the `STANDARD` storage class depends on the number of volumes in the erasure set:
-
-| Erasure Set Size | Default Parity (EC:N) |
-|------------------|-----------------------|
-| 5 or fewer       |                 EC:2  |
-| 6-7              |                 EC:3  |
-| 8 or more        |                 EC:4  |
+| Erasure Set Size | Default Parity (EC:M) |
+| --- | --- |
+| 1 | EC:0 |
+| 2–3 | EC:1 |
+| 4–5 | EC:2 |
+| 6–7 | EC:3 |
+| 8–16 | EC:4 |
 
 For more complete documentation on Erasure Set sizing, see the [Silo Documentation on Erasure Sets](https://silo.pgsty.com/operations/concepts/erasure-coding/#minio-ec-erasure-set).
 
 ### Allowed values for REDUCED_REDUNDANCY storage class
 
-`REDUCED_REDUNDANCY` implies lesser parity than `STANDARD` class. So,`REDUCED_REDUNDANCY` parity drives should be
-
-- Less than N/2, if `STANDARD` parity is not set.
-- Less than `STANDARD` Parity, if it is set.
-
-Default value for `REDUCED_REDUNDANCY` storage class is `1`.
+`REDUCED_REDUNDANCY` parity can be zero or up to `floor(N/2)`. When both storage classes have nonzero parity, its parity must be less than or equal to `STANDARD` parity. The default is `EC:1` for multi-drive sets and `EC:0` for single-drive deployments. See the [Silo storage-class reference](https://silo.pgsty.com/reference/minio-server/settings/storage-class/) for the maintained configuration documentation.
 
 ## Get started with Storage Class
 

go.mod

@@ -1,6 +1,6 @@
 module github.com/minio/minio
 
-go 1.27.1
+go 1.27.2
 
 // Console and MC retain their historical module paths for best-effort upstream
 // compatibility. Pin the maintained PGSTY implementations used by SILO.
@@ -294,7 +294,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.45.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/mod v0.41.0 // indirect
-	golang.org/x/net v0.59.0 // indirect
+	golang.org/x/net v0.60.0 // indirect
 	golang.org/x/text v0.42.0 // indirect
 	golang.org/x/tools v0.50.0 // indirect
 	google.golang.org/genproto v0.0.0-20260319201613-d00831a3d3e7 // indirect

go.sum

@@ -762,8 +762,8 @@ golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug
 golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws=
 golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
-golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues=
-golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
+golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU=
+golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
 golang.org/x/oauth2 v0.37.0 h1:JUlcxA8oAtauLfiH8FX2/FkAWHAdi0QtGCGc+hofE98=
 golang.org/x/oauth2 v0.37.0/go.mod h1:IxwZNxUULJmpBFf9K/9NTMSIfZZuvuTy1gGxhigP/58=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

internal/config/lambda/event/targetidset.go

@@ -24,7 +24,7 @@ type TargetIDSet map[TargetID]struct{}
 
 // IsEmpty returns true if the set is empty.
 func (set TargetIDSet) IsEmpty() bool {
-	return len(set) != 0
+	return len(set) == 0
 }
 
 // Clone - returns copy of this set.

internal/config/lambda/event/targetidset_test.go

@@ -22,6 +22,22 @@ import (
 	"testing"
 )
 
+func TestTargetIDSetIsEmpty(t *testing.T) {
+	testCases := []struct {
+		set      TargetIDSet
+		expected bool
+	}{
+		{NewTargetIDSet(), true},
+		{NewTargetIDSet(TargetID{"1", "webhook"}), false},
+	}
+
+	for i, testCase := range testCases {
+		if got := testCase.set.IsEmpty(); got != testCase.expected {
+			t.Fatalf("test %v: expected: %v, got: %v", i+1, testCase.expected, got)
+		}
+	}
+}
+
 func TestTargetIDSetClone(t *testing.T) {
 	testCases := []struct {
 		set           TargetIDSet