diff --git a/test/s3/retention/s3_object_lock_directory_marker_test.go b/test/s3/retention/s3_object_lock_directory_marker_test.go new file mode 100644 index 000000000..6ae813238 --- /dev/null +++ b/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) + }) +} diff --git a/weed/s3api/s3api_directory_marker.go b/weed/s3api/s3api_directory_marker.go index b70955b76..41577bae7 100644 --- a/weed/s3api/s3api_directory_marker.go +++ b/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 "/". 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 diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index 853c08e9f..99664b871 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/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 { diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 2862f1935..a60231551 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/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) diff --git a/weed/s3api/s3api_object_retention.go b/weed/s3api/s3api_object_retention.go index 42e18b9af..4cd1b039f 100644 --- a/weed/s3api/s3api_object_retention.go +++ b/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)