// Copyright (c) 2015-2026 MinIO, Inc.
//
// This file is part of MinIO Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package cmd

import (
	"bytes"
	"encoding/xml"
	"fmt"
	"net/http"
	"net/http/httptest"
	"strconv"
	"testing"

	"github.com/dustin/go-humanize"
	"github.com/minio/minio/internal/auth"
	"github.com/minio/minio/internal/hash"
	xhttp "github.com/minio/minio/internal/http"
)

// multipartChecksumTestData is the payload used by the full object checksum
// tests: one 5 MiB part (the minimum allowed non-final part size) and a small
// trailing part, so part merging is actually exercised.
func multipartChecksumTestData() (parts [][]byte, full []byte) {
	parts = [][]byte{
		bytes.Repeat([]byte("a"), 5*humanize.MiByte),
		bytes.Repeat([]byte("b"), 1*humanize.KiByte),
	}
	for _, p := range parts {
		full = append(full, p...)
	}
	return parts, full
}

func mustChecksum(t *testing.T, typ hash.ChecksumType, data []byte) string {
	t.Helper()
	cs := hash.NewChecksumFromData(typ, data)
	if cs == nil {
		t.Fatalf("unable to compute %s checksum", typ.String())
	}
	return cs.Encoded
}

// newMultipartUploadHTTP starts a multipart upload over the API router and
// returns the upload ID.
func newMultipartUploadHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
	bucket, object, algo, checksumType string,
) string {
	t.Helper()
	hdrs := map[string]string{xhttp.AmzChecksumAlgo: algo}
	if checksumType != "" {
		hdrs[xhttp.AmzChecksumType] = checksumType
	}
	req, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucket, object),
		0, nil, creds.AccessKey, creds.SecretKey, hdrs)
	if err != nil {
		t.Fatalf("failed to build NewMultipartUpload request: %v", err)
	}
	rec := httptest.NewRecorder()
	apiRouter.ServeHTTP(rec, req)
	if rec.Code != http.StatusOK {
		t.Fatalf("NewMultipartUpload failed: %d %s", rec.Code, rec.Body.String())
	}
	var res InitiateMultipartUploadResponse
	if err := xml.Unmarshal(rec.Body.Bytes(), &res); err != nil {
		t.Fatalf("failed to decode NewMultipartUpload response: %v", err)
	}
	return res.UploadID
}

// uploadPartsHTTP uploads every part carrying its own checksum header, the way
// modern AWS SDKs do by default, and returns the part ETags.
func uploadPartsHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
	bucket, object, uploadID string, typ hash.ChecksumType, parts [][]byte,
) []string {
	t.Helper()
	etags := make([]string, len(parts))
	for i, p := range parts {
		req, err := newTestSignedRequestV4(http.MethodPut,
			getPutObjectPartURL("", bucket, object, uploadID, strconv.Itoa(i+1)),
			int64(len(p)), bytes.NewReader(p), creds.AccessKey, creds.SecretKey,
			map[string]string{typ.Key(): mustChecksum(t, typ, p)})
		if err != nil {
			t.Fatalf("failed to build UploadPart %d request: %v", i+1, err)
		}
		rec := httptest.NewRecorder()
		apiRouter.ServeHTTP(rec, req)
		if rec.Code != http.StatusOK {
			t.Fatalf("UploadPart %d failed: %d %s", i+1, rec.Code, rec.Body.String())
		}
		// MinIO writes the header under a literal "ETag" map key, which
		// http.Header.Get would canonicalize to "Etag" and miss.
		etags[i] = rec.Header()[xhttp.ETag][0]
	}
	return etags
}

// completeMultipartUploadHTTP completes the upload. partCS supplies an optional
// per-part checksum for each part; an empty string omits it, which is exactly
// what boto3, aws-sdk-js and the Java SDK send when the caller does not track
// part checksums.
func completeMultipartUploadHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
	bucket, object, uploadID string, etags []string, partCS []string, hdrs map[string]string,
) *httptest.ResponseRecorder {
	t.Helper()
	var body bytes.Buffer
	body.WriteString("<CompleteMultipartUpload>")
	for i, etag := range etags {
		fmt.Fprintf(&body, "<Part><PartNumber>%d</PartNumber><ETag>%s</ETag>", i+1, etag)
		if i < len(partCS) && partCS[i] != "" {
			fmt.Fprintf(&body, "<ChecksumCRC32>%s</ChecksumCRC32>", partCS[i])
		}
		body.WriteString("</Part>")
	}
	body.WriteString("</CompleteMultipartUpload>")

	req, err := newTestSignedRequestV4(http.MethodPost,
		getCompleteMultipartUploadURL("", bucket, object, uploadID),
		int64(body.Len()), bytes.NewReader(body.Bytes()), creds.AccessKey, creds.SecretKey, hdrs)
	if err != nil {
		t.Fatalf("failed to build CompleteMultipartUpload request: %v", err)
	}
	rec := httptest.NewRecorder()
	apiRouter.ServeHTTP(rec, req)
	return rec
}

func apiErrorCode(t *testing.T, rec *httptest.ResponseRecorder) string {
	t.Helper()
	var e APIErrorResponse
	if err := xml.Unmarshal(rec.Body.Bytes(), &e); err != nil {
		t.Fatalf("unable to decode error response %q: %v", rec.Body.String(), err)
	}
	return e.Code
}

// TestAPICompleteMultipartFullObjectChecksum covers pgsty/silo#31.
//
// A multipart upload created with a full object checksum type must be
// completable by sending only PartNumber and ETag per part, plus the object
// level checksum in the request headers. That is what AWS S3 accepts, and it is
// the point of FULL_OBJECT: the client no longer has to retain per-part
// checksums, only the part numbers and ETags it already tracks.
func TestAPICompleteMultipartFullObjectChecksum(t *testing.T) {
	defer DetectTestLeak(t)()
	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
		t:          t,
		objAPITest: testAPICompleteMultipartFullObjectChecksum,
		endpoints:  []string{"NewMultipart", "PutObjectPart", "CompleteMultipart"},
	})
}

func testAPICompleteMultipartFullObjectChecksum(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
	credentials auth.Credentials, t *testing.T,
) {
	partData, full := multipartChecksumTestData()

	// Only CRC based algorithms can linearize into a full object checksum.
	// Which one a client picks by default is SDK specific - the AWS CLI v2
	// defaults to CRC64NVME while the Go and JavaScript SDKs default to CRC32 -
	// so cover all three.
	for _, typ := range []hash.ChecksumType{hash.ChecksumCRC32, hash.ChecksumCRC32C, hash.ChecksumCRC64NVME} {
		objectName := "uploads/full-object-" + typ.String()

		uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, objectName,
			typ.String(), xhttp.AmzChecksumTypeFullObject)
		etags := uploadPartsHTTP(t, apiRouter, credentials, bucketName, objectName, uploadID, typ, partData)

		rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, objectName, uploadID, etags, nil,
			map[string]string{
				typ.Key():             mustChecksum(t, typ, full),
				xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
			})
		if rec.Code != http.StatusOK {
			t.Fatalf("%s/%s: CompleteMultipartUpload failed: %d %s",
				instanceType, typ.String(), rec.Code, rec.Body.String())
		}

		oi, err := obj.GetObjectInfo(t.Context(), bucketName, objectName, ObjectOptions{})
		if err != nil {
			t.Fatalf("%s/%s: GetObjectInfo failed: %v", instanceType, typ.String(), err)
		}
		if oi.Size != int64(len(full)) {
			t.Fatalf("%s/%s: expected object size %d, got %d", instanceType, typ.String(), len(full), oi.Size)
		}

		// The persisted checksum must be the merged full object value - not a
		// composite "<checksum>-<parts>" value - and must report FULL_OBJECT.
		cs, _ := oi.decryptChecksums(0, nil)
		if got, want := cs[typ.String()], mustChecksum(t, typ, full); got != want {
			t.Fatalf("%s/%s: expected stored checksum %q, got %q", instanceType, typ.String(), want, got)
		}
		if got := cs[xhttp.AmzChecksumType]; got != xhttp.AmzChecksumTypeFullObject {
			t.Fatalf("%s/%s: expected stored checksum type %q, got %q",
				instanceType, typ.String(), xhttp.AmzChecksumTypeFullObject, got)
		}
	}
}

// TestAPICompleteMultipartFullObjectChecksumMismatch asserts that accepting
// completions without part checksums does not weaken integrity: a wrong object
// level checksum is still rejected.
func TestAPICompleteMultipartFullObjectChecksumMismatch(t *testing.T) {
	defer DetectTestLeak(t)()
	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
		t:          t,
		objAPITest: testAPICompleteMultipartFullObjectChecksumMismatch,
		endpoints:  []string{"NewMultipart", "PutObjectPart", "CompleteMultipart"},
	})
}

func testAPICompleteMultipartFullObjectChecksumMismatch(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
	credentials auth.Credentials, t *testing.T,
) {
	partData, _ := multipartChecksumTestData()
	objectName := "uploads/full-object-mismatch"
	typ := hash.ChecksumCRC32

	uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, objectName,
		typ.String(), xhttp.AmzChecksumTypeFullObject)
	etags := uploadPartsHTTP(t, apiRouter, credentials, bucketName, objectName, uploadID, typ, partData)

	rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, objectName, uploadID, etags, nil,
		map[string]string{
			typ.Key():             mustChecksum(t, typ, []byte("not the object content")),
			xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
		})
	if rec.Code != http.StatusBadRequest {
		t.Fatalf("%s: CompleteMultipartUpload with a bad full object checksum returned %d, want 400",
			instanceType, rec.Code)
	}
	// NOTE: AWS S3 documents BadDigest for a full object checksum mismatch on
	// CompleteMultipartUpload. MinIO reports XAmzContentChecksumMismatch. That
	// deviation is tracked separately; assert the current code so a future
	// change to it is a deliberate one.
	if got := apiErrorCode(t, rec); got != "XAmzContentChecksumMismatch" {
		t.Fatalf("%s: expected XAmzContentChecksumMismatch, got %q", instanceType, got)
	}

	if _, err := obj.GetObjectInfo(t.Context(), bucketName, objectName, ObjectOptions{}); err == nil {
		t.Fatalf("%s: object was created despite a failed checksum validation", instanceType)
	}
}

// TestAPICompleteMultipartCompositeStillRequiresPartChecksums locks in that the
// relaxation is scoped to full object checksums. For the algorithm/type pairs
// AWS actually supports as composite, AWS requires a checksum for every part in
// the CompleteMultipartUpload body, and so do we. (CRC64NVME is deliberately not
// covered: AWS does not support it as composite and MinIO canonicalises it to a
// full object checksum at initiation.)
func TestAPICompleteMultipartCompositeStillRequiresPartChecksums(t *testing.T) {
	defer DetectTestLeak(t)()
	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
		t:          t,
		objAPITest: testAPICompleteMultipartCompositeStillRequiresPartChecksums,
		endpoints:  []string{"NewMultipart", "PutObjectPart", "CompleteMultipart"},
	})
}

func testAPICompleteMultipartCompositeStillRequiresPartChecksums(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
	credentials auth.Credentials, t *testing.T,
) {
	partData, _ := multipartChecksumTestData()

	for _, typ := range []hash.ChecksumType{hash.ChecksumCRC32, hash.ChecksumSHA256} {
		objectName := "uploads/composite-" + typ.String()

		uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, objectName,
			typ.String(), xhttp.AmzChecksumTypeComposite)
		etags := uploadPartsHTTP(t, apiRouter, credentials, bucketName, objectName, uploadID, typ, partData)

		rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, objectName, uploadID, etags, nil, nil)
		if rec.Code != http.StatusBadRequest {
			t.Fatalf("%s/%s: composite CompleteMultipartUpload without part checksums returned %d, want 400",
				instanceType, typ.String(), rec.Code)
		}
		if got := apiErrorCode(t, rec); got != "InvalidPart" {
			t.Fatalf("%s/%s: expected InvalidPart, got %q", instanceType, typ.String(), got)
		}
		if _, err := obj.GetObjectInfo(t.Context(), bucketName, objectName, ObjectOptions{}); err == nil {
			t.Fatalf("%s/%s: object was created despite a rejected completion", instanceType, typ.String())
		}
	}
}

// TestAPICompleteMultipartFullObjectVariants pins down the surrounding
// behavior of the relaxation: what may be omitted, what must still match, and
// that a zero length object is handled like any other.
func TestAPICompleteMultipartFullObjectVariants(t *testing.T) {
	defer DetectTestLeak(t)()
	ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
		t:          t,
		objAPITest: testAPICompleteMultipartFullObjectVariants,
		endpoints:  []string{"NewMultipart", "PutObjectPart", "CompleteMultipart"},
	})
}

func testAPICompleteMultipartFullObjectVariants(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
	credentials auth.Credentials, t *testing.T,
) {
	typ := hash.ChecksumCRC32
	partData, full := multipartChecksumTestData()
	goodCS := []string{mustChecksum(t, typ, partData[0]), mustChecksum(t, typ, partData[1])}

	setup := func(name string) (string, []string) {
		uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, name,
			typ.String(), xhttp.AmzChecksumTypeFullObject)
		return uploadID, uploadPartsHTTP(t, apiRouter, credentials, bucketName, name, uploadID, typ, partData)
	}
	objCSHdr := map[string]string{
		typ.Key():             mustChecksum(t, typ, full),
		xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
	}

	t.Run("mixed-present-and-omitted", func(t *testing.T) {
		name := "variants/mixed"
		uploadID, etags := setup(name)
		rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, name, uploadID, etags,
			[]string{goodCS[0], ""}, objCSHdr)
		if rec.Code != http.StatusOK {
			t.Fatalf("%s: want 200, got %d %s", instanceType, rec.Code, rec.Body.String())
		}
	})

	t.Run("supplied-part-checksum-must-match", func(t *testing.T) {
		name := "variants/wrong-part-cs"
		uploadID, etags := setup(name)
		rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, name, uploadID, etags,
			[]string{mustChecksum(t, typ, []byte("wrong")), ""}, objCSHdr)
		if rec.Code != http.StatusBadRequest {
			t.Fatalf("%s: a non-empty but wrong part checksum must be rejected, got %d", instanceType, rec.Code)
		}
		if got := apiErrorCode(t, rec); got != "InvalidPart" {
			t.Fatalf("%s: expected InvalidPart, got %q", instanceType, got)
		}
	})

	t.Run("wrong-algorithm-part-checksum-is-rejected", func(t *testing.T) {
		// A part carrying a checksum under an algorithm other than the upload's
		// is malformed, not "omitted", and must not slip through the relaxation.
		name := "variants/wrong-algo-part-cs"
		uploadID, etags := setup(name)
		var body bytes.Buffer
		body.WriteString("<CompleteMultipartUpload>")
		for i, etag := range etags {
			fmt.Fprintf(&body, "<Part><PartNumber>%d</PartNumber><ETag>%s</ETag>"+
				"<ChecksumCRC32C>AAAAAA==</ChecksumCRC32C></Part>", i+1, etag)
		}
		body.WriteString("</CompleteMultipartUpload>")

		req, err := newTestSignedRequestV4(http.MethodPost,
			getCompleteMultipartUploadURL("", bucketName, name, uploadID),
			int64(body.Len()), bytes.NewReader(body.Bytes()), credentials.AccessKey, credentials.SecretKey, objCSHdr)
		if err != nil {
			t.Fatalf("failed to build CompleteMultipartUpload request: %v", err)
		}
		rec := httptest.NewRecorder()
		apiRouter.ServeHTTP(rec, req)
		if rec.Code != http.StatusBadRequest {
			t.Fatalf("%s: a part checksum under the wrong algorithm must be rejected, got %d",
				instanceType, rec.Code)
		}
		if got := apiErrorCode(t, rec); got != "InvalidPart" {
			t.Fatalf("%s: expected InvalidPart, got %q", instanceType, got)
		}
	})

	t.Run("no-object-checksum-supplied", func(t *testing.T) {
		// AWS treats the object level checksum on completion as optional; the
		// server stores the checksum it computed from the parts.
		name := "variants/no-object-cs"
		uploadID, etags := setup(name)
		rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, name, uploadID, etags, nil, nil)
		if rec.Code != http.StatusOK {
			t.Fatalf("%s: want 200, got %d %s", instanceType, rec.Code, rec.Body.String())
		}
		oi, err := obj.GetObjectInfo(t.Context(), bucketName, name, ObjectOptions{})
		if err != nil {
			t.Fatalf("%s: GetObjectInfo failed: %v", instanceType, err)
		}
		cs, _ := oi.decryptChecksums(0, nil)
		if got, want := cs[typ.String()], mustChecksum(t, typ, full); got != want {
			t.Fatalf("%s: expected server computed checksum %q, got %q", instanceType, want, got)
		}
	})

	t.Run("zero-length-object", func(t *testing.T) {
		// A single empty part is a legal multipart upload: only non-final parts
		// have a minimum size. The merged checksum must be the checksum of no
		// bytes, not the empty string.
		name := "variants/zero-length"
		empty := []byte{}
		uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, name,
			typ.String(), xhttp.AmzChecksumTypeFullObject)
		etags := uploadPartsHTTP(t, apiRouter, credentials, bucketName, name, uploadID, typ, [][]byte{empty})

		rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, name, uploadID, etags, nil,
			map[string]string{
				typ.Key():             mustChecksum(t, typ, empty),
				xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
			})
		if rec.Code != http.StatusOK {
			t.Fatalf("%s: want 200, got %d %s", instanceType, rec.Code, rec.Body.String())
		}
		oi, err := obj.GetObjectInfo(t.Context(), bucketName, name, ObjectOptions{})
		if err != nil {
			t.Fatalf("%s: GetObjectInfo failed: %v", instanceType, err)
		}
		if oi.Size != 0 {
			t.Fatalf("%s: expected zero length object, got %d", instanceType, oi.Size)
		}
		cs, _ := oi.decryptChecksums(0, nil)
		if got, want := cs[typ.String()], mustChecksum(t, typ, empty); got != want {
			t.Fatalf("%s: expected stored checksum %q, got %q", instanceType, want, got)
		}
	})
}