diff --git a/weed/s3api/s3api_internal_lifecycle.go b/weed/s3api/s3api_internal_lifecycle.go index b60da67871..4d267e0780 100644 --- a/weed/s3api/s3api_internal_lifecycle.go +++ b/weed/s3api/s3api_internal_lifecycle.go @@ -186,7 +186,7 @@ func (s3a *S3ApiServer) lifecycleDispatch(ctx context.Context, req *s3_lifecycle versionDeleted = true } } else { - switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm, false, metadataOnly); code { + switch code := s3a.routedDeleteSpecificVersion(owner, req.Bucket, req.ObjectPath, req.VersionId, worm && !isDeleteMarkerEntry(entry), false, metadataOnly); code { case s3err.ErrNone: versionDeleted = true case s3err.ErrAccessDenied: diff --git a/weed/s3api/s3api_internal_lifecycle_test.go b/weed/s3api/s3api_internal_lifecycle_test.go index 4b97958fa8..0009dbe753 100644 --- a/weed/s3api/s3api_internal_lifecycle_test.go +++ b/weed/s3api/s3api_internal_lifecycle_test.go @@ -8,6 +8,8 @@ import ( "time" "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -468,3 +470,53 @@ func TestLifecycleDelete_CurrentExpirationRequiresLatestVersion(t *testing.T) { } } } + +// Expired-marker cleanup removes a marker that a lock was recorded on: markers +// are never WORM-protected, so the routed delete carries no WORM guard. +func TestLifecycleDelete_LockedExpiredDeleteMarkerSkipsWormGuard(t *testing.T) { + marker := &filer_pb.Entry{ + Name: "v_m1", + Attributes: &filer_pb.FuseAttributes{Mtime: 1_700_000_000}, + Extended: map[string][]byte{ + s3_constants.ExtVersionIdKey: []byte("m1"), + s3_constants.ExtDeleteMarkerKey: []byte("true"), + s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOn), + }, + } + fake := &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{ + "/buckets/b/obj.versions": { + Name: "obj" + s3_constants.VersionsFolder, + IsDirectory: true, + Attributes: &filer_pb.FuseAttributes{}, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("m1"), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_m1"), + }, + }, + "/buckets/b/obj.versions/v_m1": proto.Clone(marker).(*filer_pb.Entry), + }} + var conditions []*filer_pb.WriteCondition + fake.txnCheck = func(req *filer_pb.ObjectTransactionRequest) { conditions = append(conditions, req.GetCondition()) } + addr := startFakeFiler(t, fake) + s3a := newPutTestServer(t, addr) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("b", &BucketConfig{ + Name: "b", + Versioning: s3_constants.VersioningEnabled, + ObjectLockConfig: &ObjectLockConfiguration{ObjectLockEnabled: s3_constants.ObjectLockEnabled}, + }) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, addr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addr}, 1) + + resp, err := s3a.LifecycleDelete(s3LocalPeerCtx(), &s3_lifecycle_pb.LifecycleDeleteRequest{ + Bucket: "b", + ObjectPath: "obj", + VersionId: "m1", + ActionKind: s3_lifecycle_pb.ActionKind_EXPIRED_DELETE_MARKER, + ExpectedIdentity: computeEntryIdentity(marker), + }) + require.NoError(t, err) + require.Equal(t, s3_lifecycle_pb.LifecycleDeleteOutcome_DONE, resp.Outcome, resp.Reason) + require.Len(t, conditions, 1) + require.Nil(t, conditions[0], "marker delete must not carry WORM guards") +} diff --git a/weed/s3api/s3api_object_handlers_delete.go b/weed/s3api/s3api_object_handlers_delete.go index 9062e7ccc5..147eb618a3 100644 --- a/weed/s3api/s3api_object_handlers_delete.go +++ b/weed/s3api/s3api_object_handlers_delete.go @@ -337,7 +337,9 @@ func (s3a *S3ApiServer) DeleteObjectHandler(w http.ResponseWriter, r *http.Reque if versionId == "null" { deleteCode, deleteHandled = s3a.routedDeleteNullVersion(owner, bucket, object, worm, bypass, false) } else { - deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm, bypass, false), true + // A version id never changes kind, so a marker read here is + // the one deleted; markers carry no WORM guard. + deleteCode, deleteHandled = s3a.routedDeleteSpecificVersion(owner, bucket, object, versionId, worm && !deleteResult.deleteMarker, bypass, false), true } } } diff --git a/weed/s3api/s3api_object_handlers_legal_hold.go b/weed/s3api/s3api_object_handlers_legal_hold.go index 326ebb59ac..e0a0d34768 100644 --- a/weed/s3api/s3api_object_handlers_legal_hold.go +++ b/weed/s3api/s3api_object_handlers_legal_hold.go @@ -50,7 +50,7 @@ func (s3a *S3ApiServer) PutObjectLegalHoldHandler(w http.ResponseWriter, r *http glog.Errorf("PutObjectLegalHoldHandler: failed to set legal hold: %v", err) // Handle specific error cases - if errors.Is(err, ErrObjectNotFound) || errors.Is(err, ErrVersionNotFound) { + if errors.Is(err, ErrObjectNotFound) || errors.Is(err, ErrVersionNotFound) || errors.Is(err, ErrLatestVersionNotFound) { s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey) return } diff --git a/weed/s3api/s3api_object_retention.go b/weed/s3api/s3api_object_retention.go index b8dddac576..6df625ed39 100644 --- a/weed/s3api/s3api_object_retention.go +++ b/weed/s3api/s3api_object_retention.go @@ -238,6 +238,14 @@ func isObjectEntryMiss(err error) bool { errors.Is(err, errInvalidVersionID) } +// missAs reports a genuine miss as sentinel and keeps any other failure as is. +func missAs(err, sentinel error) error { + if isObjectEntryMiss(err) { + return sentinel + } + return err +} + // getObjectEntryForUpdate resolves the entry a metadata update targets and the // filer path it is written under. It applies the same version resolution as // getSpecificObjectVersion - "null" names the plain pre-versioning entry, not a @@ -247,7 +255,10 @@ func (s3a *S3ApiServer) getObjectEntryForUpdate(bucket, object, versionId string if versionId != "" { entry, err = s3a.getSpecificObjectVersion(bucket, object, versionId) if err != nil { - return nil, "", fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, ErrVersionNotFound) + return nil, "", fmt.Errorf("failed to get version %s for object %s/%s: %w", versionId, bucket, object, missAs(err, ErrVersionNotFound)) + } + if isDeleteMarkerEntry(entry) { + return nil, "", fmt.Errorf("version %s of object %s/%s is a delete marker: %w", versionId, bucket, object, ErrVersionNotFound) } if versionId != "null" { entryPath = object + ".versions/" + s3a.getVersionFileName(versionId) @@ -262,14 +273,17 @@ func (s3a *S3ApiServer) getObjectEntryForUpdate(bucket, object, versionId string if !versioningEnabled { entry, err = s3a.fetchObjectEntryRequired(bucket, object) if err != nil { - return nil, "", fmt.Errorf("failed to get object %s/%s: %w", bucket, object, ErrObjectNotFound) + return nil, "", fmt.Errorf("failed to get object %s/%s: %w", bucket, object, missAs(err, ErrObjectNotFound)) } return entry, entryPath, nil } entry, err = s3a.getLatestObjectVersion(bucket, object) if err != nil { - return nil, "", fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, ErrLatestVersionNotFound) + return nil, "", fmt.Errorf("failed to get latest version for object %s/%s: %w", bucket, object, missAs(err, ErrLatestVersionNotFound)) + } + if isDeleteMarkerEntry(entry) { + return nil, "", fmt.Errorf("latest version of object %s/%s is a delete marker: %w", bucket, object, ErrLatestVersionNotFound) } if entry.Extended != nil { if versionIdBytes, exists := entry.Extended[s3_constants.ExtVersionIdKey]; exists { @@ -291,6 +305,9 @@ func (s3a *S3ApiServer) getObjectRetention(bucket, object, versionId string) (*O if err != nil { return nil, err } + if isDeleteMarkerEntry(entry) { + return nil, fmt.Errorf("object %s/%s is a delete marker: %w", bucket, object, ErrObjectNotFound) + } if entry.Extended == nil { return nil, ErrNoRetentionConfiguration @@ -406,6 +423,9 @@ func (s3a *S3ApiServer) getObjectLegalHold(bucket, object, versionId string) (*O if err != nil { return nil, err } + if isDeleteMarkerEntry(entry) { + return nil, fmt.Errorf("object %s/%s is a delete marker: %w", bucket, object, ErrObjectNotFound) + } if entry.Extended == nil { return nil, ErrNoLegalHoldConfiguration @@ -593,6 +613,11 @@ func (s3a *S3ApiServer) enforceObjectLockProtections(request *http.Request, buck // 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 { + // Delete markers are never WORM-protected; a lock recorded on one must + // not keep it from being removed. + if isDeleteMarkerEntry(entry) { + return nil + } retention, retentionActive, err := s3a.getRetentionFromEntry(entry) if err != nil { glog.Warningf("Error parsing retention for %s/%s (versionId: %s): %v", bucket, object, versionId, err) diff --git a/weed/s3api/s3api_object_retention_test.go b/weed/s3api/s3api_object_retention_test.go index 2b19c7c124..8a479485d8 100644 --- a/weed/s3api/s3api_object_retention_test.go +++ b/weed/s3api/s3api_object_retention_test.go @@ -6,17 +6,21 @@ import ( "fmt" "io" "net/http" + "net/http/httptest" + "strconv" "strings" "sync" "testing" "time" + "github.com/gorilla/mux" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" + "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -1102,3 +1106,190 @@ func TestGetObjectEntryMissStillNotFound(t *testing.T) { t.Fatalf("missing object must report ErrObjectNotFound, got: %v", err) } } + +func putLegalHoldStatus(t *testing.T, s3a *S3ApiServer, object, versionId string) int { + t.Helper() + target := "http://localhost:8333/b/" + object + "?legal-hold" + if versionId != "" { + target += "&versionId=" + versionId + } + body := `ON` + req := httptest.NewRequest(http.MethodPut, target, strings.NewReader(body)) + req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": object}) + rec := httptest.NewRecorder() + s3a.PutObjectLegalHoldHandler(rec, req) + return rec.Code +} + +// PutObjectLegalHold on a key with no version in a versioned bucket is a miss, +// answered like PutObjectRetention with NoSuchKey rather than InternalError. +func TestPutObjectLegalHoldMissingObjectIsNoSuchKey(t *testing.T) { + f := &retentionTestFiler{entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)}, + }, + }} + s3a := newRetentionTestServer(t, f) + + if got := putLegalHoldStatus(t, s3a, "missing", ""); got != http.StatusNotFound { + t.Fatalf("PutObjectLegalHold on a missing key = %d, want 404", got) + } + if len(f.updates) != 0 { + t.Fatalf("missing key must not be written, got %v", f.updates) + } +} + +// A failed read of the target version is not a miss: the lock write must +// surface it as a server error the client can retry, not NoSuchKey. +func TestPutObjectLegalHoldFailedReadIsNotNoSuchKey(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)}, + }, + }, + lookupErrs: map[string]error{ + "/buckets/b/key.versions/v_v1": status.Error(codes.Internal, "disk read failed"), + }, + } + s3a := newRetentionTestServer(t, f) + + err := s3a.setObjectLegalHold("b", "key", "v1", &ObjectLegalHold{Status: "ON"}) + if err == nil || errors.Is(err, ErrVersionNotFound) { + t.Fatalf("failed read must not report ErrVersionNotFound, got: %v", err) + } + if got := putLegalHoldStatus(t, s3a, "key", "v1"); got != http.StatusInternalServerError { + t.Fatalf("PutObjectLegalHold on a failed read = %d, want 500", got) + } +} + +func deleteMarkerTestFiler(markerExtended map[string][]byte) *retentionTestFiler { + marker := map[string][]byte{ + s3_constants.ExtDeleteMarkerKey: []byte("true"), + s3_constants.ExtVersionIdKey: []byte("m1"), + } + for k, v := range markerExtended { + marker[k] = v + } + return &retentionTestFiler{entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)}, + }, + "/buckets/b/key.versions": { + Name: "key.versions", + IsDirectory: true, + Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("m1"), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_m1"), + }, + }, + "/buckets/b/key.versions/v_m1": {Name: "v_m1", Extended: marker}, + }} +} + +// A key whose current version is a delete marker does not exist, and a marker +// takes no lock of its own: reading or writing its legal hold is NoSuchKey. +func TestLegalHoldOnDeleteMarkerIsNoSuchKey(t *testing.T) { + f := deleteMarkerTestFiler(nil) + s3a := newRetentionTestServer(t, f) + + if got := putLegalHoldStatus(t, s3a, "key", ""); got != http.StatusNotFound { + t.Fatalf("PutObjectLegalHold on a deleted key = %d, want 404", got) + } + if got := putLegalHoldStatus(t, s3a, "key", "m1"); got != http.StatusNotFound { + t.Fatalf("PutObjectLegalHold on a delete marker version = %d, want 404", got) + } + if len(f.updates) != 0 { + t.Fatalf("delete marker must not be written, got %v", f.updates) + } + + req := httptest.NewRequest(http.MethodGet, "http://localhost:8333/b/key?legal-hold", nil) + req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": "key"}) + rec := httptest.NewRecorder() + s3a.GetObjectLegalHoldHandler(rec, req) + if rec.Code != http.StatusNotFound { + t.Fatalf("GetObjectLegalHold on a deleted key = %d, want 404", rec.Code) + } + + retainUntil := time.Now().Add(time.Hour) + err := s3a.setObjectRetention("b", "key", "", &ObjectRetention{Mode: s3_constants.RetentionModeGovernance, RetainUntilDate: &retainUntil}, false) + if !errors.Is(err, ErrLatestVersionNotFound) { + t.Fatalf("setObjectRetention on a deleted key = %v, want ErrLatestVersionNotFound", err) + } + if _, err := s3a.getObjectRetention("b", "key", "m1"); !errors.Is(err, ErrObjectNotFound) { + t.Fatalf("getObjectRetention on a delete marker = %v, want ErrObjectNotFound", err) + } +} + +// A lock left on a delete marker by an older server must not keep the marker +// from being removed: markers are never WORM-protected. +func TestLockOnDeleteMarkerDoesNotBlockRemoval(t *testing.T) { + retainUntil := strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10) + f := deleteMarkerTestFiler(map[string][]byte{ + s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOn), + s3_constants.ExtObjectLockModeKey: []byte(s3_constants.RetentionModeCompliance), + s3_constants.ExtRetentionUntilDateKey: []byte(retainUntil), + }) + f.entries["/buckets/b"].Extended[s3_constants.ExtObjectLockEnabledKey] = []byte(s3_constants.ObjectLockEnabled) + s3a := newRetentionTestServer(t, f) + + for _, versionId := range []string{"m1", ""} { + if err := s3a.enforceObjectLockProtections(nil, "b", "key", versionId, false); err != nil { + t.Fatalf("enforceObjectLockProtections(versionId=%q) on a locked delete marker = %v, want nil", versionId, err) + } + } +} + +type deleteMarkerTxnFiler struct { + *retentionTestFiler + txns []*filer_pb.ObjectTransactionRequest +} + +func (f *deleteMarkerTxnFiler) ObjectTransaction(_ context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.txns = append(f.txns, req) + return &filer_pb.ObjectTransactionResponse{}, nil +} + +// The routed version delete carries the WORM guards to the owner filer, which +// would refuse a marker that holds a lock; a marker is deleted without them. +func TestRoutedDeleteOfLockedDeleteMarkerSkipsWormGuard(t *testing.T) { + retainUntil := strconv.FormatInt(time.Now().Add(time.Hour).Unix(), 10) + base := deleteMarkerTestFiler(map[string][]byte{ + s3_constants.ExtLegalHoldKey: []byte(s3_constants.LegalHoldOn), + s3_constants.ExtObjectLockModeKey: []byte(s3_constants.RetentionModeCompliance), + s3_constants.ExtRetentionUntilDateKey: []byte(retainUntil), + }) + base.entries["/buckets/b"].Extended[s3_constants.ExtObjectLockEnabledKey] = []byte(s3_constants.ObjectLockEnabled) + f := &deleteMarkerTxnFiler{retentionTestFiler: base} + addr := startFakeFiler(t, f) + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + s3a := &S3ApiServer{ + option: &S3ApiServerOption{Filers: []pb.ServerAddress{addr}, GrpcDialOption: dialOption, BucketsPath: "/buckets"}, + filerClient: wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""), + bucketConfigCache: NewBucketConfigCache(time.Minute), + } + s3a.objectWriteLockClient = cluster.NewLockClient(dialOption, addr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addr}, 1) + + req := httptest.NewRequest(http.MethodDelete, "http://localhost:8333/b/key?versionId=m1", nil) + req = mux.SetURLVars(req, map[string]string{"bucket": "b", "object": "key"}) + rec := httptest.NewRecorder() + s3a.DeleteObjectHandler(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("DeleteObject of a locked delete marker = %d, want 204: %s", rec.Code, rec.Body.String()) + } + if len(f.txns) != 1 { + t.Fatalf("expected one routed delete, got %d", len(f.txns)) + } + if cond := f.txns[0].GetCondition(); cond != nil { + t.Fatalf("delete marker removal must not carry WORM guards, got %+v", cond) + } +} diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 4cd43c9adb..9ce68ea3d7 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -32,6 +32,10 @@ import ( // ErrDeleteMarker is returned when the latest version is a delete marker (expected condition) var ErrDeleteMarker = errors.New("latest version is a delete marker") +func isDeleteMarkerEntry(entry *filer_pb.Entry) bool { + return entry != nil && string(entry.Extended[s3_constants.ExtDeleteMarkerKey]) == "true" +} + // clearCachedVersionMetadata clears only the version metadata fields (not ID/filename). // Used by setCachedListMetadata to prevent stale values when updating. func clearCachedVersionMetadata(extended map[string][]byte) {