mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-30 17:00:48 +01:00
* s3: read regular copy sources from their write owner * filer: report mutation sentinel codes in ObjectTransaction responses A failed mutation only surfaced resp.Error text, so routed writers could not distinguish an existing-directory conflict from a generic failure and could not apply the prefix-object retry the plain CreateEntry path has. * s3api: retry a routed mkfile as a prefix object over a directory mkFile retries ErrExistingIsDirectory by marking the entry a prefix object; the routed path returned the conflict instead, so a routed write to a key that is also a prefix failed where the lock path succeeds. The filer now reports the code on the transaction response, making the retry recognizable. * s3api: commit copy destinations on the key's write owner An owner-routed source read still left the destination on the gateway's default filer, so a second copy resolved the first copy's destination on the owner, found nothing, and returned NoSuchKey. Write unversioned and suspended destinations through routedMkFile, and write the version file on the owner the finalize recomputes on. * s3api: cache remote objects on the filer that served the entry An owner-routed lookup can return a remote-only entry that has not replicated to the gateway's default filer; the cache RPC then cannot find it and the copy or read degrades to a 503. Ask the entry's write owner to cache it first, falling back to the default filer. * s3: read versioned object state from the key's write owner CopyObject destinations now commit version files and the .versions pointer on the owner's object-scope ring key, so versioned reads through the default filer could miss them until replication. Generalize the routed read to dir/name lookups and paginated lists scoped to the object's owner, and point the versioned read paths at it: getSpecificObjectVersion, doGetLatestObjectVersion, the .versions quick-checks in Get/Head/attributes, the self-heal rescans, tagging, delete conditionals, directory-marker history checks, rename destination, multipart completion checks, and the version-id format probe. Reads fall back to the default filer when no ring owner is resolvable, preserving single-filer deployments. * test(s3api): fake RECOMPUTE_LATEST/DELETE and cover versioned copy reads The ambiguous-PUT fake filer now creates parent directories on PUT (mirroring InsertEntry), applies DELETE mutations, and re-derives the .versions pointer for RECOMPUTE_LATEST the way applyRecomputeLatest does, so owner-vs-replica tests can exercise versioned writes. TestVersionedCopyDestinationReadsOnOwner copies into a versioned bucket whose writes commit on the owner while the gateway's default filer is a replica with no replication, then asserts reads by versionId, latest-version resolution, and a copy chained off the new version all resolve on the owner. * test(s3api): gofmt the versioned-copy coverage * s3: only swap in a cache fill of the resolved remote generation A cache response can come back from a filer holding an older generation of the key. Replacing the authorized source or swapping its chunks under a size and range computed for the resolved entry serves the wrong object or truncates the response. Match RemoteEntry identity and size before using a cached entry, in both the copy and streaming paths. * filer: keep directory attributes when an object lands as prefix object The object-onto-directory retry replaced the existing directory's mode, owner, and extended attributes with the object's. Preserve the directory identity; merge directory-held extended keys only when the old entry never held an object, so a prior generation's object metadata is not carried forward. * s3: keep marker lock checks on the filer that replaces the marker The object-lock precondition read the key's write owner while mkdir still replaces the marker on the default filer, so a locked marker not yet replicated to the owner could be overwritten. Read the entry where the replacement lands, in both PutObjectHandler and zip extraction. Also hash a trailing-slash marker key to the same route key as the object it names, so marker and object share one owner and one write lock. * s3: route tag writes to the owner tag reads resolve GetObjectTagging reads the key's write owner, but PutObjectTagging and DeleteObjectTagging still wrote through the default filer, so a successful change returned the owner's stale tags until replication. setTags, rmTags, and the versioned tag updates now write on the owner first, falling back to the default filer when no owner resolves. * s3: keep rename lookups on the filer that performs the move AtomicRenameEntry and the retry token it carries both execute on the default filer, but the source and destination reads went to the write owner — a completed rename's token was invisible to its own retry, and a source that only exists on the default filer looked absent. Read both ends on the default filer. * s3: send the .versions pointer repair with the object's route key The heal reads the .versions entry on the object's owner, but the conditional UpdateEntry is routed by the .versions path's own ring key, which can hash to a different filer — leaving the pointer the routed reads consult unchanged. Ship the CAS repair as a PATCH_EXTENDED ObjectTransaction on the object's route key instead. * filer: preserve only directory identity on a prefix-object write Merging the old directory's extended attributes could carry a marker's encryption, version, or etag fields onto the new plaintext object, since a directory's own keys cannot be told apart from the object it held. Keep only mode, uid, gid, user name and group names. * s3: also require matching size when a cache fill replaces the source The remote-generation check covers identity; size equality is the cheap guard that keeps the streamed or copied range consistent with what was resolved. * s3: lock the object path for a routed .versions pointer repair conditionalUpdateEntryRouted held the lock on the .versions path while version writes and deletes serialize on the object path, so a repair could pass IF_ENTRY_EQUAL and then apply over a pointer a writer had advanced in between. Lock the object path like the writers and keep the condition on the .versions entry via condition_key. * s3: do not replay an owner-side tag mutation on another filer withWriteOwnerFilerClient re-ran fn on the default filer after any owner error, so an authoritative not-found could update a stale replica entry and still report success, and a post-apply transport error could apply the mutation twice. The failover walk now ends on any error fn returns; only a channel that never came up moves to another filer. * filer: keep a directory's xattrs when a prefix object replaces it The prefix-object branch preserves the directory's identity but dropped its Extended map wholesale; FUSE xattrs live there under the xattr- prefix and belong to the directory, not to the object the marker held. * s3: delete null and specific versions on the filer that serves them On the default filer an absent entry deletes as success, so a version that only reached the write owner survived a reported success while owner-routed reads kept serving it. The suspended-versioning and lifecycle deletes now go through routedDeleteNullVersion and routedDeleteSpecificVersion when an owner resolves; an unsettled attempt keeps the lock path. * s3: route a lifecycle null-version delete to the object entry The "null" version is the regular object entry, not a .versions file, so it goes through routedDeleteNullVersion; only a real version id takes the .versions-file delete. * s3: repoint .versions only while it still names the deleted version routedDeleteSpecificVersion recomputed the latest pointer on every version delete, so removing a noncurrent version re-derived the pointer to the newest remaining file — resurrecting an older version over a cleared pointer (a suspended bucket's null write keeps it clear while the null object stays current) or over a pointer that moved on. The pointer is now read on the owner and repointed under an IF_ENTRY_EQUAL guard only while it still names the deleted version; a contested pointer is left for the self-heal path rather than overwritten by a stale scan. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3: let lifecycle expiry keep its metadata-only deletes on routed paths The routed lifecycle deletes hardcoded IsDeleteData, so expiring an entry with a volume TTL queued per-chunk deletions instead of leaving the data for the volume to reclaim — extra work on the owner for deletes recordMetadataOnlyIf still counts as metadata-only. Pass the caller's cleanup choice through routedDeleteNullVersion and routedDeleteSpecificVersion; the interactive DELETE path keeps deleting data. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: let RECOMPUTE_LATEST skip a pointer that moved on A version delete repoints .versions only while the pointer still names the deleted version, but the gateway cannot carry the pointer's current state into the transaction's condition — the condition slot is already spent on the version's WORM guards. only_if_current_name moves the check under the transaction lock: the filer compares the pointer's name_to_key value before scanning and skips the recompute when it moved (or was cleared by a suspended bucket's null write), so a noncurrent delete cannot resurrect an older version. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3: repoint inside the version's locked delete repointLatestBeforeVersionDelete ran in its own transaction ahead of the delete, so a version delete denied by retention or legal hold still moved the latest pointer — GET and HEAD went on serving the older version it rolled to. The repoint and the delete now ride one transaction behind the version's WORM guards, atomically gated by the new only_if_current_name recompute guard instead of a stale-read IF_ENTRY_EQUAL. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2034 lines
83 KiB
Go
2034 lines
83 KiB
Go
package s3api
|
|
|
|
import (
|
|
"cmp"
|
|
"container/heap"
|
|
"context"
|
|
"crypto/md5"
|
|
"crypto/rand"
|
|
"encoding/base64"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"encoding/xml"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"net/url"
|
|
"path"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
|
|
"github.com/seaweedfs/seaweedfs/weed/stats"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
|
|
"github.com/aws/aws-sdk-go/aws"
|
|
"github.com/aws/aws-sdk-go/service/s3"
|
|
"github.com/google/uuid"
|
|
"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
|
|
|
|
"net/http"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/filer"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
|
)
|
|
|
|
const (
|
|
multipartExt = ".part"
|
|
multiPartMinSize = 5 * 1024 * 1024
|
|
)
|
|
|
|
type InitiateMultipartUploadResult struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ InitiateMultipartUploadResult"`
|
|
s3.CreateMultipartUploadOutput
|
|
|
|
// Checksum fields — returned as HTTP response headers, not in the XML body
|
|
ChecksumAlgorithm string `xml:"-"`
|
|
ChecksumType string `xml:"-"`
|
|
}
|
|
|
|
// getRequestScheme determines the URL scheme (http or https) from the request
|
|
// Checks X-Forwarded-Proto header first (for proxies), then TLS state
|
|
func getRequestScheme(r *http.Request) string {
|
|
// Check X-Forwarded-Proto header for proxied requests
|
|
if proto := r.Header.Get("X-Forwarded-Proto"); proto != "" {
|
|
return proto
|
|
}
|
|
// Check if connection is TLS
|
|
if r.TLS != nil {
|
|
return "https"
|
|
}
|
|
return "http"
|
|
}
|
|
|
|
func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateMultipartUploadInput) (output *InitiateMultipartUploadResult, code s3err.ErrorCode) {
|
|
|
|
glog.V(2).Infof("createMultipartUpload input %v", input)
|
|
|
|
uploadIdString := s3a.generateUploadID(*input.Key)
|
|
|
|
// a time-ordered uuid keeps uploads of one key in initiation order
|
|
uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.Must(uuid.NewV7()).String(), "-", "")
|
|
|
|
// Validate checksum algorithm before creating the upload directory
|
|
checksumAlgo, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r)
|
|
if checksumErrCode != s3err.ErrNone {
|
|
return nil, checksumErrCode
|
|
}
|
|
|
|
// Resolve and validate the requested checksum type (x-amz-checksum-type)
|
|
// against the algorithm so CompleteMultipartUpload knows whether to produce a
|
|
// COMPOSITE or FULL_OBJECT checksum.
|
|
checksumType := ""
|
|
if checksumHeaderName != "" {
|
|
resolvedType, typeErr := resolveMultipartChecksumType(checksumAlgo, r.Header.Get(s3_constants.AmzChecksumType))
|
|
if typeErr != nil {
|
|
glog.Warningf("createMultipartUpload: %v", typeErr)
|
|
return nil, s3err.ErrInvalidRequest
|
|
}
|
|
checksumType = resolvedType
|
|
}
|
|
|
|
// Prepare error handling outside callback scope
|
|
var encryptionError error
|
|
|
|
if err := s3a.mkdir(s3a.genUploadsFolder(*input.Bucket), uploadIdString, func(entry *filer_pb.Entry) {
|
|
if entry.Extended == nil {
|
|
entry.Extended = make(map[string][]byte)
|
|
}
|
|
entry.Extended[s3_constants.ExtMultipartObjectKey] = []byte(*input.Key)
|
|
// Set object owner for multipart upload
|
|
amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
|
|
if amzAccountId != "" {
|
|
entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
|
|
}
|
|
|
|
for k, v := range input.Metadata {
|
|
entry.Extended[k] = []byte(*v)
|
|
}
|
|
if input.ContentType != nil {
|
|
entry.Attributes.Mime = *input.ContentType
|
|
}
|
|
applyPutObjectACL(r, entry)
|
|
|
|
// Prepare and apply encryption configuration within directory creation
|
|
// This ensures encryption resources are only allocated if directory creation succeeds
|
|
encryptionConfig, prepErr := s3a.prepareMultipartEncryptionConfig(r, *input.Bucket, uploadIdString)
|
|
if prepErr != nil {
|
|
encryptionError = prepErr
|
|
return // Exit callback, letting mkdir handle the error
|
|
}
|
|
s3a.applyMultipartEncryptionConfig(entry, encryptionConfig)
|
|
|
|
// Store the requested checksum algorithm and type so CompleteMultipartUpload
|
|
// can compute the object checksum from per-part checksums
|
|
if checksumHeaderName != "" {
|
|
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(checksumHeaderName)
|
|
if checksumType != "" {
|
|
entry.Extended[s3_constants.ExtChecksumType] = []byte(checksumType)
|
|
}
|
|
}
|
|
|
|
// Extract and store object lock metadata from request headers
|
|
// This ensures object lock settings from create_multipart_upload are preserved
|
|
if err := s3a.extractObjectLockMetadataFromRequest(r, entry); err != nil {
|
|
glog.Errorf("createMultipartUpload: failed to extract object lock metadata: %v", err)
|
|
// Don't fail the upload - this matches AWS behavior for invalid metadata
|
|
}
|
|
}); err != nil {
|
|
_, errorCode := handleMultipartInternalError("create multipart upload directory", err)
|
|
return nil, errorCode
|
|
}
|
|
|
|
// Check for encryption configuration errors that occurred within the callback
|
|
if encryptionError != nil {
|
|
_, errorCode := handleMultipartInternalError("prepare encryption configuration", encryptionError)
|
|
return nil, errorCode
|
|
}
|
|
|
|
output = &InitiateMultipartUploadResult{
|
|
CreateMultipartUploadOutput: s3.CreateMultipartUploadOutput{
|
|
Bucket: input.Bucket,
|
|
Key: objectKey(input.Key),
|
|
UploadId: aws.String(uploadIdString),
|
|
},
|
|
ChecksumAlgorithm: checksumAlgorithmNameFromHeaderName(checksumHeaderName),
|
|
ChecksumType: checksumType,
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
type CompleteMultipartUploadResult struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CompleteMultipartUploadResult"`
|
|
Location *string `xml:"Location,omitempty"`
|
|
Bucket *string `xml:"Bucket,omitempty"`
|
|
Key *string `xml:"Key,omitempty"`
|
|
ETag *string `xml:"ETag,omitempty"`
|
|
|
|
ChecksumResult
|
|
ChecksumType string `xml:"ChecksumType,omitempty"`
|
|
|
|
// VersionId is NOT included in XML body - it should only be in x-amz-version-id HTTP header
|
|
|
|
// Store the VersionId internally for setting HTTP header, but don't marshal to XML
|
|
VersionId *string `xml:"-"`
|
|
}
|
|
|
|
// copySSEHeadersFromFirstPart copies all SSE-related headers from the first part to the destination entry
|
|
// This is critical for detectPrimarySSEType to work correctly and ensures encryption metadata is preserved
|
|
func copySSEHeadersFromFirstPart(dst *filer_pb.Entry, firstPart *filer_pb.Entry, context string) {
|
|
if firstPart == nil || firstPart.Extended == nil {
|
|
return
|
|
}
|
|
|
|
// Copy ALL SSE-related headers (not just SeaweedFSSSEKMSKey)
|
|
sseKeys := []string{
|
|
// SSE-C headers
|
|
s3_constants.SeaweedFSSSEIV,
|
|
s3_constants.AmzServerSideEncryptionCustomerAlgorithm,
|
|
s3_constants.AmzServerSideEncryptionCustomerKeyMD5,
|
|
// SSE-KMS headers
|
|
s3_constants.SeaweedFSSSEKMSKey,
|
|
s3_constants.AmzServerSideEncryptionAwsKmsKeyId,
|
|
// SSE-S3 headers
|
|
s3_constants.SeaweedFSSSES3Key,
|
|
// Common SSE header (for SSE-KMS and SSE-S3)
|
|
s3_constants.AmzServerSideEncryption,
|
|
}
|
|
|
|
for _, key := range sseKeys {
|
|
if value, exists := firstPart.Extended[key]; exists {
|
|
dst.Extended[key] = value
|
|
glog.V(4).Infof("completeMultipartUpload: copied SSE header %s from first part (%s)", key, context)
|
|
}
|
|
}
|
|
}
|
|
|
|
type multipartPartBoundary struct {
|
|
PartNumber int `json:"part"`
|
|
StartChunk int `json:"start"`
|
|
EndChunk int `json:"end"`
|
|
ETag string `json:"etag"`
|
|
// Byte offsets of the part; the chunk indexes above stop matching once
|
|
// the chunk list is folded into manifest chunks.
|
|
StartOffset int64 `json:"startOffset,omitempty"`
|
|
EndOffset int64 `json:"endOffset,omitempty"` // exclusive
|
|
}
|
|
|
|
type multipartSSES3Info struct {
|
|
keyData []byte
|
|
key *SSES3Key
|
|
baseIV []byte
|
|
}
|
|
|
|
type multipartCompletionState struct {
|
|
deleteEntries []*filer_pb.Entry
|
|
partEntries map[int][]*filer_pb.Entry
|
|
pentry *filer_pb.Entry
|
|
sses3Info *multipartSSES3Info
|
|
mime string
|
|
finalParts []*filer_pb.FileChunk
|
|
offset int64
|
|
partBoundaries []multipartPartBoundary
|
|
multipartETag string
|
|
entityWithTtl bool
|
|
checksumHeaderName string // e.g. "X-Amz-Checksum-Crc32", empty if no checksum
|
|
checksumValue string // multipart checksum: "base64-N" (COMPOSITE) or "base64" (FULL_OBJECT)
|
|
checksumType string // s3_constants.ChecksumType* (COMPOSITE or FULL_OBJECT)
|
|
|
|
newManifestChunks []*filer_pb.FileChunk // blobs folded for finalParts; orphans until the entry commits
|
|
manifestsReferenced bool // failed rollback left an entry holding newManifestChunks
|
|
supersededPartManifests []*filer_pb.FileChunk // part-entry blobs replaced by flattening; deleted after commit
|
|
metadataOnlyCleanup bool // deleteEntries share chunks with the live object; keep their data
|
|
uploadDirRemoved bool // the finalize transaction already removed .uploads/<uploadId>
|
|
}
|
|
|
|
func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult {
|
|
result := &CompleteMultipartUploadResult{
|
|
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
|
|
Bucket: input.Bucket,
|
|
ETag: aws.String(etag),
|
|
Key: objectKey(input.Key),
|
|
}
|
|
if entry != nil && entry.Extended != nil {
|
|
if versionIdBytes, ok := entry.Extended[s3_constants.ExtVersionIdKey]; ok {
|
|
versionId := string(versionIdBytes)
|
|
if versionId != "" && versionId != "null" {
|
|
result.VersionId = aws.String(versionId)
|
|
}
|
|
}
|
|
result.SetChecksum(string(entry.Extended[s3_constants.ExtChecksumAlgorithm]), string(entry.Extended[s3_constants.ExtChecksumValue]))
|
|
result.ChecksumType = string(entry.Extended[s3_constants.ExtChecksumType])
|
|
}
|
|
return result
|
|
}
|
|
|
|
func extractMultipartSSES3Info(entry *filer_pb.Entry) (*multipartSSES3Info, error) {
|
|
if entry == nil || entry.Extended == nil {
|
|
return nil, nil
|
|
}
|
|
if encryptionType := string(entry.Extended[s3_constants.SeaweedFSSSES3Encryption]); encryptionType != s3_constants.SSEAlgorithmAES256 {
|
|
return nil, nil
|
|
}
|
|
|
|
baseIVEncoded := entry.Extended[s3_constants.SeaweedFSSSES3BaseIV]
|
|
if len(baseIVEncoded) == 0 {
|
|
return nil, fmt.Errorf("missing SSE-S3 multipart base IV")
|
|
}
|
|
baseIV, err := base64.StdEncoding.DecodeString(string(baseIVEncoded))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode SSE-S3 multipart base IV: %w", err)
|
|
}
|
|
if len(baseIV) != s3_constants.AESBlockSize {
|
|
return nil, fmt.Errorf("invalid SSE-S3 multipart base IV length %d", len(baseIV))
|
|
}
|
|
|
|
keyDataEncoded := entry.Extended[s3_constants.SeaweedFSSSES3KeyData]
|
|
if len(keyDataEncoded) == 0 {
|
|
return nil, fmt.Errorf("missing SSE-S3 multipart key data")
|
|
}
|
|
keyData, err := base64.StdEncoding.DecodeString(string(keyDataEncoded))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode SSE-S3 multipart key data: %w", err)
|
|
}
|
|
|
|
key, err := DeserializeSSES3Metadata(keyData, GetSSES3KeyManager())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("deserialize SSE-S3 multipart key data: %w", err)
|
|
}
|
|
|
|
return &multipartSSES3Info{
|
|
keyData: keyData,
|
|
key: key,
|
|
baseIV: baseIV,
|
|
}, nil
|
|
}
|
|
|
|
func completedMultipartChunk(chunk *filer_pb.FileChunk, offset int64, sses3Info *multipartSSES3Info) (*filer_pb.FileChunk, error) {
|
|
finalChunk := &filer_pb.FileChunk{
|
|
FileId: chunk.GetFileIdString(),
|
|
Offset: offset,
|
|
Size: chunk.Size,
|
|
ModifiedTsNs: chunk.ModifiedTsNs,
|
|
CipherKey: chunk.CipherKey,
|
|
ETag: chunk.ETag,
|
|
IsCompressed: chunk.IsCompressed,
|
|
SseType: chunk.SseType,
|
|
SseMetadata: chunk.SseMetadata,
|
|
}
|
|
|
|
if sses3Info == nil {
|
|
return finalChunk, nil
|
|
}
|
|
if finalChunk.GetSseType() != filer_pb.SSEType_NONE && finalChunk.GetSseType() != filer_pb.SSEType_SSE_S3 {
|
|
return finalChunk, nil
|
|
}
|
|
// Trust existing per-chunk SSE-S3 metadata: if the part went through
|
|
// putToFiler's chunk loop with a non-nil sseS3Key it carries an offset
|
|
// derived IV that we cannot recover from the upload-entry alone. We do
|
|
// not validate that the embedded IV equals calculateIVWithOffset(baseIV,
|
|
// chunk.Offset); only completely missing metadata is backfilled. This
|
|
// keeps healthy parts untouched and limits backfill to the cases that
|
|
// caused #8908.
|
|
if finalChunk.GetSseType() == filer_pb.SSEType_SSE_S3 && len(finalChunk.GetSseMetadata()) > 0 {
|
|
return finalChunk, nil
|
|
}
|
|
|
|
// IV uses the chunk's offset within its part, with no partNumber term.
|
|
// This mirrors the encryption side: putToFiler hardcodes partOffset=0 when
|
|
// it calls handleSSES3MultipartEncryption for every part, so each part's
|
|
// encrypted stream begins at IV = calculateIVWithOffset(baseIV, 0) = baseIV.
|
|
// Each chunk within that stream is then at IV =
|
|
// calculateIVWithOffset(baseIV, chunk.Offset_part_local). Any deviation
|
|
// (e.g. adding (partNumber-1)*PartOffsetMultiplier) would produce IVs that
|
|
// do not match the actual encryption and would corrupt decryption.
|
|
chunkIV, _ := calculateIVWithOffset(sses3Info.baseIV, chunk.GetOffset())
|
|
chunkKey := &SSES3Key{
|
|
Key: sses3Info.key.Key,
|
|
KeyID: sses3Info.key.KeyID,
|
|
Algorithm: sses3Info.key.Algorithm,
|
|
IV: chunkIV,
|
|
}
|
|
chunkMetadata, err := SerializeSSES3Metadata(chunkKey)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("serialize SSE-S3 chunk metadata for %s: %w", chunk.GetFileIdString(), err)
|
|
}
|
|
|
|
finalChunk.SseType = filer_pb.SSEType_SSE_S3
|
|
finalChunk.SseMetadata = chunkMetadata
|
|
return finalChunk, nil
|
|
}
|
|
|
|
// applyMultipartSSES3HeadersFromUploadEntry writes the canonical object-level
|
|
// SSE-S3 attributes (SeaweedFSSSES3Key / X-Amz-Server-Side-Encryption) onto a
|
|
// completed multipart entry when they are missing. detectPrimarySSEType uses
|
|
// the object-level X-Amz-Server-Side-Encryption header to recognize SSE-S3 on
|
|
// inline / small-object reads, and IsSSES3EncryptedInternal requires both keys
|
|
// to consider an object encrypted at all. Without this, an object whose first
|
|
// part lacked the canonical attributes (e.g. a part written through a path
|
|
// that did not set them) would be served as unencrypted on GET.
|
|
func applyMultipartSSES3HeadersFromUploadEntry(dst *filer_pb.Entry, sses3Info *multipartSSES3Info) {
|
|
if dst == nil || sses3Info == nil {
|
|
return
|
|
}
|
|
if dst.Extended == nil {
|
|
dst.Extended = make(map[string][]byte)
|
|
}
|
|
if _, exists := dst.Extended[s3_constants.SeaweedFSSSES3Key]; !exists {
|
|
dst.Extended[s3_constants.SeaweedFSSSES3Key] = sses3Info.keyData
|
|
}
|
|
if _, exists := dst.Extended[s3_constants.AmzServerSideEncryption]; !exists {
|
|
dst.Extended[s3_constants.AmzServerSideEncryption] = []byte(s3_constants.SSEAlgorithmAES256)
|
|
}
|
|
}
|
|
|
|
func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload, uploadDirectory, entryName, dirName string, completedPartNumbers []int, completedPartMap map[int][]string, maxPartNo int) (*multipartCompletionState, *CompleteMultipartUploadResult, s3err.ErrorCode) {
|
|
if entry, err := s3a.resolveObjectEntry(*input.Bucket, *input.Key, ""); err == nil && entry != nil && entry.Extended != nil {
|
|
if uploadId, ok := entry.Extended[s3_constants.SeaweedFSUploadId]; ok && *input.UploadId == string(uploadId) {
|
|
cleanupEntries, _, cleanupErr := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1)
|
|
if cleanupErr != nil && !errors.Is(cleanupErr, filer_pb.ErrNotFound) {
|
|
glog.Warningf("completeMultipartUpload: failed to list stale upload directory %s for cleanup: %v", uploadDirectory, cleanupErr)
|
|
}
|
|
return &multipartCompletionState{deleteEntries: cleanupEntries, metadataOnlyCleanup: true}, completeMultipartResult(r, input, getEtagFromEntry(entry), entry), s3err.ErrNone
|
|
}
|
|
}
|
|
|
|
entries, _, err := s3a.list(uploadDirectory, "", "", false, s3_constants.MaxS3MultipartParts+1)
|
|
if err != nil {
|
|
glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
|
if isFilerNotFound(err) {
|
|
if output, code := s3a.resumeCommittedObject(r, input, dirName, entryName); code != s3err.ErrNoSuchUpload {
|
|
return &multipartCompletionState{metadataOnlyCleanup: true}, output, code
|
|
}
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
|
return nil, nil, s3err.ErrNoSuchUpload
|
|
}
|
|
// a store error must not read as "upload gone": the client would drop a fully-uploaded object
|
|
return nil, nil, s3err.ErrInternalError
|
|
}
|
|
if len(entries) == 0 {
|
|
if output, code := s3a.resumeCommittedObject(r, input, dirName, entryName); code != s3err.ErrNoSuchUpload {
|
|
return &multipartCompletionState{metadataOnlyCleanup: true}, output, code
|
|
}
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
|
return nil, nil, s3err.ErrNoSuchUpload
|
|
}
|
|
|
|
pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
|
|
if err != nil {
|
|
glog.Errorf("completeMultipartUpload %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
|
return nil, nil, s3err.ErrNoSuchUpload
|
|
}
|
|
return nil, nil, s3err.ErrInternalError
|
|
}
|
|
// only createMultipartUpload stamps the key; a directory a part write left behind is not an upload
|
|
if !isMultipartUploadEntry(pentry) {
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
|
return nil, nil, s3err.ErrNoSuchUpload
|
|
}
|
|
|
|
deleteEntries := make([]*filer_pb.Entry, 0)
|
|
partEntries := make(map[int][]*filer_pb.Entry, len(entries))
|
|
entityTooSmall := false
|
|
entityWithTtl := false
|
|
for _, entry := range entries {
|
|
foundEntry := false
|
|
glog.V(4).Infof("completeMultipartUpload part entries %s", entry.Name)
|
|
if entry.IsDirectory || !strings.HasSuffix(entry.Name, multipartExt) {
|
|
continue
|
|
}
|
|
partNumber, parseErr := parsePartNumber(entry.Name)
|
|
if parseErr != nil {
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartNumber).Inc()
|
|
glog.Errorf("completeMultipartUpload failed to parse partNumber %s:%s", entry.Name, parseErr)
|
|
continue
|
|
}
|
|
completedPartsByNumber, ok := completedPartMap[partNumber]
|
|
if !ok {
|
|
continue
|
|
}
|
|
for _, partETag := range completedPartsByNumber {
|
|
match, invalid, normalizedPartETag, normalizedEntryETag := validateCompletePartETag(partETag, entry)
|
|
if invalid {
|
|
glog.Warningf("invalid complete etag %s, storedEtag %s", normalizedPartETag, normalizedEntryETag)
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedEtagInvalid).Inc()
|
|
continue
|
|
}
|
|
if !match {
|
|
glog.Errorf("completeMultipartUpload %s ETag mismatch stored: %s part: %s", entry.Name, normalizedEntryETag, normalizedPartETag)
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedEtagMismatch).Inc()
|
|
continue
|
|
}
|
|
if len(entry.Chunks) == 0 && partNumber != maxPartNo {
|
|
glog.Warningf("completeMultipartUpload %s empty chunks", entry.Name)
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartEmpty).Inc()
|
|
continue
|
|
}
|
|
partEntries[partNumber] = append(partEntries[partNumber], entry)
|
|
foundEntry = true
|
|
}
|
|
if foundEntry {
|
|
if !entityWithTtl && entry.Attributes != nil && entry.Attributes.TtlSec > 0 {
|
|
entityWithTtl = true
|
|
}
|
|
if len(completedPartNumbers) > 1 && partNumber != completedPartNumbers[len(completedPartNumbers)-1] &&
|
|
entry.Attributes.FileSize < multiPartMinSize {
|
|
glog.Warningf("completeMultipartUpload %s part file size less 5mb", entry.Name)
|
|
entityTooSmall = true
|
|
}
|
|
} else {
|
|
deleteEntries = append(deleteEntries, entry)
|
|
}
|
|
}
|
|
if entityTooSmall {
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompleteEntityTooSmall).Inc()
|
|
return nil, nil, s3err.ErrEntityTooSmall
|
|
}
|
|
|
|
mime := ""
|
|
if pentry.Attributes != nil {
|
|
mime = pentry.Attributes.Mime
|
|
}
|
|
sses3Info, sses3Err := extractMultipartSSES3Info(pentry)
|
|
if sses3Err != nil {
|
|
glog.Errorf("completeMultipartUpload %s %s SSE-S3 metadata error: %v", *input.Bucket, *input.UploadId, sses3Err)
|
|
return nil, nil, s3err.ErrInternalError
|
|
}
|
|
finalParts := make([]*filer_pb.FileChunk, 0)
|
|
partBoundaries := make([]multipartPartBoundary, 0, len(completedPartNumbers))
|
|
var supersededPartManifests []*filer_pb.FileChunk
|
|
var offset int64
|
|
|
|
for _, partNumber := range completedPartNumbers {
|
|
partEntriesByNumber, ok := partEntries[partNumber]
|
|
if !ok {
|
|
glog.Errorf("part %d has no entry", partNumber)
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartNotFound).Inc()
|
|
return nil, nil, s3err.ErrInvalidPart
|
|
}
|
|
found := false
|
|
|
|
if len(partEntriesByNumber) > 1 {
|
|
sortEntriesByLatestChunk(partEntriesByNumber)
|
|
}
|
|
for _, entry := range partEntriesByNumber {
|
|
if found {
|
|
deleteEntries = append(deleteEntries, entry)
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedPartEntryMismatch).Inc()
|
|
continue
|
|
}
|
|
|
|
// a part may carry manifest chunks (the filer folds oversized part
|
|
// copies); the offset rebase below needs flat chunks
|
|
removedManifests, flattenErr := s3a.flattenManifestChunks(r.Context(), entry)
|
|
if flattenErr != nil {
|
|
glog.Errorf("completeMultipartUpload %s %s part %d resolve manifest chunks: %v", *input.Bucket, *input.UploadId, partNumber, flattenErr)
|
|
return nil, nil, s3err.ErrInternalError
|
|
}
|
|
supersededPartManifests = append(supersededPartManifests, removedManifests...)
|
|
|
|
partStartChunk := len(finalParts)
|
|
partStartOffset := offset
|
|
partETag := getEtagFromEntry(entry)
|
|
|
|
for _, chunk := range entry.GetChunks() {
|
|
finalChunk, chunkErr := completedMultipartChunk(chunk, offset, sses3Info)
|
|
if chunkErr != nil {
|
|
glog.Errorf("completeMultipartUpload %s %s SSE-S3 chunk metadata error: %v", *input.Bucket, *input.UploadId, chunkErr)
|
|
return nil, nil, s3err.ErrInternalError
|
|
}
|
|
finalParts = append(finalParts, finalChunk)
|
|
offset += int64(chunk.Size)
|
|
}
|
|
|
|
partEndChunk := len(finalParts)
|
|
partBoundaries = append(partBoundaries, multipartPartBoundary{
|
|
PartNumber: partNumber,
|
|
StartChunk: partStartChunk,
|
|
EndChunk: partEndChunk,
|
|
ETag: partETag,
|
|
StartOffset: partStartOffset,
|
|
EndOffset: offset,
|
|
})
|
|
|
|
found = true
|
|
}
|
|
}
|
|
|
|
// Compute the object checksum from per-part checksums if the upload was
|
|
// initiated with a checksum algorithm (stored in the upload dir entry).
|
|
// The checksum type (COMPOSITE or FULL_OBJECT) is resolved from the
|
|
// x-amz-checksum-type header captured at CreateMultipartUpload
|
|
checksumHeaderName := ""
|
|
checksumType := ""
|
|
checksumValue := ""
|
|
if pentry.Extended != nil {
|
|
if algoName, ok := pentry.Extended[s3_constants.ExtChecksumAlgorithm]; ok {
|
|
checksumHeaderName = string(algoName)
|
|
}
|
|
}
|
|
if checksumHeaderName != "" {
|
|
algo := checksumAlgorithmFromHeaderName(checksumHeaderName)
|
|
requestedType := ""
|
|
if pentry.Extended != nil {
|
|
requestedType = string(pentry.Extended[s3_constants.ExtChecksumType])
|
|
}
|
|
resolvedType, typeErr := resolveMultipartChecksumType(algo, requestedType)
|
|
if typeErr != nil {
|
|
glog.Errorf("completeMultipartUpload: %v", typeErr)
|
|
return nil, nil, s3err.ErrInvalidRequest
|
|
}
|
|
query := parseRequestQuery(r)
|
|
if completeType := lookupHeaderOrQuery(r, query, s3_constants.AmzChecksumType); completeType != "" && !strings.EqualFold(completeType, resolvedType) {
|
|
return nil, nil, s3err.ErrBadDigest
|
|
}
|
|
checksumType = resolvedType
|
|
|
|
provided := make(map[int]string, len(parts.Parts))
|
|
for _, part := range parts.Parts {
|
|
provided[part.PartNumber] = part.GetChecksum(checksumHeaderName)
|
|
}
|
|
for _, partNumber := range completedPartNumbers {
|
|
providedChecksum := provided[partNumber]
|
|
if providedChecksum == "" {
|
|
// COMPOSITE uploads must carry every part checksum in the
|
|
// request; FULL_OBJECT uploads instead carry the whole-object
|
|
// checksum in a request header.
|
|
if checksumType == s3_constants.ChecksumTypeComposite {
|
|
return nil, nil, s3err.ErrInvalidRequest
|
|
}
|
|
continue
|
|
}
|
|
raw, _, decodeErr := decodePartChecksum(partNumber, partEntries[partNumber], checksumHeaderName)
|
|
if decodeErr != nil {
|
|
glog.Errorf("completeMultipartUpload: %v", decodeErr)
|
|
return nil, nil, s3err.ErrInvalidPart
|
|
}
|
|
if providedChecksum != base64.StdEncoding.EncodeToString(raw) {
|
|
return nil, nil, s3err.ErrBadDigest
|
|
}
|
|
}
|
|
|
|
var checksumErr error
|
|
if checksumType == s3_constants.ChecksumTypeFullObject {
|
|
checksumValue, checksumErr = computeFullObjectChecksum(checksumHeaderName, partEntries, completedPartNumbers)
|
|
} else {
|
|
checksumValue, checksumErr = computeCompositeChecksum(checksumHeaderName, partEntries, completedPartNumbers)
|
|
}
|
|
if checksumErr != nil {
|
|
glog.Errorf("completeMultipartUpload: %s checksum computation failed: %v", checksumType, checksumErr)
|
|
return nil, nil, s3err.ErrInvalidPart
|
|
}
|
|
if objectChecksum := lookupHeaderOrQuery(r, query, checksumHeaderName); objectChecksum != "" && objectChecksum != checksumValue {
|
|
return nil, nil, s3err.ErrBadDigest
|
|
}
|
|
}
|
|
|
|
// Fold after the boundaries captured byte offsets; finalParts is flat
|
|
// here, so anything folded out is newly created.
|
|
finalParts = s3a.manifestizeChunks(dirName+"/"+entryName, *input.Bucket, 0, finalParts)
|
|
newManifestChunks, _ := filer.SeparateManifestChunks(finalParts)
|
|
|
|
return &multipartCompletionState{
|
|
deleteEntries: deleteEntries,
|
|
partEntries: partEntries,
|
|
pentry: pentry,
|
|
sses3Info: sses3Info,
|
|
mime: mime,
|
|
finalParts: finalParts,
|
|
offset: offset,
|
|
partBoundaries: partBoundaries,
|
|
multipartETag: calculateMultipartETag(partEntries, completedPartNumbers),
|
|
entityWithTtl: entityWithTtl,
|
|
checksumHeaderName: checksumHeaderName,
|
|
checksumValue: checksumValue,
|
|
checksumType: checksumType,
|
|
newManifestChunks: newManifestChunks,
|
|
supersededPartManifests: supersededPartManifests,
|
|
}, nil, s3err.ErrNone
|
|
}
|
|
|
|
func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.CompleteMultipartUploadInput, parts *CompleteMultipartUpload) (output *CompleteMultipartUploadResult, code s3err.ErrorCode) {
|
|
|
|
glog.V(2).Infof("completeMultipartUpload input %v", input)
|
|
if len(parts.Parts) == 0 {
|
|
stats.S3HandlerCounter.WithLabelValues(stats.ErrorCompletedNoSuchUpload).Inc()
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
completedPartNumbers := []int{}
|
|
completedPartMap := make(map[int][]string)
|
|
|
|
maxPartNo := 1
|
|
lastSeenPartNo := 0
|
|
|
|
for _, part := range parts.Parts {
|
|
if part.PartNumber < lastSeenPartNo {
|
|
return nil, s3err.ErrInvalidPartOrder
|
|
}
|
|
lastSeenPartNo = part.PartNumber
|
|
if _, ok := completedPartMap[part.PartNumber]; !ok {
|
|
completedPartNumbers = append(completedPartNumbers, part.PartNumber)
|
|
}
|
|
completedPartMap[part.PartNumber] = append(completedPartMap[part.PartNumber], part.ETag)
|
|
maxPartNo = maxInt(maxPartNo, part.PartNumber)
|
|
}
|
|
|
|
uploadDirectory := s3a.genUploadsFolder(*input.Bucket) + "/" + *input.UploadId
|
|
entryName, dirName := s3a.getEntryNameAndDir(input)
|
|
var completionState *multipartCompletionState
|
|
// Route the completion's writes to the object's owner filer when known, off
|
|
// the distributed lock. Idempotent replay is handled gateway-side in
|
|
// prepareMultipartCompletionState (it returns the existing result when the
|
|
// object already carries this UploadId), so the lock is not needed to dedupe
|
|
// retries. With no owner yet (no ring), keep the lock as the bootstrap path.
|
|
owner := s3a.objectWriteOwner(*input.Bucket, *input.Key)
|
|
routeKey := s3a.objectRouteKey(*input.Bucket, *input.Key)
|
|
completionBody := func() s3err.ErrorCode {
|
|
var prepCode s3err.ErrorCode
|
|
completionState, output, prepCode = s3a.prepareMultipartCompletionState(r, input, parts, uploadDirectory, entryName, dirName, completedPartNumbers, completedPartMap, maxPartNo)
|
|
if prepCode != s3err.ErrNone || output != nil {
|
|
return prepCode
|
|
}
|
|
|
|
etagQuote := "\"" + completionState.multipartETag + "\""
|
|
// Check if versioning is configured for this bucket BEFORE creating any files.
|
|
versioningState, vErr := s3a.getVersioningState(*input.Bucket)
|
|
if vErr != nil {
|
|
glog.Errorf("completeMultipartUpload: failed to get versioning state for bucket %s: %v", *input.Bucket, vErr)
|
|
return s3err.ErrInternalError
|
|
}
|
|
// Full object key, not just entryName, so the right .versions directory is used.
|
|
normalizedKey := s3_constants.NormalizeObjectKey(*input.Key)
|
|
|
|
if versioningState == s3_constants.VersioningEnabled {
|
|
useInvertedFormat := s3a.getVersionIdFormat(*input.Bucket, normalizedKey)
|
|
versionId := generateVersionId(useInvertedFormat)
|
|
versionFileName := s3a.getVersionFileName(versionId)
|
|
versionDir := dirName + "/" + entryName + s3_constants.VersionsFolder
|
|
|
|
// Capture timestamp and owner once for consistency between version entry and cache entry
|
|
versionMtime := time.Now().Unix()
|
|
amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
|
|
|
|
// Fills in the version entry; on the routed path this runs inside the
|
|
// finalize transaction, on the fallback inside mkFile.
|
|
decorateVersionEntry := func(versionEntry *filer_pb.Entry) {
|
|
if versionEntry.Extended == nil {
|
|
versionEntry.Extended = make(map[string][]byte)
|
|
}
|
|
versionEntry.Extended[s3_constants.ExtVersionIdKey] = []byte(versionId)
|
|
versionEntry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId)
|
|
// Store parts count for x-amz-mp-parts-count header
|
|
versionEntry.Extended[s3_constants.SeaweedFSMultipartPartsCount] = []byte(fmt.Sprintf("%d", len(completedPartNumbers)))
|
|
// Store part boundaries for GetObject with PartNumber
|
|
if partBoundariesJSON, err := json.Marshal(completionState.partBoundaries); err == nil {
|
|
versionEntry.Extended[s3_constants.SeaweedFSMultipartPartBoundaries] = partBoundariesJSON
|
|
}
|
|
|
|
// Set object owner for versioned multipart objects
|
|
if amzAccountId != "" {
|
|
versionEntry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
|
|
}
|
|
|
|
for k, v := range completionState.pentry.Extended {
|
|
if k != s3_constants.ExtMultipartObjectKey {
|
|
versionEntry.Extended[k] = v
|
|
}
|
|
}
|
|
|
|
// Persist ETag to ensure subsequent HEAD/GET uses the same value
|
|
versionEntry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
|
|
// Store the object checksum computed from per-part checksums
|
|
if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
|
|
versionEntry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
|
|
versionEntry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
|
|
if completionState.checksumType != "" {
|
|
versionEntry.Extended[s3_constants.ExtChecksumType] = []byte(completionState.checksumType)
|
|
}
|
|
}
|
|
|
|
// Preserve ALL SSE metadata from the first part (if any)
|
|
// SSE metadata is stored in individual parts, not the upload directory
|
|
if len(completedPartNumbers) > 0 && len(completionState.partEntries[completedPartNumbers[0]]) > 0 {
|
|
firstPartEntry := completionState.partEntries[completedPartNumbers[0]][0]
|
|
copySSEHeadersFromFirstPart(versionEntry, firstPartEntry, "versioned")
|
|
}
|
|
applyMultipartSSES3HeadersFromUploadEntry(versionEntry, completionState.sses3Info)
|
|
if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
|
|
versionEntry.Attributes.Mime = completionState.pentry.Attributes.Mime
|
|
} else if completionState.mime != "" {
|
|
versionEntry.Attributes.Mime = completionState.mime
|
|
}
|
|
versionEntry.Attributes.FileSize = uint64(completionState.offset)
|
|
versionEntry.Attributes.Mtime = versionMtime
|
|
}
|
|
|
|
// Construct entry with metadata for caching in .versions directory
|
|
// Reuse versionMtime to keep list vs. HEAD timestamps aligned
|
|
// multipartETag is precomputed
|
|
versionEntryForCache := &filer_pb.Entry{
|
|
Attributes: &filer_pb.FuseAttributes{
|
|
FileSize: uint64(completionState.offset),
|
|
Mtime: versionMtime,
|
|
},
|
|
Extended: map[string][]byte{
|
|
s3_constants.ExtETagKey: []byte(completionState.multipartETag),
|
|
},
|
|
}
|
|
if amzAccountId != "" {
|
|
versionEntryForCache.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
|
|
}
|
|
|
|
// objectTxnOnFiler needs an owner to route to or a filerClient to
|
|
// pick a filer from; only the bootstrap fallback keeps the
|
|
// mkFile + pointer-update sequence.
|
|
if owner != "" || s3a.filerClient != nil {
|
|
// The transaction removes the upload directory metadata-only,
|
|
// so the part entries the object does not reference are freed
|
|
// first or their chunks leak.
|
|
if err := s3a.deleteUnusedPartEntries(r.Context(), uploadDirectory, *input.Bucket, *input.UploadId, completionState); err != nil {
|
|
glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
if code := s3a.routedMultipartFinalize(owner, *input.Bucket, *input.Key, useInvertedFormat, versionDir, versionFileName, completionState.finalParts, decorateVersionEntry, *input.UploadId); code != s3err.ErrNone {
|
|
if code == s3err.ErrNoSuchUpload {
|
|
return code
|
|
}
|
|
// Roll back only while the upload directory survives: once the
|
|
// transaction removed it, the version file is the only record
|
|
// left and deleting it would make a retry impossible.
|
|
if exists, _ := s3a.exists(s3a.genUploadsFolder(*input.Bucket), *input.UploadId, true); exists {
|
|
if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil {
|
|
glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after routed finalize error: %v", versionId, *input.Bucket, *input.Key, rollbackErr)
|
|
completionState.manifestsReferenced = true
|
|
}
|
|
} else {
|
|
completionState.manifestsReferenced = true
|
|
}
|
|
return code
|
|
}
|
|
completionState.uploadDirRemoved = true
|
|
} else {
|
|
if err := s3a.mkFile(versionDir, versionFileName, completionState.finalParts, decorateVersionEntry); err != nil {
|
|
glog.Errorf("completeMultipartUpload: failed to create version %s: %v", versionId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
if err := s3a.updateLatestVersionInDirectory(*input.Bucket, *input.Key, versionId, versionFileName, versionEntryForCache); err != nil {
|
|
if rollbackErr := s3a.rollbackMultipartVersion(versionDir, versionFileName); rollbackErr != nil {
|
|
glog.Errorf("completeMultipartUpload: failed to rollback version %s for %s/%s after latest pointer update error: %v", versionId, *input.Bucket, *input.Key, rollbackErr)
|
|
completionState.manifestsReferenced = true
|
|
}
|
|
glog.Errorf("completeMultipartUpload: failed to update latest version in directory: %v", err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
}
|
|
|
|
// For versioned buckets, all content is stored in .versions directory
|
|
// The latest version information is tracked in the .versions directory metadata
|
|
output = &CompleteMultipartUploadResult{
|
|
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
|
|
Bucket: input.Bucket,
|
|
ETag: aws.String(etagQuote),
|
|
Key: objectKey(input.Key),
|
|
VersionId: aws.String(versionId),
|
|
ChecksumType: completionState.checksumType,
|
|
}
|
|
output.SetChecksum(completionState.checksumHeaderName, completionState.checksumValue)
|
|
return s3err.ErrNone
|
|
}
|
|
|
|
if versioningState == s3_constants.VersioningSuspended {
|
|
// For suspended versioning, add "null" version ID metadata and return "null" version ID
|
|
removal, err := s3a.routedUploadRemoval(r.Context(), owner, uploadDirectory, *input.Bucket, *input.UploadId, completionState)
|
|
if err != nil {
|
|
glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
|
|
if entry.Extended == nil {
|
|
entry.Extended = make(map[string][]byte)
|
|
}
|
|
entry.Extended[s3_constants.ExtVersionIdKey] = []byte("null")
|
|
entry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId)
|
|
// Store parts count for x-amz-mp-parts-count header
|
|
entry.Extended[s3_constants.SeaweedFSMultipartPartsCount] = []byte(fmt.Sprintf("%d", len(completedPartNumbers)))
|
|
// Store part boundaries for GetObject with PartNumber
|
|
if partBoundariesJSON, jsonErr := json.Marshal(completionState.partBoundaries); jsonErr == nil {
|
|
entry.Extended[s3_constants.SeaweedFSMultipartPartBoundaries] = partBoundariesJSON
|
|
}
|
|
|
|
// Set object owner for suspended versioning multipart objects
|
|
amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
|
|
if amzAccountId != "" {
|
|
entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
|
|
}
|
|
|
|
for k, v := range completionState.pentry.Extended {
|
|
if k != s3_constants.ExtMultipartObjectKey {
|
|
entry.Extended[k] = v
|
|
}
|
|
}
|
|
|
|
// Preserve ALL SSE metadata from the first part (if any)
|
|
// SSE metadata is stored in individual parts, not the upload directory
|
|
if len(completedPartNumbers) > 0 && len(completionState.partEntries[completedPartNumbers[0]]) > 0 {
|
|
firstPartEntry := completionState.partEntries[completedPartNumbers[0]][0]
|
|
copySSEHeadersFromFirstPart(entry, firstPartEntry, "suspended versioning")
|
|
}
|
|
applyMultipartSSES3HeadersFromUploadEntry(entry, completionState.sses3Info)
|
|
// Persist ETag to ensure subsequent HEAD/GET uses the same value
|
|
entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
|
|
// Store the object checksum computed from per-part checksums
|
|
if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
|
|
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
|
|
entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
|
|
if completionState.checksumType != "" {
|
|
entry.Extended[s3_constants.ExtChecksumType] = []byte(completionState.checksumType)
|
|
}
|
|
}
|
|
if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
|
|
entry.Attributes.Mime = completionState.pentry.Attributes.Mime
|
|
} else if completionState.mime != "" {
|
|
entry.Attributes.Mime = completionState.mime
|
|
}
|
|
entry.Attributes.FileSize = uint64(completionState.offset)
|
|
}, removal); err != nil {
|
|
if errors.Is(err, errUploadRemoved) {
|
|
return s3err.ErrNoSuchUpload
|
|
}
|
|
// The transaction may have committed the object before failing on
|
|
// the upload removal; a surviving entry references the manifests.
|
|
if exists, err := s3a.exists(dirName, entryName, false); err != nil || exists {
|
|
completionState.manifestsReferenced = true
|
|
}
|
|
glog.Errorf("completeMultipartUpload: failed to create suspended versioning object: %v", err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
completionState.uploadDirRemoved = removal != nil
|
|
|
|
// A failed finalize leaves the key reading as deleted, so fail rather than
|
|
// return 200 — a non-ErrNone finalize keeps the upload directory, so the
|
|
// caller's retry replays.
|
|
if err := s3a.finalizeSuspendedNullWrite(owner, *input.Bucket, normalizedKey, s3_constants.SeaweedFSUploadId, *input.UploadId); err != nil {
|
|
completionState.manifestsReferenced = true
|
|
glog.Errorf("completeMultipartUpload: failed to retire the null delete marker for %s/%s: %v", *input.Bucket, normalizedKey, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
|
|
// Note: Suspended versioning should NOT return VersionId field according to AWS S3 spec
|
|
output = &CompleteMultipartUploadResult{
|
|
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
|
|
Bucket: input.Bucket,
|
|
ETag: aws.String(etagQuote),
|
|
Key: objectKey(input.Key),
|
|
ChecksumType: completionState.checksumType,
|
|
// VersionId field intentionally omitted for suspended versioning
|
|
}
|
|
output.SetChecksum(completionState.checksumHeaderName, completionState.checksumValue)
|
|
return s3err.ErrNone
|
|
}
|
|
|
|
// For non-versioned buckets, create main object file
|
|
removal, err := s3a.routedUploadRemoval(r.Context(), owner, uploadDirectory, *input.Bucket, *input.UploadId, completionState)
|
|
if err != nil {
|
|
glog.Errorf("completeMultipartUpload %s upload %s unused part cleanup: %v", *input.Bucket, *input.UploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
if err := s3a.writeMultipartObject(owner, routeKey, dirName, entryName, completionState.finalParts, func(entry *filer_pb.Entry) {
|
|
if entry.Extended == nil {
|
|
entry.Extended = make(map[string][]byte)
|
|
}
|
|
entry.Extended[s3_constants.SeaweedFSUploadId] = []byte(*input.UploadId)
|
|
// Store parts count for x-amz-mp-parts-count header
|
|
entry.Extended[s3_constants.SeaweedFSMultipartPartsCount] = []byte(fmt.Sprintf("%d", len(completedPartNumbers)))
|
|
// Store part boundaries for GetObject with PartNumber
|
|
if partBoundariesJSON, err := json.Marshal(completionState.partBoundaries); err == nil {
|
|
entry.Extended[s3_constants.SeaweedFSMultipartPartBoundaries] = partBoundariesJSON
|
|
}
|
|
|
|
// Set object owner for non-versioned multipart objects
|
|
amzAccountId := r.Header.Get(s3_constants.AmzAccountId)
|
|
if amzAccountId != "" {
|
|
entry.Extended[s3_constants.ExtAmzOwnerKey] = []byte(amzAccountId)
|
|
}
|
|
|
|
for k, v := range completionState.pentry.Extended {
|
|
if k != s3_constants.ExtMultipartObjectKey {
|
|
entry.Extended[k] = v
|
|
}
|
|
}
|
|
|
|
// Preserve ALL SSE metadata from the first part (if any)
|
|
// SSE metadata is stored in individual parts, not the upload directory
|
|
if len(completedPartNumbers) > 0 && len(completionState.partEntries[completedPartNumbers[0]]) > 0 {
|
|
firstPartEntry := completionState.partEntries[completedPartNumbers[0]][0]
|
|
copySSEHeadersFromFirstPart(entry, firstPartEntry, "non-versioned")
|
|
}
|
|
applyMultipartSSES3HeadersFromUploadEntry(entry, completionState.sses3Info)
|
|
// Persist ETag to ensure subsequent HEAD/GET uses the same value
|
|
entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag)
|
|
// Store the object checksum computed from per-part checksums
|
|
if completionState.checksumHeaderName != "" && completionState.checksumValue != "" {
|
|
entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName)
|
|
entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue)
|
|
if completionState.checksumType != "" {
|
|
entry.Extended[s3_constants.ExtChecksumType] = []byte(completionState.checksumType)
|
|
}
|
|
}
|
|
if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" {
|
|
entry.Attributes.Mime = completionState.pentry.Attributes.Mime
|
|
} else if completionState.mime != "" {
|
|
entry.Attributes.Mime = completionState.mime
|
|
}
|
|
entry.Attributes.FileSize = uint64(completionState.offset)
|
|
// Set TTL-based S3 expiry (modification time)
|
|
if completionState.entityWithTtl {
|
|
entry.Extended[s3_constants.SeaweedFSExpiresS3] = []byte("true")
|
|
}
|
|
}, removal); err != nil {
|
|
if errors.Is(err, errUploadRemoved) {
|
|
return s3err.ErrNoSuchUpload
|
|
}
|
|
// The transaction may have committed the object before failing on
|
|
// the upload removal; a surviving entry references the manifests.
|
|
if exists, err := s3a.exists(dirName, entryName, false); err != nil || exists {
|
|
completionState.manifestsReferenced = true
|
|
}
|
|
glog.Errorf("completeMultipartUpload %s/%s error: %v", dirName, entryName, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
completionState.uploadDirRemoved = removal != nil
|
|
|
|
// For non-versioned buckets, return response without VersionId
|
|
output = &CompleteMultipartUploadResult{
|
|
Location: aws.String(fmt.Sprintf("%s://%s/%s/%s", getRequestScheme(r), r.Host, url.PathEscape(*input.Bucket), urlPathEscape(*input.Key))),
|
|
Bucket: input.Bucket,
|
|
ETag: aws.String(etagQuote),
|
|
Key: objectKey(input.Key),
|
|
ChecksumType: completionState.checksumType,
|
|
}
|
|
output.SetChecksum(completionState.checksumHeaderName, completionState.checksumValue)
|
|
return s3err.ErrNone
|
|
}
|
|
var finalizeCode s3err.ErrorCode
|
|
if owner != "" {
|
|
if code := s3a.checkConditionalHeaders(r, *input.Bucket, *input.Key); code != s3err.ErrNone {
|
|
finalizeCode = code
|
|
} else {
|
|
finalizeCode = completionBody()
|
|
}
|
|
} else {
|
|
finalizeCode = s3a.withObjectWriteLock(*input.Bucket, *input.Key, func() s3err.ErrorCode {
|
|
return s3a.checkConditionalHeaders(r, *input.Bucket, *input.Key)
|
|
}, completionBody)
|
|
}
|
|
if finalizeCode != s3err.ErrNone {
|
|
// this attempt's manifests are orphans unless a failed rollback left an entry holding them
|
|
if completionState != nil && !completionState.manifestsReferenced && len(completionState.newManifestChunks) > 0 {
|
|
s3a.deleteOrphanedChunks(completionState.newManifestChunks)
|
|
}
|
|
return nil, finalizeCode
|
|
}
|
|
|
|
if completionState != nil {
|
|
if !completionState.uploadDirRemoved {
|
|
// The object is already committed and the client is still waiting, so the
|
|
// cleanup below runs on its own context but spends one allowance between
|
|
// all of it rather than a retry backoff per unused entry.
|
|
cleanupCtx := withFilerRetryBudget(context.Background(), filerRetryRequestBudget)
|
|
// Keep the directory when an entry delete failed so its metadata
|
|
// still references the chunks; removing it metadata-only orphans them.
|
|
if err := s3a.deleteUnusedPartEntries(cleanupCtx, uploadDirectory, *input.Bucket, *input.UploadId, completionState); err != nil {
|
|
glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
|
|
} else if err := s3a.rm(cleanupCtx, s3a.genUploadsFolder(*input.Bucket), *input.UploadId, false, true); err != nil {
|
|
glog.V(1).Infof("completeMultipartUpload cleanup %s upload %s: %v", *input.Bucket, *input.UploadId, err)
|
|
}
|
|
}
|
|
if len(completionState.supersededPartManifests) > 0 {
|
|
s3a.deleteOrphanedChunks(completionState.supersededPartManifests)
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
// deleteUnusedPartEntries frees the part entries the completed object does not
|
|
// reference. A finalize that removes the whole upload directory metadata-only
|
|
// frees them first, or their chunks leak; a failure here must abort the
|
|
// completion, since the surviving entries keep the upload retriable.
|
|
func (s3a *S3ApiServer) deleteUnusedPartEntries(ctx context.Context, uploadDirectory, bucket, uploadId string, completionState *multipartCompletionState) error {
|
|
var lastErr error
|
|
for _, deleteEntry := range completionState.deleteEntries {
|
|
if err := s3a.rm(ctx, uploadDirectory, deleteEntry.Name, !completionState.metadataOnlyCleanup, true); err != nil {
|
|
glog.Warningf("completeMultipartUpload cleanup %s upload %s unused %s : %v", bucket, uploadId, deleteEntry.Name, err)
|
|
lastErr = err
|
|
}
|
|
}
|
|
return lastErr
|
|
}
|
|
|
|
// resumeCommittedObject finds an entry this upload already committed at the
|
|
// regular path when the upload directory is gone. Suspended versioning can
|
|
// leave such an entry hidden behind a delete marker the finalize failed to
|
|
// retire, so re-running it repairs the key and lets the retry succeed. Any
|
|
// other versioning state means a newer write owns the key and the marker must
|
|
// not be demoted.
|
|
func (s3a *S3ApiServer) resumeCommittedObject(r *http.Request, input *s3.CompleteMultipartUploadInput, dirName, entryName string) (*CompleteMultipartUploadResult, s3err.ErrorCode) {
|
|
entry, err := s3a.getEntry(dirName, entryName)
|
|
if err != nil {
|
|
if isFilerNotFound(err) {
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
if entry == nil || string(entry.Extended[s3_constants.SeaweedFSUploadId]) != *input.UploadId {
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
state, err := s3a.getVersioningState(*input.Bucket)
|
|
if err != nil {
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
if state != s3_constants.VersioningSuspended {
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
if err := s3a.finalizeSuspendedNullWrite(s3a.objectWriteOwner(*input.Bucket, *input.Key), *input.Bucket, s3_constants.NormalizeObjectKey(*input.Key), s3_constants.SeaweedFSUploadId, *input.UploadId); err != nil {
|
|
glog.Errorf("completeMultipartUpload: failed to retire the null delete marker for %s/%s: %v", *input.Bucket, *input.Key, err)
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
return completeMultipartResult(r, input, getEtagFromEntry(entry), entry), s3err.ErrNone
|
|
}
|
|
|
|
// Metadata-only: the version file's chunks are the still-registered parts'
|
|
// chunks, which a retried completion needs.
|
|
func (s3a *S3ApiServer) rollbackMultipartVersion(versionDir, versionFileName string) error {
|
|
return s3a.rmObject(context.Background(), versionDir, versionFileName, false, false)
|
|
}
|
|
|
|
func (s3a *S3ApiServer) getEntryNameAndDir(input *s3.CompleteMultipartUploadInput) (string, string) {
|
|
entryName := path.Base(*input.Key)
|
|
dirName := path.Dir(*input.Key)
|
|
if dirName == "." {
|
|
dirName = ""
|
|
}
|
|
dirName = strings.TrimPrefix(dirName, "/")
|
|
dirName = fmt.Sprintf("%s/%s", s3a.bucketDir(*input.Bucket), dirName)
|
|
|
|
// remove suffix '/'
|
|
dirName = strings.TrimSuffix(dirName, "/")
|
|
return entryName, dirName
|
|
}
|
|
|
|
func parsePartNumber(fileName string) (int, error) {
|
|
var partNumberString string
|
|
index := strings.Index(fileName, "_")
|
|
if index != -1 {
|
|
partNumberString = fileName[:index]
|
|
} else {
|
|
partNumberString = fileName[:len(fileName)-len(multipartExt)]
|
|
}
|
|
return strconv.Atoi(partNumberString)
|
|
}
|
|
|
|
func (s3a *S3ApiServer) abortMultipartUpload(input *s3.AbortMultipartUploadInput) (output *s3.AbortMultipartUploadOutput, code s3err.ErrorCode) {
|
|
|
|
glog.V(2).Infof("abortMultipartUpload input %v", input)
|
|
|
|
uploadsFolder := s3a.genUploadsFolder(*input.Bucket)
|
|
uploadEntry, err := s3a.getEntry(uploadsFolder, *input.UploadId)
|
|
if err != nil {
|
|
if isFilerNotFound(err) {
|
|
return &s3.AbortMultipartUploadOutput{}, s3err.ErrNone
|
|
}
|
|
glog.Errorf("bucket %s abort upload %s: %v", *input.Bucket, *input.UploadId, err)
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
if uploadEntry == nil {
|
|
return &s3.AbortMultipartUploadOutput{}, s3err.ErrNone
|
|
}
|
|
|
|
object := string(uploadEntry.Extended[s3_constants.ExtMultipartObjectKey])
|
|
if object == "" && input.Key != nil {
|
|
object = *input.Key
|
|
}
|
|
return &s3.AbortMultipartUploadOutput{}, s3a.withObjectWriteLock(*input.Bucket, object, nil, func() s3err.ErrorCode {
|
|
return s3a.removeUploadDir(*input.Bucket, *input.UploadId, object)
|
|
})
|
|
}
|
|
|
|
// removeUploadDir removes a leftover upload directory under the object write
|
|
// lock: a directory that outlived the object it completed into shares chunks
|
|
// with it and goes metadata-only; anything else frees the parts' chunks.
|
|
func (s3a *S3ApiServer) removeUploadDir(bucket, uploadId, object string) s3err.ErrorCode {
|
|
completed, err := s3a.uploadCompleted(bucket, uploadId, object)
|
|
if err != nil {
|
|
glog.Errorf("bucket %s remove upload %s completed check: %v", bucket, uploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
if completed {
|
|
if err := s3a.rm(context.Background(), s3a.genUploadsFolder(bucket), uploadId, false, true); err != nil {
|
|
glog.V(1).Infof("bucket %s remove upload %s: %v", bucket, uploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
return s3err.ErrNone
|
|
}
|
|
return s3a.routedUploadDirDelete(bucket, uploadId, object)
|
|
}
|
|
|
|
// routedUploadDirDelete frees an open upload's chunks. The delete rides an
|
|
// ObjectTransaction on the object's lock key, so a racing routed commit either
|
|
// fails its upload-exists precondition afterwards or has already stamped the
|
|
// object — the condition then rejects the data delete and it falls back to
|
|
// metadata-only. With no owner there is no routed commit to exclude, and the
|
|
// caller's object write lock already serializes the mkFile fallback.
|
|
func (s3a *S3ApiServer) routedUploadDirDelete(bucket, uploadId, object string) s3err.ErrorCode {
|
|
uploadsFolder := s3a.genUploadsFolder(bucket)
|
|
rmUploadDir := func(isDeleteData bool) s3err.ErrorCode {
|
|
if err := s3a.rm(context.Background(), uploadsFolder, uploadId, isDeleteData, true); err != nil {
|
|
glog.V(1).Infof("bucket %s remove upload %s: %v", bucket, uploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
return s3err.ErrNone
|
|
}
|
|
if object == "" {
|
|
return rmUploadDir(true)
|
|
}
|
|
owner := s3a.objectWriteOwner(bucket, object)
|
|
if owner == "" {
|
|
return rmUploadDir(true)
|
|
}
|
|
objectPath := s3a.toFilerPath(bucket, object)
|
|
resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{
|
|
LockKey: objectPath,
|
|
RouteKey: s3a.objectRouteKey(bucket, object),
|
|
ConditionKey: objectPath,
|
|
Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{
|
|
Kind: filer_pb.WriteCondition_IF_EXTENDED_NOT_EQUAL,
|
|
ExtKey: s3_constants.SeaweedFSUploadId,
|
|
ExtValue: uploadId,
|
|
}}},
|
|
Mutations: []*filer_pb.ObjectMutation{{
|
|
Type: filer_pb.ObjectMutation_DELETE,
|
|
Directory: uploadsFolder,
|
|
Name: uploadId,
|
|
IsDeleteData: true,
|
|
IsRecursive: true,
|
|
}},
|
|
})
|
|
if err != nil {
|
|
glog.Errorf("bucket %s abort upload %s transaction: %v", bucket, uploadId, err)
|
|
return s3err.ErrInternalError
|
|
}
|
|
if resp.ErrorCode == filer_pb.FilerError_PRECONDITION_FAILED {
|
|
return rmUploadDir(false)
|
|
}
|
|
if resp.Error != "" {
|
|
glog.Errorf("bucket %s abort upload %s transaction: %s", bucket, uploadId, resp.Error)
|
|
return s3err.ErrInternalError
|
|
}
|
|
return s3err.ErrNone
|
|
}
|
|
|
|
// uploadCompleted reports whether the upload assembled into an object: the
|
|
// object entry, or any version file under <key>.versions, still carries the
|
|
// upload id completion stamps on it.
|
|
func (s3a *S3ApiServer) uploadCompleted(bucket, uploadId, objectKey string) (bool, error) {
|
|
if objectKey == "" {
|
|
return false, nil
|
|
}
|
|
bucketDir := s3a.bucketDir(bucket)
|
|
name := path.Base(objectKey)
|
|
dir := path.Dir(objectKey)
|
|
if dir == "." {
|
|
dir = ""
|
|
}
|
|
objectDir := path.Join(bucketDir, dir)
|
|
|
|
entry, err := s3a.getObjectEntryRoutedByKey(bucket, objectKey)
|
|
if err != nil && !isFilerNotFound(err) {
|
|
return false, err
|
|
}
|
|
if entry != nil && string(entry.Extended[s3_constants.SeaweedFSUploadId]) == uploadId {
|
|
return true, nil
|
|
}
|
|
|
|
versions, _, err := s3a.listObjectScopedEntriesRoutedByKey(bucket, objectKey, objectDir+"/"+name+s3_constants.VersionsFolder, "", "", false, math.MaxInt32)
|
|
if err != nil {
|
|
if isFilerNotFound(err) {
|
|
return false, nil
|
|
}
|
|
return false, err
|
|
}
|
|
for _, version := range versions {
|
|
if string(version.Extended[s3_constants.SeaweedFSUploadId]) == uploadId {
|
|
return true, nil
|
|
}
|
|
}
|
|
return false, nil
|
|
}
|
|
|
|
type ListMultipartUploadsResult struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListMultipartUploadsResult"`
|
|
|
|
// copied from s3.ListMultipartUploadsOutput, the Uploads is not converting to <Upload></Upload>
|
|
Bucket *string `type:"string"`
|
|
Delimiter *string `type:"string"`
|
|
EncodingType *string `type:"string" enum:"EncodingType"`
|
|
IsTruncated *bool `type:"boolean"`
|
|
KeyMarker *string `type:"string"`
|
|
MaxUploads *int64 `type:"integer"`
|
|
NextKeyMarker *string `type:"string"`
|
|
NextUploadIdMarker *string `type:"string"`
|
|
Prefix *string `type:"string"`
|
|
UploadIdMarker *string `type:"string"`
|
|
Upload []*s3.MultipartUpload `locationName:"Upload" type:"list" flattened:"true"`
|
|
CommonPrefixes []PrefixEntry `xml:"CommonPrefixes,omitempty"`
|
|
}
|
|
|
|
func (s3a *S3ApiServer) listMultipartUploads(input *s3.ListMultipartUploadsInput) (output *ListMultipartUploadsResult, code s3err.ErrorCode) {
|
|
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListMultipartUploads.html
|
|
|
|
glog.V(2).Infof("listMultipartUploads input %v", input)
|
|
|
|
uploadsFolder := s3a.genUploadsFolder(*input.Bucket)
|
|
var page *multipartUploadPage
|
|
err := listRetry(uploadsFolder, func() error {
|
|
page = newMultipartUploadPage(input)
|
|
return filer_pb.List(context.Background(), s3a, uploadsFolder, "", func(entry *filer_pb.Entry, isLastEntry bool) error {
|
|
page.add(entry)
|
|
return nil
|
|
}, "", false, math.MaxInt32)
|
|
})
|
|
if err != nil {
|
|
// A missing .uploads folder normally lists as empty with no error; a
|
|
// store that reports it as not-found still means an empty list.
|
|
if isFilerNotFound(err) {
|
|
return selectMultipartUploads(input, nil), s3err.ErrNone
|
|
}
|
|
// surface a real store error; a masked empty 200 makes a resuming client treat the upload as gone
|
|
glog.Errorf("listMultipartUploads %s error: %v", *input.Bucket, err)
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
|
|
return page.result(), s3err.ErrNone
|
|
}
|
|
|
|
// selectMultipartUploads orders uploads by key, then by upload id, and resumes
|
|
// after the (key-marker, upload-id-marker) position, so an aborted marker
|
|
// upload does not hide the rest of its key.
|
|
func selectMultipartUploads(input *s3.ListMultipartUploadsInput, entries []*filer_pb.Entry) *ListMultipartUploadsResult {
|
|
page := newMultipartUploadPage(input)
|
|
for _, entry := range entries {
|
|
page.add(entry)
|
|
}
|
|
return page.result()
|
|
}
|
|
|
|
// multipartUploadPage accumulates one page of a ListMultipartUploads result.
|
|
// Upload order is the object key kept in extended attributes, not the upload
|
|
// id the filer sorts by, so every entry has to be seen; keeping only the
|
|
// maxUploads smallest candidates bounds the heap by the page, not the folder.
|
|
// prefixes dedups common prefixes still in the page — an evicted one is
|
|
// unmarked; it could only lose again, so the map stays bounded by the heap.
|
|
type multipartUploadPage struct {
|
|
input *s3.ListMultipartUploadsInput
|
|
limit int
|
|
top candidateHeap
|
|
more bool
|
|
prefixes map[string]bool
|
|
}
|
|
|
|
// multipartCandidate is an upload, or with a nil entry a common prefix that
|
|
// groups the uploads whose keys continue past the delimiter.
|
|
type multipartCandidate struct {
|
|
key string
|
|
entry *filer_pb.Entry
|
|
}
|
|
|
|
func (c multipartCandidate) uploadId() string {
|
|
if c.entry == nil {
|
|
return ""
|
|
}
|
|
return c.entry.Name
|
|
}
|
|
|
|
func compareMultipartCandidate(a, b multipartCandidate) int {
|
|
return cmp.Or(strings.Compare(a.key, b.key), strings.Compare(a.uploadId(), b.uploadId()))
|
|
}
|
|
|
|
// candidateHeap is a max-heap on (key, upload id): the root sorts last of the
|
|
// candidates held, so popping it keeps the page's smallest entries.
|
|
type candidateHeap []multipartCandidate
|
|
|
|
func (h candidateHeap) Len() int { return len(h) }
|
|
func (h candidateHeap) Less(i, j int) bool {
|
|
return compareMultipartCandidate(h[i], h[j]) > 0
|
|
}
|
|
func (h candidateHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
|
|
func (h *candidateHeap) Push(x any) { *h = append(*h, x.(multipartCandidate)) }
|
|
func (h *candidateHeap) Pop() any {
|
|
old := *h
|
|
n := len(old)
|
|
*h = old[:n-1]
|
|
return old[n-1]
|
|
}
|
|
|
|
func newMultipartUploadPage(input *s3.ListMultipartUploadsInput) *multipartUploadPage {
|
|
limit := maxUploadsList
|
|
if *input.MaxUploads < int64(limit) {
|
|
limit = int(*input.MaxUploads)
|
|
}
|
|
return &multipartUploadPage{input: input, limit: limit, prefixes: map[string]bool{}}
|
|
}
|
|
|
|
func (p *multipartUploadPage) add(entry *filer_pb.Entry) {
|
|
if p.limit <= 0 || !isMultipartUploadEntry(entry) {
|
|
return
|
|
}
|
|
input := p.input
|
|
key := *objectKey(aws.String(string(entry.Extended[s3_constants.ExtMultipartObjectKey])))
|
|
if *input.Prefix != "" && !strings.HasPrefix(key, *input.Prefix) {
|
|
return
|
|
}
|
|
if *input.KeyMarker != "" && (key < *input.KeyMarker || key == *input.KeyMarker && (*input.UploadIdMarker == "" || entry.Name <= *input.UploadIdMarker)) {
|
|
return
|
|
}
|
|
c := multipartCandidate{key: key, entry: entry}
|
|
// with a delimiter, keys that go on past it after the prefix are listed
|
|
// once as their common prefix, which counts toward max-uploads
|
|
if delimiter := aws.StringValue(input.Delimiter); delimiter != "" {
|
|
if i := strings.Index(key[len(*input.Prefix):], delimiter); i >= 0 {
|
|
commonPrefix := key[:len(*input.Prefix)+i+len(delimiter)]
|
|
if p.prefixes[commonPrefix] || *input.KeyMarker != "" && commonPrefix <= *input.KeyMarker {
|
|
return
|
|
}
|
|
c = multipartCandidate{key: commonPrefix}
|
|
}
|
|
}
|
|
if p.top.Len() >= p.limit {
|
|
p.more = true
|
|
if compareMultipartCandidate(c, p.top[0]) >= 0 {
|
|
return
|
|
}
|
|
if evicted := heap.Pop(&p.top).(multipartCandidate); evicted.entry == nil {
|
|
delete(p.prefixes, evicted.key)
|
|
}
|
|
}
|
|
heap.Push(&p.top, c)
|
|
if c.entry == nil {
|
|
p.prefixes[c.key] = true
|
|
}
|
|
}
|
|
|
|
func (p *multipartUploadPage) result() *ListMultipartUploadsResult {
|
|
input := p.input
|
|
output := &ListMultipartUploadsResult{
|
|
Bucket: input.Bucket,
|
|
Delimiter: input.Delimiter,
|
|
KeyMarker: input.KeyMarker,
|
|
MaxUploads: aws.Int64(int64(p.limit)),
|
|
Prefix: input.Prefix,
|
|
UploadIdMarker: input.UploadIdMarker,
|
|
IsTruncated: aws.Bool(false),
|
|
}
|
|
candidates := make([]multipartCandidate, len(p.top))
|
|
copy(candidates, p.top)
|
|
slices.SortFunc(candidates, compareMultipartCandidate)
|
|
for _, c := range candidates {
|
|
if c.entry == nil {
|
|
output.CommonPrefixes = append(output.CommonPrefixes, PrefixEntry{Prefix: c.key})
|
|
continue
|
|
}
|
|
upload := &s3.MultipartUpload{
|
|
Key: aws.String(c.key),
|
|
UploadId: aws.String(c.entry.Name),
|
|
}
|
|
if c.entry.Attributes != nil {
|
|
upload.Initiated = aws.Time(time.Unix(c.entry.Attributes.Crtime, int64(c.entry.Attributes.CrtimeNs)))
|
|
}
|
|
output.Upload = append(output.Upload, upload)
|
|
}
|
|
if p.more {
|
|
last := candidates[len(candidates)-1]
|
|
output.IsTruncated = aws.Bool(true)
|
|
output.NextKeyMarker = aws.String(last.key)
|
|
if last.entry != nil {
|
|
output.NextUploadIdMarker = aws.String(last.entry.Name)
|
|
}
|
|
}
|
|
if aws.StringValue(input.EncodingType) == s3.EncodingTypeUrl {
|
|
// AWS url-encodes these key-named elements when encoding-type=url is
|
|
// requested; without it XML 1.0 cannot carry the keys at all and the
|
|
// encoder replaces their characters with U+FFFD.
|
|
output.EncodingType = input.EncodingType
|
|
output.Delimiter = encodeMultipartUploadURLField(output.Delimiter)
|
|
output.Prefix = encodeMultipartUploadURLField(output.Prefix)
|
|
output.KeyMarker = encodeMultipartUploadURLField(output.KeyMarker)
|
|
output.NextKeyMarker = encodeMultipartUploadURLField(output.NextKeyMarker)
|
|
for _, upload := range output.Upload {
|
|
upload.Key = encodeMultipartUploadURLField(upload.Key)
|
|
}
|
|
for i := range output.CommonPrefixes {
|
|
output.CommonPrefixes[i].Prefix = urlQueryEscape(output.CommonPrefixes[i].Prefix)
|
|
}
|
|
} else {
|
|
output.KeyMarker = encodeListContinuationTokenField(output.KeyMarker)
|
|
output.NextKeyMarker = encodeListContinuationTokenField(output.NextKeyMarker)
|
|
}
|
|
return output
|
|
}
|
|
|
|
// urlQueryEscape url-encodes key text per segment: every "/"-separated part is
|
|
// query-escaped and the separators are kept, so a client recovers the exact key
|
|
// with url.QueryUnescape.
|
|
func urlQueryEscape(key string) string {
|
|
segments := strings.Split(key, "/")
|
|
for i, segment := range segments {
|
|
segments[i] = url.QueryEscape(segment)
|
|
}
|
|
return strings.Join(segments, "/")
|
|
}
|
|
|
|
func encodeMultipartUploadURLField(value *string) *string {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
return aws.String(urlQueryEscape(*value))
|
|
}
|
|
|
|
type ListPartsResult struct {
|
|
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ ListPartsResult"`
|
|
|
|
// copied from s3.ListPartsOutput, the Parts is not converting to <Part></Part>
|
|
Bucket *string `type:"string"`
|
|
IsTruncated *bool `type:"boolean"`
|
|
Key *string `min:"1" type:"string"`
|
|
MaxParts *int64 `type:"integer"`
|
|
NextPartNumberMarker *int64 `type:"integer"`
|
|
PartNumberMarker *int64 `type:"integer"`
|
|
Part []*s3.Part `locationName:"Part" type:"list" flattened:"true"`
|
|
StorageClass *string `type:"string" enum:"StorageClass"`
|
|
UploadId *string `type:"string"`
|
|
}
|
|
|
|
func (s3a *S3ApiServer) listObjectParts(input *s3.ListPartsInput) (output *ListPartsResult, code s3err.ErrorCode) {
|
|
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListParts.html
|
|
|
|
glog.V(2).Infof("listObjectParts input %v", input)
|
|
|
|
// complete/abort delete the upload directory, but most stores list a missing
|
|
// directory as empty instead of erroring, so listing alone cannot tell a
|
|
// completed upload (NoSuchUpload on AWS) from an open upload with no parts
|
|
// yet (200 with an empty list). Probe the upload record instead.
|
|
pentry, err := s3a.getEntry(s3a.genUploadsFolder(*input.Bucket), *input.UploadId)
|
|
if err != nil {
|
|
if errors.Is(err, filer_pb.ErrNotFound) {
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
// only createMultipartUpload stamps the key; a directory a part write left behind is not an upload
|
|
if !isMultipartUploadEntry(pentry) {
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
|
|
output = &ListPartsResult{
|
|
Bucket: input.Bucket,
|
|
Key: objectKey(input.Key),
|
|
UploadId: input.UploadId,
|
|
MaxParts: input.MaxParts, // the maximum number of parts to return.
|
|
PartNumberMarker: input.PartNumberMarker, // the part number starts after this, exclusive
|
|
StorageClass: aws.String("STANDARD"),
|
|
}
|
|
|
|
// part files are "%04d_<uuid>.part" (legacy "%04d.part"). '_' sorts after
|
|
// '.', so an exclusive start at "%04d.part" still returns the marker part
|
|
// and a page of max-parts=1 never advances. The next part number sorts
|
|
// past both name shapes of the marker part.
|
|
if *input.PartNumberMarker >= math.MaxInt64 {
|
|
output.IsTruncated = aws.Bool(false)
|
|
return output, s3err.ErrNone
|
|
}
|
|
entries, isLast, err := s3a.list(s3a.genUploadsFolder(*input.Bucket)+"/"+*input.UploadId, "", fmt.Sprintf("%04d", *input.PartNumberMarker+1), true, uint32(*input.MaxParts))
|
|
if err != nil {
|
|
// A store that reports the missing upload directory as not-found means
|
|
// the upload is gone (completed or aborted), not a store error.
|
|
if isFilerNotFound(err) {
|
|
return nil, s3err.ErrNoSuchUpload
|
|
}
|
|
glog.Errorf("listObjectParts %s %s error: %v", *input.Bucket, *input.UploadId, err)
|
|
return nil, s3err.ErrInternalError
|
|
}
|
|
|
|
// Note: The upload directory is sort of a marker of the existence of an multipart upload request.
|
|
// So can not just delete empty upload folders.
|
|
|
|
output.IsTruncated = aws.Bool(!isLast)
|
|
|
|
for _, entry := range entries {
|
|
if strings.HasSuffix(entry.Name, multipartExt) && !entry.IsDirectory {
|
|
partNumber, err := parsePartNumber(entry.Name)
|
|
if err != nil {
|
|
glog.Errorf("listObjectParts %s %s parse %s: %v", *input.Bucket, *input.UploadId, entry.Name, err)
|
|
continue
|
|
}
|
|
// the marker is exclusive even if a store ignores startFrom
|
|
if partNumber <= int(*input.PartNumberMarker) {
|
|
continue
|
|
}
|
|
|
|
partETag := getEtagFromEntry(entry)
|
|
part := &s3.Part{
|
|
PartNumber: aws.Int64(int64(partNumber)),
|
|
LastModified: aws.Time(time.Unix(entry.Attributes.Mtime, 0).UTC()),
|
|
Size: aws.Int64(int64(filer.FileSize(entry))),
|
|
ETag: aws.String(partETag),
|
|
}
|
|
output.Part = append(output.Part, part)
|
|
glog.V(3).Infof("listObjectParts: Added part %d, size=%d, etag=%s",
|
|
partNumber, filer.FileSize(entry), partETag)
|
|
if !isLast {
|
|
output.NextPartNumberMarker = aws.Int64(int64(partNumber))
|
|
}
|
|
}
|
|
}
|
|
|
|
// a page that surfaced no parts has no next page to claim
|
|
if len(output.Part) == 0 {
|
|
output.IsTruncated = aws.Bool(false)
|
|
output.NextPartNumberMarker = nil
|
|
}
|
|
|
|
glog.V(2).Infof("listObjectParts: Returning %d parts for uploadId=%s", len(output.Part), *input.UploadId)
|
|
return
|
|
}
|
|
|
|
// maxInt returns the maximum of two int values
|
|
func maxInt(a, b int) int {
|
|
if a > b {
|
|
return a
|
|
}
|
|
return b
|
|
}
|
|
|
|
// MultipartEncryptionConfig holds pre-prepared encryption configuration to avoid error handling in callbacks
|
|
type MultipartEncryptionConfig struct {
|
|
// SSE-KMS configuration
|
|
IsSSEKMS bool
|
|
KMSKeyID string
|
|
BucketKeyEnabled bool
|
|
EncryptionContext string
|
|
KMSBaseIVEncoded string
|
|
|
|
// SSE-S3 configuration
|
|
IsSSES3 bool
|
|
S3BaseIVEncoded string
|
|
S3KeyDataEncoded string
|
|
}
|
|
|
|
// prepareMultipartEncryptionConfig prepares encryption configuration with proper error handling
|
|
// This eliminates the need for criticalError variable in callback functions
|
|
// Updated to support bucket-default encryption (matches putToFiler behavior)
|
|
func (s3a *S3ApiServer) prepareMultipartEncryptionConfig(r *http.Request, bucket string, uploadIdString string) (*MultipartEncryptionConfig, error) {
|
|
config := &MultipartEncryptionConfig{}
|
|
|
|
// Check for explicit encryption headers first (priority over bucket defaults)
|
|
hasExplicitSSEKMS := IsSSEKMSRequest(r)
|
|
hasExplicitSSES3 := IsSSES3RequestInternal(r)
|
|
|
|
// Prepare SSE-KMS configuration (explicit request headers)
|
|
if hasExplicitSSEKMS {
|
|
config.IsSSEKMS = true
|
|
config.KMSKeyID = r.Header.Get(s3_constants.AmzServerSideEncryptionAwsKmsKeyId)
|
|
config.BucketKeyEnabled = strings.ToLower(r.Header.Get(s3_constants.AmzServerSideEncryptionBucketKeyEnabled)) == "true"
|
|
config.EncryptionContext = r.Header.Get(s3_constants.AmzServerSideEncryptionContext)
|
|
|
|
// Generate and encode base IV with proper error handling
|
|
baseIV := make([]byte, s3_constants.AESBlockSize)
|
|
n, err := rand.Read(baseIV)
|
|
if err != nil || n != len(baseIV) {
|
|
return nil, fmt.Errorf("failed to generate secure IV for SSE-KMS multipart upload: %v (read %d/%d bytes)", err, n, len(baseIV))
|
|
}
|
|
config.KMSBaseIVEncoded = base64.StdEncoding.EncodeToString(baseIV)
|
|
glog.V(4).Infof("Generated base IV %x for explicit SSE-KMS multipart upload %s", baseIV[:8], uploadIdString)
|
|
}
|
|
|
|
// Prepare SSE-S3 configuration (explicit request headers)
|
|
if hasExplicitSSES3 {
|
|
config.IsSSES3 = true
|
|
|
|
// Generate and encode base IV with proper error handling
|
|
baseIV := make([]byte, s3_constants.AESBlockSize)
|
|
n, err := rand.Read(baseIV)
|
|
if err != nil || n != len(baseIV) {
|
|
return nil, fmt.Errorf("failed to generate secure IV for SSE-S3 multipart upload: %v (read %d/%d bytes)", err, n, len(baseIV))
|
|
}
|
|
config.S3BaseIVEncoded = base64.StdEncoding.EncodeToString(baseIV)
|
|
glog.V(4).Infof("Generated base IV %x for explicit SSE-S3 multipart upload %s", baseIV[:8], uploadIdString)
|
|
|
|
// Generate and serialize SSE-S3 key with proper error handling
|
|
keyManager := GetSSES3KeyManager()
|
|
sseS3Key, err := keyManager.GetOrCreateKey("")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to generate SSE-S3 key for multipart upload: %v", err)
|
|
}
|
|
|
|
keyData, serErr := SerializeSSES3Metadata(sseS3Key)
|
|
if serErr != nil {
|
|
return nil, fmt.Errorf("failed to serialize SSE-S3 metadata for multipart upload: %v", serErr)
|
|
}
|
|
|
|
config.S3KeyDataEncoded = base64.StdEncoding.EncodeToString(keyData)
|
|
|
|
// Store key in manager for later retrieval
|
|
keyManager.StoreKey(sseS3Key)
|
|
glog.V(4).Infof("Stored SSE-S3 key %s for explicit multipart upload %s", sseS3Key.KeyID, uploadIdString)
|
|
}
|
|
|
|
// If no explicit encryption headers, check bucket-default encryption
|
|
// This matches AWS S3 behavior and putToFiler() implementation
|
|
if !hasExplicitSSEKMS && !hasExplicitSSES3 {
|
|
encryptionConfig, err := s3a.GetBucketEncryptionConfig(bucket)
|
|
if err != nil {
|
|
// Check if this is just "no encryption configured" vs a real error
|
|
if !errors.Is(err, ErrNoEncryptionConfig) {
|
|
// Real error - propagate to prevent silent encryption bypass
|
|
return nil, fmt.Errorf("failed to read bucket encryption config for multipart upload: %v", err)
|
|
}
|
|
// No default encryption configured, continue without encryption
|
|
} else if encryptionConfig != nil && encryptionConfig.SseAlgorithm != "" {
|
|
glog.V(3).Infof("prepareMultipartEncryptionConfig: applying bucket-default encryption %s for bucket %s, upload %s",
|
|
encryptionConfig.SseAlgorithm, bucket, uploadIdString)
|
|
|
|
switch encryptionConfig.SseAlgorithm {
|
|
case EncryptionTypeKMS:
|
|
// Apply SSE-KMS as bucket default
|
|
config.IsSSEKMS = true
|
|
config.KMSKeyID = encryptionConfig.KmsKeyId
|
|
config.BucketKeyEnabled = encryptionConfig.BucketKeyEnabled
|
|
// No encryption context for bucket defaults
|
|
|
|
// Generate and encode base IV
|
|
baseIV := make([]byte, s3_constants.AESBlockSize)
|
|
n, readErr := rand.Read(baseIV)
|
|
if readErr != nil || n != len(baseIV) {
|
|
return nil, fmt.Errorf("failed to generate secure IV for bucket-default SSE-KMS multipart upload: %v (read %d/%d bytes)", readErr, n, len(baseIV))
|
|
}
|
|
config.KMSBaseIVEncoded = base64.StdEncoding.EncodeToString(baseIV)
|
|
glog.V(4).Infof("Generated base IV %x for bucket-default SSE-KMS multipart upload %s", baseIV[:8], uploadIdString)
|
|
|
|
case EncryptionTypeAES256:
|
|
// Apply SSE-S3 (AES256) as bucket default
|
|
config.IsSSES3 = true
|
|
|
|
// Generate and encode base IV
|
|
baseIV := make([]byte, s3_constants.AESBlockSize)
|
|
n, readErr := rand.Read(baseIV)
|
|
if readErr != nil || n != len(baseIV) {
|
|
return nil, fmt.Errorf("failed to generate secure IV for bucket-default SSE-S3 multipart upload: %v (read %d/%d bytes)", readErr, n, len(baseIV))
|
|
}
|
|
config.S3BaseIVEncoded = base64.StdEncoding.EncodeToString(baseIV)
|
|
glog.V(4).Infof("Generated base IV %x for bucket-default SSE-S3 multipart upload %s", baseIV[:8], uploadIdString)
|
|
|
|
// Generate and serialize SSE-S3 key
|
|
keyManager := GetSSES3KeyManager()
|
|
sseS3Key, keyErr := keyManager.GetOrCreateKey("")
|
|
if keyErr != nil {
|
|
return nil, fmt.Errorf("failed to generate SSE-S3 key for bucket-default multipart upload: %v", keyErr)
|
|
}
|
|
|
|
keyData, serErr := SerializeSSES3Metadata(sseS3Key)
|
|
if serErr != nil {
|
|
return nil, fmt.Errorf("failed to serialize SSE-S3 metadata for bucket-default multipart upload: %v", serErr)
|
|
}
|
|
|
|
config.S3KeyDataEncoded = base64.StdEncoding.EncodeToString(keyData)
|
|
|
|
// Store key in manager for later retrieval
|
|
keyManager.StoreKey(sseS3Key)
|
|
glog.V(4).Infof("Stored SSE-S3 key %s for bucket-default multipart upload %s", sseS3Key.KeyID, uploadIdString)
|
|
|
|
default:
|
|
glog.V(3).Infof("prepareMultipartEncryptionConfig: unsupported bucket-default encryption algorithm %s for bucket %s",
|
|
encryptionConfig.SseAlgorithm, bucket)
|
|
}
|
|
}
|
|
}
|
|
|
|
return config, nil
|
|
}
|
|
|
|
// applyMultipartEncryptionConfig applies pre-prepared encryption configuration to filer entry
|
|
// This function is guaranteed not to fail since all error-prone operations were done during preparation
|
|
func (s3a *S3ApiServer) applyMultipartEncryptionConfig(entry *filer_pb.Entry, config *MultipartEncryptionConfig) {
|
|
// Apply SSE-KMS configuration
|
|
if config.IsSSEKMS {
|
|
entry.Extended[s3_constants.SeaweedFSSSEKMSKeyID] = []byte(config.KMSKeyID)
|
|
if config.BucketKeyEnabled {
|
|
entry.Extended[s3_constants.SeaweedFSSSEKMSBucketKeyEnabled] = []byte("true")
|
|
}
|
|
if config.EncryptionContext != "" {
|
|
entry.Extended[s3_constants.SeaweedFSSSEKMSEncryptionContext] = []byte(config.EncryptionContext)
|
|
}
|
|
entry.Extended[s3_constants.SeaweedFSSSEKMSBaseIV] = []byte(config.KMSBaseIVEncoded)
|
|
glog.V(3).Infof("applyMultipartEncryptionConfig: applied SSE-KMS settings with keyID %s", config.KMSKeyID)
|
|
}
|
|
|
|
// Apply SSE-S3 configuration
|
|
if config.IsSSES3 {
|
|
entry.Extended[s3_constants.SeaweedFSSSES3Encryption] = []byte(s3_constants.SSEAlgorithmAES256)
|
|
entry.Extended[s3_constants.SeaweedFSSSES3BaseIV] = []byte(config.S3BaseIVEncoded)
|
|
entry.Extended[s3_constants.SeaweedFSSSES3KeyData] = []byte(config.S3KeyDataEncoded)
|
|
glog.V(3).Infof("applyMultipartEncryptionConfig: applied SSE-S3 settings")
|
|
}
|
|
}
|
|
|
|
func sortEntriesByLatestChunk(entries []*filer_pb.Entry) {
|
|
slices.SortFunc(entries, func(a, b *filer_pb.Entry) int {
|
|
var aTs, bTs int64
|
|
if len(a.Chunks) > 0 {
|
|
aTs = a.Chunks[0].ModifiedTsNs
|
|
}
|
|
if len(b.Chunks) > 0 {
|
|
bTs = b.Chunks[0].ModifiedTsNs
|
|
}
|
|
return cmp.Compare(bTs, aTs)
|
|
})
|
|
}
|
|
|
|
func calculateMultipartETag(partEntries map[int][]*filer_pb.Entry, completedPartNumbers []int) string {
|
|
var etags []byte
|
|
for _, partNumber := range completedPartNumbers {
|
|
entries, ok := partEntries[partNumber]
|
|
if !ok || len(entries) == 0 {
|
|
continue
|
|
}
|
|
if len(entries) > 1 {
|
|
sortEntriesByLatestChunk(entries)
|
|
}
|
|
entry := entries[0]
|
|
etag := getEtagFromEntry(entry)
|
|
glog.V(4).Infof("calculateMultipartETag: part %d, entry %s, getEtagFromEntry result: %s", partNumber, entry.Name, etag)
|
|
etag = strings.Trim(etag, "\"")
|
|
if before, _, found := strings.Cut(etag, "-"); found {
|
|
etag = before
|
|
}
|
|
if etagBytes, err := hex.DecodeString(etag); err == nil {
|
|
etags = append(etags, etagBytes...)
|
|
} else {
|
|
glog.Warningf("calculateMultipartETag: failed to decode etag '%s' for part %d: %v", etag, partNumber, err)
|
|
}
|
|
}
|
|
return fmt.Sprintf("%x-%d", md5.Sum(etags), len(completedPartNumbers))
|
|
}
|
|
|
|
func decodePartChecksum(partNumber int, entries []*filer_pb.Entry, checksumHeaderName string) ([]byte, *filer_pb.Entry, error) {
|
|
if len(entries) == 0 {
|
|
return nil, nil, fmt.Errorf("part %d not found", partNumber)
|
|
}
|
|
if len(entries) > 1 {
|
|
sortEntriesByLatestChunk(entries)
|
|
}
|
|
entry := entries[0]
|
|
if entry.Extended == nil {
|
|
return nil, nil, fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName)
|
|
}
|
|
// Validate the part's checksum algorithm matches the upload's expected algorithm
|
|
partAlgo, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm]
|
|
if !ok || len(partAlgo) == 0 {
|
|
return nil, nil, fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName)
|
|
}
|
|
if string(partAlgo) != checksumHeaderName {
|
|
return nil, nil, fmt.Errorf("part %d checksum algorithm mismatch: upload expects %s but part has %s", partNumber, checksumHeaderName, string(partAlgo))
|
|
}
|
|
partChecksumB64, ok := entry.Extended[s3_constants.ExtChecksumValue]
|
|
if !ok || len(partChecksumB64) == 0 {
|
|
return nil, nil, fmt.Errorf("part %d missing checksum value: upload initiated with %s but part has no checksum value", partNumber, checksumHeaderName)
|
|
}
|
|
raw, err := base64.StdEncoding.DecodeString(string(partChecksumB64))
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("part %d has invalid checksum encoding: %w", partNumber, err)
|
|
}
|
|
return raw, entry, nil
|
|
}
|
|
|
|
// computeCompositeChecksum computes a composite checksum from per-part checksums.
|
|
// It concatenates the raw (decoded) per-part checksums, hashes the result with the
|
|
// same algorithm, and returns the value as "base64-N" where N is the part count.
|
|
// This follows the AWS S3 multipart checksum specification.
|
|
// Returns an error if a part is missing its checksum (the upload was initiated with
|
|
// a checksum algorithm, so all parts must have been uploaded with checksums).
|
|
func computeCompositeChecksum(checksumHeaderName string, partEntries map[int][]*filer_pb.Entry, completedPartNumbers []int) (string, error) {
|
|
// Determine the algorithm from the header name
|
|
algo := checksumAlgorithmFromHeaderName(checksumHeaderName)
|
|
if algo == ChecksumAlgorithmNone {
|
|
return "", fmt.Errorf("unknown checksum algorithm for header %q", checksumHeaderName)
|
|
}
|
|
|
|
// Collect raw per-part checksums
|
|
var combined []byte
|
|
for _, partNumber := range completedPartNumbers {
|
|
raw, _, err := decodePartChecksum(partNumber, partEntries[partNumber], checksumHeaderName)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
combined = append(combined, raw...)
|
|
}
|
|
|
|
// Hash the concatenated raw checksums
|
|
h := getCheckSumWriter(algo)
|
|
if h == nil {
|
|
return "", fmt.Errorf("failed to create hash writer for %s", checksumHeaderName)
|
|
}
|
|
h.Write(combined)
|
|
compositeRaw := h.Sum(nil)
|
|
return fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(compositeRaw), len(completedPartNumbers)), nil
|
|
}
|
|
|
|
// computeFullObjectChecksum computes a FULL_OBJECT checksum from per-part checksums
|
|
// by combining the per-part CRCs into the CRC of the whole object. The result is the
|
|
// base64-encoded whole-object checksum with NO "-N" suffix (a full-object checksum
|
|
// is indistinguishable from a single-part upload's checksum). This is required for
|
|
// CRC64NVME, which AWS only supports as a full-object checksum.
|
|
func computeFullObjectChecksum(checksumHeaderName string, partEntries map[int][]*filer_pb.Entry, completedPartNumbers []int) (string, error) {
|
|
algo := checksumAlgorithmFromHeaderName(checksumHeaderName)
|
|
params, ok := crcCombineParams[algo]
|
|
if !ok {
|
|
return "", fmt.Errorf("full object checksum not supported for %s", checksumHeaderName)
|
|
}
|
|
|
|
checksumBytes := int(params.width / 8)
|
|
|
|
var combined uint64
|
|
for i, partNumber := range completedPartNumbers {
|
|
raw, entry, err := decodePartChecksum(partNumber, partEntries[partNumber], checksumHeaderName)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if len(raw) != checksumBytes {
|
|
return "", fmt.Errorf("part %d checksum has unexpected length %d for %s", partNumber, len(raw), checksumHeaderName)
|
|
}
|
|
|
|
crc := util.BytesToUint64(raw)
|
|
|
|
if i == 0 {
|
|
combined = crc
|
|
} else {
|
|
partLen := filer.FileSize(entry)
|
|
combined = combineCRC(combined, crc, partLen, params)
|
|
}
|
|
}
|
|
|
|
// Write the low checksumBytes bytes of the combined CRC big-endian. We can't
|
|
// use util.Uint64toBytes here because it unconditionally writes 8 bytes, which
|
|
// would panic for narrower CRCs (e.g. 4-byte CRC32/CRC32C).
|
|
out := make([]byte, checksumBytes)
|
|
for i := 0; i < checksumBytes; i++ {
|
|
out[checksumBytes-1-i] = byte(combined >> (i * 8))
|
|
}
|
|
|
|
return base64.StdEncoding.EncodeToString(out), nil
|
|
}
|
|
|
|
// checksumAlgorithmFromHeaderName maps a canonical header name back to its algorithm.
|
|
func checksumAlgorithmFromHeaderName(headerName string) ChecksumAlgorithm {
|
|
for _, entry := range checksumHeaders {
|
|
if entry.name == headerName {
|
|
return entry.alg
|
|
}
|
|
}
|
|
return ChecksumAlgorithmNone
|
|
}
|
|
|
|
func checksumAlgorithmNameFromHeaderName(headerName string) string {
|
|
for name, entry := range checksumAlgorithmMapping {
|
|
if entry.name == headerName {
|
|
return name
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func getEtagFromEntry(entry *filer_pb.Entry) string {
|
|
if entry.Extended != nil {
|
|
if etagBytes, ok := entry.Extended[s3_constants.ExtETagKey]; ok {
|
|
etag := string(etagBytes)
|
|
if len(etag) > 0 {
|
|
if !strings.HasPrefix(etag, "\"") {
|
|
return "\"" + etag + "\""
|
|
}
|
|
return etag
|
|
}
|
|
// Empty stored ETag — fall through to filer.ETag calculation
|
|
}
|
|
}
|
|
// Fallback to filer.ETag which handles Attributes.Md5 consistently
|
|
etag := filer.ETag(entry)
|
|
entryName := entry.Name
|
|
if entryName == "" {
|
|
entryName = "entry"
|
|
}
|
|
glog.V(4).Infof("getEtagFromEntry: fallback to filer.ETag for %s: %s, chunkCount: %d", entryName, etag, len(entry.Chunks))
|
|
return "\"" + etag + "\""
|
|
}
|
|
|
|
func validateCompletePartETag(partETag string, entry *filer_pb.Entry) (match bool, invalid bool, normalizedPartETag string, normalizedEntryETag string) {
|
|
normalizedPartETag = strings.Trim(strings.TrimSpace(partETag), `"`)
|
|
if normalizedPartETag == "" {
|
|
return false, true, normalizedPartETag, ""
|
|
}
|
|
|
|
normalizedEntryETag = strings.Trim(getEtagFromEntry(entry), `"`)
|
|
if normalizedEntryETag == "" {
|
|
return false, true, normalizedPartETag, normalizedEntryETag
|
|
}
|
|
|
|
return normalizedPartETag == normalizedEntryETag, false, normalizedPartETag, normalizedEntryETag
|
|
}
|
|
|
|
type checksumTypeSupport struct {
|
|
composite bool
|
|
fullObject bool
|
|
}
|
|
|
|
var checksumTypeSupportByAlgo = map[ChecksumAlgorithm]checksumTypeSupport{
|
|
ChecksumAlgorithmCRC32: {composite: true, fullObject: true},
|
|
ChecksumAlgorithmCRC32C: {composite: true, fullObject: true},
|
|
ChecksumAlgorithmCRC64NVMe: {fullObject: true},
|
|
ChecksumAlgorithmSHA1: {composite: true}, // Technically, fullObject could be supported, but is not yet implemented
|
|
ChecksumAlgorithmSHA256: {composite: true}, // Technically, fullObject could be supported, but is not yet implemented
|
|
}
|
|
|
|
func resolveMultipartChecksumType(algo ChecksumAlgorithm, requested string) (string, error) {
|
|
support, ok := checksumTypeSupportByAlgo[algo]
|
|
if !ok {
|
|
return "", fmt.Errorf("unsupported checksum algorithm %v", algo)
|
|
}
|
|
|
|
switch strings.ToUpper(strings.TrimSpace(requested)) {
|
|
case "":
|
|
if support.composite {
|
|
return s3_constants.ChecksumTypeComposite, nil
|
|
}
|
|
return s3_constants.ChecksumTypeFullObject, nil
|
|
case s3_constants.ChecksumTypeComposite:
|
|
if !support.composite {
|
|
return "", fmt.Errorf("checksum algorithm %v does not support %s checksums", algo, s3_constants.ChecksumTypeComposite)
|
|
}
|
|
return s3_constants.ChecksumTypeComposite, nil
|
|
case s3_constants.ChecksumTypeFullObject:
|
|
if !support.fullObject {
|
|
return "", fmt.Errorf("checksum algorithm %v does not support %s checksums", algo, s3_constants.ChecksumTypeFullObject)
|
|
}
|
|
return s3_constants.ChecksumTypeFullObject, nil
|
|
default:
|
|
return "", fmt.Errorf("invalid checksum type %q", requested)
|
|
}
|
|
}
|