mirror of
https://github.com/pgsty/minio.git
synced 2026-10-30 12:30:39 +03:00
Merge pull request #134 from pgsty/fix/issue-120-ssec-replica-retransmit
fix: retransmit and re-order Object Lock for SSE-C replicas (single erasure set)
This commit is contained in:
@@ -631,6 +631,7 @@
|
|||||||
"x-minio-replication-encrypted-multipart",
|
"x-minio-replication-encrypted-multipart",
|
||||||
"x-minio-replication-ready",
|
"x-minio-replication-ready",
|
||||||
"x-minio-replication-reset-status",
|
"x-minio-replication-reset-status",
|
||||||
|
"x-minio-replication-server-side-encryption",
|
||||||
"x-minio-replication-server-side-encryption-iv",
|
"x-minio-replication-server-side-encryption-iv",
|
||||||
"x-minio-replication-server-side-encryption-seal-algorithm",
|
"x-minio-replication-server-side-encryption-seal-algorithm",
|
||||||
"x-minio-replication-server-side-encryption-sealed-key",
|
"x-minio-replication-server-side-encryption-sealed-key",
|
||||||
|
|||||||
@@ -25,6 +25,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/minio/minio/internal/amztime"
|
||||||
"github.com/minio/minio/internal/auth"
|
"github.com/minio/minio/internal/auth"
|
||||||
objectlock "github.com/minio/minio/internal/bucket/object/lock"
|
objectlock "github.com/minio/minio/internal/bucket/object/lock"
|
||||||
xhttp "github.com/minio/minio/internal/http"
|
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
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+27
-11
@@ -872,19 +872,29 @@ func putReplicationOpts(ctx context.Context, sc string, objInfo ObjectInfo) (put
|
|||||||
if cc, ok := lkMap.Lookup(xhttp.CacheControl); ok {
|
if cc, ok := lkMap.Lookup(xhttp.CacheControl); ok {
|
||||||
putOpts.CacheControl = cc
|
putOpts.CacheControl = cc
|
||||||
}
|
}
|
||||||
if mode, ok := lkMap.Lookup(xhttp.AmzObjectLockMode); ok {
|
mode, hasMode := lkMap.Lookup(xhttp.AmzObjectLockMode)
|
||||||
rmode := minio.RetentionMode(mode)
|
retainDateStr, hasRetainDate := lkMap.Lookup(xhttp.AmzObjectLockRetainUntilDate)
|
||||||
putOpts.Mode = rmode
|
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)
|
rdate, err := amztime.ISO8601Parse(retainDateStr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return putOpts, false, err
|
return putOpts, false, err
|
||||||
}
|
}
|
||||||
putOpts.RetainUntilDate = rdate
|
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
|
retTimestamp := objInfo.ModTime
|
||||||
if retainTmstampStr, ok := objInfo.UserDefined[ReservedMetadataPrefixLower+ObjectLockRetentionTimestamp]; ok {
|
if hasRetainTmstamp {
|
||||||
|
var err error
|
||||||
retTimestamp, err = time.Parse(time.RFC3339Nano, retainTmstampStr)
|
retTimestamp, err = time.Parse(time.RFC3339Nano, retainTmstampStr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return putOpts, false, err
|
return putOpts, false, err
|
||||||
@@ -1615,11 +1625,14 @@ func (ri ReplicateObjectInfo) replicateAll(ctx context.Context, objectAPI Object
|
|||||||
return rinfo
|
return rinfo
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// SSEC objects will refuse HeadObject without the decryption key.
|
// The sender holds no customer key, so the target refuses HeadObject on
|
||||||
// Ignore the error, since we know the object exists and versioning prevents overwriting existing versions.
|
// 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) {
|
if isSSEC && strings.Contains(cerr.Error(), errorCodes[ErrSSEEncryptedObject].Description) {
|
||||||
rinfo.ReplicationStatus = replication.Completed
|
rAction = replicateAll
|
||||||
rinfo.ReplicationAction = replicateNone
|
|
||||||
goto applyAction
|
goto applyAction
|
||||||
}
|
}
|
||||||
// if target returns error other than NoSuchKey, defer replication attempt
|
// if target returns error other than NoSuchKey, defer replication attempt
|
||||||
@@ -1704,8 +1717,11 @@ applyAction:
|
|||||||
} else {
|
} else {
|
||||||
putOpts, isMP, err := putReplicationOpts(ctx, tgt.StorageClass, objInfo)
|
putOpts, isMP, err := putReplicationOpts(ctx, tgt.StorageClass, objInfo)
|
||||||
if err != nil {
|
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.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))
|
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{
|
sendEvent(eventArgs{
|
||||||
EventName: event.ObjectReplicationNotTracked,
|
EventName: event.ObjectReplicationNotTracked,
|
||||||
|
|||||||
+54
-12
@@ -1114,7 +1114,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
|
|||||||
auditObjectErasureSet(ctx, "CompleteMultipartUpload", object, &er)
|
auditObjectErasureSet(ctx, "CompleteMultipartUpload", object, &er)
|
||||||
}
|
}
|
||||||
|
|
||||||
if opts.CheckPrecondFn != nil {
|
if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile {
|
||||||
if !opts.NoLock {
|
if !opts.NoLock {
|
||||||
ns := er.NewNSLock(bucket, object)
|
ns := er.NewNSLock(bucket, object)
|
||||||
lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
|
lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
|
||||||
@@ -1126,18 +1126,24 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
|
|||||||
opts.NoLock = true
|
opts.NoLock = true
|
||||||
}
|
}
|
||||||
|
|
||||||
obj, err := er.getObjectInfo(ctx, bucket, object, opts)
|
// The Object Lock reconcile below needs the version being committed, read
|
||||||
if err == nil && opts.CheckPrecondFn(obj) {
|
// after checkUploadIDExists, so only the precondition read happens here;
|
||||||
return ObjectInfo{}, PreConditionFailed{}
|
// both run under this same write lock, held until the version is renamed
|
||||||
}
|
// into place.
|
||||||
if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
if opts.CheckPrecondFn != nil {
|
||||||
return ObjectInfo{}, err
|
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 object doesn't exist return error for If-Match conditional requests
|
||||||
// If-None-Match should be allowed to proceed for non-existent objects
|
// If-None-Match should be allowed to proceed for non-existent objects
|
||||||
if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
|
if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
|
||||||
return ObjectInfo{}, err
|
return ObjectInfo{}, err
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1149,6 +1155,42 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
|
|||||||
return oi, toObjectErr(err, bucket, object, uploadID)
|
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)
|
uploadIDPath := er.getUploadIDDir(bucket, object, uploadID)
|
||||||
onlineDisks := er.getDisks()
|
onlineDisks := er.getDisks()
|
||||||
writeQuorum := fi.WriteQuorum(er.defaultWQuorum())
|
writeQuorum := fi.WriteQuorum(er.defaultWQuorum())
|
||||||
|
|||||||
+24
-8
@@ -1268,7 +1268,7 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
|
|||||||
|
|
||||||
data := r.Reader
|
data := r.Reader
|
||||||
|
|
||||||
if opts.CheckPrecondFn != nil {
|
if opts.CheckPrecondFn != nil || opts.ReplicaLockReconcile {
|
||||||
if !opts.NoLock {
|
if !opts.NoLock {
|
||||||
ns := er.NewNSLock(bucket, object)
|
ns := er.NewNSLock(bucket, object)
|
||||||
lkctx, err := ns.GetLock(ctx, globalOperationTimeout)
|
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)
|
obj, err := er.getObjectInfo(ctx, bucket, object, opts)
|
||||||
if err == nil && opts.CheckPrecondFn(obj) {
|
// A destination read that fails for a reason other than not-found must not
|
||||||
return objInfo, PreConditionFailed{}
|
// be taken as a passed precondition or as absent lock state.
|
||||||
}
|
|
||||||
if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
|
||||||
return objInfo, 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
|
// Order this trusted SSE-C replica's Object Lock against the addressed
|
||||||
// If-None-Match should be allowed to proceed for non-existent objects
|
// version's stored state, read on this erasure set under the write lock,
|
||||||
if err != nil && opts.HasIfMatch && (isErrObjectNotFound(err) || isErrVersionNotFound(err)) {
|
// so a value that lost the ordering cannot overwrite a newer one committed
|
||||||
return objInfo, err
|
// 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))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -99,6 +99,7 @@ type ObjectOptions struct {
|
|||||||
ReplicationSourceTaggingTimestamp time.Time // set if MinIOSourceTaggingTimestamp received
|
ReplicationSourceTaggingTimestamp time.Time // set if MinIOSourceTaggingTimestamp received
|
||||||
ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received
|
ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received
|
||||||
ReplicationSourceRetentionTimestamp time.Time // set if MinIOSourceObjectRetentionTimestamp 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,
|
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)
|
DeletePrefixObject bool // set true when object's erasure set is resolvable by object name (using getHashedSetIndex)
|
||||||
|
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ import (
|
|||||||
|
|
||||||
"github.com/minio/minio/internal/amztime"
|
"github.com/minio/minio/internal/amztime"
|
||||||
"github.com/minio/minio/internal/bucket/lifecycle"
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
||||||
|
"github.com/minio/minio/internal/crypto"
|
||||||
"github.com/minio/minio/internal/event"
|
"github.com/minio/minio/internal/event"
|
||||||
"github.com/minio/minio/internal/hash"
|
"github.com/minio/minio/internal/hash"
|
||||||
xhttp "github.com/minio/minio/internal/http"
|
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)
|
etagMatch := opts.PreserveETag != "" && isETagEqual(objInfo.ETag, opts.PreserveETag)
|
||||||
vidMatch := opts.VersionID != "" && opts.VersionID == objInfo.VersionID
|
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()
|
writeHeaders()
|
||||||
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrPreconditionFailed), r.URL)
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrPreconditionFailed), r.URL)
|
||||||
return true
|
return true
|
||||||
|
|||||||
+43
-53
@@ -1771,50 +1771,9 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
|
|||||||
// apply default bucket configuration/governance headers for dest side.
|
// apply default bucket configuration/governance headers for dest side.
|
||||||
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted)
|
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, dstBucket, dstObject, getObjectInfo, retPerms, holdPerms, replicaTrusted)
|
||||||
if s3Err == ErrNone {
|
if s3Err == ErrNone {
|
||||||
// A replica update is ordered by its source timestamp alone, whether or
|
applyReplicatedObjectLock(srcInfo.UserDefined, storedLock, replicaTrusted,
|
||||||
// not it still carries a value: an update newer than the stored state
|
retentionMode, retentionDate, legalHold,
|
||||||
// applies, and a removal is just an update with no value. An update that
|
dstOpts.ReplicationSourceRetentionTimestamp, dstOpts.ReplicationSourceLegalholdTimestamp)
|
||||||
// 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)
|
|
||||||
}
|
|
||||||
|
|
||||||
if replicaTrusted {
|
if replicaTrusted {
|
||||||
// An SSE-C key rotation snapshots every stored reserved key into
|
// 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.IfMatch) != "" ||
|
||||||
r.Header.Get(xhttp.IfNoneMatch) != "" {
|
r.Header.Get(xhttp.IfNoneMatch) != "" {
|
||||||
opts.CheckPrecondFn = func(oi ObjectInfo) bool {
|
opts.CheckPrecondFn = func(oi ObjectInfo) bool {
|
||||||
if _, err := DecryptObjectInfo(&oi, r); err != nil {
|
// A pure raw SSE-C replica overwrite (no public precondition) fully
|
||||||
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
// replaces the stored object, so the destination must not first
|
||||||
return true
|
// 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)
|
return checkPreconditionsPUT(ctx, w, r, oi, opts)
|
||||||
}
|
}
|
||||||
@@ -2282,12 +2253,31 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
|
|||||||
getObjectInfo := objectAPI.GetObjectInfo
|
getObjectInfo := objectAPI.GetObjectInfo
|
||||||
|
|
||||||
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, isReplicaTrusted(ctx))
|
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, isReplicaTrusted(ctx))
|
||||||
if s3Err == ErrNone && retentionMode.Valid() {
|
if s3Err == ErrNone {
|
||||||
metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
|
// A trusted replica write addressing a specific version can be a full
|
||||||
metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
|
// retransmit over an existing version whose lock state is newer than the
|
||||||
}
|
// source snapshot (issue #120). Order the incoming update against what is
|
||||||
if s3Err == ErrNone && legalHold.Status.Valid() {
|
// stored so a stale value cannot overwrite it; a non-replica write (and a
|
||||||
metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
|
// 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 {
|
if s3Err != ErrNone {
|
||||||
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
||||||
|
|||||||
@@ -34,7 +34,6 @@ import (
|
|||||||
"github.com/minio/minio-go/v7"
|
"github.com/minio/minio-go/v7"
|
||||||
"github.com/minio/minio-go/v7/pkg/encrypt"
|
"github.com/minio/minio-go/v7/pkg/encrypt"
|
||||||
"github.com/minio/minio-go/v7/pkg/tags"
|
"github.com/minio/minio-go/v7/pkg/tags"
|
||||||
"github.com/minio/minio/internal/amztime"
|
|
||||||
sse "github.com/minio/minio/internal/bucket/encryption"
|
sse "github.com/minio/minio/internal/bucket/encryption"
|
||||||
objectlock "github.com/minio/minio/internal/bucket/object/lock"
|
objectlock "github.com/minio/minio/internal/bucket/object/lock"
|
||||||
"github.com/minio/minio/internal/bucket/replication"
|
"github.com/minio/minio/internal/bucket/replication"
|
||||||
@@ -260,12 +259,32 @@ func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r
|
|||||||
getObjectInfo := objectAPI.GetObjectInfo
|
getObjectInfo := objectAPI.GetObjectInfo
|
||||||
|
|
||||||
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, replicaTrusted)
|
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, replicaTrusted)
|
||||||
if s3Err == ErrNone && retentionMode.Valid() {
|
if s3Err == ErrNone {
|
||||||
metadata[strings.ToLower(xhttp.AmzObjectLockMode)] = string(retentionMode)
|
// A trusted replica NewMultipartUpload addressing a specific version can
|
||||||
metadata[strings.ToLower(xhttp.AmzObjectLockRetainUntilDate)] = amztime.ISO8601Format(retentionDate.UTC())
|
// be a full retransmit over an existing version whose lock state is newer
|
||||||
}
|
// than the source snapshot (issue #120). Order the incoming update against
|
||||||
if s3Err == ErrNone && legalHold.Status.Valid() {
|
// what is stored so a stale value cannot overwrite it. opts is built below
|
||||||
metadata[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = string(legalHold.Status)
|
// (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 {
|
if s3Err != ErrNone {
|
||||||
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
||||||
@@ -1166,6 +1185,14 @@ func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWrite
|
|||||||
}
|
}
|
||||||
opts.Versioned = versioned
|
opts.Versioned = versioned
|
||||||
opts.VersionSuspended = suspended
|
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.
|
// First, we compute the ETag of the multipart object.
|
||||||
// The ETag of a multi-part object is always:
|
// The ETag of a multi-part object is always:
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user