s3: cover the directory marker key with object lock (#10988)

* s3: enforce object lock when deleting a directory marker

The key "dir/" is deleted the unversioned way, ahead of the branches
that enforce Object Lock, so a principal with plain delete permission
could remove a key the gateway was reporting as COMPLIANCE-retained --
retention set through PutObjectRetention is stored on the directory
entry and served back by GetObjectRetention, only the delete ignored it.

The same path also takes any key ending in "/" regardless of size, while
a PUT only makes a marker of one up to 1KiB. A larger one is a genuine
versioned object, and deleting it here dropped its whole history after
the versioned delete of the same key had been refused.

Enforce in the marker delete itself, so the single, versioned and
multi-object delete paths are all covered.

* s3: apply object lock headers on a directory marker PUT

The trailing-slash branch runs before the versioning and Object Lock
handling, so it accepted x-amz-object-lock-* headers and stored none of
them: a bucket owner could believe a key was retained while nothing
recorded it, and an invalid mode or a past retention date that a regular
key rejects came back 200 here.

Validate the headers the way the regular path does, store what they ask
for beside the owner the same callback already sets, and refuse to
replace a key that is already retained.

* s3: check every version a marker delete would remove

The marker delete clears any history under the key in one recursive
removal, while the lock check ahead of it resolves the latest version
only. A version retained under an unretained one was taken with the
rest, so enforce against each version the removal covers.

* test: pin the marker lock refusals to AccessDenied

A bare require.Error passes on any failure, including one that has
nothing to do with the lock. Assert the code, the key the batch delete
reports, and that the marker survives each refusal.

* s3: check the history entries a version list leaves out

The version list skips an entry without a version id, while the removal
takes it with the rest, so an entry an older build left unnamed escaped
the check. Walk the history directly instead, and refuse when an unnamed
entry is still under a retention or a legal hold of its own.

* s3: let a governance bypass reach an unnamed history entry

The unnamed branch refused every active retention, so a caller allowed
to bypass governance could not clear one, which the named path lets
through. Refuse a legal hold and compliance mode as before, and take the
bypass into account for governance.

* s3: keep the object lock decision in one place

The unnamed history entry had to repeat the retention and legal hold
rules inline because the enforcement helper only takes a key to look up.
Split the part that judges an entry out of it and call that from both.

* s3: guard a marker PUT on the entry it replaces

The overwrite check resolved the key's latest version, but mkdir builds
a fresh entry for the marker itself, dropping the lock metadata the old
one carried. Once the key had a history, an unlocked version answered
for a retained marker and a plain PUT replaced it. Judge the entry the
write is about to replace instead; a versioned write of the same key
still adds a version, which is its own to allow.

* s3: guard a marker delete on the entry it removes

The check ran against the key rather than the entry, so once the key had
a history it answered with a version and the retention recorded on the
marker itself went unseen. Judge the entry that is about to be removed,
the same way the PUT side now does; the versions under it are still
covered by the walk that follows.

* s3: take the object write lock for a marker PUT

The overwrite check read the entry that the mkdir after it replaces, so
two marker PUTs could both pass while one was still unlocked. The marker
delete already runs under this lock; hold it across the check and the
mkdir so the entry cannot change in between, and so the two paths are
serialized against each other.

test/s3/retention/s3_object_lock_directory_marker_test.go

@@ -0,0 +1,231 @@
+package retention
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"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/aws/smithy-go"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func requireAccessDenied(t *testing.T, err error, msg string) {
+	t.Helper()
+	require.Error(t, err, msg)
+	var apiErr smithy.APIError
+	require.True(t, errors.As(err, &apiErr), "expected an API error, got %T", err)
+	assert.Equal(t, "AccessDenied", apiErr.ErrorCode(), msg)
+}
+
+// A key ending in "/" is stored as the filer directory rather than as an object
+// beside it, and is deleted the unversioned way. Object Lock still covers it: the
+// gateway lists it as an object and serves retention set on it.
+func TestObjectLockDirectoryMarker(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+
+	createBucketWithObjectLock(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+
+	retainUntil := time.Now().Add(24 * time.Hour)
+
+	t.Run("retention headers are honored, not dropped", func(t *testing.T) {
+		key := "records/evidence/"
+		_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket:                    aws.String(bucketName),
+			Key:                       aws.String(key),
+			Body:                      strings.NewReader("marker"),
+			ObjectLockMode:            types.ObjectLockModeCompliance,
+			ObjectLockRetainUntilDate: aws.Time(retainUntil),
+		})
+		require.NoError(t, err)
+
+		_, err = client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		requireAccessDenied(t, err, "a retained marker must not be deletable")
+
+		_, err = client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		assert.NoError(t, err, "the marker must still be there after the refused delete")
+	})
+
+	t.Run("retention set through PutObjectRetention is honored", func(t *testing.T) {
+		key := "records/ledger/"
+		_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Body:   strings.NewReader("marker"),
+		})
+		require.NoError(t, err)
+
+		_, err = client.PutObjectRetention(context.TODO(), &s3.PutObjectRetentionInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Retention: &types.ObjectLockRetention{
+				Mode:            types.ObjectLockRetentionModeCompliance,
+				RetainUntilDate: aws.Time(retainUntil),
+			},
+		})
+		require.NoError(t, err)
+
+		_, err = client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		requireAccessDenied(t, err, "retention the gateway serves back must also block the delete")
+
+		_, err = client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		assert.NoError(t, err, "the marker must still be there after the refused delete")
+	})
+
+	t.Run("multi-object delete is refused too", func(t *testing.T) {
+		key := "records/batch/"
+		_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket:                    aws.String(bucketName),
+			Key:                       aws.String(key),
+			Body:                      strings.NewReader("marker"),
+			ObjectLockMode:            types.ObjectLockModeCompliance,
+			ObjectLockRetainUntilDate: aws.Time(retainUntil),
+		})
+		require.NoError(t, err)
+
+		resp, err := client.DeleteObjects(context.TODO(), &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucketName),
+			Delete: &types.Delete{Objects: []types.ObjectIdentifier{{Key: aws.String(key)}}},
+		})
+		require.NoError(t, err)
+		assert.Empty(t, resp.Deleted, "a retained marker must not be reported deleted")
+		require.Len(t, resp.Errors, 1)
+		assert.Equal(t, key, aws.ToString(resp.Errors[0].Key))
+		assert.Equal(t, "AccessDenied", aws.ToString(resp.Errors[0].Code))
+
+		_, err = client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		assert.NoError(t, err, "the marker must survive the batch delete")
+	})
+
+	t.Run("invalid lock headers are rejected, as on a regular key", func(t *testing.T) {
+		_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket:                    aws.String(bucketName),
+			Key:                       aws.String("records/bad-mode/"),
+			Body:                      strings.NewReader("marker"),
+			ObjectLockMode:            "INVALID_MODE",
+			ObjectLockRetainUntilDate: aws.Time(retainUntil),
+		})
+		require.Error(t, err)
+
+		_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket:         aws.String(bucketName),
+			Key:            aws.String("records/no-date/"),
+			Body:           strings.NewReader("marker"),
+			ObjectLockMode: types.ObjectLockModeGovernance,
+		})
+		require.Error(t, err)
+	})
+
+	// Past 1KiB a trailing-slash key is a real versioned object, and the delete
+	// takes the whole history at once, so an older retained version has to block
+	// it even when the version on top carries no retention of its own.
+	t.Run("a retained version under an unretained one still blocks", func(t *testing.T) {
+		key := "records/history/"
+		body := strings.Repeat("x", 2048)
+
+		first, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket:                    aws.String(bucketName),
+			Key:                       aws.String(key),
+			Body:                      strings.NewReader(body),
+			ObjectLockMode:            types.ObjectLockModeCompliance,
+			ObjectLockRetainUntilDate: aws.Time(retainUntil),
+		})
+		require.NoError(t, err)
+		require.NotNil(t, first.VersionId)
+
+		_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Body:   strings.NewReader(body),
+		})
+		require.NoError(t, err)
+
+		_, err = client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		requireAccessDenied(t, err, "the retained version underneath must block the delete")
+
+		_, err = client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+			Bucket:    aws.String(bucketName),
+			Key:       aws.String(key),
+			VersionId: first.VersionId,
+		})
+		assert.NoError(t, err, "the retained version must survive")
+	})
+
+	// mkdir replaces the marker entry outright, taking its lock metadata with it,
+	// so a plain PUT over a retained marker has to be refused - including once the
+	// key has grown a version history that the latest-version lookup would find
+	// instead of the marker.
+	t.Run("a plain PUT cannot replace a retained marker", func(t *testing.T) {
+		key := "records/overwrite/"
+		_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket:                    aws.String(bucketName),
+			Key:                       aws.String(key),
+			Body:                      strings.NewReader("marker"),
+			ObjectLockMode:            types.ObjectLockModeCompliance,
+			ObjectLockRetainUntilDate: aws.Time(retainUntil),
+		})
+		require.NoError(t, err)
+
+		_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Body:   strings.NewReader("replacement"),
+		})
+		requireAccessDenied(t, err, "a retained marker must not be replaceable")
+
+		_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Body:   strings.NewReader(strings.Repeat("x", 2048)),
+		})
+		require.NoError(t, err, "a versioned write of the same key adds a version")
+
+		_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Body:   strings.NewReader("replacement"),
+		})
+		requireAccessDenied(t, err, "the history must not hide the marker's own lock")
+	})
+
+	t.Run("an unretained marker still deletes", func(t *testing.T) {
+		key := "records/plain/"
+		_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+			Body:   strings.NewReader("marker"),
+		})
+		require.NoError(t, err)
+
+		_, err = client.DeleteObject(context.TODO(), &s3.DeleteObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(key),
+		})
+		require.NoError(t, err)
+	})
+}

weed/s3api/s3api_directory_marker.go

@@ -2,6 +2,7 @@ package s3api
 
 import (
 	"errors"
+	"net/http"
 	"strings"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -25,7 +26,9 @@ import (
 
 // deleteDirectoryMarker removes the key "<dir>/". Callers hold the object write lock,
 // so the entry this decides about cannot change between the read and the delete.
-func (s3a *S3ApiServer) deleteDirectoryMarker(bucket, object string) s3err.ErrorCode {
+func (s3a *S3ApiServer) deleteDirectoryMarker(r *http.Request, bucket, object string) s3err.ErrorCode {
+	governanceBypassAllowed := s3a.evaluateGovernanceBypassRequest(r, bucket, object)
+
 	markerDir := s3a.bucketDir(bucket) + "/" + strings.TrimSuffix(strings.TrimPrefix(object, "/"), "/")
 	dir, name := util.FullPath(markerDir).DirAndName()
 
@@ -45,11 +48,51 @@ func (s3a *S3ApiServer) deleteDirectoryMarker(bucket, object string) s3err.Error
 		return s3err.ErrNone
 	}
 
+	// The key is deleted the unversioned way, but Object Lock still covers it: the
+	// gateway lists it as an object and serves retention set on it. The lock that
+	// matters is the one on this entry, since that is what is removed -- looking the
+	// key up instead would answer with a version once the key has a history.
+	if err := s3a.enforceObjectLockOnEntry(entry, bucket, object, "", governanceBypassAllowed); err != nil {
+		glog.V(2).Infof("deleteDirectoryMarker: %s/%s is locked: %v", bucket, object, err)
+		return s3err.ErrAccessDenied
+	}
+
 	// Drop a history an older build recorded for this key. Nothing writes one now, and
 	// leaving it behind keeps reporting the key in ListObjectVersions, so a history we
 	// cannot read or remove fails the delete rather than half finishing it.
 	switch _, historyErr := s3a.getEntry(markerDir, s3_constants.VersionsFolder); {
 	case historyErr == nil:
+		// The removal below takes every entry under the key, so each has to be clear
+		// of a lock of its own.
+		versionsDir := markerDir + "/" + s3_constants.VersionsFolder
+		for startFrom := ""; ; {
+			entries, isLast, listErr := s3a.list(versionsDir, "", startFrom, false, 1000)
+			if listErr != nil {
+				glog.Errorf("deleteDirectoryMarker: cannot list history of %s/%s: %v", bucket, object, listErr)
+				return s3err.ErrInternalError
+			}
+			for _, entry := range entries {
+				startFrom = entry.Name
+				versionId, named := entry.Extended[s3_constants.ExtVersionIdKey]
+				if !named {
+					// An entry an older build left without a version id is what this
+					// removal is here to clear, but one still under a lock cannot be
+					// named to check it, so judge it on what it carries itself.
+					if err := s3a.enforceObjectLockOnEntry(entry, bucket, object, "", governanceBypassAllowed); err != nil {
+						glog.V(2).Infof("deleteDirectoryMarker: unnamed history entry %s of %s/%s is locked: %v", entry.Name, bucket, object, err)
+						return s3err.ErrAccessDenied
+					}
+					continue
+				}
+				if err := s3a.enforceObjectLockProtections(r, bucket, object, string(versionId), governanceBypassAllowed); err != nil {
+					glog.V(2).Infof("deleteDirectoryMarker: version %s of %s/%s is locked: %v", versionId, bucket, object, err)
+					return s3err.ErrAccessDenied
+				}
+			}
+			if isLast || len(entries) == 0 {
+				break
+			}
+		}
 		if rmErr := s3a.rm(markerDir, s3_constants.VersionsFolder, true, true); rmErr != nil {
 			glog.Errorf("deleteDirectoryMarker: failed to remove stale history of %s/%s: %v", bucket, object, rmErr)
 			return s3err.ErrInternalError

weed/s3api/s3api_object_handlers_delete.go

@@ -126,7 +126,7 @@ func (s3a *S3ApiServer) deleteVersionedObject(r *http.Request, bucket, object, v
 	// in for without hiding the children underneath it. It is not a versioned object,
 	// so it is deleted the way an unversioned bucket deletes it.
 	if versionId == "" && strings.HasSuffix(object, "/") {
-		return result, s3a.deleteDirectoryMarker(bucket, object)
+		return result, s3a.deleteDirectoryMarker(r, bucket, object)
 	}
 
 	switch {
@@ -245,7 +245,7 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque
 		deleteCode, deleteHandled = s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
 			return s3a.checkDeleteIfMatch(bucket, object, versionId, versioningState, r.Header.Get(s3_constants.IfMatch), s3err.ErrPreconditionFailed)
 		}, func() s3err.ErrorCode {
-			return s3a.deleteDirectoryMarker(bucket, object)
+			return s3a.deleteDirectoryMarker(r, bucket, object)
 		}), true
 	}
 
@@ -486,7 +486,7 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 				}
 
 				if strings.HasSuffix(object.Key, "/") {
-					return s3a.deleteDirectoryMarker(bucket, object.Key)
+					return s3a.deleteDirectoryMarker(r, bucket, object.Key)
 				}
 
 				if err := s3a.deleteUnversionedObjectWithClient(client, bucket, object.Key, false); err != nil {

weed/s3api/s3api_object_handlers_put.go

@@ -142,6 +142,17 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			return
 		}
 
+		objectLockEnabled, lockErr := s3a.isObjectLockEnabled(bucket)
+		if lockErr != nil && !errors.Is(lockErr, filer_pb.ErrNotFound) {
+			glog.Errorf("PutObjectHandler: failed to check object lock for bucket %s: %v", bucket, lockErr)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			return
+		}
+		if validationErr := s3a.validateObjectLockHeaders(r, objectLockEnabled); validationErr != nil {
+			glog.V(2).Infof("PutObjectHandler: object lock header validation failed for %s/%s: %v", bucket, object, validationErr)
+			s3err.WriteErrorResponse(w, r, mapValidationErrorToS3Error(validationErr))
+			return
+		}
 		// Split the object into directory path and name
 		objectWithoutSlash := strings.TrimSuffix(object, "/")
 		dirName := path.Dir(objectWithoutSlash)
@@ -176,28 +187,55 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 
 		glog.Infof("PutObjectHandler: explicit directory marker %s/%s (contentType=%q, len=%d)",
 			bucket, object, objectContentType, r.ContentLength)
-		if err := s3a.mkdir(
-			fullDirPath, entryName,
-			func(entry *filer_pb.Entry) {
-				if objectContentType == "" {
-					objectContentType = s3_constants.FolderMimeType
-				}
-				if len(dirContent) > 0 {
-					entry.Content = dirContent
-				}
-				entry.Attributes.Mime = objectContentType
-				entry.Attributes.Md5 = dirMd5[:]
+		// mkdir replaces this entry outright, so a lock recorded on the entry itself
+		// is what stands in the way -- not the latest version, which a versioned
+		// write of the same key is free to add to. Check it under the same lock the
+		// marker delete takes, so the entry cannot change in between.
+		markerCode := s3a.withObjectWriteLock(bucket, object, func() s3err.ErrorCode {
+			if !objectLockEnabled {
+				return s3err.ErrNone
+			}
+			existing, existErr := s3a.getEntry(fullDirPath, entryName)
+			if existErr != nil {
+				return s3err.ErrNone
+			}
+			if lockErr := s3a.enforceObjectLockOnEntry(existing, bucket, object, "", s3a.evaluateGovernanceBypassRequest(r, bucket, object)); lockErr != nil {
+				glog.V(2).Infof("PutObjectHandler: object lock permissions check failed for %s/%s: %v", bucket, object, lockErr)
+				return s3err.ErrAccessDenied
+			}
+			return s3err.ErrNone
+		}, func() s3err.ErrorCode {
+			if err := s3a.mkdir(
+				fullDirPath, entryName,
+				func(entry *filer_pb.Entry) {
+					if objectContentType == "" {
+						objectContentType = s3_constants.FolderMimeType
+					}
+					if len(dirContent) > 0 {
+						entry.Content = dirContent
+					}
+					entry.Attributes.Mime = objectContentType
+					entry.Attributes.Md5 = dirMd5[:]
 
-				// Store ETag in extended attributes for consistency with regular objects
-				if entry.Extended == nil {
-					entry.Extended = make(map[string][]byte)
-				}
-				entry.Extended[s3_constants.ExtETagKey] = []byte(dirEtag)
+					// Store ETag in extended attributes for consistency with regular objects
+					if entry.Extended == nil {
+						entry.Extended = make(map[string][]byte)
+					}
+					entry.Extended[s3_constants.ExtETagKey] = []byte(dirEtag)
 
-				// Set object owner for directory objects (same as regular objects)
-				s3a.setObjectOwnerFromRequest(r, bucket, entry)
-			}); err != nil {
-			s3err.WriteErrorResponse(w, r, filerErrorToS3Error(err))
+					// Set object owner for directory objects (same as regular objects)
+					s3a.setObjectOwnerFromRequest(r, bucket, entry)
+
+					if lockErr := s3a.extractObjectLockMetadataFromRequest(r, entry); lockErr != nil {
+						glog.Errorf("PutObjectHandler: failed to extract object lock metadata for %s/%s: %v", bucket, object, lockErr)
+					}
+				}); err != nil {
+				return filerErrorToS3Error(err)
+			}
+			return s3err.ErrNone
+		})
+		if markerCode != s3err.ErrNone {
+			s3err.WriteErrorResponse(w, r, markerCode)
 			return
 		}
 		setEtag(w, dirEtag)

weed/s3api/s3api_object_retention.go

@@ -608,14 +608,20 @@ func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, buck
 		return err
 	}
 
-	// Extract retention information from the entry
+	return s3a.enforceObjectLockOnEntry(entry, bucket, object, versionId, governanceBypassAllowed)
+}
+
+// enforceObjectLockOnEntry reports whether a lock recorded on an entry stops the
+// operation. Callers holding the entry already -- one reached without a version id
+// of its own, or the one a write is about to replace -- use it directly, so that
+// the decision lives in one place.
+func (s3a *S3ApiServer) enforceObjectLockOnEntry(entry *filer_pb.Entry, bucket, object, versionId string, governanceBypassAllowed bool) error {
 	retention, retentionActive, err := s3a.getRetentionFromEntry(entry)
 	if err != nil {
 		glog.Warningf("Error parsing retention for %s/%s (versionId: %s): %v", bucket, object, versionId, err)
 		// Continue with legal hold check even if retention parsing fails
 	}
 
-	// Extract legal hold information from the entry
 	_, legalHoldActive, err := s3a.getLegalHoldFromEntry(entry)
 	if err != nil {
 		glog.Warningf("Error parsing legal hold for %s/%s (versionId: %s): %v", bucket, object, versionId, err)