configuration for checksum algorithm for PutObject to S3

backupstoragelocation.md

@@ -113,7 +113,15 @@ spec:
 
     # Tags that need to be placed on AWS S3 objects. 
     # For example "Key1=Value1&Key2=Value2"
+    #
     # Optional (defaults to empty "")
     tagging: ""
 
+    # The checksum algorithm to use for uploading objects to S3.
+    # The Supported values are  "CRC32",  "CRC32C", "SHA1", "SHA256".
+    # If the value is set as empty string "", no checksum will be calculated and attached to 
+    # the request headers.
+    #
+    # Optional (defaults to "CRC32")
+    checksumAlgorithm: "CRC32"
 ```

velero-plugin-for-aws/object_store.go

@@ -50,6 +50,7 @@ const (
 	caCertKey                    = "caCert"
 	enableSharedConfigKey        = "enableSharedConfig"
 	taggingKey                   = "tagging"
+	checksumAlgKey               = "checksumAlgorithm"
 )
 
 type s3Interface interface {
@@ -73,6 +74,7 @@ type ObjectStore struct {
 	signatureVersion     string
 	serverSideEncryption string
 	tagging              string
+	checksumAlg          string
 }
 
 func newObjectStore(logger logrus.FieldLogger) *ObjectStore {
@@ -94,6 +96,7 @@ func (o *ObjectStore) Init(config map[string]string) error {
 		insecureSkipTLSVerifyKey,
 		enableSharedConfigKey,
 		taggingKey,
+		checksumAlgKey,
 	); err != nil {
 		return err
 	}
@@ -187,9 +190,22 @@ func (o *ObjectStore) Init(config map[string]string) error {
 			return err
 		}
 	}
+	if alg, ok := config[checksumAlgKey]; ok {
+		if !validChecksumAlg(alg) {
+			return errors.Errorf("invalid checksum algorithm: %s", alg)
+		}
+		o.checksumAlg = alg
+	} else {
+		o.checksumAlg = string(types.ChecksumAlgorithmCrc32)
+	}
 	return nil
 }
 
+func validChecksumAlg(alg string) bool {
+	return alg == string(types.ChecksumAlgorithmCrc32) || alg == string(types.ChecksumAlgorithmCrc32c) ||
+		alg == string(types.ChecksumAlgorithmSha1) || alg == string(types.ChecksumAlgorithmSha256) || alg == ""
+}
+
 func readCustomerKey(customerKeyEncryptionFile string) (string, error) {
 	if _, err := os.Stat(customerKeyEncryptionFile); err != nil {
 		if os.IsNotExist(err) {
@@ -241,6 +257,10 @@ func (o *ObjectStore) PutObject(bucket, key string, body io.Reader) error {
 		input.ServerSideEncryption = types.ServerSideEncryption(o.serverSideEncryption)
 	}
 
+	if o.checksumAlg != "" {
+		input.ChecksumAlgorithm = types.ChecksumAlgorithm(o.checksumAlg)
+	}
+
 	_, err := o.s3Uploader.Upload(context.Background(), input)
 
 	return errors.Wrapf(err, "error putting object %s", key)

velero-plugin-for-aws/object_store_test.go

@@ -112,3 +112,43 @@ func TestObjectExists(t *testing.T) {
 		})
 	}
 }
+
+func TestValidChecksumAlg(t *testing.T) {
+	tests := []struct {
+		name     string
+		input    string
+		expected bool
+	}{
+		{
+			name:     "md5 is invalid",
+			input:    "MD5",
+			expected: false,
+		},
+		{
+			name:     "sha256 is invalid",
+			input:    "sha256",
+			expected: false,
+		},
+		{
+			name:     "SHA256 is valid",
+			input:    "SHA256",
+			expected: true,
+		},
+		{
+			name:     "empty is valid",
+			input:    "",
+			expected: true,
+		},
+		{
+			name:     "blank string with space is invalid",
+			input:    "  ",
+			expected: false,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, validChecksumAlg(tc.input))
+		})
+	}
+}