fix: stop posix mixing up keys that differ only by a trailing slash

In posix a key and the same key with a trailing slash, such as `foo` and `foo/`, map to one path and share one set of attributes: a file there is the object of `foo`, a directory object is the object of `foo/`. Several operations used the entry at that path without checking that it belongs to the requested key. They now check it through the new `isLiveObject`, `statLiveObject` and `objVersionAttrPath` helpers.

`DeleteObject` of `foo` with a version id of `foo/` treated the directory as the current version of `foo` and removed it. That left the noncurrent versions of `foo/` unlisted and undeletable, so `DeleteBucket` failed with `BucketNotEmpty`. A delete of `foo` without a version id failed with an internal error while trying to version the directory as a file. Both now succeed without touching `foo/`, as for any object that doesn't exist.

`CompleteMultipartUpload` of `foo` cleared the `delete-marker` attribute of `foo/` and then failed with an internal error when linking the object onto the directory, which turned the delete marker back into a live version. It now returns `ExistingObjectIsDirectory` before any attribute is changed, both before the parts are assembled and again under the object publish lock. The idempotent completion path also no longer reports a missing upload as completed just because `foo/` exists.

The object tagging, legal hold and retention APIs read and wrote the attributes of the other key. For example, `PutObjectLegalHold` on `foo` could turn off the legal hold of `foo/`, and the object lock check could block a delete because of the other key's retention. When the requested key has no object, they now return `NoSuchKey` for the current version and `NoSuchVersion` for a specific version. They also resolve the `null` version id to a current null version instead of looking for it in the versioning directory. The object lock check treats `NoSuchVersion` like `NoSuchKey`, since a version that doesn't exist has nothing to protect. Conditional writes no longer evaluate `If-Match` and `If-None-Match` against the other key's ETag. `CreateMultipartUpload` now stores the upload's tagging and object lock settings directly on the upload directory, and `PutObject` of a directory object sets its tagging after the directory gets its ETag.

auth/object_lock.go

@@ -495,7 +495,9 @@ func (s objectLockState) checkObject(ctx context.Context, be backend.Backend, ia
 
 	checkRetention := true
 	retentionData, err := be.GetObjectRetention(ctx, bucket, key, versionId)
-	if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) {
+	// an object or version that doesn't exist has nothing to protect
+	if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) ||
+		errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchVersion)) {
 		return nil
 	}
 	// the object is a delete marker, if a `MethodNotAllowed` error is returned
@@ -538,7 +540,8 @@ func (s objectLockState) checkObject(ctx context.Context, be backend.Backend, ia
 
 	status, err := be.GetObjectLegalHold(ctx, bucket, key, versionId)
 	if err != nil {
-		if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) {
+		if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchKey)) ||
+			errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchVersion)) {
 			return nil
 		}
 		if errors.Is(err, s3err.GetAPIError(s3err.ErrNoSuchObjectLockConfiguration)) {

backend/posix/posix.go

@@ -631,21 +631,20 @@ func (p *Posix) doesBucketExist(bucket string) error {
 	return nil
 }
 
-func (p *Posix) doesBucketAndObjectExist(bucket, object string) error {
+// doesBucketAndObjectExist checks that bucket exists and, when versionId is
+// empty, that object has a current version. A specific version is looked up
+// once versionId is validated.
+func (p *Posix) doesBucketAndObjectExist(bucket, object, versionId string) error {
 	err := p.doesBucketExist(bucket)
 	if err != nil {
 		return err
 	}
-
-	_, err = os.Stat(p.ObjectPath(bucket, object))
-	if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-		return s3err.GetAPIError(s3err.ErrNoSuchKey)
-	}
-	if err != nil {
-		return fmt.Errorf("stat object: %w", err)
+	if versionId != "" {
+		return nil
 	}
 
-	return nil
+	_, _, err = p.objVersionAttrPath(bucket, object, "")
+	return err
 }
 
 func (p *Posix) ListBuckets(ctx context.Context, input s3response.ListBucketsInput) (s3response.ListAllMyBucketsResult, error) {
@@ -1300,15 +1299,79 @@ func (p *Posix) isDirObject(bucket, key string) (bool, error) {
 	return true, nil
 }
 
-// isLiveDirObject reports whether fi, the entry at the path of the
-// directory object key, is a directory object
-func (p *Posix) isLiveDirObject(fi os.FileInfo, bucket, key string) (bool, error) {
-	if !fi.IsDir() {
+// isLiveObject reports whether fi, the entry at the path of key, is the
+// current version of key. A key and the same key with a trailing slash
+// share one path and one set of attributes: a file there is the object of
+// the key without the slash, a directory object the object of the key with
+// it.
+func (p *Posix) isLiveObject(fi os.FileInfo, bucket, key string) (bool, error) {
+	if fi.IsDir() != strings.HasSuffix(key, "/") {
 		return false, nil
 	}
+	if !fi.IsDir() {
+		return true, nil
+	}
 	return p.isDirObject(bucket, key)
 }
 
+// statLiveObject returns the file info of the current version of key. An
+// error matching fs.ErrNotExist is returned when key has no current
+// version, including when the entry at its path is another key's object.
+func (p *Posix) statLiveObject(bucket, key string) (os.FileInfo, error) {
+	fi, err := os.Stat(p.ObjectPath(bucket, key))
+	if isErrNotDir(err) {
+		return nil, fs.ErrNotExist
+	}
+	if err != nil {
+		return nil, err
+	}
+	isObj, err := p.isLiveObject(fi, bucket, key)
+	if err != nil {
+		return nil, err
+	}
+	if !isObj {
+		return nil, fs.ErrNotExist
+	}
+	return fi, nil
+}
+
+// objVersionAttrPath returns the bucket and object that the attributes of
+// the version versionId of key are stored at: key itself for its current
+// version, an entry of the versioning directory for other versions. An
+// empty versionId selects the current version, and NoSuchKey is returned if
+// key has none.
+func (p *Posix) objVersionAttrPath(bucket, key, versionId string) (string, string, error) {
+	_, err := p.statLiveObject(bucket, key)
+	if isErrNameTooLong(err) {
+		return "", "", s3err.GetKeyTooLongErr(int64(len(key)), 1024)
+	}
+	if err != nil && !errors.Is(err, fs.ErrNotExist) {
+		return "", "", fmt.Errorf("stat object: %w", err)
+	}
+	isLive := err == nil
+
+	if versionId == "" {
+		if !isLive {
+			return "", "", s3err.GetAPIError(s3err.ErrNoSuchKey)
+		}
+		return bucket, key, nil
+	}
+
+	if isLive {
+		vId, err := p.meta.RetrieveAttribute(nil, bucket, key, versionIdKey)
+		if errors.Is(err, meta.ErrNoSuchKey) {
+			vId = []byte(nullVersionId)
+		} else if err != nil {
+			return "", "", fmt.Errorf("get obj versionId: %w", err)
+		}
+		if string(vId) == versionId {
+			return bucket, key, nil
+		}
+	}
+
+	return filepath.Join(p.versioningDir, bucket), filepath.Join(genObjVersionKey(key), versionId), nil
+}
+
 // clearDirObjectAttrs removes the directory object attributes, including
 // legacy metadata attributes, from the directory at bucket/key. The etag is
 // kept: it marks the directory as an object, so a failure before the new
@@ -1962,7 +2025,7 @@ func (p *Posix) CreateMultipartUpload(ctx context.Context, mpu s3response.Create
 
 	// set object tagging
 	if tags != nil {
-		err := p.PutObjectTagging(withCtxNoSlot(ctx), bucket, filepath.Join(objdir, uploadID), "", tags)
+		err := p.storeObjectTags(bucket, filepath.Join(objdir, uploadID), tags)
 		if err != nil {
 			// cleanup object if returning error
 			os.RemoveAll(filepath.Join(tmppath, uploadID))
@@ -1993,7 +2056,10 @@ func (p *Posix) CreateMultipartUpload(ctx context.Context, mpu s3response.Create
 
 	// set object legal hold
 	if mpu.ObjectLockLegalHoldStatus == types.ObjectLockLegalHoldStatusOn {
-		err := p.PutObjectLegalHold(withCtxNoSlot(ctx), bucket, filepath.Join(objdir, uploadID), "", true)
+		err := p.isBucketObjectLockEnabled(bucket)
+		if err == nil {
+			err = p.meta.StoreAttribute(nil, bucket, filepath.Join(objdir, uploadID), objectLegalHoldKey, []byte{1})
+		}
 		if err != nil {
 			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
 				err = s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
@@ -2020,7 +2086,10 @@ func (p *Posix) CreateMultipartUpload(ctx context.Context, mpu s3response.Create
 			_ = p.meta.DeleteAttributes(bucket, filepath.Join(objdir, uploadID))
 			return s3response.InitiateMultipartUploadResult{}, fmt.Errorf("parse object lock retention: %w", err)
 		}
-		err = p.PutObjectRetention(withCtxNoSlot(ctx), bucket, filepath.Join(objdir, uploadID), "", retParsed)
+		err = p.isBucketObjectLockEnabled(bucket)
+		if err == nil {
+			err = p.meta.StoreAttribute(nil, bucket, filepath.Join(objdir, uploadID), objectRetentionKey, retParsed)
+		}
 		if err != nil {
 			if errors.Is(err, s3err.GetAPIError(s3err.ErrMissingObjectLockConfiguration)) {
 				err = s3err.GetAPIError(s3err.ErrMissingObjectLockConfigurationNoSpaces)
@@ -2313,7 +2382,9 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 			}, "", nil
 		}
 		// Directory is gone: the concurrent call already completed and cleaned up.
-		if _, statErr := os.Stat(p.ObjectPath(bucket, object)); statErr == nil {
+		// A directory at the object path is the object of the key with a
+		// trailing slash, not the completed upload.
+		if fi, statErr := os.Stat(p.ObjectPath(bucket, object)); statErr == nil && !fi.IsDir() {
 			etag := multipartClaimToken
 			if p.dataIntegrityEtag {
 				etagBytes, etagErr := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
@@ -2383,13 +2454,17 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 	defer os.Rename(uploadIDInProgress, uploadIDDir)
 	defer p.meta.RenameObject(bucket, newMetaObj, oldMetaObj)
 
-	// Fast-fail precondition check before the parts are assembled. This is
-	// only advisory: the authoritative check is repeated while holding the
-	// object publish lock just before the final link.
+	// Fast-fail precondition and directory checks before the parts are
+	// assembled. These are only advisory: the authoritative checks are
+	// repeated while holding the object publish lock just before the final
+	// link.
 	err = p.checkPutPreconditions(bucket, object, input.IfMatch, input.IfNoneMatch)
 	if err != nil {
 		return res, "", err
 	}
+	if d, err := os.Stat(p.ObjectPath(bucket, object)); err == nil && d.IsDir() {
+		return res, "", s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)
+	}
 
 	checksums, err := p.retrieveChecksums(nil, bucket, filepath.Join(objdir, activeUploadName))
 	if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
@@ -2717,9 +2792,14 @@ func (p *Posix) CompleteMultipartUploadWithCopy(ctx context.Context, input *s3.C
 	vEnabled := p.isBucketVersioningEnabled(vStatus)
 
 	d, err := os.Stat(objname)
+	if err == nil && d.IsDir() {
+		// the directory is the object of the key with a trailing slash, or
+		// the parent of other objects: its attributes are not this object's
+		return res, "", s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)
+	}
 
 	// if the versioning is enabled first create the file object version
-	if p.versioningEnabled() && vEnabled && err == nil && !d.IsDir() {
+	if p.versioningEnabled() && vEnabled && err == nil {
 		_, err := p.createObjVersion(bucket, object, d.Size(), acct, false)
 		if err != nil {
 			return res, "", fmt.Errorf("create object version: %w", err)
@@ -4189,6 +4269,13 @@ func (p *Posix) checkPutPreconditions(bucket, object string, ifMatch, ifNoneMatc
 		return s3err.GetAPIError(s3err.ErrNotImplemented)
 	}
 
+	// the etag at the object path may be the one of the key with or
+	// without the trailing slash
+	_, err := p.statLiveObject(bucket, object)
+	if errors.Is(err, fs.ErrNotExist) {
+		return backend.EvaluateObjectPutPreconditions("", ifMatch, ifNoneMatch, false)
+	}
+
 	etagBytes, err := p.meta.RetrieveAttribute(nil, bucket, object, etagkey)
 	if err == nil || errors.Is(err, fs.ErrNotExist) || errors.Is(err, meta.ErrNoSuchKey) {
 		return backend.EvaluateObjectPutPreconditions(string(etagBytes), ifMatch, ifNoneMatch, err == nil)
@@ -4420,14 +4507,6 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			return s3response.PutObjectOutput{}, fmt.Errorf("set object metadata: %w", err)
 		}
 
-		// Set object tagging
-		if tags != nil {
-			err := p.PutObjectTagging(withCtxNoSlot(ctx), *po.Bucket, *po.Key, "", tags)
-			if err != nil {
-				return s3response.PutObjectOutput{}, err
-			}
-		}
-
 		dirETag := emptyMD5
 		if p.dataIntegrityEtag {
 			dirETag = fmt.Sprintf("\"%s-%s\"", strings.ToUpper(string(checksumAlgorithm)), expectedSum)
@@ -4483,6 +4562,14 @@ func (p *Posix) PutObjectWithPostFunc(ctx context.Context, po s3response.PutObje
 			_ = os.Chtimes(name, now, now)
 		}
 
+		// Set object tagging once the etag makes the directory an object
+		if tags != nil {
+			err := p.PutObjectTagging(withCtxNoSlot(ctx), *po.Bucket, *po.Key, "", tags)
+			if err != nil {
+				return s3response.PutObjectOutput{}, err
+			}
+		}
+
 		err = p.putObjectLockSettings(ctx, po)
 		if err != nil {
 			return s3response.PutObjectOutput{}, err
@@ -4896,15 +4983,15 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 			if err != nil {
 				return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
 			}
-			if isDir {
-				isObj, err := p.isLiveDirObject(fi, bucket, object)
-				if err != nil {
-					return nil, err
-				}
-				if !isObj {
-					// AWS returns success if the object does not exist
-					return &s3.DeleteObjectOutput{}, nil
-				}
+			// the entry at the object path may be the object of the key
+			// with or without the trailing slash
+			isObj, err := p.isLiveObject(fi, bucket, object)
+			if err != nil {
+				return nil, err
+			}
+			if !isObj {
+				// AWS returns success if the object does not exist
+				return &s3.DeleteObjectOutput{}, nil
 			}
 
 			err = evalPreconditions(fi, bucket, object)
@@ -4962,28 +5049,19 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 		} else {
 			versionPath := p.genObjVersionPath(bucket, object)
 
-			if isDir {
-				// the attributes at a directory object path may belong to a
-				// file or to a directory that isn't an object
-				fi, err := os.Stat(objpath)
-				if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-					// AWS returns success if the object does not exist
-					return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
-				}
-				if isErrNameTooLong(err) {
-					return nil, s3err.GetKeyTooLongErr(int64(len(object)), 1024)
-				}
-				if err != nil {
-					return nil, fmt.Errorf("stat object: %w", err)
-				}
-				isObj, err := p.isLiveDirObject(fi, bucket, object)
-				if err != nil {
-					return nil, err
-				}
-				if !isObj {
-					// AWS returns success if the object does not exist
-					return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
-				}
+			// the attributes at the object path may belong to the key with
+			// or without the trailing slash, or to a directory that isn't an
+			// object
+			_, err := p.statLiveObject(bucket, object)
+			if errors.Is(err, fs.ErrNotExist) {
+				// AWS returns success if the object does not exist
+				return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
+			}
+			if isErrNameTooLong(err) {
+				return nil, s3err.GetKeyTooLongErr(int64(len(object)), 1024)
+			}
+			if err != nil {
+				return nil, fmt.Errorf("stat object: %w", err)
 			}
 
 			vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
@@ -4997,16 +5075,6 @@ func (p *Posix) DeleteObject(ctx context.Context, input *s3.DeleteObjectInput) (
 				return nil, fmt.Errorf("get obj versionId: %w", err)
 			}
 			if errors.Is(err, meta.ErrNoSuchKey) {
-				// With sidecar, ErrNoSuchKey means "attribute absent" regardless of
-				// whether the data file exists.  If the file is absent the object
-				// does not exist at all → AWS returns success for DeleteObject.
-				// Also handle ENOTDIR: when a key such as "foo/bar" is requested
-				// but "foo" is a regular file (not a directory), the path cannot
-				// contain any object.
-				_, statErr := os.Stat(p.ObjectPath(bucket, object))
-				if errors.Is(statErr, fs.ErrNotExist) || isErrNotDir(statErr) {
-					return &s3.DeleteObjectOutput{VersionId: input.VersionId}, nil
-				}
 				vId = []byte(nullVersionId)
 			}
 
@@ -6836,36 +6904,14 @@ func (p *Posix) GetObjectTagging(ctx context.Context, bucket, object, versionId
 		return nil, err
 	}
 
-	if versionId == "" {
-		_, err = os.Stat(p.ObjectPath(bucket, object))
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if isErrNameTooLong(err) {
-			return nil, s3err.GetAPIError(s3err.ErrKeyTooLong)
-		}
-		if err != nil {
-			return nil, fmt.Errorf("stat object: %w", err)
-		}
+	if versionId != "" && !p.versioningEnabled() {
+		//TODO: Maybe we need to return our custom error here?
+		return nil, s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
 	}
 
-	if versionId != "" {
-		if !p.versioningEnabled() {
-			//TODO: Maybe we need to return our custom error here?
-			return nil, s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
-		}
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
-			return nil, fmt.Errorf("get obj versionId: %w", err)
-		}
-
-		if string(vId) != versionId {
-			bucket = filepath.Join(p.versioningDir, bucket)
-			object = filepath.Join(genObjVersionKey(object), versionId)
-		}
+	bucket, object, err = p.objVersionAttrPath(bucket, object, versionId)
+	if err != nil {
+		return nil, err
 	}
 
 	err = p.ensureNotDeleteMarker(bucket, object, versionId)
@@ -6923,36 +6969,14 @@ func (p *Posix) PutObjectTagging(ctx context.Context, bucket, object, versionId
 		return err
 	}
 
-	if versionId == "" {
-		_, err = os.Stat(p.ObjectPath(bucket, object))
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if isErrNameTooLong(err) {
-			return s3err.GetAPIError(s3err.ErrKeyTooLong)
-		}
-		if err != nil {
-			return fmt.Errorf("stat object: %w", err)
-		}
+	if versionId != "" && !p.versioningEnabled() {
+		//TODO: Maybe we need to return our custom error here?
+		return s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
 	}
 
-	if versionId != "" {
-		if !p.versioningEnabled() {
-			//TODO: Maybe we need to return our custom error here?
-			return s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
-		}
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
-			return fmt.Errorf("get obj versionId: %w", err)
-		}
-
-		if string(vId) != versionId {
-			bucket = filepath.Join(p.versioningDir, bucket)
-			object = filepath.Join(genObjVersionKey(object), versionId)
-		}
+	bucket, object, err = p.objVersionAttrPath(bucket, object, versionId)
+	if err != nil {
+		return err
 	}
 
 	err = p.ensureNotDeleteMarker(bucket, object, versionId)
@@ -6977,12 +7001,7 @@ func (p *Posix) PutObjectTagging(ctx context.Context, bucket, object, versionId
 		return nil
 	}
 
-	b, err := json.Marshal(tags)
-	if err != nil {
-		return fmt.Errorf("marshal tags: %w", err)
-	}
-
-	err = p.meta.StoreAttribute(nil, bucket, object, tagHdr, b)
+	err = p.storeObjectTags(bucket, object, tags)
 	if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
 		if versionId != "" {
 			return s3err.GetNoSuchVersionErr(object, versionId)
@@ -6996,6 +7015,16 @@ func (p *Posix) PutObjectTagging(ctx context.Context, bucket, object, versionId
 	return nil
 }
 
+// storeObjectTags stores tags as the tagging attribute of bucket/object
+func (p *Posix) storeObjectTags(bucket, object string, tags map[string]string) error {
+	b, err := json.Marshal(tags)
+	if err != nil {
+		return fmt.Errorf("marshal tags: %w", err)
+	}
+
+	return p.meta.StoreAttribute(nil, bucket, object, tagHdr, b)
+}
+
 func (p *Posix) DeleteObjectTagging(ctx context.Context, bucket, object, versionId string) error {
 	if !p.isBucketValid(bucket) {
 		return s3err.GetBucketErr(s3err.ErrInvalidBucketName, bucket)
@@ -7317,7 +7346,7 @@ func (p *Posix) PutObjectLegalHold(ctx context.Context, bucket, object, versionI
 	if !p.isBucketValid(bucket) {
 		return s3err.GetBucketErr(s3err.ErrInvalidBucketName, bucket)
 	}
-	err = p.doesBucketAndObjectExist(bucket, object)
+	err = p.doesBucketAndObjectExist(bucket, object, versionId)
 	if err != nil {
 		return err
 	}
@@ -7342,17 +7371,9 @@ func (p *Posix) PutObjectLegalHold(ctx context.Context, bucket, object, versionI
 			//TODO: Maybe we need to return our custom error here?
 			return s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
 		}
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
-			return fmt.Errorf("get obj versionId: %w", err)
-		}
-
-		if string(vId) != versionId {
-			bucket = filepath.Join(p.versioningDir, bucket)
-			object = filepath.Join(genObjVersionKey(object), versionId)
+		bucket, object, err = p.objVersionAttrPath(bucket, object, versionId)
+		if err != nil {
+			return err
 		}
 	}
 
@@ -7385,7 +7406,7 @@ func (p *Posix) GetObjectLegalHold(ctx context.Context, bucket, object, versionI
 	if !p.isBucketValid(bucket) {
 		return nil, s3err.GetBucketErr(s3err.ErrInvalidBucketName, bucket)
 	}
-	err = p.doesBucketAndObjectExist(bucket, object)
+	err = p.doesBucketAndObjectExist(bucket, object, versionId)
 	if err != nil {
 		return nil, err
 	}
@@ -7403,17 +7424,9 @@ func (p *Posix) GetObjectLegalHold(ctx context.Context, bucket, object, versionI
 			//TODO: Maybe we need to return our custom error here?
 			return nil, s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
 		}
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
-			return nil, fmt.Errorf("get obj versionId: %w", err)
-		}
-
-		if string(vId) != versionId {
-			bucket = filepath.Join(p.versioningDir, bucket)
-			object = filepath.Join(genObjVersionKey(object), versionId)
+		bucket, object, err = p.objVersionAttrPath(bucket, object, versionId)
+		if err != nil {
+			return nil, err
 		}
 	}
 
@@ -7451,7 +7464,7 @@ func (p *Posix) PutObjectRetention(ctx context.Context, bucket, object, versionI
 	if !p.isBucketValid(bucket) {
 		return s3err.GetBucketErr(s3err.ErrInvalidBucketName, bucket)
 	}
-	err = p.doesBucketAndObjectExist(bucket, object)
+	err = p.doesBucketAndObjectExist(bucket, object, versionId)
 	if err != nil {
 		return err
 	}
@@ -7469,17 +7482,9 @@ func (p *Posix) PutObjectRetention(ctx context.Context, bucket, object, versionI
 			//TODO: Maybe we need to return our custom error here?
 			return s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
 		}
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
-			return fmt.Errorf("get obj versionId: %w", err)
-		}
-
-		if string(vId) != versionId {
-			bucket = filepath.Join(p.versioningDir, bucket)
-			object = filepath.Join(genObjVersionKey(object), versionId)
+		bucket, object, err = p.objVersionAttrPath(bucket, object, versionId)
+		if err != nil {
+			return err
 		}
 	}
 
@@ -7506,7 +7511,7 @@ func (p *Posix) GetObjectRetention(ctx context.Context, bucket, object, versionI
 	if !p.isBucketValid(bucket) {
 		return nil, s3err.GetBucketErr(s3err.ErrInvalidBucketName, bucket)
 	}
-	err = p.doesBucketAndObjectExist(bucket, object)
+	err = p.doesBucketAndObjectExist(bucket, object, versionId)
 	if err != nil {
 		return nil, err
 	}
@@ -7524,17 +7529,9 @@ func (p *Posix) GetObjectRetention(ctx context.Context, bucket, object, versionI
 			//TODO: Maybe we need to return our custom error here?
 			return nil, s3err.GetInvalidArgumentErr(s3err.InvalidArgVersionId, versionId)
 		}
-		vId, err := p.meta.RetrieveAttribute(nil, bucket, object, versionIdKey)
-		if errors.Is(err, fs.ErrNotExist) || isErrNotDir(err) {
-			return nil, s3err.GetAPIError(s3err.ErrNoSuchKey)
-		}
-		if err != nil && !errors.Is(err, meta.ErrNoSuchKey) {
-			return nil, fmt.Errorf("get obj versionId: %w", err)
-		}
-
-		if string(vId) != versionId {
-			bucket = filepath.Join(p.versioningDir, bucket)
-			object = filepath.Join(genObjVersionKey(object), versionId)
+		bucket, object, err = p.objVersionAttrPath(bucket, object, versionId)
+		if err != nil {
+			return nil, err
 		}
 	}
 

tests/integration/group-tests.go

@@ -990,6 +990,12 @@ func TestPosix(ts *TestState) {
 	ts.Run(DeleteObject_name_too_long)
 	ts.Run(CopyObject_overwrite_same_dir_object)
 	ts.Run(CopyObject_overwrite_same_file_object)
+	ts.Run(CompleteMultipartUpload_overwrite_dir_obj)
+	if ts.conf.versioningEnabled {
+		ts.Run(CompleteMultipartUpload_overwrite_dir_obj_delete_marker)
+	}
+	ts.Run(ObjectTagging_trailing_slash_counterpart)
+	ts.Run(ObjectLock_trailing_slash_counterpart)
 	ts.Run(DeleteObject_directory_not_empty)
 	if !ts.conf.windowsTests {
 		ts.Run(PutObject_race_with_delete)
@@ -1094,6 +1100,9 @@ func TestScoutfs(ts *TestState) {
 	ts.Run(DeleteObject_name_too_long)
 	ts.Run(CopyObject_overwrite_same_dir_object)
 	ts.Run(CopyObject_overwrite_same_file_object)
+	ts.Run(CompleteMultipartUpload_overwrite_dir_obj)
+	ts.Run(ObjectTagging_trailing_slash_counterpart)
+	ts.Run(ObjectLock_trailing_slash_counterpart)
 	ts.Run(DeleteObject_directory_not_empty)
 }
 
@@ -1977,6 +1986,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_DeleteObjectTagging_invalid_versionId)
 	ts.Run(Versioning_DeleteObjectTagging_non_existing_object_version)
 	ts.Run(Versioning_PutGetDeleteObjectTagging_success)
+	ts.Run(Versioning_ObjectTagging_trailing_slash_counterpart)
 	// GetObjectAttributes action
 	ts.Run(Versioning_GetObjectAttributes_invalid_versionId)
 	ts.Run(Versioning_GetObjectAttributes_object_version)
@@ -1987,6 +1997,7 @@ func TestVersioning(ts *TestState) {
 	ts.Run(Versioning_DeleteObject_dir_object_latest_version)
 	ts.Run(Versioning_DeleteObject_non_existing_object)
 	ts.Run(Versioning_DeleteObject_implicit_dir)
+	ts.Run(Versioning_DeleteObject_trailing_slash_counterpart)
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_DeleteObject_delete_a_delete_marker)
 		ts.Run(Versioning_DeleteObject_dir_object_with_children)
@@ -2047,6 +2058,8 @@ func TestVersioning(ts *TestState) {
 	if !ts.conf.windowsTests {
 		ts.Run(Versioning_WORM_remove_delete_marker_under_bucket_default_retention)
 	}
+	ts.Run(Versioning_WORM_trailing_slash_counterpart)
+	ts.Run(Versioning_WORM_null_version_locked_with_legal_hold)
 	// Concurrent requests
 	// Versioninig_concurrent_upload_object
 	ts.Run(Versioning_AccessControl_GetObjectVersion)
@@ -2989,6 +3002,10 @@ func GetIntTests() IntTests {
 		"DeleteObject_name_too_long":                                                       DeleteObject_name_too_long,
 		"CopyObject_overwrite_same_dir_object":                                             CopyObject_overwrite_same_dir_object,
 		"CopyObject_overwrite_same_file_object":                                            CopyObject_overwrite_same_file_object,
+		"CompleteMultipartUpload_overwrite_dir_obj":                                        CompleteMultipartUpload_overwrite_dir_obj,
+		"CompleteMultipartUpload_overwrite_dir_obj_delete_marker":                          CompleteMultipartUpload_overwrite_dir_obj_delete_marker,
+		"ObjectTagging_trailing_slash_counterpart":                                         ObjectTagging_trailing_slash_counterpart,
+		"ObjectLock_trailing_slash_counterpart":                                            ObjectLock_trailing_slash_counterpart,
 		"DeleteObject_non_existing_dir_object":                                             DeleteObject_non_existing_dir_object,
 		"DeleteObject_directory_object":                                                    DeleteObject_directory_object,
 		"DeleteObject_success":                                                             DeleteObject_success,
@@ -3499,6 +3516,7 @@ func GetIntTests() IntTests {
 		"Versioning_DeleteObjectTagging_invalid_versionId":                                 Versioning_DeleteObjectTagging_invalid_versionId,
 		"Versioning_DeleteObjectTagging_non_existing_object_version":                       Versioning_DeleteObjectTagging_non_existing_object_version,
 		"Versioning_PutGetDeleteObjectTagging_success":                                     Versioning_PutGetDeleteObjectTagging_success,
+		"Versioning_ObjectTagging_trailing_slash_counterpart":                              Versioning_ObjectTagging_trailing_slash_counterpart,
 		"Versioning_GetObjectAttributes_invalid_versionId":                                 Versioning_GetObjectAttributes_invalid_versionId,
 		"Versioning_GetObjectAttributes_object_version":                                    Versioning_GetObjectAttributes_object_version,
 		"Versioning_GetObjectAttributes_delete_marker":                                     Versioning_GetObjectAttributes_delete_marker,
@@ -3507,6 +3525,7 @@ func GetIntTests() IntTests {
 		"Versioning_DeleteObject_dir_object_latest_version":                                Versioning_DeleteObject_dir_object_latest_version,
 		"Versioning_DeleteObject_non_existing_object":                                      Versioning_DeleteObject_non_existing_object,
 		"Versioning_DeleteObject_implicit_dir":                                             Versioning_DeleteObject_implicit_dir,
+		"Versioning_DeleteObject_trailing_slash_counterpart":                               Versioning_DeleteObject_trailing_slash_counterpart,
 		"Versioning_DeleteObject_delete_a_delete_marker":                                   Versioning_DeleteObject_delete_a_delete_marker,
 		"Versioning_DeleteObject_dir_object_with_children":                                 Versioning_DeleteObject_dir_object_with_children,
 		"Versioning_Delete_null_versionId_object":                                          Versioning_Delete_null_versionId_object,
@@ -3557,6 +3576,8 @@ func GetIntTests() IntTests {
 		"Versioning_WORM_CopyObject_overwrite_locked_object":                               Versioning_WORM_CopyObject_overwrite_locked_object,
 		"Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object":                  Versioning_WORM_CompleteMultipartUpload_overwrite_locked_object,
 		"Versioning_WORM_remove_delete_marker_under_bucket_default_retention":              Versioning_WORM_remove_delete_marker_under_bucket_default_retention,
+		"Versioning_WORM_trailing_slash_counterpart":                                       Versioning_WORM_trailing_slash_counterpart,
+		"Versioning_WORM_null_version_locked_with_legal_hold":                              Versioning_WORM_null_version_locked_with_legal_hold,
 		"Versioning_AccessControl_GetObjectVersion":                                        Versioning_AccessControl_GetObjectVersion,
 		"Versioning_AccessControl_HeadObjectVersion":                                       Versioning_AccessControl_HeadObjectVersion,
 		"Versioning_AccessControl_object_tagging_policy":                                   Versioning_AccessControl_object_tagging_policy,

tests/integration/posix.go

@@ -17,6 +17,7 @@ package integration
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -222,6 +223,255 @@ func CopyObject_overwrite_same_file_object(s *S3Conf) error {
 	})
 }
 
+func CompleteMultipartUpload_overwrite_dir_obj(s *S3Conf) error {
+	testName := "CompleteMultipartUpload_overwrite_dir_obj"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dir, obj := "foo/", "foo"
+		_, err := putObjects(s3client, []string{dir}, bucket)
+		if err != nil {
+			return err
+		}
+
+		mp, err := createMp(s3client, bucket, obj)
+		if err != nil {
+			return err
+		}
+
+		parts, _, err := uploadParts(s3client, 100, 1, bucket, obj, *mp.UploadId)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+			Bucket:   &bucket,
+			Key:      &obj,
+			UploadId: mp.UploadId,
+			MultipartUpload: &types.CompletedMultipartUpload{
+				Parts: []types.CompletedPart{
+					{ETag: parts[0].ETag, PartNumber: parts[0].PartNumber},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)); err != nil {
+			return err
+		}
+
+		// the directory object isn't taken for the object of a completed upload
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+			Bucket:   &bucket,
+			Key:      &obj,
+			UploadId: getPtr("non-existing-upload-id"),
+			MultipartUpload: &types.CompletedMultipartUpload{
+				Parts: []types.CompletedPart{
+					{ETag: parts[0].ETag, PartNumber: parts[0].PartNumber},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrNoSuchUpload)); err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &dir,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		// the failed upload can still be completed or aborted
+		return checkAndAbortUpload(s3client, bucket, obj, *mp.UploadId)
+	})
+}
+
+func CompleteMultipartUpload_overwrite_dir_obj_delete_marker(s *S3Conf) error {
+	testName := "CompleteMultipartUpload_overwrite_dir_obj_delete_marker"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		dir, obj := "foo/", "foo"
+		versions, err := createObjVersions(s3client, bucket, dir, 1)
+		if err != nil {
+			return err
+		}
+		versions[0].IsLatest = getPtr(false)
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: &bucket,
+			Key:    &dir,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+		delMarkers := []types.DeleteMarkerEntry{
+			{Key: &dir, VersionId: out.VersionId, IsLatest: getPtr(true)},
+		}
+
+		mp, err := createMp(s3client, bucket, obj)
+		if err != nil {
+			return err
+		}
+
+		parts, _, err := uploadParts(s3client, 100, 1, bucket, obj, *mp.UploadId)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.CompleteMultipartUpload(ctx, &s3.CompleteMultipartUploadInput{
+			Bucket:   &bucket,
+			Key:      &obj,
+			UploadId: mp.UploadId,
+			MultipartUpload: &types.CompletedMultipartUpload{
+				Parts: []types.CompletedPart{
+					{ETag: parts[0].ETag, PartNumber: parts[0].PartNumber},
+				},
+			},
+		})
+		cancel()
+		if err := checkApiErr(err, s3err.GetAPIError(s3err.ErrExistingObjectIsDirectory)); err != nil {
+			return err
+		}
+
+		// the delete marker of the directory object is left in place
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(versions, res.Versions) {
+			return fmt.Errorf("expected the versions to be %v, instead got %v",
+				versions, res.Versions)
+		}
+		if !compareDelMarkers(delMarkers, res.DeleteMarkers) {
+			return fmt.Errorf("expected the delete markers to be %v, instead got %v",
+				delMarkers, res.DeleteMarkers)
+		}
+
+		ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: &bucket,
+			Key:    &dir,
+		})
+		cancel()
+		if err := checkSdkApiErr(err, "NotFound"); err != nil {
+			return err
+		}
+
+		return checkAndAbortUpload(s3client, bucket, obj, *mp.UploadId)
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
+func ObjectTagging_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "ObjectTagging_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagSet := []types.Tag{{Key: getPtr("key"), Value: getPtr("value")}}
+
+		// a key and the same key with a trailing slash have one path: the
+		// object of one of them isn't an object of the other
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			_, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket:  &bucket,
+				Key:     &obj,
+				Tagging: getPtr("key=value"),
+			}, s3client)
+			if err != nil {
+				return err
+			}
+
+			err = checkObjectTaggingErr(s3client, bucket, other, "", s3err.GetAPIError(s3err.ErrNoSuchKey))
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+			if !areTagsSame(res.TagSet, tagSet) {
+				return fmt.Errorf("%v: expected the tag set to be %v, instead got %v",
+					obj, tagSet, res.TagSet)
+			}
+		}
+
+		// neither key names the parent directory of an object
+		_, err := putObjects(s3client, []string{"my-parent/obj"}, bucket)
+		if err != nil {
+			return err
+		}
+		for _, key := range []string{"my-parent/", "my-parent"} {
+			err := checkObjectTaggingErr(s3client, bucket, key, "", s3err.GetAPIError(s3err.ErrNoSuchKey))
+			if err != nil {
+				return fmt.Errorf("%v: %w", key, err)
+			}
+		}
+
+		return nil
+	})
+}
+
+func ObjectLock_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "ObjectLock_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		rDate := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
+		lockedObjs := []objToDelete{}
+
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			_, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket:                    &bucket,
+				Key:                       &obj,
+				ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+				ObjectLockMode:            types.ObjectLockModeGovernance,
+				ObjectLockRetainUntilDate: &rDate,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+			lockedObjs = append(lockedObjs, objToDelete{key: obj, removeOnlyLeglHold: true})
+
+			err = checkObjectLockErr(s3client, bucket, other, "", s3err.GetAPIError(s3err.ErrNoSuchKey))
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			// the lock of the object doesn't protect the other key
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: &bucket,
+				Key:    &other,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			err = checkObjectLock(s3client, bucket, obj, "", rDate)
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+		}
+
+		return cleanupLockedObjects(s3client, bucket, lockedObjs)
+	}, withLock())
+}
+
 // PutObject_race_with_delete tests the race between PutObject and DeleteObject
 // in the same subdirectory.
 // One goroutine sequentially puts "race-dir/0.txt" … "race-dir/N-1.txt".

tests/integration/utils.go

@@ -3989,3 +3989,170 @@ func checkDeleteObjectsErrsInOrder(got []types.Error, want []keyDenial) error {
 	}
 	return nil
 }
+
+// checkObjectTaggingErr checks that getting, putting and deleting the
+// tagging of the version versionId of key fail with expected. An empty
+// versionId selects the current version.
+func checkObjectTaggingErr(client *s3.Client, bucket, key, versionId string, expected s3err.S3Error) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err := client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("get object tagging: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.PutObjectTagging(ctx, &s3.PutObjectTaggingInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+		Tagging: &types.Tagging{
+			TagSet: []types.Tag{{Key: getPtr("other-key"), Value: getPtr("other-value")}},
+		},
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("put object tagging: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.DeleteObjectTagging(ctx, &s3.DeleteObjectTaggingInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("delete object tagging: %w", err)
+	}
+
+	return nil
+}
+
+// checkObjectLockErr checks that getting and putting the legal hold and the
+// retention of the version versionId of key fail with expected. An empty
+// versionId selects the current version.
+func checkObjectLockErr(client *s3.Client, bucket, key, versionId string, expected s3err.S3Error) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	_, err := client.GetObjectLegalHold(ctx, &s3.GetObjectLegalHoldInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("get object legal hold: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.PutObjectLegalHold(ctx, &s3.PutObjectLegalHoldInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+		LegalHold: &types.ObjectLockLegalHold{
+			Status: types.ObjectLockLegalHoldStatusOff,
+		},
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("put object legal hold: %w", err)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.GetObjectRetention(ctx, &s3.GetObjectRetentionInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("get object retention: %w", err)
+	}
+
+	rDate := time.Now().Add(time.Hour * 2)
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.PutObjectRetention(ctx, &s3.PutObjectRetentionInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+		Retention: &types.ObjectLockRetention{
+			Mode:            types.ObjectLockRetentionModeGovernance,
+			RetainUntilDate: &rDate,
+		},
+	})
+	cancel()
+	if err := checkApiErr(err, expected); err != nil {
+		return fmt.Errorf("put object retention: %w", err)
+	}
+
+	return nil
+}
+
+// checkObjectLock checks that the version versionId of key is under legal
+// hold and has a governance retention until rDate. An empty versionId
+// selects the current version.
+func checkObjectLock(client *s3.Client, bucket, key, versionId string, rDate time.Time) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	lHold, err := client.GetObjectLegalHold(ctx, &s3.GetObjectLegalHoldInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	if lHold.LegalHold == nil || lHold.LegalHold.Status != types.ObjectLockLegalHoldStatusOn {
+		return fmt.Errorf("expected the legal hold status to be %q, instead got %v",
+			types.ObjectLockLegalHoldStatusOn, lHold.LegalHold)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	ret, err := client.GetObjectRetention(ctx, &s3.GetObjectRetentionInput{
+		Bucket:    &bucket,
+		Key:       &key,
+		VersionId: getNonEmptyPtr(versionId),
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	if ret.Retention == nil || ret.Retention.Mode != types.ObjectLockRetentionModeGovernance ||
+		ret.Retention.RetainUntilDate == nil || ret.Retention.RetainUntilDate.Unix() != rDate.Unix() {
+		return fmt.Errorf("expected a %q retention until %v, instead got %+v",
+			types.ObjectLockRetentionModeGovernance, rDate.Format(time.RFC3339), ret.Retention)
+	}
+
+	return nil
+}
+
+// checkAndAbortUpload checks that the upload uploadId of key is the only
+// multipart upload in the bucket and aborts it
+func checkAndAbortUpload(client *s3.Client, bucket, key, uploadId string) error {
+	ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+	res, err := client.ListMultipartUploads(ctx, &s3.ListMultipartUploadsInput{
+		Bucket: &bucket,
+	})
+	cancel()
+	if err != nil {
+		return err
+	}
+	if len(res.Uploads) != 1 || getString(res.Uploads[0].Key) != key ||
+		getString(res.Uploads[0].UploadId) != uploadId {
+		return fmt.Errorf("expected the upload %v of %v to be listed, instead got %+v",
+			uploadId, key, res.Uploads)
+	}
+
+	ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+	_, err = client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{
+		Bucket:   &bucket,
+		Key:      &key,
+		UploadId: &uploadId,
+	})
+	cancel()
+	return err
+}

tests/integration/versioning.go

@@ -2016,6 +2016,83 @@ func Versioning_DeleteObject_dir_object_with_children(s *S3Conf) error {
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
 
+func Versioning_DeleteObject_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "Versioning_DeleteObject_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		expected := []types.ObjectVersion{}
+		// deleting the key with or without a trailing slash doesn't delete
+		// the object or any of its versions
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			versions, err := createObjVersions(s3client, bucket, obj, 2)
+			if err != nil {
+				return err
+			}
+			expected = append(expected, versions...)
+
+			for _, versionId := range []*string{versions[0].VersionId, versions[1].VersionId} {
+				ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+				out, err := s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+					Bucket:    &bucket,
+					Key:       &other,
+					VersionId: versionId,
+				})
+				cancel()
+				if err != nil {
+					return fmt.Errorf("%v: %w", other, err)
+				}
+				if getString(out.VersionId) != *versionId {
+					return fmt.Errorf("%v: expected the versionId to be %v, instead got %v",
+						other, *versionId, getString(out.VersionId))
+				}
+				if out.DeleteMarker != nil && *out.DeleteMarker {
+					return fmt.Errorf("%v: expected the response DeleteMarker to be false", other)
+				}
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: &bucket,
+				Key:    &other,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.HeadObject(ctx, &s3.HeadObjectInput{
+				Bucket: &bucket,
+				Key:    &obj,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+			if getString(res.VersionId) != getString(versions[0].VersionId) {
+				return fmt.Errorf("%v: expected the versionId to be %v, instead got %v",
+					obj, getString(versions[0].VersionId), getString(res.VersionId))
+			}
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectVersions(ctx, &s3.ListObjectVersionsInput{
+			Bucket: &bucket,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if !compareVersions(expected, res.Versions) {
+			return fmt.Errorf("expected the versions to be %v, instead got %v",
+				expected, res.Versions)
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}
+
 func Versioning_Delete_null_versionId_object(s *S3Conf) error {
 	testName := "Versioning_Delete_null_versionId_object"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -4196,6 +4273,139 @@ func Versioning_WORM_remove_delete_marker_under_bucket_default_retention(s *S3Co
 	}, withLock())
 }
 
+func Versioning_WORM_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "Versioning_WORM_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		rDate := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
+		lockedObjs := []objToDelete{}
+
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			res, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket:                    &bucket,
+				Key:                       &obj,
+				ObjectLockLegalHoldStatus: types.ObjectLockLegalHoldStatusOn,
+				ObjectLockMode:            types.ObjectLockModeGovernance,
+				ObjectLockRetainUntilDate: &rDate,
+			}, s3client)
+			if err != nil {
+				return err
+			}
+			versionId := getString(res.res.VersionId)
+			lockedObjs = append(lockedObjs, objToDelete{
+				key:                obj,
+				versionId:          versionId,
+				removeOnlyLeglHold: true,
+			})
+
+			// the version belongs to the object, not to the other key
+			err = checkObjectLockErr(s3client, bucket, other, versionId, s3err.GetAPIError(s3err.ErrNoSuchVersion))
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket:    &bucket,
+				Key:       &other,
+				VersionId: &versionId,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			err = checkObjectLock(s3client, bucket, obj, versionId, rDate)
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+		}
+
+		return cleanupLockedObjects(s3client, bucket, lockedObjs)
+	}, withLock())
+}
+
+func Versioning_WORM_null_version_locked_with_legal_hold(s *S3Conf) error {
+	testName := "Versioning_WORM_null_version_locked_with_legal_hold"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		objs := []string{"my-obj", "my-dir/"}
+		// the objects are put before versioning is enabled: their
+		// current versions are the null versions
+		_, err := putObjects(s3client, objs, bucket)
+		if err != nil {
+			return err
+		}
+
+		err = putBucketVersioningStatus(s3client, bucket, types.BucketVersioningStatusEnabled)
+		if err != nil {
+			return err
+		}
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		_, err = s3client.PutObjectLockConfiguration(ctx, &s3.PutObjectLockConfigurationInput{
+			Bucket: &bucket,
+			ObjectLockConfiguration: &types.ObjectLockConfiguration{
+				ObjectLockEnabled: types.ObjectLockEnabledEnabled,
+			},
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		lockedObjs := []objToDelete{}
+		err = forEachKey(objs, func(obj string) error {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			_, err := s3client.PutObjectLegalHold(ctx, &s3.PutObjectLegalHoldInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: getPtr(nullVersionId),
+				LegalHold: &types.ObjectLockLegalHold{
+					Status: types.ObjectLockLegalHoldStatusOn,
+				},
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+			lockedObjs = append(lockedObjs, objToDelete{
+				key:                obj,
+				versionId:          nullVersionId,
+				removeOnlyLeglHold: true,
+			})
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			res, err := s3client.GetObjectLegalHold(ctx, &s3.GetObjectLegalHoldInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: getPtr(nullVersionId),
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+			if res.LegalHold == nil || res.LegalHold.Status != types.ObjectLockLegalHoldStatusOn {
+				return fmt.Errorf("expected the legal hold status to be %q, instead got %v",
+					types.ObjectLockLegalHoldStatusOn, res.LegalHold)
+			}
+
+			ctx, cancel = context.WithTimeout(context.Background(), shortTimeout)
+			_, err = s3client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: getPtr(nullVersionId),
+			})
+			cancel()
+			return checkApiErr(err, s3err.GetAPIError(s3err.ErrObjectLocked))
+		})
+		if err != nil {
+			return err
+		}
+
+		return cleanupLockedObjects(s3client, bucket, lockedObjs)
+	})
+}
+
 func Versioning_AccessControl_GetObjectVersion(s *S3Conf) error {
 	testName := "Versioning_AccessControl_GetObjectVersion"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5032,3 +5242,46 @@ func Versioning_PutGetDeleteObjectTagging_success(s *S3Conf) error {
 		})
 	}, withVersioning(types.BucketVersioningStatusEnabled))
 }
+
+func Versioning_ObjectTagging_trailing_slash_counterpart(s *S3Conf) error {
+	testName := "Versioning_ObjectTagging_trailing_slash_counterpart"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		tagSet := []types.Tag{{Key: getPtr("key"), Value: getPtr("value")}}
+
+		for _, keys := range [][2]string{{"my-dir/", "my-dir"}, {"my-obj", "my-obj/"}} {
+			obj, other := keys[0], keys[1]
+			res, err := putObjectWithData(objDataLen(obj, 10), &s3.PutObjectInput{
+				Bucket:  &bucket,
+				Key:     &obj,
+				Tagging: getPtr("key=value"),
+			}, s3client)
+			if err != nil {
+				return err
+			}
+			versionId := getString(res.res.VersionId)
+
+			// the version belongs to the object, not to the other key
+			err = checkObjectTaggingErr(s3client, bucket, other, versionId, s3err.GetAPIError(s3err.ErrNoSuchVersion))
+			if err != nil {
+				return fmt.Errorf("%v: %w", other, err)
+			}
+
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.GetObjectTagging(ctx, &s3.GetObjectTaggingInput{
+				Bucket:    &bucket,
+				Key:       &obj,
+				VersionId: &versionId,
+			})
+			cancel()
+			if err != nil {
+				return fmt.Errorf("%v: %w", obj, err)
+			}
+			if !areTagsSame(out.TagSet, tagSet) {
+				return fmt.Errorf("%v: expected the tag set to be %v, instead got %v",
+					obj, tagSet, out.TagSet)
+			}
+		}
+
+		return nil
+	}, withVersioning(types.BucketVersioningStatusEnabled))
+}