//go:build integration // +build integration package manager_test import ( "bytes" "context" "crypto/md5" "crypto/sha1" "crypto/sha256" "encoding/base64" "encoding/hex" "errors" "fmt" "hash" "hash/crc32" "io" "log" "regexp" "strconv" "testing" "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/feature/s3/manager" "github.com/aws/aws-sdk-go-v2/service/s3" "github.com/aws/aws-sdk-go-v2/service/s3/types" s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/aws/smithy-go/middleware" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" ) var integBuf12MB = make([]byte, 1024*1024*12) var integMD512MB = fmt.Sprintf("%x", md5.Sum(integBuf12MB)) func hexEncodeSum(sum []byte) string { sumHex := make([]byte, hex.EncodedLen(len(sum))) hex.Encode(sumHex, sum) return string(sumHex) } func base64EncodeSum(sum []byte) string { sum64 := make([]byte, base64.StdEncoding.EncodedLen(len(sum))) base64.StdEncoding.Encode(sum64, sum) return string(sum64) } func base64Sum(h hash.Hash, b []byte) string { h.Write(b) return base64EncodeSum(h.Sum(nil)) } func hexSum(h hash.Hash, b []byte) string { h.Write(b) return hexEncodeSum(h.Sum(nil)) } func base64StringDecode(v string) []byte { vv, err := base64.StdEncoding.DecodeString(v) if err != nil { panic(err.Error()) } return vv } func hexStringDecode(v string) []byte { vv, err := base64.StdEncoding.DecodeString(v) if err != nil { panic(err.Error()) } return vv } func base64SumOfSums(h hash.Hash, sums []string) string { for _, v := range sums { h.Write(base64StringDecode(v)) } return base64EncodeSum(h.Sum(nil)) + "-" + strconv.Itoa(len(sums)) } func hexSumOfSums(h hash.Hash, sums []string) string { for _, v := range sums { h.Write(hexStringDecode(unquote(v))) } return hexEncodeSum(h.Sum(nil)) + "-" + strconv.Itoa(len(sums)) } func TestInteg_UploadConcurrently(t *testing.T) { key := "12mb-1" mgr := manager.NewUploader(client) out, err := mgr.Upload(context.Background(), &s3.PutObjectInput{ Bucket: bucketName, Key: &key, Body: bytes.NewReader(integBuf12MB), ChecksumAlgorithm: s3types.ChecksumAlgorithmCrc32c, }) if err != nil { t.Fatalf("expect no error, got %v", err) } if len(out.UploadID) == 0 { t.Errorf("expect upload ID but was empty") } re := regexp.MustCompile(`^https?://.+/` + key + `$`) if e, a := re.String(), out.Location; !re.MatchString(a) { t.Errorf("expect %s to match URL regexp %q, did not", e, a) } validate(t, key, integMD512MB) } func unquote(v string) string { log.Printf("unquote: %v", v) vv, err := strconv.Unquote(v) if err != nil { // Unquote returns error if string doesn't contain quotes if err == strconv.ErrSyntax { return v } panic(err.Error()) } return vv } func TestInteg_UploadPresetChecksum(t *testing.T) { singlePartBytes := integBuf12MB[0:manager.DefaultUploadPartSize] singlePartCRC32 := base64Sum(crc32.NewIEEE(), singlePartBytes) singlePartCRC32C := base64Sum(crc32.New(crc32.MakeTable(crc32.Castagnoli)), singlePartBytes) singlePartSHA1 := base64Sum(sha1.New(), singlePartBytes) singlePartSHA256 := base64Sum(sha256.New(), singlePartBytes) singlePartMD5 := base64Sum(md5.New(), singlePartBytes) singlePartETag := fmt.Sprintf("%q", hexSum(md5.New(), singlePartBytes)) multiPartTailBytes := integBuf12MB[manager.DefaultUploadPartSize*2:] multiPartTailCRC32 := base64Sum(crc32.NewIEEE(), multiPartTailBytes) multiPartTailCRC32C := base64Sum(crc32.New(crc32.MakeTable(crc32.Castagnoli)), multiPartTailBytes) multiPartTailSHA1 := base64Sum(sha1.New(), multiPartTailBytes) multiPartTailSHA256 := base64Sum(sha256.New(), multiPartTailBytes) multiPartTailETag := fmt.Sprintf("%q", hexSum(md5.New(), multiPartTailBytes)) multiPartBytes := integBuf12MB multiPartCRC32 := base64SumOfSums(crc32.NewIEEE(), []string{singlePartCRC32, singlePartCRC32, multiPartTailCRC32}) multiPartCRC32C := base64SumOfSums(crc32.New(crc32.MakeTable(crc32.Castagnoli)), []string{singlePartCRC32C, singlePartCRC32C, multiPartTailCRC32C}) multiPartSHA1 := base64SumOfSums(sha1.New(), []string{singlePartSHA1, singlePartSHA1, multiPartTailSHA1}) multiPartSHA256 := base64SumOfSums(sha256.New(), []string{singlePartSHA256, singlePartSHA256, multiPartTailSHA256}) multiPartETag := `"4e982d58b6c2ce178ae042c23f9bca6e-3"` // Not obvious how this is computed cases := map[string]map[string]struct { algorithm s3types.ChecksumAlgorithm payload io.Reader checksumCRC32 string checksumCRC32C string checksumSHA1 string checksumSHA256 string contentMD5 string expectParts []s3types.CompletedPart expectChecksumCRC32 string expectChecksumCRC32C string expectChecksumSHA1 string expectChecksumSHA256 string expectETag string }{ "auto single part": { "no checksum": { payload: bytes.NewReader(singlePartBytes), expectETag: singlePartETag, }, "CRC32": { algorithm: s3types.ChecksumAlgorithmCrc32, payload: bytes.NewReader(singlePartBytes), expectChecksumCRC32: singlePartCRC32, expectETag: singlePartETag, }, "CRC32C": { algorithm: s3types.ChecksumAlgorithmCrc32c, payload: bytes.NewReader(singlePartBytes), expectChecksumCRC32C: singlePartCRC32C, expectETag: singlePartETag, }, "SHA1": { algorithm: s3types.ChecksumAlgorithmSha1, payload: bytes.NewReader(singlePartBytes), expectChecksumSHA1: singlePartSHA1, expectETag: singlePartETag, }, "SHA256": { algorithm: s3types.ChecksumAlgorithmSha256, payload: bytes.NewReader(singlePartBytes), expectChecksumSHA256: singlePartSHA256, expectETag: singlePartETag, }, }, "preset single part": { "CRC32": { payload: bytes.NewReader(singlePartBytes), checksumCRC32: singlePartCRC32, expectChecksumCRC32: singlePartCRC32, expectETag: singlePartETag, }, "CRC32C": { payload: bytes.NewReader(singlePartBytes), checksumCRC32C: singlePartCRC32C, expectChecksumCRC32C: singlePartCRC32C, expectETag: singlePartETag, }, "SHA1": { payload: bytes.NewReader(singlePartBytes), checksumSHA1: singlePartSHA1, expectChecksumSHA1: singlePartSHA1, expectETag: singlePartETag, }, "SHA256": { payload: bytes.NewReader(singlePartBytes), checksumSHA256: singlePartSHA256, expectChecksumSHA256: singlePartSHA256, expectETag: singlePartETag, }, "MD5": { payload: bytes.NewReader(singlePartBytes), contentMD5: singlePartMD5, expectETag: singlePartETag, }, }, "auto multipart part": { "no checksum": { payload: bytes.NewReader(multiPartBytes), expectParts: []s3types.CompletedPart{ { ETag: aws.String(singlePartETag), PartNumber: 1, }, { ETag: aws.String(singlePartETag), PartNumber: 2, }, { ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectETag: multiPartETag, }, "CRC32": { algorithm: s3types.ChecksumAlgorithmCrc32, payload: bytes.NewReader(multiPartBytes), expectParts: []s3types.CompletedPart{ { ChecksumCRC32: aws.String(singlePartCRC32), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumCRC32: aws.String(singlePartCRC32), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumCRC32: aws.String(multiPartTailCRC32), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumCRC32: multiPartCRC32, expectETag: multiPartETag, }, "CRC32C": { algorithm: s3types.ChecksumAlgorithmCrc32c, payload: bytes.NewReader(multiPartBytes), expectParts: []s3types.CompletedPart{ { ChecksumCRC32C: aws.String(singlePartCRC32C), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumCRC32C: aws.String(singlePartCRC32C), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumCRC32C: aws.String(multiPartTailCRC32C), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumCRC32C: multiPartCRC32C, expectETag: multiPartETag, }, "SHA1": { algorithm: s3types.ChecksumAlgorithmSha1, payload: bytes.NewReader(multiPartBytes), expectParts: []s3types.CompletedPart{ { ChecksumSHA1: aws.String(singlePartSHA1), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumSHA1: aws.String(singlePartSHA1), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumSHA1: aws.String(multiPartTailSHA1), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumSHA1: multiPartSHA1, expectETag: multiPartETag, }, "SHA256": { algorithm: s3types.ChecksumAlgorithmSha256, payload: bytes.NewReader(multiPartBytes), expectParts: []s3types.CompletedPart{ { ChecksumSHA256: aws.String(singlePartSHA256), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumSHA256: aws.String(singlePartSHA256), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumSHA256: aws.String(multiPartTailSHA256), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumSHA256: multiPartSHA256, expectETag: multiPartETag, }, }, "preset multipart part": { "CRC32": { algorithm: s3types.ChecksumAlgorithmCrc32, payload: bytes.NewReader(multiPartBytes), checksumCRC32: multiPartCRC32, expectParts: []s3types.CompletedPart{ { ChecksumCRC32: aws.String(singlePartCRC32), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumCRC32: aws.String(singlePartCRC32), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumCRC32: aws.String(multiPartTailCRC32), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumCRC32: multiPartCRC32, expectETag: multiPartETag, }, "CRC32C": { algorithm: s3types.ChecksumAlgorithmCrc32c, payload: bytes.NewReader(multiPartBytes), checksumCRC32C: multiPartCRC32C, expectParts: []s3types.CompletedPart{ { ChecksumCRC32C: aws.String(singlePartCRC32C), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumCRC32C: aws.String(singlePartCRC32C), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumCRC32C: aws.String(multiPartTailCRC32C), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumCRC32C: multiPartCRC32C, expectETag: multiPartETag, }, "SHA1": { algorithm: s3types.ChecksumAlgorithmSha1, payload: bytes.NewReader(multiPartBytes), checksumSHA1: multiPartSHA1, expectParts: []s3types.CompletedPart{ { ChecksumSHA1: aws.String(singlePartSHA1), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumSHA1: aws.String(singlePartSHA1), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumSHA1: aws.String(multiPartTailSHA1), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumSHA1: multiPartSHA1, expectETag: multiPartETag, }, "SHA256": { algorithm: s3types.ChecksumAlgorithmSha256, payload: bytes.NewReader(multiPartBytes), checksumSHA256: multiPartSHA256, expectParts: []s3types.CompletedPart{ { ChecksumSHA256: aws.String(singlePartSHA256), ETag: aws.String(singlePartETag), PartNumber: 1, }, { ChecksumSHA256: aws.String(singlePartSHA256), ETag: aws.String(singlePartETag), PartNumber: 2, }, { ChecksumSHA256: aws.String(multiPartTailSHA256), ETag: aws.String(multiPartTailETag), PartNumber: 3, }, }, expectChecksumSHA256: multiPartSHA256, expectETag: multiPartETag, }, }, } for group, cs := range cases { t.Run(group, func(t *testing.T) { for name, c := range cs { t.Run(name, func(t *testing.T) { mgr := manager.NewUploader(client) out, err := mgr.Upload(context.Background(), &s3.PutObjectInput{ Bucket: bucketName, Key: aws.String(t.Name()), Body: c.payload, ChecksumAlgorithm: c.algorithm, ChecksumCRC32: toStringPtr(c.checksumCRC32), ChecksumCRC32C: toStringPtr(c.checksumCRC32C), ChecksumSHA1: toStringPtr(c.checksumSHA1), ChecksumSHA256: toStringPtr(c.checksumSHA256), ContentMD5: toStringPtr(c.contentMD5), }) if err != nil { t.Fatalf("expect no error, got %v", err) } if diff := cmp.Diff(c.expectParts, out.CompletedParts, cmpopts.IgnoreUnexported(types.CompletedPart{})); diff != "" { t.Errorf("expect parts match\n%s", diff) } if e, a := c.expectChecksumCRC32, aws.ToString(out.ChecksumCRC32); e != a { t.Errorf("expect %v CRC32 checksum, got %v", e, a) } if e, a := c.expectChecksumCRC32C, aws.ToString(out.ChecksumCRC32C); e != a { t.Errorf("expect %v CRC32C checksum, got %v", e, a) } if e, a := c.expectChecksumSHA1, aws.ToString(out.ChecksumSHA1); e != a { t.Errorf("expect %v SHA1 checksum, got %v", e, a) } if e, a := c.expectChecksumSHA256, aws.ToString(out.ChecksumSHA256); e != a { t.Errorf("expect %v SHA256 checksum, got %v", e, a) } if e, a := c.expectETag, aws.ToString(out.ETag); e != a { t.Errorf("expect %v ETag, got %v", e, a) } }) } }) } } func toStringPtr(v string) *string { if v == "" { return nil } return &v } type invalidateHash struct{} func (b *invalidateHash) ID() string { return "s3manager:InvalidateHash" } func (b *invalidateHash) RegisterMiddleware(stack *middleware.Stack) error { return stack.Serialize.Add(b, middleware.After) } func (b *invalidateHash) HandleSerialize(ctx context.Context, in middleware.SerializeInput, next middleware.SerializeHandler) ( out middleware.SerializeOutput, metadata middleware.Metadata, err error, ) { if input, ok := in.Parameters.(*s3.UploadPartInput); ok && input.PartNumber == 2 { ctx = v4.SetPayloadHash(ctx, "000") } return next.HandleSerialize(ctx, in) } func TestInteg_UploadFailCleanup(t *testing.T) { key := "12mb-leave" mgr := manager.NewUploader(client, func(u *manager.Uploader) { u.LeavePartsOnError = false u.ClientOptions = append(u.ClientOptions, func(options *s3.Options) { options.APIOptions = append(options.APIOptions, (&invalidateHash{}).RegisterMiddleware) }) }) _, err := mgr.Upload(context.Background(), &s3.PutObjectInput{ Bucket: bucketName, Key: &key, Body: bytes.NewReader(integBuf12MB), }) if err == nil { t.Fatalf("expect error, but did not get one") } uploadID := "" var uf manager.MultiUploadFailure if !errors.As(err, &uf) { t.Errorf("") } else if uploadID = uf.UploadID(); len(uploadID) == 0 { t.Errorf("expect upload ID to not be empty, but was") } _, err = client.ListParts(context.Background(), &s3.ListPartsInput{ Bucket: bucketName, Key: &key, UploadId: &uploadID, }) if err == nil { t.Errorf("expect error for list parts, but got none") } }