From 561768a42691d00677d8e0bc16bd55f08be54cf2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 14 Jun 2026 00:16:14 -0700 Subject: [PATCH] [s3]: preserve multipart copy checksums (#9948) * s3: preserve checksums for copied multipart parts * s3: return checksums from multipart copy * s3: pin the upload's checksum algorithm on copy-part re-stream * s3: note why UploadPartCopy uses the re-stream slow path * s3: explain the TLS proxy in the multipart copy checksum test * s3: cover nil and unknown-algorithm edge cases in copy checksum tests * s3: cover all checksum algorithms in the multipart copy test * s3: run all checksum integration tests, not just presigned --- test/s3/checksum/Makefile | 4 +- .../s3/checksum/s3_checksum_multipart_test.go | 124 ++++++++++++++++ weed/s3api/filer_multipart.go | 9 ++ weed/s3api/s3api_object_handlers_copy.go | 43 ++++-- ...3api_object_handlers_copy_checksum_test.go | 135 ++++++++++++++++++ .../s3api_object_handlers_copy_part_sse.go | 60 ++++---- 6 files changed, 331 insertions(+), 44 deletions(-) create mode 100644 test/s3/checksum/s3_checksum_multipart_test.go create mode 100644 weed/s3api/s3api_object_handlers_copy_checksum_test.go diff --git a/test/s3/checksum/Makefile b/test/s3/checksum/Makefile index e88005618..fd25a4e06 100644 --- a/test/s3/checksum/Makefile +++ b/test/s3/checksum/Makefile @@ -1,5 +1,5 @@ # S3 Checksum Integration Tests -# Covers flexible-checksum behavior on presigned URL uploads (issue #9075). +# Covers flexible-checksum behavior on presigned uploads and multipart copy. .PHONY: help build-weed check-deps start-server stop-server test test-with-server logs clean health-check @@ -8,7 +8,7 @@ S3_PORT := 8333 ACCESS_KEY ?= some_access_key1 SECRET_KEY ?= some_secret_key1 TEST_TIMEOUT := 10m -TEST_PATTERN ?= TestPresignedPut +TEST_PATTERN ?= . SERVER_DIR := ./test-volume-data/server-data help: diff --git a/test/s3/checksum/s3_checksum_multipart_test.go b/test/s3/checksum/s3_checksum_multipart_test.go new file mode 100644 index 000000000..1482fe753 --- /dev/null +++ b/test/s3/checksum/s3_checksum_multipart_test.go @@ -0,0 +1,124 @@ +package checksum_test + +import ( + "bytes" + "context" + "fmt" + "net/http/httptest" + "net/http/httputil" + "net/url" + "testing" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" + "github.com/stretchr/testify/require" +) + +func TestMultipartCopyPreservesChecksum(t *testing.T) { + // aws-sdk-go-v2 sends flexible checksums as unsigned streaming trailers, which + // it refuses over plain HTTP, so front the HTTP endpoint with a TLS proxy. + target, err := url.Parse(defaultConfig.Endpoint) + require.NoError(t, err) + + proxy := httputil.NewSingleHostReverseProxy(target) + server := httptest.NewTLSServer(proxy) + defer server.Close() + + cfg, err := config.LoadDefaultConfig(context.Background(), + config.WithRegion(defaultConfig.Region), + config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider( + defaultConfig.AccessKey, defaultConfig.SecretKey, "")), + config.WithHTTPClient(server.Client()), + ) + require.NoError(t, err) + client := s3.NewFromConfig(cfg, func(o *s3.Options) { + o.BaseEndpoint = aws.String(server.URL) + o.UsePathStyle = true + }) + + bucket := uniqueBucket() + createBucket(t, client, bucket) + defer cleanupBucket(t, client, bucket) + + cases := []struct { + algorithm types.ChecksumAlgorithm + srcSum func(*s3.PutObjectOutput) *string + partSum func(*types.CopyPartResult) *string + setPart func(*types.CompletedPart, *string) + }{ + { + algorithm: types.ChecksumAlgorithmCrc32, + srcSum: func(o *s3.PutObjectOutput) *string { return o.ChecksumCRC32 }, + partSum: func(r *types.CopyPartResult) *string { return r.ChecksumCRC32 }, + setPart: func(p *types.CompletedPart, v *string) { p.ChecksumCRC32 = v }, + }, + { + algorithm: types.ChecksumAlgorithmCrc32c, + srcSum: func(o *s3.PutObjectOutput) *string { return o.ChecksumCRC32C }, + partSum: func(r *types.CopyPartResult) *string { return r.ChecksumCRC32C }, + setPart: func(p *types.CompletedPart, v *string) { p.ChecksumCRC32C = v }, + }, + { + algorithm: types.ChecksumAlgorithmCrc64nvme, + srcSum: func(o *s3.PutObjectOutput) *string { return o.ChecksumCRC64NVME }, + partSum: func(r *types.CopyPartResult) *string { return r.ChecksumCRC64NVME }, + setPart: func(p *types.CompletedPart, v *string) { p.ChecksumCRC64NVME = v }, + }, + { + algorithm: types.ChecksumAlgorithmSha1, + srcSum: func(o *s3.PutObjectOutput) *string { return o.ChecksumSHA1 }, + partSum: func(r *types.CopyPartResult) *string { return r.ChecksumSHA1 }, + setPart: func(p *types.CompletedPart, v *string) { p.ChecksumSHA1 = v }, + }, + { + algorithm: types.ChecksumAlgorithmSha256, + srcSum: func(o *s3.PutObjectOutput) *string { return o.ChecksumSHA256 }, + partSum: func(r *types.CopyPartResult) *string { return r.ChecksumSHA256 }, + setPart: func(p *types.CompletedPart, v *string) { p.ChecksumSHA256 = v }, + }, + } + + for _, tc := range cases { + t.Run(string(tc.algorithm), func(t *testing.T) { + sourceKey := "source-" + string(tc.algorithm) + source, err := client.PutObject(context.Background(), &s3.PutObjectInput{ + Bucket: aws.String(bucket), + Key: aws.String(sourceKey), + Body: bytes.NewBufferString("trailing checksum"), + ChecksumAlgorithm: tc.algorithm, + }) + require.NoError(t, err) + + key := "multipart-copy-" + string(tc.algorithm) + create, err := client.CreateMultipartUpload(context.Background(), &s3.CreateMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + ChecksumAlgorithm: tc.algorithm, + }) + require.NoError(t, err) + + part, err := client.UploadPartCopy(context.Background(), &s3.UploadPartCopyInput{ + Bucket: aws.String(bucket), + CopySource: aws.String(fmt.Sprintf("%s/%s", bucket, sourceKey)), + Key: aws.String(key), + UploadId: create.UploadId, + PartNumber: aws.Int32(1), + }) + require.NoError(t, err) + require.Equal(t, aws.ToString(tc.srcSum(source)), aws.ToString(tc.partSum(part.CopyPartResult))) + + completed := types.CompletedPart{ETag: part.CopyPartResult.ETag, PartNumber: aws.Int32(1)} + tc.setPart(&completed, tc.partSum(part.CopyPartResult)) + _, err = client.CompleteMultipartUpload(context.Background(), &s3.CompleteMultipartUploadInput{ + Bucket: aws.String(bucket), + Key: aws.String(key), + UploadId: create.UploadId, + MultipartUpload: &types.CompletedMultipartUpload{Parts: []types.CompletedPart{completed}}, + }) + require.NoError(t, err) + }) + } +} diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 1acd19dfe..5010bce3d 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -1313,6 +1313,15 @@ func checksumAlgorithmFromHeaderName(headerName string) ChecksumAlgorithm { 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 { diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 79fe4f063..77651e7f7 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -713,8 +713,33 @@ func pathToBucketObjectAndVersion(rawPath, decodedPath string) (bucket, object, } type CopyPartResult struct { - LastModified time.Time `xml:"LastModified"` - ETag string `xml:"ETag"` + LastModified time.Time `xml:"LastModified"` + ETag string `xml:"ETag"` + ChecksumCRC32 string `xml:"ChecksumCRC32,omitempty"` + ChecksumCRC32C string `xml:"ChecksumCRC32C,omitempty"` + ChecksumCRC64NVME string `xml:"ChecksumCRC64NVME,omitempty"` + ChecksumSHA1 string `xml:"ChecksumSHA1,omitempty"` + ChecksumSHA256 string `xml:"ChecksumSHA256,omitempty"` +} + +func buildCopyPartResult(etag string, lastModified time.Time, metadata SSEResponseMetadata) CopyPartResult { + result := CopyPartResult{ + ETag: etag, + LastModified: lastModified, + } + switch metadata.ChecksumHeaderName { + case s3_constants.AmzChecksumCRC32: + result.ChecksumCRC32 = metadata.ChecksumValue + case s3_constants.AmzChecksumCRC32C: + result.ChecksumCRC32C = metadata.ChecksumValue + case s3_constants.AmzChecksumCRC64NVME: + result.ChecksumCRC64NVME = metadata.ChecksumValue + case s3_constants.AmzChecksumSHA1: + result.ChecksumSHA1 = metadata.ChecksumValue + case s3_constants.AmzChecksumSHA256: + result.ChecksumSHA256 = metadata.ChecksumValue + } + return result } // copyPartLocation returns the destination directory and filename for a @@ -906,7 +931,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req return } - if uploadEntryHasSSE(uploadEntry) || sourceEntryHasSSE(entry) { + if uploadEntryHasSSE(uploadEntry) || sourceEntryHasSSE(entry) || uploadEntryHasChecksum(uploadEntry) { etag, sseMetadata, errCode := s3a.copyObjectPartViaReencryption(r, entry, startOffset, endOffset, dstBucket, uploadID, partID, uploadEntry) if errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) @@ -917,10 +942,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req // x-amz-server-side-encryption-aws-kms-key-id headers on the response // so clients can see the destination's encryption state. s3a.setSSEResponseHeaders(w, r, sseMetadata) - writeSuccessResponseXML(w, r, CopyPartResult{ - ETag: etag, - LastModified: t, - }) + writeSuccessResponseXML(w, r, buildCopyPartResult(etag, t, sseMetadata)) return } @@ -990,12 +1012,7 @@ func (s3a *S3ApiServer) CopyObjectPartHandler(w http.ResponseWriter, r *http.Req etag := copyEntryETag(dstEntry) setEtag(w, etag) - response := CopyPartResult{ - ETag: etag, - LastModified: t, - } - - writeSuccessResponseXML(w, r, response) + writeSuccessResponseXML(w, r, buildCopyPartResult(etag, t, SSEResponseMetadata{})) } func replaceDirective(reqHeader http.Header) (replaceMeta, replaceTagging bool) { diff --git a/weed/s3api/s3api_object_handlers_copy_checksum_test.go b/weed/s3api/s3api_object_handlers_copy_checksum_test.go new file mode 100644 index 000000000..b6ed9247a --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_checksum_test.go @@ -0,0 +1,135 @@ +package s3api + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" +) + +func TestApplyDestChecksumHeaderToCopyRequest(t *testing.T) { + entry := &filer_pb.Entry{Extended: map[string][]byte{ + s3_constants.ExtChecksumAlgorithm: []byte(s3_constants.AmzChecksumCRC64NVME), + }} + request := httptest.NewRequest(http.MethodPut, "http://example.com/bucket/object", nil) + + applyDestChecksumHeaderToCopyRequest(request, entry) + + algorithm, headerName, errCode := detectRequestedChecksumAlgorithm(request) + if errCode != s3err.ErrNone { + t.Fatalf("detect checksum returned %v", errCode) + } + if algorithm != ChecksumAlgorithmCRC64NVMe { + t.Fatalf("algorithm = %v, want %v", algorithm, ChecksumAlgorithmCRC64NVMe) + } + if headerName != s3_constants.AmzChecksumCRC64NVME { + t.Fatalf("header = %q, want %q", headerName, s3_constants.AmzChecksumCRC64NVME) + } + + for _, noChecksum := range []*filer_pb.Entry{ + nil, + {}, + {Extended: map[string][]byte{s3_constants.ExtChecksumAlgorithm: []byte("unknown")}}, + } { + req := httptest.NewRequest(http.MethodPut, "http://example.com/bucket/object", nil) + applyDestChecksumHeaderToCopyRequest(req, noChecksum) + if got := req.Header.Get(s3_constants.AmzChecksumAlgorithm); got != "" { + t.Fatalf("expected no checksum header, got %q", got) + } + } +} + +func TestUploadEntryHasChecksum(t *testing.T) { + entry := &filer_pb.Entry{Extended: map[string][]byte{ + s3_constants.ExtChecksumAlgorithm: []byte(s3_constants.AmzChecksumCRC64NVME), + }} + if !uploadEntryHasChecksum(entry) { + t.Fatal("checksum-enabled upload was not detected") + } + + entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte("unknown") + if uploadEntryHasChecksum(entry) { + t.Fatal("unknown checksum algorithm was accepted") + } + + if uploadEntryHasChecksum(nil) || uploadEntryHasChecksum(&filer_pb.Entry{}) { + t.Fatal("nil entry or nil Extended map reported a checksum") + } +} + +func TestBuildCopyPartResult(t *testing.T) { + modified := time.Unix(123, 0).UTC() + tests := []struct { + name string + header string + element string + expected CopyPartResult + }{ + { + name: "CRC32", + header: s3_constants.AmzChecksumCRC32, + element: "value", + expected: CopyPartResult{ + ETag: "etag", LastModified: modified, ChecksumCRC32: "value", + }, + }, + { + name: "CRC32C", + header: s3_constants.AmzChecksumCRC32C, + element: "value", + expected: CopyPartResult{ + ETag: "etag", LastModified: modified, ChecksumCRC32C: "value", + }, + }, + { + name: "CRC64NVME", + header: s3_constants.AmzChecksumCRC64NVME, + element: "value", + expected: CopyPartResult{ + ETag: "etag", LastModified: modified, ChecksumCRC64NVME: "value", + }, + }, + { + name: "SHA1", + header: s3_constants.AmzChecksumSHA1, + element: "value", + expected: CopyPartResult{ + ETag: "etag", LastModified: modified, ChecksumSHA1: "value", + }, + }, + { + name: "SHA256", + header: s3_constants.AmzChecksumSHA256, + element: "value", + expected: CopyPartResult{ + ETag: "etag", LastModified: modified, ChecksumSHA256: "value", + }, + }, + { + name: "Unknown", + header: "x-amz-checksum-unknown", + element: "", + expected: CopyPartResult{ETag: "etag", LastModified: modified}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + result := buildCopyPartResult("etag", modified, SSEResponseMetadata{ + ChecksumHeaderName: test.header, + ChecksumValue: "value", + }) + if result != test.expected { + t.Fatalf("result = %#v, want %#v", result, test.expected) + } + if encoded := string(s3err.EncodeXMLResponse(result)); test.element != "" && !strings.Contains(encoded, test.element) { + t.Fatalf("response %q does not contain %q", encoded, test.element) + } + }) + } +} diff --git a/weed/s3api/s3api_object_handlers_copy_part_sse.go b/weed/s3api/s3api_object_handlers_copy_part_sse.go index 66442a6d7..163543212 100644 --- a/weed/s3api/s3api_object_handlers_copy_part_sse.go +++ b/weed/s3api/s3api_object_handlers_copy_part_sse.go @@ -65,6 +65,14 @@ func uploadEntryHasSSE(uploadEntry *filer_pb.Entry) bool { return false } +func uploadEntryHasChecksum(uploadEntry *filer_pb.Entry) bool { + if uploadEntry == nil || uploadEntry.Extended == nil { + return false + } + headerName := string(uploadEntry.Extended[s3_constants.ExtChecksumAlgorithm]) + return checksumAlgorithmFromHeaderName(headerName) != ChecksumAlgorithmNone +} + // sourceEntryHasSSE reports whether the source object's chunks are SSE // ciphertext on disk and therefore cannot be raw-copied — they must be // decrypted on read. @@ -315,10 +323,18 @@ func (s3a *S3ApiServer) applyDestSSEHeadersToCopyRequest( return s3a.handleSSES3MultipartHeaders(r, uploadEntry, uploadID) } -// fakeContentRequest builds a minimal request representing "PUT this body" for -// the multipart-part write path used by UploadPartCopy. It clones the original -// request's headers (so things like AmzAccountId carry over) and clears the -// copy-only headers; SSE setup is added later by applyDestSSEHeadersToCopyRequest. +func applyDestChecksumHeaderToCopyRequest(r *http.Request, uploadEntry *filer_pb.Entry) { + if uploadEntry == nil || uploadEntry.Extended == nil { + return + } + headerName := string(uploadEntry.Extended[s3_constants.ExtChecksumAlgorithm]) + if algorithm := checksumAlgorithmNameFromHeaderName(headerName); algorithm != "" { + // Drop any inherited sdk-checksum selector; it outranks the header we set. + r.Header.Del(s3_constants.AmzSdkChecksumAlgorithm) + r.Header.Set(s3_constants.AmzChecksumAlgorithm, algorithm) + } +} + func fakeContentRequest(orig *http.Request, body io.ReadCloser, contentLength int64) *http.Request { cloned := orig.Clone(orig.Context()) cloned.Body = body @@ -339,22 +355,11 @@ func fakeContentRequest(orig *http.Request, body io.ReadCloser, contentLength in return cloned } -// copyObjectPartViaReencryption implements the slow path of UploadPartCopy when -// either the source object is SSE-encrypted or the destination multipart upload -// is configured for SSE encryption. It: -// -// 1. Opens a plaintext reader of the source range (decrypting if needed). -// 2. Stages the destination's SSE-S3 / SSE-KMS multipart headers on a cloned -// request so handleAllSSEEncryption (called from putToFiler) routes the -// body through the matching multipart-encryption helper. -// 3. Calls putToFiler with the plaintext reader, which encrypts using the -// destination upload session's key+baseIV (consistent with PutObjectPart), -// auto-chunks, and writes the part entry with proper per-chunk SSE metadata. -// -// Without this path, copyChunksForRange's raw byte copy leaves destination -// chunks SseType=NONE; completedMultipartChunk then "backfills" SSE-S3 metadata -// with destination-baseIV-derived IVs, but the bytes on disk were encrypted -// with the source's key — yielding deterministic byte corruption on GET (#8908). +// copyObjectPartViaReencryption is the UploadPartCopy slow path: it re-streams the +// source range through putToFiler so the destination's SSE re-encryption and/or +// requested checksum are produced on write. A raw chunk copy can't: it would leave +// dest chunks under the source key (corrupt GET) and parts with no checksum +// (completion fails). func (s3a *S3ApiServer) copyObjectPartViaReencryption( r *http.Request, srcEntry *filer_pb.Entry, @@ -363,11 +368,14 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption( partID int, uploadEntry *filer_pb.Entry, ) (etag string, sseMetadata SSEResponseMetadata, errCode s3err.ErrorCode) { - if endOffset < startOffset { + if endOffset < startOffset && !uploadEntryHasChecksum(uploadEntry) { tag, code := s3a.writeEmptyCopyPart(dstBucket, uploadID, partID) return tag, SSEResponseMetadata{}, code } - sliceLen := endOffset - startOffset + 1 + sliceLen := int64(0) + if endOffset >= startOffset { + sliceLen = endOffset - startOffset + 1 + } srcReader, err := s3a.openSourcePlaintextReader(r.Context(), srcEntry, startOffset, endOffset) if err != nil { @@ -388,15 +396,9 @@ func (s3a *S3ApiServer) copyObjectPartViaReencryption( glog.Errorf("UploadPartCopy: apply destination SSE headers: %v", err) return "", SSEResponseMetadata{}, s3err.ErrInternalError } + applyDestChecksumHeaderToCopyRequest(cloned, uploadEntry) - // Surface putToFiler's SSE response metadata to the caller so the handler - // can mirror PutObjectPart's behavior of writing - // x-amz-server-side-encryption / x-amz-server-side-encryption-aws-kms-key-id - // on the UploadPartCopy response. Without this, clients have no way to - // see that the destination was encrypted. filePath := s3a.genPartUploadPath(dstBucket, uploadID, partID) - // Copy-part is an MPU part write under .uploads//; lifecycle - // TTL only applies to the eventual completed object. Pass 0. tag, code, putSSE := s3a.putToFiler(cloned, filePath, srcReader, dstBucket, "", partID, 0, nil, false) if code != s3err.ErrNone { return "", SSEResponseMetadata{}, code