package handler

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"mime/multipart"
	"net/http"
	"net/http/httptest"
	"strconv"
	"strings"
	"testing"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/auth"
	v4 "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/auth/signer/v4"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/data"
	apierr "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/errors"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/layer"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/middleware"
	"git.frostfs.info/TrueCloudLab/frostfs-s3-gw/creds/accessbox"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/stretchr/testify/require"
)

const (
	awsChunkedRequestExampleDecodedContentLength = 66560
	awsChunkedRequestExampleContentLength        = 66824
)

func TestCheckBucketName(t *testing.T) {
	for _, tc := range []struct {
		name string
		err  bool
	}{
		{name: "bucket"},
		{name: "2bucket"},
		{name: "buc.ket"},
		{name: "buc-ket"},
		{name: "abc"},
		{name: "63aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
		{name: "buc.-ket", err: true},
		{name: "bucket.", err: true},
		{name: ".bucket", err: true},
		{name: "bucket.", err: true},
		{name: "bucket-", err: true},
		{name: "-bucket", err: true},
		{name: "Bucket", err: true},
		{name: "buc.-ket", err: true},
		{name: "buc-.ket", err: true},
		{name: "Bucket", err: true},
		{name: "buc!ket", err: true},
		{name: "buc_ket", err: true},
		{name: "xn--bucket", err: true},
		{name: "bucket-s3alias", err: true},
		{name: "192.168.0.1", err: true},
		{name: "as", err: true},
		{name: "64aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", err: true},
	} {
		err := checkBucketName(tc.name)
		if tc.err {
			require.Error(t, err, "bucket name: %s", tc.name)
		} else {
			require.NoError(t, err, "bucket name: %s", tc.name)
		}
	}
}

func TestCustomJSONMarshal(t *testing.T) {
	data := []byte(`
{ "expiration": "2015-12-30T12:00:00.000Z",
  "conditions": [
	["content-length-range", 1048576, 10485760],
    {"bucket": "bucketName"},
    ["starts-with", "$key", "user/user1/"]
  ]
}`)

	parsedTime, err := time.Parse(time.RFC3339, "2015-12-30T12:00:00.000Z")
	require.NoError(t, err)

	expectedPolicy := &postPolicy{
		Expiration: parsedTime,
		Conditions: []*policyCondition{
			{
				Matching: "content-length-range",
				Key:      "1048576",
				Value:    "10485760",
			},
			{
				Matching: "eq",
				Key:      "bucket",
				Value:    "bucketName",
			},
			{
				Matching: "starts-with",
				Key:      "key",
				Value:    "user/user1/",
			},
		},
	}

	policy := &postPolicy{}
	err = json.Unmarshal(data, policy)
	require.NoError(t, err)

	require.Equal(t, expectedPolicy, policy)
}

func TestEmptyPostPolicy(t *testing.T) {
	r := &http.Request{
		MultipartForm: &multipart.Form{
			Value: map[string][]string{
				"key": {"some-key"},
			},
		},
	}
	reqInfo := &middleware.ReqInfo{}
	metadata := make(map[string]string)

	_, err := checkPostPolicy(r, reqInfo, metadata)
	require.NoError(t, err)
}

// if content length is greater than this value
// data will be writen to file location.
const maxContentSizeForFormData = 10

func TestPostObject(t *testing.T) {
	hc := prepareHandlerContext(t)

	ns, bktName := "", "bucket"
	createTestBucket(hc, bktName)

	for _, tc := range []struct {
		key      string
		filename string
		content  string
		objName  string
		err      bool
	}{
		{
			key:      "user/user1/${filename}",
			filename: "object",
			content:  "content",
			objName:  "user/user1/object",
		},
		{
			key:      "user/user1/${filename}",
			filename: "object",
			content:  "maxContentSizeForFormData",
			objName:  "user/user1/object",
		},
		{
			key:      "user/user1/key-object",
			filename: "object",
			content:  "",
			objName:  "user/user1/key-object",
		},
		{
			key:      "user/user1/key-object",
			filename: "object",
			content:  "maxContentSizeForFormData",
			objName:  "user/user1/key-object",
		},
		{
			key:      "",
			filename: "object",
			content:  "",
			objName:  "object",
		},
		{
			key:      "",
			filename: "object",
			content:  "maxContentSizeForFormData",
			objName:  "object",
		},
		{
			// RFC 7578, Section 4.2 requires that if a filename is provided, the
			// directory path information must not be used.
			key:      "",
			filename: "dir/object",
			content:  "content",
			objName:  "object",
		},
		{
			key:      "object",
			filename: "",
			content:  "content",
			objName:  "object",
		},
		{
			key:      "",
			filename: "",
			err:      true,
		},
	} {
		t.Run(tc.key+";"+tc.filename, func(t *testing.T) {
			w := postObjectBase(hc, ns, bktName, tc.key, tc.filename, tc.content)
			if tc.err {
				assertS3Error(hc.t, w, apierr.GetAPIError(apierr.ErrInternalError))
				return
			}
			assertStatus(hc.t, w, http.StatusNoContent)
			content, _ := getObject(hc, bktName, tc.objName)
			require.Equal(t, tc.content, string(content))
		})
	}
}

func TestPutObjectOverrideCopiesNumber(t *testing.T) {
	tc := prepareHandlerContext(t)

	bktName, objName := "bucket-for-copies-number", "object-for-copies-number"
	bktInfo := createTestBucket(tc, bktName)

	w, r := prepareTestRequest(tc, bktName, objName, nil)
	r.Header.Set(api.MetadataPrefix+strings.ToUpper(layer.AttributeFrostfsCopiesNumber), "1")
	tc.Handler().PutObjectHandler(w, r)

	p := &layer.HeadObjectParams{
		BktInfo: bktInfo,
		Object:  objName,
	}

	objInfo, err := tc.Layer().GetObjectInfo(tc.Context(), p)
	require.NoError(t, err)
	require.Equal(t, "1", objInfo.Headers[layer.AttributeFrostfsCopiesNumber])
}

func TestPutObjectWithNegativeContentLength(t *testing.T) {
	tc := prepareHandlerContext(t)

	bktName, objName := "bucket-for-put", "object-for-put"
	createTestBucket(tc, bktName)

	content := []byte("content")
	w, r := prepareTestPayloadRequest(tc, bktName, objName, bytes.NewReader(content))
	r.ContentLength = -1
	tc.Handler().PutObjectHandler(w, r)
	assertStatus(t, w, http.StatusOK)

	w, r = prepareTestRequest(tc, bktName, objName, nil)
	tc.Handler().HeadObjectHandler(w, r)
	assertStatus(t, w, http.StatusOK)
	require.Equal(t, strconv.Itoa(len(content)), w.Header().Get(api.ContentLength))

	result := listVersions(t, tc, bktName)
	require.Len(t, result.Version, 1)
	require.EqualValues(t, len(content), result.Version[0].Size)
}

func TestPutObjectWithStreamBodyError(t *testing.T) {
	tc := prepareHandlerContext(t)

	bktName, objName := "bucket-for-put", "object-for-put"
	createTestBucket(tc, bktName)

	content := []byte("content")
	w, r := prepareTestPayloadRequest(tc, bktName, objName, bytes.NewReader(content))
	r.Header.Set(api.AmzContentSha256, api.StreamingContentSHA256)
	r.Header.Set(api.ContentEncoding, api.AwsChunked)
	tc.Handler().PutObjectHandler(w, r)
	assertS3Error(t, w, apierr.GetAPIError(apierr.ErrMissingContentLength))

	checkNotFound(t, tc, bktName, objName, emptyVersion)
}

func TestPutObjectWithInvalidContentMD5(t *testing.T) {
	tc := prepareHandlerContext(t)
	tc.config.md5Enabled = true

	bktName, objName := "bucket-for-put", "object-for-put"
	createTestBucket(tc, bktName)

	content := []byte("content")
	w, r := prepareTestPayloadRequest(tc, bktName, objName, bytes.NewReader(content))
	r.Header.Set(api.ContentMD5, base64.StdEncoding.EncodeToString([]byte("invalid")))
	tc.Handler().PutObjectHandler(w, r)
	assertS3Error(t, w, apierr.GetAPIError(apierr.ErrInvalidDigest))

	checkNotFound(t, tc, bktName, objName, emptyVersion)
}

func TestPutObjectWithEnabledMD5(t *testing.T) {
	tc := prepareHandlerContext(t)
	tc.config.md5Enabled = true

	bktName, objName := "bucket-for-put", "object-for-put"
	createTestBucket(tc, bktName)

	content := []byte("content")
	md5Hash := md5.New()
	md5Hash.Write(content)
	w, r := prepareTestPayloadRequest(tc, bktName, objName, bytes.NewReader(content))
	tc.Handler().PutObjectHandler(w, r)
	require.Equal(t, data.Quote(hex.EncodeToString(md5Hash.Sum(nil))), w.Header().Get(api.ETag))
}

func TestPutObjectCheckContentSHA256(t *testing.T) {
	hc := prepareHandlerContext(t)

	bktName, objName := "bucket-for-put", "object-for-put"
	createTestBucket(hc, bktName)

	for _, tc := range []struct {
		name    string
		hash    string
		content []byte
		error   bool
	}{
		{
			name:    "invalid hash value",
			hash:    "d1b2a59fbea7e20077af9f91b27e95e865061b270be03ff539ab3b73587882e8",
			content: []byte("content"),
			error:   true,
		},
		{
			name:    "correct hash for empty payload",
			hash:    "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
			content: []byte(""),
			error:   false,
		},
		{
			name:    "unsigned payload",
			hash:    "UNSIGNED-PAYLOAD",
			content: []byte("content"),
			error:   false,
		},
		{
			name:    "correct hash",
			hash:    "ed7002b439e9ac845f22357d822bac1444730fbdb6016d3ec9432297b9ec9f73",
			content: []byte("content"),
			error:   false,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			w, r := prepareTestPayloadRequest(hc, bktName, objName, bytes.NewReader(tc.content))
			r.Header.Set("X-Amz-Content-Sha256", tc.hash)
			hc.Handler().PutObjectHandler(w, r)

			if tc.error {
				assertS3Error(t, w, apierr.GetAPIError(apierr.ErrContentSHA256Mismatch))

				w, r := prepareTestRequest(hc, bktName, objName, nil)
				hc.Handler().GetObjectHandler(w, r)

				assertStatus(t, w, http.StatusNotFound)
				return
			}

			assertStatus(t, w, http.StatusOK)
		})
	}
}

func TestPutObjectWithStreamBodyAWSExample(t *testing.T) {
	hc := prepareHandlerContext(t)

	bktName, objName := "examplebucket", "chunkObject.txt"
	createTestBucket(hc, bktName)

	w, req, chunk := getChunkedRequest(hc.context, t, bktName, objName)
	hc.Handler().PutObjectHandler(w, req)
	assertStatus(t, w, http.StatusOK)

	w, req = prepareTestRequest(hc, bktName, objName, nil)
	hc.Handler().HeadObjectHandler(w, req)
	assertStatus(t, w, http.StatusOK)
	require.Equal(t, strconv.Itoa(awsChunkedRequestExampleDecodedContentLength), w.Header().Get(api.ContentLength))

	data := getObjectRange(t, hc, bktName, objName, 0, awsChunkedRequestExampleDecodedContentLength)
	for i := range chunk {
		require.Equal(t, chunk[i], data[i])
	}
}

func TestPutChunkedTestContentEncoding(t *testing.T) {
	hc := prepareHandlerContext(t)

	bktName, objName := "examplebucket", "chunkObject.txt"
	createTestBucket(hc, bktName)

	w, req, _ := getChunkedRequest(hc.context, t, bktName, objName)
	req.Header.Set(api.ContentEncoding, api.AwsChunked+",gzip")

	hc.Handler().PutObjectHandler(w, req)
	assertStatus(t, w, http.StatusOK)

	resp := headObjectBase(hc, bktName, objName, emptyVersion)
	require.Equal(t, "gzip", resp.Header().Get(api.ContentEncoding))

	w, req, _ = getChunkedRequest(hc.context, t, bktName, objName)
	req.Header.Set(api.ContentEncoding, "gzip")
	hc.Handler().PutObjectHandler(w, req)
	assertS3Error(t, w, apierr.GetAPIError(apierr.ErrInvalidEncodingMethod))

	hc.config.bypassContentEncodingInChunks = true
	w, req, _ = getChunkedRequest(hc.context, t, bktName, objName)
	req.Header.Set(api.ContentEncoding, "gzip")
	hc.Handler().PutObjectHandler(w, req)
	assertStatus(t, w, http.StatusOK)

	resp = headObjectBase(hc, bktName, objName, emptyVersion)
	require.Equal(t, "gzip", resp.Header().Get(api.ContentEncoding))
}

// getChunkedRequest implements request example from
// https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming.html
func getChunkedRequest(ctx context.Context, t *testing.T, bktName, objName string) (*httptest.ResponseRecorder, *http.Request, []byte) {
	chunk := make([]byte, 65*1024)
	for i := range chunk {
		chunk[i] = 'a'
	}
	chunk1 := chunk[:64*1024]
	chunk2 := chunk[64*1024:]

	AWSAccessKeyID := "AKIAIOSFODNN7EXAMPLE"
	AWSSecretAccessKey := "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

	awsCreds := credentials.NewStaticCredentials(AWSAccessKeyID, AWSSecretAccessKey, "")
	signer := v4.NewSigner(awsCreds)

	reqBody := bytes.NewBufferString("10000;chunk-signature=ad80c730a21e5b8d04586a2213dd63b9a0e99e0e2307b0ade35a65485a288648\r\n")
	_, err := reqBody.Write(chunk1)
	require.NoError(t, err)
	_, err = reqBody.WriteString("\r\n400;chunk-signature=0055627c9e194cb4542bae2aa5492e3c1575bbb81b612b7d234b86a503ef5497\r\n")
	require.NoError(t, err)
	_, err = reqBody.Write(chunk2)
	require.NoError(t, err)
	_, err = reqBody.WriteString("\r\n0;chunk-signature=b6c6ea8a5354eaf15b3cb7646744f4275b71ea724fed81ceb9323e279d449df9\r\n\r\n")
	require.NoError(t, err)

	req, err := http.NewRequest("PUT", "https://s3.amazonaws.com/"+bktName+"/"+objName, nil)
	require.NoError(t, err)
	req.Header.Set("content-encoding", "aws-chunked")
	req.Header.Set("content-length", strconv.Itoa(awsChunkedRequestExampleContentLength))
	req.Header.Set("x-amz-content-sha256", "STREAMING-AWS4-HMAC-SHA256-PAYLOAD")
	req.Header.Set("x-amz-decoded-content-length", strconv.Itoa(awsChunkedRequestExampleDecodedContentLength))
	req.Header.Set("x-amz-storage-class", "REDUCED_REDUNDANCY")

	signTime, err := time.Parse("20060102T150405Z", "20130524T000000Z")
	require.NoError(t, err)

	_, err = signer.Sign(req, nil, "s3", "us-east-1", signTime)
	require.NoError(t, err)

	req.Body = io.NopCloser(reqBody)

	w := httptest.NewRecorder()
	reqInfo := middleware.NewReqInfo(w, req, middleware.ObjectRequest{Bucket: bktName, Object: objName}, "")
	req = req.WithContext(middleware.SetReqInfo(ctx, reqInfo))
	req = req.WithContext(middleware.SetBox(req.Context(), &middleware.Box{
		ClientTime: signTime,
		AuthHeaders: &middleware.AuthHeader{
			AccessKeyID: AWSAccessKeyID,
			SignatureV4: "4f232c4386841ef735655705268965c44a0e4690baa4adea153f7db9fa80a0a9",
			Region:      "us-east-1",
		},
		AccessBox: &accessbox.Box{
			Gate: &accessbox.GateData{
				SecretKey: AWSSecretAccessKey,
			},
		},
	}))

	return w, req, chunk
}

func TestCreateBucket(t *testing.T) {
	hc := prepareHandlerContext(t)
	bktName := "bkt-name"

	info := createBucket(hc, bktName)
	createBucketAssertS3Error(hc, bktName, info.Box, apierr.ErrBucketAlreadyOwnedByYou)

	box2, _ := createAccessBox(t)
	createBucketAssertS3Error(hc, bktName, box2, apierr.ErrBucketAlreadyExists)
}

func TestCreateBucketWithoutPermissions(t *testing.T) {
	hc := prepareHandlerContext(t)
	bktName := "bkt-name"

	hc.h.ape.(*apeMock).err = errors.New("no permissions")

	box, _ := createAccessBox(t)
	createBucketAssertS3Error(hc, bktName, box, s3errors.ErrInternalError)

	_, err := hc.tp.ContainerID(bktName)
	require.Errorf(t, err, "container exists after failed creation, but shouldn't")
}

func TestCreateNamespacedBucket(t *testing.T) {
	hc := prepareHandlerContext(t)
	bktName := "bkt-name"
	namespace := "yabloko"

	box, _ := createAccessBox(t)
	w, r := prepareTestRequest(hc, bktName, "", nil)
	ctx := middleware.SetBox(r.Context(), &middleware.Box{AccessBox: box})
	reqInfo := middleware.GetReqInfo(ctx)
	reqInfo.Namespace = namespace
	r = r.WithContext(middleware.SetReqInfo(ctx, reqInfo))
	hc.Handler().CreateBucketHandler(w, r)
	assertStatus(t, w, http.StatusOK)

	bktInfo, err := hc.Layer().GetBucketInfo(middleware.SetReqInfo(hc.Context(), reqInfo), bktName)
	require.NoError(t, err)

	require.Equal(t, namespace+".ns", bktInfo.Zone)
}

func TestPutObjectClientCut(t *testing.T) {
	hc := prepareHandlerContext(t)
	bktName, objName1, objName2 := "bkt-name", "obj-name1", "obj-name2"
	createTestBucket(hc, bktName)

	putObject(hc, bktName, objName1)
	obj1 := getObjectFromLayer(hc, objName1)[0]
	require.Empty(t, getObjectAttribute(obj1, "s3-client-cut"))

	hc.layerFeatures.SetClientCut(true)
	putObject(hc, bktName, objName2)
	obj2 := getObjectFromLayer(hc, objName2)[0]
	require.Equal(t, "true", getObjectAttribute(obj2, "s3-client-cut"))
}

func getObjectFromLayer(hc *handlerContext, objName string) []*object.Object {
	var res []*object.Object
	for _, o := range hc.tp.Objects() {
		if objName == getObjectAttribute(o, object.AttributeFilePath) {
			res = append(res, o)
		}
	}
	return res
}

func getObjectAttribute(obj *object.Object, attrName string) string {
	for _, attr := range obj.Attributes() {
		if attr.Key() == attrName {
			return attr.Value()
		}
	}
	return ""
}

func TestPutObjectWithContentLanguage(t *testing.T) {
	tc := prepareHandlerContext(t)

	expectedContentLanguage := "en"
	bktName, objName := "bucket-1", "object-1"
	createTestBucket(tc, bktName)

	w, r := prepareTestRequest(tc, bktName, objName, nil)
	r.Header.Set(api.ContentLanguage, expectedContentLanguage)
	tc.Handler().PutObjectHandler(w, r)

	tc.Handler().HeadObjectHandler(w, r)
	require.Equal(t, expectedContentLanguage, w.Header().Get(api.ContentLanguage))
}

func postObjectBase(hc *handlerContext, ns, bktName, key, filename, content string) *httptest.ResponseRecorder {
	policy := "eyJleHBpcmF0aW9uIjogIjIwMjUtMTItMDFUMTI6MDA6MDAuMDAwWiIsImNvbmRpdGlvbnMiOiBbCiBbInN0YXJ0cy13aXRoIiwgIiR4LWFtei1jcmVkZW50aWFsIiwgIiJdLAogWyJzdGFydHMtd2l0aCIsICIkeC1hbXotZGF0ZSIsICIiXSwKIFsic3RhcnRzLXdpdGgiLCAiJGtleSIsICIiXQpdfQ=="

	timeToSign := time.Now()
	timeToSignStr := timeToSign.Format("20060102T150405Z")
	region := "default"
	service := "s3"

	accessKeyID := "5jizSbYu8hX345aqCKDgRWKCJYHxnzxRS8e6SUYHZ8Fw0HiRkf3KbJAWBn5mRzmiyHQ3UHADGyzVXLusn1BrmAfLn"
	secretKey := "abf066d77c6744cd956a123a0b9612df587f5c14d3350ecb01b363f182dd7279"

	creds := getCredsStr(accessKeyID, timeToSignStr, region, service)
	sign := auth.SignStr(secretKey, service, region, timeToSign, policy)

	body, contentType, err := getMultipartFormBody(policy, creds, timeToSignStr, sign, key, filename, content)
	require.NoError(hc.t, err)

	w, r := prepareTestPostRequest(hc, bktName, body)
	r.Header.Set(auth.ContentTypeHdr, contentType)
	r.Header.Set("X-Frostfs-Namespace", ns)

	err = r.ParseMultipartForm(50 * 1024 * 1024)
	require.NoError(hc.t, err)

	hc.Handler().PostObject(w, r)
	return w
}

func getCredsStr(accessKeyID, timeToSign, region, service string) string {
	return accessKeyID + "/" + timeToSign + "/" + region + "/" + service + "/aws4_request"
}

func getMultipartFormBody(policy, creds, date, sign, key, filename, content string) (io.Reader, string, error) {
	body := &bytes.Buffer{}
	writer := multipart.NewWriter(body)
	defer writer.Close()

	if err := writer.WriteField("policy", policy); err != nil {
		return nil, "", err
	}

	if err := writer.WriteField("key", key); err != nil {
		return nil, "", err
	}
	if err := writer.WriteField(strings.ToLower(auth.AmzCredential), creds); err != nil {
		return nil, "", err
	}
	if err := writer.WriteField(strings.ToLower(auth.AmzDate), date); err != nil {
		return nil, "", err
	}
	if err := writer.WriteField(strings.ToLower(auth.AmzSignature), sign); err != nil {
		return nil, "", err
	}

	file, err := writer.CreateFormFile("file", filename)
	if err != nil {
		return nil, "", err
	}

	if len(content) < maxContentSizeForFormData {
		if err = writer.WriteField("file", content); err != nil {
			return nil, "", err
		}
	} else {
		if _, err = file.Write([]byte(content)); err != nil {
			return nil, "", err
		}
	}

	return body, writer.FormDataContentType(), nil
}

func prepareTestPostRequest(hc *handlerContext, bktName string, payload io.Reader) (*httptest.ResponseRecorder, *http.Request) {
	w := httptest.NewRecorder()
	r := httptest.NewRequest(http.MethodPost, defaultURL+bktName, payload)

	reqInfo := middleware.NewReqInfo(w, r, middleware.ObjectRequest{Bucket: bktName}, "")
	r = r.WithContext(middleware.SetReqInfo(hc.Context(), reqInfo))

	return w, r
}