fix: fixes unsigned streaming upload parsing and checksum calculation

Fixes #1600
Fixes #1603
Fixes #1607
Fixes #1626
Fixes #1632
Fixes #1652
Fixes #1653
Fixes #1656
Fixes #1657
Fixes #1659

This PR focuses mainly on unsigned streaming payload **trailer request payload parsing** and **checksum calculation**. For streaming uploads, there are essentially two ways to specify checksums:

1. via `x-amz-checksum-*` headers,
2. via `x-amz-trailer`,
   or none — in which case the checksum should default to **crc64nvme**.

Previously, the implementation calculated the checksum only from `x-amz-checksum-*` headers. Now, `x-amz-trailer` is also treated as a checksum-related header and indicates the checksum algorithm for streaming requests. If `x-amz-trailer` is present, the payload must include a trailing checksum; otherwise, an error is returned.

`x-amz-trailer` and any `x-amz-checksum-*` header **cannot** be used together — doing so results in an error.

If `x-amz-sdk-checksum-algorithm` is specified, then either `x-amz-trailer` or one of the `x-amz-checksum-*` headers must also be present, and the algorithms must match. If they don’t, an error is returned.

The old implementation used to return an internal error when no `x-amz-trailer` was received in streaming requests or when the payload didn’t contain a trailer. This is now fixed.

Checksum calculation used to happen twice in the gateway (once in the chunk reader and once in the backend). A new `ChecksumReader` is introduced to prevent double computation, and the trailing checksum is now read by the backend from the chunk reader. The logic for stacking `io.Reader`s in the Fiber context is preserved, but extended: once a `ChecksumReader` is stacked, all following `io.Reader`s are wrapped with `MockChecksumReader`, which simply delegates to the underlying checksum reader. In the backend, a simple type assertion on `io.Reader` provides the necessary checksum metadata (algorithm, value, etc.).

backend/posix/posix.go

@@ -40,6 +40,7 @@ import (
 	"github.com/versity/versitygw/backend"
 	"github.com/versity/versitygw/backend/meta"
 	"github.com/versity/versitygw/debuglogger"
+	"github.com/versity/versitygw/s3api/middlewares"
 	"github.com/versity/versitygw/s3api/utils"
 	"github.com/versity/versitygw/s3err"
 	"github.com/versity/versitygw/s3response"
@@ -2880,10 +2881,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 	if po.Key == nil {
 		return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrNoSuchKey)
 	}
-	// Override the checksum algorithm with default: CRC64NVME
-	if po.ChecksumAlgorithm == "" {
-		po.ChecksumAlgorithm = types.ChecksumAlgorithmCrc64nvme
-	}
 	if !p.isBucketValid(*po.Bucket) {
 		return s3response.PutObjectOutput{}, s3err.GetAPIError(s3err.ErrInvalidBucketName)
 	}
@@ -3017,35 +3014,54 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 	hash := md5.New()
 	rdr := io.TeeReader(po.Body, hash)
 
-	hashConfigs := []hashConfig{
-		{po.ChecksumCRC32, utils.HashTypeCRC32},
-		{po.ChecksumCRC32C, utils.HashTypeCRC32C},
-		{po.ChecksumSHA1, utils.HashTypeSha1},
-		{po.ChecksumSHA256, utils.HashTypeSha256},
-		{po.ChecksumCRC64NVME, utils.HashTypeCRC64NVME},
-	}
 	var hashRdr *utils.HashReader
 
-	for _, config := range hashConfigs {
-		if config.value != nil {
-			hashRdr, err = utils.NewHashReader(rdr, *config.value, config.hashType)
+	chRdr, chunkUpload := po.Body.(middlewares.ChecksumReader)
+	isTrailingChecksum := chunkUpload && chRdr.Algorithm() != ""
+
+	if !isTrailingChecksum {
+		hashConfigs := []hashConfig{
+			{po.ChecksumCRC32, utils.HashTypeCRC32},
+			{po.ChecksumCRC32C, utils.HashTypeCRC32C},
+			{po.ChecksumSHA1, utils.HashTypeSha1},
+			{po.ChecksumSHA256, utils.HashTypeSha256},
+			{po.ChecksumCRC64NVME, utils.HashTypeCRC64NVME},
+		}
+
+		for _, config := range hashConfigs {
+			if config.value != nil {
+				hashRdr, err = utils.NewHashReader(rdr, *config.value, config.hashType)
+				if err != nil {
+					return s3response.PutObjectOutput{}, fmt.Errorf("initialize hash reader: %w", err)
+				}
+
+				rdr = hashRdr
+			}
+		}
+
+		// If only the checksum algorithm is provided register
+		// a new HashReader to calculate the object checksum
+		// This can never happen with PutObject direct call
+		// it's there for CopyObject to add a new checksum
+		if hashRdr == nil && po.ChecksumAlgorithm != "" {
+			hashRdr, err = utils.NewHashReader(rdr, "", utils.HashType(strings.ToLower(string(po.ChecksumAlgorithm))))
 			if err != nil {
 				return s3response.PutObjectOutput{}, fmt.Errorf("initialize hash reader: %w", err)
 			}
 
 			rdr = hashRdr
 		}
-	}
 
-	// If only the checksum algorithm is provided register
-	// a new HashReader to calculate the object checksum
-	if hashRdr == nil && po.ChecksumAlgorithm != "" {
-		hashRdr, err = utils.NewHashReader(rdr, "", utils.HashType(strings.ToLower(string(po.ChecksumAlgorithm))))
-		if err != nil {
-			return s3response.PutObjectOutput{}, fmt.Errorf("initialize hash reader: %w", err)
+		if hashRdr == nil {
+			// if no precalculated checksum or sdk checksum algorithm is provided
+			// and no streaming upload has checksum, default to crc64nvme
+			hashRdr, err = utils.NewHashReader(rdr, "", utils.HashTypeCRC64NVME)
+			if err != nil {
+				return s3response.PutObjectOutput{}, fmt.Errorf("initialize hash reader: %w", err)
+			}
+
+			rdr = hashRdr
 		}
-
-		rdr = hashRdr
 	}
 
 	_, err = io.Copy(f, rdr)
@@ -3095,15 +3111,24 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 		}
 	}
 
+	var chAlgo utils.HashType
+	var sum string
+
+	if isTrailingChecksum {
+		fmt.Println("reading from result reader")
+		chAlgo = utils.HashType(chRdr.Algorithm())
+		sum = chRdr.Checksum()
+	} else if hashRdr != nil {
+		chAlgo = hashRdr.Type()
+		sum = hashRdr.Sum()
+	}
+
 	checksum := s3response.Checksum{}
 
 	// Store the calculated checksum in the object metadata
-	if hashRdr != nil {
-		// The checksum type is always FULL_OBJECT for PutObject
+	if sum != "" {
 		checksum.Type = types.ChecksumTypeFullObject
-
-		sum := hashRdr.Sum()
-		switch hashRdr.Type() {
+		switch chAlgo {
 		case utils.HashTypeCRC32:
 			checksum.CRC32 = &sum
 			checksum.Algorithm = types.ChecksumAlgorithmCrc32
@@ -3120,7 +3145,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			checksum.CRC64NVME = &sum
 			checksum.Algorithm = types.ChecksumAlgorithmCrc64nvme
 		}
-
 		err := p.storeChecksums(f.File(), *po.Bucket, *po.Key, checksum)
 		if err != nil {
 			return s3response.PutObjectOutput{}, fmt.Errorf("store checksum: %w", err)

s3api/middlewares/body-reader.go

@@ -22,17 +22,83 @@ import (
 	"github.com/versity/versitygw/s3api/utils"
 )
 
+// ChecksumReader extends io.Reader with checksum-related metadata.
+// It is used to differentiate normal readers from readers that can
+// report a checksum and the algorithm used to produce it.
+type ChecksumReader interface {
+	io.Reader
+	Algorithm() string
+	Checksum() string
+}
+
+// NewChecksumReader wraps a stackedReader and returns a reader that
+// preserves checksum behavior when the *original* bodyReader implemented
+// ChecksumReader.
+//
+// If bodyReader already supports ChecksumReader, we wrap stackedReader
+// with MockChecksumReader so that reading continues from stackedReader,
+// but Algorithm() and Checksum() still delegate to the underlying reader.
+//
+// If bodyReader is not a ChecksumReader, we simply return stackedReader.
+func NewChecksumReader(bodyReader io.Reader, stackedReader io.Reader) io.Reader {
+	_, ok := bodyReader.(ChecksumReader)
+	if ok {
+		return &MockChecksumReader{rdr: stackedReader}
+	}
+
+	return stackedReader
+}
+
+// MockChecksumReader is a wrapper around an io.Reader that forwards Read()
+// but also conditionally exposes checksum metadata if the underlying reader
+// implements the ChecksumReader interface.
+type MockChecksumReader struct {
+	rdr io.Reader
+}
+
+// Read simply forwards data reads to the underlying reader.
+func (rr *MockChecksumReader) Read(buffer []byte) (int, error) {
+	return rr.rdr.Read(buffer)
+}
+
+// Algorithm returns the checksum algorithm used by the underlying reader,
+// but only if the wrapped reader implements ChecksumReader.
+func (rr *MockChecksumReader) Algorithm() string {
+	r, ok := rr.rdr.(ChecksumReader)
+	if ok {
+		return r.Algorithm()
+	}
+
+	return ""
+}
+
+// Checksum returns the checksum value from the underlying reader,
+// if it implements ChecksumReader. Otherwise returns an empty string.
+func (rr *MockChecksumReader) Checksum() string {
+	r, ok := rr.rdr.(ChecksumReader)
+	if ok {
+		return r.Checksum()
+	}
+
+	return ""
+}
+
+var _ ChecksumReader = &MockChecksumReader{}
+
 func wrapBodyReader(ctx *fiber.Ctx, wr func(io.Reader) io.Reader) {
-	r, ok := utils.ContextKeyBodyReader.Get(ctx).(io.Reader)
+	rdr, ok := utils.ContextKeyBodyReader.Get(ctx).(io.Reader)
 	if !ok {
-		r = ctx.Request().BodyStream()
+		rdr = ctx.Request().BodyStream()
 		// Override the body reader with an empty reader to prevent panics
 		// in case of unexpected or malformed HTTP requests.
-		if r == nil {
-			r = bytes.NewBuffer([]byte{})
+		if rdr == nil {
+			rdr = bytes.NewBuffer([]byte{})
 		}
 	}
 
-	r = wr(r)
+	r := wr(rdr)
+	// Ensure checksum behavior is stacked if the original body reader had it.
+	r = NewChecksumReader(rdr, r)
+
 	utils.ContextKeyBodyReader.Set(ctx, r)
 }

s3api/utils/chunk-reader.go

@@ -170,7 +170,7 @@ func NewChunkReader(ctx *fiber.Ctx, r io.Reader, authdata AuthData, region, secr
 	//TODO: not sure if InvalidRequest should be returned in this case
 	if err != nil {
 		debuglogger.Logf("invalid value for 'X-Amz-Decoded-Content-Length': %v", decContLengthStr)
-		return nil, s3err.GetAPIError(s3err.ErrInvalidRequest)
+		return nil, s3err.GetAPIError(s3err.ErrMissingContentLength)
 	}
 
 	if decContLength > maxObjSizeLimit {
@@ -189,10 +189,6 @@ func NewChunkReader(ctx *fiber.Ctx, r io.Reader, authdata AuthData, region, secr
 	if err != nil {
 		return nil, err
 	}
-	if contentSha256 != payloadTypeStreamingSigned && checksumType == "" {
-		debuglogger.Logf("empty value for required trailer header 'X-Amz-Trailer': %v", checksumType)
-		return nil, s3err.GetAPIError(s3err.ErrTrailerHeaderNotSupported)
-	}
 
 	switch contentSha256 {
 	case payloadTypeStreamingUnsignedTrailer:

s3api/utils/unsigned-chunk-reader.go

@@ -21,7 +21,6 @@ import (
 	"crypto/sha256"
 	"encoding/base64"
 	"errors"
-	"fmt"
 	"hash"
 	"hash/crc32"
 	"hash/crc64"
@@ -30,7 +29,9 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/versity/versitygw/debuglogger"
+	"github.com/versity/versitygw/s3err"
 )
 
 var (
@@ -39,20 +40,25 @@ var (
 )
 
 type UnsignedChunkReader struct {
-	reader           *bufio.Reader
-	checksumType     checksumType
-	expectedChecksum string
-	hasher           hash.Hash
-	stash            []byte
-	offset           int
+	reader         *bufio.Reader
+	checksumType   checksumType
+	parsedChecksum string
+	hasher         hash.Hash
+	stash          []byte
+	offset         int
 }
 
 func NewUnsignedChunkReader(r io.Reader, ct checksumType) (*UnsignedChunkReader, error) {
-	hasher, err := getHasher(ct)
+	var hasher hash.Hash
+	var err error
+	if ct != "" {
+		hasher, err = getHasher(ct)
+	}
 	if err != nil {
 		debuglogger.Logf("failed to initialize hash calculator: %v", err)
 		return nil, err
 	}
+
 	debuglogger.Infof("initializing unsigned chunk reader")
 	return &UnsignedChunkReader{
 		reader:       bufio.NewReader(r),
@@ -62,6 +68,16 @@ func NewUnsignedChunkReader(r io.Reader, ct checksumType) (*UnsignedChunkReader,
 	}, nil
 }
 
+// Algorithm returns the checksum algorithm
+func (ucr *UnsignedChunkReader) Algorithm() string {
+	return strings.TrimPrefix(string(ucr.checksumType), "x-amz-checksum-")
+}
+
+// Checksum returns the parsed trailing checksum
+func (ucr *UnsignedChunkReader) Checksum() string {
+	return ucr.parsedChecksum
+}
+
 func (ucr *UnsignedChunkReader) Read(p []byte) (int, error) {
 	// First read any stashed data
 	if len(ucr.stash) != 0 {
@@ -87,7 +103,10 @@ func (ucr *UnsignedChunkReader) Read(p []byte) (int, error) {
 			// Stop reading parsing payloads as 0 sized chunk is reached
 			break
 		}
-		rdr := io.TeeReader(ucr.reader, ucr.hasher)
+		var rdr io.Reader = ucr.reader
+		if ucr.hasher != nil {
+			rdr = io.TeeReader(ucr.reader, ucr.hasher)
+		}
 		payload := make([]byte, chunkSize)
 		// Read and cache the payload
 		_, err = io.ReadFull(rdr, payload)
@@ -167,6 +186,7 @@ func (ucr *UnsignedChunkReader) extractChunkSize() (int64, error) {
 // Reads and validates the trailer at the end
 func (ucr *UnsignedChunkReader) readTrailer() error {
 	var trailerBuffer bytes.Buffer
+	var hasChecksum bool
 
 	for {
 		v, err := ucr.reader.ReadByte()
@@ -178,9 +198,27 @@ func (ucr *UnsignedChunkReader) readTrailer() error {
 			return err
 		}
 		if v != '\r' {
+			hasChecksum = true
 			trailerBuffer.WriteByte(v)
 			continue
 		}
+
+		if !hasChecksum {
+			// in case the payload doesn't contain trailer
+			// the first 2 bytes(\r\n) have been read
+			// only read the last byte: \n
+			err := ucr.readAndSkip('\n')
+			if err != nil {
+				debuglogger.Logf("failed to read chunk last byte: \\n: %v", err)
+				if err == io.EOF {
+					return io.ErrUnexpectedEOF
+				}
+				return err
+			}
+
+			break
+		}
+
 		var tmp [3]byte
 		_, err = io.ReadFull(ucr.reader, tmp[:])
 		if err != nil {
@@ -200,20 +238,35 @@ func (ucr *UnsignedChunkReader) readTrailer() error {
 	// Parse the trailer
 	trailerHeader := trailerBuffer.String()
 	trailerHeader = strings.TrimSpace(trailerHeader)
+	if trailerHeader == "" {
+		if ucr.checksumType != "" {
+			debuglogger.Logf("expected %s checksum in the paylod, but it's missing", ucr.checksumType)
+			return s3err.GetAPIError(s3err.ErrMalformedTrailer)
+		}
+
+		return nil
+	}
 	trailerHeaderParts := strings.Split(trailerHeader, ":")
 	if len(trailerHeaderParts) != 2 {
 		debuglogger.Logf("invalid trailer header parts: %v", trailerHeaderParts)
-		return errMalformedEncoding
+		return s3err.GetAPIError(s3err.ErrMalformedTrailer)
 	}
 
-	if trailerHeaderParts[0] != string(ucr.checksumType) {
-		debuglogger.Logf("invalid checksum type: %v", trailerHeaderParts[0])
-		//TODO: handle the error
-		return errMalformedEncoding
+	checksumKey := checksumType(trailerHeaderParts[0])
+	checksum := trailerHeaderParts[1]
+
+	if !checksumKey.isValid() {
+		debuglogger.Logf("invalid checksum header key: %s", checksumKey)
+		return s3err.GetAPIError(s3err.ErrMalformedTrailer)
 	}
 
-	ucr.expectedChecksum = trailerHeaderParts[1]
-	debuglogger.Infof("parsed the trailing header:\n%v:%v", trailerHeaderParts[0], trailerHeaderParts[1])
+	if checksumKey != ucr.checksumType {
+		debuglogger.Logf("incorrect checksum type (expected): %s, (actual): %s", ucr.checksumType, checksumKey)
+		return s3err.GetAPIError(s3err.ErrMalformedTrailer)
+	}
+
+	ucr.parsedChecksum = checksum
+	debuglogger.Infof("parsed the trailing header:\n%v:%v", checksumKey, checksum)
 
 	// Validate checksum
 	return ucr.validateChecksum()
@@ -221,17 +274,30 @@ func (ucr *UnsignedChunkReader) readTrailer() error {
 
 // Validates the trailing checksum sent at the end
 func (ucr *UnsignedChunkReader) validateChecksum() error {
-	csum := ucr.hasher.Sum(nil)
-	checksum := base64.StdEncoding.EncodeToString(csum)
+	algo := types.ChecksumAlgorithm(strings.ToUpper(strings.TrimPrefix(string(ucr.checksumType), "x-amz-checksum-")))
+	// validate the checksum
+	if !IsValidChecksum(ucr.parsedChecksum, algo) {
+		debuglogger.Logf("invalid checksum: (algo): %s, (checksum): %s", algo, ucr.parsedChecksum)
+		return s3err.GetInvalidTrailingChecksumHeaderErr(string(ucr.checksumType))
+	}
 
-	if checksum != ucr.expectedChecksum {
-		debuglogger.Logf("incorrect checksum: (expected): %v, (got): %v", ucr.expectedChecksum, checksum)
-		return fmt.Errorf("actual checksum: %v, expected checksum: %v", checksum, ucr.expectedChecksum)
+	checksum := ucr.calculateChecksum()
+
+	// compare the calculated and parsed checksums
+	if checksum != ucr.parsedChecksum {
+		debuglogger.Logf("incorrect checksum: (expected): %v, (got): %v", ucr.parsedChecksum, checksum)
+		return s3err.GetChecksumBadDigestErr(algo)
 	}
 
 	return nil
 }
 
+// calculateChecksum calculates the checksum with the unsigned reader hasher
+func (ucr *UnsignedChunkReader) calculateChecksum() string {
+	csum := ucr.hasher.Sum(nil)
+	return base64.StdEncoding.EncodeToString(csum)
+}
+
 // Retruns the hash calculator based on the hash type provided
 func getHasher(ct checksumType) (hash.Hash, error) {
 	switch ct {

s3api/utils/utils.go

@@ -512,14 +512,33 @@ func ParseChecksumHeadersAndSdkAlgo(ctx *fiber.Ctx) (types.ChecksumAlgorithm, Ch
 		return sdkAlgorithm, checksums, err
 	}
 
-	if len(checksums) == 0 && sdkAlgorithm != "" {
-		if ctx.Get("X-Amz-Trailer") == "" {
-			// This is a special case when x-amz-trailer is there
-			// it means the upload is done with chunked encoding
-			// where the checksum verification is handled in the chunk reader
+	trailer := strings.ToUpper(ctx.Get("X-Amz-Trailer"))
+
+	if len(checksums) != 0 && trailer != "" {
+		// both x-amz-trailer and one of x-amz-checksum-* is not allowed
+		debuglogger.Logf("x-amz-checksum-* header is used with x-amz-trailer: trailer: %s", trailer)
+		return sdkAlgorithm, checksums, s3err.GetAPIError(s3err.ErrMultipleChecksumHeaders)
+	}
+
+	trailerAlgo := strings.TrimPrefix(trailer, "X-AMZ-CHECKSUM-")
+
+	if sdkAlgorithm != "" {
+		if len(checksums) == 0 && trailerAlgo == "" {
+			// in case x-amz-sdk-algorithm is specified, but no corresponging
+			// x-amz-checksum-* or x-amz-trailer is sent
 			debuglogger.Logf("'x-amz-sdk-checksum-algorithm : %s' is used without corresponding x-amz-checksum-* header", sdkAlgorithm)
 			return sdkAlgorithm, checksums, s3err.GetAPIError(s3err.ErrChecksumSDKAlgoMismatch)
 		}
+
+		if trailerAlgo != "" && string(sdkAlgorithm) != trailerAlgo {
+			// x-amz-sdk-checksum-algorithm and x-amz-trailer should match
+			debuglogger.Logf("x-amz-sdk-checksum-algorithm: (%s) and x-amz-trailer: (%s) doesn't match", sdkAlgorithm, trailerAlgo)
+			return sdkAlgorithm, checksums, s3err.GetInvalidChecksumHeaderErr("x-amz-sdk-checksum-algorithm")
+		}
+	}
+
+	if trailerAlgo != "" {
+		sdkAlgorithm = types.ChecksumAlgorithm(trailerAlgo)
 	}
 
 	for al, val := range checksums {

s3err/s3err.go

@@ -178,6 +178,7 @@ const (
 	ErrNotModified
 	ErrInvalidLocationConstraint
 	ErrInvalidArgument
+	ErrMalformedTrailer
 
 	// Non-AWS errors
 	ErrExistingObjectIsDirectory
@@ -792,6 +793,11 @@ var errorCodeResponse = map[ErrorCode]APIError{
 		Description:    "",
 		HTTPStatusCode: http.StatusBadRequest,
 	},
+	ErrMalformedTrailer: {
+		Code:           "MalformedTrailerError",
+		Description:    "The request contained trailing data that was not well-formed or did not conform to our published schema.",
+		HTTPStatusCode: http.StatusBadRequest,
+	},
 
 	// non aws errors
 	ErrExistingObjectIsDirectory: {