mirrors/versitygw
1
0
Fork 0
mirror of https://github.com/versity/versitygw.git synced 2026-08-17 20:56:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add optional data-integrity ETag mode for posix/scoutfs

Browse Source 
Introduce an opt-in mode that derives ETags from checksums instead of relying on
MD5 assumptions, aligning with AWS-compatible scenarios where ETag should be
treated as an opaque identifier rather than a content-MD5 signal.

This can significantly reduce CPU load on the server to disable MD5 checksums
when we are already validating data with other checksums.

This change adds --data-integrity-etag and VGW_DATA_INTEGRITY_ETAG support for
both posix and scoutfs backends. In this mode, PUT object ETags, multipart part
ETags, and completed multipart object ETags are checksum-derived.
This commit is contained in:
Ben McClelland
2026-07-31 14:58:21 -07:00
parent b590f9c8ba
commit 29256f0ab5
14 changed files with 651 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/common.go
+3 -1
View File
@@ -381,7 +381,9 @@ func isValidTagComponent(str string) bool {
return validTagComponent.Match([]byte(str))
}
func GetMultipartMD5(parts []types.CompletedPart) (string, error) {
// ComputeMultipartETagFromPartETags computes the S3 multipart ETag
// ("<md5>-<partCount>") from the completed-part ETags.
func ComputeMultipartETagFromPartETags(parts []types.CompletedPart) (string, error) {
var partsEtagBytes []byte
for _, part := range parts {
if part.ETag == nil {
backend/posix/posix.go
+255 -24
View File
@@ -21,6 +21,7 @@ import (
"encoding/json"
"errors"
"fmt"
"hash"
"io"
"io/fs"
"net/http"
@@ -117,6 +118,16 @@ type Posix struct {
// ioBufferSize is the buffer size used by buffered copy/read paths.
ioBufferSize int
ioBufferPool sync.Pool
// dataIntegrityEtag, when true, replaces the standard MD5-based ETag with
// a checksum-derived ETag that embeds the algorithm name and value.
// For multipart uploads this applies to both part ETags and the final
// completed object ETag:
// single PUT: "ALGO-<checksum>"
// multipart part: "CRC64NVME-<part-checksum>"
// multipart CRC64NVME: "CRC64NVME-<whole-file-checksum>"
// multipart composite: "ALGO-<composite-checksum>-<part-count>"
dataIntegrityEtag bool
}
var _ backend.Backend = &Posix{}
@@ -216,6 +227,13 @@ type PosixOpts struct {
// IOBufferSize sets the buffer size (in bytes) for copy/read paths that use
// io.CopyBuffer or buffered readers. Defaults to 1MiB when unset or invalid.
IOBufferSize int
// DataIntegrityEtag, when enabled, replaces the standard MD5-based ETag
// with a checksum-derived value that embeds the algorithm name and checksum
// (e.g. "CRC64NVME-<base64>"). For multipart uploads, part ETags become
// CRC64NVME-based values and the completed object ETag is checksum-derived.
// For CRC64NVME full-object checksums, the composable whole-file checksum is
// used; other algorithms embed the part count (e.g. "SHA256-<composite>-<N>").
DataIntegrityEtag bool
}
func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, error) {
@@ -289,6 +307,7 @@ func New(rootdir string, meta meta.MetadataStorer, opts PosixOpts) (*Posix, erro
b := make([]byte, ioBufferSize)
return &b
}},
dataIntegrityEtag: opts.DataIntegrityEtag,
}, nil
}
@@ -1864,30 +1883,141 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
sum := sha256.Sum256([]byte(object))
objdirFull := filepath.Join(bucket, MetaTmpMultipartDir, fmt.Sprintf("%x", sum))
uploadIDDir := filepath.Join(objdirFull, uploadID)
// Calculate s3 compatible md5sum for complete multipart.
s3MD5, err := backend.GetMultipartMD5(parts)
// Compute the default multipart ETag token used for claim naming.
// In standard mode this is the S3-compatible multipart MD5 ETag; in
// dataIntegrityEtag mode it may fall back to a deterministic claim token.
multipartClaimToken, err := backend.ComputeMultipartETagFromPartETags(parts)
if err != nil {
return res, "", err
if !p.dataIntegrityEtag {
return res, "", err
}
// dataIntegrityEtag allows non-MD5 multipart part ETags (for example,
// "CRC64NVME-<checksum>"). In that mode, ComputeMultipartETagFromPartETags
// cannot decode part ETags as hex MD5 bytes. Build a deterministic
// claim token from part-number/etag pairs so concurrent complete calls
// still contend on the same in-progress directory name.
h := sha256.New()
for _, part := range parts {
if part.ETag == nil || part.PartNumber == nil {
return res, "", s3err.GetAPIError(s3err.ErrMalformedXML)
}
_, _ = h.Write([]byte(strconv.FormatInt(int64(*part.PartNumber), 10)))
_, _ = h.Write([]byte{':'})
_, _ = h.Write([]byte(strings.Trim(*part.ETag, "\"")))
_, _ = h.Write([]byte{';'})
}
multipartClaimToken = fmt.Sprintf("\"%x-%d\"", h.Sum(nil), len(parts))
}
activeUploadName := fmt.Sprintf("%s.%s%s", uploadID, strings.Trim(s3MD5, "\""), inProgressSuffix)
activeUploadName := fmt.Sprintf("%s.%s%s", uploadID, strings.Trim(multipartClaimToken, "\""), inProgressSuffix)
uploadIDInProgress := filepath.Join(objdirFull, activeUploadName)
objdir := filepath.Join(MetaTmpMultipartDir, fmt.Sprintf("%x", sum))
predictDataIntegrityFinalETag := func(uploadName string) (string, error) {
checksums, err := p.retrieveChecksums(nil, bucket, filepath.Join(objdir, uploadName))
if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
return "", fmt.Errorf("get mp checksums: %w", err)
}
mpChecksumType := checksums.Type
if checksums.Type == "" {
checksums.Type = types.ChecksumTypeFullObject
checksums.Algorithm = types.ChecksumAlgorithmCrc64nvme
}
var compositeChecksumRdr *utils.CompositeChecksumReader
if checksums.Type == types.ChecksumTypeComposite {
compositeChecksumRdr, err = utils.NewCompositeChecksumReader(utils.HashType(strings.ToLower(string(checksums.Algorithm))))
if err != nil {
return "", fmt.Errorf("initialize composite checksum reader: %w", err)
}
}
var value string
var composableCsum string
for i, part := range parts {
if part.PartNumber == nil || part.ETag == nil {
return "", s3err.GetAPIError(s3err.ErrMalformedXML)
}
partObjPath := filepath.Join(objdir, uploadName, fmt.Sprintf("%v", *part.PartNumber))
fi, err := os.Lstat(filepath.Join(bucket, partObjPath))
if err != nil {
return "", s3err.GetInvalidPartErr(uploadID, *part.PartNumber, backend.GetStringFromPtr(part.ETag))
}
switch checksums.Type {
case types.ChecksumTypeFullObject:
var pcs string
if mpChecksumType != "" {
pcs = getPartChecksum(checksums.Algorithm, part)
} else {
crc64nvme, err := p.meta.RetrieveAttribute(nil, bucket, partObjPath, partCrc64nvme)
if err != nil {
return "", fmt.Errorf("retrieve part internal crc64nvme: %w", err)
}
pcs = string(crc64nvme)
}
if i == 0 {
composableCsum = pcs
} else {
composableCsum, err = utils.AddCRCChecksum(checksums.Algorithm, composableCsum, pcs, fi.Size())
if err != nil {
return "", fmt.Errorf("add part %v checksum: %w", *part.PartNumber, err)
}
}
case types.ChecksumTypeComposite:
if err := compositeChecksumRdr.Process(getPartChecksum(checksums.Algorithm, part)); err != nil {
return "", fmt.Errorf("process %v part checksum: %w", *part.PartNumber, err)
}
}
}
switch checksums.Type {
case types.ChecksumTypeComposite:
value = fmt.Sprintf("%s-%v", compositeChecksumRdr.Sum(), len(parts))
case types.ChecksumTypeFullObject:
value = composableCsum
}
if checksums.Algorithm == "" || value == "" {
return multipartClaimToken, nil
}
return fmt.Sprintf("\"%s-%s\"", strings.ToUpper(string(checksums.Algorithm)), value), nil
}
err = os.Rename(uploadIDDir, uploadIDInProgress)
if errors.Is(err, fs.ErrNotExist) {
// Another call already claimed this slot and is still assembling the object.
if _, statErr := os.Stat(uploadIDInProgress); statErr == nil {
etag := multipartClaimToken
if p.dataIntegrityEtag {
etag, err = predictDataIntegrityFinalETag(activeUploadName)
if err != nil {
return res, "", err
}
}
// Still in progress — treat as success for idempotency.
return s3response.CompleteMultipartUploadResult{
Bucket: &bucket,
ETag: &s3MD5,
ETag: &etag,
Key: &object,
}, "", nil
}
// Directory is gone: the concurrent call already completed and cleaned up.
if _, statErr := os.Stat(filepath.Join(bucket, object)); statErr == nil {
etag := multipartClaimToken
if p.dataIntegrityEtag {
etagBytes, etagErr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
if etagErr != nil {
return res, "", fmt.Errorf("get object etag: %w", etagErr)
}
etag = string(etagBytes)
}
return s3response.CompleteMultipartUploadResult{
Bucket: &bucket,
ETag: &s3MD5,
ETag: &etag,
Key: &object,
}, "", nil
}
@@ -1909,9 +2039,17 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
return res, "", s3err.GetAPIError(s3err.ErrNoSuchUpload)
}
etag := multipartClaimToken
if p.dataIntegrityEtag {
etagBytes, etagErr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
if etagErr != nil {
return res, "", fmt.Errorf("get object etag: %w", etagErr)
}
etag = string(etagBytes)
}
return s3response.CompleteMultipartUploadResult{
Bucket: &bucket,
ETag: &s3MD5,
ETag: &etag,
Key: &object,
}, "", nil
}
@@ -1946,8 +2084,6 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
}
}
objdir := filepath.Join(MetaTmpMultipartDir, fmt.Sprintf("%x", sum))
checksums, err := p.retrieveChecksums(nil, bucket, filepath.Join(objdir, activeUploadName))
if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
return res, "", fmt.Errorf("get mp checksums: %w", err)
@@ -2148,6 +2284,18 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
}
}
// Determine the ETag that will be stored on the final object and returned
// to the client. By default this is the S3-compatible MD5-of-part-ETags
// value. When dataIntegrityEtag is enabled, embed the checksum algorithm
// and computed value instead:
// CRC64NVME full-object: "CRC64NVME-<whole-file-checksum>"
// composite other algo: "ALGO-<composite-checksum>-<part-count>"
// full-object other algo: "ALGO-<whole-file-checksum>"
finalEtag := multipartClaimToken
if p.dataIntegrityEtag && checksums.Algorithm != "" && value != "" {
finalEtag = fmt.Sprintf("\"%s-%s\"", strings.ToUpper(string(checksums.Algorithm)), value)
}
f, err := p.openTmpFile(filepath.Join(bucket, MetaTmpDir), bucket, object,
totalsize, acct, skipFalloc, p.forceNoTmpFile, odirectNotAllowed)
if err != nil {
@@ -2312,7 +2460,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
}
}
err = p.meta.StoreAttribute(f.File(), bucket, object, etagkey, []byte(s3MD5))
err = p.meta.StoreAttribute(f.File(), bucket, object, etagkey, []byte(finalEtag))
if err != nil {
return res, "", fmt.Errorf("set etag attr: %w", err)
}
@@ -2343,7 +2491,7 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
return s3response.CompleteMultipartUploadResult{
Bucket: &bucket,
ETag: &s3MD5,
ETag: &finalEtag,
Key: &object,
ChecksumCRC32: crc32,
ChecksumCRC32C: crc32c,
@@ -3104,8 +3252,16 @@ func (p *Posix) UploadPartWithPostFunc(ctx context.Context, input *s3.UploadPart
}
defer f.cleanup()
hash := md5.New()
tr := io.TeeReader(r, hash)
// When dataIntegrityEtag is enabled, MD5 is never used for the part ETag,
// so skip both md5.New() and the per-byte TeeReader overhead.
var md5hash hash.Hash
var tr io.Reader
if p.dataIntegrityEtag {
tr = r
} else {
md5hash = md5.New()
tr = io.TeeReader(r, md5hash)
}
chRdr, chunkUpload := input.Body.(middlewares.ChecksumReader)
isTrailingChecksum := chunkUpload && chRdr.Algorithm() != ""
@@ -3210,6 +3366,16 @@ func (p *Posix) UploadPartWithPostFunc(ctx context.Context, input *s3.UploadPart
tr = hashRdr
}
// When dataIntegrityEtag is enabled and the mp algo isn't CRC64NVME,
// wrap the reader chain with an internal CRC64NVME reader so the part
// ETag can be CRC64NVME-based (trailing or non-trailing).
if p.dataIntegrityEtag && checksums.Algorithm != types.ChecksumAlgorithmCrc64nvme {
crc64nvmeRdr, err = utils.NewHashReader(tr, "", utils.HashTypeCRC64NVME)
if err != nil {
return nil, fmt.Errorf("initialize internal crc64nvme reader: %w", err)
}
tr = crc64nvmeRdr
}
}
buf := p.getIOBuffer()
@@ -3232,7 +3398,30 @@ func (p *Posix) UploadPartWithPostFunc(ctx context.Context, input *s3.UploadPart
return nil, fmt.Errorf("write part data: %w", err)
}
etag := backend.GenerateEtag(hash)
// Generate the part ETag.
// When dataIntegrityEtag is enabled, use CRC64NVME instead of MD5.
// CRC64NVME is available from: hashRdr/chRdr when the mp or input algo
// is already CRC64NVME, or from crc64nvmeRdr which is always set up in
// the non-CRC64NVME paths (checksums.Type=="" branch) or was just added
// above in the storeChecksum branch.
var etag string
if p.dataIntegrityEtag {
var crc64Sum string
isCrc64 := checksums.Algorithm == types.ChecksumAlgorithmCrc64nvme ||
(checksums.Type == "" && inputChAlgo == utils.HashTypeCRC64NVME)
if isCrc64 {
if isTrailingChecksum {
crc64Sum = chRdr.Checksum()
} else {
crc64Sum = hashRdr.Sum()
}
} else {
crc64Sum = crc64nvmeRdr.Sum()
}
etag = fmt.Sprintf("\"CRC64NVME-%s\"", crc64Sum)
} else {
etag = backend.GenerateEtag(md5hash)
}
err = p.meta.StoreAttribute(f.File(), bucket, partPath, etagkey, []byte(etag))
if err != nil {
return nil, fmt.Errorf("set etag attr: %w", err)
@@ -3491,8 +3680,12 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
defer f.cleanup()
rdr := io.NewSectionReader(srcf, startOffset, length)
hash := md5.New()
tr := io.TeeReader(rdr, hash)
var md5hash hash.Hash
var tr io.Reader = rdr
if !p.dataIntegrityEtag {
md5hash = md5.New()
tr = io.TeeReader(rdr, md5hash)
}
mpChecksums, err := p.retrieveChecksums(nil, *upi.Bucket, filepath.Join(objdir, *upi.UploadId))
if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
@@ -3526,6 +3719,15 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
tr = crc64nvmeRdr
}
if p.dataIntegrityEtag && crc64nvmeRdr == nil {
// still need a crc64nvme-derived ETag even though a different checksum is already being computed
crc64nvmeRdr, err = utils.NewHashReader(tr, "", utils.HashTypeCRC64NVME)
if err != nil {
return s3response.CopyPartResult{}, fmt.Errorf("initialize etag crc64nvme reader: %w", err)
}
tr = crc64nvmeRdr
}
_, err = io.Copy(f, tr)
if err != nil {
if errors.Is(err, syscall.EDQUOT) {
@@ -3574,7 +3776,12 @@ func (p *Posix) UploadPartCopy(ctx context.Context, upi *s3.UploadPartCopyInput)
}
}
etag := backend.GenerateEtag(hash)
var etag string
if p.dataIntegrityEtag {
etag = fmt.Sprintf("\"CRC64NVME-%s\"", crc64nvmeRdr.Sum())
} else {
etag = backend.GenerateEtag(md5hash)
}
err = p.meta.StoreAttribute(f.File(), *upi.Bucket, partPath, etagkey, []byte(etag))
if err != nil {
return s3response.CopyPartResult{}, fmt.Errorf("set etag attr: %w", err)
@@ -3758,9 +3965,16 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
}
}
expectedSum := getEmptyChecksumValue(checksumAlgorithm)
dirETag := emptyMD5
if p.dataIntegrityEtag {
dirETag = fmt.Sprintf("\"%s-%s\"", strings.ToUpper(string(checksumAlgorithm)), expectedSum)
}
// set etag attribute to signify this dir was specifically put
err = p.meta.StoreAttribute(nil, *po.Bucket, *po.Key, etagkey,
[]byte(emptyMD5))
[]byte(dirETag))
if err != nil {
return s3response.PutObjectOutput{}, fmt.Errorf("set etag attr: %w", err)
}
@@ -3780,7 +3994,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
}
}
expectedSum := getEmptyChecksumValue(checksumAlgorithm)
if checksumValue != "" && expectedSum != checksumValue {
return s3response.PutObjectOutput{}, s3err.GetChecksumBadDigestErr(checksumAlgorithm)
}
@@ -3800,7 +4013,7 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
// for directory object no version is created
return s3response.PutObjectOutput{
ETag: emptyMD5,
ETag: dirETag,
Size: &contentLength,
ChecksumType: checksum.Type,
ChecksumCRC32: checksum.CRC32,
@@ -3882,8 +4095,16 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
objsize := f.size
hash := md5.New()
rdr := io.TeeReader(po.Body, hash)
// When dataIntegrityEtag is enabled the MD5 is never used, so skip
// both the allocating md5.New() and the per-byte TeeReader overhead.
var md5hash hash.Hash
var rdr io.Reader
if p.dataIntegrityEtag {
rdr = po.Body
} else {
md5hash = md5.New()
rdr = io.TeeReader(po.Body, md5hash)
}
var hashRdr *utils.HashReader
if !isTrailingChecksum {
@@ -3934,8 +4155,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
}
}
etag := backend.GenerateEtag(hash)
// if the versioning is enabled, generate a new versionID for the object
var versionID string
if p.versioningEnabled() && vEnabled {
@@ -3966,6 +4185,18 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
sum = hashRdr.Sum()
}
// Generate the ETag for this object.
// When dataIntegrityEtag is enabled, embed the checksum algorithm name and
// value directly in the ETag (e.g. "CRC64NVME-<base64>") instead of the
// standard MD5 digest. The checksum is always available here: either
// supplied by the client or computed internally as CRC64NVME by default.
var etag string
if p.dataIntegrityEtag {
etag = fmt.Sprintf("\"%s-%s\"", strings.ToUpper(string(checksumAlgorithm)), sum)
} else {
etag = backend.GenerateEtag(md5hash)
}
checksum := s3response.Checksum{
Type: types.ChecksumTypeFullObject,
Algorithm: checksumAlgorithm,
backend/scoutfs/scoutfs.go
+5
View File
@@ -54,6 +54,11 @@ type ScoutfsOpts struct {
// attribute (e.g. files placed on the filesystem outside of versitygw).
// When empty, such objects are served with an empty ETag.
DefaultEtag string
// DataIntegrityEtag, when enabled, replaces the standard MD5-based ETag
// with a checksum-derived value that embeds the algorithm name and checksum
// (e.g. "CRC64NVME-<base64>"). For multipart uploads, part ETags become
// CRC64NVME-based values and the completed object ETag is checksum-derived.
DataIntegrityEtag bool
}
var _ backend.Backend = &ScoutFS{}
backend/scoutfs/scoutfs_compat.go
+1
View File
@@ -84,6 +84,7 @@ func New(rootdir string, opts ScoutfsOpts) (*ScoutFS, error) {
Concurrency: opts.Concurrency,
CopyObjectThreshold: opts.CopyObjectThreshold,
DefaultEtag: opts.DefaultEtag,
DataIntegrityEtag: opts.DataIntegrityEtag,
})
if err != nil {
return nil, err
cmd/versitygw/posix.go
+8
View File
@@ -37,6 +37,7 @@ var (
actionsConcurrency int
ioBufferSize int
defaultEtag string
dataIntegrityEtag bool
)
func posixCommand() *cli.Command {
@@ -137,6 +138,12 @@ will be translated into the file /mnt/fs/gwroot/mybucket/a/b/c/myobject`,
EnvVars: []string{"VGW_DEFAULT_ETAG"},
Destination: &defaultEtag,
},
&cli.BoolFlag{
Name: "data-integrity-etag",
Usage: "use data-integrity checksum-derived ETags instead of MD5-based ETags (PUT object ETag, multipart part ETags, and completed multipart object ETag)",
EnvVars: []string{"VGW_DATA_INTEGRITY_ETAG"},
Destination: &dataIntegrityEtag,
},
},
}
}
@@ -174,6 +181,7 @@ func runPosix(ctx *cli.Context) error {
IOBufferSize: ioBufferSize,
CopyObjectThreshold: copyObjectThreshold,
DefaultEtag: defaultEtag,
DataIntegrityEtag: dataIntegrityEtag,
}
var ms meta.MetadataStorer
cmd/versitygw/scoutfs.go
+7
View File
@@ -112,6 +112,12 @@ move interfaces as well as support for tiered filesystems.`,
EnvVars: []string{"VGW_DEFAULT_ETAG"},
Destination: &defaultEtag,
},
&cli.BoolFlag{
Name: "data-integrity-etag",
Usage: "use data-integrity checksum-derived ETags instead of MD5-based ETags (PUT object ETag, multipart part ETags, and completed multipart object ETag)",
EnvVars: []string{"VGW_DATA_INTEGRITY_ETAG"},
Destination: &dataIntegrityEtag,
},
},
}
}
@@ -142,6 +148,7 @@ func runScoutfs(ctx *cli.Context) error {
opts.Concurrency = actionsConcurrency
opts.CopyObjectThreshold = copyObjectThreshold
opts.DefaultEtag = defaultEtag
opts.DataIntegrityEtag = dataIntegrityEtag
be, err := scoutfs.New(ctx.Args().Get(0), opts)
if err != nil {
cmd/versitygw/test.go
+5
View File
@@ -211,6 +211,11 @@ func initTestCommands() []*cli.Command {
Usage: "Tests gateway in ACL-disabled mode",
Action: getAction(integration.TestNoAclMode),
},
{
Name: "data-integrity-etag",
Usage: "Tests checksum-derived ETag behavior",
Action: getAction(integration.TestDataIntegrityETag),
},
{
Name: "bench",
Usage: "Runs download/upload performance test on the gateway",
extra/example.conf
+18
View File
@@ -648,6 +648,15 @@ ROOT_SECRET_ACCESS_KEY=
# unknown-1. When not set, such objects are served with an empty ETag.
#VGW_DEFAULT_ETAG=
# The VGW_DATA_INTEGRITY_ETAG option enables checksum-derived ETags instead of
# MD5-based ETags for uploads. When enabled, this affects:
# - PUT object ETag
# - multipart part ETags
# - completed multipart object ETag
# This mode is intended for AWS-compatible flows where ETag should be treated
# as an opaque identifier instead of a content-MD5 signal.
#VGW_DATA_INTEGRITY_ETAG=false
###########
# scoutfs #
###########
@@ -719,6 +728,15 @@ ROOT_SECRET_ACCESS_KEY=
# unknown-1. When not set, such objects are served with an empty ETag.
#VGW_DEFAULT_ETAG=
# The VGW_DATA_INTEGRITY_ETAG option enables checksum-derived ETags instead of
# MD5-based ETags for uploads. When enabled, this affects:
# - PUT object ETag
# - multipart part ETags
# - completed multipart object ETag
# This mode is intended for AWS-compatible flows where ETag should be treated
# as an opaque identifier instead of a content-MD5 signal.
#VGW_DATA_INTEGRITY_ETAG=false
######
# s3 #
######
runtests.sh
+27
View File
@@ -2,9 +2,11 @@
# parse options
USE_SIDECAR=false
RUN_DATA_INTEGRITY_ETAG_TESTS=true
for arg in "$@"; do
case "$arg" in
--sidecar) USE_SIDECAR=true ;;
--skip-data-integrity-etag-tests) RUN_DATA_INTEGRITY_ETAG_TESTS=false ;;
esac
done
@@ -178,6 +180,31 @@ fi
# kill off server
kill $GW_VS_HTTPS_PID
if $RUN_DATA_INTEGRITY_ETAG_TESTS; then
ECHO "Running data-integrity-etag integration tests"
# run server in background with data-integrity-etag enabled
# port: 7075
GOCOVERDIR=/tmp/covdata ./versitygw -p :7075 -a user -s pass --iam-dir /tmp/gw posix $SIDECAR_FLAG --data-integrity-etag /tmp/gw &
GW_DI_ETAG_PID=$!
# wait a second for server to start up
sleep 1
# check if data-integrity-etag gateway process is still running
if ! kill -0 $GW_DI_ETAG_PID; then
echo "data-integrity-etag server no longer running"
exit 1
fi
if ! ./versitygw test -a user -s pass -e http://127.0.0.1:7075 data-integrity-etag; then
echo "data-integrity-etag tests failed"
kill $GW_DI_ETAG_PID
exit 1
fi
kill $GW_DI_ETAG_PID
fi
ECHO "Running No ACL integration tests"
# run server in background versioning-enabled
# port: 7073
tests/integration/CompleteMultipartUpload.go
+83
View File
@@ -29,6 +29,7 @@ import (
"sync"
"time"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/versity/versitygw/s3err"
@@ -1850,6 +1851,88 @@ func CompleteMultipartUpload_success(s *S3Conf) error {
})
}
func CompleteMultipartUpload_data_integrity_etag(s *S3Conf) error {
testName := "CompleteMultipartUpload_data_integrity_etag"
return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
})
obj := "my-obj"
out, err := createMp(customClient, bucket, obj)
if err != nil {
return err
}
partNumber := int32(1)
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
partOut, err := customClient.UploadPart(ctx, &s3.UploadPartInput{
Bucket: &bucket,
Key: &obj,
UploadId: out.UploadId,
PartNumber: &partNumber,
Body: bytes.NewReader([]byte("payload-data")),
})
cancel()
if err != nil {
return err
}
if !isQuotedEtag(getString(partOut.ETag)) {
return fmt.Errorf("expected UploadPart ETag to be quoted, instead got %s", getString(partOut.ETag))
}
if !strings.HasPrefix(getString(partOut.ETag), "\"CRC64NVME-") {
return fmt.Errorf("expected UploadPart ETag to be CRC64NVME-based, instead got %s", getString(partOut.ETag))
}
ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
res, err := customClient.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
Bucket: &bucket,
Key: &obj,
UploadId: out.UploadId,
MultipartUpload: &types.CompletedMultipartUpload{
Parts: []types.CompletedPart{{
ETag: partOut.ETag,
PartNumber: &partNumber,
}},
},
})
cancel()
if err != nil {
return err
}
if !isQuotedEtag(getString(res.ETag)) {
return fmt.Errorf("expected CompleteMultipartUpload ETag to be quoted, instead got %s", getString(res.ETag))
}
if !strings.HasPrefix(getString(res.ETag), "\"CRC64NVME-") {
return fmt.Errorf("expected CompleteMultipartUpload ETag to be CRC64NVME-based, instead got %s", getString(res.ETag))
}
if getString(res.ETag) != getString(partOut.ETag) {
return fmt.Errorf("expected single-part complete ETag %s, instead got %s", getString(partOut.ETag), getString(res.ETag))
}
ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
head, err := customClient.HeadObject(ctx, &s3.HeadObjectInput{
Bucket: &bucket,
Key: &obj,
ChecksumMode: types.ChecksumModeEnabled,
})
cancel()
if err != nil {
return err
}
if got := getString(head.ETag); got != getString(res.ETag) {
return fmt.Errorf("expected HeadObject ETag to be %s, instead got %s", getString(res.ETag), got)
}
if getString(head.ChecksumCRC64NVME) == "" {
return fmt.Errorf("expected HeadObject CRC64NVME checksum to be set")
}
return nil
})
}
func CompleteMultipartUpload_racey_success(s *S3Conf) error {
testName := "CompleteMultipartUpload_racey_success"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
tests/integration/PutObject.go
+98
View File
@@ -891,6 +891,104 @@ func PutObject_default_checksum(s *S3Conf) error {
})
}
func PutObject_data_integrity_etag(s *S3Conf) error {
testName := "PutObject_data_integrity_etag"
return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
})
obj := "my-obj"
out, err := putObjectWithData(256, &s3.PutObjectInput{
Bucket: &bucket,
Key: &obj,
}, customClient)
if err != nil {
return err
}
if out.res.ChecksumCRC64NVME == nil {
return fmt.Errorf("expected non nil crc64nvme checksum in PutObject response")
}
expectedETag := fmt.Sprintf("\"CRC64NVME-%s\"", getString(out.res.ChecksumCRC64NVME))
if got := getString(out.res.ETag); got != expectedETag {
return fmt.Errorf("expected PutObject ETag to be %s, instead got %s", expectedETag, got)
}
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
head, err := customClient.HeadObject(ctx, &s3.HeadObjectInput{
Bucket: &bucket,
Key: &obj,
ChecksumMode: types.ChecksumModeEnabled,
})
cancel()
if err != nil {
return err
}
if got := getString(head.ETag); got != expectedETag {
return fmt.Errorf("expected HeadObject ETag to be %s, instead got %s", expectedETag, got)
}
if got := getString(head.ChecksumCRC64NVME); got != getString(out.res.ChecksumCRC64NVME) {
return fmt.Errorf("expected HeadObject checksum to be %s, instead got %s", getString(out.res.ChecksumCRC64NVME), got)
}
return nil
})
}
func PutObject_dir_object_data_integrity_etag(s *S3Conf) error {
testName := "PutObject_dir_object_data_integrity_etag"
return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
})
obj := "dir/obj/"
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
res, err := customClient.PutObject(ctx, &s3.PutObjectInput{
Bucket: &bucket,
Key: &obj,
})
cancel()
if err != nil {
return err
}
if res.ChecksumCRC64NVME == nil {
return fmt.Errorf("expected non nil crc64nvme checksum in PutObject response")
}
expectedETag := fmt.Sprintf("\"CRC64NVME-%s\"", getString(res.ChecksumCRC64NVME))
if got := getString(res.ETag); got != expectedETag {
return fmt.Errorf("expected PutObject ETag to be %s, instead got %s", expectedETag, got)
}
ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
head, err := customClient.HeadObject(ctx, &s3.HeadObjectInput{
Bucket: &bucket,
Key: &obj,
ChecksumMode: types.ChecksumModeEnabled,
})
cancel()
if err != nil {
return err
}
if got := getString(head.ETag); got != expectedETag {
return fmt.Errorf("expected HeadObject ETag to be %s, instead got %s", expectedETag, got)
}
if got := getString(head.ChecksumCRC64NVME); got != getString(res.ChecksumCRC64NVME) {
return fmt.Errorf("expected HeadObject checksum to be %s, instead got %s", getString(res.ChecksumCRC64NVME), got)
}
return nil
})
}
func PutObject_dir_object_default_checksum(s *S3Conf) error {
testName := "PutObject_dir_object_default_checksum"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
tests/integration/UploadPart.go
+62
View File
@@ -502,6 +502,68 @@ func UploadPart_success(s *S3Conf) error {
})
}
func UploadPart_data_integrity_etag(s *S3Conf) error {
testName := "UploadPart_data_integrity_etag"
partNumber := int32(1)
return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
})
obj := "my-obj"
out, err := createMp(customClient, bucket, obj)
if err != nil {
return err
}
payload := []byte("payload-data")
h := crc64.New(crc64.MakeTable(bits.Reverse64(0xad93d23594c93659)))
h.Write(payload)
expectedChecksum := base64.StdEncoding.EncodeToString(h.Sum(nil))
expectedETag := fmt.Sprintf("\"CRC64NVME-%s\"", expectedChecksum)
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
res, err := customClient.UploadPart(ctx, &s3.UploadPartInput{
Bucket: &bucket,
Key: &obj,
UploadId: out.UploadId,
PartNumber: &partNumber,
Body: bytes.NewReader(payload),
})
cancel()
if err != nil {
return err
}
if got := getString(res.ETag); got != expectedETag {
return fmt.Errorf("expected UploadPart ETag to be %s, instead got %s", expectedETag, got)
}
if res.ChecksumCRC64NVME != nil && getString(res.ChecksumCRC64NVME) != expectedChecksum {
return fmt.Errorf("expected UploadPart checksum to be %s, instead got %s", expectedChecksum, getString(res.ChecksumCRC64NVME))
}
ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
listOut, err := customClient.ListParts(ctx, &s3.ListPartsInput{
Bucket: &bucket,
Key: &obj,
UploadId: out.UploadId,
})
cancel()
if err != nil {
return err
}
if len(listOut.Parts) != 1 {
return fmt.Errorf("expected 1 uploaded part, instead got %d", len(listOut.Parts))
}
if got := getString(listOut.Parts[0].ETag); got != expectedETag {
return fmt.Errorf("expected ListParts ETag to be %s, instead got %s", expectedETag, got)
}
return nil
})
}
// isQuotedEtag reports whether an ETag is a non-empty double-quoted string,
// as required by the S3 contract (e.g. "\"abc\"").
func isQuotedEtag(etag string) bool {
tests/integration/UploadPartCopy.go
+66
View File
@@ -20,6 +20,7 @@ import (
"strings"
"time"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/aws-sdk-go-v2/service/s3/types"
"github.com/versity/versitygw/s3err"
@@ -1007,6 +1008,71 @@ func UploadPartCopy_should_calculate_the_checksum(s *S3Conf) error {
})
}
func UploadPartCopy_data_integrity_etag(s *S3Conf) error {
testName := "UploadPartCopy_data_integrity_etag"
return actionHandler(s, testName, func(_ *s3.Client, bucket string) error {
customClient := s3.NewFromConfig(s.Config(), func(o *s3.Options) {
o.RequestChecksumCalculation = aws.RequestChecksumCalculationUnset
})
obj := "my-obj"
srcObj := "source-object"
mp, err := createMp(customClient, bucket, obj)
if err != nil {
return err
}
if _, err := putObjectWithData(300, &s3.PutObjectInput{
Bucket: &bucket,
Key: &srcObj,
}, customClient); err != nil {
return err
}
partNumber := int32(1)
ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
res, err := customClient.UploadPartCopy(ctx, &s3.UploadPartCopyInput{
Bucket: &bucket,
Key: &obj,
UploadId: mp.UploadId,
PartNumber: &partNumber,
CopySource: getPtr(fmt.Sprintf("%v/%v", bucket, srcObj)),
})
cancel()
if err != nil {
return err
}
if !isQuotedEtag(getString(res.CopyPartResult.ETag)) {
return fmt.Errorf("expected UploadPartCopy ETag to be quoted, instead got %s", getString(res.CopyPartResult.ETag))
}
if !strings.HasPrefix(getString(res.CopyPartResult.ETag), "\"CRC64NVME-") {
return fmt.Errorf("expected UploadPartCopy ETag to be CRC64NVME-based, instead got %s", getString(res.CopyPartResult.ETag))
}
ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
parts, err := customClient.ListParts(ctx, &s3.ListPartsInput{
Bucket: &bucket,
Key: &obj,
UploadId: mp.UploadId,
})
cancel()
if err != nil {
return err
}
if len(parts.Parts) != 1 {
return fmt.Errorf("expected one uploaded part, instead got %d", len(parts.Parts))
}
if got := getString(parts.Parts[0].ETag); got != getString(res.CopyPartResult.ETag) {
return fmt.Errorf("expected ListParts ETag to be %s, instead got %s", getString(res.CopyPartResult.ETag), got)
}
return nil
})
}
func UploadPartCopy_incorrect_source_bucket_expected_owner(s *S3Conf) error {
testName := "UploadPartCopy_incorrect_source_bucket_expected_owner"
return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
tests/integration/group-tests.go
+13
View File
@@ -1356,6 +1356,14 @@ func TestNoAclMode(ts *TestState) {
ts.Run(NoAclMode_PutBucketAcl)
}
func TestDataIntegrityETag(ts *TestState) {
ts.Run(PutObject_data_integrity_etag)
ts.Run(PutObject_dir_object_data_integrity_etag)
ts.Run(UploadPart_data_integrity_etag)
ts.Run(UploadPartCopy_data_integrity_etag)
ts.Run(CompleteMultipartUpload_data_integrity_etag)
}
type IntTest func(s3 *S3Conf) error
type IntTests map[string]IntTest
@@ -1431,6 +1439,8 @@ func GetIntTests() IntTests {
"PutObject_invalid_checksum_header": PutObject_invalid_checksum_header,
"PutObject_incorrect_checksums": PutObject_incorrect_checksums,
"PutObject_default_checksum": PutObject_default_checksum,
"PutObject_data_integrity_etag": PutObject_data_integrity_etag,
"PutObject_dir_object_data_integrity_etag": PutObject_dir_object_data_integrity_etag,
"PutObject_dir_object_default_checksum": PutObject_dir_object_default_checksum,
"PutObject_checksums_success": PutObject_checksums_success,
"PutObject_dir_object_checksums_success": PutObject_dir_object_checksums_success,
@@ -1694,6 +1704,7 @@ func GetIntTests() IntTests {
"UploadPart_with_checksums_success": UploadPart_with_checksums_success,
"UploadPart_success": UploadPart_success,
"UploadPart_etag_quoting_consistency": UploadPart_etag_quoting_consistency,
"UploadPart_data_integrity_etag": UploadPart_data_integrity_etag,
"UploadPartCopy_non_existing_bucket": UploadPartCopy_non_existing_bucket,
"UploadPartCopy_incorrect_uploadId": UploadPartCopy_incorrect_uploadId,
"UploadPartCopy_incorrect_object_key": UploadPartCopy_incorrect_object_key,
@@ -1711,6 +1722,7 @@ func GetIntTests() IntTests {
"UploadPartCopy_should_copy_the_checksum": UploadPartCopy_should_copy_the_checksum,
"UploadPartCopy_should_not_copy_the_checksum": UploadPartCopy_should_not_copy_the_checksum,
"UploadPartCopy_should_calculate_the_checksum": UploadPartCopy_should_calculate_the_checksum,
"UploadPartCopy_data_integrity_etag": UploadPartCopy_data_integrity_etag,
"ListParts_incorrect_uploadId": ListParts_incorrect_uploadId,
"ListParts_incorrect_object_key": ListParts_incorrect_object_key,
"ListParts_invalid_max_parts": ListParts_invalid_max_parts,
@@ -1767,6 +1779,7 @@ func GetIntTests() IntTests {
"CompleteMultipartUpload_should_ignore_the_final_checksum": CompleteMultipartUpload_should_ignore_the_final_checksum,
"CompleteMultipartUpload_should_succeed_without_final_checksum_type": CompleteMultipartUpload_should_succeed_without_final_checksum_type,
"CompleteMultipartUpload_success": CompleteMultipartUpload_success,
"CompleteMultipartUpload_data_integrity_etag": CompleteMultipartUpload_data_integrity_etag,
"CompleteMultipartUpload_already_completed": CompleteMultipartUpload_already_completed,
"CompleteMultipartUpload_racey_success": CompleteMultipartUpload_racey_success,
"CompleteMultipartUpload_racey_data_integrity": CompleteMultipartUpload_racey_data_integrity,