// 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"
	"fmt"
	"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"
	"golang.org/x/sync/errgroup"
)

func PutObject_non_existing_bucket(s *S3Conf) error {
	testName := "PutObject_non_existing_bucket"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		_, err := putObjects(s3client, []string{"my-obj"}, "non-existing-bucket")
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchBucket)); err != nil {
			return err
		}
		return nil
	})
}

func PutObject_special_chars(s *S3Conf) error {
	testName := "PutObject_special_chars"

	objnames := []string{
		"my!key", "my-key", "my_key", "my.key", "my'key", "my(key", "my)key",
		"my&key", "my@key", "my=key", "my;key", "my:key", "my key", "my,key",
		"my?key", "my^key", "my{}key", "my%key", "my`key",
		"my[]key", "my~key", "my<>key", "my|key", "my#key",
	}
	if !s.azureTests {
		// azure currently can't handle backslashes in object names
		objnames = append(objnames, "my\\key")
	}

	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		objs, err := putObjects(s3client, objnames, bucket)
		if err != nil {
			return err
		}
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
			Bucket: &bucket,
		})
		cancel()
		if err != nil {
			return err
		}

		if !compareObjects(objs, res.Contents) {
			return fmt.Errorf("expected the objects to be %vß, instead got %v",
				objStrings(objs), objStrings(res.Contents))
		}

		return nil
	})
}

func PutObject_tagging(s *S3Conf) error {
	testName := "PutObject_tagging"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		testTagging := func(object, taggging string, result map[string]string, expectedErr error) error {
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)

			_, err := s3client.PutObject(ctx, &s3.PutObjectInput{
				Bucket:  &bucket,
				Key:     &object,
				Tagging: &taggging,
			})
			cancel()
			if err == nil && expectedErr != nil {
				return fmt.Errorf("expected err %w, instead got nil", expectedErr)
			}
			if err != nil {
				if expectedErr == nil {
					return err
				}
				switch eErr := expectedErr.(type) {
				case s3err.APIError:
					return checkApiErr(err, eErr)
				default:
					return fmt.Errorf("invalid err provided: %w", expectedErr)
				}
			}

			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
			res, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
				Bucket: &bucket,
				Key:    &object,
			})
			cancel()
			if err != nil {
				return err
			}

			if len(res.TagSet) != len(result) {
				return fmt.Errorf("tag lengths are not equal: (expected): %v, (got): %v", len(result), len(res.TagSet))
			}

			for _, tag := range res.TagSet {
				val, ok := result[getString(tag.Key)]
				if !ok {
					return fmt.Errorf("tag key not found: %v", getString(tag.Key))
				}

				if val != getString(tag.Value) {
					return fmt.Errorf("expected the %v tag value to be %v, instead got %v", getString(tag.Key), val, getString(tag.Value))
				}
			}

			return nil
		}

		fileObj, dirObj := "file-object", "dir-object/"

		for i, el := range []struct {
			tagging     string
			result      map[string]string
			expectedErr error
		}{
			// success cases
			{"&", map[string]string{}, nil},
			{"&&&", map[string]string{}, nil},
			{"key", map[string]string{"key": ""}, nil},
			{"key&", map[string]string{"key": ""}, nil},
			{"key=&", map[string]string{"key": ""}, nil},
			{"key=val&", map[string]string{"key": "val"}, nil},
			{"key1&key2", map[string]string{"key1": "", "key2": ""}, nil},
			{"key1=val1&key2=val2", map[string]string{"key1": "val1", "key2": "val2"}, nil},
			{"key@=val@", map[string]string{"key@": "val@"}, nil},
			// invalid url-encoded
			{"=", nil, s3err.GetAPIError(s3err.ErrInvalidURLEncodedTagging)},
			{"key%", nil, s3err.GetAPIError(s3err.ErrInvalidURLEncodedTagging)},
			// duplicate keys
			{"key=val&key=val", nil, s3err.GetAPIError(s3err.ErrInvalidURLEncodedTagging)},
			// invalid tag keys
			{"key?=val", nil, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
			{"key(=val", nil, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
			{"key*=val", nil, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
			{"key$=val", nil, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
			{"key#=val", nil, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
			{"key!=val", nil, s3err.GetAPIError(s3err.ErrInvalidTagKey)},
			// invalid tag values
			{"key=val?", nil, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
			{"key=val(", nil, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
			{"key=val*", nil, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
			{"key=val$", nil, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
			{"key=val#", nil, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
			{"key=val!", nil, s3err.GetAPIError(s3err.ErrInvalidTagValue)},
			// success special chars
			{"key-key_key.key/key=value-value_value.value/value", map[string]string{"key-key_key.key/key": "value-value_value.value/value"}, nil},
			// should handle supported encoded characters
			{"key%2E=value%2F", map[string]string{"key.": "value/"}, nil},
			{"key%2D=value%2B", map[string]string{"key-": "value+"}, nil},
			{"key++key=value++value", map[string]string{"key  key": "value  value"}, nil},
			{"key%20key=value%20value", map[string]string{"key key": "value value"}, nil},
			{"key%5Fkey=value%5Fvalue", map[string]string{"key_key": "value_value"}, nil},
		} {
			if s.azureTests {
				// azure doesn't support '@' character
				if strings.Contains(el.tagging, "@") {
					continue
				}
			}
			// once test for file object
			err := testTagging(fileObj, el.tagging, el.result, el.expectedErr)
			if err != nil {
				return fmt.Errorf("test case %v failed for file object: %w", i+1, err)
			}

			// the test for directory object
			err = testTagging(dirObj, el.tagging, el.result, el.expectedErr)
			if err != nil {
				return fmt.Errorf("test case %v failed for directory object: %w", i+1, err)
			}
		}
		return nil
	})
}

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

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err := s3client.PutObject(ctx, &s3.PutObjectInput{
			Bucket:         &bucket,
			Key:            &key,
			ObjectLockMode: types.ObjectLockModeCompliance,
		})
		cancel()
		if err := checkSdkApiErr(err, "InvalidRequest"); err != nil {
			return err
		}
		// client sdk regression issue prevents getting full error message,
		// change back to below once this is fixed:
		// https://github.com/aws/aws-sdk-go-v2/issues/2921
		// if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrObjectLockInvalidHeaders)); err != nil {
		// 	return err
		// }

		retainDate := time.Now().Add(time.Hour * 48)

		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
		_, err = s3client.PutObject(ctx, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       &key,
			ObjectLockRetainUntilDate: &retainDate,
		})
		cancel()
		if err := checkSdkApiErr(err, "InvalidRequest"); err != nil {
			return err
		}
		// client sdk regression issue prevents getting full error message,
		// change back to below once this is fixed:
		// https://github.com/aws/aws-sdk-go-v2/issues/2921
		// if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrObjectLockInvalidHeaders)); err != nil {
		// 	return err
		// }

		return nil
	})
}

func PutObject_with_object_lock(s *S3Conf) error {
	testName := "PutObject_with_object_lock"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		retainUntilDate := time.Now().AddDate(1, 0, 0)

		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       &obj,
			ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
			ObjectLockMode:            types.ObjectLockModeCompliance,
			ObjectLockRetainUntilDate: &retainUntilDate,
		}, s3client)
		if err != nil {
			return 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.ObjectLockMode != types.ObjectLockModeCompliance {
			return fmt.Errorf("expected object lock mode to be %v, instead got %v", types.ObjectLockModeCompliance, out.ObjectLockMode)
		}
		if out.ObjectLockLegalHoldStatus != types.ObjectLockLegalHoldStatusOn {
			return fmt.Errorf("expected object lock mode to be %v, instead got %v", types.ObjectLockLegalHoldStatusOn, out.ObjectLockLegalHoldStatus)
		}

		return cleanupLockedObjects(s3client, bucket, []objToDelete{{key: obj, removeLegalHold: true, isCompliance: true}})
	}, withLock())
}

func PutObject_missing_bucket_lock(s *S3Conf) error {
	testName := "PutObject_missing_bucket_lock"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		// with retention
		_, err := putObjectWithData(3, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       getPtr("my-object"),
			ObjectLockMode:            types.ObjectLockModeGovernance,
			ObjectLockRetainUntilDate: getPtr(time.Now().AddDate(0, 0, 2)),
		}, s3client)
		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)); err != nil {
			return err
		}

		// with legal hold
		_, err = putObjectWithData(2, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       getPtr("my-object"),
			ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
		}, s3client)
		return checkApiErr(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces))
	})
}

func PutObject_invalid_legal_hold(s *S3Conf) error {
	testName := "PutObject_invalid_legal_hold"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       getPtr("foo"),
			ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatus("invalid_status"),
		}, s3client)
		return checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidLegalHoldStatus))
	}, withLock())
}

func PutObject_invalid_object_lock_mode(s *S3Conf) error {
	testName := "PutObject_invalid_object_lock_mode"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		rDate := time.Now().Add(time.Hour * 10)
		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       getPtr("foo"),
			ObjectLockRetainUntilDate: &rDate,
			ObjectLockMode:            types.ObjectLockMode("invalid_mode"),
		}, s3client)
		return checkApiErr(err, s3err.GetAPIError(s3err.ErrInvalidObjectLockMode))
	}, withLock())
}

func PutObject_past_retain_until_date(s *S3Conf) error {
	testName := "PutObject_past_retain_until_date"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		rDate := time.Now().AddDate(-1, 0, 0)
		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket:                    &bucket,
			Key:                       getPtr("my-object"),
			ObjectLockMode:            types.ObjectLockModeGovernance,
			ObjectLockRetainUntilDate: &rDate,
		}, s3client)

		return checkApiErr(err, s3err.GetAPIError(s3err.ErrPastObjectLockRetainDate))
	}, withLock())
}

func PutObject_invalid_retain_until_date(s *S3Conf) error {
	testName := "PutObject_invalid_retain_until_date"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		req, err := createSignedReq(
			http.MethodPut,
			s.endpoint,
			fmt.Sprintf("%s/my-object", bucket),
			s.awsID,
			s.awsSecret,
			"s3",
			s.awsRegion,
			nil,
			time.Now(),
			map[string]string{
				"x-amz-object-lock-retain-until-date": "invalid_date",
				"x-amz-object-lock-mode":              "GOVERNANCE",
			},
		)
		if err != nil {
			return err
		}

		resp, err := s.httpClient.Do(req)
		if err != nil {
			return err
		}

		return checkHTTPResponseApiErr(resp, s3err.GetAPIError(s3err.ErrInvalidRetainUntilDate))
	}, withLock())
}

func PutObject_conditional_writes(s *S3Conf) error {
	testName := "PutObject_conditional_writes"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-obj"
		res, err := putObjectWithData(0, &s3.PutObjectInput{
			Bucket: &bucket,
			Key:    &obj,
			Body:   bytes.NewReader([]byte("dummy")),
		}, s3client)
		if err != nil {
			return err
		}

		etag := res.res.ETag
		etagTrimmed := strings.Trim(*etag, `"`)
		incorrectEtag := getPtr("incorrect_etag")
		errPrecond := s3err.GetAPIError(s3err.ErrPreconditionFailed)
		errNoSuchKey := s3err.GetAPIError(s3err.ErrNoSuchKey)
		errNotImplemented := s3err.GetAPIError(s3err.ErrNotImplemented)

		for i, test := range []struct {
			obj         string
			ifMatch     *string
			ifNoneMatch *string
			err         error
		}{
			{obj, etag, nil, nil},
			{obj, etag, etag, errNotImplemented},
			{obj, etag, incorrectEtag, errNotImplemented},
			{obj, incorrectEtag, incorrectEtag, errNotImplemented},
			{obj, incorrectEtag, etag, errNotImplemented},
			{obj, incorrectEtag, nil, errPrecond},
			{obj, nil, incorrectEtag, errNotImplemented},
			{obj, nil, etag, errNotImplemented},
			{obj, nil, getPtr("*"), errPrecond},
			{obj, etag, getPtr("*"), errNotImplemented},
			{obj, nil, nil, nil},

			// precondition headers without quotes
			{obj, &etagTrimmed, nil, nil},
			{obj, &etagTrimmed, &etagTrimmed, errNotImplemented},
			{obj, &etagTrimmed, incorrectEtag, errNotImplemented},
			{obj, incorrectEtag, &etagTrimmed, errNotImplemented},
			{obj, nil, &etagTrimmed, errNotImplemented},

			// object deson't exist tests
			{"obj-1", incorrectEtag, etag, errNotImplemented},
			{"obj-2", etag, etag, errNotImplemented},
			{"obj-3", etag, nil, errNoSuchKey},
			{"obj-4", etag, incorrectEtag, errNotImplemented},
			{"obj-5", incorrectEtag, nil, errNoSuchKey},
			{"obj-6", nil, etag, errNotImplemented},
			{"obj-7", nil, getPtr("*"), nil},
			{"obj-8", etag, getPtr("*"), errNotImplemented},
		} {
			res, err := putObjectWithData(0, &s3.PutObjectInput{
				Bucket:      &bucket,
				Key:         &test.obj,
				Body:        bytes.NewReader([]byte("dummy")),
				IfMatch:     test.ifMatch,
				IfNoneMatch: test.ifNoneMatch,
			}, s3client)
			if err == nil {
				// azure blob storage generates different ETags for
				// the exact same data.
				// to avoid ETag collision reassign the etag value
				*etag = *res.res.ETag
				etagTrimmed = strings.Trim(*res.res.ETag, `"`)
			}
			if test.err == nil && err != nil {
				return fmt.Errorf("test case %v: expected no error, instead got %w", i, err)
			}
			if test.err != nil {
				apierr, ok := test.err.(s3err.APIError)
				if !ok {
					return fmt.Errorf("test case %v: invalid error type: %w", i, test.err)
				}

				if err := checkApiErr(err, apierr); err != nil {
					return fmt.Errorf("test case %v: %w", i, err)
				}
			}
		}

		return nil
	})
}

func PutObject_should_combine_metadata(s *S3Conf) error {
	testName := "PutObject_should_combine_metadata"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-object"
		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		req, err := http.NewRequestWithContext(ctx, http.MethodPut, fmt.Sprintf("%s/%s/%s", s.endpoint, bucket, obj), strings.NewReader("dummy data"))
		if err != nil {
			return err
		}

		req.Header.Set("x-amz-content-sha256", "UNSIGNED-PAYLOAD")
		req.Header.Add("x-amz-meta-key", "value-1")
		req.Header.Add("x-amz-meta-key", "value-2")
		req.Header.Add("x-amz-meta-key", "value-3")
		req.Header.Add("x-amz-meta-Key", "value-4")
		req.Header.Add("x-amz-meta-keY", "value-5")
		req.Header.Add("x-amz-meta-foo", "bar")
		req.Header.Add("x-amz-meta-baz", "abc")
		req.Header.Add("x-amz-meta-xyzz", "xxx-3")
		req.Header.Add("x-amz-meta-Xyzz", "xxx-1")
		req.Header.Add("x-amz-meta-xyzz", "xxx-2")
		req.Header.Add("x-amz-meta-hello", "world")
		req.Header.Add("x-amz-meta-boo", "bar")
		req.Header.Add("x-amz-meta-asdf", "ghk")

		signer := v4.NewSigner()

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

		resp, err := s.httpClient.Do(req)
		cancel()
		if err != nil {
			return fmt.Errorf("send request: %w", err)
		}

		if resp.StatusCode != http.StatusOK {
			return fmt.Errorf("expected the response status code to be %v, instead got %v", http.StatusOK, resp.StatusCode)
		}

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

		expectedMeta := map[string]string{
			"key":   "value-1,value-2,value-3,value-4,value-5",
			"foo":   "bar",
			"baz":   "abc",
			"xyzz":  "xxx-3,xxx-1,xxx-2",
			"hello": "world",
			"boo":   "bar",
			"asdf":  "ghk",
		}

		if !areMapsSame(expectedMeta, out.Metadata) {
			return fmt.Errorf("expected the object metadata to be %v, instead got %v", expectedMeta, out.Metadata)
		}

		return nil
	})
}

func PutObject_long_metadata(s *S3Conf) error {
	testName := "PutObject_long_metadata"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket: &bucket,
			Key:    getPtr("obj"),
			Metadata: map[string]string{
				"key":            "value",
				"foo":            "bar",
				"baz":            "quxx",
				"something_long": strings.Repeat("a", 2048),
			},
		}, s3client)

		return checkApiErr(err, s3err.GetAPIError(s3err.ErrMetadataTooLarge))
	})
}

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

		inputMeta := map[string]string{
			"Key":                    "Val",
			"X-Test":                 "Example",
			"UPPERCASE":              "should-remain",
			"MiXeD-CaSe":             "normalize-to-lower",
			"with-number-123":        "numeric-test",
			"123numeric-prefix":      "value123",
			"key_with_underscore":    "underscore-ok",
			"key-with-dash":          "dash-ok",
			"key.with.dot":           "dot-ok",
			"KeyURL":                 "https://example.com/test?query=1",
			"EmptyValue":             "",
			strings.Repeat("a", 256): "some long metadata value to ensure nothing breaks at higher header sizes",
			"WhitespaceKey ":         " trailing-key",
		}

		expectedMeta := map[string]string{
			"key":                    "Val",
			"x-test":                 "Example",
			"uppercase":              "should-remain",
			"mixed-case":             "normalize-to-lower",
			"with-number-123":        "numeric-test",
			"123numeric-prefix":      "value123",
			"key_with_underscore":    "underscore-ok",
			"key-with-dash":          "dash-ok",
			"key.with.dot":           "dot-ok",
			"keyurl":                 "https://example.com/test?query=1",
			"emptyvalue":             "",
			strings.Repeat("a", 256): "some long metadata value to ensure nothing breaks at higher header sizes",
			"whitespacekey":          "trailing-key",
		}

		for i, test := range []struct {
			obj        string
			dataLength int64
		}{
			// test for file object
			{"file-object", 100},
			// test for directory object
			{"dir-object/", 0},
		} {
			_, err := putObjectWithData(test.dataLength, &s3.PutObjectInput{
				Bucket:   &bucket,
				Key:      &test.obj,
				Metadata: inputMeta,
			}, s3client)
			if err != nil {
				return fmt.Errorf("test %v failed: %w", i+1, err)
			}

			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
				Bucket: &bucket,
				Key:    &test.obj,
			})
			cancel()
			if err != nil {
				return fmt.Errorf("test %v failed: %w", i+1, err)
			}

			if !areMapsSame(expectedMeta, res.Metadata) {
				return fmt.Errorf("test %v failed: expected the object metadata to be %v, instead got %v", i+1, expectedMeta, res.Metadata)
			}
		}

		return nil
	})
}

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

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		_, err := s3client.PutObject(ctx, &s3.PutObjectInput{
			Bucket:            &bucket,
			Key:               &obj,
			ChecksumAlgorithm: types.ChecksumAlgorithmCrc32,
			ChecksumCRC32C:    getPtr("m0cB1Q=="),
		})
		cancel()
		// FIXME: The error message for PutObject is not properly serialized by the sdk
		// References to aws sdk issue https://github.com/aws/aws-sdk-go-v2/issues/2921

		// if err := checkApiErr(err, s3err.GetInvalidChecksumHeaderErr("x-amz-sdk-checksum-algorithm"); err != nil {
		// 	return err
		// }
		if err := checkSdkApiErr(err, "InvalidRequest"); err != nil {
			return err
		}

		return nil
	})
}

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

		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket:         &bucket,
			Key:            &obj,
			ChecksumSHA1:   getPtr("Kq5sNclPz7QV2+lfQIuc6R7oRu0="),
			ChecksumCRC32C: getPtr("m0cB1Q=="),
		}, s3client)
		// FIXME: The error message for PutObject is not properly serialized by the sdk
		// References to aws sdk issue https://github.com/aws/aws-sdk-go-v2/issues/2921

		// if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMultipleChecksumHeaders)); err != nil {
		// 	return err
		// }
		if err := checkSdkApiErr(err, "InvalidRequest"); err != nil {
			return err
		}

		// Empty checksums case
		_, err = putObjectWithData(10, &s3.PutObjectInput{
			Bucket:         &bucket,
			Key:            &obj,
			ChecksumSHA1:   getPtr(""),
			ChecksumCRC32C: getPtr(""),
		}, s3client)
		// FIXME: The error message for PutObject is not properly serialized by the sdk
		// References to aws sdk issue https://github.com/aws/aws-sdk-go-v2/issues/2921

		// if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMultipleChecksumHeaders)); err != nil {
		// 	return err
		// }
		if err := checkSdkApiErr(err, "InvalidRequest"); err != nil {
			return err
		}

		return nil
	})
}

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

		i := 0
		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.PutObjectInput{
					Bucket: &bucket,
					Key:    &obj,
				}
				setPutObjectChecksum(input, algo, getPtr(checksum))

				_, err := putObjectWithData(int64((i+1)*100), input, s3client)
				i++

				// FIXME: The error message for PutObject is not properly serialized by the sdk
				// References to aws sdk issue https://github.com/aws/aws-sdk-go-v2/issues/2921

				// if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrMultipleChecksumHeaders)); err != nil {
				// 	return err
				// }
				if err := checkSdkApiErr(err, "InvalidRequest"); err != nil {
					return err
				}
			}
		}

		return nil
	})
}

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

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

			// test for file object
			input := &s3.PutObjectInput{
				Bucket: &bucket,
				Key:    &obj,
			}
			setPutObjectChecksum(input, algo, &wrongChecksum)
			_, err = putObjectWithData(int64((i+1)*100), input, s3client)
			if err := checkApiErr(err, s3err.GetChecksumBadDigestErr(algo)); err != nil {
				return err
			}

			// test for directory object
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			input = &s3.PutObjectInput{
				Bucket: &bucket,
				Key:    &dirObj,
			}
			setPutObjectChecksum(input, algo, &wrongChecksum)
			_, err = s3client.PutObject(ctx, input)
			cancel()
			if err := checkApiErr(err, s3err.GetChecksumBadDigestErr(algo)); err != nil {
				return err
			}
		}

		return nil
	})
}

func PutObject_default_checksum(s *S3Conf) error {
	testName := "PutObject_default_checksum"
	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 := putObjectWithData(100, &s3.PutObjectInput{
			Bucket: &bucket,
			Key:    &obj,
		}, customClient)
		if err != nil {
			return err
		}

		if out.res.ChecksumCRC64NVME == nil {
			return fmt.Errorf("expected non nil default crc64nvme checksum")
		}

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := customClient.HeadObject(ctx, &s3.HeadObjectInput{
			Bucket:       &bucket,
			Key:          &obj,
			ChecksumMode: types.ChecksumModeEnabled,
		})
		cancel()
		if err != nil {
			return err
		}

		if getString(res.ChecksumCRC64NVME) != getString(out.res.ChecksumCRC64NVME) {
			return fmt.Errorf("expected the object crc64nvme checksum to be %v, instead got %v", getString(res.ChecksumCRC64NVME), getString(out.res.ChecksumCRC64NVME))
		}

		return nil
	})
}

func PutObject_dir_object_default_checksum(s *S3Conf) error {
	testName := "PutObject_dir_object_default_checksum"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "dir/obj/"

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := s3client.PutObject(ctx, &s3.PutObjectInput{
			Bucket: &bucket,
			Key:    &obj,
		}, func(o *s3.Options) {
			o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
		})
		cancel()
		if err != nil {
			return err
		}

		if res.ChecksumType != types.ChecksumTypeFullObject {
			return fmt.Errorf("expected the object checksum type to be %s, instead got %s", types.ChecksumTypeFullObject, res.ChecksumType)
		}
		expectedcrc64nvme := "AAAAAAAAAAA="
		if getString(res.ChecksumCRC64NVME) != expectedcrc64nvme {
			return fmt.Errorf("expected the crc64nvme checksum to be %s, instead got %s", expectedcrc64nvme, getString(res.ChecksumCRC64NVME))
		}

		return nil
	})
}

func PutObject_dir_object_checksums_success(s *S3Conf) error {
	testName := "PutObject_dir_object_checksums_success"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		for i, test := range []struct {
			algo          types.ChecksumAlgorithm
			checksumValue string
		}{
			{types.ChecksumAlgorithmCrc32, "AAAAAA=="},
			{types.ChecksumAlgorithmCrc32c, "AAAAAA=="},
			{types.ChecksumAlgorithmCrc64nvme, "AAAAAAAAAAA="},
			{types.ChecksumAlgorithmSha1, "2jmj7l5rSw0yVb/vlWAYkK/YBwk="},
			{types.ChecksumAlgorithmSha256, "47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU="},
			{types.ChecksumAlgorithmSha512, "z4PhNX7vuL3xVChQ1m2AB9Yg5AULVxXcg/SpIdNs6c5H0NE8XYXysP+DGNKHfuwvY7kxvUdBeoGlODJ6+SfaPg=="},
			{types.ChecksumAlgorithmMd5, "1B2M2Y8AsgTpgAmY7PhCfg=="},
			{types.ChecksumAlgorithmXxhash64, "70bbN1HY6Zk="},
			{types.ChecksumAlgorithmXxhash3, "LQaABTjTlMI="},
			{types.ChecksumAlgorithmXxhash128, "maoG0wFHmNhgAcMkRo1Jfw=="},
		} {
			input := &s3.PutObjectInput{
				Bucket:            &bucket,
				Key:               getPtr(fmt.Sprintf("obj-%v/", i)),
				ChecksumAlgorithm: test.algo,
			}
			setPutObjectChecksum(input, test.algo, &test.checksumValue)
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			res, err := s3client.PutObject(ctx, input)
			cancel()
			if err != nil {
				return err
			}

			if res.ChecksumType != types.ChecksumTypeFullObject {
				return fmt.Errorf("expected the checksum type to be %s, instead got %s", types.ChecksumTypeFullObject, res.ChecksumType)
			}

			if got := getString(getPutObjectChecksum(res, test.algo)); got != test.checksumValue {
				return fmt.Errorf("expected the %s checksum value to be %s, instead got %s", test.algo, test.checksumValue, got)
			}
		}

		return nil
	})
}

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

		for i, algo := range types.ChecksumAlgorithmCrc32.Values() {
			res, err := putObjectWithData(int64(i*200), &s3.PutObjectInput{
				Bucket:            &bucket,
				Key:               &obj,
				ChecksumAlgorithm: algo,
			}, s3client, withPutObjectChecksumAlgo(algo))
			if err != nil {
				return err
			}

			if res.res.ChecksumType != types.ChecksumTypeFullObject {
				return fmt.Errorf("expected the object checksum type to be %v, instead got %v", types.ChecksumTypeFullObject, res.res.ChecksumType)
			}

			if getPutObjectChecksum(res.res, algo) == nil {
				return fmt.Errorf("expected non empty %s checksum in the response", algo)
			}
		}

		return nil
	})
}

func PutObject_racey_success(s *S3Conf) error {
	testName := "PutObject_racey_success"
	runF(testName)
	bucket, obj, lockStatus := getBucketName(), "my-obj", true

	client := s.GetClient()
	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
	_, err := client.CreateBucket(ctx, &s3.CreateBucketInput{
		Bucket:                     &bucket,
		ObjectLockEnabledForBucket: &lockStatus,
	})
	cancel()
	if err != nil {
		failF("%v: %v", testName, err)
		return fmt.Errorf("%v: %w", testName, err)
	}

	eg := errgroup.Group{}
	for range 10 {
		eg.Go(func() error {
			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
			_, err := client.PutObject(ctx, &s3.PutObjectInput{
				Bucket: &bucket,
				Key:    &obj,
			})
			cancel()
			return err
		})
	}
	err = eg.Wait()

	if err != nil {
		failF("%v: %v", testName, err)
		return fmt.Errorf("%v: %w", testName, err)
	}

	err = teardown(s, bucket)
	if err != nil {
		failF("%v: %v", testName, err)
		return fmt.Errorf("%v: %w", testName, err)
	}

	passF(testName)
	return nil
}

func PutObject_success(s *S3Conf) error {
	testName := "PutObject_success"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		lgth := int64(100)
		res, err := putObjectWithData(lgth, &s3.PutObjectInput{
			Bucket: &bucket,
			Key:    getPtr("my-obj"),
		}, s3client)
		if err != nil {
			return err
		}

		// skip the ETag check for azure tests
		if !s.azureTests {
			etag, err := calculateEtag(res.data)
			if err != nil {
				return err
			}

			if getString(res.res.ETag) != etag {
				return fmt.Errorf("expected ETag to be %s, intead got %s", getString(res.res.ETag), etag)
			}
		}
		if res.res.Size == nil {
			return fmt.Errorf("unexpected nil object Size")
		}
		if *res.res.Size != lgth {
			return fmt.Errorf("expected the object size to be %v, instead got %v", lgth, *res.res.Size)
		}

		return nil
	})
}

func PutObject_default_content_type(s *S3Conf) error {
	testName := "PutObject_default_content_type"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		key := "my-object"
		_, err := putObjectWithData(10, &s3.PutObjectInput{
			Bucket:      &bucket,
			Key:         &key,
			ContentType: nil,
		}, s3client)
		if err != nil {
			return err
		}

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
			Bucket: &bucket,
			Key:    &key,
		})
		cancel()
		if err != nil {
			return err
		}

		if getString(res.ContentType) != defaultContentType {
			return fmt.Errorf("expected default %s Content-Type, instead got %s", defaultContentType, getString(res.ContentType))
		}

		return nil
	})
}

func PutObject_invalid_credentials(s *S3Conf) error {
	testName := "PutObject_invalid_credentials"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		newconf := *s
		newconf.awsSecret = newconf.awsSecret + "badpassword"
		client := newconf.GetClient()
		_, err := putObjects(client, []string{"my-obj"}, bucket)
		return checkApiErr(err, s3err.GetAPIError(s3err.ErrSignatureDoesNotMatch))
	})
}

func PutObject_invalid_object_names(s *S3Conf) error {
	testName := "PutObject_invalid_object_names"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		for _, obj := range []string{
			".",
			"..",
			"./",
			"/.",
			"//",
			"../",
			"/..",
			"/..",
			"../.",
			"../../../.",
			"../../../etc/passwd",
			"../../../../tmp/foo",
			"for/../../bar/",
			"a/a/a/../../../../../etc/passwd",
			"/a/../../b/../../c/../../../etc/passwd",
		} {
			_, err := putObjects(s3client, []string{obj}, bucket)
			if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrBadRequest)); err != nil {
				return err
			}
		}

		return nil
	})
}

func PutObject_object_acl_not_supported(s *S3Conf) error {
	testName := "PutObject_object_acl_not_supported"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		obj := "my-object"
		testuser := getUser("user")
		err := createUsers(s, []user{testuser})
		if err != nil {
			return err
		}

		for i, modifyInput := range []func(*s3.PutObjectInput){
			func(poi *s3.PutObjectInput) { poi.ACL = types.ObjectCannedACLPublicRead },
			func(poi *s3.PutObjectInput) { poi.GrantFullControl = &testuser.access },
			func(poi *s3.PutObjectInput) { poi.GrantRead = &testuser.access },
			func(poi *s3.PutObjectInput) { poi.GrantReadACP = &testuser.access },
			func(poi *s3.PutObjectInput) { poi.GrantWriteACP = &testuser.access },
		} {
			input := &s3.PutObjectInput{
				Bucket: &bucket,
				Key:    &obj,
			}

			modifyInput(input)
			_, err := putObjectWithData(0, input, s3client)
			if err != nil {
				return fmt.Errorf("test %v failed: %w", i+1, err)
			}
		}

		return nil
	})
}

func PutObject_false_negative_object_names(s *S3Conf) error {
	testName := "PutObject_false_negative_object_names"
	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
		objs := []string{
			"%252e%252e%252fetc/passwd",            // double encoding
			"%2e%2e/%2e%2e/%2e%2e/.ssh/id_rsa",     // double URL-encoded
			"%u002e%u002e/%u002e%u002e/etc/passwd", // unicode escape
			"..%2f..%2f..%2fsecret/file.txt",       // URL-encoded
			"..%c0%af..%c0%afetc/passwd",           // UTF-8 overlong trick
			".../.../.../target.txt",
			"..\\u2215..\\u2215etc/passwd",             // Unicode division slash
			"dir/%20../file.txt",                       // encoded space
			"dir/%c0%ae%c0%ae/%c0%ae%c0%ae/etc/passwd", // overlong UTF-8 encoding
			"logs/latest -> /etc/passwd",               // symlink attacks
			//TODO: add this test case in advanced routing
			// "/etc/passwd" // absolute path injection
		}
		_, err := putObjects(s3client, objs, bucket)
		if err != nil {
			return err
		}

		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
		res, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
			Bucket: &bucket,
		})
		cancel()
		if err != nil {
			return err
		}

		if len(res.Contents) != len(objs) {
			return fmt.Errorf("expected %v objects, instead got %v", len(objs), len(res.Contents))
		}

		for i, obj := range res.Contents {
			if *obj.Key != objs[i] {
				return fmt.Errorf("expected the %vth object name to be %s, instead got %s", i+1, objs[i], *obj.Key)
			}
		}

		return nil
	})
}