[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

@@ -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:

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)
+		})
+	}
+}

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 {

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) {

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: "<ChecksumCRC32>value</ChecksumCRC32>",
+			expected: CopyPartResult{
+				ETag: "etag", LastModified: modified, ChecksumCRC32: "value",
+			},
+		},
+		{
+			name:    "CRC32C",
+			header:  s3_constants.AmzChecksumCRC32C,
+			element: "<ChecksumCRC32C>value</ChecksumCRC32C>",
+			expected: CopyPartResult{
+				ETag: "etag", LastModified: modified, ChecksumCRC32C: "value",
+			},
+		},
+		{
+			name:    "CRC64NVME",
+			header:  s3_constants.AmzChecksumCRC64NVME,
+			element: "<ChecksumCRC64NVME>value</ChecksumCRC64NVME>",
+			expected: CopyPartResult{
+				ETag: "etag", LastModified: modified, ChecksumCRC64NVME: "value",
+			},
+		},
+		{
+			name:    "SHA1",
+			header:  s3_constants.AmzChecksumSHA1,
+			element: "<ChecksumSHA1>value</ChecksumSHA1>",
+			expected: CopyPartResult{
+				ETag: "etag", LastModified: modified, ChecksumSHA1: "value",
+			},
+		},
+		{
+			name:    "SHA256",
+			header:  s3_constants.AmzChecksumSHA256,
+			element: "<ChecksumSHA256>value</ChecksumSHA256>",
+			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)
+			}
+		})
+	}
+}

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/<id>/<n>; 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