From eefe63496238798e558042a9d029b83f67ef1984 Mon Sep 17 00:00:00 2001 From: Cole Helbling Date: Sun, 5 Jul 2026 20:19:20 -0700 Subject: [PATCH] s3api: FULL_OBJECT checksums for CRC multipart uploads (#10236) CRC64NVME multipart objects now emit a full-object checksum instead of a composite base64-N value, matching AWS (CRC64NVME supports full-object only). Adds x-amz-checksum-type handling (COMPOSITE/FULL_OBJECT) for CRC32/CRC32C/CRC64NVME via CRC combination, resolved at CreateMultipartUpload and applied at completion. --- weed/s3api/crc_combine.go | 226 ++++++++++++++++++ weed/s3api/crc_combine_test.go | 124 ++++++++++ weed/s3api/filer_multipart.go | 215 ++++++++++++++--- weed/s3api/filer_multipart_checksum_test.go | 146 +++++++++++ weed/s3api/s3_constants/extend_key.go | 1 + weed/s3api/s3_constants/header.go | 7 + weed/s3api/s3api_object_handlers.go | 3 + weed/s3api/s3api_object_handlers_multipart.go | 5 +- 8 files changed, 690 insertions(+), 37 deletions(-) create mode 100644 weed/s3api/crc_combine.go create mode 100644 weed/s3api/crc_combine_test.go create mode 100644 weed/s3api/filer_multipart_checksum_test.go diff --git a/weed/s3api/crc_combine.go b/weed/s3api/crc_combine.go new file mode 100644 index 000000000..582d12f77 --- /dev/null +++ b/weed/s3api/crc_combine.go @@ -0,0 +1,226 @@ +package s3api + +import "math" + +// crcParams describes a CRC algorithm in the Rocksoft/Williams parameterization (see "15. A +// Parameterized Model For CRC Algorithms" in http://www.ross.net/crc/download/crc_v3.txt) +type crcParams struct { + width uint32 // the width of the algorithm expressed in bits; 1 less than the width of poly + poly uint64 // the unreflected poly + init uint64 // initial register value + xorout uint64 // value XORed into the final register + refin bool // reflect input bytes + refout bool // reflect final register +} + +var crcCombineParams = map[ChecksumAlgorithm]crcParams{ + ChecksumAlgorithmCRC64NVMe: { + width: 64, + poly: 0xad93d23594c93659, + init: 0xffffffffffffffff, + xorout: 0xffffffffffffffff, + refin: true, + refout: true, + }, + ChecksumAlgorithmCRC32: { + width: 32, + poly: 0x04c11db7, + init: 0xffffffff, + xorout: 0xffffffff, + refin: true, + refout: true, + }, + ChecksumAlgorithmCRC32C: { + width: 32, + poly: 0x1edc6f41, + init: 0xffffffff, + xorout: 0xffffffff, + refin: true, + refout: true, + }, +} + +// Ported from: https://github.com/awesomized/crc-fast-rust/blob/3a853cc7daf2cd47cc4466f198680cabdfb0b5fa/src/combine.rs + +/* + Derived from this excellent answer by Mark Adler on StackOverflow: + https://stackoverflow.com/questions/29915764/generic-crc-8-16-32-64-combine-implementation/29928573#29928573 +*/ + +/* crccomb.c -- generalized combination of CRCs + * Copyright (C) 2015 Mark Adler + * Version 1.1 29 Apr 2015 Mark Adler + */ + +/* + This software is provided 'as-is', without any express or implied + warranty. In no event will the author be held liable for any damages + arising from the use of this software. + + Permission is granted to anyone to use this software for any purpose, + including commercial applications, and to alter it and redistribute it + freely, subject to the following restrictions: + + 1. The origin of this software must not be misrepresented; you must not + claim that you wrote the original software. If you use this software + in a product, an acknowledgment in the product documentation would be + appreciated but is not required. + 2. Altered source versions must be plainly marked as such, and must not be + misrepresented as being the original software. + 3. This notice may not be removed or altered from any source distribution. + + Mark Adler + madler@alumni.caltech.edu +*/ + +/* + zlib provides a fast operation to combine the CRCs of two sequences of bytes + into a single CRC, which is the CRC of the two sequences concatenated. That + operation requires only the two CRC's and the length of the second sequence. + The routine in zlib only works on the particular CRC-32 used by zlib. The + code provided here generalizes that operation to apply to a wide range of + CRCs. The CRC is specified in a series of #defines, based on the + parameterization found in Ross William's excellent CRC tutorial here: + + http://www.ross.net/crc/download/crc_v3.txt + + A comprehensive catalogue of known CRCs, their parameters, check values, and + references can be found here: + + http://reveng.sourceforge.net/crc-catalogue/all.htm +*/ + +// Multiply the GF(2) vector vec by the GF(2) matrix mat, returning the +// resulting vector. The vector is stored as bits in a crc_t. The matrix is +// similarly stored with each column as a crc_t, where the number of columns is +// at least enough to cover the position of the most significant 1 bit in the +// vector (so a dimension parameter is not needed). +func gf2MatrixTimes(mat *[64]uint64, vec uint64) uint64 { + var sum uint64 + idx := 0 + + for vec > 0 { + if vec&1 == 1 { + sum ^= mat[idx] + } + vec >>= 1 + idx++ + } + + return sum +} + +// Multiply the matrix mat by itself, returning the result in square. WIDTH is +// the dimension of the matrices, i.e., the number of bits in each crc_t +// (rows), and the number of crc_t's (columns). +func gf2MatrixSquare(square *[64]uint64, mat *[64]uint64) { + for n := 0; n < 64; n++ { + square[n] = gf2MatrixTimes(mat, mat[n]) + } +} + +// Combine the CRCs of two successive sequences, where crc1 is the CRC of the +// first sequence of bytes, crc2 is the CRC of the immediately following +// sequence of bytes, and len2 is the length of the second sequence. The CRC +// of the combined sequence is returned. +func combineCRC(crc1 uint64, crc2 uint64, len2 uint64, params crcParams) uint64 { + even := [64]uint64{} /* even-power-of-two zeros operator */ + odd := [64]uint64{} /* odd-power-of-two zeros operator */ + + /* exclusive-or the result with len2 zeros applied to the CRC of an empty + sequence */ + crc1 ^= params.init ^ params.xorout + + /* construct the operator for one zero bit and put in odd[] */ + if params.refin && params.refout { + // use the reflected POLY + odd[0] = reflectPoly(params.poly, params.width) + col := uint64(1) + for n := uint32(1); n < params.width; n++ { + odd[n] = col + col <<= 1 + } + } else if !params.refin && !params.refout { + col := uint64(2) + for n := uint32(0); n < params.width-1; n++ { + odd[n] = col + col <<= 1 + } + odd[params.width-1] = params.poly + } else { + panic("Unsupported CRC configuration") + } + + /* put operator for two zero bits in even */ + gf2MatrixSquare(&even, &odd) + + /* put operator for four zero bits in odd */ + gf2MatrixSquare(&odd, &even) + + /* apply len2 zeros to crc1 (first square will put the operator for one + zero byte, eight zero bits, in even) */ + for { + /* apply zeros operator for this bit of len2 */ + gf2MatrixSquare(&even, &odd) + if len2&1 == 1 { + crc1 = gf2MatrixTimes(&even, crc1) + } + len2 >>= 1 + + /* if no more bits set, then done */ + if len2 == 0 { + break + } + + /* another iteration of the loop with odd and even swapped */ + gf2MatrixSquare(&odd, &even) + if len2&1 == 1 { + crc1 = gf2MatrixTimes(&odd, crc1) + } + len2 >>= 1 + + /* if no more bits set, then done */ + if len2 == 0 { + break + } + } + + /* return combined crc */ + return crc1 ^ crc2 +} + +func reflectPoly(poly uint64, width uint32) uint64 { + if width > 64 { + panic("Width must be <= 64 bits") + } + + // First reverse all bits + reversed := bitReverse(poly) + + // Shift right to get the significant bits in the correct position + // For a 32-bit poly, we need to shift right by (64 - 32) = 32 bits + shifted := reversed >> (64 - width) + + // Create mask for the target width + var mask uint64 + if width == 64 { + mask = math.MaxUint64 + } else { + mask = (1 << width) - 1 + } + + // Apply mask to ensure we only keep the bits we want + return shifted & mask +} + +func bitReverse(forward uint64) uint64 { + reversed := uint64(0) + + for i := 0; i < 64; i++ { + reversed <<= 1 + reversed |= forward & 1 + forward >>= 1 + } + + return reversed +} diff --git a/weed/s3api/crc_combine_test.go b/weed/s3api/crc_combine_test.go new file mode 100644 index 000000000..ff9850098 --- /dev/null +++ b/weed/s3api/crc_combine_test.go @@ -0,0 +1,124 @@ +package s3api + +import ( + "bytes" + "hash/crc32" + "testing" + + "github.com/minio/crc64nvme" +) + +// Tests that combining per-block CRC-64/NVME values reproduces the CRC of the concatenated data. +func TestCombineCRC64NVME(t *testing.T) { + params := crcCombineParams[ChecksumAlgorithmCRC64NVMe] + + blocks := [][]byte{ + []byte("abc"), + []byte("def"), + []byte("hello world, this is a longer block of data to combine"), + bytes.Repeat([]byte{0x00}, 5*1024*1024), + bytes.Repeat([]byte{0xAB, 0xCD}, 1024), + []byte("z"), + } + + var combined uint64 + var concatenated []byte + for i, block := range blocks { + blockCRC := crc64nvme.Checksum(block) + + if i == 0 { + combined = blockCRC + } else { + combined = combineCRC(combined, blockCRC, uint64(len(block)), params) + } + concatenated = append(concatenated, block...) + + want := crc64nvme.Checksum(concatenated) + if combined != want { + t.Fatalf("after block %d: combined = %016x, want %016x", i, combined, want) + } + } +} + +// Tests that combining per-block CRC-32 values reproduces the CRC of the concatenated data. +func TestCombineCRC32(t *testing.T) { + params := crcCombineParams[ChecksumAlgorithmCRC32] + + blocks := [][]byte{ + []byte("abc"), + []byte("def"), + []byte("hello world, this is a longer block of data to combine"), + bytes.Repeat([]byte{0x00}, 5*1024*1024), + bytes.Repeat([]byte{0xAB, 0xCD}, 1024), + []byte("z"), + } + + var combined uint64 + var concatenated []byte + for i, block := range blocks { + blockCRC := uint64(crc32.Checksum(block, crc32.MakeTable(crc32.IEEE))) + + if i == 0 { + combined = blockCRC + } else { + combined = combineCRC(combined, blockCRC, uint64(len(block)), params) + } + concatenated = append(concatenated, block...) + + want := uint64(crc32.Checksum(concatenated, crc32.MakeTable(crc32.IEEE))) + if combined != want { + t.Fatalf("after block %d: combined = %016x, want %016x", i, combined, want) + } + } +} + +// Tests that combining per-block CRC-32C values reproduces the CRC of the concatenated data. +func TestCombineCRC32C(t *testing.T) { + params := crcCombineParams[ChecksumAlgorithmCRC32C] + table := crc32.MakeTable(crc32.Castagnoli) + + blocks := [][]byte{ + []byte("abc"), + []byte("def"), + []byte("hello world, this is a longer block of data to combine"), + bytes.Repeat([]byte{0x00}, 5*1024*1024), + bytes.Repeat([]byte{0xAB, 0xCD}, 1024), + []byte("z"), + } + + var combined uint64 + var concatenated []byte + for i, block := range blocks { + blockCRC := uint64(crc32.Checksum(block, table)) + + if i == 0 { + combined = blockCRC + } else { + combined = combineCRC(combined, blockCRC, uint64(len(block)), params) + } + concatenated = append(concatenated, block...) + + want := uint64(crc32.Checksum(concatenated, table)) + if combined != want { + t.Fatalf("after block %d: combined = %016x, want %016x", i, combined, want) + } + } +} + +func TestCombineCRC64NVMEEmptySecond(t *testing.T) { + params := crcCombineParams[ChecksumAlgorithmCRC64NVMe] + crc1 := crc64nvme.Checksum([]byte("abc")) + + if got := combineCRC(crc1, crc64nvme.Checksum(nil), 0, params); got != crc1 { + t.Fatalf("combining with empty second block changed crc: got %016x, want %016x", got, crc1) + } +} + +func TestCombineCRC32EmptySecond(t *testing.T) { + params := crcCombineParams[ChecksumAlgorithmCRC32] + crc1 := uint64(crc32.Checksum([]byte("abc"), crc32.MakeTable(crc32.IEEE))) + + if got := combineCRC(crc1, uint64(crc32.Checksum(nil, crc32.MakeTable(crc32.IEEE))), 0, params); got != crc1 { + t.Fatalf("combining with empty second block changed crc: got %016x, want %016x", got, crc1) + } +} diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 4bf4f39dc..b02fcf501 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -20,6 +20,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/s3" @@ -66,11 +67,24 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM uploadIdString = uploadIdString + "_" + strings.ReplaceAll(uuid.New().String(), "-", "") // Validate checksum algorithm before creating the upload directory - _, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r) + checksumAlgo, checksumHeaderName, checksumErrCode := detectRequestedChecksumAlgorithm(r) if checksumErrCode != s3err.ErrNone { return nil, checksumErrCode } + // Resolve and validate the requested checksum type (x-amz-checksum-type) + // against the algorithm so CompleteMultipartUpload knows whether to produce a + // COMPOSITE or FULL_OBJECT checksum. + checksumType := "" + if checksumHeaderName != "" { + resolvedType, typeErr := resolveMultipartChecksumType(checksumAlgo, r.Header.Get(s3_constants.AmzChecksumType)) + if typeErr != nil { + glog.Warningf("createMultipartUpload: %v", typeErr) + return nil, s3err.ErrInvalidRequest + } + checksumType = resolvedType + } + // Prepare error handling outside callback scope var encryptionError error @@ -101,10 +115,13 @@ func (s3a *S3ApiServer) createMultipartUpload(r *http.Request, input *s3.CreateM } s3a.applyMultipartEncryptionConfig(entry, encryptionConfig) - // Store the requested checksum algorithm so CompleteMultipartUpload can compute - // a composite checksum from per-part checksums + // Store the requested checksum algorithm and type so CompleteMultipartUpload + // can compute the object checksum from per-part checksums if checksumHeaderName != "" { entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(checksumHeaderName) + if checksumType != "" { + entry.Extended[s3_constants.ExtChecksumType] = []byte(checksumType) + } } // Extract and store object lock metadata from request headers @@ -145,6 +162,7 @@ type CompleteMultipartUploadResult struct { // Checksum fields — returned as HTTP response headers, not in the XML body ChecksumHeaderName string `xml:"-"` ChecksumValue string `xml:"-"` + ChecksumType string `xml:"-"` // VersionId is NOT included in XML body - it should only be in x-amz-version-id HTTP header @@ -207,7 +225,8 @@ type multipartCompletionState struct { multipartETag string entityWithTtl bool checksumHeaderName string // e.g. "X-Amz-Checksum-Crc32", empty if no checksum - checksumValue string // composite base64 checksum with "-N" suffix + checksumValue string // multipart checksum: "base64-N" (COMPOSITE) or "base64" (FULL_OBJECT) + checksumType string // s3_constants.ChecksumType* (COMPOSITE or FULL_OBJECT) } func completeMultipartResult(r *http.Request, input *s3.CompleteMultipartUploadInput, etag string, entry *filer_pb.Entry) *CompleteMultipartUploadResult { @@ -498,9 +517,12 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * } } - // Compute composite checksum from per-part checksums if the upload - // was initiated with a checksum algorithm (stored in upload dir entry) + // Compute the object checksum from per-part checksums if the upload was + // initiated with a checksum algorithm (stored in the upload dir entry). + // The checksum type (COMPOSITE or FULL_OBJECT) is resolved from the + // x-amz-checksum-type header captured at CreateMultipartUpload checksumHeaderName := "" + checksumType := "" checksumValue := "" if pentry.Extended != nil { if algoName, ok := pentry.Extended[s3_constants.ExtChecksumAlgorithm]; ok { @@ -508,10 +530,26 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * } } if checksumHeaderName != "" { + algo := checksumAlgorithmFromHeaderName(checksumHeaderName) + requestedType := "" + if pentry.Extended != nil { + requestedType = string(pentry.Extended[s3_constants.ExtChecksumType]) + } + resolvedType, typeErr := resolveMultipartChecksumType(algo, requestedType) + if typeErr != nil { + glog.Errorf("completeMultipartUpload: %v", typeErr) + return nil, nil, s3err.ErrInvalidRequest + } + checksumType = resolvedType + var checksumErr error - checksumValue, checksumErr = computeCompositeChecksum(checksumHeaderName, partEntries, completedPartNumbers) + if checksumType == s3_constants.ChecksumTypeFullObject { + checksumValue, checksumErr = computeFullObjectChecksum(checksumHeaderName, partEntries, completedPartNumbers) + } else { + checksumValue, checksumErr = computeCompositeChecksum(checksumHeaderName, partEntries, completedPartNumbers) + } if checksumErr != nil { - glog.Errorf("completeMultipartUpload: composite checksum computation failed: %v", checksumErr) + glog.Errorf("completeMultipartUpload: %s checksum computation failed: %v", checksumType, checksumErr) return nil, nil, s3err.ErrInvalidPart } } @@ -529,6 +567,7 @@ func (s3a *S3ApiServer) prepareMultipartCompletionState(r *http.Request, input * entityWithTtl: entityWithTtl, checksumHeaderName: checksumHeaderName, checksumValue: checksumValue, + checksumType: checksumType, }, nil, s3err.ErrNone } @@ -620,10 +659,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl // Persist ETag to ensure subsequent HEAD/GET uses the same value versionEntry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag) - // Store composite checksum if computed from per-part checksums + // Store the object checksum computed from per-part checksums if completionState.checksumHeaderName != "" && completionState.checksumValue != "" { versionEntry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName) versionEntry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue) + if completionState.checksumType != "" { + versionEntry.Extended[s3_constants.ExtChecksumType] = []byte(completionState.checksumType) + } } // Preserve ALL SSE metadata from the first part (if any) @@ -690,6 +732,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl VersionId: aws.String(versionId), ChecksumHeaderName: completionState.checksumHeaderName, ChecksumValue: completionState.checksumValue, + ChecksumType: completionState.checksumType, } return s3err.ErrNone } @@ -730,10 +773,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl applyMultipartSSES3HeadersFromUploadEntry(entry, completionState.sses3Info) // Persist ETag to ensure subsequent HEAD/GET uses the same value entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag) - // Store composite checksum if computed from per-part checksums + // Store the object checksum computed from per-part checksums if completionState.checksumHeaderName != "" && completionState.checksumValue != "" { entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName) entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue) + if completionState.checksumType != "" { + entry.Extended[s3_constants.ExtChecksumType] = []byte(completionState.checksumType) + } } if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" { entry.Attributes.Mime = completionState.pentry.Attributes.Mime @@ -754,6 +800,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl Key: objectKey(input.Key), ChecksumHeaderName: completionState.checksumHeaderName, ChecksumValue: completionState.checksumValue, + ChecksumType: completionState.checksumType, // VersionId field intentionally omitted for suspended versioning } return s3err.ErrNone @@ -793,10 +840,13 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl applyMultipartSSES3HeadersFromUploadEntry(entry, completionState.sses3Info) // Persist ETag to ensure subsequent HEAD/GET uses the same value entry.Extended[s3_constants.ExtETagKey] = []byte(completionState.multipartETag) - // Store composite checksum if computed from per-part checksums + // Store the object checksum computed from per-part checksums if completionState.checksumHeaderName != "" && completionState.checksumValue != "" { entry.Extended[s3_constants.ExtChecksumAlgorithm] = []byte(completionState.checksumHeaderName) entry.Extended[s3_constants.ExtChecksumValue] = []byte(completionState.checksumValue) + if completionState.checksumType != "" { + entry.Extended[s3_constants.ExtChecksumType] = []byte(completionState.checksumType) + } } if completionState.pentry.Attributes != nil && completionState.pentry.Attributes.Mime != "" { entry.Attributes.Mime = completionState.pentry.Attributes.Mime @@ -821,6 +871,7 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl Key: objectKey(input.Key), ChecksumHeaderName: completionState.checksumHeaderName, ChecksumValue: completionState.checksumValue, + ChecksumType: completionState.checksumType, } return s3err.ErrNone } @@ -1267,6 +1318,36 @@ func calculateMultipartETag(partEntries map[int][]*filer_pb.Entry, completedPart return fmt.Sprintf("%x-%d", md5.Sum(etags), len(completedPartNumbers)) } +func decodePartChecksum(partNumber int, entries []*filer_pb.Entry, checksumHeaderName string) ([]byte, *filer_pb.Entry, error) { + if len(entries) == 0 { + return nil, nil, fmt.Errorf("part %d not found", partNumber) + } + if len(entries) > 1 { + sortEntriesByLatestChunk(entries) + } + entry := entries[0] + if entry.Extended == nil { + return nil, nil, fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName) + } + // Validate the part's checksum algorithm matches the upload's expected algorithm + partAlgo, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm] + if !ok || len(partAlgo) == 0 { + return nil, nil, fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName) + } + if string(partAlgo) != checksumHeaderName { + return nil, nil, fmt.Errorf("part %d checksum algorithm mismatch: upload expects %s but part has %s", partNumber, checksumHeaderName, string(partAlgo)) + } + partChecksumB64, ok := entry.Extended[s3_constants.ExtChecksumValue] + if !ok || len(partChecksumB64) == 0 { + return nil, nil, fmt.Errorf("part %d missing checksum value: upload initiated with %s but part has no checksum value", partNumber, checksumHeaderName) + } + raw, err := base64.StdEncoding.DecodeString(string(partChecksumB64)) + if err != nil { + return nil, nil, fmt.Errorf("part %d has invalid checksum encoding: %w", partNumber, err) + } + return raw, entry, nil +} + // computeCompositeChecksum computes a composite checksum from per-part checksums. // It concatenates the raw (decoded) per-part checksums, hashes the result with the // same algorithm, and returns the value as "base64-N" where N is the part count. @@ -1283,32 +1364,9 @@ func computeCompositeChecksum(checksumHeaderName string, partEntries map[int][]* // Collect raw per-part checksums var combined []byte for _, partNumber := range completedPartNumbers { - entries, ok := partEntries[partNumber] - if !ok || len(entries) == 0 { - return "", fmt.Errorf("part %d not found", partNumber) - } - if len(entries) > 1 { - sortEntriesByLatestChunk(entries) - } - entry := entries[0] - if entry.Extended == nil { - return "", fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName) - } - // Validate the part's checksum algorithm matches the upload's expected algorithm - partAlgo, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm] - if !ok || len(partAlgo) == 0 { - return "", fmt.Errorf("part %d missing checksum: upload initiated with %s but part was uploaded without a checksum", partNumber, checksumHeaderName) - } - if string(partAlgo) != checksumHeaderName { - return "", fmt.Errorf("part %d checksum algorithm mismatch: upload expects %s but part has %s", partNumber, checksumHeaderName, string(partAlgo)) - } - partChecksumB64, ok := entry.Extended[s3_constants.ExtChecksumValue] - if !ok || len(partChecksumB64) == 0 { - return "", fmt.Errorf("part %d missing checksum value: upload initiated with %s but part has no checksum value", partNumber, checksumHeaderName) - } - raw, err := base64.StdEncoding.DecodeString(string(partChecksumB64)) + raw, _, err := decodePartChecksum(partNumber, partEntries[partNumber], checksumHeaderName) if err != nil { - return "", fmt.Errorf("part %d has invalid checksum encoding: %w", partNumber, err) + return "", err } combined = append(combined, raw...) } @@ -1323,6 +1381,51 @@ func computeCompositeChecksum(checksumHeaderName string, partEntries map[int][]* return fmt.Sprintf("%s-%d", base64.StdEncoding.EncodeToString(compositeRaw), len(completedPartNumbers)), nil } +// computeFullObjectChecksum computes a FULL_OBJECT checksum from per-part checksums +// by combining the per-part CRCs into the CRC of the whole object. The result is the +// base64-encoded whole-object checksum with NO "-N" suffix (a full-object checksum +// is indistinguishable from a single-part upload's checksum). This is required for +// CRC64NVME, which AWS only supports as a full-object checksum. +func computeFullObjectChecksum(checksumHeaderName string, partEntries map[int][]*filer_pb.Entry, completedPartNumbers []int) (string, error) { + algo := checksumAlgorithmFromHeaderName(checksumHeaderName) + params, ok := crcCombineParams[algo] + if !ok { + return "", fmt.Errorf("full object checksum not supported for %s", checksumHeaderName) + } + + checksumBytes := int(params.width / 8) + + var combined uint64 + for i, partNumber := range completedPartNumbers { + raw, entry, err := decodePartChecksum(partNumber, partEntries[partNumber], checksumHeaderName) + if err != nil { + return "", err + } + if len(raw) != checksumBytes { + return "", fmt.Errorf("part %d checksum has unexpected length %d for %s", partNumber, len(raw), checksumHeaderName) + } + + crc := util.BytesToUint64(raw) + + if i == 0 { + combined = crc + } else { + partLen := filer.FileSize(entry) + combined = combineCRC(combined, crc, partLen, params) + } + } + + // Write the low checksumBytes bytes of the combined CRC big-endian. We can't + // use util.Uint64toBytes here because it unconditionally writes 8 bytes, which + // would panic for narrower CRCs (e.g. 4-byte CRC32/CRC32C). + out := make([]byte, checksumBytes) + for i := 0; i < checksumBytes; i++ { + out[checksumBytes-1-i] = byte(combined >> (i * 8)) + } + + return base64.StdEncoding.EncodeToString(out), nil +} + // checksumAlgorithmFromHeaderName maps a canonical header name back to its algorithm. func checksumAlgorithmFromHeaderName(headerName string) ChecksumAlgorithm { for _, entry := range checksumHeaders { @@ -1378,3 +1481,43 @@ func validateCompletePartETag(partETag string, entry *filer_pb.Entry) (match boo return normalizedPartETag == normalizedEntryETag, false, normalizedPartETag, normalizedEntryETag } + +type checksumTypeSupport struct { + composite bool + fullObject bool +} + +var checksumTypeSupportByAlgo = map[ChecksumAlgorithm]checksumTypeSupport{ + ChecksumAlgorithmCRC32: {composite: true, fullObject: true}, + ChecksumAlgorithmCRC32C: {composite: true, fullObject: true}, + ChecksumAlgorithmCRC64NVMe: {fullObject: true}, + ChecksumAlgorithmSHA1: {composite: true}, // Technically, fullObject could be supported, but is not yet implemented + ChecksumAlgorithmSHA256: {composite: true}, // Technically, fullObject could be supported, but is not yet implemented +} + +func resolveMultipartChecksumType(algo ChecksumAlgorithm, requested string) (string, error) { + support, ok := checksumTypeSupportByAlgo[algo] + if !ok { + return "", fmt.Errorf("unsupported checksum algorithm %v", algo) + } + + switch strings.ToUpper(strings.TrimSpace(requested)) { + case "": + if support.composite { + return s3_constants.ChecksumTypeComposite, nil + } + return s3_constants.ChecksumTypeFullObject, nil + case s3_constants.ChecksumTypeComposite: + if !support.composite { + return "", fmt.Errorf("checksum algorithm %v does not support %s checksums", algo, s3_constants.ChecksumTypeComposite) + } + return s3_constants.ChecksumTypeComposite, nil + case s3_constants.ChecksumTypeFullObject: + if !support.fullObject { + return "", fmt.Errorf("checksum algorithm %v does not support %s checksums", algo, s3_constants.ChecksumTypeFullObject) + } + return s3_constants.ChecksumTypeFullObject, nil + default: + return "", fmt.Errorf("invalid checksum type %q", requested) + } +} diff --git a/weed/s3api/filer_multipart_checksum_test.go b/weed/s3api/filer_multipart_checksum_test.go new file mode 100644 index 000000000..f0558e8f1 --- /dev/null +++ b/weed/s3api/filer_multipart_checksum_test.go @@ -0,0 +1,146 @@ +package s3api + +import ( + "bytes" + "encoding/base64" + "hash/crc32" + "testing" + + "github.com/minio/crc64nvme" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +func makeCRC64NVMEPartEntry(data []byte) *filer_pb.Entry { + sum := crc64nvme.New() + sum.Write(data) + return &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(data))}, + Extended: map[string][]byte{ + s3_constants.ExtChecksumAlgorithm: []byte(s3_constants.AmzChecksumCRC64NVME), + s3_constants.ExtChecksumValue: []byte(base64.StdEncoding.EncodeToString(sum.Sum(nil))), + }, + } +} + +func makeCRC32PartEntry(data []byte) *filer_pb.Entry { + sum := crc32.NewIEEE() + sum.Write(data) + return &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(len(data))}, + Extended: map[string][]byte{ + s3_constants.ExtChecksumAlgorithm: []byte(s3_constants.AmzChecksumCRC32), + s3_constants.ExtChecksumValue: []byte(base64.StdEncoding.EncodeToString(sum.Sum(nil))), + }, + } +} + +// Tests the FULL_OBJECT checksum of a multipart upload equals the CRC64NVME of the concatenated +// part data, with no composite "-N" suffix. +func TestComputeFullObjectChecksumCRC64NVME(t *testing.T) { + parts := [][]byte{ + bytes.Repeat([]byte("a"), 5*1024*1024), + bytes.Repeat([]byte("b"), 5*1024*1024), + []byte("tail part, smaller than the rest"), + } + + partEntries := map[int][]*filer_pb.Entry{} + var whole []byte + completed := []int{} + for i, data := range parts { + partNumber := i + 1 + partEntries[partNumber] = []*filer_pb.Entry{makeCRC64NVMEPartEntry(data)} + completed = append(completed, partNumber) + whole = append(whole, data...) + } + + got, err := computeFullObjectChecksum(s3_constants.AmzChecksumCRC64NVME, partEntries, completed) + if err != nil { + t.Fatalf("computeFullObjectChecksum: %v", err) + } + + wholeSum := crc64nvme.New() + wholeSum.Write(whole) + expected := base64.StdEncoding.EncodeToString(wholeSum.Sum(nil)) + + if got != expected { + t.Fatalf("full object checksum = %q, want %q", got, expected) + } + if bytes.ContainsRune([]byte(got), '-') { + t.Fatalf("full object checksum must not carry a -N suffix: %q", got) + } +} + +// Tests the FULL_OBJECT checksum of a multipart upload equals the CRC32 of the concatenated +// part data, with no composite "-N" suffix. +func TestComputeFullObjectChecksumCRC32(t *testing.T) { + parts := [][]byte{ + bytes.Repeat([]byte("a"), 5*1024*1024), + bytes.Repeat([]byte("b"), 5*1024*1024), + []byte("tail part, smaller than the rest"), + } + + partEntries := map[int][]*filer_pb.Entry{} + var whole []byte + completed := []int{} + for i, data := range parts { + partNumber := i + 1 + partEntries[partNumber] = []*filer_pb.Entry{makeCRC32PartEntry(data)} + completed = append(completed, partNumber) + whole = append(whole, data...) + } + + got, err := computeFullObjectChecksum(s3_constants.AmzChecksumCRC32, partEntries, completed) + if err != nil { + t.Fatalf("computeFullObjectChecksum: %v", err) + } + + wholeSum := crc32.NewIEEE() + wholeSum.Write(whole) + expected := base64.StdEncoding.EncodeToString(wholeSum.Sum(nil)) + + if got != expected { + t.Fatalf("full object checksum = %q, want %q", got, expected) + } + if bytes.ContainsRune([]byte(got), '-') { + t.Fatalf("full object checksum must not carry a -N suffix: %q", got) + } +} + +func TestResolveMultipartChecksumType(t *testing.T) { + cases := []struct { + name string + algo ChecksumAlgorithm + requested string + want string + wantErr bool + }{ + {"crc64 default full", ChecksumAlgorithmCRC64NVMe, "", s3_constants.ChecksumTypeFullObject, false}, + {"crc64 explicit full", ChecksumAlgorithmCRC64NVMe, s3_constants.ChecksumTypeFullObject, s3_constants.ChecksumTypeFullObject, false}, + {"crc64 composite rejected", ChecksumAlgorithmCRC64NVMe, s3_constants.ChecksumTypeComposite, "", true}, + {"crc32 default composite", ChecksumAlgorithmCRC32, "", s3_constants.ChecksumTypeComposite, false}, + {"crc32 explicit composite", ChecksumAlgorithmCRC32, s3_constants.ChecksumTypeComposite, s3_constants.ChecksumTypeComposite, false}, + {"crc32 explicit full", ChecksumAlgorithmCRC32, s3_constants.ChecksumTypeFullObject, s3_constants.ChecksumTypeFullObject, false}, + {"crc32c explicit full", ChecksumAlgorithmCRC32C, s3_constants.ChecksumTypeFullObject, s3_constants.ChecksumTypeFullObject, false}, + {"sha256 default composite", ChecksumAlgorithmSHA256, "", s3_constants.ChecksumTypeComposite, false}, + {"sha256 full rejected", ChecksumAlgorithmSHA256, s3_constants.ChecksumTypeFullObject, "", true}, + {"invalid checksum type", ChecksumAlgorithmCRC64NVMe, "bogus", "", true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, err := resolveMultipartChecksumType(tc.algo, tc.requested) + if tc.wantErr { + if err == nil { + t.Fatalf("expected error, got %q", got) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != tc.want { + t.Fatalf("got %q, want %q", got, tc.want) + } + }) + } +} diff --git a/weed/s3api/s3_constants/extend_key.go b/weed/s3api/s3_constants/extend_key.go index e3ad07e5f..21c996b19 100644 --- a/weed/s3api/s3_constants/extend_key.go +++ b/weed/s3api/s3_constants/extend_key.go @@ -36,6 +36,7 @@ const ( // S3 checksum storage keys (use x-seaweedfs- prefix to avoid leaking in generic header loop) ExtChecksumAlgorithm = "x-seaweedfs-checksum-algorithm" ExtChecksumValue = "x-seaweedfs-checksum-value" + ExtChecksumType = "x-seaweedfs-checksum-type" // Bucket Policy ExtBucketPolicyKey = "Seaweed-X-Amz-Bucket-Policy" diff --git a/weed/s3api/s3_constants/header.go b/weed/s3api/s3_constants/header.go index 81314eca1..cdfddb9cb 100644 --- a/weed/s3api/s3_constants/header.go +++ b/weed/s3api/s3_constants/header.go @@ -86,6 +86,13 @@ const ( AmzChecksumSHA256 = "X-Amz-Checksum-Sha256" AmzTrailer = "X-Amz-Trailer" AmzSdkChecksumAlgorithm = "X-Amz-Sdk-Checksum-Algorithm" + AmzChecksumType = "X-Amz-Checksum-Type" + + // S3 checksum type values (x-amz-checksum-type). A COMPOSITE checksum is a + // checksum-of-per-part-checksums ("base64-N"); a FULL_OBJECT checksum is the + // checksum of the whole object as if uploaded in a single request ("base64"). + ChecksumTypeComposite = "COMPOSITE" + ChecksumTypeFullObject = "FULL_OBJECT" // S3 conditional headers IfMatch = "If-Match" diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index eb67e6cbe..887a9f2a4 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -2091,6 +2091,9 @@ func (s3a *S3ApiServer) setResponseHeaders(w http.ResponseWriter, r *http.Reques if algoName, ok := entry.Extended[s3_constants.ExtChecksumAlgorithm]; ok { if checksumVal, ok := entry.Extended[s3_constants.ExtChecksumValue]; ok { w.Header().Set(string(algoName), string(checksumVal)) + if checksumType, ok := entry.Extended[s3_constants.ExtChecksumType]; ok && len(checksumType) > 0 { + w.Header().Set(s3_constants.AmzChecksumType, string(checksumType)) + } } } } diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go index 7f2feca5d..8b7b6b05b 100644 --- a/weed/s3api/s3api_object_handlers_multipart.go +++ b/weed/s3api/s3api_object_handlers_multipart.go @@ -180,9 +180,12 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r w.Header().Set("x-amz-version-id", *response.VersionId) } - // Set composite checksum header if present + // Set checksum header if present if response.ChecksumHeaderName != "" && response.ChecksumValue != "" { w.Header().Set(response.ChecksumHeaderName, response.ChecksumValue) + if response.ChecksumType != "" { + w.Header().Set(s3_constants.AmzChecksumType, response.ChecksumType) + } } stats_collect.RecordBucketActiveTime(bucket)