Files
minio/cmd/object-multipart-handlers.go
Feng Ruohang 109d824e5f fix: retransmit and re-order Object Lock for SSE-C replicas (single erasure set)
Issue #120 routes an existing SSE-C replica through PutObjectHandler and
NewMultipartUploadHandler. On main those handlers assigned the incoming retention
and legal hold directly, without the source-timestamp ordering #111 added to
CopyObjectHandler and without persisting the ordering timestamps, so in
active-active replication a retransmit carrying an older value could overwrite a
destination version's newer lock state.

Share #111's ordering decision as applyReplicatedObjectLock in
cmd/bucket-object-lock.go and call it from CopyObject, PUT and multipart
initiation. A request that is not an actual trusted replica keeps ordinary write
semantics (a validated value is applied and stamped now); only a real replica
update is ordered against the stored version, so a marker-only peer write no
longer drops a validated hold or default retention. CopyObject keeps its SSE-C
key-rotation encMetadata reconciliation inline. putReplicationOpts now emits a
stored retention ordering timestamp even when the value keys are absent, so a
removal recorded on the retransmit PUT path still replicates onward. replicateAll
marks Failed and carries the error when putReplicationOpts fails.

The handler decision is made against the version as it stands then, which a
concurrent lock update can outrun before the write commits, and for multipart
across the whole initiation-to-completion span. Close that window under the
object write lock the receiving erasure set holds: a trusted SSE-C replica full
write sets ObjectOptions.ReplicaLockReconcile, and erasureObjects.PutObject and
CompleteMultipartUpload re-run the ordering (reconcileStoredObjectLock, which
orders retention and legal hold independently by their reserved timestamps)
against the destination version read on that set before committing. Persisted
upload metadata records the null version as an empty VersionID, so completion
looks that up as the null version rather than the latest. The reconcile runs only
against an existing version; a not-found destination keeps the write's own
accepted lock, including a pre-upgrade upload that persisted values without
ordering timestamps, and a non-not-found read error fails the write. Scoped to
the SSE-C paths this issue enables; CopyObject is left as #111 wrote it.

Scope: this orders Object Lock against the destination version under the write
lock and is correct for a single erasure set. A multi-pool deployment -- where a
version can have duplicate copies across pools, object ModTime ties do not track
per-field lock timestamps, and the object namespace lock is per-pool -- needs a
cross-pool lock-safe reconcile and is deliberately out of scope here, tracked in
pgsty/silo#TBD-multipool-lock.

Tests: TestAPISSECReplicaRetransmitObjectLockOrdering and its multipart sibling;
TestAPIReplicaMultipartNewerHoldSurvivesCompletion and
TestReplicaPutObjectLockReconcileUnderWriteLock (a hold or retention reaching the
version after the handler decision, or after multipart initiation, survives the
commit; a pre-upgrade upload on an absent version keeps its lock);
TestReplicaLockReconcileNullVersion (a null-version completion reconciles the null
version, not a coexisting UUID version, and an absent null version keeps its
accepted lock); TestAPIReplicaMarkerOnlyAppliesObjectLock; TestReplicaStoredLock;
the timestamp-only putReplicationOpts round trip; and the retransmit, exemption
and target-head tests. The #111 CopyObject replica suite and the existing #120
suite stay green, as do the PUT/multipart handler and object-layer regression
suites. Compatibility: the shared helper preserves #111's CopyObject behavior; a
non-replica PUT or multipart initiation that sets Object Lock now also stamps the
reserved ordering timestamp, matching CopyObject since #111; only trusted SSE-C
replica writes take the in-lock reconcile.

Refs pgsty/silo#120

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L7qJqWwy8oFA6aCXWRzXQe
Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-09-07 00:11:04 +08:00

1424 lines
48 KiB
Go

// Copyright (c) 2015-2023 MinIO, Inc.
//
// 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.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bufio"
"context"
"fmt"
"io"
"maps"
"net/http"
"net/textproto"
"net/url"
"strconv"
"strings"
"time"
"github.com/google/uuid"
"github.com/minio/minio-go/v7"
"github.com/minio/minio-go/v7/pkg/encrypt"
"github.com/minio/minio-go/v7/pkg/tags"
sse "github.com/minio/minio/internal/bucket/encryption"
objectlock "github.com/minio/minio/internal/bucket/object/lock"
"github.com/minio/minio/internal/bucket/replication"
"github.com/minio/minio/internal/config/dns"
"github.com/minio/minio/internal/config/storageclass"
"github.com/minio/minio/internal/crypto"
"github.com/minio/minio/internal/etag"
"github.com/minio/minio/internal/event"
"github.com/minio/minio/internal/handlers"
"github.com/minio/minio/internal/hash"
"github.com/minio/minio/internal/hash/sha256"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/logger"
"github.com/minio/mux"
"github.com/minio/sio"
"github.com/pgsty/silo-pkg/v3/policy"
)
// Multipart objectAPIHandlers
// isFederatedInternalRequest reports whether User-Agent carries the minio-go
// application token attached by getRemoteInstanceClient.
//
// This is only a response-shape hint. User-Agent is not authenticated and must
// never gate authorization, object visibility, or request validation. It is
// safe here because the only effect is returning the checksum of the body the
// caller was already authorized to upload.
func isFederatedInternalRequest(userAgent string) bool {
for _, product := range strings.Fields(userAgent) {
name, version, ok := strings.Cut(product, "/")
if ok && name == federatedInternalAppName && version != "" {
return true
}
}
return false
}
// partChecksumMap returns the non-empty part checksums in the form expected by
// hash.AddChecksumHeader. x-amz-checksum-type is deliberately excluded because
// UploadPart does not return it and minio-go cannot carry it in ObjectPart.
func partChecksumMap(partInfo PartInfo) map[string]string {
checksums := make(map[string]string, 1)
if partInfo.ChecksumCRC32 != "" {
checksums[hash.ChecksumCRC32.String()] = partInfo.ChecksumCRC32
}
if partInfo.ChecksumCRC32C != "" {
checksums[hash.ChecksumCRC32C.String()] = partInfo.ChecksumCRC32C
}
if partInfo.ChecksumSHA1 != "" {
checksums[hash.ChecksumSHA1.String()] = partInfo.ChecksumSHA1
}
if partInfo.ChecksumSHA256 != "" {
checksums[hash.ChecksumSHA256.String()] = partInfo.ChecksumSHA256
}
if partInfo.ChecksumCRC64NVME != "" {
checksums[hash.ChecksumCRC64NVME.String()] = partInfo.ChecksumCRC64NVME
}
return checksums
}
// multipartChecksumType returns the base checksum type recorded when a
// multipart upload was created. The boolean reports whether an algorithm was
// recorded at all.
func multipartChecksumType(metadata map[string]string) (hash.ChecksumType, bool) {
algorithm := metadata[hash.MinIOMultipartChecksum]
if algorithm == "" {
return hash.ChecksumNone, false
}
t := hash.NewChecksumType(algorithm, metadata[hash.MinIOMultipartChecksumType])
if !t.IsSet() {
return t, true
}
return t.Base(), true
}
// prepareMultipartChecksumReader validates a supplied part checksum algorithm,
// or installs a server-side hasher when the client omitted the optional
// checksum. It must run before compression or encryption can consume reader.
func prepareMultipartChecksumReader(reader *hash.Reader, metadata map[string]string, bucket, object string) error {
want, ok := multipartChecksumType(metadata)
if !ok {
return nil
}
got := reader.ContentCRCType()
if !got.IsSet() && reader.ServerSideChecksumType.IsSet() {
got = reader.ServerSideChecksumType
}
if !want.IsSet() || (got.IsSet() && got.Base() != want) {
return InvalidArgument{
Bucket: bucket,
Object: object,
Err: fmt.Errorf("checksum missing, want %q, got %q",
metadata[hash.MinIOMultipartChecksum], got.String()),
}
}
if !got.IsSet() {
reader.AddServerSideChecksumHasher(want)
}
return nil
}
// NewMultipartUploadHandler - New multipart upload.
// Notice: The S3 client can send secret keys in headers for encryption related jobs,
// the handler should ensure to remove these keys before sending them to the object layer.
// Currently these keys are:
// - X-Amz-Server-Side-Encryption-Customer-Key
// - X-Amz-Copy-Source-Server-Side-Encryption-Customer-Key
func (api objectAPIHandlers) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "NewMultipartUpload")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
vars := mux.Vars(r)
bucket := vars["bucket"]
object, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
requestTags, hasRequestTags := getRequestHeaderOrQueryValue(r, xhttp.AmzObjectTagging)
var requestTagsPtr *string
if hasRequestTags {
requestTagsPtr = &requestTags
}
if s3Error := checkRequestAuthTypeWithRequestTags(ctx, r, policy.PutObjectAction, bucket, object, requestTagsPtr); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
trustedReplication, replicaTrusted, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
// A validated raw SSE-C replica upload carries source ciphertext in every
// part; the destination must not add its own encryption or compression.
rawSSECReplica := isRawSSECReplica(r.Header, replicaTrusted)
if !rawSSECReplica {
// Check if bucket encryption is enabled
sseConfig, _ := globalBucketSSEConfigSys.Get(bucket)
sseConfig.Apply(r.Header, sse.ApplyOptions{
AutoEncrypt: globalAutoEncryption,
})
}
// Validate the storage class header if present. Query values retain the
// existing compatibility path, including its historical validation behavior.
if sc := r.Header.Get(xhttp.AmzStorageClass); sc != "" {
if !storageclass.IsValid(sc) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidStorageClass), r.URL)
return
}
}
encMetadata := map[string]string{}
if crypto.Requested(r.Header) {
if crypto.SSECopy.IsRequested(r.Header) {
writeErrorResponse(ctx, w, toAPIError(ctx, errInvalidEncryptionParameters), r.URL)
return
}
if crypto.SSEC.IsRequested(r.Header) && crypto.S3.IsRequested(r.Header) {
writeErrorResponse(ctx, w, toAPIError(ctx, crypto.ErrIncompatibleEncryptionMethod), r.URL)
return
}
if crypto.SSEC.IsRequested(r.Header) && crypto.S3KMS.IsRequested(r.Header) {
writeErrorResponse(ctx, w, toAPIError(ctx, crypto.ErrIncompatibleEncryptionMethod), r.URL)
return
}
if !rawSSECReplica {
if err = setEncryptionMetadata(r, bucket, object, encMetadata); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
// Set this for multipart only operations, we need to differentiate during
// decryption if the file was actually multipart or not.
encMetadata[ReservedMetadataPrefix+"Encrypted-Multipart"] = ""
}
// Extract metadata that needs to be saved.
metadata, err := extractMetadataFromReq(ctx, r)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if objTags := metadata[xhttp.AmzObjectTagging]; objTags != "" {
if _, err := tags.ParseObjectTags(objTags); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
if replicaTrusted {
if err = extractReplicationMetadataFromMime(ctx, textproto.MIMEHeader(r.Header), metadata); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
metadata[xhttp.AmzBucketReplicationStatus] = replication.Replica.String()
metadata[ReservedMetadataPrefixLower+ReplicaStatus] = replication.Replica.String()
metadata[ReservedMetadataPrefixLower+ReplicaTimestamp] = UTCNow().Format(time.RFC3339Nano)
} else {
delete(metadata, xhttp.AmzBucketReplicationStatus)
}
retPerms := isPutActionAllowed(ctx, getRequestAuthType(r), bucket, object, r, policy.PutObjectRetentionAction)
holdPerms := isPutActionAllowed(ctx, getRequestAuthType(r), bucket, object, r, policy.PutObjectLegalHoldAction)
getObjectInfo := objectAPI.GetObjectInfo
retentionMode, retentionDate, legalHold, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, getObjectInfo, retPerms, holdPerms, replicaTrusted)
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)
return
}
if dsc := mustReplicate(ctx, bucket, object, getMustReplicateOptions(metadata, "", "", replication.ObjectReplicationType, ObjectOptions{})); dsc.ReplicateAny() {
metadata[ReservedMetadataPrefixLower+ReplicationTimestamp] = UTCNow().Format(time.RFC3339Nano)
metadata[ReservedMetadataPrefixLower+ReplicationStatus] = dsc.PendingStatus()
}
// We need to preserve the encryption headers set in EncryptRequest,
// so we do not want to override them, copy them instead.
maps.Copy(metadata, encMetadata)
// Ensure that metadata does not contain sensitive information
crypto.RemoveSensitiveEntries(metadata)
if !rawSSECReplica && isCompressible(r.Header, object) {
// Storing the compression metadata.
metadata[ReservedMetadataPrefix+"compression"] = compressionAlgorithmV2
}
opts, err := putOptsFromReq(ctx, r, bucket, object, metadata)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if r.Header.Get(xhttp.IfMatch) != "" {
opts.HasIfMatch = true
}
if opts.PreserveETag != "" ||
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
}
return checkPreconditionsPUT(ctx, w, r, oi, opts)
}
}
if _, err := hash.GetContentChecksum(r.Header); err != nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
}
checksumType := hash.NewChecksumHeader(r.Header)
if checksumType.Is(hash.ChecksumInvalid) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
} else if checksumType.IsSet() && !checksumType.Is(hash.ChecksumTrailing) {
opts.WantChecksum = &hash.Checksum{Type: checksumType}
}
if opts.WantChecksum != nil {
opts.WantChecksum.Type |= hash.ChecksumMultipart | hash.ChecksumIncludesMultipart
}
newMultipartUpload := objectAPI.NewMultipartUpload
res, err := newMultipartUpload(ctx, bucket, object, opts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
response := generateInitiateMultipartUploadResponse(bucket, object, res.UploadID)
if res.ChecksumAlgo != "" {
w.Header().Set(xhttp.AmzChecksumAlgo, res.ChecksumAlgo)
if res.ChecksumType != "" {
w.Header().Set(xhttp.AmzChecksumType, res.ChecksumType)
}
}
encodedSuccessResponse := encodeResponse(response)
// Write success response.
writeSuccessResponseXML(w, encodedSuccessResponse)
}
// CopyObjectPartHandler - uploads a part by copying data from an existing object as data source.
func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "CopyObjectPart")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
if crypto.S3KMS.IsRequested(r.Header) { // SSE-KMS is not supported
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
return
}
vars := mux.Vars(r)
dstBucket := vars["bucket"]
dstObject, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if s3Error := checkRequestAuthType(ctx, r, policy.PutObjectAction, dstBucket, dstObject); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
// Read escaped copy source path to check for parameters.
cpSrcPath := r.Header.Get(xhttp.AmzCopySource)
var vid string
if u, err := url.Parse(cpSrcPath); err == nil {
vid = strings.TrimSpace(u.Query().Get(xhttp.VersionID))
// Note that url.Parse does the unescaping
cpSrcPath = u.Path
}
srcBucket, srcObject := path2BucketObject(cpSrcPath)
// If source object is empty or bucket is empty, reply back invalid copy source.
if srcObject == "" || srcBucket == "" {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidCopySource), r.URL)
return
}
if vid != "" && vid != nullVersionID {
_, err := uuid.Parse(vid)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, VersionNotFound{
Bucket: srcBucket,
Object: srcObject,
VersionID: vid,
}), r.URL)
return
}
}
if s3Error := checkRequestAuthType(ctx, r, policy.GetObjectAction, srcBucket, srcObject); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
if hasReplicaStatus(r.Header) &&
!replicationPermissionAllowed(ctx, r, dstBucket, dstObject, policy.ReplicateObjectAction) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, false, false)
}
uploadID := r.Form.Get(xhttp.UploadID)
partIDString := r.Form.Get(xhttp.PartNumber)
partID, err := strconv.Atoi(partIDString)
if err != nil || partID <= 0 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidPart), r.URL)
return
}
// check partID with maximum part ID for multipart objects
if isMaxPartID(partID) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidMaxParts), r.URL)
return
}
var srcOpts, dstOpts ObjectOptions
srcOpts, err = copySrcOpts(ctx, r, srcBucket, srcObject)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
srcOpts.VersionID = vid
// convert copy src and dst encryption options for GET/PUT calls
getOpts := ObjectOptions{VersionID: srcOpts.VersionID}
if srcOpts.ServerSideEncryption != nil {
getOpts.ServerSideEncryption = encrypt.SSE(srcOpts.ServerSideEncryption)
}
dstOpts, err = copyDstOpts(ctx, r, dstBucket, dstObject, nil)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
getObjectNInfo := objectAPI.GetObjectNInfo
// Get request range.
var rs *HTTPRangeSpec
var parseRangeErr error
if rangeHeader := r.Header.Get(xhttp.AmzCopySourceRange); rangeHeader != "" {
rs, parseRangeErr = parseCopyPartRangeSpec(rangeHeader)
} else {
// This check is to see if client specified a header but the value
// is empty for 'x-amz-copy-source-range'
_, ok := r.Header[xhttp.AmzCopySourceRange]
if ok {
parseRangeErr = errInvalidRange
}
}
checkCopyPartPrecondFn := func(o ObjectInfo) bool {
if _, err := DecryptObjectInfo(&o, r); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return true
}
if checkCopyObjectPartPreconditions(ctx, w, r, o) {
return true
}
if parseRangeErr != nil {
writeCopyPartErr(ctx, w, parseRangeErr, r.URL)
// Range header mismatch is pre-condition like failure
// so return true to indicate Range precondition failed.
return true
}
return false
}
getOpts.CheckPrecondFn = checkCopyPartPrecondFn
gr, err := getObjectNInfo(ctx, srcBucket, srcObject, rs, r.Header, getOpts)
if err != nil {
if isErrPreconditionFailed(err) {
return
}
if globalBucketVersioningSys.PrefixEnabled(srcBucket, srcObject) && gr != nil {
// Versioning enabled quite possibly object is deleted might be delete-marker
// if present set the headers, no idea why AWS S3 sets these headers.
if gr.ObjInfo.VersionID != "" && gr.ObjInfo.DeleteMarker {
w.Header()[xhttp.AmzVersionID] = []string{gr.ObjInfo.VersionID}
w.Header()[xhttp.AmzDeleteMarker] = []string{strconv.FormatBool(gr.ObjInfo.DeleteMarker)}
}
}
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
defer gr.Close()
srcInfo := gr.ObjInfo
actualPartSize, err := srcInfo.GetActualSize()
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err := enforceBucketQuotaHard(ctx, dstBucket, actualPartSize); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// Special care for CopyObjectPart
if partRangeErr := checkCopyPartRangeWithSize(rs, actualPartSize); partRangeErr != nil {
writeCopyPartErr(ctx, w, partRangeErr, r.URL)
return
}
// Get the object offset & length
startOffset, length, err := rs.GetOffsetLength(actualPartSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// maximum copy size for multipart objects in a single operation
if isMaxObjectSize(length) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrEntityTooLarge), r.URL)
return
}
if isRemoteCopyRequired(ctx, srcBucket, dstBucket, objectAPI) {
var dstRecords []dns.SrvRecord
dstRecords, err = globalDNSConfig.Get(dstBucket)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// Send PutObject request to appropriate instance (in federated deployment)
core, rerr := getRemoteInstanceClient(r, getHostFromSrv(dstRecords))
if rerr != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, rerr), r.URL)
return
}
popts := minio.PutObjectPartOptions{
SSE: dstOpts.ServerSideEncryption,
}
partInfo, err := core.PutObjectPart(ctx, dstBucket, dstObject, uploadID, partID, gr, length, popts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
response := generateCopyObjectPartResponse(PartInfo{
ETag: partInfo.ETag,
LastModified: partInfo.LastModified,
ChecksumCRC32: partInfo.ChecksumCRC32,
ChecksumCRC32C: partInfo.ChecksumCRC32C,
ChecksumSHA1: partInfo.ChecksumSHA1,
ChecksumSHA256: partInfo.ChecksumSHA256,
ChecksumCRC64NVME: partInfo.ChecksumCRC64NVME,
})
encodedSuccessResponse := encodeResponse(response)
// Write success response.
writeSuccessResponseXML(w, encodedSuccessResponse)
return
}
actualPartSize = length
var reader io.Reader = etag.NewReader(ctx, gr, nil, nil)
var checksumReader *hash.Reader
mi, err := objectAPI.GetMultipartInfo(ctx, dstBucket, dstObject, uploadID, dstOpts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if _, ok := multipartChecksumType(mi.UserDefined); ok {
checksumReader, err = hash.NewReader(ctx, reader, length, "", "", actualPartSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err = prepareMultipartChecksumReader(checksumReader, mi.UserDefined, dstBucket, dstObject); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
reader = checksumReader
}
_, isEncrypted := crypto.IsEncrypted(mi.UserDefined)
// Read compression metadata preserved in the init multipart for the decision.
_, isCompressed := mi.UserDefined[ReservedMetadataPrefix+"compression"]
// Compress only if the compression is enabled during initial multipart.
var idxCb func() []byte
if isCompressed {
wantEncryption := crypto.Requested(r.Header) || isEncrypted
s2c, cb := newS2CompressReader(reader, actualPartSize, wantEncryption)
idxCb = cb
defer s2c.Close()
reader = etag.Wrap(s2c, reader)
length = -1
}
srcInfo.Reader, err = hash.NewReader(ctx, reader, length, "", "", actualPartSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
dstOpts, err = copyDstOpts(ctx, r, dstBucket, dstObject, mi.UserDefined)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
dstOpts.IndexCB = idxCb
rawReader := srcInfo.Reader
pReader := NewPutObjReader(rawReader)
pReader.setChecksumReader(checksumReader)
var objectEncryptionKey crypto.ObjectKey
if isEncrypted {
if !crypto.SSEC.IsRequested(r.Header) && crypto.SSEC.IsEncrypted(mi.UserDefined) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrSSEMultipartEncrypted), r.URL)
return
}
if crypto.S3.IsEncrypted(mi.UserDefined) && crypto.SSEC.IsRequested(r.Header) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrSSEMultipartEncrypted), r.URL)
return
}
var key []byte
if crypto.SSEC.IsRequested(r.Header) {
key, err = ParseSSECustomerRequest(r)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
key, err = decryptObjectMeta(key, dstBucket, dstObject, mi.UserDefined)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
copy(objectEncryptionKey[:], key)
var nonce [12]byte
tmp := sha256.Sum256(fmt.Append(nil, uploadID, partID))
copy(nonce[:], tmp[:12])
partEncryptionKey := objectEncryptionKey.DerivePartKey(uint32(partID))
encReader, err := sio.EncryptReader(reader, sio.Config{
Key: partEncryptionKey[:],
Nonce: &nonce,
})
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
reader = etag.Wrap(encReader, reader)
wantSize := int64(-1)
if length >= 0 {
info := ObjectInfo{Size: length}
wantSize = info.EncryptedSize()
}
srcInfo.Reader, err = hash.NewReader(ctx, reader, wantSize, "", "", actualPartSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
pReader, err = pReader.WithEncryption(srcInfo.Reader, &objectEncryptionKey)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if dstOpts.IndexCB != nil {
dstOpts.IndexCB = compressionIndexEncrypter(objectEncryptionKey, dstOpts.IndexCB)
}
}
srcInfo.PutObjReader = pReader
copyObjectPart := objectAPI.CopyObjectPart
// Copy source object to destination, if source and destination
// object is same then only metadata is updated.
partInfo, err := copyObjectPart(ctx, srcBucket, srcObject, dstBucket, dstObject, uploadID, partID,
startOffset, length, srcInfo, srcOpts, dstOpts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if isEncrypted {
sseS3 := crypto.S3.IsRequested(r.Header) || crypto.S3.IsEncrypted(mi.UserDefined)
partInfo.ETag = tryDecryptETag(objectEncryptionKey[:], partInfo.ETag, sseS3)
}
response := generateCopyObjectPartResponse(partInfo)
encodedSuccessResponse := encodeResponse(response)
// Write success response.
writeSuccessResponseXML(w, encodedSuccessResponse)
}
// PutObjectPartHandler - uploads an incoming part for an ongoing multipart operation.
func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "PutObjectPart")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
vars := mux.Vars(r)
bucket := vars["bucket"]
object, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// X-Amz-Copy-Source shouldn't be set for this call.
if _, ok := r.Header[xhttp.AmzCopySource]; ok {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidCopySource), r.URL)
return
}
clientETag, err := etag.FromContentMD5(r.Header)
if err != nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidDigest), r.URL)
return
}
// if Content-Length is unknown/missing, throw away
size := r.ContentLength
rAuthType := getRequestAuthType(r)
// For auth type streaming signature, we need to gather a different content length.
switch rAuthType {
// Check signature types that must have content length
case authTypeStreamingSigned, authTypeStreamingSignedTrailer, authTypeStreamingUnsignedTrailer:
if sizeStr, ok := r.Header[xhttp.AmzDecodedContentLength]; ok {
if sizeStr[0] == "" {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentLength), r.URL)
return
}
size, err = strconv.ParseInt(sizeStr[0], 10, 64)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
}
if size == -1 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentLength), r.URL)
return
}
uploadID := r.Form.Get(xhttp.UploadID)
partIDString := r.Form.Get(xhttp.PartNumber)
partID, err := strconv.Atoi(partIDString)
if err != nil || partID <= 0 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidPart), r.URL)
return
}
// maximum size for multipart objects in a single operation
if isMaxObjectSize(size) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrEntityTooLarge), r.URL)
return
}
// check partID with maximum part ID for multipart objects
if isMaxPartID(partID) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidMaxParts), r.URL)
return
}
var (
md5hex = clientETag.String()
sha256hex = ""
reader io.Reader = r.Body
s3Error APIErrorCode
)
if s3Error = isPutActionAllowed(ctx, rAuthType, bucket, object, r, policy.PutObjectAction); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
switch rAuthType {
case authTypeStreamingSigned, authTypeStreamingSignedTrailer:
// Initialize stream signature verifier.
reader, s3Error = newSignV4ChunkedReader(r, rAuthType == authTypeStreamingSignedTrailer)
if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
case authTypeStreamingUnsignedTrailer:
// Initialize stream chunked reader with optional trailers.
reader, s3Error = newUnsignedV4ChunkedReader(r, true)
if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
case authTypeSignedV2, authTypePresignedV2:
if s3Error = isReqAuthenticatedV2(r); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
case authTypePresigned, authTypeSigned:
if s3Error = reqSignatureV4Verify(r, globalSite.Region(), serviceS3); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
if !skipContentSha256Cksum(r) {
sha256hex = getContentSha256Cksum(r, serviceS3)
}
}
if err := enforceBucketQuotaHard(ctx, bucket, size); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
actualSize := size
// get encryption options
var opts ObjectOptions
if crypto.SSEC.IsRequested(r.Header) {
opts, err = getOpts(ctx, r, bucket, object)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
mi, err := objectAPI.GetMultipartInfo(ctx, bucket, object, uploadID, opts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
trustedReplication, _, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
storedReplica := mi.UserDefined[xhttp.AmzBucketReplicationStatus] == replication.Replica.String()
replicaTrusted := trustedReplication && storedReplica
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
// Read compression metadata preserved in the init multipart for the decision.
_, isCompressed := mi.UserDefined[ReservedMetadataPrefix+"compression"]
var idxCb func() []byte
var checksumReader *hash.Reader
if isCompressed {
actualReader, err := hash.NewReader(ctx, reader, size, md5hex, sha256hex, actualSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err = actualReader.AddChecksum(r, false); err != nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
}
if err = prepareMultipartChecksumReader(actualReader, mi.UserDefined, bucket, object); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
checksumReader = actualReader
// Set compression metrics.
wantEncryption := crypto.Requested(r.Header)
s2c, cb := newS2CompressReader(actualReader, actualSize, wantEncryption)
idxCb = cb
defer s2c.Close()
reader = etag.Wrap(s2c, actualReader)
size = -1 // Since compressed size is un-predictable.
md5hex = "" // Do not try to verify the content.
sha256hex = ""
}
var forceMD5 []byte
// Optimization: If SSE-KMS and SSE-C did not request Content-Md5. Use uuid as etag. Optionally enable this also
// for server that is started with `--no-compat`.
if !etag.ContentMD5Requested(r.Header) && (crypto.S3KMS.IsEncrypted(mi.UserDefined) || crypto.SSEC.IsRequested(r.Header) || !globalServerCtxt.StrictS3Compat) {
forceMD5 = mustGetUUIDBytes()
}
hashReader, err := hash.NewReaderWithOpts(ctx, reader, hash.Options{
Size: size,
MD5Hex: md5hex,
SHA256Hex: sha256hex,
ActualSize: actualSize,
DisableMD5: false,
ForceMD5: forceMD5,
})
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err := hashReader.AddChecksum(r, size < 0); err != nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
}
if checksumReader == nil {
if err = prepareMultipartChecksumReader(hashReader, mi.UserDefined, bucket, object); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
checksumReader = hashReader
}
pReader := NewPutObjReader(hashReader)
pReader.setChecksumReader(checksumReader)
_, isEncrypted := crypto.IsEncrypted(mi.UserDefined)
var objectEncryptionKey crypto.ObjectKey
if isEncrypted {
if !crypto.SSEC.IsRequested(r.Header) && crypto.SSEC.IsEncrypted(mi.UserDefined) && !replicaTrusted {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrSSEMultipartEncrypted), r.URL)
return
}
opts, err = putOptsFromReq(ctx, r, bucket, object, mi.UserDefined)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
var key []byte
if crypto.SSEC.IsRequested(r.Header) {
key, err = ParseSSECustomerRequest(r)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
if !replicaTrusted || !crypto.SSEC.IsEncrypted(mi.UserDefined) {
// Calculating object encryption key
key, err = decryptObjectMeta(key, bucket, object, mi.UserDefined)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
copy(objectEncryptionKey[:], key)
partEncryptionKey := objectEncryptionKey.DerivePartKey(uint32(partID))
in := io.Reader(hashReader)
if size > encryptBufferThreshold {
// The encryption reads in blocks of 64KB.
// We add a buffer on bigger files to reduce the number of syscalls upstream.
in = bufio.NewReaderSize(hashReader, encryptBufferSize)
}
var nonce [12]byte
tmp := sha256.Sum256(fmt.Append(nil, uploadID, partID))
copy(nonce[:], tmp[:12])
reader, err = sio.EncryptReader(in, sio.Config{
Key: partEncryptionKey[:],
Nonce: &nonce,
})
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
wantSize := int64(-1)
if size >= 0 {
info := ObjectInfo{Size: size}
wantSize = info.EncryptedSize()
}
// do not try to verify encrypted content
hashReader, err = hash.NewReader(ctx, etag.Wrap(reader, hashReader), wantSize, "", "", actualSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err := hashReader.AddChecksum(r, true); err != nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
}
pReader, err = pReader.WithEncryption(hashReader, &objectEncryptionKey)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if idxCb != nil {
idxCb = compressionIndexEncrypter(objectEncryptionKey, idxCb)
}
opts.EncryptFn = metadataEncrypter(objectEncryptionKey)
}
}
opts.IndexCB = idxCb
opts.ReplicationRequest = trustedReplication
putObjectPart := objectAPI.PutObjectPart
partInfo, err := putObjectPart(ctx, bucket, object, uploadID, partID, pReader, opts)
if err != nil {
// Verify if the underlying error is signature mismatch.
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
etag := partInfo.ETag
if kind, encrypted := crypto.IsEncrypted(mi.UserDefined); encrypted {
switch kind {
case crypto.S3KMS:
w.Header().Set(xhttp.AmzServerSideEncryption, xhttp.AmzEncryptionKMS)
w.Header().Set(xhttp.AmzServerSideEncryptionKmsID, mi.KMSKeyID())
if kmsCtx, ok := mi.UserDefined[crypto.MetaContext]; ok {
w.Header().Set(xhttp.AmzServerSideEncryptionKmsContext, kmsCtx)
}
if len(etag) >= 32 && strings.Count(etag, "-") != 1 {
etag = etag[len(etag)-32:]
}
case crypto.S3:
w.Header().Set(xhttp.AmzServerSideEncryption, xhttp.AmzEncryptionAES)
etag, _ = DecryptETag(objectEncryptionKey, ObjectInfo{ETag: etag})
case crypto.SSEC:
w.Header().Set(xhttp.AmzServerSideEncryptionCustomerAlgorithm, r.Header.Get(xhttp.AmzServerSideEncryptionCustomerAlgorithm))
w.Header().Set(xhttp.AmzServerSideEncryptionCustomerKeyMD5, r.Header.Get(xhttp.AmzServerSideEncryptionCustomerKeyMD5))
if len(etag) >= 32 && strings.Count(etag, "-") != 1 {
etag = etag[len(etag)-32:]
}
}
}
// We must not use the http.Header().Set method here because some (broken)
// clients expect the ETag header key to be literally "ETag" - not "Etag" (case-sensitive).
// Therefore, we have to set the ETag directly as map entry.
w.Header()[xhttp.ETag] = []string{"\"" + etag + "\""}
hash.TransferChecksumHeader(w, r)
if isFederatedInternalRequest(r.UserAgent()) {
// Legacy federation proxies UploadPartCopy through minio-go
// Core.PutObjectPart, which can only recover checksums from response
// headers. Use the PartInfo returned by this exact write so the ETag and
// checksum cannot be mixed with a concurrent overwrite.
hash.AddChecksumHeader(w, partChecksumMap(partInfo))
}
writeSuccessResponseHeadersOnly(w)
}
// CompleteMultipartUploadHandler - Complete multipart upload.
func (api objectAPIHandlers) CompleteMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "CompleteMultipartUpload")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
vars := mux.Vars(r)
bucket := vars["bucket"]
object, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
if s3Error := checkRequestAuthType(ctx, r, policy.PutObjectAction, bucket, object); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
trustedReplication, replicaTrusted, trustErr := evaluateReplicationTrust(ctx, r, bucket, object, policy.ReplicateObjectAction)
if trustErr != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(trustErr), r.URL)
return
}
if hasReplicationRequestHeaders(r.Header) {
ctx, r = applyReplicationTrust(ctx, r, trustedReplication, replicaTrusted)
}
// Get upload id.
uploadID, _, _, _, s3Error := getObjectResources(r.Form)
if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
// Content-Length is required and should be non-zero
if r.ContentLength <= 0 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingPart), r.URL)
return
}
complMultipartUpload := &CompleteMultipartUpload{}
if err = xmlDecoder(r.Body, complMultipartUpload, r.ContentLength); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if len(complMultipartUpload.Parts) == 0 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingPart), r.URL)
return
}
// The parts list must be strictly increasing by part number. Gaps are
// allowed, repeats are not - sort.SliceIsSorted() with a '<' predicate
// considers equal neighbors sorted, so it is checked explicitly here,
// before anything is assembled into the target object.
for i := 1; i < len(complMultipartUpload.Parts); i++ {
if complMultipartUpload.Parts[i-1].PartNumber >= complMultipartUpload.Parts[i].PartNumber {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidPartOrder), r.URL)
return
}
}
// Reject retention or governance headers if set, CompleteMultipartUpload spec
// does not use these headers, and should not be passed down to checkPutObjectLockAllowed
if objectlock.IsObjectLockRequested(r.Header) || objectlock.IsObjectLockGovernanceBypassSet(r.Header) {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
return
}
if _, _, _, s3Err := checkPutObjectLockAllowed(ctx, r, bucket, object, objectAPI.GetObjectInfo, ErrNone, ErrNone, false); s3Err != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
return
}
completeMultiPartUpload := objectAPI.CompleteMultipartUpload
versioned := globalBucketVersioningSys.PrefixEnabled(bucket, object)
suspended := globalBucketVersioningSys.PrefixSuspended(bucket, object)
os := newObjSweeper(bucket, object).WithVersioning(versioned, suspended)
if !globalTierConfigMgr.Empty() {
// Get appropriate object info to identify the remote object to delete
goiOpts := os.GetOpts()
if goi, gerr := objectAPI.GetObjectInfo(ctx, bucket, object, goiOpts); gerr == nil {
os.SetTransitionState(goi.TransitionedObject)
}
}
opts, err := completeMultipartOpts(ctx, r, bucket, object)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
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:
// ETag := MD5(ETag_p1, ETag_p2, ...)+"-N" (N being the number of parts)
//
// This is independent of encryption. An encrypted multipart
// object also has an ETag that is the MD5 of its part ETags.
// The fact the in case of encryption the ETag of a part is
// not the MD5 of the part content does not change that.
var completeETags []etag.ETag
for _, part := range complMultipartUpload.Parts {
ETag, err := etag.Parse(part.ETag)
if err != nil {
continue
}
completeETags = append(completeETags, ETag)
}
multipartETag := etag.Multipart(completeETags...)
opts.UserDefined["etag"] = multipartETag.String()
if r.Header.Get(xhttp.IfMatch) != "" {
opts.HasIfMatch = true
}
if opts.PreserveETag != "" ||
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
}
return checkPreconditionsPUT(ctx, w, r, oi, opts)
}
}
objInfo, err := completeMultiPartUpload(ctx, bucket, object, uploadID, complMultipartUpload.Parts, opts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
opts.EncryptFn, err = objInfo.metadataEncryptFn(r.Header)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if r.Header.Get(xMinIOExtract) == "true" && HasSuffix(object, archiveExt) {
opts := ObjectOptions{VersionID: objInfo.VersionID, MTime: objInfo.ModTime}
if _, err := updateObjectMetadataWithZipInfo(ctx, objectAPI, bucket, object, opts); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
setPutObjHeaders(w, objInfo, false, r.Header)
if dsc := mustReplicate(ctx, bucket, object, objInfo.getMustReplicateOptions(replication.ObjectReplicationType, opts)); dsc.ReplicateAny() {
scheduleReplication(ctx, objInfo, objectAPI, dsc, replication.ObjectReplicationType)
}
if isTrustedReplication(ctx) {
actualSize, _ := objInfo.GetActualSize()
defer globalReplicationStats.Load().UpdateReplicaStat(bucket, actualSize)
}
// Get object location.
location := getObjectLocation(r, globalDomainNames, bucket, object)
// Generate complete multipart response.
response := generateCompleteMultipartUploadResponse(bucket, object, location, objInfo, r.Header)
encodedSuccessResponse := encodeResponse(response)
// Write success response.
writeSuccessResponseXML(w, encodedSuccessResponse)
// Notify object created event.
evt := eventArgs{
EventName: event.ObjectCreatedCompleteMultipartUpload,
BucketName: bucket,
Object: objInfo,
ReqParams: extractReqParams(r),
RespElements: extractRespElements(w),
UserAgent: r.UserAgent(),
Host: handlers.GetSourceIP(r),
}
sendEvent(evt)
if objInfo.NumVersions > int(scannerExcessObjectVersions.Load()) {
evt.EventName = event.ObjectManyVersions
sendEvent(evt)
auditLogInternal(context.Background(), AuditLogOptions{
Event: "scanner:manyversions",
APIName: "CompleteMultipartUpload",
Bucket: objInfo.Bucket,
Object: objInfo.Name,
VersionID: objInfo.VersionID,
Status: http.StatusText(http.StatusOK),
})
}
// Remove the transitioned object whose object version is being overwritten.
if !globalTierConfigMgr.Empty() {
// Schedule object for immediate transition if eligible.
enqueueTransitionImmediate(objInfo, lcEventSrc_s3CompleteMultipartUpload)
os.Sweep()
}
}
// AbortMultipartUploadHandler - Abort multipart upload
func (api objectAPIHandlers) AbortMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "AbortMultipartUpload")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
vars := mux.Vars(r)
bucket := vars["bucket"]
object, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
abortMultipartUpload := objectAPI.AbortMultipartUpload
if s3Error := checkRequestAuthType(ctx, r, policy.AbortMultipartUploadAction, bucket, object); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
uploadID, _, _, _, s3Error := getObjectResources(r.Form)
if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
opts := ObjectOptions{}
if err := abortMultipartUpload(ctx, bucket, object, uploadID, opts); err != nil {
switch err.(type) {
case InvalidUploadID:
// Do not have return an error for non-existent upload-id
default:
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
writeSuccessNoContent(w)
}
// ListObjectPartsHandler - List object parts
func (api objectAPIHandlers) ListObjectPartsHandler(w http.ResponseWriter, r *http.Request) {
ctx := newContext(r, w, "ListObjectParts")
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
vars := mux.Vars(r)
bucket := vars["bucket"]
object, err := unescapePath(vars["object"])
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
objectAPI := api.ObjectAPI()
if objectAPI == nil {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
if s3Error := checkRequestAuthType(ctx, r, policy.ListMultipartUploadPartsAction, bucket, object); s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
uploadID, partNumberMarker, maxParts, encodingType, s3Error := getObjectResources(r.Form)
if s3Error != ErrNone {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL)
return
}
if partNumberMarker < 0 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidPartNumberMarker), r.URL)
return
}
if maxParts < 0 {
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidMaxParts), r.URL)
return
}
opts := ObjectOptions{}
listPartsInfo, err := objectAPI.ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// We have to adjust the size of encrypted parts since encrypted parts
// are slightly larger due to encryption overhead.
// Further, we have to adjust the ETags of parts when using SSE-S3.
// Due to AWS S3, SSE-S3 encrypted parts return the plaintext ETag
// being the content MD5 of that particular part. This is not the
// case for SSE-C and SSE-KMS objects.
if kind, ok := crypto.IsEncrypted(listPartsInfo.UserDefined); ok {
var objectEncryptionKey []byte
if kind == crypto.S3 {
objectEncryptionKey, err = decryptObjectMeta(nil, bucket, object, listPartsInfo.UserDefined)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
}
for i, p := range listPartsInfo.Parts {
listPartsInfo.Parts[i].ETag = tryDecryptETag(objectEncryptionKey, p.ETag, kind == crypto.S3)
listPartsInfo.Parts[i].Size = p.ActualSize
}
} else if _, ok := listPartsInfo.UserDefined[ReservedMetadataPrefix+"compression"]; ok {
for i, p := range listPartsInfo.Parts {
listPartsInfo.Parts[i].Size = p.ActualSize
}
}
response := generateListPartsResponse(listPartsInfo, encodingType)
encodedSuccessResponse := encodeResponse(response)
// Write success response.
writeSuccessResponseXML(w, encodedSuccessResponse)
}