versitygw/tests/integration/UploadPart.go

// Copyright 2023 Versity Software
// This file is licensed under the Apache License, Version 2.0
// (the "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package integration

import (
	"bytes"
	"context"
	"crypto/md5"
	"crypto/rand"
	"encoding/base64"
	"fmt"
	"hash"
	"hash/crc32"
	"hash/crc64"
	"math/bits"
	"net/http"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/versity/versitygw/s3err"
)

func UploadPart_non_existing_bucket(s *S3Conf) error {
	testName := "UploadPart_non_existing_bucket"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		bucketName := getBucketName()
		partNumber := int32(1)
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err := s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:     &bucketName,
			Key:        getPtr("my-obj"),
			UploadId:   getPtr("uploadId"),
			PartNumber: &partNumber,
		})
		cancel()
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
			return err
		}

		return nil
	})
}

func UploadPart_invalid_part_number(s *S3Conf) error {
	testName := "UploadPart_invalid_part_number"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		key := "my-obj"
		mp, err := createMp(s3client, bucket, key)
		if err != nil {
			return err
		}
		for _, el := range []int32{0, -1, 10001, 2300000} {
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			_, err := s3client.UploadPart(ctx, &s3.UploadPartInput{
				Bucket:     &bucket,
				Key:        &key,
				UploadId:   mp.UploadId,
				PartNumber: &el,
			})
			cancel()
			if err := checkApiErr(err, s3err.GetInvalidArgumentErr(s3err.InvalidArgPartNumber, fmt.Sprint(el))); err != nil {
				return err
			}
		}

		return nil
	})
}

func UploadPart_non_existing_mp_upload(s *S3Conf) error {
	testName := "UploadPart_non_existing_mp_upload"
	partNumber := int32(1)
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err := s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:     &bucket,
			Key:        getPtr("my-obj"),
			UploadId:   getPtr("uploadId"),
			PartNumber: &partNumber,
		})
		cancel()
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil {
			return err
		}
		return nil
	})
}

func UploadPart_upload_id_path_traversal(s *S3Conf) error {
	testName := "UploadPart_upload_id_path_traversal"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		objs, err := putObjects(s3client, []string{"victim-dir/1"}, bucket)
		if err != nil {
			return err
		}

		// A posix upload directory is three levels below its bucket, so
		// part 1 of the upload ID would be written over "victim-dir/1".
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:     &bucket,
			Key:        getPtr("my-obj"),
			UploadId:   getPtr("../../../victim-dir"),
			PartNumber: getPtr(int32(1)),
			Body:       strings.NewReader("part data"),
		})
		cancel()
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil {
			return err
		}

		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
			Bucket: &bucket,
			Key:    objs[0].Key,
		})
		cancel()
		if err != nil {
			return err
		}
		if getString(out.ETag) != getString(objs[0].ETag) {
			return fmt.Errorf("expected the object etag to be %v, instead got %v",
				getString(objs[0].ETag), getString(out.ETag))
		}

		return nil
	})
}

func UploadPart_malformed_upload_id(s *S3Conf) error {
	testName := "UploadPart_malformed_upload_id"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		// A malformed upload ID is reported ahead of an invalid part
		// number and of invalid or mismatched checksums.
		for i, input := range []*s3.UploadPartInput{
			{PartNumber: getPtr(int32(-10))},
			{PartNumber: getPtr(int32(1)), ChecksumCRC32: getPtr("invalid_base64!")},
			{PartNumber: getPtr(int32(1)), ChecksumCRC32: getPtr("AAAAAA==")},
			{PartNumber: getPtr(int32(1)), ContentMD5: getPtr("nZ8ysQ5XDHwZxKA3cpfRYQ==")},
		} {
			input.Bucket = &bucket
			input.Key = getPtr("my-obj")
			input.UploadId = getPtr("../invalid-upload")
			input.Body = strings.NewReader("part data")

			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			_, err := s3client.UploadPart(ctx, input)
			cancel()
			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil {
				return fmt.Errorf("test case %d: %w", i+1, err)
			}
		}

		return nil
	})
}

func UploadPart_multiple_checksum_headers(s *S3Conf) error {
	testName := "UploadPart_multiple_checksum_headers"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"

		mp, err := createMp(s3client, bucket, obj, withChecksum(types.ChecksumAlgorithmCrc32c))
		if err != nil {
			return err
		}

		partNumber := int32(1)

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:         &bucket,
			Key:            &obj,
			ChecksumSHA1:   getPtr("Kq5sNclPz7QV2+lfQIuc6R7oRu0="),
			ChecksumCRC32C: getPtr("m0cB1Q=="),
			UploadId:       mp.UploadId,
			PartNumber:     &partNumber,
		})
		cancel()
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMultipleChecksumHeaders)); err != nil {
			return err
		}

		// multiple empty checksums
		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:         &bucket,
			Key:            &obj,
			ChecksumSHA1:   getPtr(""),
			ChecksumCRC32C: getPtr(""),
			UploadId:       mp.UploadId,
			PartNumber:     &partNumber,
		})
		cancel()
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMultipleChecksumHeaders)); err != nil {
			return err
		}

		return nil
	})
}

func UploadPart_invalid_checksum_header(s *S3Conf) error {
	testName := "UploadPart_invalid_checksum_header"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"

		mp, err := createMp(s3client, bucket, obj)
		if err != nil {
			return err
		}

		partNumber := int32(1)

		for _, algo := range types.ChecksumAlgorithmCrc32.Values() {
			// tests against:
			// - empty string
			// - invalid base64
			// - valid base64, but invalid checksum
			for _, checksum := range []string{"", "invalid_base64!", "c2RhZnNhZGZzZGFm"} {
				input := &s3.UploadPartInput{
					Bucket:     &bucket,
					Key:        &obj,
					PartNumber: &partNumber,
					UploadId:   mp.UploadId,
				}
				setUploadPartChecksum(input, algo, getPtr(checksum))

				ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
				_, err := s3client.UploadPart(ctx, input)
				cancel()
				if err := checkApiErr(err, s3err.GetInvalidChecksumHeaderErr(checksumHeaderName(algo))); err != nil {
					return err
				}
			}
		}

		return nil
	})
}

func UploadPart_checksum_header_and_algo_mismatch(s *S3Conf) error {
	testName := "UploadPart_checksum_header_and_algo_mismatch"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-object"
		mp, err := createMp(s3client, bucket, obj)
		if err != nil {
			return err
		}

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:            &bucket,
			Key:               &obj,
			UploadId:          mp.UploadId,
			PartNumber:        getPtr(int32(1)),
			Body:              strings.NewReader("dummy"),
			ChecksumAlgorithm: types.ChecksumAlgorithmCrc32,
			ChecksumCRC32C:    getPtr("muDarg=="),
		})
		cancel()
		return checkApiErr(err, s3err.GetInvalidChecksumHeaderErr("x-amz-sdk-checksum-algorithm"))
	})
}

func UploadPart_checksum_algorithm_mistmatch_on_initialization(s *S3Conf) error {
	testName := "UploadPart_checksum_algorithm_mistmatch_on_initialization"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		mp, err := createMp(s3client, bucket, obj, withChecksum(types.ChecksumAlgorithmCrc32))
		if err != nil {
			return err
		}

		partNumber := int32(1)

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:            &bucket,
			Key:               &obj,
			UploadId:          mp.UploadId,
			PartNumber:        &partNumber,
			ChecksumAlgorithm: types.ChecksumAlgorithmSha1,
		})
		cancel()
		if err := checkApiErr(err, s3err.GetChecksumTypeMismatchErr(types.ChecksumAlgorithmCrc32, types.ChecksumAlgorithmSha1)); err != nil {
			return err
		}

		return nil
	})
}

func UploadPart_checksum_algorithm_mistmatch_on_initialization_with_value(s *S3Conf) error {
	testName := "UploadPart_checksum_algorithm_mistmatch_on_initialization_with_value"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		mp, err := createMp(s3client, bucket, obj, withChecksum(types.ChecksumAlgorithmCrc32))
		if err != nil {
			return err
		}

		partNumber := int32(1)

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:         &bucket,
			Key:            &obj,
			UploadId:       mp.UploadId,
			PartNumber:     &partNumber,
			ChecksumSHA256: getPtr("uU0nuZNNPgilLlLX2n2r+sSE7+N6U4DukIj3rOLvzek="),
		})
		cancel()
		if err := checkApiErr(err, s3err.GetChecksumTypeMismatchErr(types.ChecksumAlgorithmCrc32, types.ChecksumAlgorithmSha256)); err != nil {
			return err
		}

		return nil
	})
}

func UploadPart_incorrect_checksums(s *S3Conf) error {
	testName := "UploadPart_incorrect_checksums"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"

		for _, algo := range types.ChecksumAlgorithmCrc32.Values() {
			wrongChecksum, err := wrongChecksumForAlgorithm(algo)
			if err != nil {
				return err
			}

			mp, err := createMp(s3client, bucket, obj, withChecksum(algo))
			if err != nil {
				return err
			}

			body := strings.NewReader("random string body")
			partNumber := int32(1)

			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			input := &s3.UploadPartInput{
				Bucket:     &bucket,
				Key:        &obj,
				UploadId:   mp.UploadId,
				PartNumber: &partNumber,
				Body:       body,
			}
			setUploadPartChecksum(input, algo, &wrongChecksum)
			_, err = s3client.UploadPart(ctx, input)
			cancel()
			if err := checkApiErr(err, s3err.GetChecksumBadDigestErr(algo)); err != nil {
				return err
			}
		}

		return nil
	})
}

func UploadPart_no_checksum_with_full_object_checksum_type(s *S3Conf) error {
	testName := "UploadPart_no_checksum_with_full_object_checksum_type"
	return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
		customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
			o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
		})
		obj := "my-obj"

		for _, algo := range []types.ChecksumAlgorithm{
			types.ChecksumAlgorithmCrc32,
			types.ChecksumAlgorithmCrc32c,
			types.ChecksumAlgorithmCrc64nvme,
		} {
			mp, err := createMp(customClient, bucket, obj, withChecksum(algo), withChecksumType(types.ChecksumTypeFullObject))
			if err != nil {
				return err
			}

			var hashRdr hash.Hash

			switch algo {
			case types.ChecksumAlgorithmCrc32:
				hashRdr = crc32.NewIEEE()
			case types.ChecksumAlgorithmCrc32c:
				hashRdr = crc32.New(crc32.MakeTable(crc32.Castagnoli))
			case types.ChecksumAlgorithmCrc64nvme:
				hashRdr = crc64.New(crc64.MakeTable(bits.Reverse64(0xad93d23594c93659)))
			default:
				return fmt.Errorf("invalid checksum algorithm provided: %s", algo)
			}

			partBuffer := make([]byte, 5*1024*1024)
			rand.Read(partBuffer)
			hashRdr.Write(partBuffer)
			partNumber := int32(1)

			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			res, err := customClient.UploadPart(ctx, &s3.UploadPartInput{
				Bucket:     &bucket,
				Key:        &obj,
				UploadId:   mp.UploadId,
				Body:       bytes.NewReader(partBuffer),
				PartNumber: &partNumber,
			})
			cancel()
			if err != nil {
				return err
			}

			csum := base64.StdEncoding.EncodeToString(hashRdr.Sum(nil))

			switch algo {
			case types.ChecksumAlgorithmCrc32:
				if getString(res.ChecksumCRC32) != csum {
					return fmt.Errorf("expected the uploaded part checksum %s to be %s, instead got %s", algo, csum, getString(res.ChecksumCRC32))
				}
			case types.ChecksumAlgorithmCrc32c:
				if getString(res.ChecksumCRC32C) != csum {
					return fmt.Errorf("expected the uploaded part checksum %s to be %s, instead got %s", algo, csum, getString(res.ChecksumCRC32C))
				}
			case types.ChecksumAlgorithmCrc64nvme:
				if getString(res.ChecksumCRC64NVME) != csum {
					return fmt.Errorf("expected the uploaded part checksum %s to be %s, instead got %s", algo, csum, getString(res.ChecksumCRC64NVME))
				}
			}
		}
		return nil
	})
}

func UploadPart_no_checksum_with_composite_checksum_type(s *S3Conf) error {
	testName := "UploadPart_no_checksum_with_composite_checksum_type"
	return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
		customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
			o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
		})
		obj := "my-obj"

		for _, algo := range []types.ChecksumAlgorithm{
			types.ChecksumAlgorithmCrc32,
			types.ChecksumAlgorithmCrc32c,
			types.ChecksumAlgorithmSha1,
			types.ChecksumAlgorithmSha256,
			types.ChecksumAlgorithmSha512,
			types.ChecksumAlgorithmMd5,
			types.ChecksumAlgorithmXxhash64,
			types.ChecksumAlgorithmXxhash3,
			types.ChecksumAlgorithmXxhash128,
		} {
			mp, err := createMp(customClient, bucket, obj, withChecksum(algo), withChecksumType(types.ChecksumTypeComposite))
			if err != nil {
				return err
			}
			_, _, err = uploadParts(customClient, 10, 1, bucket, obj, *mp.UploadId)
			if err := checkApiErr(err, s3err.GetChecksumTypeMismatchErr(algo, "null")); err != nil {
				return err
			}
		}
		return nil
	})
}

func UploadPart_with_checksums_success(s *S3Conf) error {
	testName := "UploadPart_with_checksums_success"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
			o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
		})
		obj := "my-obj"

		for _, test := range []struct {
			chType types.ChecksumType
			chAlgo types.ChecksumAlgorithm
		}{
			{types.ChecksumTypeFullObject, types.ChecksumAlgorithmCrc32},
			{types.ChecksumTypeFullObject, types.ChecksumAlgorithmCrc32c},
			{types.ChecksumTypeFullObject, types.ChecksumAlgorithmCrc64nvme},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmCrc32},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmCrc32c},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmSha1},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmSha256},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmSha512},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmMd5},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmXxhash64},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmXxhash3},
			{types.ChecksumTypeComposite, types.ChecksumAlgorithmXxhash128},
		} {
			mp, err := createMp(customClient, bucket, obj, withChecksum(test.chAlgo), withChecksumType(test.chType))
			if err != nil {
				return err
			}

			parts, csum, err := uploadParts(customClient, 5*1024*1024, 1, bucket, obj, *mp.UploadId, withChecksum(test.chAlgo))
			if err != nil {
				return err
			}

			if len(parts) != 1 {
				return fmt.Errorf("expected 1 uploaded part, instaed got %d", len(parts))
			}

			if got := getString(getPartChecksum(parts[0], test.chAlgo)); got != csum {
				return fmt.Errorf("expected the uploaded part checksum %s to be %s, instead got %s", test.chAlgo, csum, got)
			}
		}
		return nil
	})
}

func UploadPart_non_existing_key(s *S3Conf) error {
	testName := "UploadPart_non_existing_key"
	partNumber := int32(1)
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		out, err := createMp(s3client, bucket, obj)
		if err != nil {
			return err
		}

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:     &bucket,
			Key:        getPtr("non-existing-object-key"),
			UploadId:   out.UploadId,
			PartNumber: &partNumber,
		})
		cancel()
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil {
			return err
		}
		return nil
	})
}

func UploadPart_success(s *S3Conf) error {
	testName := "UploadPart_success"
	partNumber := int32(1)
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		out, err := createMp(s3client, bucket, obj)
		if err != nil {
			return err
		}
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := s3client.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:     &bucket,
			Key:        &obj,
			UploadId:   out.UploadId,
			PartNumber: &partNumber,
		})
		cancel()
		if err != nil {
			return err
		}
		if getString(res.ETag) == "" {
			return fmt.Errorf("expected a valid etag, instead got empty")
		}
		return nil
	})
}

// UploadPart_plain_body_with_decoded_length checks that a complete plain part
// uses Content-Length rather than X-Amz-Decoded-Content-Length. The latter is
// only meaningful for aws-chunked uploads and must not preallocate a zero tail.
func UploadPart_plain_body_with_decoded_length(s *S3Conf) error {
	testName := "UploadPart_plain_body_with_decoded_length"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "plain-part-with-decoded-length"
		upload, err := createMp(s3client, bucket, obj)
		if err != nil {
			return err
		}

		data := "hello"
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		req, err := http.NewRequestWithContext(ctx, http.MethodPut,
			fmt.Sprintf("%s/%s/%s?uploadId=%s&partNumber=1", s.endpoint, bucket, obj, *upload.UploadId),
			strings.NewReader(data))
		if err != nil {
			cancel()
			return fmt.Errorf("create upload part request: %w", err)
		}
		req.Header.Set("x-amz-content-sha256", "UNSIGNED-PAYLOAD")
		req.Header.Set("X-Amz-Decoded-Content-Length", "99999")

		signer := v4.NewSigner()
		if err := signer.SignHTTP(req.Context(),
			aws.Credentials{AccessKeyID: s.awsID, SecretAccessKey: s.awsSecret},
			req, "UNSIGNED-PAYLOAD", "s3", s.awsRegion, time.Now()); err != nil {
			cancel()
			return fmt.Errorf("sign upload part request: %w", err)
		}

		resp, err := s.httpClient.Do(req)
		cancel()
		if err != nil {
			return fmt.Errorf("send upload part request: %w", err)
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusOK {
			return fmt.Errorf("expected upload part status %v, got %v", http.StatusOK, resp.StatusCode)
		}

		partNumber := int32(1)
		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
			Bucket:   &bucket,
			Key:      &obj,
			UploadId: upload.UploadId,
			MultipartUpload: &types.CompletedMultipartUpload{Parts: []types.CompletedPart{{
				ETag:       aws.String(resp.Header.Get("ETag")),
				PartNumber: &partNumber,
			}}},
		})
		cancel()
		if err != nil {
			return fmt.Errorf("complete multipart upload: %w", err)
		}

		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		out, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: &bucket, Key: &obj})
		cancel()
		if err != nil {
			return err
		}
		if out.ContentLength == nil || *out.ContentLength != int64(len(data)) {
			return fmt.Errorf("expected completed object to be %v bytes, got %v", len(data), out.ContentLength)
		}
		return nil
	})
}

func UploadPart_data_integrity_etag(s *S3Conf) error {
	testName := "UploadPart_data_integrity_etag"
	partNumber := int32(1)
	return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
		customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
			o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
		})

		obj := "my-obj"
		out, err := createMp(customClient, bucket, obj)
		if err != nil {
			return err
		}

		payload := []byte("payload-data")
		h := crc64.New(crc64.MakeTable(bits.Reverse64(0xad93d23594c93659)))
		h.Write(payload)
		expectedChecksum := base64.StdEncoding.EncodeToString(h.Sum(nil))
		expectedETag := fmt.Sprintf("\"CRC64NVME-%s\"", expectedChecksum)

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := customClient.UploadPart(ctx, &s3.UploadPartInput{
			Bucket:     &bucket,
			Key:        &obj,
			UploadId:   out.UploadId,
			PartNumber: &partNumber,
			Body:       bytes.NewReader(payload),
		})
		cancel()
		if err != nil {
			return err
		}

		if got := getString(res.ETag); got != expectedETag {
			return fmt.Errorf("expected UploadPart ETag to be %s, instead got %s", expectedETag, got)
		}
		if res.ChecksumCRC64NVME != nil && getString(res.ChecksumCRC64NVME) != expectedChecksum {
			return fmt.Errorf("expected UploadPart checksum to be %s, instead got %s", expectedChecksum, getString(res.ChecksumCRC64NVME))
		}

		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		listOut, err := customClient.ListParts(ctx, &s3.ListPartsInput{
			Bucket:   &bucket,
			Key:      &obj,
			UploadId: out.UploadId,
		})
		cancel()
		if err != nil {
			return err
		}

		if len(listOut.Parts) != 1 {
			return fmt.Errorf("expected 1 uploaded part, instead got %d", len(listOut.Parts))
		}
		if got := getString(listOut.Parts[0].ETag); got != expectedETag {
			return fmt.Errorf("expected ListParts ETag to be %s, instead got %s", expectedETag, got)
		}

		return nil
	})
}

// isQuotedEtag reports whether an ETag is a non-empty double-quoted string,
// as required by the S3 contract (e.g. "\"abc\"").
func isQuotedEtag(etag string) bool {
	return len(etag) >= 3 &&
		strings.HasPrefix(etag, "\"") &&
		strings.HasSuffix(etag, "\"")
}

// UploadPart_etag_quoting_consistency verifies that multipart part ETags are
// returned as quoted strings and stay consistent across the UploadPart
// response, ListParts, and CompleteMultipartUpload. This is the S3 contract
// (ETags are quoted) and matches the posix backend's GenerateEtag convention;
// it is the regression guard for the Azure backend, which previously returned
// raw, unquoted block ids.
func UploadPart_etag_quoting_consistency(s *S3Conf) error {
	testName := "UploadPart_etag_quoting_consistency"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		out, err := createMp(s3client, bucket, obj)
		if err != nil {
			return err
		}

		parts, _, err := uploadParts(s3client, 18*1024*1024, 3, bucket, obj, *out.UploadId)
		if err != nil {
			return err
		}

		// Every ETag returned by UploadPart must be a quoted string.
		for _, p := range parts {
			etag := getString(p.ETag)
			if !isQuotedEtag(etag) {
				return fmt.Errorf("expected UploadPart etag to be quoted, instead got %q", etag)
			}
		}

		// ListParts must report the same quoted ETags as UploadPart.
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := s3client.ListParts(ctx, &s3.ListPartsInput{
			Bucket:   &bucket,
			Key:      &obj,
			UploadId: out.UploadId,
		})
		cancel()
		if err != nil {
			return err
		}
		for _, p := range res.Parts {
			etag := getString(p.ETag)
			if !isQuotedEtag(etag) {
				return fmt.Errorf("expected ListParts etag to be quoted, instead got %q", etag)
			}
		}
		if ok := compareParts(parts, res.Parts); !ok {
			return fmt.Errorf("expected ListParts parts %+v to match UploadPart parts %+v",
				res.Parts, parts)
		}

		// CompleteMultipartUpload must accept the quoted ETags returned above.
		compParts := []types.CompletedPart{}
		for _, p := range parts {
			compParts = append(compParts, types.CompletedPart{
				ETag:       p.ETag,
				PartNumber: p.PartNumber,
			})
		}
		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
			Bucket:   &bucket,
			Key:      &obj,
			UploadId: out.UploadId,
			MultipartUpload: &types.CompletedMultipartUpload{
				Parts: compParts,
			},
		})
		cancel()
		if err != nil {
			return fmt.Errorf("complete multipart upload with quoted etags: %w", err)
		}

		return nil
	})
}

// UploadPart_object_lock_checksum_required covers parts of multipart uploads
// created with Object Lock parameters, a legal hold of either status or a
// retention: each part needs an integrity check of its body, so one with
// neither Content-MD5 nor a checksum is rejected and one with Content-MD5
// is stored. A part of an upload created without them needs none.
func UploadPart_object_lock_checksum_required(s *S3Conf) error {
	testName := "UploadPart_object_lock_checksum_required"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		data := []byte("data")
		md5sum := md5.Sum(data)
		contentMD5 := base64.StdEncoding.EncodeToString(md5sum[:])
		retainUntilDate := time.Now().Add(time.Hour)

		for _, test := range []struct {
			name       string
			input      s3.CreateMultipartUploadInput
			lockParams bool
		}{
			{
				name: "no lock parameters",
			},
			{
				name:       "legal hold ON",
				input:      s3.CreateMultipartUploadInput{ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn},
				lockParams: true,
			},
			{
				name:       "legal hold OFF",
				input:      s3.CreateMultipartUploadInput{ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOff},
				lockParams: true,
			},
			{
				name: "retention",
				input: s3.CreateMultipartUploadInput{
					ObjectLockMode:            types.ObjectLockModeGovernance,
					ObjectLockRetainUntilDate: &retainUntilDate,
				},
				lockParams: true,
			},
		} {
			input := test.input
			input.Bucket = &bucket
			input.Key = &obj
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			mp, err := s3client.CreateMultipartUpload(ctx, &input)
			cancel()
			if err != nil {
				return fmt.Errorf("%s: %w", test.name, err)
			}

			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
			_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
				Bucket:     &bucket,
				Key:        &obj,
				UploadId:   mp.UploadId,
				PartNumber: getPtr(int32(1)),
				Body:       bytes.NewReader(data),
			})
			cancel()
			if test.lockParams {
				if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrObjectLockPartChecksumRequired)); err != nil {
					return fmt.Errorf("%s, part without an integrity check: %w", test.name, err)
				}
			} else if err != nil {
				return fmt.Errorf("%s, part without an integrity check: %w", test.name, err)
			}

			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
			_, err = s3client.UploadPart(ctx, &s3.UploadPartInput{
				Bucket:     &bucket,
				Key:        &obj,
				UploadId:   mp.UploadId,
				PartNumber: getPtr(int32(2)),
				Body:       bytes.NewReader(data),
				ContentMD5: &contentMD5,
			})
			cancel()
			if err != nil {
				return fmt.Errorf("%s, part with Content-MD5: %w", test.name, err)
			}

			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
			_, err = s3client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
				Bucket:   &bucket,
				Key:      &obj,
				UploadId: mp.UploadId,
			})
			cancel()
			if err != nil {
				return err
			}
		}

		return nil
	}, withLock())
}

// UploadPart_default_retention_checksum_required covers parts of multipart
// uploads to a bucket with a default retention rule. Whether an upload
// carries Object Lock parameters is settled when it is created: the parts of
// one created under the rule need an integrity check of their body even
// after the rule is removed, and the parts of one created before the rule
// was set need none.
func UploadPart_default_retention_checksum_required(s *S3Conf) error {
	testName := "UploadPart_default_retention_checksum_required"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		putLockConfig := func(rule *types.ObjectLockRule) error {
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			_, err := s3client.PutObjectLockConfiguration(ctx, &s3.PutObjectLockConfigurationInput{
				Bucket: &bucket,
				ObjectLockConfiguration: &types.ObjectLockConfiguration{
					ObjectLockEnabled: types.ObjectLockEnabledEnabled,
					Rule:              rule,
				},
			})
			cancel()
			return err
		}

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		before, err := s3client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
			Bucket: &bucket,
			Key:    &obj,
		})
		cancel()
		if err != nil {
			return err
		}

		if err := putLockConfig(&types.ObjectLockRule{
			DefaultRetention: &types.DefaultRetention{
				Mode: types.ObjectLockRetentionModeGovernance,
				Days: getPtr(int32(1)),
			},
		}); err != nil {
			return err
		}

		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		under, err := s3client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{
			Bucket: &bucket,
			Key:    &obj,
		})
		cancel()
		if err != nil {
			return err
		}

		if err := putLockConfig(nil); err != nil {
			return err
		}

		uploadPart := func(uploadId *string) error {
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			_, err := s3client.UploadPart(ctx, &s3.UploadPartInput{
				Bucket:     &bucket,
				Key:        &obj,
				UploadId:   uploadId,
				PartNumber: getPtr(int32(1)),
				Body:       bytes.NewReader([]byte("data")),
			})
			cancel()
			return err
		}

		if err := uploadPart(before.UploadId); err != nil {
			return fmt.Errorf("part of the upload created before the rule: %w", err)
		}
		err = uploadPart(under.UploadId)
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrObjectLockPartChecksumRequired)); err != nil {
			return fmt.Errorf("part of the upload created under the rule: %w", err)
		}

		for _, uploadId := range []*string{before.UploadId, under.UploadId} {
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			_, err := s3client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
				Bucket:   &bucket,
				Key:      &obj,
				UploadId: uploadId,
			})
			cancel()
			if err != nil {
				return err
			}
		}

		return nil
	}, withLock())
}