Make it possible to disable integrity checks for multipart uploads

As announced in #2960,
AWS SDK for Go v2 service/s3 v1.73.0 shipped a change
(#2808) that adopted the new
default integrity protections.

While it is possible to revert to the previous behavior by setting
`AWS_REQUEST_CHECKSUM_CALCULATION=when_required`, this config setting
did not apply to the S3 Manager multipart uploader, which always enabled
CRC32 checksums by default.

This commit checks the S3 options and only attaches the CRC32 checksums
if `when_required` is not set.

Relates to #3007

Signed-off-by: Stan Hu <[email protected]>

Author: stanhu

feature/s3/manager/upload.go

@@ -341,6 +341,18 @@ type uploader struct {
 	totalSize int64 // set to -1 if the size is not known
 }
 
+// getRequestChecksumCalculation extracts the RequestChecksumCalculation setting
+// from the uploader's ClientOptions configuration.
+func (u *uploader) getRequestChecksumCalculation() aws.RequestChecksumCalculation {
+	opts := &s3.Options{}
+
+	for _, opt := range u.cfg.ClientOptions {
+		opt(opts)
+	}
+
+	return opts.RequestChecksumCalculation
+}
+
 // internal logic for deciding whether to upload a single part or use a
 // multipart upload.
 func (u *uploader) upload() (*UploadOutput, error) {
@@ -776,7 +788,11 @@ func (u *multiuploader) initChecksumAlgorithm() {
 	case u.in.ChecksumSHA256 != nil:
 		u.in.ChecksumAlgorithm = types.ChecksumAlgorithmSha256
 	default:
-		u.in.ChecksumAlgorithm = types.ChecksumAlgorithmCrc32
+		checksumCalc := u.getRequestChecksumCalculation()
+
+		if checksumCalc != aws.RequestChecksumCalculationWhenRequired {
+			u.in.ChecksumAlgorithm = types.ChecksumAlgorithmCrc32
+		}
 	}
 }
 

feature/s3/manager/upload_test.go

@@ -1160,6 +1160,95 @@ func (h *failPartHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	h.failsRemaining--
 }
 
+// TestUploadRequestChecksumCalculation tests that checksum behavior respects
+// the RequestChecksumCalculation setting for backwards compatibility.
+func TestUploadRequestChecksumCalculation(t *testing.T) {
+	testCases := []struct {
+		name                       string
+		requestChecksumCalculation aws.RequestChecksumCalculation
+		inputChecksumAlgorithm     types.ChecksumAlgorithm
+		expectedChecksumAlgorithm  types.ChecksumAlgorithm
+		description                string
+	}{
+		{
+			name:                       "WhenRequired_NoDefault",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+			inputChecksumAlgorithm:     "",
+			expectedChecksumAlgorithm:  "",
+			description:                "When RequestChecksumCalculationWhenRequired is set, no default checksum should be applied (backwards compatibility)",
+		},
+		{
+			name:                       "WhenSupported_HasDefault",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenSupported,
+			inputChecksumAlgorithm:     "",
+			expectedChecksumAlgorithm:  types.ChecksumAlgorithmCrc32,
+			description:                "When RequestChecksumCalculationWhenSupported is set, default CRC32 checksum should be applied",
+		},
+		{
+			name:                       "WhenRequired_ExplicitPreserved",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
+			inputChecksumAlgorithm:     types.ChecksumAlgorithmSha256,
+			expectedChecksumAlgorithm:  types.ChecksumAlgorithmSha256,
+			description:                "Explicit checksums should always be preserved regardless of RequestChecksumCalculation setting",
+		},
+		{
+			name:                       "WhenSupported_ExplicitPreserved",
+			requestChecksumCalculation: aws.RequestChecksumCalculationWhenSupported,
+			inputChecksumAlgorithm:     types.ChecksumAlgorithmSha1,
+			expectedChecksumAlgorithm:  types.ChecksumAlgorithmSha1,
+			description:                "Explicit checksums should override defaults even with WhenSupported",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			c, invocations, args := s3testing.NewUploadLoggingClient(nil)
+
+			mgr := manager.NewUploader(c,
+				manager.WithUploaderRequestOptions(func(o *s3.Options) {
+					o.RequestChecksumCalculation = tc.requestChecksumCalculation
+				}))
+
+			input := &s3.PutObjectInput{
+				Bucket: aws.String("Bucket"),
+				Key:    aws.String("Key"),
+				Body:   bytes.NewReader(buf12MB), // Large enough to trigger multipart upload
+			}
+			if tc.inputChecksumAlgorithm != "" {
+				input.ChecksumAlgorithm = tc.inputChecksumAlgorithm
+			}
+
+			resp, err := mgr.Upload(context.Background(), input)
+			if err != nil {
+				t.Errorf("Expected no error but received %v", err)
+			}
+
+			expectedOps := []string{"CreateMultipartUpload", "UploadPart", "UploadPart", "UploadPart", "CompleteMultipartUpload"}
+			if diff := cmpDiff(expectedOps, *invocations); len(diff) > 0 {
+				t.Error(diff)
+			}
+
+			if resp.UploadID != "UPLOAD-ID" {
+				t.Errorf("expect %q, got %q", "UPLOAD-ID", resp.UploadID)
+			}
+
+			cmu := (*args)[0].(*s3.CreateMultipartUploadInput)
+			if cmu.ChecksumAlgorithm != tc.expectedChecksumAlgorithm {
+				t.Errorf("%s: Expected checksum algorithm %v in CreateMultipartUpload, but got %v",
+					tc.description, tc.expectedChecksumAlgorithm, cmu.ChecksumAlgorithm)
+			}
+
+			for i := 1; i <= 3; i++ {
+				uploadPart := (*args)[i].(*s3.UploadPartInput)
+				if uploadPart.ChecksumAlgorithm != tc.expectedChecksumAlgorithm {
+					t.Errorf("%s: Expected checksum algorithm %v in UploadPart %d, but got %v",
+						tc.description, tc.expectedChecksumAlgorithm, i, uploadPart.ChecksumAlgorithm)
+				}
+			}
+		})
+	}
+}
+
 type recordedBufferProvider struct {
 	content       []byte
 	size          int