diff --git a/backend/common.go b/backend/common.go index 299735f6..b073dff9 100644 --- a/backend/common.go +++ b/backend/common.go @@ -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 +// ("-") from the completed-part ETags. +func ComputeMultipartETagFromPartETags(parts []types.CompletedPart) (string, error) { var partsEtagBytes []byte for _, part := range parts { if part.ETag == nil { diff --git a/backend/posix/posix.go b/backend/posix/posix.go index bb34ca12..6259c74c 100644 --- a/backend/posix/posix.go +++ b/backend/posix/posix.go @@ -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-" + // multipart part: "CRC64NVME-" + // multipart CRC64NVME: "CRC64NVME-" + // multipart composite: "ALGO--" + 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-"). 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--"). + 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-"). 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-" + // composite other algo: "ALGO--" + // full-object other algo: "ALGO-" + 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-") 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, diff --git a/backend/scoutfs/scoutfs.go b/backend/scoutfs/scoutfs.go index 98836b12..a0d94318 100644 --- a/backend/scoutfs/scoutfs.go +++ b/backend/scoutfs/scoutfs.go @@ -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-"). For multipart uploads, part ETags become + // CRC64NVME-based values and the completed object ETag is checksum-derived. + DataIntegrityEtag bool } var _ backend.Backend = &ScoutFS{} diff --git a/backend/scoutfs/scoutfs_compat.go b/backend/scoutfs/scoutfs_compat.go index 15076417..e4461c0c 100644 --- a/backend/scoutfs/scoutfs_compat.go +++ b/backend/scoutfs/scoutfs_compat.go @@ -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 diff --git a/cmd/versitygw/posix.go b/cmd/versitygw/posix.go index a28227ea..d7f98d95 100644 --- a/cmd/versitygw/posix.go +++ b/cmd/versitygw/posix.go @@ -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 diff --git a/cmd/versitygw/scoutfs.go b/cmd/versitygw/scoutfs.go index bd0c012f..6c2a24f7 100644 --- a/cmd/versitygw/scoutfs.go +++ b/cmd/versitygw/scoutfs.go @@ -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 { diff --git a/cmd/versitygw/test.go b/cmd/versitygw/test.go index 8eadca23..e65b33fc 100644 --- a/cmd/versitygw/test.go +++ b/cmd/versitygw/test.go @@ -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", diff --git a/extra/example.conf b/extra/example.conf index 4cebfc94..c58920bb 100644 --- a/extra/example.conf +++ b/extra/example.conf @@ -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 # ###### diff --git a/runtests.sh b/runtests.sh index a86f3659..6dbeabb1 100755 --- a/runtests.sh +++ b/runtests.sh @@ -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 diff --git a/tests/integration/CompleteMultipartUpload.go b/tests/integration/CompleteMultipartUpload.go index 6a4f28b9..a9d5a6d4 100644 --- a/tests/integration/CompleteMultipartUpload.go +++ b/tests/integration/CompleteMultipartUpload.go @@ -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 { diff --git a/tests/integration/PutObject.go b/tests/integration/PutObject.go index 365bf1ca..835232b2 100644 --- a/tests/integration/PutObject.go +++ b/tests/integration/PutObject.go @@ -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 { diff --git a/tests/integration/UploadPart.go b/tests/integration/UploadPart.go index 1314387e..573d3baa 100644 --- a/tests/integration/UploadPart.go +++ b/tests/integration/UploadPart.go @@ -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 { diff --git a/tests/integration/UploadPartCopy.go b/tests/integration/UploadPartCopy.go index 677c0b36..8a71ed46 100644 --- a/tests/integration/UploadPartCopy.go +++ b/tests/integration/UploadPartCopy.go @@ -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 { diff --git a/tests/integration/group-tests.go b/tests/integration/group-tests.go index ddfbae1d..383123da 100644 --- a/tests/integration/group-tests.go +++ b/tests/integration/group-tests.go @@ -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,