[#607] Support sigV4 streaming with trailers

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>

api/auth/signer/v4sdk2/signer/v4/stream.go

@@ -1,4 +1,6 @@
 // This is https://github.com/aws/aws-sdk-go-v2/blob/a2b751d1ba71f59175a41f9cae5f159f1044360f/aws/signer/v4/stream.go
+// with changes
+//   * add GetTrailingSignature
 
 package v4
 
@@ -87,3 +89,32 @@ func (s *StreamSigner) buildEventStreamStringToSign(headers, payload, previousSi
 		hex.EncodeToString(makeHash(hash, payload)),
 	}, "\n")
 }
+
+// GetTrailerSignature signs the provided header and payload bytes.
+func (s *StreamSigner) GetTrailerSignature(payload []byte, signingTime time.Time) ([]byte, error) {
+	prevSignature := s.prevSignature
+
+	st := v4Internal.NewSigningTime(signingTime)
+
+	sigKey := s.signingKeyDeriver.DeriveKey(s.credentials, s.service, s.region, st)
+
+	scope := v4Internal.BuildCredentialScope(st, s.region, s.service)
+
+	stringToSign := s.buildEventStreamStringToSignTrailer(payload, prevSignature, scope, &st)
+
+	signature := v4Internal.HMACSHA256(sigKey, []byte(stringToSign))
+	s.prevSignature = signature
+
+	return signature, nil
+}
+
+func (s *StreamSigner) buildEventStreamStringToSignTrailer(payload, previousSignature []byte, credentialScope string, signingTime *v4Internal.SigningTime) string {
+	hash := sha256.New()
+	return strings.Join([]string{
+		"AWS4-HMAC-SHA256-TRAILER",
+		signingTime.TimeFormat(),
+		credentialScope,
+		hex.EncodeToString(previousSignature),
+		hex.EncodeToString(makeHash(hash, payload)),
+	}, "\n")
+}

api/handler/put_test.go

@@ -425,17 +425,27 @@ func TestPutObjectWithStreamBodyAWSExampleTrailing(t *testing.T) {
 	bktName, objName := "examplebucket", "chunkObject.txt"
 	createTestBucket(hc, bktName)
 
-	w, req, chunk := getChunkedRequestTrailing(hc.context, t, bktName, objName)
-	hc.Handler().PutObjectHandler(w, req)
-	assertStatus(t, w, http.StatusOK)
+	t.Run("valid trailer signature", func(t *testing.T) {
+		w, req, chunk := getChunkedRequestTrailing(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))
+		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)
-	equalDataSlices(t, chunk, data)
+		data := getObjectRange(t, hc, bktName, objName, 0, awsChunkedRequestExampleDecodedContentLength)
+		equalDataSlices(t, chunk, data)
+	})
+
+	t.Run("invalid trailer signature", func(t *testing.T) {
+		w, req, _ := getChunkedRequestTrailing(hc.context, t, bktName, objName)
+		body := req.Body.(*customNopCloser)
+		body.Bytes()[body.Len()-2] = 'a'
+		hc.Handler().PutObjectHandler(w, req)
+		assertStatus(t, w, http.StatusForbidden)
+	})
 }
 
 func TestPutObjectWithStreamBodyAWSExample(t *testing.T) {
@@ -571,6 +581,14 @@ func getChunkedRequest(ctx context.Context, t *testing.T, bktName, objName strin
 	return w, req, chunk
 }
 
+type customNopCloser struct {
+	*bytes.Buffer
+}
+
+func (c *customNopCloser) Close() error {
+	return nil
+}
+
 // getChunkedRequestTrailing implements request example from
 // https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-streaming-trailers.html
 func getChunkedRequestTrailing(ctx context.Context, t *testing.T, bktName, objName string) (*httptest.ResponseRecorder, *http.Request, []byte) {
@@ -596,9 +614,9 @@ func getChunkedRequestTrailing(ctx context.Context, t *testing.T, bktName, objNa
 	require.NoError(t, err)
 	_, err = reqBody.WriteString("\r\n0;chunk-signature=2ca2aba2005185cf7159c6277faf83795951dd77a3a99e6e65d5c9f85863f992\r\n")
 	require.NoError(t, err)
-	_, err = reqBody.WriteString("\r\nx-amz-checksum-crc32c:sOO8/Q==\r\n")
+	_, err = reqBody.WriteString("\r\nx-amz-checksum-crc32c:sOO8/Q==\n")
 	require.NoError(t, err)
-	_, err = reqBody.WriteString("\r\nx-amz-trailer-signature:63bddb248ad2590c92712055f51b8e78ab024eead08276b24f010b0efd74843f\r\n\r\n")
+	_, err = reqBody.WriteString("x-amz-trailer-signature:63bddb248ad2590c92712055f51b8e78ab024eead08276b24f010b0efd74843f")
 	require.NoError(t, err)
 
 	req, err := http.NewRequest("PUT", "https://s3.amazonaws.com/"+bktName+"/"+objName, nil)
@@ -608,7 +626,7 @@ func getChunkedRequestTrailing(ctx context.Context, t *testing.T, bktName, objNa
 	req.Header.Set("x-amz-content-sha256", api.StreamingContentSHA256Trailer)
 	req.Header.Set("x-amz-decoded-content-length", strconv.Itoa(awsChunkedRequestExampleDecodedContentLength))
 	req.Header.Set("x-amz-storage-class", "REDUCED_REDUNDANCY")
-	req.Header.Set("x-amz-trailer", "x-amz-checksum-crc32")
+	req.Header.Set("x-amz-trailer", "x-amz-checksum-crc32c")
 
 	signTime, err := time.Parse("20060102T150405Z", "20130524T000000Z")
 	require.NoError(t, err)
@@ -616,7 +634,7 @@ func getChunkedRequestTrailing(ctx context.Context, t *testing.T, bktName, objNa
 	err = signer.SignHTTP(ctx, awsCreds, req, api.StreamingContentSHA256Trailer, "s3", "us-east-1", signTime)
 	require.NoError(t, err)
 
-	req.Body = io.NopCloser(reqBody)
+	req.Body = &customNopCloser{Buffer: reqBody}
 
 	w := httptest.NewRecorder()
 	reqInfo := middleware.NewReqInfo(w, req, middleware.ObjectRequest{Bucket: bktName, Object: objName}, "")

api/handler/s3reader.go

@@ -8,6 +8,8 @@ import (
 	"errors"
 	"io"
 	"net/http"
+	"slices"
+	"strings"
 	"time"
 
 	v4 "git.frostfs.info/TrueCloudLab/frostfs-s3-gw/api/auth/signer/v4sdk2/signer/v4"
@@ -27,10 +29,12 @@ type (
 		reader       *bufio.Reader
 		streamSigner *v4.StreamSigner
 
-		requestTime time.Time
-		buffer      []byte
-		offset      int
-		err         error
+		trailerHeaders []string
+		trailers       map[string]string
+		requestTime    time.Time
+		buffer         []byte
+		offset         int
+		err            error
 	}
 )
 
@@ -108,29 +112,9 @@ func (c *s3ChunkReader) Read(buf []byte) (num int, err error) {
 		c.err = errMalformedChunkedEncoding
 		return num, c.err
 	}
-	b, err := c.reader.ReadByte()
-	if err == io.EOF {
-		err = io.ErrUnexpectedEOF
-	}
-	if err != nil {
-		c.err = err
-		return num, c.err
-	}
-	if b != '\r' {
-		c.err = errMalformedChunkedEncoding
-		return num, c.err
-	}
-	b, err = c.reader.ReadByte()
-	if err == io.EOF {
-		err = io.ErrUnexpectedEOF
-	}
-	if err != nil {
-		c.err = err
-		return num, c.err
-	}
-	if b != '\n' {
-		c.err = errMalformedChunkedEncoding
-		return num, c.err
+
+	if err = c.readCRLF(); err != nil {
+		return num, err
 	}
 
 	if cap(c.buffer) < size {
@@ -148,23 +132,6 @@ func (c *s3ChunkReader) Read(buf []byte) (num int, err error) {
 		c.err = err
 		return num, c.err
 	}
-	b, err = c.reader.ReadByte()
-	if b != '\r' || err != nil {
-		c.err = errMalformedChunkedEncoding
-		return num, c.err
-	}
-	b, err = c.reader.ReadByte()
-	if err == io.EOF {
-		err = io.ErrUnexpectedEOF
-	}
-	if err != nil {
-		c.err = err
-		return num, c.err
-	}
-	if b != '\n' {
-		c.err = errMalformedChunkedEncoding
-		return num, c.err
-	}
 
 	// Once we have read the entire chunk successfully, we verify
 	// that the received signature matches our computed signature.
@@ -182,19 +149,98 @@ func (c *s3ChunkReader) Read(buf []byte) (num int, err error) {
 	// If the chunk size is zero we return io.EOF. As specified by AWS,
 	// only the last chunk is zero-sized.
 	if size == 0 {
+		if len(c.trailerHeaders) != 0 {
+			if err = c.readTrailers(); err != nil {
+				c.err = err
+				return num, c.err
+			}
+		} else if err = c.readCRLF(); err != nil {
+			return num, err
+		}
+
 		c.err = io.EOF
 		return num, c.err
 	}
 
+	if err = c.readCRLF(); err != nil {
+		return num, err
+	}
+
 	c.offset = copy(buf, c.buffer)
 	num += c.offset
 	return num, err
 }
 
-func (c *s3ChunkReader) TrailerHeaders() map[string]string {
+func (c *s3ChunkReader) readCRLF() error {
+	for _, ch := range [2]byte{'\r', '\n'} {
+		b, err := c.reader.ReadByte()
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
+		}
+
+		if err != nil {
+			c.err = err
+			return c.err
+		}
+		if b != ch {
+			c.err = errMalformedChunkedEncoding
+			return c.err
+		}
+	}
+
 	return nil
 }
 
+func (c *s3ChunkReader) readTrailers() error {
+	var k, v []byte
+	var err error
+	for err == nil {
+		k, err = c.reader.ReadBytes(':')
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			c.err = errMalformedTrailerHeaders
+			return c.err
+		}
+		v, err = c.reader.ReadBytes('\n')
+		if err != nil && err != io.EOF {
+			c.err = errMalformedTrailerHeaders
+			return c.err
+		}
+		if len(v) >= 2 && v[len(v)-2] == '\r' {
+			v[len(v)-2] = '\n'
+			v = v[:len(v)-1]
+		}
+
+		switch {
+		case slices.Contains(c.trailerHeaders, string(k[:len(k)-1])):
+			c.buffer = append(append(c.buffer, k...), v...) // todo use copy
+		case string(k) == "x-amz-trailer-signature:":
+			calculatedSignature, err := c.streamSigner.GetTrailerSignature(c.buffer, c.requestTime)
+			if err != nil {
+				c.err = err
+				return c.err
+			}
+			if string(v[:64]) != hex.EncodeToString(calculatedSignature) {
+				c.err = errs.GetAPIError(errs.ErrSignatureDoesNotMatch)
+				return c.err
+			}
+		default:
+			c.err = errMalformedTrailerHeaders
+			return c.err
+		}
+
+		c.trailers[string(k[:len(k)-1])] = string(v[:len(v)-1])
+	}
+
+	return nil
+}
+
+func (c *s3ChunkReader) TrailerHeaders() map[string]string {
+	return c.trailers
+}
+
 func newSignV4ChunkedReader(req *http.Request) (*s3ChunkReader, error) {
 	ctx := req.Context()
 	box, err := middleware.GetBoxData(ctx)
@@ -219,11 +265,19 @@ func newSignV4ChunkedReader(req *http.Request) (*s3ChunkReader, error) {
 	}
 	newStreamSigner := v4.NewStreamSigner(currentCredentials, "s3", authHeaders.Region, seed)
 
+	var trailerHeaders []string
+	trailer := req.Header.Get("x-amz-trailer")
+	if trailer != "" {
+		trailerHeaders = strings.Split(trailer, ";")
+	}
+
 	return &s3ChunkReader{
-		ctx:          ctx,
-		reader:       bufio.NewReader(req.Body),
-		streamSigner: newStreamSigner,
-		requestTime:  reqTime,
-		buffer:       make([]byte, 64*1024),
+		ctx:            ctx,
+		reader:         bufio.NewReader(req.Body),
+		streamSigner:   newStreamSigner,
+		requestTime:    reqTime,
+		buffer:         make([]byte, 64*1024),
+		trailerHeaders: trailerHeaders,
+		trailers:       make(map[string]string, len(trailerHeaders)),
 	}, nil
 }

api/handler/s3reader_test.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -59,7 +60,77 @@ func TestSigV4AStreaming(t *testing.T) {
 	require.Equal(t, chunk1, string(data))
 }
 
-func TestStreamingUnsigned(t *testing.T) {
+func TestSigV4ChunkedReader(t *testing.T) {
+	accessKeyID := "9uEm8zMrGWsEDWiPCnVuQLKTiGtCEXpYXt8eBG7agupw0JDySJZMFuej7PTcPzRqBUyPtFowNu1RtvHULU8XHjie6"
+	secretKey := "9f546428957ed7e189b7be928906ce7d1d9cb3042dd4d2d5194e28ce8c4c3b8e"
+
+	signature := "b740b3b2a08c541c3fc4bd155a448e25408b509a29af98a86356b894930b93e8"
+	signingTime, err := time.Parse("20060102T150405Z", "20250203T134442Z")
+	require.NoError(t, err)
+
+	key, err := keys.NewPrivateKey()
+	require.NoError(t, err)
+
+	accessBox, err := newTestAccessBox(key)
+	require.NoError(t, err)
+	accessBox.Gate.SecretKey = secretKey
+
+	setBoxFn := func(ctx context.Context) context.Context {
+		return middleware.SetBox(ctx, &middleware.Box{
+			AccessBox: accessBox,
+			AuthHeaders: &middleware.AuthHeader{
+				AccessKeyID: accessKeyID,
+				SignatureV4: signature,
+				Region:      "us-east-1",
+			},
+			ClientTime: signingTime,
+		})
+	}
+
+	chunk1 := "Testing with the {sdk-java}"
+
+	t.Run("with trailers", func(t *testing.T) {
+		body := "1b;chunk-signature=a6a9be5fff05db0b542aedb2203d892b4162250885d06b1422b173ee0ea92ba5\r\n" +
+			chunk1 +
+			"\r\n0;chunk-signature=31afd083a57c416c46afaf101649d7f0c6c0627cfa60c0f93d1f7ea84396ee42\r\n" +
+			"x-amz-checksum-crc32:Np6zMg==\r\n" +
+			"x-amz-trailer-signature:40ec0046ac730fa27a1451d00d849056c49553ee753f5d158306d05671a42125\r\n\r\n"
+
+		reqBody := bytes.NewBufferString(body)
+		req, err := http.NewRequest("PUT", "https://localhost:8184/test2/tmp", reqBody)
+		require.NoError(t, err)
+		req.Header.Set("x-amz-trailer", "x-amz-checksum-crc32")
+
+		req = req.WithContext(setBoxFn(req.Context()))
+
+		r, err := newSignV4ChunkedReader(req)
+		require.NoError(t, err)
+
+		data, err := io.ReadAll(r)
+		require.NoError(t, err)
+		require.Equal(t, chunk1, string(data))
+	})
+
+	t.Run("without trailers", func(t *testing.T) {
+		body := "1b;chunk-signature=a6a9be5fff05db0b542aedb2203d892b4162250885d06b1422b173ee0ea92ba5\r\n" +
+			chunk1 +
+			"\r\n0;chunk-signature=31afd083a57c416c46afaf101649d7f0c6c0627cfa60c0f93d1f7ea84396ee42\r\n\r\n"
+		reqBody := bytes.NewBufferString(body)
+		req, err := http.NewRequest("PUT", "https://localhost:8184/test2/tmp", reqBody)
+		require.NoError(t, err)
+
+		req = req.WithContext(setBoxFn(req.Context()))
+
+		r, err := newSignV4ChunkedReader(req)
+		require.NoError(t, err)
+
+		data, err := io.ReadAll(r)
+		require.NoError(t, err)
+		require.Equal(t, chunk1, string(data))
+	})
+}
+
+func TestUnsignedChunkReader(t *testing.T) {
 	chunk1 := "chunk1"
 	chunk2 := "chunk2"