diff --git a/buildscripts/rebrand-guard/compat-baseline.json b/buildscripts/rebrand-guard/compat-baseline.json
index 84ee7a4bc..8ae0f32dc 100644
--- a/buildscripts/rebrand-guard/compat-baseline.json
+++ b/buildscripts/rebrand-guard/compat-baseline.json
@@ -631,6 +631,7 @@
"x-minio-replication-encrypted-multipart",
"x-minio-replication-ready",
"x-minio-replication-reset-status",
+ "x-minio-replication-server-side-encryption",
"x-minio-replication-server-side-encryption-iv",
"x-minio-replication-server-side-encryption-seal-algorithm",
"x-minio-replication-server-side-encryption-sealed-key",
diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go
index e8cb31114..e5828e2dd 100644
--- a/cmd/bucket-object-lock.go
+++ b/cmd/bucket-object-lock.go
@@ -25,6 +25,7 @@ import (
"strings"
"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"
@@ -407,3 +408,116 @@ func (s objectLockState) restoreLegalHold(metadata map[string]string) {
}
metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = s.legalHold
}
+
+// replicaStoredLock reads the Object Lock state stored on the addressed version
+// so a trusted replica write can order its update against it. A missing object
+// or version yields an empty state, which is correct for the first write of a
+// version; any other read error is returned so the caller fails the write rather
+// than ordering an incoming update against lock state it merely failed to read
+// (an older incoming value must not win over a newer stored one just because the
+// read timed out).
+func replicaStoredLock(ctx context.Context, getObjectInfo GetObjectInfoFn, bucket, object, versionID string) (objectLockState, error) {
+ oi, err := getObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: versionID})
+ switch {
+ case err == nil:
+ return storedObjectLockState(oi.UserDefined), nil
+ case isErrObjectNotFound(err) || isErrVersionNotFound(err):
+ return objectLockState{}, nil
+ default:
+ return objectLockState{}, err
+ }
+}
+
+// applyReplicatedObjectLock writes the retention and legal-hold decision into
+// metadata for a PUT, CopyObject, or multipart-initiation request. A request
+// that is not an actual trusted replica -- a normal user write, or a trusted
+// peer that carried the replication marker without REPLICA status -- takes
+// ordinary write semantics: a validated value is applied and stamped now, and a
+// missing value is left as is. Only an actual replica update is ordered against
+// the state already stored on the addressed version, so a stale value cannot
+// overwrite a newer one and a full retransmit cannot roll a destination back.
+// The stored argument is meaningful only for a replica; callers pass an empty
+// state otherwise. Only the two Object Lock keys and their reserved ordering
+// timestamps are touched; any encryption-metadata reconciliation stays with the
+// caller.
+func applyReplicatedObjectLock(metadata map[string]string, stored objectLockState,
+ replicaTrusted bool,
+ retentionMode objectlock.RetMode, retentionDate objectlock.RetentionDate,
+ legalHold objectlock.ObjectLegalHold, srcRetentionTimestamp, srcLegalholdTimestamp time.Time,
+) {
+ switch {
+ case !replicaTrusted:
+ // Ordinary write semantics: apply a validated retention and stamp it now;
+ // a missing value carries no instruction, so leave the metadata as it is.
+ if retentionMode.Valid() {
+ metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
+ metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
+ metadata[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ }
+ case !stored.retentionIsOlderThan(srcRetentionTimestamp):
+ // The stored update is at least as new as this replica's, or the replica
+ // carries no ordering timestamp: keep what is stored. This is also how a
+ // stale retransmit is rejected.
+ stored.restoreRetention(metadata)
+ default:
+ // The replica update wins. A removal carries no value but still records
+ // the source timestamp that orders it.
+ if retentionMode.Valid() {
+ metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
+ metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
+ }
+ metadata[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = srcRetentionTimestamp.UTC().Format(time.RFC3339Nano)
+ }
+
+ // Legal hold has no removal in S3: an explicitly empty header is already
+ // rejected as an invalid status, so the only value-less shape that gets here
+ // is an absent one, which conveys no legal-hold change. Only a valid status
+ // can win.
+ switch {
+ case !replicaTrusted:
+ if legalHold.Status.Valid() {
+ metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
+ metadata[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano)
+ }
+ case legalHold.Status.Valid() && stored.legalHoldIsOlderThan(srcLegalholdTimestamp):
+ metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
+ metadata[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = srcLegalholdTimestamp.UTC().Format(time.RFC3339Nano)
+ default:
+ stored.restoreLegalHold(metadata)
+ }
+}
+
+// reconcileStoredObjectLock re-orders the Object Lock already written into
+// metadata against the state currently stored on the destination version, both
+// compared by their reserved ordering timestamps. It runs inside the object
+// layer under the namespace write lock that guards the version replacement,
+// after the destination version is read and before the new one is committed, so
+// a replica update whose ordering was decided at handler time (or, for multipart,
+// at initiation) cannot overwrite a newer lock update that reached the version in
+// between. metadata already carries the incoming update with its source
+// timestamps; a stored value that is not older than the incoming one is put back,
+// which for a stored removal means clearing the incoming value and keeping only
+// the removal's timestamp. Only the two lock keys and their reserved timestamps
+// move; a non-replica write never sets the flag that invokes this.
+func reconcileStoredObjectLock(metadata map[string]string, stored objectLockState) {
+ incoming := storedObjectLockState(metadata)
+
+ incomingRetentionTS, _ := time.Parse(time.RFC3339Nano, incoming.retentionTimestamp)
+ if !stored.retentionIsOlderThan(incomingRetentionTS) {
+ // The stored retention is at least as new as the incoming one (or the
+ // incoming update is unordered): drop the incoming value and put the stored
+ // state back, which may itself be a removal (value keys absent, timestamp
+ // present).
+ delete(metadata, strings.ToLower(xhttp.AmzObjectLockMode))
+ delete(metadata, strings.ToLower(xhttp.AmzObjectLockRetainUntilDate))
+ delete(metadata, ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp)
+ stored.restoreRetention(metadata)
+ }
+
+ incomingLegalHoldTS, _ := time.Parse(time.RFC3339Nano, incoming.legalHoldTimestamp)
+ if incoming.legalHold == "" || !stored.legalHoldIsOlderThan(incomingLegalHoldTS) {
+ delete(metadata, strings.ToLower(xhttp.AmzObjectLockLegalHold))
+ delete(metadata, ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp)
+ stored.restoreLegalHold(metadata)
+ }
+}
diff --git a/cmd/bucket-replication.go b/cmd/bucket-replication.go
index 920c543d0..def8a74f8 100644
--- a/cmd/bucket-replication.go
+++ b/cmd/bucket-replication.go
@@ -872,19 +872,29 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put
if cc, ok := lkMap.Lookup(xhttp.CacheControl); ok {
putOpts.CacheControl = cc
}
- if mode, ok := lkMap.Lookup(xhttp.AmzObjectLockMode); ok {
- rmode := minio.RetentionMode(mode)
- putOpts.Mode = rmode
+ mode, hasMode := lkMap.Lookup(xhttp.AmzObjectLockMode)
+ retainDateStr, hasRetainDate := lkMap.Lookup(xhttp.AmzObjectLockRetainUntilDate)
+ if hasMode {
+ putOpts.Mode = minio.RetentionMode(mode)
}
- if retainDateStr, ok := lkMap.Lookup(xhttp.AmzObjectLockRetainUntilDate); ok {
+ // A removed retention is stored as an empty or absent mode and date; it is
+ // sent as a value-less update that still carries its ordering timestamp.
+ if hasRetainDate && retainDateStr != "" {
rdate, err := amztime.ISO8601Parse(retainDateStr)
if err != nil {
return putOpts, false, err
}
putOpts.RetainUntilDate = rdate
- // set retention timestamp in opts
+ }
+ retainTmstampStr, hasRetainTmstamp := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]
+ if hasMode || hasRetainDate || hasRetainTmstamp {
+ // Send the ordering timestamp whenever the version carries one, even for a
+ // removal whose value keys are absent (the shape a retransmit PUT leaves),
+ // so the next hop can order the removal instead of keeping obsolete
+ // retention.
retTimestamp := objInfo.ModTime
- if retainTmstampStr, ok := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]; ok {
+ if hasRetainTmstamp {
+ var err error
retTimestamp, err = time.Parse(time.RFC3339Nano, retainTmstampStr)
if err != nil {
return putOpts, false, err
@@ -1615,11 +1625,14 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
return rinfo
}
} else {
- // SSEC objects will refuse HeadObject without the decryption key.
- // Ignore the error, since we know the object exists and versioning prevents overwriting existing versions.
+ // The sender holds no customer key, so the target refuses HeadObject on
+ // an SSE-C object and the replica cannot be compared. The metadata-only
+ // CopyObject that a replicateMetadata action would run then fails on any
+ // non-empty object, because the undecryptable source checksum makes the
+ // target recompute one and rewrite the data. A full retransmit is the
+ // only action that completes.
if isSSEC && strings.Contains(cerr.Error(), errorCodes[ErrSSEEncryptedObject].Description) {
- rinfo.ReplicationStatus = replication.Completed
- rinfo.ReplicationAction = replicateNone
+ rAction = replicateAll
goto applyAction
}
// if target returns error other than NoSuchKey, defer replication attempt
@@ -1704,8 +1717,11 @@ applyAction:
} else {
putOpts, isMP, err := putReplicationOpts(ctx, tgt.StorageClass, objInfo)
if err != nil {
- rinfo.Err = err
+ // rinfo was primed Completed above; a failure to build the write
+ // options means nothing reached the target, so mark it Failed and
+ // carry the error instead of reporting a phantom success.
rinfo.ReplicationStatus = replication.Failed
+ rinfo.Err = err
replLogIf(ctx, fmt.Errorf("failed to set replicate options for object %s/%s(%s) (target %s) err:%w", bucket, objInfo.Name, objInfo.VersionID, tgt.EndpointURL(), err))
sendEvent(eventArgs{
EventName: event.ObjectReplicationNotTracked,
diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go
index 35c790805..9de330e28 100644
--- a/cmd/erasure-multipart.go
+++ b/cmd/erasure-multipart.go
@@ -1114,7 +1114,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
auditObjectErasureSet(ctx, "CompleteMultipartUpload", object, &er)
}
- if opts.CheckPrecondFn != nil {
+ if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile {
if !opts.NoLock {
ns := er.NewNSLock(bucket, object)
lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
@@ -1126,18 +1126,24 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
opts.NoLock = true
}
- obj, err := er.getObjectInfo(ctx, bucket, object, opts)
- if err == nil && opts.CheckPrecondFn(obj) {
- return ObjectInfo{}, PreConditionFailed{}
- }
- if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
- return ObjectInfo{}, err
- }
+ // The Object Lock reconcile below needs the version being committed, read
+ // after checkUploadIDExists, so only the precondition read happens here;
+ // both run under this same write lock, held until the version is renamed
+ // into place.
+ if opts.CheckPrecondFn != nil {
+ obj, err := er.getObjectInfo(ctx, bucket, object, opts)
+ if err == nil && opts.CheckPrecondFn(obj) {
+ return ObjectInfo{}, PreConditionFailed{}
+ }
+ if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
+ return ObjectInfo{}, err
+ }
- // if object doesn't exist return error for If-Match conditional requests
- // If-None-Match should be allowed to proceed for non-existent objects
- if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
- return ObjectInfo{}, err
+ // if object doesn't exist return error for If-Match conditional requests
+ // If-None-Match should be allowed to proceed for non-existent objects
+ if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
+ return ObjectInfo{}, err
+ }
}
}
@@ -1149,6 +1155,42 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
return oi, toObjectErr(err, bucket, object, uploadID)
}
+ // A trusted SSE-C replica completion re-orders the Object Lock carried in the
+ // upload metadata against the version it is about to replace, read on this
+ // erasure set under the write lock held above, so a hold or retention that
+ // reached the version after this upload was initiated is not rolled back at
+ // completion (issue #120). Scoped to SSE-C uploads, the only ones this issue
+ // routes through completion.
+ //
+ // Scope: correct for a single erasure set. A multi-pool deployment (duplicate
+ // versions across pools, ModTime ties, cross-pool lock authority) is out of
+ // scope and tracked in pgsty/silo#133.
+ if opts.ReplicaLockReconcile && crypto.SSEC.IsEncrypted(fi.Metadata) {
+ // A persisted upload records the null version as an empty VersionID; look
+ // it up as the null version so the reconcile reads the addressed version's
+ // stored lock, not the latest version's.
+ lookupVersionID := fi.VersionID
+ if lookupVersionID == "" {
+ lookupVersionID = nullVersionID
+ }
+ curr, gerr := er.getObjectInfo(ctx, bucket, object, ObjectOptions{
+ VersionID: lookupVersionID,
+ Versioned: opts.Versioned,
+ VersionSuspended: opts.VersionSuspended,
+ NoLock: true,
+ })
+ switch {
+ case gerr == nil:
+ reconcileStoredObjectLock(fi.Metadata, storedObjectLockState(curr.UserDefined))
+ case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr):
+ // No existing version to order against: keep the upload's own accepted
+ // lock, including a pre-upgrade upload that persisted values without
+ // their ordering timestamps.
+ default:
+ return oi, toObjectErr(gerr, bucket, object)
+ }
+ }
+
uploadIDPath := er.getUploadIDDir(bucket, object, uploadID)
onlineDisks := er.getDisks()
writeQuorum := fi.WriteQuorum(er.defaultWQuorum())
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index db5e77ed2..bade406e6 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -1268,7 +1268,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
data := r.Reader
- if opts.CheckPrecondFn != nil {
+ if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile {
if !opts.NoLock {
ns := er.NewNSLock(bucket, object)
lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
@@ -1281,17 +1281,33 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
}
obj, err := er.getObjectInfo(ctx, bucket, object, opts)
- if err == nil && opts.CheckPrecondFn(obj) {
- return objInfo, PreConditionFailed{}
- }
+ // A destination read that fails for a reason other than not-found must not
+ // be taken as a passed precondition or as absent lock state.
if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
return objInfo, err
}
+ if opts.CheckPrecondFn != nil {
+ if err == nil && opts.CheckPrecondFn(obj) {
+ return objInfo, PreConditionFailed{}
+ }
+ // if object doesn't exist return error for If-Match conditional requests
+ // If-None-Match should be allowed to proceed for non-existent objects
+ if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
+ return objInfo, err
+ }
+ }
- // if object doesn't exist return error for If-Match conditional requests
- // If-None-Match should be allowed to proceed for non-existent objects
- if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
- return objInfo, err
+ // Order this trusted SSE-C replica's Object Lock against the addressed
+ // version's stored state, read on this erasure set under the write lock,
+ // so a value that lost the ordering cannot overwrite a newer one committed
+ // after the handler decided (issue #120). Only reconcile against an
+ // existing version; on not-found the write's own accepted lock is kept.
+ //
+ // Scope: correct for a single erasure set. A multi-pool deployment
+ // (duplicate versions across pools, ModTime ties, cross-pool lock
+ // authority) is out of scope and tracked in pgsty/silo#133.
+ if opts.ReplicaLockReconcile && err == nil {
+ reconcileStoredObjectLock(opts.UserDefined, storedObjectLockState(obj.UserDefined))
}
}
diff --git a/cmd/object-api-interface.go b/cmd/object-api-interface.go
index f8664310d..3361a9ced 100644
--- a/cmd/object-api-interface.go
+++ b/cmd/object-api-interface.go
@@ -99,6 +99,7 @@ type ObjectOptions struct {
ReplicationSourceTaggingTimestamp time.Time // set if MinIOSourceTaggingTimestamp received
ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received
ReplicationSourceRetentionTimestamp time.Time // set if MinIOSourceObjectRetentionTimestamp received
+ ReplicaLockReconcile bool // set for a trusted SSE-C replica full write/completion: re-order Object Lock against the destination version read under the write lock (single erasure set; see pgsty/silo#133)
DeletePrefix bool // set true to enforce a prefix deletion, only application for DeleteObject API,
DeletePrefixObject bool // set true when object's erasure set is resolvable by object name (using getHashedSetIndex)
diff --git a/cmd/object-handlers-common.go b/cmd/object-handlers-common.go
index 90009112a..f8de11ab8 100644
--- a/cmd/object-handlers-common.go
+++ b/cmd/object-handlers-common.go
@@ -28,6 +28,7 @@ import (
"github.com/minio/minio/internal/amztime"
"github.com/minio/minio/internal/bucket/lifecycle"
+ "github.com/minio/minio/internal/crypto"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/hash"
xhttp "github.com/minio/minio/internal/http"
@@ -193,7 +194,15 @@ func checkPreconditionsPUT(ctx context.Context, w http.ResponseWriter, r *http.R
etagMatch := opts.PreserveETag != "" && isETagEqual(objInfo.ETag, opts.PreserveETag)
vidMatch := opts.VersionID != "" && opts.VersionID == objInfo.VersionID
- if etagMatch && vidMatch {
+ // A matching version and ETag normally mean the destination already holds
+ // this version, so the write is skipped. They do not establish that for an
+ // authenticated SSE-C replica write: the destination cannot decrypt or
+ // re-encrypt the body without the customer key, so it cannot verify the
+ // replica, and this retransmission is how such a replica is repaired or
+ // updated. The predicate is the incoming request's restored SSE-C metadata,
+ // not what the destination happens to hold.
+ ssecReplica := isReplicaTrusted(r.Context()) && crypto.SSEC.IsEncrypted(opts.UserDefined)
+ if etagMatch && vidMatch && !ssecReplica {
writeHeaders()
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrPreconditionFailed), r.URL)
return true
diff --git a/cmd/object-handlers.go b/cmd/object-handlers.go
index fcb834821..331393292 100644
--- a/cmd/object-handlers.go
+++ b/cmd/object-handlers.go
@@ -1771,50 +1771,9 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
// apply default bucket configuration/governance headers for dest side.
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted)
if s3Err == ErrNone {
- // A replica update is ordered by its source timestamp alone, whether or
- // not it still carries a value: an update newer than the stored state
- // applies, and a removal is just an update with no value. An update that
- // is stale, or that carries no source timestamp at all and is therefore
- // unordered, leaves the stored state in place instead of erasing it.
- switch {
- case !dstOpts.ReplicationRequest:
- if retentionMode.Valid() {
- srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
- srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
- srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = UTCNow().Format(time.RFC3339Nano)
- }
- case !replicaTrusted && !retentionMode.Valid():
- // A trusted peer that did not mark this request as a replica sends
- // no replicated state, so a missing value carries no instruction and
- // the rebuilt metadata is left as it is.
- case !storedLock.retentionIsOlderThan(dstOpts.ReplicationSourceRetentionTimestamp):
- storedLock.restoreRetention(srcInfo.UserDefined)
- default:
- if retentionMode.Valid() {
- srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
- srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
- }
- srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp] = dstOpts.ReplicationSourceRetentionTimestamp.UTC().Format(time.RFC3339Nano)
- }
-
- // Legal hold has no removal in S3: an explicitly empty header is already
- // rejected as an invalid status, so the only value-less shape that gets
- // here is an absent one, and that conveys no legal-hold change even when
- // an orphaned timestamp comes with it. Only a valid status can win.
- switch {
- case !dstOpts.ReplicationRequest:
- if legalHold.Status.Valid() {
- srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
- srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = UTCNow().Format(time.RFC3339Nano)
- }
- case !replicaTrusted && !legalHold.Status.Valid():
- // As above: a marker-only request carries no legal-hold update.
- case legalHold.Status.Valid() && storedLock.legalHoldIsOlderThan(dstOpts.ReplicationSourceLegalholdTimestamp):
- srcInfo.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
- srcInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = dstOpts.ReplicationSourceLegalholdTimestamp.UTC().Format(time.RFC3339Nano)
- default:
- storedLock.restoreLegalHold(srcInfo.UserDefined)
- }
+ applyReplicatedObjectLock(srcInfo.UserDefined, storedLock, replicaTrusted,
+ retentionMode, retentionDate, legalHold,
+ dstOpts.ReplicationSourceRetentionTimestamp, dstOpts.ReplicationSourceLegalholdTimestamp)
if replicaTrusted {
// An SSE-C key rotation snapshots every stored reserved key into
@@ -2268,9 +2227,21 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
r.Header.Get(xhttp.IfMatch) != "" ||
r.Header.Get(xhttp.IfNoneMatch) != "" {
opts.CheckPrecondFn = func(oi ObjectInfo) bool {
- if _, err := DecryptObjectInfo(&oi, r); err != nil {
- writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
- return true
+ // A pure raw SSE-C replica overwrite (no public precondition) fully
+ // replaces the stored object, so the destination must not first
+ // require the stored object to decrypt: a replica an older destination
+ // bug left as compress(ciphertext) or double-encrypted has an invalid
+ // decrypted length, and requiring it here blocks the retransmission
+ // that repairs it. The predicate is the incoming request's restored
+ // SSE-C metadata, the same one checkPreconditionsPUT uses to exempt the
+ // version/ETag duplicate. A conditional request (If-Match/If-None-Match)
+ // still needs the decrypted, client-visible ETag, so it keeps the check.
+ ssecReplica := isReplicaTrusted(ctx) && crypto.SSEC.IsEncrypted(opts.UserDefined)
+ if !ssecReplica || r.Header.Get(xhttp.IfMatch) != "" || r.Header.Get(xhttp.IfNoneMatch) != "" {
+ if _, err := DecryptObjectInfo(&oi, r); err != nil {
+ writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
+ return true
+ }
}
return checkPreconditionsPUT(ctx, w, r, oi, opts)
}
@@ -2282,12 +2253,31 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
getObjectInfo := objectAPI.GetObjectInfo
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, isReplicaTrusted(ctx))
- if s3Err == ErrNone && retentionMode.Valid() {
- metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
- metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
- }
- if s3Err == ErrNone && legalHold.Status.Valid() {
- metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
+ if s3Err == ErrNone {
+ // A trusted replica write addressing a specific version can be a full
+ // retransmit over an existing version whose lock state is newer than the
+ // source snapshot (issue #120). Order the incoming update against what is
+ // stored so a stale value cannot overwrite it; a non-replica write (and a
+ // marker-only peer write) has no stored state to order against and takes
+ // the helper's ordinary-write branch.
+ var storedLock objectLockState
+ if isReplicaTrusted(ctx) && opts.VersionID != "" {
+ var lerr error
+ if storedLock, lerr = replicaStoredLock(ctx, getObjectInfo, bucket, object, opts.VersionID); lerr != nil {
+ writeErrorResponse(ctx, w, toAPIError(ctx, lerr), r.URL)
+ return
+ }
+ }
+ applyReplicatedObjectLock(metadata, storedLock, isReplicaTrusted(ctx),
+ retentionMode, retentionDate, legalHold,
+ opts.ReplicationSourceRetentionTimestamp, opts.ReplicationSourceLegalholdTimestamp)
+ // The decision above orders against the version as read here; let the
+ // object layer re-run it against the version read under the write lock
+ // that guards the replacement, so a newer hold or retention committed in
+ // between is not rolled back (issue #120). Scoped to the SSE-C replica
+ // retransmit this issue enables, keyed on the incoming write's restored
+ // SSE-C seal, the same predicate as the duplicate-version exemption.
+ opts.ReplicaLockReconcile = isReplicaTrusted(ctx) && opts.VersionID != "" && crypto.SSEC.IsEncrypted(metadata)
}
if s3Err != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
diff --git a/cmd/object-multipart-handlers.go b/cmd/object-multipart-handlers.go
index 22d2839f1..983644223 100644
--- a/cmd/object-multipart-handlers.go
+++ b/cmd/object-multipart-handlers.go
@@ -34,7 +34,6 @@ import (
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/encrypt"
"github.com/minio/minio-go/v7/pkg/tags"
- "github.com/minio/minio/internal/amztime"
sse "github.com/minio/minio/internal/bucket/encryption"
objectlock "github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/replication"
@@ -260,12 +259,32 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
getObjectInfo := objectAPI.GetObjectInfo
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, replicaTrusted)
- if s3Err == ErrNone && retentionMode.Valid() {
- metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
- metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
- }
- if s3Err == ErrNone && legalHold.Status.Valid() {
- metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
+ if s3Err == ErrNone {
+ // A trusted replica NewMultipartUpload addressing a specific version can
+ // be a full retransmit over an existing version whose lock state is newer
+ // than the source snapshot (issue #120). Order the incoming update against
+ // what is stored so a stale value cannot overwrite it. opts is built below
+ // (its ServerSideEncryption depends on the encMetadata merge that has not
+ // happened yet), so read the replica ordering inputs the way
+ // putOptsFromHeaders will; a malformed timestamp fails the request when
+ // opts is built, so a parse error here is left as a zero time.
+ var (
+ storedLock objectLockState
+ srcRetentionTS, srcLegalholdTS time.Time
+ )
+ if replicaTrusted {
+ srcRetentionTS, _ = time.Parse(time.RFC3339, strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceObjectRetentionTimestamp)))
+ srcLegalholdTS, _ = time.Parse(time.RFC3339, strings.TrimSpace(r.Header.Get(xhttp.MinIOSourceObjectLegalHoldTimestamp)))
+ if versionID := strings.TrimSpace(r.Form.Get(xhttp.VersionID)); versionID != "" {
+ var lerr error
+ if storedLock, lerr = replicaStoredLock(ctx, getObjectInfo, bucket, object, versionID); lerr != nil {
+ writeErrorResponse(ctx, w, toAPIError(ctx, lerr), r.URL)
+ return
+ }
+ }
+ }
+ applyReplicatedObjectLock(metadata, storedLock, replicaTrusted,
+ retentionMode, retentionDate, legalHold, srcRetentionTS, srcLegalholdTS)
}
if s3Err != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
@@ -1166,6 +1185,14 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
}
opts.Versioned = versioned
opts.VersionSuspended = suspended
+ // A replicated multipart completion carries the internal replication marker
+ // (the sender does not re-assert REPLICA status on Complete, so this is keyed
+ // on trusted replication, matching completeMultipartOpts). The object layer
+ // re-orders the Object Lock it carries against the destination version read
+ // under the write lock, but only for an SSE-C upload -- the scope this issue
+ // enables -- so a marker-only non-SSE-C completion keeps ordinary write
+ // semantics (issue #120).
+ opts.ReplicaLockReconcile = trustedReplication
// First, we compute the ETag of the multipart object.
// The ETag of a multi-part object is always:
diff --git a/cmd/replication-ssec-retransmit_test.go b/cmd/replication-ssec-retransmit_test.go
new file mode 100644
index 000000000..c845b4668
--- /dev/null
+++ b/cmd/replication-ssec-retransmit_test.go
@@ -0,0 +1,1944 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+// Copyright (c) 2026 PGSTY
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+package cmd
+
+import (
+ "bytes"
+ "context"
+ "crypto/md5"
+ "encoding/base64"
+ "encoding/xml"
+ "io"
+ "maps"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "slices"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/minio/minio/internal/auth"
+ "github.com/minio/minio/internal/crypto"
+ xhttp "github.com/minio/minio/internal/http"
+ "github.com/minio/sio"
+)
+
+// TestAPISSECReplicationTargetHead pins what the replication sender's target
+// HEAD sees for an SSE-C object, which is what replicateAll's dispatch relies
+// on: a keyless HEAD answers 400 InvalidRequest (so the sender must retransmit
+// rather than compare), a missing key still answers 404 NoSuchKey (so a missing
+// replica keeps healing), a HEAD carrying the internal replication marker
+// answers with the replica metadata (what the resync accounting HEAD now
+// sends), and the metadata-only CopyObject the sender used to fall into fails
+// with ExcessData on any non-empty object. See pgsty/silo#120.
+func TestAPISSECReplicationTargetHead(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPISSECReplicationTargetHead,
+ makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+ })
+}
+
+func testAPISSECReplicationTargetHead(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ previousTLS := globalIsTLS
+ globalIsTLS = true
+ defer func() { globalIsTLS = previousTLS }()
+
+ key := bytes.Repeat([]byte{0x42}, 32)
+ keyMD5 := md5.Sum(key)
+ data := bytes.Repeat([]byte("ssec-keyless-head-"), 512)
+ sseHeaders := map[string]string{
+ xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+ xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
+ xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
+ }
+ object := "ssec-keyless-head/replica"
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders)
+
+ // The replication target credential holds the standard replication actions.
+ replicator := newObjectAttributesAuthzUser(t, instanceType, bucketName,
+ `"s3:GetObject","s3:PutObject","s3:ReplicateObject","s3:ReplicateDelete","s3:ReplicateTags"`)
+
+ // Exactly the header set replicateAll's StatObject sends today.
+ senderHeaders := map[string]string{
+ "X-Minio-Source-Proxy-Request": "false",
+ xhttp.AmzTagDirective: "ACCESS",
+ }
+ // The same request with the internal replication marker added.
+ markedHeaders := map[string]string{
+ "X-Minio-Source-Proxy-Request": "false",
+ xhttp.AmzTagDirective: "ACCESS",
+ xhttp.MinIOSourceReplicationRequest: "true",
+ }
+
+ head := func(t *testing.T, creds auth.Credentials, obj, versionID string, headers map[string]string) *httptest.ResponseRecorder {
+ t.Helper()
+ headURL := getGetObjectURL("", bucketName, obj)
+ if versionID != "" {
+ // replicateAll addresses the source version (minio-go
+ // api-get-options.go toQueryValues).
+ headURL += "?versionId=" + versionID
+ }
+ req, err := newTestSignedRequestV4(http.MethodHead, headURL, 0, nil,
+ creds.AccessKey, creds.SecretKey, headers)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ return rec
+ }
+
+ sseDesc := errorCodes[ErrSSEEncryptedObject].Description
+
+ baseInfo, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ sourceVersion := baseInfo.VersionID
+
+ t.Run("sender-head-today-is-rejected", func(t *testing.T) {
+ rec := head(t, replicator, object, sourceVersion, senderHeaders)
+ if rec.Code != http.StatusBadRequest {
+ t.Fatalf("%s: keyless HEAD status %d, want 400", instanceType, rec.Code)
+ }
+ if got := rec.Header().Get("x-minio-error-code"); got != "InvalidRequest" {
+ t.Fatalf("%s: error code %q, want InvalidRequest", instanceType, got)
+ }
+ desc := strings.Trim(rec.Header().Get("x-minio-error-desc"), `"`)
+ if !strings.Contains(desc, sseDesc) {
+ t.Fatalf("%s: error desc %q does not carry %q", instanceType, desc, sseDesc)
+ }
+ t.Logf("%s: keyless HEAD -> %d %s / %s", instanceType, rec.Code,
+ rec.Header().Get("x-minio-error-code"), desc)
+ })
+
+ t.Run("missing-object-head-is-distinguishable", func(t *testing.T) {
+ rec := head(t, replicator, "ssec-keyless-head/absent", "", senderHeaders)
+ if rec.Code != http.StatusNotFound {
+ t.Fatalf("%s: missing-object HEAD status %d, want 404", instanceType, rec.Code)
+ }
+ if got := rec.Header().Get("x-minio-error-code"); got != "NoSuchKey" {
+ t.Fatalf("%s: missing-object error code %q, want NoSuchKey", instanceType, got)
+ }
+ })
+
+ t.Run("marked-head-answers-with-metadata", func(t *testing.T) {
+ rec := head(t, replicator, object, sourceVersion, markedHeaders)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: marked keyless HEAD status %d, want 200: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ // setObjectHeaders assigns the ETag through the raw header map with the
+ // non canonical key "ETag", so read it the same way.
+ etag := ""
+ if v := rec.Header()[xhttp.ETag]; len(v) > 0 {
+ etag = strings.Trim(v[0], `"`)
+ }
+ clen := rec.Header().Get(xhttp.ContentLength)
+ lastMod := rec.Header().Get(xhttp.LastModified)
+ if etag == "" || clen == "" || lastMod == "" {
+ t.Fatalf("%s: marked HEAD lacks comparison metadata etag=%q len=%q mtime=%q",
+ instanceType, etag, clen, lastMod)
+ }
+
+ // What replicateAll would compare this against: the source ObjectInfo it
+ // obtained from GetObjectNInfo(..., ReplicationRequest: true).
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ gr.Close()
+ srcSize, err := srcInfo.GetActualSize()
+ if err != nil {
+ t.Fatal(err)
+ }
+ headSize, err := strconv.ParseInt(clen, 10, 64)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Logf("%s: source ETag=%q (len %d) size=%d ; target HEAD ETag=%q (len %d) size=%d",
+ instanceType, srcInfo.ETag, len(srcInfo.ETag), srcSize, etag, len(etag), headSize)
+ if headSize != srcSize {
+ t.Errorf("%s: getReplicationAction size mismatch: source %d target %d", instanceType, srcSize, headSize)
+ }
+ if srcInfo.ETag != etag {
+ t.Logf("%s: NOTE getReplicationAction would see an ETag mismatch (source keeps the sealed ETag, "+
+ "the target HEAD returns the last 32 bytes) and therefore return replicateAll", instanceType)
+ }
+ })
+
+ t.Run("zero-byte-metadata-copy-succeeds", func(t *testing.T) {
+ // A zero-byte SSE-C object has nothing for the plaintext-sized reader to
+ // overrun, so the same copy request succeeds. The ExcessData failure is a
+ // property of non-empty objects.
+ zero := "ssec-keyless-head/zero"
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, zero, nil, sseHeaders)
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, zero, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ zi := gr.ObjInfo
+ gr.Close()
+
+ copySrc := url.QueryEscape(SlashSeparator+bucketName+SlashSeparator+zero) + "?versionId=" + zi.VersionID
+ headers := map[string]string{
+ xhttp.AmzCopySource: copySrc,
+ xhttp.MinIOSourceReplicationRequest: "true",
+ }
+ for k, v := range getCopyObjMetadata(zi, "") {
+ if strings.EqualFold(k, "content-length") {
+ continue
+ }
+ headers[k] = v
+ }
+ headers[xhttp.AmzObjectTagging] = "keyless-head=zero"
+ req, err := newTestSignedRequestV4(http.MethodPut,
+ getCopyObjectURL("", bucketName, zero)+"?versionId="+zi.VersionID, 0, nil,
+ replicator.AccessKey, replicator.SecretKey, headers)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ after, gerr := obj.GetObjectInfo(t.Context(), bucketName, zero, ObjectOptions{})
+ if gerr != nil {
+ t.Fatal(gerr)
+ }
+ t.Logf("%s: zero-byte metadata CopyObject -> %d; stored SSE-C=%v size=%d tags=%q",
+ instanceType, rec.Code, crypto.SSEC.IsEncrypted(after.UserDefined), after.Size, after.UserTags)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: zero-byte metadata CopyObject status %d: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ })
+
+ t.Run("metadata-copy-is-what-resync-runs", func(t *testing.T) {
+ // Exactly what replicateAll runs at cmd/bucket-replication.go:1582 once
+ // the keyless HEAD has been misread: a same bucket, same key CopyObject
+ // built from getCopyObjMetadata plus the replication marker, carrying no
+ // customer key. getCopyObjMetadata already sets REPLICA status and
+ // x-amz-tagging-directive: REPLACE, so the request is replica trusted.
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ gr.Close()
+
+ // minio-go addresses the source version on the copy source and in the
+ // destination query (api-compose-object.go:307,314).
+ copySrc := url.QueryEscape(SlashSeparator+bucketName+SlashSeparator+object) + "?versionId=" + srcInfo.VersionID
+ headers := map[string]string{
+ xhttp.AmzCopySource: copySrc,
+ xhttp.MinIOSourceReplicationRequest: "true",
+ }
+ copyMeta := getCopyObjMetadata(srcInfo, "")
+ for k, v := range copyMeta {
+ // net/http derives the request body length from a literal
+ // Content-Length header; minio-go relies on req.ContentLength, so
+ // drop it here to keep the in-process request faithful.
+ if strings.EqualFold(k, "content-length") {
+ t.Logf("%s: dropping content-length=%q from the copy metadata", instanceType, v)
+ continue
+ }
+ headers[k] = v
+ }
+ t.Logf("%s: copy metadata keys: %v", instanceType, slices.Sorted(maps.Keys(copyMeta)))
+ copyURL := getCopyObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID
+ req, err := newTestSignedRequestV4(http.MethodPut, copyURL, 0, nil,
+ replicator.AccessKey, replicator.SecretKey, headers)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ t.Logf("%s: replication metadata CopyObject (version %s) -> %d %s", instanceType, srcInfo.VersionID, rec.Code,
+ strings.ReplaceAll(rec.Body.String(), "\n", " "))
+ if rec.Code != http.StatusBadRequest {
+ t.Fatalf("%s: metadata CopyObject status %d, want 400", instanceType, rec.Code)
+ }
+ if !strings.Contains(rec.Body.String(), "ExcessData") {
+ t.Fatalf("%s: metadata CopyObject did not fail with ExcessData: %s", instanceType, rec.Body.String())
+ }
+
+ // Whatever the status, the object must still read back with the key.
+ greq, gerr := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object), 0, nil,
+ credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if gerr != nil {
+ t.Fatal(gerr)
+ }
+ grec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(grec, greq)
+ if grec.Code != http.StatusOK || !bytes.Equal(grec.Body.Bytes(), data) {
+ t.Errorf("%s: after the replication metadata CopyObject the object no longer reads back: %d (%d bytes)",
+ instanceType, grec.Code, grec.Body.Len())
+ } else {
+ t.Logf("%s: object still reads back correctly with the customer key", instanceType)
+ }
+ })
+}
+
+// TestAPISSECReplicaRetransmitOverExistingVersion asserts that a full
+// retransmit of an SSE-C object reaches the destination when the replica
+// already exists with the source version and ETag. checkPreconditionsPUT used
+// to reject a matching PreserveETag plus VersionID with 412 for the multipart
+// path (the single-part sealed ETag is truncated before the comparison), and
+// the sender turns 412 into success, so no part was ever sent and an SSE-C
+// replica could never be repaired or updated. See pgsty/silo#120.
+func TestAPISSECReplicaRetransmitOverExistingVersion(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPISSECReplicaRetransmitOverExistingVersion,
+ makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+ })
+}
+
+func testAPISSECReplicaRetransmitOverExistingVersion(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ previousTLS := globalIsTLS
+ globalIsTLS = true
+ defer func() { globalIsTLS = previousTLS }()
+
+ key := bytes.Repeat([]byte{0x43}, 32)
+ keyMD5 := md5.Sum(key)
+ sseHeaders := map[string]string{
+ xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+ xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
+ xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
+ }
+ replicator := newObjectAttributesAuthzUser(t, instanceType, bucketName,
+ `"s3:GetObject","s3:PutObject","s3:ReplicateObject"`)
+
+ replicaHeaders := func(t *testing.T, oi ObjectInfo) map[string]string {
+ t.Helper()
+ opts, _, err := putReplicationOpts(t.Context(), "", oi)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opts.Internal.SourceMTime = time.Time{}
+ out := make(map[string]string)
+ for name, values := range opts.Header() {
+ if len(values) > 0 {
+ out[name] = values[0]
+ }
+ }
+ out[xhttp.MinIOSourceReplicationRequest] = "true"
+ out[xhttp.AmzBucketReplicationStatus] = "REPLICA"
+ out[xhttp.MinIOSourceETag] = oi.ETag
+ return out
+ }
+
+ t.Run("single-part-replica-put", func(t *testing.T) {
+ data := bytes.Repeat([]byte("single-part-ssec-replica-"), 400)
+ object := "ssec-duplicate/single"
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders)
+
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, err := io.ReadAll(gr)
+ gr.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The source changed a tag after the replica was written: the
+ // retransmit must carry it onto the same version.
+ srcInfo.UserTags = "retransmit=single"
+ hdrs := replicaHeaders(t, srcInfo)
+ putURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID
+ req, err := newTestSignedRequestV4(http.MethodPut, putURL, int64(len(cipher)),
+ bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: single-part replica PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=single", data, sseHeaders)
+ })
+
+ t.Run("zero-byte-replica-put", func(t *testing.T) {
+ object := "ssec-duplicate/zero"
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, nil, sseHeaders)
+
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, err := io.ReadAll(gr)
+ gr.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo.UserTags = "retransmit=zero"
+ hdrs := replicaHeaders(t, srcInfo)
+ putURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID
+ req, err := newTestSignedRequestV4(http.MethodPut, putURL, int64(len(cipher)),
+ bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: zero-byte replica PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=zero", nil, sseHeaders)
+ })
+
+ t.Run("multipart-replica-newmpu", func(t *testing.T) {
+ data := bytes.Repeat([]byte("multipart-ssec-replica-"), 4096)
+ object := "ssec-duplicate/multipart"
+
+ newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object),
+ 0, nil, credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ newRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(newRec, newReq)
+ if newRec.Code != http.StatusOK {
+ t.Fatalf("%s: source NewMultipart %d: %s", instanceType, newRec.Code, newRec.Body.String())
+ }
+ var srcInit InitiateMultipartUploadResponse
+ if err = xmlDecoder(newRec.Body, &srcInit, int64(newRec.Body.Len())); err != nil {
+ t.Fatal(err)
+ }
+ partReq, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectPartURL("", bucketName, object, srcInit.UploadID, "1"), int64(len(data)),
+ bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ partRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(partRec, partReq)
+ if partRec.Code != http.StatusOK {
+ t.Fatalf("%s: source PutPart %d: %s", instanceType, partRec.Code, partRec.Body.String())
+ }
+ body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{
+ {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])},
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeReq, err := newTestSignedRequestV4(http.MethodPost,
+ getCompleteMultipartUploadURL("", bucketName, object, srcInit.UploadID), int64(len(body)),
+ bytes.NewReader(body), credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(completeRec, completeReq)
+ if completeRec.Code != http.StatusOK {
+ t.Fatalf("%s: source Complete %d: %s", instanceType, completeRec.Code, completeRec.Body.String())
+ }
+
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ gr.Close()
+
+ srcInfo.UserTags = "retransmit=multipart"
+ hdrs := replicaHeaders(t, srcInfo)
+ mpuURL := getNewMultipartURL("", bucketName, object) + "&versionId=" + srcInfo.VersionID
+ req, err := newTestSignedRequestV4(http.MethodPost, mpuURL, 0, nil,
+ replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ t.Logf("%s: multipart replica NewMultipartUpload over the same version+ETag -> %d %s (source ETag %q, multipart=%v)",
+ instanceType, rec.Code, strings.ReplaceAll(rec.Body.String(), "\n", " "),
+ srcInfo.ETag, crypto.IsMultiPart(srcInfo.UserDefined))
+ if rec.Code == http.StatusPreconditionFailed {
+ t.Fatalf("%s: multipart replica upload short-circuited with 412; the sender turns that into "+
+ "success, so no part is ever sent", instanceType)
+ }
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: replica NewMultipart status %d: %s", instanceType, rec.Code, rec.Body.String())
+ }
+
+ // Finish the replica upload the way the sender does and prove the object
+ // is still readable with the customer key afterwards.
+ gr2, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ rawPart, err := io.ReadAll(gr2)
+ gr2.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ var replicaInit InitiateMultipartUploadResponse
+ if err = xmlDecoder(rec.Body, &replicaInit, int64(rec.Body.Len())); err != nil {
+ t.Fatal(err)
+ }
+ partReq2, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectPartURL("", bucketName, object, replicaInit.UploadID, "1"), int64(len(rawPart)),
+ bytes.NewReader(rawPart), replicator.AccessKey, replicator.SecretKey,
+ map[string]string{xhttp.MinIOSourceReplicationRequest: "true"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ partRec2 := httptest.NewRecorder()
+ apiRouter.ServeHTTP(partRec2, partReq2)
+ if partRec2.Code != http.StatusOK {
+ t.Fatalf("%s: replica PutPart %d: %s", instanceType, partRec2.Code, partRec2.Body.String())
+ }
+ actualSize, err := srcInfo.GetActualSize()
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeBody2, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{
+ {PartNumber: 1, ETag: canonicalizeETag(partRec2.Header()[xhttp.ETag][0])},
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeReq2, err := newTestSignedRequestV4(http.MethodPost,
+ getCompleteMultipartUploadURL("", bucketName, object, replicaInit.UploadID), int64(len(completeBody2)),
+ bytes.NewReader(completeBody2), replicator.AccessKey, replicator.SecretKey, map[string]string{
+ xhttp.MinIOSourceReplicationRequest: "true",
+ xhttp.MinIOSourceMTime: srcInfo.ModTime.Format(time.RFC3339Nano),
+ xhttp.MinIOSourceETag: srcInfo.ETag,
+ xhttp.MinIOReplicationActualObjectSize: strconv.FormatInt(actualSize, 10),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeRec2 := httptest.NewRecorder()
+ apiRouter.ServeHTTP(completeRec2, completeReq2)
+ if completeRec2.Code != http.StatusOK {
+ t.Fatalf("%s: replica Complete %d: %s", instanceType, completeRec2.Code, completeRec2.Body.String())
+ }
+ assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=multipart", data, sseHeaders)
+ })
+
+ t.Run("repairs-an-undecodable-existing-version", func(t *testing.T) {
+ // A pre-fix destination (issue #109) could persist an SSE-C replica as
+ // compress(ciphertext) or a re-encrypted body, leaving a stored length
+ // that is not a valid encryption stream. Resync repairs such a version by
+ // retransmitting the source's raw ciphertext, but the write must not first
+ // require the stored, damaged object to decrypt. See issue #120.
+ data := bytes.Repeat([]byte("SILO raw SSE-C recovery\n"), 400)
+ object := "ssec-duplicate/undecodable"
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders)
+
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, err := io.ReadAll(gr)
+ gr.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // Stage the damage: overwrite the version with a body too short to be a
+ // valid encryption stream, standing in for the compression/re-encryption
+ // an old destination left behind. The staging write itself is a raw
+ // replica over the still-valid version, so it stores verbatim.
+ damaged := []byte("dmg!!")
+ if _, derr := sio.DecryptedSize(uint64(len(damaged))); derr == nil {
+ t.Fatalf("%s: fixture body of %d bytes is a valid stream length, not undecodable", instanceType, len(damaged))
+ }
+ stageHdrs := replicaHeaders(t, srcInfo)
+ stageURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID
+ stageReq, err := newTestSignedRequestV4(http.MethodPut, stageURL, int64(len(damaged)),
+ bytes.NewReader(damaged), replicator.AccessKey, replicator.SecretKey, stageHdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ stageRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(stageRec, stageReq)
+ if stageRec.Code != http.StatusOK {
+ t.Fatalf("%s: could not stage the damaged replica: %d %s", instanceType, stageRec.Code, stageRec.Body.String())
+ }
+ // The staged version is genuinely undecodable at the object layer, which
+ // is exactly what makes DecryptObjectInfo fail during the overwrite.
+ staged, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{VersionID: srcInfo.VersionID})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, derr := staged.DecryptedSize(); derr == nil {
+ t.Fatalf("%s: staged replica is decodable, cannot exercise the repair path", instanceType)
+ }
+
+ // Retransmit the correct ciphertext over the same version. Before the
+ // raw-replica precondition exemption this failed with XMinioObjectTampered
+ // because the damaged object could not decrypt; it must now repair.
+ srcInfo.UserTags = "retransmit=undecodable"
+ hdrs := replicaHeaders(t, srcInfo)
+ putURL := getPutObjectURL("", bucketName, object) + "?versionId=" + srcInfo.VersionID
+ req, err := newTestSignedRequestV4(http.MethodPut, putURL, int64(len(cipher)),
+ bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: retransmit over an undecodable version status %d, want 200: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ assertRetransmittedVersion(t, obj, apiRouter, credentials, bucketName, object, srcInfo.VersionID, "retransmit=undecodable", data, sseHeaders)
+ })
+
+ // setupHealthySSECVersion writes a normal SSE-C object and returns its
+ // ObjectInfo (for building replica headers), its ciphertext, and the
+ // client-visible ETag a keyed reader sees -- the decrypted ETag, which is
+ // distinct from the stored sealed ETag.
+ setupHealthySSECVersion := func(t *testing.T, object string, data []byte) (srcInfo ObjectInfo, cipher []byte, clientETag string) {
+ t.Helper()
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders)
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{ReplicationRequest: true})
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo = gr.ObjInfo
+ cipher, err = io.ReadAll(gr)
+ gr.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The client-visible ETag is the decrypted one, which the handler derives
+ // with the customer key through DecryptObjectInfo; compute it the same way.
+ keyHeader := http.Header{}
+ for k, v := range sseHeaders {
+ keyHeader.Set(k, v)
+ }
+ clientETag = getDecryptedETag(keyHeader, srcInfo, false)
+ if clientETag == "" || clientETag == srcInfo.ETag {
+ t.Fatalf("%s: client ETag %q is not distinct from the sealed ETag %q", instanceType, clientETag, srcInfo.ETag)
+ }
+ return srcInfo, cipher, clientETag
+ }
+
+ // A conditional replica PUT must compare the public precondition against the
+ // client-visible ETag, not the stored sealed one. Skipping DecryptObjectInfo
+ // for every raw SSE-C replica (not only a pure overwrite) left oi.ETag sealed
+ // and inverted both conditions.
+ t.Run("if-match-on-the-client-etag-proceeds", func(t *testing.T) {
+ object := "ssec-duplicate/cond-if-match"
+ srcInfo, cipher, clientETag := setupHealthySSECVersion(t, object, bytes.Repeat([]byte("cond-if-match-"), 64))
+ hdrs := replicaHeaders(t, srcInfo)
+ hdrs[xhttp.IfMatch] = "\"" + clientETag + "\""
+ req, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectURL("", bucketName, object)+"?versionId="+srcInfo.VersionID,
+ int64(len(cipher)), bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: If-Match on the client ETag status %d, want 200: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ })
+
+ t.Run("if-none-match-on-the-client-etag-fails", func(t *testing.T) {
+ object := "ssec-duplicate/cond-if-none-match"
+ srcInfo, cipher, clientETag := setupHealthySSECVersion(t, object, bytes.Repeat([]byte("cond-if-none-"), 64))
+ hdrs := replicaHeaders(t, srcInfo)
+ hdrs[xhttp.IfNoneMatch] = "\"" + clientETag + "\""
+ req, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectURL("", bucketName, object)+"?versionId="+srcInfo.VersionID,
+ int64(len(cipher)), bytes.NewReader(cipher), replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusPreconditionFailed {
+ t.Fatalf("%s: If-None-Match on the client ETag status %d, want 412: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ })
+}
+
+// assertRetransmittedVersion checks that a retransmit landed on the addressed
+// version: the changed tag is stored on it and it still reads back with the
+// customer key.
+func assertRetransmittedVersion(t *testing.T, obj ObjectLayer, apiRouter http.Handler, credentials auth.Credentials,
+ bucketName, object, versionID, wantTags string, want []byte, sseHeaders map[string]string,
+) {
+ t.Helper()
+ info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{VersionID: versionID})
+ if err != nil {
+ t.Fatalf("version %s after retransmit: %v", versionID, err)
+ }
+ if info.UserTags != wantTags {
+ t.Errorf("version %s tags after retransmit %q, want %q", versionID, info.UserTags, wantTags)
+ }
+ if !crypto.SSEC.IsEncrypted(info.UserDefined) {
+ t.Errorf("version %s lost its SSE-C seal after retransmit", versionID)
+ }
+ getReq, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object)+"?versionId="+versionID,
+ 0, nil, credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ getRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(getRec, getReq)
+ if getRec.Code != http.StatusOK || !bytes.Equal(getRec.Body.Bytes(), want) {
+ t.Fatalf("version %s does not read back with the customer key after retransmit: %d (%d bytes, want %d)",
+ versionID, getRec.Code, getRec.Body.Len(), len(want))
+ }
+}
+
+// TestPutReplicationOptsRetentionRemoval asserts that a source version whose
+// retention was removed (stored as an empty mode and date) still builds
+// replication options, carrying the removal's ordering timestamp and no value.
+func TestPutReplicationOptsRetentionRemoval(t *testing.T) {
+ removedAt := time.Date(2026, 9, 5, 10, 0, 0, 0, time.UTC)
+ oi := ObjectInfo{
+ Bucket: "b", Name: "o", VersionID: "v1", ModTime: removedAt.Add(-time.Hour),
+ UserDefined: map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockMode): "",
+ strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): "",
+ ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: removedAt.Format(time.RFC3339Nano),
+ },
+ }
+ opts, _, err := putReplicationOpts(t.Context(), "", oi)
+ if err != nil {
+ t.Fatalf("putReplicationOpts on a removed retention: %v", err)
+ }
+ if opts.Mode != "" || !opts.RetainUntilDate.IsZero() {
+ t.Errorf("removal sent as a retention: mode %q date %v", opts.Mode, opts.RetainUntilDate)
+ }
+ if !opts.Internal.RetentionTimestamp.Equal(removedAt) {
+ t.Errorf("removal timestamp %v, want %v", opts.Internal.RetentionTimestamp, removedAt)
+ }
+ if hdr := opts.Header(); hdr.Get(xhttp.AmzObjectLockMode) != "" || hdr.Get(xhttp.AmzObjectLockRetainUntilDate) != "" ||
+ hdr.Get(xhttp.MinIOSourceObjectRetentionTimestamp) == "" {
+ t.Errorf("removal headers %v: want no lock value and a retention timestamp", hdr)
+ }
+}
+
+// TestAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite pins the predicate
+// of the duplicate version and ETag exemption: it applies to an authenticated
+// replica write that carries an SSE-C seal, whatever the destination holds, and
+// not to a plaintext replica write that happens to match an SSE-C destination
+// version. See pgsty/silo#120.
+func TestAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite,
+ makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+ })
+}
+
+func testAPISSECReplicaWriteExemptionIsKeyedOnTheIncomingWrite(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ previousTLS := globalIsTLS
+ globalIsTLS = true
+ defer func() { globalIsTLS = previousTLS }()
+
+ key := bytes.Repeat([]byte{0x44}, 32)
+ keyMD5 := md5.Sum(key)
+ sseHeaders := map[string]string{
+ xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+ xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
+ xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
+ }
+ replicator := newObjectAttributesAuthzUser(t, instanceType, bucketName,
+ `"s3:GetObject","s3:PutObject","s3:ReplicateObject"`)
+
+ rawOf := func(t *testing.T, object string) (ObjectInfo, []byte) {
+ t.Helper()
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer gr.Close()
+ raw, err := io.ReadAll(gr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return gr.ObjInfo, raw
+ }
+ replicaHeaders := func(t *testing.T, oi ObjectInfo) map[string]string {
+ t.Helper()
+ opts, _, err := putReplicationOpts(t.Context(), "", oi)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opts.Internal.SourceMTime = time.Time{}
+ out := make(map[string]string)
+ for name, values := range opts.Header() {
+ if len(values) > 0 {
+ out[name] = values[0]
+ }
+ }
+ out[xhttp.MinIOSourceReplicationRequest] = "true"
+ out[xhttp.AmzBucketReplicationStatus] = "REPLICA"
+ out[xhttp.MinIOSourceETag] = oi.ETag
+ return out
+ }
+ replicaPut := func(t *testing.T, object, versionID string, body []byte, hdrs map[string]string) *httptest.ResponseRecorder {
+ t.Helper()
+ req, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(body)),
+ bytes.NewReader(body), replicator.AccessKey, replicator.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ return rec
+ }
+
+ // putSSECMultipart stores a one-part SSE-C multipart object: unlike a
+ // single-part SSE-C object, whose stored ETag is the sealed one and never
+ // matches the sender's, a multipart ETag compares equal, which is what
+ // makes the duplicate version and ETag check reachable at all.
+ putSSECMultipart := func(t *testing.T, object string, data []byte) {
+ t.Helper()
+ newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object),
+ 0, nil, credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ newRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(newRec, newReq)
+ if newRec.Code != http.StatusOK {
+ t.Fatalf("%s: NewMultipart %d: %s", instanceType, newRec.Code, newRec.Body.String())
+ }
+ var init InitiateMultipartUploadResponse
+ if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil {
+ t.Fatal(err)
+ }
+ partReq, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(data)),
+ bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ partRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(partRec, partReq)
+ if partRec.Code != http.StatusOK {
+ t.Fatalf("%s: PutPart %d: %s", instanceType, partRec.Code, partRec.Body.String())
+ }
+ body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{
+ {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])},
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeReq, err := newTestSignedRequestV4(http.MethodPost,
+ getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(body)),
+ bytes.NewReader(body), credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(completeRec, completeReq)
+ if completeRec.Code != http.StatusOK {
+ t.Fatalf("%s: Complete %d: %s", instanceType, completeRec.Code, completeRec.Body.String())
+ }
+ }
+
+ t.Run("plaintext-replica-over-ssec-version-is-still-rejected", func(t *testing.T) {
+ object := "ssec-exemption/ssec-destination"
+ putSSECMultipart(t, object, bytes.Repeat([]byte("ssec-destination-"), 4096))
+ srcInfo, raw := rawOf(t, object)
+ hdrs := replicaHeaders(t, srcInfo)
+ // Without the seal the incoming write is a plaintext replica that merely
+ // carries the stored version and ETag: the duplicate check still applies.
+ for name := range hdrs {
+ if strings.HasPrefix(name, "X-Minio-Replication-Server-Side-Encryption-") {
+ delete(hdrs, name)
+ }
+ }
+ rec := replicaPut(t, object, srcInfo.VersionID, raw, hdrs)
+ if rec.Code != http.StatusPreconditionFailed {
+ t.Fatalf("%s: plaintext replica write over an existing SSE-C version answered %d, want 412: %s",
+ instanceType, rec.Code, rec.Body.String())
+ }
+ })
+
+ t.Run("ssec-replica-over-plaintext-version-is-exempted", func(t *testing.T) {
+ object := "ssec-exemption/plain-destination"
+ // A plaintext version first, then an SSE-C version of the same key so
+ // the seal is bound to this object path.
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, []byte("plaintext destination version"), nil)
+ plainInfo, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ data := bytes.Repeat([]byte("ssec-source-"), 400)
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, sseHeaders)
+ srcInfo, raw := rawOf(t, object)
+
+ // The incoming write carries the SSE-C seal and addresses the plaintext
+ // version with its ETag: the exemption is decided by the incoming
+ // write, not by what the destination holds.
+ hdrs := replicaHeaders(t, srcInfo)
+ hdrs[xhttp.MinIOSourceETag] = plainInfo.ETag
+ rec := replicaPut(t, object, plainInfo.VersionID, raw, hdrs)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: SSE-C replica write over a matching plaintext version answered %d, want 200: %s",
+ instanceType, rec.Code, rec.Body.String())
+ }
+ getReq, err := newTestSignedRequestV4(http.MethodGet,
+ getGetObjectURL("", bucketName, object)+"?versionId="+plainInfo.VersionID, 0, nil,
+ credentials.AccessKey, credentials.SecretKey, sseHeaders)
+ if err != nil {
+ t.Fatal(err)
+ }
+ getRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(getRec, getReq)
+ if getRec.Code != http.StatusOK || !bytes.Equal(getRec.Body.Bytes(), data) {
+ t.Fatalf("%s: the retransmitted version does not read back with the customer key: %d (%d bytes)",
+ instanceType, getRec.Code, getRec.Body.Len())
+ }
+ })
+}
+
+// Retain-until values in the millisecond form ISO8601Format round-trips to, so
+// a value applied through the handler reads back byte-for-byte.
+const (
+ retransmitRetainUntilNewer = "2031-01-01T00:00:00.000Z"
+ retransmitRetainUntilStale = "2028-01-01T00:00:00.000Z"
+)
+
+// TestAPISSECReplicaRetransmitObjectLockOrdering proves that the full SSE-C
+// replica retransmit orders the Object Lock update it carries against the state
+// already stored on the addressed version, the same way the metadata CopyObject
+// path does. Before issue #120 routed these writes through PutObjectHandler the
+// handler applied the incoming retention and legal hold directly and never
+// persisted the ordering timestamps, so a retransmit carrying an older value
+// could overwrite a destination version's newer one. See pgsty/silo#120.
+func TestAPISSECReplicaRetransmitObjectLockOrdering(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPISSECReplicaRetransmitObjectLockOrdering,
+ makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ })
+}
+
+func testAPISSECReplicaRetransmitObjectLockOrdering(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ previousTLS := globalIsTLS
+ globalIsTLS = true
+ defer func() { globalIsTLS = previousTLS }()
+
+ key := bytes.Repeat([]byte{0x45}, 32)
+ keyMD5 := md5.Sum(key)
+ sseHeaders := map[string]string{
+ xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+ xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
+ xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
+ }
+
+ // seed writes a fresh SSE-C source version with no Object Lock state and
+ // returns its version id.
+ seed := func(t *testing.T, object string) string {
+ t.Helper()
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object,
+ bytes.Repeat([]byte("ssec-lock-ordering-"), 64), sseHeaders)
+ info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return info.VersionID
+ }
+
+ // retransmit sends the full SSE-C replica retransmit the sender emits for the
+ // addressed version, over its raw ciphertext, carrying exactly the given
+ // Object Lock headers on top of the replica seal. The credential is the admin
+ // user so the retention and legal-hold permission checks pass; the request is
+ // still a trusted replica because it carries the marker and REPLICA status.
+ retransmit := func(t *testing.T, object, versionID string, lockHeaders map[string]string) {
+ t.Helper()
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true,
+ VersionID: versionID,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, err := io.ReadAll(gr)
+ gr.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ opts, _, err := putReplicationOpts(t.Context(), "", srcInfo)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opts.Internal.SourceMTime = time.Time{}
+ hdrs := make(map[string]string)
+ for name, values := range opts.Header() {
+ if len(values) > 0 {
+ hdrs[name] = values[0]
+ }
+ }
+ hdrs[xhttp.MinIOSourceReplicationRequest] = "true"
+ hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA"
+ hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag
+ // The case owns the lock instruction: drop any lock header the sender
+ // derived from the source version.
+ for _, name := range []string{
+ xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold,
+ xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp,
+ } {
+ delete(hdrs, name)
+ }
+ maps.Copy(hdrs, lockHeaders)
+
+ req, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)),
+ bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, hdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: replica retransmit PUT status %d, want 200: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ }
+
+ t.Run("older-legal-hold-off-does-not-clear-newer-on", func(t *testing.T) {
+ object := "ssec-lock-ordering/legal-hold"
+ versionID := seed(t, object)
+
+ // Establish the newer legal hold ON. Its timestamp persistence is proven
+ // by the retention sibling test, so here only require the value took
+ // effect before the older OFF arrives, so the clobber below is what tells
+ // a fixed handler from a broken one.
+ retransmit(t, object, versionID, map[string]string{
+ xhttp.AmzObjectLockLegalHold: "ON",
+ xhttp.MinIOSourceObjectLegalHoldTimestamp: objectLockTestStamp1000,
+ })
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" {
+ t.Fatalf("%s: seeding legal hold ON failed: got %+v", instanceType, got)
+ }
+
+ // A retransmit carrying an older legal-hold OFF must not clear it.
+ retransmit(t, object, versionID, map[string]string{
+ xhttp.AmzObjectLockLegalHold: "OFF",
+ xhttp.MinIOSourceObjectLegalHoldTimestamp: objectLockTestStamp0900,
+ })
+ want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000}
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: an older legal-hold OFF cleared the newer ON: got %+v, want %+v", instanceType, got, want)
+ }
+ })
+
+ t.Run("newer-retention-applies-and-persists-its-timestamp", func(t *testing.T) {
+ object := "ssec-lock-ordering/retention-applies"
+ versionID := seed(t, object)
+
+ retransmit(t, object, versionID, map[string]string{
+ xhttp.AmzObjectLockMode: "GOVERNANCE",
+ xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilNewer,
+ xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp1000,
+ })
+ // The value applies and, crucially, the ordering timestamp is persisted so
+ // a later stale update can be recognized as older.
+ want := objectLockFields{
+ mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000,
+ }
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: newer retention did not apply and persist its timestamp: got %+v, want %+v", instanceType, got, want)
+ }
+ })
+
+ t.Run("stale-retention-update-is-ignored", func(t *testing.T) {
+ object := "ssec-lock-ordering/retention-stale"
+ versionID := seed(t, object)
+
+ // Establish the newer retention first; its timestamp persistence is proven
+ // by the sibling test above, so here only require the value took effect, so
+ // the stale overwrite below is what tells a fixed handler from a broken one.
+ retransmit(t, object, versionID, map[string]string{
+ xhttp.AmzObjectLockMode: "GOVERNANCE",
+ xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilNewer,
+ xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp1000,
+ })
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.mode != "GOVERNANCE" || got.retainUntil != retransmitRetainUntilNewer {
+ t.Fatalf("%s: seeding the newer retention failed: got %+v", instanceType, got)
+ }
+
+ // A retransmit carrying an older retention with a different date must be
+ // ignored; the newer date and its ordering timestamp survive.
+ retransmit(t, object, versionID, map[string]string{
+ xhttp.AmzObjectLockMode: "GOVERNANCE",
+ xhttp.AmzObjectLockRetainUntilDate: retransmitRetainUntilStale,
+ xhttp.MinIOSourceObjectRetentionTimestamp: objectLockTestStamp0900,
+ })
+ want := objectLockFields{
+ mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000,
+ }
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: a stale retention update overwrote the newer one: got %+v, want %+v", instanceType, got, want)
+ }
+ })
+}
+
+// TestAPISSECReplicaRetransmitMultipartObjectLockOrdering proves the ordering
+// fix also covers the multipart initiation path #120 routes an SSE-C replica
+// through: NewMultipartUploadHandler reads the addressed version's stored lock
+// state and orders the incoming update against it, so a large-object retransmit
+// carrying an older legal-hold OFF cannot clear a newer ON. See pgsty/silo#120.
+func TestAPISSECReplicaRetransmitMultipartObjectLockOrdering(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPISSECReplicaRetransmitMultipartObjectLockOrdering,
+ makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ })
+}
+
+func testAPISSECReplicaRetransmitMultipartObjectLockOrdering(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ previousTLS := globalIsTLS
+ globalIsTLS = true
+ defer func() { globalIsTLS = previousTLS }()
+
+ key := bytes.Repeat([]byte{0x46}, 32)
+ keyMD5 := md5.Sum(key)
+ sseHeaders := map[string]string{
+ xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+ xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
+ xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
+ }
+
+ // replicaSealHeaders returns the sender's replica seal and marker headers for
+ // the addressed version, with every Object Lock header stripped so the case
+ // owns the lock instruction.
+ replicaSealHeaders := func(t *testing.T, srcInfo ObjectInfo) map[string]string {
+ t.Helper()
+ opts, _, err := putReplicationOpts(t.Context(), "", srcInfo)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opts.Internal.SourceMTime = time.Time{}
+ hdrs := make(map[string]string)
+ for name, values := range opts.Header() {
+ if len(values) > 0 {
+ hdrs[name] = values[0]
+ }
+ }
+ hdrs[xhttp.MinIOSourceReplicationRequest] = "true"
+ hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA"
+ hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag
+ for _, name := range []string{
+ xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold,
+ xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp,
+ } {
+ delete(hdrs, name)
+ }
+ return hdrs
+ }
+
+ object := "ssec-lock-ordering/multipart"
+ // A single-part SSE-C source is enough; the retransmit re-uploads its bytes
+ // as one multipart part and commits the addressed version.
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object,
+ bytes.Repeat([]byte("ssec-multipart-lock-ordering-"), 64), sseHeaders)
+ base, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ versionID := base.VersionID
+
+ // Establish the newer legal hold ON through the single-part PUT retransmit.
+ {
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true, VersionID: versionID,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, rerr := io.ReadAll(gr)
+ gr.Close()
+ if rerr != nil {
+ t.Fatal(rerr)
+ }
+ hdrs := replicaSealHeaders(t, srcInfo)
+ hdrs[xhttp.AmzObjectLockLegalHold] = "ON"
+ hdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp1000
+ req, rerr := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)),
+ bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, hdrs)
+ if rerr != nil {
+ t.Fatal(rerr)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: seeding legal hold ON via PUT status %d: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" {
+ t.Fatalf("%s: seeding legal hold ON failed: got %+v", instanceType, got)
+ }
+ }
+
+ // A multipart retransmit carrying an older legal-hold OFF must not clear it.
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true, VersionID: versionID,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, err := io.ReadAll(gr)
+ gr.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ actualSize, err := srcInfo.GetActualSize()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ newHdrs := replicaSealHeaders(t, srcInfo)
+ newHdrs[xhttp.AmzObjectLockLegalHold] = "OFF"
+ newHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0900
+ newReq, err := newTestSignedRequestV4(http.MethodPost,
+ getNewMultipartURL("", bucketName, object)+"&versionId="+versionID, 0, nil,
+ credentials.AccessKey, credentials.SecretKey, newHdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ newRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(newRec, newReq)
+ if newRec.Code != http.StatusOK {
+ t.Fatalf("%s: replica NewMultipartUpload status %d: %s", instanceType, newRec.Code, newRec.Body.String())
+ }
+ var init InitiateMultipartUploadResponse
+ if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil {
+ t.Fatal(err)
+ }
+
+ partReq, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(cipher)),
+ bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey,
+ map[string]string{xhttp.MinIOSourceReplicationRequest: "true"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ partRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(partRec, partReq)
+ if partRec.Code != http.StatusOK {
+ t.Fatalf("%s: replica PutObjectPart status %d: %s", instanceType, partRec.Code, partRec.Body.String())
+ }
+
+ completeBody, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{
+ {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])},
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeReq, err := newTestSignedRequestV4(http.MethodPost,
+ getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(completeBody)),
+ bytes.NewReader(completeBody), credentials.AccessKey, credentials.SecretKey, map[string]string{
+ xhttp.MinIOSourceReplicationRequest: "true",
+ xhttp.MinIOSourceMTime: srcInfo.ModTime.Format(time.RFC3339Nano),
+ xhttp.MinIOSourceETag: srcInfo.ETag,
+ xhttp.MinIOReplicationActualObjectSize: strconv.FormatInt(actualSize, 10),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(completeRec, completeReq)
+ if completeRec.Code != http.StatusOK {
+ t.Fatalf("%s: replica CompleteMultipartUpload status %d: %s", instanceType, completeRec.Code, completeRec.Body.String())
+ }
+
+ want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000}
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: an older legal-hold OFF via multipart cleared the newer ON: got %+v, want %+v", instanceType, got, want)
+ }
+}
+
+// TestReplicaStoredLock verifies how a replica write reads the destination
+// version's stored lock state before ordering its update: a present version
+// yields its state, a missing object or version yields an empty state so a first
+// write is not blocked, and any other read error (a quorum loss, a timeout) is
+// returned so the caller fails the write instead of ordering an incoming update
+// against lock state it merely could not read. See pgsty/silo#120.
+func TestReplicaStoredLock(t *testing.T) {
+ fixed := func(oi ObjectInfo, err error) GetObjectInfoFn {
+ return func(context.Context, string, string, ObjectOptions) (ObjectInfo, error) { return oi, err }
+ }
+ stored := ObjectInfo{UserDefined: map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000,
+ }}
+
+ t.Run("present-version-returns-its-state", func(t *testing.T) {
+ got, err := replicaStoredLock(context.Background(), fixed(stored, nil), "b", "o", "v")
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if got.legalHold != "ON" || got.legalHoldTimestamp != objectLockTestStamp1000 {
+ t.Fatalf("stored lock state = %+v, want legal hold ON stamped %q", got, objectLockTestStamp1000)
+ }
+ })
+
+ for name, notFound := range map[string]error{
+ "object-not-found": ObjectNotFound{Bucket: "b", Object: "o"},
+ "version-not-found": VersionNotFound{Bucket: "b", Object: "o", VersionID: "v"},
+ } {
+ t.Run(name+"-is-empty-state", func(t *testing.T) {
+ got, err := replicaStoredLock(context.Background(), fixed(ObjectInfo{}, notFound), "b", "o", "v")
+ if err != nil {
+ t.Fatalf("a not-found read must not error: %v", err)
+ }
+ if got != (objectLockState{}) {
+ t.Fatalf("a not-found read must yield empty state, got %+v", got)
+ }
+ })
+ }
+
+ t.Run("transient-read-error-propagates", func(t *testing.T) {
+ boom := InsufficientReadQuorum{}
+ got, err := replicaStoredLock(context.Background(), fixed(ObjectInfo{}, boom), "b", "o", "v")
+ if err == nil {
+ t.Fatal("a transient read error must propagate, not be treated as absent lock state")
+ }
+ if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
+ t.Fatalf("transient error misclassified as not-found: %v", err)
+ }
+ if got != (objectLockState{}) {
+ t.Fatalf("on a read error the caller must get empty state and fail the write, got %+v", got)
+ }
+ })
+}
+
+// TestPutReplicationOptsRetentionRemovalTimestampOnly asserts that a version
+// whose retention was removed on the retransmit PUT path -- stored as an
+// ordering timestamp with the value keys absent, not empty -- still builds
+// replication options that carry the removal timestamp, so the next hop can
+// order the removal instead of keeping obsolete retention. See pgsty/silo#120.
+func TestPutReplicationOptsRetentionRemovalTimestampOnly(t *testing.T) {
+ removedAt := time.Date(2026, 9, 5, 10, 0, 0, 0, time.UTC)
+ oi := ObjectInfo{
+ Bucket: "b", Name: "o", VersionID: "v1", ModTime: removedAt.Add(-time.Hour),
+ UserDefined: map[string]string{
+ // Only the reserved ordering timestamp; no mode/date keys at all.
+ ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: removedAt.Format(time.RFC3339Nano),
+ },
+ }
+ opts, _, err := putReplicationOpts(t.Context(), "", oi)
+ if err != nil {
+ t.Fatalf("putReplicationOpts on a timestamp-only removal: %v", err)
+ }
+ if opts.Mode != "" || !opts.RetainUntilDate.IsZero() {
+ t.Errorf("removal sent as a retention: mode %q date %v", opts.Mode, opts.RetainUntilDate)
+ }
+ if !opts.Internal.RetentionTimestamp.Equal(removedAt) {
+ t.Errorf("removal timestamp %v, want %v", opts.Internal.RetentionTimestamp, removedAt)
+ }
+ if hdr := opts.Header(); hdr.Get(xhttp.AmzObjectLockMode) != "" || hdr.Get(xhttp.AmzObjectLockRetainUntilDate) != "" ||
+ hdr.Get(xhttp.MinIOSourceObjectRetentionTimestamp) == "" {
+ t.Errorf("removal headers %v: want no lock value and a retention timestamp", hdr)
+ }
+}
+
+// TestAPIReplicaMarkerOnlyAppliesObjectLock guards the regression the shared
+// ordering helper could introduce: a trusted peer write that carries the
+// internal replication marker but NOT REPLICA status (replicationRequest true,
+// replicaTrusted false) must keep ordinary write semantics and apply its
+// validated Object Lock, not restore an empty stored state. It covers an
+// explicit legal hold and a bucket-default retention, on both the PUT and the
+// multipart-initiation paths. See pgsty/silo#120 (Codex finding 3).
+func TestAPIReplicaMarkerOnlyAppliesObjectLock(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPIReplicaMarkerOnlyAppliesObjectLock,
+ makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ })
+}
+
+func testAPIReplicaMarkerOnlyAppliesObjectLock(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ // A trusted peer credential holding ReplicateObject plus the lock permissions
+ // checkPutObjectLockAllowed enforces even for a marker-only write.
+ peer := newObjectAttributesAuthzUser(t, instanceType, bucketName,
+ `"s3:GetObject","s3:PutObject","s3:ReplicateObject","s3:PutObjectRetention","s3:PutObjectLegalHold"`)
+
+ // Bucket default retention, so a marker-only write with no lock headers still
+ // has a validated retention to apply.
+ lockCfg := []byte(`EnabledGOVERNANCE30`)
+ if _, err := globalBucketMetadataSys.Update(t.Context(), bucketName, objectLockConfig, lockCfg); err != nil {
+ t.Fatalf("%s: configure bucket default retention: %v", instanceType, err)
+ }
+
+ // markerOnly returns the trusted-but-not-REPLICA header set plus extra: the
+ // internal marker with no REPLICA status.
+ markerOnly := func(extra map[string]string) map[string]string {
+ h := map[string]string{xhttp.MinIOSourceReplicationRequest: "true"}
+ maps.Copy(h, extra)
+ return h
+ }
+ data := bytes.Repeat([]byte("marker-only-lock-"), 32)
+
+ markerOnlyMPU := func(t *testing.T, object string, lockHeaders map[string]string) {
+ t.Helper()
+ newReq, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object), 0, nil,
+ peer.AccessKey, peer.SecretKey, markerOnly(lockHeaders))
+ if err != nil {
+ t.Fatal(err)
+ }
+ newRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(newRec, newReq)
+ if newRec.Code != http.StatusOK {
+ t.Fatalf("%s: marker-only NewMultipartUpload %d: %s", instanceType, newRec.Code, newRec.Body.String())
+ }
+ var init InitiateMultipartUploadResponse
+ if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil {
+ t.Fatal(err)
+ }
+ partReq, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(data)),
+ bytes.NewReader(data), peer.AccessKey, peer.SecretKey, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ partRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(partRec, partReq)
+ if partRec.Code != http.StatusOK {
+ t.Fatalf("%s: marker-only PutObjectPart %d: %s", instanceType, partRec.Code, partRec.Body.String())
+ }
+ body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{
+ {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])},
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeReq, err := newTestSignedRequestV4(http.MethodPost,
+ getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(body)),
+ bytes.NewReader(body), peer.AccessKey, peer.SecretKey, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(completeRec, completeReq)
+ if completeRec.Code != http.StatusOK {
+ t.Fatalf("%s: marker-only CompleteMultipartUpload %d: %s", instanceType, completeRec.Code, completeRec.Body.String())
+ }
+ }
+ markerOnlyPUT := func(t *testing.T, object string, lockHeaders map[string]string) {
+ t.Helper()
+ req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucketName, object), int64(len(data)),
+ bytes.NewReader(data), peer.AccessKey, peer.SecretKey, markerOnly(lockHeaders))
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(rec, req)
+ if rec.Code != http.StatusOK {
+ t.Fatalf("%s: marker-only PUT %d: %s", instanceType, rec.Code, rec.Body.String())
+ }
+ }
+ lockOf := func(t *testing.T, object string) objectLockState {
+ t.Helper()
+ info, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return storedObjectLockState(info.UserDefined)
+ }
+
+ // An explicit legal hold on a marker-only write must be applied, not dropped.
+ // A legal-hold request suppresses the bucket default retention, so the version
+ // carries only the hold.
+ explicitHold := map[string]string{xhttp.AmzObjectLockLegalHold: "ON"}
+
+ t.Run("put-explicit-legal-hold", func(t *testing.T) {
+ object := "marker-only/put-legal-hold"
+ markerOnlyPUT(t, object, explicitHold)
+ if got := lockOf(t, object); got.legalHold != "ON" {
+ t.Fatalf("%s: marker-only PUT dropped the explicit legal hold: got %+v", instanceType, got)
+ }
+ })
+ t.Run("mpu-explicit-legal-hold", func(t *testing.T) {
+ object := "marker-only/mpu-legal-hold"
+ markerOnlyMPU(t, object, explicitHold)
+ if got := lockOf(t, object); got.legalHold != "ON" {
+ t.Fatalf("%s: marker-only multipart dropped the explicit legal hold: got %+v", instanceType, got)
+ }
+ })
+ t.Run("put-bucket-default-retention", func(t *testing.T) {
+ object := "marker-only/put-default-retention"
+ markerOnlyPUT(t, object, nil)
+ if got := lockOf(t, object); got.mode != "GOVERNANCE" || got.retainUntil == "" {
+ t.Fatalf("%s: marker-only PUT dropped the bucket default retention: got %+v", instanceType, got)
+ }
+ })
+ t.Run("mpu-bucket-default-retention", func(t *testing.T) {
+ object := "marker-only/mpu-default-retention"
+ markerOnlyMPU(t, object, nil)
+ if got := lockOf(t, object); got.mode != "GOVERNANCE" || got.retainUntil == "" {
+ t.Fatalf("%s: marker-only multipart dropped the bucket default retention: got %+v", instanceType, got)
+ }
+ })
+}
+
+// TestAPIReplicaMultipartNewerHoldSurvivesCompletion verifies that a legal hold
+// that reaches a destination version AFTER a replica multipart upload was
+// initiated is not rolled back when that upload completes. The initiation
+// resolves the lock against the version as it then stands, but completion
+// re-orders the carried lock against the version read under the namespace write
+// lock that guards the replacement, so a newer hold (with its newer timestamp)
+// survives. See pgsty/silo#120 (Codex finding 1).
+func TestAPIReplicaMultipartNewerHoldSurvivesCompletion(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testAPIReplicaMultipartNewerHoldSurvivesCompletion,
+ makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ })
+}
+
+func testAPIReplicaMultipartNewerHoldSurvivesCompletion(obj ObjectLayer, instanceType, bucketName string,
+ apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
+) {
+ previousTLS := globalIsTLS
+ globalIsTLS = true
+ defer func() { globalIsTLS = previousTLS }()
+
+ key := bytes.Repeat([]byte{0x47}, 32)
+ keyMD5 := md5.Sum(key)
+ sseHeaders := map[string]string{
+ xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
+ xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
+ xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
+ }
+ object := "mpu-lock-boundary/obj"
+
+ // seal returns the sender's SSE-C replica seal and marker headers for the
+ // addressed version, with every Object Lock header stripped so the case owns
+ // the lock instruction.
+ seal := func(t *testing.T, srcInfo ObjectInfo) map[string]string {
+ t.Helper()
+ opts, _, err := putReplicationOpts(t.Context(), "", srcInfo)
+ if err != nil {
+ t.Fatal(err)
+ }
+ opts.Internal.SourceMTime = time.Time{}
+ hdrs := make(map[string]string)
+ for name, values := range opts.Header() {
+ if len(values) > 0 {
+ hdrs[name] = values[0]
+ }
+ }
+ hdrs[xhttp.MinIOSourceReplicationRequest] = "true"
+ hdrs[xhttp.AmzBucketReplicationStatus] = "REPLICA"
+ hdrs[xhttp.MinIOSourceETag] = srcInfo.ETag
+ for _, name := range []string{
+ xhttp.AmzObjectLockMode, xhttp.AmzObjectLockRetainUntilDate, xhttp.AmzObjectLockLegalHold,
+ xhttp.MinIOSourceObjectRetentionTimestamp, xhttp.MinIOSourceObjectLegalHoldTimestamp,
+ } {
+ delete(hdrs, name)
+ }
+ return hdrs
+ }
+ rawOf := func(t *testing.T, versionID string) (ObjectInfo, []byte) {
+ t.Helper()
+ gr, err := obj.GetObjectNInfo(t.Context(), bucketName, object, nil, http.Header{}, ObjectOptions{
+ ReplicationRequest: true, VersionID: versionID,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ srcInfo := gr.ObjInfo
+ cipher, rerr := io.ReadAll(gr)
+ gr.Close()
+ if rerr != nil {
+ t.Fatal(rerr)
+ }
+ return srcInfo, cipher
+ }
+
+ // A destination SSE-C version with no lock.
+ putCopyChecksumSource(t, apiRouter, credentials, bucketName, object,
+ bytes.Repeat([]byte("mpu-lock-boundary-"), 64), sseHeaders)
+ base, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ versionID := base.VersionID
+ srcInfo, cipher := rawOf(t, versionID)
+
+ // Initiate an SSE-C replica multipart carrying legal hold OFF stamped 09:00.
+ // The destination has no lock yet, so the decision made now is OFF@09:00.
+ newHdrs := seal(t, srcInfo)
+ newHdrs[xhttp.AmzObjectLockLegalHold] = "OFF"
+ newHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp0900
+ newReq, err := newTestSignedRequestV4(http.MethodPost,
+ getNewMultipartURL("", bucketName, object)+"&versionId="+versionID, 0, nil,
+ credentials.AccessKey, credentials.SecretKey, newHdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ newRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(newRec, newReq)
+ if newRec.Code != http.StatusOK {
+ t.Fatalf("%s: replica NewMultipartUpload %d: %s", instanceType, newRec.Code, newRec.Body.String())
+ }
+ var init InitiateMultipartUploadResponse
+ if err = xmlDecoder(newRec.Body, &init, int64(newRec.Body.Len())); err != nil {
+ t.Fatal(err)
+ }
+
+ // After initiation, a newer legal hold ON@10:00 lands on the same version
+ // through an independent SSE-C replica PUT retransmit.
+ putHdrs := seal(t, srcInfo)
+ putHdrs[xhttp.AmzObjectLockLegalHold] = "ON"
+ putHdrs[xhttp.MinIOSourceObjectLegalHoldTimestamp] = objectLockTestStamp1000
+ putReq, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectURL("", bucketName, object)+"?versionId="+versionID, int64(len(cipher)),
+ bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey, putHdrs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ putRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(putRec, putReq)
+ if putRec.Code != http.StatusOK {
+ t.Fatalf("%s: interleaving replica PUT %d: %s", instanceType, putRec.Code, putRec.Body.String())
+ }
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got.legalHold != "ON" {
+ t.Fatalf("%s: the interleaving PUT did not set the newer ON: got %+v", instanceType, got)
+ }
+
+ // Finish the multipart upload the sender's way and prove the newer ON survives.
+ partReq, err := newTestSignedRequestV4(http.MethodPut,
+ getPutObjectPartURL("", bucketName, object, init.UploadID, "1"), int64(len(cipher)),
+ bytes.NewReader(cipher), credentials.AccessKey, credentials.SecretKey,
+ map[string]string{xhttp.MinIOSourceReplicationRequest: "true"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ partRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(partRec, partReq)
+ if partRec.Code != http.StatusOK {
+ t.Fatalf("%s: replica PutObjectPart %d: %s", instanceType, partRec.Code, partRec.Body.String())
+ }
+ actualSize, err := srcInfo.GetActualSize()
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := xml.Marshal(CompleteMultipartUpload{Parts: []CompletePart{
+ {PartNumber: 1, ETag: canonicalizeETag(partRec.Header()[xhttp.ETag][0])},
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeReq, err := newTestSignedRequestV4(http.MethodPost,
+ getCompleteMultipartUploadURL("", bucketName, object, init.UploadID), int64(len(body)),
+ bytes.NewReader(body), credentials.AccessKey, credentials.SecretKey, map[string]string{
+ xhttp.MinIOSourceReplicationRequest: "true",
+ xhttp.MinIOSourceMTime: srcInfo.ModTime.Format(time.RFC3339Nano),
+ xhttp.MinIOSourceETag: srcInfo.ETag,
+ xhttp.MinIOReplicationActualObjectSize: strconv.FormatInt(actualSize, 10),
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ completeRec := httptest.NewRecorder()
+ apiRouter.ServeHTTP(completeRec, completeReq)
+ if completeRec.Code != http.StatusOK {
+ t.Fatalf("%s: replica CompleteMultipartUpload %d: %s", instanceType, completeRec.Code, completeRec.Body.String())
+ }
+
+ // The completion re-ordered the carried OFF@09:00 against the ON@10:00 that
+ // reached the version after initiation: the newer hold and its timestamp win.
+ want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000}
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: a newer legal hold that arrived after initiation was rolled back at completion: got %+v, want %+v",
+ instanceType, got, want)
+ }
+}
+
+// TestReplicaPutObjectLockReconcileUnderWriteLock exercises the PUT counterpart
+// of the multipart reconcile: a replica full write reaches the object layer
+// carrying the Object Lock its handler resolved, but the addressed version has
+// since taken a newer lock update. PutObject must re-order the incoming lock
+// against the version read under the write lock, so the newer stored value is
+// kept. Driving the object layer directly stands in for the handler-read /
+// backend-commit interleave without a timing race. See pgsty/silo#120.
+func TestReplicaPutObjectLockReconcileUnderWriteLock(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testReplicaPutObjectLockReconcileUnderWriteLock,
+ makeBucketOptions: MakeBucketOptions{LockEnabled: true},
+ })
+}
+
+func testReplicaPutObjectLockReconcileUnderWriteLock(obj ObjectLayer, instanceType, bucketName string,
+ _ http.Handler, _ auth.Credentials, t *testing.T,
+) {
+ ctx := t.Context()
+
+ seedVersion := func(t *testing.T, object string, meta map[string]string) string {
+ t.Helper()
+ info, err := obj.PutObject(ctx, bucketName, object,
+ mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+ ObjectOptions{Versioned: true, UserDefined: meta})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return info.VersionID
+ }
+ // replicaWrite overwrites the addressed version the way a replica retransmit
+ // reaches the object layer, with the in-lock reconcile enabled.
+ replicaWrite := func(t *testing.T, object, versionID string, meta map[string]string) {
+ t.Helper()
+ _, err := obj.PutObject(ctx, bucketName, object,
+ mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+ ObjectOptions{Versioned: true, VersionID: versionID, ReplicaLockReconcile: true, UserDefined: meta})
+ if err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ t.Run("older-legal-hold-off-does-not-clear-newer-on", func(t *testing.T) {
+ object := "reconcile/put-legal-hold"
+ versionID := seedVersion(t, object, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000,
+ })
+ replicaWrite(t, object, versionID, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900,
+ })
+ want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000}
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: in-lock reconcile let an older OFF overwrite the newer stored ON: got %+v, want %+v",
+ instanceType, got, want)
+ }
+ })
+
+ t.Run("stale-retention-does-not-overwrite-newer", func(t *testing.T) {
+ object := "reconcile/put-retention"
+ versionID := seedVersion(t, object, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE",
+ strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilNewer,
+ ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: objectLockTestStamp1000,
+ })
+ replicaWrite(t, object, versionID, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE",
+ strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilStale,
+ ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: objectLockTestStamp0900,
+ })
+ want := objectLockFields{
+ mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, retentionStamp: objectLockTestStamp1000,
+ }
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: in-lock reconcile let a stale retention overwrite the newer stored one: got %+v, want %+v",
+ instanceType, got, want)
+ }
+ })
+
+ t.Run("newer-incoming-hold-applies", func(t *testing.T) {
+ object := "reconcile/put-newer-applies"
+ versionID := seedVersion(t, object, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900,
+ })
+ replicaWrite(t, object, versionID, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000,
+ })
+ want := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000}
+ if got := readObjectLockFields(t, obj, bucketName, object, versionID); got != want {
+ t.Fatalf("%s: in-lock reconcile did not apply the newer incoming hold: got %+v, want %+v",
+ instanceType, got, want)
+ }
+ })
+
+ t.Run("pre-upgrade-shape-on-absent-version-is-preserved", func(t *testing.T) {
+ // A pre-upgrade upload persisted validated lock values WITHOUT their
+ // ordering timestamps. Completing it while the destination version is
+ // absent must keep those values: there is nothing to order against, so the
+ // reconcile is skipped rather than deleting the accepted lock. The same
+ // not-found handling guards CompleteMultipartUpload.
+ object := "reconcile/put-absent-version"
+ absentVersion := mustGetUUID()
+ _, err := obj.PutObject(ctx, bucketName, object,
+ mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+ ObjectOptions{Versioned: true, VersionID: absentVersion, ReplicaLockReconcile: true, UserDefined: map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockMode): "GOVERNANCE",
+ strings.ToLower(xhttp.AmzObjectLockRetainUntilDate): retransmitRetainUntilNewer,
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := objectLockFields{mode: "GOVERNANCE", retainUntil: retransmitRetainUntilNewer, legalHold: "ON"}
+ if got := readObjectLockFields(t, obj, bucketName, object, absentVersion); got != want {
+ t.Fatalf("%s: a pre-upgrade lock (no ordering timestamps) on an absent version was stripped: got %+v, want %+v",
+ instanceType, got, want)
+ }
+ })
+}
+
+// TestReplicaLockReconcileNullVersion covers the null version, which persisted
+// upload metadata records as an empty VersionID. The completion reconcile must
+// order the incoming lock against the null version's own stored state -- looked
+// up as the null version, not the latest -- so a retransmit addressing the null
+// version cannot be reconciled against an unrelated UUID version, and a null
+// version absent while a UUID version exists is not mistaken for present. Single
+// erasure set. See pgsty/silo#120.
+func TestReplicaLockReconcileNullVersion(t *testing.T) {
+ defer DetectTestLeak(t)()
+ ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
+ t: t,
+ objAPITest: testReplicaLockReconcileNullVersion,
+ makeBucketOptions: MakeBucketOptions{VersioningEnabled: true},
+ })
+}
+
+func testReplicaLockReconcileNullVersion(obj ObjectLayer, instanceType, bucketName string,
+ _ http.Handler, _ auth.Credentials, t *testing.T,
+) {
+ ctx := t.Context()
+
+ // completeNullMPU runs an SSE-C replica multipart upload addressing the null
+ // version (VersionSuspended) carrying uploadLock, through the reconcile.
+ completeNullMPU := func(t *testing.T, object string, uploadLock map[string]string) {
+ t.Helper()
+ meta := map[string]string{crypto.MetaSealedKeySSEC: "dummy-sealed-key"}
+ maps.Copy(meta, uploadLock)
+ res, err := obj.NewMultipartUpload(ctx, bucketName, object, ObjectOptions{VersionSuspended: true, UserDefined: meta})
+ if err != nil {
+ t.Fatal(err)
+ }
+ // PutObjectPart now validates the DARE stream length (#119). This
+ // object-layer fixture needs an encrypted body before it can exercise
+ // the completion-time null-version metadata reconciliation.
+ var ciphertext bytes.Buffer
+ if _, err := sio.Encrypt(&ciphertext, bytes.NewReader([]byte("data")), sio.Config{Key: bytes.Repeat([]byte{1}, 32)}); err != nil {
+ t.Fatal(err)
+ }
+ part, err := obj.PutObjectPart(ctx, bucketName, object, res.UploadID, 1,
+ mustGetPutObjReader(t, bytes.NewReader(ciphertext.Bytes()), int64(ciphertext.Len()), "", ""), ObjectOptions{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := obj.CompleteMultipartUpload(ctx, bucketName, object, res.UploadID,
+ []CompletePart{{PartNumber: 1, ETag: part.ETag}},
+ ObjectOptions{VersionSuspended: true, ReplicaLockReconcile: true}); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ t.Run("null-and-uuid-present-reconciles-the-null-version", func(t *testing.T) {
+ object := "reconcile/null-vs-uuid"
+ // The null version holds a newer legal hold ON@10:00.
+ if _, err := obj.PutObject(ctx, bucketName, object,
+ mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+ ObjectOptions{VersionSuspended: true, MTime: time.Date(2026, 9, 3, 10, 0, 0, 0, time.UTC), UserDefined: map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1000,
+ }}); err != nil {
+ t.Fatal(err)
+ }
+ // A later UUID version holds an unrelated OFF@11:00 and is the latest.
+ uuidInfo, err := obj.PutObject(ctx, bucketName, object,
+ mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+ ObjectOptions{Versioned: true, MTime: time.Date(2026, 9, 3, 11, 0, 0, 0, time.UTC), UserDefined: map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1100,
+ }})
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // A null-version SSE-C retransmit carrying an older OFF@09:00.
+ completeNullMPU(t, object, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp0900,
+ })
+
+ // The null version keeps its own newer ON@10:00; the UUID is untouched.
+ wantNull := objectLockFields{legalHold: "ON", legalHoldStamp: objectLockTestStamp1000}
+ if got := readObjectLockFields(t, obj, bucketName, object, nullVersionID); got != wantNull {
+ t.Fatalf("%s: null-version completion reconciled against the wrong version: got %+v, want %+v", instanceType, got, wantNull)
+ }
+ wantUUID := objectLockFields{legalHold: "OFF", legalHoldStamp: objectLockTestStamp1100}
+ if got := readObjectLockFields(t, obj, bucketName, object, uuidInfo.VersionID); got != wantUUID {
+ t.Fatalf("%s: the UUID version was changed by a null-version completion: got %+v, want %+v", instanceType, got, wantUUID)
+ }
+ })
+
+ t.Run("absent-null-with-uuid-keeps-accepted-lock", func(t *testing.T) {
+ object := "reconcile/null-absent"
+ // Only a UUID version exists; there is no null version.
+ if _, err := obj.PutObject(ctx, bucketName, object,
+ mustGetPutObjReader(t, bytes.NewReader([]byte("data")), 4, "", ""),
+ ObjectOptions{Versioned: true, MTime: time.Date(2026, 9, 3, 11, 0, 0, 0, time.UTC), UserDefined: map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "OFF",
+ ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: objectLockTestStamp1100,
+ }}); err != nil {
+ t.Fatal(err)
+ }
+
+ // A null-version retransmit with its own validated legal hold ON. The null
+ // version does not exist, so the write must keep its accepted lock rather
+ // than order against the unrelated UUID version.
+ completeNullMPU(t, object, map[string]string{
+ strings.ToLower(xhttp.AmzObjectLockLegalHold): "ON",
+ })
+ if got := readObjectLockFields(t, obj, bucketName, object, nullVersionID); got.legalHold != "ON" {
+ t.Fatalf("%s: an absent null version was reconciled against the UUID version: got %+v", instanceType, got)
+ }
+ })
+}