Merge pull request #1701 from versity/sis/signed-streaming-upload-error-handling

fix: bunch of fixes in signed streaming requests
This commit is contained in:
Ben McClelland
2025-12-31 00:20:27 -08:00
committed by GitHub
8 changed files with 892 additions and 77 deletions
+1 -1
View File
@@ -134,7 +134,7 @@ func VerifyV4Signature(root RootUserConfig, iam auth.IAMService, region string,
var err error
wrapBodyReader(ctx, func(r io.Reader) io.Reader {
var cr io.Reader
cr, err = utils.NewChunkReader(ctx, r, authData, region, account.Secret, tdate)
cr, err = utils.NewChunkReader(ctx, r, authData, account.Secret, tdate)
return cr
})
if err != nil {
+3 -3
View File
@@ -182,7 +182,7 @@ func ParseDecodedContentLength(ctx *fiber.Ctx) (int64, error) {
return decContLength, nil
}
func NewChunkReader(ctx *fiber.Ctx, r io.Reader, authdata AuthData, region, secret string, date time.Time) (io.Reader, error) {
func NewChunkReader(ctx *fiber.Ctx, r io.Reader, authdata AuthData, secret string, date time.Time) (io.Reader, error) {
cLength, err := ParseDecodedContentLength(ctx)
if err != nil {
return nil, err
@@ -204,9 +204,9 @@ func NewChunkReader(ctx *fiber.Ctx, r io.Reader, authdata AuthData, region, secr
case payloadTypeStreamingUnsignedTrailer:
return NewUnsignedChunkReader(r, checksumType, cLength)
case payloadTypeStreamingSignedTrailer:
return NewSignedChunkReader(r, authdata, region, secret, date, checksumType)
return NewSignedChunkReader(r, authdata, secret, date, checksumType, true, cLength)
case payloadTypeStreamingSigned:
return NewSignedChunkReader(r, authdata, region, secret, date, "")
return NewSignedChunkReader(r, authdata, secret, date, "", false, cLength)
// return not supported for:
// - STREAMING-AWS4-ECDSA-P256-SHA256-PAYLOAD
// - STREAMING-AWS4-ECDSA-P256-SHA256-PAYLOAD-TRAILER
+152 -70
View File
@@ -43,9 +43,16 @@ const (
awsV4 = "AWS4"
awsS3Service = "s3"
awsV4Request = "aws4_request"
trailerSignatureHeader = "x-amz-trailer-signature"
trailerSignatureHeader = "x-amz-trailer-signature:"
streamPayloadAlgo = "AWS4-HMAC-SHA256-PAYLOAD"
streamPayloadTrailerAlgo = "AWS4-HMAC-SHA256-TRAILER"
maxHeaderSize = 1024
)
var (
errskipHeader = errors.New("skip to next header")
delimiter = []byte{'\r', '\n'}
)
// ChunkReader reads from chunked upload request body, and returns
@@ -66,24 +73,31 @@ type ChunkReader struct {
isFirstHeader bool
region string
date time.Time
requireTrailer bool
chunkSizes []int64
cLength int64
dataRead int64
}
// NewChunkReader reads from request body io.Reader and parses out the
// chunk metadata in stream. The headers are validated for proper signatures.
// Reading from the chunk reader will read only the object data stream
// without the chunk headers/trailers.
func NewSignedChunkReader(r io.Reader, authdata AuthData, region, secret string, date time.Time, chType checksumType) (io.Reader, error) {
func NewSignedChunkReader(r io.Reader, authdata AuthData, secret string, date time.Time, chType checksumType, requireTrailer bool, cLength int64) (io.Reader, error) {
chRdr := &ChunkReader{
r: r,
signingKey: getSigningKey(secret, region, date),
signingKey: getSigningKey(secret, authdata.Region, date),
// the authdata.Signature is validated in the auth-reader,
// so we can use that here without any other checks
prevSig: authdata.Signature,
chunkHash: sha256.New(),
isFirstHeader: true,
date: date,
region: region,
trailer: chType,
prevSig: authdata.Signature,
chunkHash: sha256.New(),
isFirstHeader: true,
date: date,
region: authdata.Region,
trailer: chType,
requireTrailer: requireTrailer,
chunkSizes: []int64{},
cLength: cLength,
}
if chType != "" {
@@ -95,7 +109,7 @@ func NewSignedChunkReader(r io.Reader, authdata AuthData, region, secret string,
chRdr.checksumHash = checksumHasher
}
if chType == "" {
if !requireTrailer {
debuglogger.Infof("initializing signed chunk reader")
} else {
debuglogger.Infof("initializing signed chunk reader with '%v' trailing checksum", chType)
@@ -121,7 +135,17 @@ func (cr *ChunkReader) Read(p []byte) (int, error) {
}
}
n, err := cr.parseAndRemoveChunkInfo(p[chunkSize:n])
if err != nil && err != io.EOF {
return 0, err
}
n += int(chunkSize)
cr.dataRead += int64(n)
if cr.isEOF {
if cr.cLength != cr.dataRead {
debuglogger.Logf("number of bytes expected: (%v), number of bytes read: (%v)", cr.cLength, cr.dataRead)
return 0, s3err.GetAPIError(s3err.ErrContentLengthMismatch)
}
}
return n, err
}
@@ -130,6 +154,13 @@ func (cr *ChunkReader) Read(p []byte) (int, error) {
if cr.checksumHash != nil {
cr.checksumHash.Write(p[:n])
}
cr.dataRead += int64(n)
if cr.isEOF {
if cr.cLength != cr.dataRead {
debuglogger.Logf("number of bytes expected: (%v), number of bytes read: (%v)", cr.cLength, cr.dataRead)
return 0, s3err.GetAPIError(s3err.ErrContentLengthMismatch)
}
}
return n, err
}
@@ -328,15 +359,6 @@ func hmac256(key []byte, data []byte) []byte {
return hash.Sum(nil)
}
var (
errInvalidChunkFormat = errors.New("invalid chunk header format")
errskipHeader = errors.New("skip to next header")
)
const (
maxHeaderSize = 1024
)
// This returns the chunk payload size, signature, data start offset, and
// error if any. See the AWS documentation for the chunk header format. The
// header[0] byte is expected to be the first byte of the chunk size here.
@@ -344,7 +366,7 @@ func (cr *ChunkReader) parseChunkHeaderBytes(header []byte) (int64, string, int,
stashLen := len(cr.stash)
if stashLen > maxHeaderSize {
debuglogger.Logf("the stash length exceeds the maximum allowed chunk header size: (stash len): %v, (header limit): %v", stashLen, maxHeaderSize)
return 0, "", 0, errInvalidChunkFormat
return 0, "", 0, s3err.GetAPIError(s3err.ErrIncompleteBody)
}
if cr.stash != nil {
debuglogger.Logf("recovering the stash: (stash len): %v", stashLen)
@@ -360,45 +382,39 @@ func (cr *ChunkReader) parseChunkHeaderBytes(header []byte) (int64, string, int,
// After the first chunk each chunk header should start
// with "\n\r\n"
if !cr.isFirstHeader {
err := readAndSkip(rdr, '\r', '\n')
err := readAndSkip(rdr, delimiter...)
if err != nil {
debuglogger.Logf("failed to read chunk header first 2 bytes: (should be): \\r\\n, (got): %q", header[:min(2, len(header))])
return cr.handleRdrErr(err, header)
}
}
// read and parse the chunk size
chunkSizeStr, err := readAndTrim(rdr, ';')
chunkSize, err := cr.parseChunkSize(rdr, header)
if err != nil {
debuglogger.Logf("failed to read chunk size: %v", err)
return cr.handleRdrErr(err, header)
}
chunkSize, err := strconv.ParseInt(chunkSizeStr, 16, 64)
if err != nil {
debuglogger.Logf("failed to parse chunk size: (size): %v, (err): %v", chunkSizeStr, err)
return 0, "", 0, errInvalidChunkFormat
return 0, "", 0, err
}
// read the chunk signature
err = readAndSkip(rdr, 'c', 'h', 'u', 'n', 'k', '-', 's', 'i', 'g', 'n', 'a', 't', 'u', 'r', 'e', '=')
err = readAndSkip(rdr, []byte("chunk-signature=")...)
if err != nil {
debuglogger.Logf("failed to read 'chunk-signature=': %v", err)
return cr.handleRdrErr(err, header)
}
sig, err := readAndTrim(rdr, '\r')
sig, err := readBytes(rdr, 64)
if err != nil {
debuglogger.Logf("failed to read '\\r', after chunk signature: %v", err)
debuglogger.Logf("failed to read the chunk signature: %v", err)
return cr.handleRdrErr(err, header)
}
err = readAndSkip(rdr, delimiter...)
if err != nil {
debuglogger.Logf("failed to read '\\r\\n' after chunk signature")
return cr.handleRdrErr(err, header)
}
// read and parse the final chunk trailer and checksum
if chunkSize == 0 {
if cr.trailer != "" {
err = readAndSkip(rdr, '\n')
if err != nil {
debuglogger.Logf("failed to read \\n before the trailer: %v", err)
return cr.handleRdrErr(err, header)
}
if cr.requireTrailer {
// parse and validate the trailing header
trailer, err := readAndTrim(rdr, ':')
if err != nil {
@@ -407,7 +423,7 @@ func (cr *ChunkReader) parseChunkHeaderBytes(header []byte) (int64, string, int,
}
if trailer != string(cr.trailer) {
debuglogger.Logf("incorrect trailer prefix: (expected): %v, (got): %v", cr.trailer, trailer)
return 0, "", 0, errInvalidChunkFormat
return 0, "", 0, s3err.GetAPIError(s3err.ErrMalformedTrailer)
}
algo := types.ChecksumAlgorithm(strings.ToUpper(strings.TrimPrefix(trailer, "x-amz-checksum-")))
@@ -419,19 +435,19 @@ func (cr *ChunkReader) parseChunkHeaderBytes(header []byte) (int64, string, int,
return cr.handleRdrErr(err, header)
}
if !IsValidChecksum(checksum, algo) {
debuglogger.Logf("invalid checksum value: %v", checksum)
return 0, "", 0, s3err.GetInvalidTrailingChecksumHeaderErr(trailer)
}
err = readAndSkip(rdr, '\n')
if err != nil {
debuglogger.Logf("failed to read \\n after checksum: %v", err)
return cr.handleRdrErr(err, header)
}
if !IsValidChecksum(checksum, algo) {
debuglogger.Logf("invalid checksum value: %v", checksum)
return 0, "", 0, s3err.GetInvalidTrailingChecksumHeaderErr(trailer)
}
// parse the trailing signature
trailerSigPrefix, err := readAndTrim(rdr, ':')
trailerSigPrefix, err := readBytes(rdr, 24)
if err != nil {
debuglogger.Logf("failed to read trailing signature prefix: %v", err)
return cr.handleRdrErr(err, header)
@@ -439,40 +455,44 @@ func (cr *ChunkReader) parseChunkHeaderBytes(header []byte) (int64, string, int,
if trailerSigPrefix != trailerSignatureHeader {
debuglogger.Logf("invalid trailing signature prefix: (expected): %v, (got): %v", trailerSignatureHeader, trailerSigPrefix)
return 0, "", 0, errInvalidChunkFormat
return 0, "", 0, s3err.GetAPIError(s3err.ErrIncompleteBody)
}
trailerSig, err := readAndTrim(rdr, '\r')
trailerSig, err := readBytes(rdr, 64)
if err != nil {
debuglogger.Logf("failed to read trailing signature: %v", err)
return cr.handleRdrErr(err, header)
}
err = readAndSkip(rdr, delimiter...)
if err != nil {
debuglogger.Logf("failed to read '\\r\\n' after last chunk signature")
return cr.handleRdrErr(err, header)
}
cr.trailerSig = trailerSig
cr.parsedChecksum = checksum
}
// "\r\n\r\n" is followed after the last chunk
err = readAndSkip(rdr, '\n', '\r', '\n')
err = readAndSkip(rdr, delimiter...)
if err != nil {
debuglogger.Logf("failed to read \\n\\r\\n at the end of chunk header: %v", err)
debuglogger.Logf("failed to read \\r\\n at the end of chunk header: %v", err)
return cr.handleRdrErr(err, header)
}
return 0, sig, 0, nil
}
err = readAndSkip(rdr, '\n')
if err != nil {
debuglogger.Logf("failed to read \\n at the end of chunk header: %v", err)
return cr.handleRdrErr(err, header)
}
// add the chunk size at the end of header parsing
// to avoid duplication because of header stashing
cr.addChunkSize(chunkSize)
// find the index of chunk ending: '\r\n'
// skip the first 2 bytes as it is the starting '\r\n'
// the first chunk doesn't contain the starting '\r\n', but
// anyway, trimming the first 2 bytes doesn't pollute the logic.
ind := bytes.Index(header[2:], []byte{'\r', '\n'})
ind := bytes.Index(header[2:], delimiter)
cr.isFirstHeader = false
// the offset is the found index + 4 - the stash length
@@ -495,30 +515,85 @@ func (cr *ChunkReader) stashAndSkipHeader(header []byte) (int64, string, int, er
// calls "cr.stashAndSkipHeader" if the passed err is "io.EOF" and cr.isEOF is false
// Returns the error otherwise
func (cr *ChunkReader) handleRdrErr(err error, header []byte) (int64, string, int, error) {
if err == io.EOF {
if err == io.EOF || err == io.ErrUnexpectedEOF {
if cr.isEOF {
debuglogger.Logf("incomplete chunk encoding, EOF reached")
return 0, "", 0, errInvalidChunkFormat
return 0, "", 0, s3err.GetAPIError(s3err.ErrIncompleteBody)
}
return cr.stashAndSkipHeader(header)
}
return 0, "", 0, err
return 0, "", 0, s3err.GetAPIError(s3err.ErrIncompleteBody)
}
// parseChunkSize parses and validates the chunk size
func (cr *ChunkReader) parseChunkSize(rdr *bufio.Reader, header []byte) (int64, error) {
// read and parse the chunk size
chunkSizeStr, err := readAndTrim(rdr, ';')
if err != nil {
debuglogger.Logf("failed to read chunk size: %v", err)
_, _, _, err := cr.handleRdrErr(err, header)
return 0, err
}
chunkSize, err := strconv.ParseInt(chunkSizeStr, 16, 64)
if err != nil {
debuglogger.Logf("failed to parse chunk size: (size): %v, (err): %v", chunkSizeStr, err)
return 0, s3err.GetAPIError(s3err.ErrIncompleteBody)
}
if !cr.isValidChunkSize(chunkSize) {
return 0, s3err.GetAPIError(s3err.ErrInvalidChunkSize)
}
return chunkSize, nil
}
// addChunkSize adds the input chunk size to chunkSizes slice
func (cr *ChunkReader) addChunkSize(size int64) {
cr.chunkSizes = append(cr.chunkSizes, size)
}
// isValidChunkSize checks if the parsed chunk size is valid
// they follow one rule: all chunk sizes except for the last one
// should be greater than 8192
func (cr *ChunkReader) isValidChunkSize(size int64) bool {
if len(cr.chunkSizes) == 0 {
// any valid number is valid as a first chunk size
return true
}
lastChunkSize := cr.chunkSizes[len(cr.chunkSizes)-1]
// any chunk size, except the last one should be greater than 8192
if size != 0 && lastChunkSize < minChunkSize {
debuglogger.Logf("invalid chunk size %v", lastChunkSize)
return false
}
return true
}
// Algorithm returns the checksum algorithm
func (cr *ChunkReader) Algorithm() string {
return strings.TrimPrefix(string(cr.trailer), "x-amz-checksum-")
}
// Checksum returns the parsed trailing checksum
func (cr *ChunkReader) Checksum() string {
return cr.parsedChecksum
}
// reads data from the "rdr" and validates the passed data bytes
func readAndSkip(rdr *bufio.Reader, data ...byte) error {
for _, d := range data {
b, err := rdr.ReadByte()
if err != nil {
return err
}
if b != d {
return errMalformedEncoding
}
func readAndSkip(rdr *bufio.Reader, expected ...byte) error {
buf := make([]byte, len(expected))
_, err := io.ReadFull(rdr, buf)
if err != nil {
return err
}
return nil
if bytes.Equal(buf, expected) {
return nil
}
return s3err.GetAPIError(s3err.ErrIncompleteBody)
}
// reads string by "delim" and trims the delimiter at the end
@@ -530,3 +605,10 @@ func readAndTrim(r *bufio.Reader, delim byte) (string, error) {
return strings.TrimSuffix(str, string(delim)), nil
}
func readBytes(r *bufio.Reader, count int) (string, error) {
buf := make([]byte, count)
_, err := io.ReadFull(r, buf)
return string(buf), err
}
+2 -3
View File
@@ -35,9 +35,8 @@ import (
)
var (
trailerDelim = []byte{'\n', '\r', '\n'}
minChunkSize int64 = 8192
errMalformedEncoding = errors.New("malformed chunk encoding")
trailerDelim = []byte{'\n', '\r', '\n'}
minChunkSize int64 = 8192
)
type UnsignedChunkReader struct {
+30
View File
@@ -808,6 +808,8 @@ func TestFullFlow(ts *TestState) {
TestAccessControl(ts)
TestRouter(ts)
TestUnsignedStreaminPayloadTrailer(ts)
TestSignedStreaminPayload(ts)
TestSignedStreaminPayloadTrailer(ts)
// FIXME: The tests should pass for azure as well
// but this issue should be fixed with https://github.com/versity/versitygw/issues/1336
if !ts.conf.azureTests {
@@ -1114,6 +1116,25 @@ func TestUnsignedStreaminPayloadTrailer(ts *TestState) {
}
}
func TestSignedStreaminPayload(ts *TestState) {
if !ts.conf.azureTests {
ts.Run(SignedStreamingPayload_invalid_encoding)
ts.Run(SignedStreamingPayload_invalid_chunk_size)
ts.Run(SignedStreamingPayload_decoded_content_length_mismatch)
}
}
func TestSignedStreaminPayloadTrailer(ts *TestState) {
if !ts.conf.azureTests {
ts.Run(SignedStreamingPayloadTrailer_malformed_trailer)
ts.Run(SignedStreamingPayloadTrailer_incomplete_body)
ts.Run(SignedStreamingPayloadTrailer_missing_x_amz_trailer_header)
ts.Run(SignedStreamingPayloadTrailer_invalid_checksum)
ts.Run(SignedStreamingPayloadTrailer_bad_digest)
ts.Run(SignedStreamingPayloadTrailer_success)
}
}
type IntTest func(s3 *S3Conf) error
type IntTests map[string]IntTest
@@ -1767,5 +1788,14 @@ func GetIntTests() IntTests {
"UnsignedStreamingPayloadTrailer_UploadPart_trailer_and_mp_algo_mismatch": UnsignedStreamingPayloadTrailer_UploadPart_trailer_and_mp_algo_mismatch,
"UnsignedStreamingPayloadTrailer_UploadPart_success_with_trailer": UnsignedStreamingPayloadTrailer_UploadPart_success_with_trailer,
"UnsignedStreamingPayloadTrailer_not_allowed": UnsignedStreamingPayloadTrailer_not_allowed,
"SignedStreamingPayload_invalid_encoding": SignedStreamingPayload_invalid_encoding,
"SignedStreamingPayload_invalid_chunk_size": SignedStreamingPayload_invalid_chunk_size,
"SignedStreamingPayload_decoded_content_length_mismatch": SignedStreamingPayload_decoded_content_length_mismatch,
"SignedStreamingPayloadTrailer_malformed_trailer": SignedStreamingPayloadTrailer_malformed_trailer,
"SignedStreamingPayloadTrailer_incomplete_body": SignedStreamingPayloadTrailer_incomplete_body,
"SignedStreamingPayloadTrailer_missing_x_amz_trailer_header": SignedStreamingPayloadTrailer_missing_x_amz_trailer_header,
"SignedStreamingPayloadTrailer_invalid_checksum": SignedStreamingPayloadTrailer_invalid_checksum,
"SignedStreamingPayloadTrailer_bad_digest": SignedStreamingPayloadTrailer_bad_digest,
"SignedStreamingPayloadTrailer_success": SignedStreamingPayloadTrailer_success,
}
}
@@ -0,0 +1,128 @@
// 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"
"fmt"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/versity/versitygw/s3err"
)
func SignedStreamingPayload_invalid_encoding(s *S3Conf) error {
testName := "SignedStreamingPayload_invalid_encoding"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "object"
for i, test := range []struct {
from int
to int
buffer []byte
}{
{0, 2, []byte{'j'}}, // invalid chunk size
// missing/invalid delimiters
{83, 85, nil},
{83, 85, []byte("dd")},
{103, 105, nil},
{103, 105, []byte("something invalid")},
// invalid trailing delimiter
{187, 191, []byte("bbbb")},
// only last character changed
{190, 191, []byte("s")},
// invalid chunksize delimiter (;)
{2, 3, []byte(":")},
// missing chunk-signature
{3, 19, nil},
// short signature
{19, 24, nil},
} {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, object, []byte("dummy data paylaod"), withModifyPayload(test.from, test.to, test.buffer))
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if err := compareS3ApiError(s3err.GetAPIError(s3err.ErrIncompleteBody), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
return nil
})
}
func SignedStreamingPayload_invalid_chunk_size(s *S3Conf) error {
testName := "SignedStreamingPayload_invalid_chunk_size"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
chunkSize int64
payload []byte
expectErr bool
}{
{10, bytes.Repeat([]byte{'b'}, 100), true},
{1000, bytes.Repeat([]byte{'a'}, 200), false},
{8192, bytes.Repeat([]byte{'c'}, 10000), false},
{8192, bytes.Repeat([]byte{'c'}, 20000), false},
{1000, bytes.Repeat([]byte{'c'}, 1024*64), true},
} {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, object, test.payload, withChunkSize(test.chunkSize), withCustomHeaders(map[string]string{
"Content-Length": "-1",
"Transfer-Encoding": "chunked",
}))
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if !test.expectErr && apiErr != nil {
return fmt.Errorf("test %v failed: expected no error, instead got: (%s) %s", i+1, apiErr.Code, apiErr.Message)
}
if test.expectErr {
if err := compareS3ApiError(s3err.GetAPIError(s3err.ErrInvalidChunkSize), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
}
return nil
})
}
func SignedStreamingPayload_decoded_content_length_mismatch(s *S3Conf) error {
testName := "SignedStreamingPayload_decoded_content_length_mismatch"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
cLength int64
payload []byte
}{
{10, bytes.Repeat([]byte{'a'}, 8)},
{10, bytes.Repeat([]byte{'a'}, 12)},
} {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, object, test.payload, withCustomHeaders(map[string]string{
"x-amz-decoded-content-length": fmt.Sprint(test.cLength),
}))
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if err := compareS3ApiError(s3err.GetAPIError(s3err.ErrContentLengthMismatch), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
return nil
})
}
@@ -0,0 +1,207 @@
// 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 (
"fmt"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/versity/versitygw/s3err"
)
func SignedStreamingPayloadTrailer_malformed_trailer(s *S3Conf) error {
testName := "SignedStreamingPayloadTrailer_malformed_trailer"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
trailerHdr string
trailingChecksum string
}{
{"x-amz-checksum-crc64nvme", "x-amz-invalid:invalid"},
{"x-amz-checksum-crc64nvme", ""},
// x-amz-trailer and trailing checksum mismatch
{"x-amz-checksum-sha1", "x-amz-checksum-crc32:QWaN2w=="},
{"x-amz-checksum-crc32c", "x-amz-checksum-sha1:YR/1TvTYOJz5gtqVFoBJBtmTibY="},
} {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, object, []byte("dummy data"), withTrailingChecksum(test.trailingChecksum), withCustomHeaders(map[string]string{
"x-amz-trailer": test.trailerHdr,
}))
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if err := compareS3ApiError(s3err.GetAPIError(s3err.ErrMalformedTrailer), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
return nil
})
}
func SignedStreamingPayloadTrailer_incomplete_body(s *S3Conf) error {
testName := "SignedStreamingPayloadTrailer_incomplete_body"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
modifFrom int
modifTo int
modifPayload []byte
}{
{175, 176, []byte("k")},
{175, 177, []byte("cc")},
{215, 216, []byte("bcd")},
{220, 223, []byte("invalid")},
{230, 235, []byte("abcd")},
{241, 245, []byte("abcde")},
{306, 308, []byte("pp")},
{304, 308, []byte("erty")},
} {
_, apiErr, err := testSignedStreamingObjectPut(
s,
bucket,
object,
[]byte("abcdefg"),
withTrailingChecksum("x-amz-checksum-crc64nvme:SmzZ/LTp1CA="),
withCustomHeaders(map[string]string{"x-amz-trailer": "x-amz-checksum-crc64nvme"}),
withModifyPayload(test.modifFrom, test.modifTo, test.modifPayload),
)
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if err := compareS3ApiError(s3err.GetAPIError(s3err.ErrIncompleteBody), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
return nil
})
}
func SignedStreamingPayloadTrailer_missing_x_amz_trailer_header(s *S3Conf) error {
testName := "SignedStreamingPayloadTrailer_missing_x_amz_trailer_header"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, "my-object", []byte("hello"), withTrailingChecksum("x-amz-checksum-crc32:NhCmhg=="))
if err != nil {
return err
}
return compareS3ApiError(s3err.GetAPIError(s3err.ErrMalformedTrailer), apiErr)
})
}
func SignedStreamingPayloadTrailer_invalid_checksum(s *S3Conf) error {
testName := "SignedStreamingPayloadTrailer_invalid_checksum"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
trailerHdr string
trailingChecksum string
}{
{"x-amz-checksum-crc32", "x-amz-checksum-crc32:invalid"},
{"x-amz-checksum-crc32c", "x-amz-checksum-crc32c:invalid"},
{"x-amz-checksum-crc64nvme", "x-amz-checksum-crc64nvme:invalid"},
{"x-amz-checksum-sha1", "x-amz-checksum-sha1:invalid"},
{"x-amz-checksum-sha256", "x-amz-checksum-sha256:invalid"},
} {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, object, []byte("dummy data"), withTrailingChecksum(test.trailingChecksum), withCustomHeaders(map[string]string{
"x-amz-trailer": test.trailerHdr,
}))
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if err := compareS3ApiError(s3err.GetInvalidTrailingChecksumHeaderErr(test.trailerHdr), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
return nil
})
}
func SignedStreamingPayloadTrailer_bad_digest(s *S3Conf) error {
testName := "SignedStreamingPayloadTrailer_bad_digest"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
algo types.ChecksumAlgorithm
trailerHdr string
trailingChecksum string
}{
{types.ChecksumAlgorithmCrc32, "x-amz-checksum-crc32", "x-amz-checksum-crc32:NhCmhg=="},
{types.ChecksumAlgorithmCrc32c, "x-amz-checksum-crc32c", "x-amz-checksum-crc32c:+Cy97w=="},
{types.ChecksumAlgorithmCrc64nvme, "x-amz-checksum-crc64nvme", "x-amz-checksum-crc64nvme:QFRKMGE3tuw="},
{types.ChecksumAlgorithmSha1, "x-amz-checksum-sha1", "x-amz-checksum-sha1:qvTGHdzF6KLavt4PO0gs2a6pQ00="},
{types.ChecksumAlgorithmSha256, "x-amz-checksum-sha256", "x-amz-checksum-sha256:LPJNul+wow4m6DsqxbninhsWHlwfp0JecwQzYpOLmCQ="},
} {
_, apiErr, err := testSignedStreamingObjectPut(s, bucket, object, []byte("some random data"), withTrailingChecksum(test.trailingChecksum), withCustomHeaders(map[string]string{
"x-amz-trailer": test.trailerHdr,
}))
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if err := compareS3ApiError(s3err.GetChecksumBadDigestErr(test.algo), apiErr); err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
}
return nil
})
}
func SignedStreamingPayloadTrailer_success(s *S3Conf) error {
testName := "SignedStreamingPayloadTrailer_success"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
object := "my-object"
for i, test := range []struct {
checksumKey string
checksumValue string
}{
{"x-amz-checksum-crc32", "z3mWAA=="},
{"x-amz-checksum-crc32c", "rxvjPA=="},
{"x-amz-checksum-crc64nvme", "dYnI3/Fh0gM="},
{"x-amz-checksum-sha1", "8O8FwCfmd5fCbCBvH09mrKMVoHU="},
{"x-amz-checksum-sha256", "OoSow5X4zTIPl27MtdFdYT+9O3C367C75+Cb2MFtRBc="},
} {
headers, apiErr, err := testSignedStreamingObjectPut(
s,
bucket,
object,
[]byte("the object data"),
withTrailingChecksum(fmt.Sprintf("%s:%s", test.checksumKey, test.checksumValue)),
withCustomHeaders(map[string]string{
"x-amz-trailer": test.checksumKey,
}),
)
if err != nil {
return fmt.Errorf("test %v failed: %w", i+1, err)
}
if apiErr != nil {
return fmt.Errorf("test %v failed: (%s) %s", i+1, apiErr.Code, apiErr.Message)
}
if headers[test.checksumKey] != test.checksumValue {
return fmt.Errorf("test %v failed: expected %s header value to be %s, instead got %s", i+1, test.checksumKey, test.checksumValue, headers[test.checksumKey])
}
}
return nil
})
}
+369
View File
@@ -17,6 +17,7 @@ package integration
import (
"bytes"
"context"
"crypto/hmac"
"crypto/md5"
"crypto/rand"
"crypto/sha1"
@@ -2118,3 +2119,371 @@ func constructUnsignedPaylod(chunkSizes ...int64) (int64, []byte, error) {
return cLength, buffer.Bytes(), nil
}
type signedReqCfg struct {
headers map[string]string
chunkSize int64
modifFrom *int
modifTo *int
modifPayload []byte
trailingChecksum *string
isTrailer bool
}
type signedReqOpt func(*signedReqCfg)
func withCustomHeaders(h map[string]string) signedReqOpt {
return func(src *signedReqCfg) { src.headers = h }
}
func withChunkSize(s int64) signedReqOpt {
return func(src *signedReqCfg) { src.chunkSize = s }
}
func withModifyPayload(from int, to int, p []byte) signedReqOpt {
return func(src *signedReqCfg) {
src.modifPayload = p
src.modifFrom = &from
src.modifTo = &to
}
}
func withTrailingChecksum(checksum string) signedReqOpt {
return func(src *signedReqCfg) {
src.trailingChecksum = &checksum
src.isTrailer = true
}
}
func testSignedStreamingObjectPut(s *S3Conf, bucket, object string, payload []byte, opts ...signedReqOpt) (map[string]string, *s3err.APIErrorResponse, error) {
cfg := &signedReqCfg{
chunkSize: 8192, // minimal valid chunk size
}
for _, opt := range opts {
opt(cfg)
}
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
// create a request with no body
req, err := http.NewRequestWithContext(ctx, http.MethodPut, fmt.Sprintf("%s/%s/%s", s.endpoint, bucket, object), nil)
if err != nil {
return nil, nil, cancelAndError(fmt.Errorf("failed to create a request: %w", err), cancel)
}
var payloadOffset int64
var trailerLength int
// any planned modification which is going to affect the
// Content-Length header value
if cfg.modifFrom != nil && cfg.modifTo != nil {
diff := len(cfg.modifPayload) - *cfg.modifTo + *cfg.modifFrom
payloadOffset = int64(diff)
}
if cfg.isTrailer {
trailerLength = len(*cfg.trailingChecksum)
}
// precalculated the Content-Length header to correctly sign the request
req.ContentLength = calculateSignedReqContentLength(int64(len(payload)), cfg.chunkSize, payloadOffset, cfg.isTrailer, int64(trailerLength))
sha256Header := "STREAMING-AWS4-HMAC-SHA256-PAYLOAD"
if cfg.isTrailer {
sha256Header = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER"
}
req.Header.Set("x-amz-decoded-content-length", fmt.Sprint(len(payload)))
req.Header.Set("x-amz-content-sha256", sha256Header)
// set custom request headers
for key, val := range cfg.headers {
req.Header.Set(key, val)
}
signer := v4.NewSigner()
signingTime := time.Now()
// sign the request
err = signer.SignHTTP(ctx, aws.Credentials{AccessKeyID: s.awsID, SecretAccessKey: s.awsSecret}, req, sha256Header, "s3", s.awsRegion, signingTime)
if err != nil {
return nil, nil, cancelAndError(fmt.Errorf("failed to sign the request: %w", err), cancel)
}
// extract the seed signature
seedSignature, err := extractSignature(req)
if err != nil {
return nil, nil, cancelAndError(fmt.Errorf("failed to extract seed signature: %w", err), cancel)
}
// initialize v4 stream signed
streamSigner := v4.NewStreamSigner(aws.Credentials{AccessKeyID: s.awsID, SecretAccessKey: s.awsSecret}, "s3", s.awsRegion, seedSignature)
// create the signed payload
body, err := constructSignedStreamingPayload(ctx, streamSigner, signingTime, payload, cfg.chunkSize, cfg.trailingChecksum, s.awsRegion, s.awsSecret)
if err != nil {
return nil, nil, cancelAndError(fmt.Errorf("failed to encode req body: %w", err), cancel)
}
// overwrite body bytes by configuration
if cfg.modifFrom != nil && cfg.modifTo != nil {
body, err = replaceRange(body, cfg.modifPayload, *cfg.modifFrom, *cfg.modifTo)
if err != nil {
return nil, nil, cancelAndError(fmt.Errorf("failed replace body bytes: %w", err), cancel)
}
}
// assign req.Body and req.GetBody for the http client
// to handle the request
req.Body = io.NopCloser(bytes.NewReader(body))
req.GetBody = func() (io.ReadCloser, error) {
return io.NopCloser(bytes.NewReader(body)), nil
}
// send the request
resp, err := s.httpClient.Do(req)
cancel()
if err != nil {
return nil, nil, fmt.Errorf("failed to send the request: %w", err)
}
if resp.StatusCode >= 300 {
defer resp.Body.Close()
bodyBytes, err := io.ReadAll(resp.Body)
if err != nil {
return nil, nil, fmt.Errorf("failed to read the response body: %w", err)
}
var errResp s3err.APIErrorResponse
err = xml.Unmarshal(bodyBytes, &errResp)
if err != nil {
return nil, nil, fmt.Errorf("failed to unmarshal response body: %w", err)
}
return nil, &errResp, nil
}
headers := map[string]string{}
for key, val := range resp.Header {
headers[strings.ToLower(key)] = val[0]
}
return headers, nil, nil
}
func cancelAndError(err error, cancel context.CancelFunc) error {
cancel()
return err
}
const (
chunkSigHdrLength int64 = 81
trailerSigLength int64 = 88
)
// calculateSignedReqContentLength calculates the value of `Content-Length` header
// sizeOffset marks any planned changes on the body, which will affect the size
func calculateSignedReqContentLength(decPayloadSize int64, chunkSize int64, sizeOffset int64, withTrailer bool, trailerLength int64) int64 {
payloadSize := decPayloadSize
var chunkHeadersLength int64
if withTrailer {
chunkHeadersLength += trailerLength + 4 + trailerSigLength
}
// special case when chunk size is greater or equal than decoded content length
if chunkSize >= decPayloadSize {
chSizeLgth := len(fmt.Sprintf("%x", decPayloadSize))
return decPayloadSize + sizeOffset + int64(chSizeLgth) + 2*chunkSigHdrLength + 9 + chunkHeadersLength
}
for {
if payloadSize == 0 {
chunkHeadersLength += chunkSigHdrLength + 5
break
}
if payloadSize < chunkSize {
chunkHeadersLength += 2*chunkSigHdrLength + 9 + int64(len(fmt.Sprintf("%x", payloadSize)))
break
}
chSizeLgth := len(fmt.Sprintf("%x", chunkSize))
chunkHeadersLength += int64(chSizeLgth) + chunkSigHdrLength + 4
payloadSize -= chunkSize
}
return chunkHeadersLength + decPayloadSize + sizeOffset
}
// constructSignedStreamingPayload creates chunk encoded payload with signatures.
func constructSignedStreamingPayload(ctx context.Context, signer *v4.StreamSigner, signingTime time.Time, payload []byte, chunkSize int64, trailer *string, region, secret string) ([]byte, error) {
buf := bytes.NewBuffer(nil)
payloadLen := int64(len(payload))
if chunkSize > payloadLen {
chunkSize = payloadLen
}
for i := int64(0); i < payloadLen; i += chunkSize {
if i+chunkSize > payloadLen {
offset := payloadLen - i
sig, err := signer.GetSignature(ctx, nil, payload[i:i+offset], signingTime)
if err != nil {
return nil, err
}
_, err = buf.WriteString(fmt.Sprintf("%x;chunk-signature=%x\r\n%s\r\n", offset, sig, payload[i:i+offset]))
if err != nil {
return nil, err
}
break
}
sig, err := signer.GetSignature(ctx, nil, payload[i:i+chunkSize], signingTime)
if err != nil {
return nil, err
}
_, err = buf.WriteString(fmt.Sprintf("%x;chunk-signature=%x\r\n%s\r\n", chunkSize, sig, payload[i:i+chunkSize]))
if err != nil {
return nil, err
}
}
sig, err := signer.GetSignature(ctx, nil, nil, signingTime)
if err != nil {
return nil, err
}
if trailer != nil {
_, err = buf.WriteString(fmt.Sprintf("0;chunk-signature=%x\r\n", sig))
if err != nil {
return nil, err
}
sigKey := getSigningKey(secret, signingTime.Format("20060102"), region)
trailerSig, err := getAWS4StreamingTrailer(sigKey, sig, signingTime, region, *trailer)
if err != nil {
return nil, err
}
_, err = buf.WriteString(fmt.Sprintf("%s\r\nx-amz-trailer-signature:%s\r\n\r\n", *trailer, trailerSig))
if err != nil {
return nil, err
}
return buf.Bytes(), nil
}
_, err = buf.WriteString(fmt.Sprintf("0;chunk-signature=%x\r\n\r\n", sig))
if err != nil {
return nil, err
}
return buf.Bytes(), nil
}
// extractSignature extracts the signature from Authorization header
func extractSignature(req *http.Request) ([]byte, error) {
const key = "Signature="
authHdr := req.Header.Get("Authorization")
i := strings.Index(authHdr, key)
if i == -1 {
return nil, errors.New("signature not found")
}
sig := authHdr[i+len(key):]
return hex.DecodeString(sig)
}
// replaceRange replaces dst[start:end] with src and returns the modified slice.
// Used for custom overwrite of request payload bytes.
func replaceRange(dst, src []byte, start, end int) ([]byte, error) {
if start < 0 || end < start || end > len(dst) {
return nil, fmt.Errorf("invalid start/end indexes")
}
newLen := len(dst) - (end - start) + len(src)
// Fast path: reuse dst capacity if possible
if cap(dst) >= newLen {
// Extend or shrink dst
dst = dst[:newLen]
// Move the tail if sizes differ
copy(dst[start+len(src):], dst[end:])
// Copy replacement
copy(dst[start:], src)
return dst, nil
}
// Fallback: allocate new slice
out := make([]byte, newLen)
copy(out, dst[:start])
copy(out[start:], src)
copy(out[start+len(src):], dst[end:])
return out, nil
}
func getAWS4StreamingTrailer(
signingKey,
lastSignature []byte,
signingTime time.Time,
awsRegion,
trailer string,
) (string, error) {
// yyyyMMdd
yearMonthDay := signingTime.Format("20060102")
// ISO8601 basic format: yyyyMMdd'T'HHmmss'Z'
currentDateTime := signingTime.UTC().Format("20060102T150405Z")
// <date>/<region>/<service>/aws4_request
serviceString := fmt.Sprintf(
"%s/%s/s3/aws4_request",
yearMonthDay,
awsRegion,
)
// Trailer must be newline-terminated for hashing/signing
trailerWithNL := trailer + "\n"
// Hash of trailer
trailerHash := sha256.Sum256([]byte(trailerWithNL))
trailerHashHex := hex.EncodeToString(trailerHash[:])
// String-to-sign prefix
stringToSignPrefix := fmt.Sprintf(
"%s\n%s\n%s",
"AWS4-HMAC-SHA256-TRAILER",
currentDateTime,
serviceString,
)
// Full string-to-sign
stringToSign := fmt.Sprintf(
"%s\n%x\n%s",
stringToSignPrefix,
lastSignature,
trailerHashHex,
)
// Final trailer signature
finalSignature := hex.EncodeToString(
hmacSHA256(signingKey, stringToSign),
)
return finalSignature, nil
}
func hmacSHA256(key []byte, data string) []byte {
h := hmac.New(sha256.New, key)
h.Write([]byte(data))
return h.Sum(nil)
}
func getSigningKey(secret, yearMonthDay, region string) []byte {
dateKey := hmacSHA256([]byte("AWS4"+secret), yearMonthDay)
dateRegionKey := hmacSHA256(dateKey, region)
dateRegionServiceKey := hmacSHA256(dateRegionKey, "s3")
return hmacSHA256(dateRegionServiceKey, "aws4_request")
}