From 16e66b1bad624aa6fc99305a69c18810a639f63b Mon Sep 17 00:00:00 2001 From: zhao-yc <43179751+zhao-yc@users.noreply.github.com> Date: Mon, 5 Oct 2026 18:22:16 +0800 Subject: [PATCH] fix(s3): initialize destination ACLs for CopyObject (#11599) * fix(s3): initialize destination ACLs for CopyObject Signed-off-by: zhaoyuchen <43179751+zhao-yc@users.noreply.github.com> * fix(s3): re-check routed self-copy eligibility on the locked read routeInPlace was decided on the pre-lock entry, but the PATCH body re-reads the entry. A concurrent write changing file mode or MIME in between left the routed PATCH installing new ACL keys while Attributes kept the stale mode. Evaluate eligibility against the re-read entry and retry the self-copy under the distributed lock when it no longer qualifies. * fix(s3): guard metadata self-copies against concurrent writes Signed-off-by: zhaoyuchen <43179751+zhao-yc@users.noreply.github.com> * ci: raise s3api unit-test timeout to 9m The suite crossed the 5m binary timeout on the hosted runner (local run is ~4.3m and still growing). The job-level limit is already 10m. * ci: allow setup time before the S3 API test suite Signed-off-by: zhaoyuchen <43179751+zhao-yc@users.noreply.github.com> --------- Signed-off-by: zhaoyuchen <43179751+zhao-yc@users.noreply.github.com> Co-authored-by: Chris Lu --- .github/workflows/s3-parquet-tests.yml | 5 +- weed/s3api/s3api_object_handlers_copy.go | 101 +++- .../s3api_object_handlers_copy_acl_test.go | 484 ++++++++++++++++++ weed/s3api/s3api_object_routed_write.go | 65 +-- weed/s3api/s3api_object_upload_acl.go | 19 +- weed/s3api/s3api_object_upload_acl_test.go | 8 +- weed/s3api/s3err/s3api_errors.go | 7 + ...er_grpc_server_self_copy_condition_test.go | 87 ++++ 8 files changed, 703 insertions(+), 73 deletions(-) create mode 100644 weed/s3api/s3api_object_handlers_copy_acl_test.go create mode 100644 weed/server/filer_grpc_server_self_copy_condition_test.go diff --git a/.github/workflows/s3-parquet-tests.yml b/.github/workflows/s3-parquet-tests.yml index 52e602a63..b86c3869e 100644 --- a/.github/workflows/s3-parquet-tests.yml +++ b/.github/workflows/s3-parquet-tests.yml @@ -139,7 +139,8 @@ jobs: unit-tests: name: Go Unit Tests (Implicit Directory) runs-on: ubuntu-latest - timeout-minutes: 10 + # Leave time for setup and the focused run before the nine-minute suite. + timeout-minutes: 20 steps: - name: Checkout code @@ -159,5 +160,5 @@ jobs: - name: Run all S3 API tests run: | cd weed/s3api - go test -v -timeout 5m + go test -v -timeout 9m diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 1c48d243f..4855b29bb 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -184,6 +184,15 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return } + // A copy creates destination ownership and grants independently of the + // source ACL. Validate explicit grants before caching or writing any data. + var aclCode s3err.ErrorCode + r, aclCode = s3a.preparePutObjectACL(r, dstBucket) + if aclCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, aclCode) + return + } + // Get detailed versioning state for source bucket srcVersioningState, err := s3a.getVersioningState(srcBucket) if err != nil { @@ -257,45 +266,82 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request if replacesSource && (replaceMeta || replaceTagging) && s3a.canUseMetadataOnlySelfCopy(entry, r, dstBucket, dstObject) { var dstVersionId string var etag string - // An in-place metadata replace routes to the owner as a serialized PATCH - // (off the distributed lock); the no-owner bootstrap keeps the lock. - // - // REPLACE can also change Content-Type, which lives on Attributes.Mime, - // not Extended. The routed PATCH only carries Extended keys, so when the - // Mime actually changes keep the lock and take the clone path below: it is - // still metadata-only (reuses the source chunks) but can set the Mime. + // Bind the entire self-copy update to the snapshot read on each attempt. + // ACLs and their file mode must commit together, even when another writer + // changes the mode between the gateway's read and the owner's commit. owner := s3a.objectWriteOwner(dstBucket, dstObject) - sourceMime := entry.GetAttributes().GetMime() - mimeChanged := resolveDestinationMime(r.Header, sourceMime, replaceMeta) != sourceMime - routeInPlace := owner != "" && !mimeChanged + fileMode := s3a.resolveFileMode(r) selfCopyBody := func() s3err.ErrorCode { - currentEntry, currentErr := s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState) - currentEntry = prefixObjectSource(currentEntry) + var rawEntry *filer_pb.Entry + var currentErr error + if owner != "" { + // Read the same owner that will check the snapshot, so a lagging + // replica does not make every retry conflict with a completed PUT. + rawEntry, currentErr = s3a.getObjectEntryRoutedByKey(srcBucket, srcObject) + } else { + rawEntry, currentErr = s3a.resolveCopySourceEntry(srcBucket, srcObject, srcVersionId, srcVersioningState) + } + currentEntry := prefixObjectSource(rawEntry) if errCode := classifyCopySourceError(currentEntry, currentErr); errCode != s3err.ErrNone { return errCode } if errCode := s3a.validateConditionalCopyHeaders(r, currentEntry); errCode != s3err.ErrNone { return errCode } + // Source and destination name the same unversioned entry. Evaluate + // destination conditions against the exact snapshot guarded at commit. + headers, errCode := parseConditionalHeaders(r) + if errCode != s3err.ErrNone { + return errCode + } + if errCode = s3a.validateConditionalHeaders(r, headers, currentEntry, dstBucket, dstObject); errCode != s3err.ErrNone { + return errCode + } + if err := ValidateCopyEncryption(currentEntry.Extended, r.Header); err != nil { + return MapCopyValidationError(err) + } + if !s3a.canUseMetadataOnlySelfCopy(currentEntry, r, dstBucket, dstObject) { + // A new encryption/storage state needs the ordinary copy path on a + // fresh request; never reuse chunks from an incompatible snapshot. + return s3err.ErrConditionalRequestConflict + } updatedMetadata, metadataErr := processMetadataBytes(r.Header, currentEntry.Extended, replaceMeta, replaceTagging) if metadataErr != nil { glog.Errorf("CopyObjectHandler ValidateTags error %s: %v", r.URL, metadataErr) return s3err.ErrInvalidTag } - if routeInPlace { - if err := s3a.routedMetadataReplace(owner, dstBucket, dstObject, currentEntry, updatedMetadata); err != nil { - return filerErrorToS3Error(err) - } - etag = getEtagFromEntry(currentEntry) - return s3err.ErrNone - } + // Both ACL keys replace the source's grants on every fresh attempt. + applyPutObjectACL(r, &filer_pb.Entry{Extended: updatedMetadata}) updatedEntry := cloneProtoEntry(currentEntry) + if owner != "" { + // Prefix objects remain directories containing child keys. Use + // their raw form both for the condition and for the update. + updatedEntry = cloneProtoEntry(rawEntry) + } updatedEntry.Extended = mergeCopyMetadata(updatedEntry.Extended, updatedMetadata) if updatedEntry.Attributes == nil { updatedEntry.Attributes = &filer_pb.FuseAttributes{} } updatedEntry.Attributes.Mime = resolveDestinationMime(r.Header, currentEntry.GetAttributes().GetMime(), replaceMeta) + updatedEntry.Attributes.FileMode = fileMode + if updatedEntry.IsDirectory { + updatedEntry.Attributes.FileMode |= uint32(os.ModeDir) + } updatedEntry.Attributes.Mtime = t.Unix() + if owner != "" { + if err := s3a.routedSelfCopy(r.Context(), owner, dstBucket, dstObject, rawEntry, updatedEntry); err != nil { + switch status.Code(err) { + case codes.FailedPrecondition: + return s3err.ErrConditionalRequestConflict + case codes.NotFound: + return s3err.ErrNoSuchKey + default: + return filerErrorToS3Error(err) + } + } + etag = getEtagFromEntry(currentEntry) + return s3err.ErrNone + } var finErr error dstVersionId, etag, finErr = s3a.finalizeCopyDestination(dstBucket, dstObject, dstVersioningState, updatedEntry) if finErr != nil { @@ -304,9 +350,14 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return s3err.ErrNone } var updateCode s3err.ErrorCode - if routeInPlace { - if updateCode = s3a.checkConditionalHeaders(r, dstBucket, dstObject); updateCode == s3err.ErrNone { + if owner != "" { + // A rejected snapshot has not mutated the object. Re-read and + // revalidate on bounded conflicts, without an unconditional fallback. + for attempt := 0; attempt < 3; attempt++ { updateCode = selfCopyBody() + if updateCode != s3err.ErrConditionalRequestConflict { + break + } } } else { updateCode = s3a.withObjectWriteLock(dstBucket, dstObject, func() s3err.ErrorCode { @@ -367,6 +418,10 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request dstWantsSSES3 := IsSSES3RequestInternal(r) for k, v := range entry.Extended { + // Object ACLs and ownership are not copied, regardless of the metadata directive. + if k == s3_constants.ExtAmzOwnerKey || k == s3_constants.ExtAmzAclKey { + continue + } // Skip encryption-specific headers that might conflict with destination encryption type skipHeader := false @@ -400,8 +455,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request } // mergeCopyMetadata drops stale managed keys before applying the new set, - // so REPLACE doesn't leak source values through the merge. Mirrors the - // self-copy path's routedMetadataReplace. + // so REPLACE doesn't leak source values through the merge, as on self-copies. dstEntry.Extended = mergeCopyMetadata(dstEntry.Extended, processedMetadata) // For zero-size files or files without chunks, handle inline content @@ -468,6 +522,7 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request var dstVersionId string var etag string + applyPutObjectACL(r, dstEntry) finalizeCode := s3a.withObjectWriteLock(dstBucket, dstObject, func() s3err.ErrorCode { return s3a.checkConditionalHeaders(r, dstBucket, dstObject) diff --git a/weed/s3api/s3api_object_handlers_copy_acl_test.go b/weed/s3api/s3api_object_handlers_copy_acl_test.go new file mode 100644 index 000000000..ef2997e58 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_acl_test.go @@ -0,0 +1,484 @@ +package s3api + +import ( + "context" + "encoding/xml" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/signer/v4" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" + "github.com/seaweedfs/seaweedfs/weed/wdclient" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// copyACLFiler records both ordinary creates and the metadata-only self-copy +// transaction, so tests inspect the actual entry sent to the filer. +type copyACLFiler struct { + *ambiguousPutFiler + transactions []*filer_pb.ObjectTransactionRequest + updates []*filer_pb.UpdateEntryRequest + afterLookup func(*filer_pb.LookupDirectoryEntryRequest) + beforeCommit func() +} + +// Count two-node RPCs to distinguish a business rejection from failover. +type selfCopyReplyFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + code codes.Code + calls atomic.Int32 +} + +func (f *selfCopyReplyFiler) UpdateEntry(context.Context, *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + f.calls.Add(1) + return nil, status.Error(f.code, "test self-copy reply") +} + +// A snapshot conflict or deletion is an authoritative owner reply. It must +// preserve its status without consulting a peer or tripping owner health. +func TestRoutedSelfCopyBusinessErrors(t *testing.T) { + for _, code := range []codes.Code{codes.FailedPrecondition, codes.NotFound} { + t.Run(code.String(), func(t *testing.T) { + owner := &selfCopyReplyFiler{code: code} + peer := &selfCopyReplyFiler{code: codes.Unavailable} + ownerAddress := startFakeFiler(t, owner) + peerAddress := startFakeFiler(t, peer) + s3a := newPutTestServer(t, peerAddress) + current := &filer_pb.Entry{Name: "obj", Attributes: &filer_pb.FuseAttributes{FileMode: defaultFileMode}} + err := s3a.routedSelfCopy(context.Background(), ownerAddress, "b", "obj", current, proto.Clone(current).(*filer_pb.Entry)) + require.Equal(t, code, status.Code(err), "a peer transport error must not mask the owner's business rejection") + require.Equal(t, int32(1), owner.calls.Load()) + require.Zero(t, peer.calls.Load(), "rejected commits must not be replayed") + require.False(t, s3a.ownerRecentlyUnreachable(ownerAddress)) + }) + } +} + +// Race hooks place an intervening write after the snapshot or immediately +// before the filer's serialized precondition check, without timing-dependent sleeps. +func (f *copyACLFiler) LookupDirectoryEntry(ctx context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + resp, err := f.ambiguousPutFiler.LookupDirectoryEntry(ctx, req) + f.mu.Lock() + defer f.mu.Unlock() + if f.afterLookup != nil { + f.afterLookup(req) + } + return resp, err +} + +// Match the filer's normalized whole-entry condition, then atomically replace +// the entry; native condition/commit behavior is covered by server tests too. +func (f *copyACLFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest) (*filer_pb.UpdateEntryResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.updates = append(f.updates, proto.Clone(req).(*filer_pb.UpdateEntryRequest)) + if f.beforeCommit != nil { + f.beforeCommit() + } + current := f.entries[req.Directory+"/"+req.Entry.Name] + if current == nil { + return nil, status.Error(codes.NotFound, "missing entry") + } + if req.Condition == nil || len(req.Condition.Clauses) != 1 || req.Condition.Clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL { + return nil, status.Error(codes.InvalidArgument, "self-copy requires an entry condition") + } + normalize := func(e *filer_pb.Entry) *filer_pb.Entry { return filer.FromPbEntry("", e).ToProtoEntry() } + if !proto.Equal(normalize(current), normalize(req.Condition.Clauses[0].ExpectedEntry)) { + return nil, status.Error(codes.FailedPrecondition, "entry changed") + } + f.entries[req.Directory+"/"+req.Entry.Name] = proto.Clone(req.Entry).(*filer_pb.Entry) + return &filer_pb.UpdateEntryResponse{}, nil +} + +// These fixtures have no prior destination versions to list during a null write. +func (f *copyACLFiler) ListEntries(_ *filer_pb.ListEntriesRequest, _ filer_pb.SeaweedFiler_ListEntriesServer) error { + return nil +} + +func (f *copyACLFiler) ObjectTransaction(_ context.Context, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.transactions = append(f.transactions, proto.Clone(req).(*filer_pb.ObjectTransactionRequest)) + if f.beforeCommit != nil { + f.beforeCommit() + } + for _, mutation := range req.Mutations { + if mutation.Type != filer_pb.ObjectMutation_PATCH_EXTENDED { + return &filer_pb.ObjectTransactionResponse{Error: "unexpected mutation"}, nil + } + entry := f.entries[mutation.Directory+"/"+mutation.Name] + if entry == nil { + return &filer_pb.ObjectTransactionResponse{Error: "missing entry"}, nil + } + for _, key := range mutation.DeleteExtended { + delete(entry.Extended, key) + } + for key, value := range mutation.SetExtended { + entry.Extended[key] = append([]byte(nil), value...) + } + } + return &filer_pb.ObjectTransactionResponse{}, nil +} + +// TestCopyObjectDestinationACL uses authenticated CopyObject requests to verify +// that source grants never become destination grants, including self-copies. +func TestCopyObjectDestinationACL(t *testing.T) { + const bucket, writer, bucketOwner, sourceOwner = "copy-acl", "copy-writer", "bucket-owner", "source-owner" + const object, sourceObject = "destination", "source" + const versionID = "672a75d526ef29c79fe6b5680cad4a0d" + tests := []struct { + name, acl, grantHeader, grant, ownership, versioning, denyAction, errorCode string + self, routed, replace, presigned, v2, noRead, noWrite, noWriteACP, sourceVersion bool + query url.Values + sourceMode uint32 + mime string + race, condition string + prefix bool + }{ + {name: "default private"}, + {name: "replace metadata also defaults private", replace: true}, + {name: "explicit private", acl: "private"}, + {name: "explicit public read", acl: "public-read"}, + {name: "signed query public read", acl: "public-read", presigned: true}, + {name: "custom read", grantHeader: s3_constants.AmzAclRead, grant: `id="bucket-owner"`}, + {name: "custom full control", grantHeader: s3_constants.AmzAclFullControl, grant: `id="bucket-owner"`}, + {name: "custom read acp", grantHeader: s3_constants.AmzAclReadAcp, grant: `id="bucket-owner"`}, + {name: "custom write acp", grantHeader: s3_constants.AmzAclWriteAcp, grant: `id="bucket-owner"`, presigned: true}, + {name: "self copy bootstrap", self: true, replace: true}, + {name: "self copy routed private", self: true, replace: true, routed: true}, + {name: "self copy routed explicit public", self: true, replace: true, routed: true, acl: "public-read", sourceMode: 0644}, + {name: "self copy routed custom", self: true, replace: true, routed: true, grantHeader: s3_constants.AmzAclRead, grant: `id="bucket-owner"`}, + {name: "self copy changes file mode", self: true, replace: true, routed: true, sourceMode: 0644}, + {name: "self copy changes mime", self: true, replace: true, routed: true, mime: "image/png"}, + {name: "versioned destination", versioning: s3_constants.VersioningEnabled}, + {name: "versioned public destination", versioning: s3_constants.VersioningEnabled, acl: "public-read"}, + {name: "suspended destination", versioning: s3_constants.VersioningSuspended}, + {name: "specific source version", sourceVersion: true}, + {name: "versioned self copy", versioning: s3_constants.VersioningEnabled, sourceVersion: true, self: true}, + {name: "preferred bucket owner", ownership: s3_constants.OwnershipBucketOwnerPreferred, acl: "bucket-owner-full-control"}, + {name: "enforced bucket owner", ownership: s3_constants.OwnershipBucketOwnerEnforced}, + {name: "enforced accepts full control", ownership: s3_constants.OwnershipBucketOwnerEnforced, acl: "bucket-owner-full-control"}, + {name: "enforced rejects public", ownership: s3_constants.OwnershipBucketOwnerEnforced, acl: "public-read", errorCode: "AccessControlListNotSupported"}, + {name: "default needs no write acp", noWriteACP: true}, + {name: "public requires write acp", acl: "public-read", noWriteACP: true, errorCode: "AccessDenied"}, + {name: "custom requires write acp", grantHeader: s3_constants.AmzAclRead, grant: `id="bucket-owner"`, noWriteACP: true, errorCode: "AccessDenied"}, + {name: "requires source read", noRead: true, errorCode: "AccessDenied"}, + {name: "requires destination write", noWrite: true, errorCode: "AccessDenied"}, + {name: "query put deny", acl: "public-read", presigned: true, denyAction: "s3:PutObject", errorCode: "AccessDenied"}, + {name: "query acl deny", acl: "public-read", presigned: true, denyAction: "s3:PutObjectAcl", errorCode: "AccessDenied"}, + {name: "query custom put deny", grantHeader: s3_constants.AmzAclRead, grant: `id="bucket-owner"`, presigned: true, denyAction: "s3:PutObject", errorCode: "AccessDenied"}, + {name: "query custom acl deny", grantHeader: s3_constants.AmzAclRead, grant: `id="bucket-owner"`, presigned: true, denyAction: "s3:PutObjectAcl", errorCode: "AccessDenied"}, + {name: "invalid canned", acl: "invalid", errorCode: "InvalidRequest"}, + {name: "conflicting canned and grants", acl: "private", grantHeader: s3_constants.AmzAclRead, grant: `id="bucket-owner"`, errorCode: "InvalidRequest"}, + {name: "duplicate signed query", query: url.Values{"x-amz-acl": {"private", "public-read"}}, presigned: true, errorCode: "InvalidRequest"}, + {name: "v2 unsigned query ignored", v2: true, query: url.Values{"x-amz-acl": {"public-read"}}}, + {name: "private mode changes after first read", self: true, replace: true, routed: true, race: "read"}, + {name: "public mode changes after first read", self: true, replace: true, routed: true, race: "read", acl: "public-read", sourceMode: 0644}, + {name: "private mode changes before commit", self: true, replace: true, routed: true, race: "commit"}, + {name: "public mode changes before commit", self: true, replace: true, routed: true, race: "commit", acl: "public-read", sourceMode: 0644}, + {name: "mime changes before commit", self: true, replace: true, routed: true, race: "mime"}, + {name: "content changes before commit", self: true, replace: true, routed: true, race: "content"}, + {name: "source condition rechecked on retry", self: true, replace: true, routed: true, race: "content", condition: "source", errorCode: "PreconditionFailed"}, + {name: "destination condition rechecked on retry", self: true, replace: true, routed: true, race: "content", condition: "destination", errorCode: "PreconditionFailed"}, + {name: "deleted before commit", self: true, replace: true, routed: true, race: "delete", errorCode: "NoSuchKey"}, + {name: "bounded self copy conflicts", self: true, replace: true, routed: true, race: "repeat", errorCode: "ConditionalRequestConflict"}, + {name: "prefix private control", self: true, replace: true, routed: true, prefix: true}, + {name: "prefix mode changes before commit", self: true, replace: true, routed: true, prefix: true, race: "commit"}, + {name: "self copy reads owner ahead of replica", self: true, replace: true, routed: true, race: "replica"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + filer := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{apply: true, entries: map[string]*filer_pb.Entry{}}} + address := startFakeFiler(t, filer) + s3a := newPutTestServer(t, address) + s3a.iam = NewIdentityAccessManagementWithStore(s3a.option, nil, "memory") + t.Cleanup(s3a.iam.Shutdown) + s3a.iam.isAuthEnabled = true + account := &Account{Id: writer, DisplayName: writer} + identity := &Identity{Name: "copy-acl-test", Account: account, IsStatic: true, + Credentials: []*Credential{{AccessKey: routingTestAccessKey, SecretKey: routingTestSecretKey}}} + for action, enabled := range map[Action]bool{"Read:" + bucket: !tt.noRead, "Write:" + bucket: !tt.noWrite, "WriteAcp:" + bucket: !tt.noWriteACP} { + if enabled { + identity.Actions = append(identity.Actions, action) + } + } + s3a.iam.accessKeyIdent[routingTestAccessKey] = identity + s3a.iam.nameToIdentity[identity.Name] = identity + s3a.iam.accounts[writer] = account + s3a.iam.accounts[bucketOwner] = &Account{Id: bucketOwner, DisplayName: bucketOwner} + ownership := tt.ownership + if ownership == "" { + ownership = s3_constants.OwnershipObjectWriter + } + bucketEntry := &filer_pb.Entry{Name: bucket, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}, Extended: map[string][]byte{ + s3_constants.ExtAmzOwnerKey: []byte(bucketOwner), s3_constants.ExtOwnershipKey: []byte(ownership), + }} + filer.entries["/buckets/"+bucket] = bucketEntry + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set(bucket, &BucketConfig{Name: bucket, Owner: bucketOwner, Ownership: ownership, Versioning: tt.versioning}) + s3a.bucketRegistry = NewBucketRegistry(s3a) + s3a.bucketRegistry.LoadBucketMetadata(bucketEntry) + mode := tt.sourceMode + if mode == 0 { + mode = defaultFileMode + } + source := &filer_pb.Entry{Name: sourceObject, Attributes: &filer_pb.FuseAttributes{FileMode: mode, FileSize: 4, Mime: "text/plain"}, Content: []byte("data"), + Extended: map[string][]byte{"X-Amz-Meta-Source": []byte("kept"), "internal-other": []byte("retained")}} + require.Equal(t, s3err.ErrNone, AssembleEntryWithAcp(source, sourceOwner, []*s3.Grant{ + {Grantee: &s3.Grantee{Type: aws.String(s3_constants.GrantTypeCanonicalUser), ID: aws.String(sourceOwner)}, Permission: aws.String(s3_constants.PermissionFullControl)}, + {Grantee: &s3.Grantee{Type: aws.String(s3_constants.GrantTypeGroup), URI: aws.String(s3_constants.GranteeGroupAllUsers)}, Permission: aws.String(s3_constants.PermissionRead)}, + })) + if tt.prefix { + source.IsDirectory = true + source.Attributes.FileMode |= uint32(os.ModeDir) + source.Extended[s3_constants.SeaweedFSPrefixObject] = []byte("true") + filer.entries["/buckets/"+bucket+"/"+sourceObject+"/child"] = &filer_pb.Entry{Name: "child", Content: []byte("child data")} + } + key := object + if tt.self { + key = sourceObject + } + sourcePath := "/buckets/" + bucket + "/" + sourceObject + copySource := "/" + bucket + "/" + sourceObject + if tt.sourceVersion { + source.Name = s3a.getVersionFileName(versionID) + source.Extended[s3_constants.ExtVersionIdKey] = []byte(versionID) + sourcePath += s3_constants.VersionsFolder + "/" + source.Name + copySource += "?versionId=" + versionID + } + filer.entries[sourcePath] = source + originalSource := proto.Clone(source).(*filer_pb.Entry) + if tt.race != "" { + injected := false + intervene := func() { + if injected && tt.race != "repeat" { + return + } + injected = true + if tt.race == "delete" { + delete(filer.entries, sourcePath) + return + } + current := filer.entries[sourcePath] + current.Attributes.FileMode ^= 0024 + current.Extended["internal-other"] = []byte("concurrent retained") + current.Extended[s3_constants.ExtLegalHoldKey] = []byte("ON") + if tt.race == "mime" { + current.Attributes.Mime = "image/png" + } + if tt.race == "content" || tt.race == "replica" { + current.Content = []byte("next") + current.Extended[s3_constants.ExtETagKey] = []byte("next-etag") + } + } + if tt.race == "replica" { + // The default filer still has the pre-PUT snapshot. The owner + // already committed the newer entry before this copy starts. + replica := ©ACLFiler{ambiguousPutFiler: &ambiguousPutFiler{entries: map[string]*filer_pb.Entry{sourcePath: proto.Clone(originalSource).(*filer_pb.Entry)}}} + replicaAddress := startFakeFiler(t, replica) + s3a.filerClient = wdclient.NewFilerClient([]pb.ServerAddress{replicaAddress, address}, s3a.option.GrpcDialOption, "") + intervene() + } else if tt.race == "read" { + filer.afterLookup = func(req *filer_pb.LookupDirectoryEntryRequest) { + if req.Directory+"/"+req.Name == sourcePath { + intervene() + } + } + } else { + filer.beforeCommit = intervene + } + } + if tt.versioning != "" || tt.sourceVersion { + filer.entries["/buckets/"+bucket+"/"+key+s3_constants.VersionsFolder] = &filer_pb.Entry{Name: key + s3_constants.VersionsFolder, IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}} + } + if tt.routed { + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, address) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{address}, 1) + } + if tt.denyAction != "" { + s3a.policyEngine = NewBucketPolicyEngine() + s3a.iam.policyEngine = s3a.policyEngine + conditionKey, conditionValue := "s3:x-amz-acl", "public-read" + if tt.grantHeader != "" { + conditionKey, conditionValue = "s3:"+strings.ToLower(tt.grantHeader), tt.grant + } + policy := fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Principal":"*","Effect":"Deny","Action":%q,"Resource":"arn:aws:s3:::copy-acl/*","Condition":{"StringEquals":{%q:%q}}}]}`, tt.denyAction, conditionKey, conditionValue) + require.NoError(t, s3a.policyEngine.engine.SetBucketPolicy(bucket, policy)) + } + req := httptest.NewRequest(http.MethodPut, "http://s3/"+bucket+"/"+key, nil) + req = mux.SetURLVars(req, map[string]string{"bucket": bucket, "object": key}) + req.Header.Set("X-Amz-Copy-Source", copySource) + if tt.condition != "" { + header := "X-Amz-Copy-Source-If-Match" + if tt.condition == "destination" { + header = "If-Match" + } + req.Header.Set(header, getEtagFromEntry(prefixObjectSource(originalSource))) + } + // Forged internal headers must not influence the new owner or grants. + req.Header.Set(s3_constants.ExtAmzOwnerKey, "forged-owner") + req.Header.Set(s3_constants.ExtAmzAclKey, string(source.Extended[s3_constants.ExtAmzAclKey])) + if tt.replace { + req.Header.Set(s3_constants.AmzUserMetaDirective, "REPLACE") + req.Header.Set("X-Amz-Meta-Destination", "new") + req.Header.Set("Content-Type", "text/plain") + } + if tt.mime != "" { + req.Header.Set("Content-Type", tt.mime) + } + if tt.acl != "" { + req.Header.Set(s3_constants.AmzCannedAcl, tt.acl) + } + if tt.grantHeader != "" { + req.Header.Set(tt.grantHeader, tt.grant) + } + req.URL.RawQuery = tt.query.Encode() + if tt.presigned { + query := req.URL.Query() + for _, header := range []string{s3_constants.AmzCannedAcl, tt.grantHeader} { + if value := req.Header.Get(header); value != "" { + query.Set(header, value) + req.Header.Del(header) + } + } + req.URL.RawQuery = query.Encode() + signer := v4.NewSigner(credentials.NewStaticCredentials(routingTestAccessKey, routingTestSecretKey, "")) + _, err := signer.Presign(req, nil, "s3", "us-east-1", time.Minute, time.Now()) + require.NoError(t, err) + } else if tt.v2 { + req.Header.Set("Date", time.Now().UTC().Format(http.TimeFormat)) + req.Header.Set("Authorization", signatureV2(&Credential{AccessKey: routingTestAccessKey, SecretKey: routingTestSecretKey}, req.Method, req.URL.EscapedPath(), req.URL.RawQuery, req.Header)) + } else { + signRoutingTestRequest(t, req, "", "s3") + } + rr := httptest.NewRecorder() + s3a.iam.Auth(s3a.CopyObjectHandler, s3_constants.ACTION_WRITE)(rr, req) + filer.mu.Lock() + defer filer.mu.Unlock() + if tt.errorCode != "" { + require.Contains(t, rr.Body.String(), ""+tt.errorCode+"") + require.NotEqual(t, http.StatusOK, rr.Code) + if tt.race != "" { + if tt.race == "repeat" { + require.Equal(t, http.StatusConflict, rr.Code) + } + if tt.race == "delete" { + require.Nil(t, filer.entries[sourcePath]) + } else { + require.Equal(t, originalSource.Extended[s3_constants.ExtAmzAclKey], filer.entries[sourcePath].Extended[s3_constants.ExtAmzAclKey], "conflicts must not commit the self-copy ACL") + } + require.LessOrEqual(t, len(filer.updates), 3, "CAS retries must be bounded") + return + } + require.Empty(t, filer.transactions, "rejected copies must not issue mutations") + require.Nil(t, filer.entries["/buckets/"+bucket+"/"+object]) + require.True(t, proto.Equal(originalSource, filer.entries[sourcePath]), "rejected copies must preserve the source") + return + } + require.Equal(t, http.StatusOK, rr.Code, rr.Body.String()) + storedPath := "/buckets/" + bucket + "/" + key + if tt.versioning == s3_constants.VersioningEnabled { + newVersion := rr.Header().Get("X-Amz-Version-Id") + require.NotEmpty(t, newVersion) + storedPath += s3_constants.VersionsFolder + "/" + s3a.getVersionFileName(newVersion) + } + stored := filer.entries[storedPath] + require.NotNil(t, stored) + wantContent, wantOther := "data", "retained" + if tt.race != "" { + wantOther = "concurrent retained" + require.Equal(t, []byte("ON"), stored.Extended[s3_constants.ExtLegalHoldKey]) + } + if tt.race == "content" || tt.race == "replica" { + wantContent = "next" + var result CopyObjectResult + require.NoError(t, xml.Unmarshal(rr.Body.Bytes(), &result)) + require.Equal(t, `"next-etag"`, result.ETag) + } + require.Equal(t, []byte(wantContent), stored.Content) + require.Equal(t, []byte(wantOther), stored.Extended["internal-other"]) + if tt.replace { + require.NotContains(t, stored.Extended, "X-Amz-Meta-Source") + require.Equal(t, []byte("new"), stored.Extended["X-Amz-Meta-Destination"]) + } else { + require.Equal(t, []byte("kept"), stored.Extended["X-Amz-Meta-Source"]) + } + wantOwner := writer + if tt.ownership == s3_constants.OwnershipBucketOwnerEnforced || (tt.ownership == s3_constants.OwnershipBucketOwnerPreferred && tt.acl == "bucket-owner-full-control") { + wantOwner = bucketOwner + } + require.Equal(t, wantOwner, string(stored.Extended[s3_constants.ExtAmzOwnerKey])) + grants := GetAcpGrants(stored.Extended) + ownerFullControl := false + publicRead := false + customPermission := map[string]string{s3_constants.AmzAclRead: s3_constants.PermissionRead, s3_constants.AmzAclFullControl: s3_constants.PermissionFullControl, s3_constants.AmzAclReadAcp: s3_constants.PermissionReadAcp, s3_constants.AmzAclWriteAcp: s3_constants.PermissionWriteAcp}[tt.grantHeader] + for _, grant := range grants { + require.NotEqual(t, sourceOwner, aws.StringValue(grant.Grantee.ID), "source owner grants must not transfer") + ownerFullControl = ownerFullControl || (aws.StringValue(grant.Grantee.ID) == wantOwner && aws.StringValue(grant.Permission) == s3_constants.PermissionFullControl) + publicRead = publicRead || (aws.StringValue(grant.Grantee.URI) == s3_constants.GranteeGroupAllUsers && aws.StringValue(grant.Permission) == s3_constants.PermissionRead) + } + require.True(t, ownerFullControl) + require.Equal(t, tt.acl == "public-read", publicRead) + if customPermission != "" { + require.Len(t, grants, 2) + require.Equal(t, bucketOwner, aws.StringValue(grants[0].Grantee.ID)) + require.Equal(t, customPermission, aws.StringValue(grants[0].Permission)) + } else if tt.acl == "" || tt.acl == "private" { + require.Len(t, grants, 1, "private copies must not inherit any source grants") + } + wantMode := defaultFileMode + if tt.acl == "public-read" { + wantMode = 0644 + } + if tt.prefix { + wantMode |= uint32(os.ModeDir) + require.True(t, stored.IsDirectory) + require.Equal(t, []byte("true"), stored.Extended[s3_constants.SeaweedFSPrefixObject]) + require.Equal(t, []byte("child data"), filer.entries[sourcePath+"/child"].Content) + } + require.Equal(t, wantMode, stored.Attributes.FileMode) + if tt.race == "mime" { + require.Equal(t, "text/plain", stored.Attributes.Mime) + } + if tt.mime != "" { + require.Equal(t, tt.mime, stored.Attributes.Mime) + } + require.Empty(t, filer.transactions, "self-copies must not use an unguarded metadata PATCH") + if tt.self && tt.routed && tt.versioning == "" && !tt.sourceVersion { + attempts := 1 + if tt.race != "" && tt.race != "read" && tt.race != "replica" { + attempts = 2 + } + require.Len(t, filer.updates, attempts) + update := filer.updates[attempts-1] + require.Equal(t, filer_pb.WriteCondition_IF_ENTRY_EQUAL, update.Condition.Clauses[0].Kind) + require.Equal(t, stored.Extended[s3_constants.ExtAmzOwnerKey], update.Entry.Extended[s3_constants.ExtAmzOwnerKey]) + require.Equal(t, stored.Extended[s3_constants.ExtAmzAclKey], update.Entry.Extended[s3_constants.ExtAmzAclKey]) + } else { + require.Empty(t, filer.updates) + } + if !tt.self || tt.sourceVersion { + require.True(t, proto.Equal(originalSource, filer.entries[sourcePath]), "copy must preserve its source ACL and content") + } + }) + } +} diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index d1865b58c..67c027908 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -14,6 +14,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // objectWriteRouteKeyPrefix namespaces an object's full path into the ring key @@ -303,41 +305,42 @@ func (s3a *S3ApiServer) routedDelete(owner pb.ServerAddress, bucket, object stri }) } -// routedMetadataReplace applies a metadata-only self-copy (REPLACE directive) to -// an existing object in place via a routed PATCH_EXTENDED. The owner merges the -// new managed metadata onto a fresh read of the entry under its per-path lock — -// so a concurrent change to non-managed keys (legal hold, retention, version id) -// is preserved rather than clobbered by a whole-entry rewrite — and bumps mtime. -// updatedMetadata is the full managed-metadata set (processMetadataBytes); the -// delete list is the managed keys the replace dropped. -func (s3a *S3ApiServer) routedMetadataReplace(owner pb.ServerAddress, bucket, object string, current *filer_pb.Entry, updatedMetadata map[string][]byte) error { +// routedSelfCopy updates metadata and attributes together, guarded by the raw +// source snapshot. UpdateEntry routes conditional writes to the owner and checks +// the condition under its path lock, so concurrent content, ACL or retention +// updates cannot be overwritten by a stale self-copy. Chunks are reused in place. +func (s3a *S3ApiServer) routedSelfCopy(ctx context.Context, owner pb.ServerAddress, bucket, object string, current, updated *filer_pb.Entry) error { fullpath := util.NewFullPath(s3a.bucketDir(bucket), object) - dir, name := fullpath.DirAndName() - var del []string - for k := range current.Extended { - if isManagedCopyMetadataKey(k) { - if _, keep := updatedMetadata[k]; !keep { - del = append(del, k) - } + dir, _ := fullpath.DirAndName() + req := &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: updated, + Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ + Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, + ExpectedEntry: current, + }}}, + } + var conditionErr error + update := func(client filer_pb.SeaweedFilerClient) error { + err := filer_pb.UpdateEntry(ctx, client, req) + switch status.Code(err) { + case codes.FailedPrecondition, codes.NotFound: + // These are authoritative replies from a healthy owner, not a + // transport failure. Do not replay them or mark the owner down. + conditionErr = err + return nil + default: + return err } } - resp, err := s3a.objectTxnOnFiler(owner, &filer_pb.ObjectTransactionRequest{ - LockKey: string(fullpath), - RouteKey: s3a.objectRouteKey(bucket, object), - Mutations: []*filer_pb.ObjectMutation{{ - Type: filer_pb.ObjectMutation_PATCH_EXTENDED, - Directory: dir, - Name: name, - SetExtended: updatedMetadata, - DeleteExtended: del, - TouchMtime: true, - }}, - }) + var err error + if s3a.filerClient == nil { + err = pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, update) + } else { + err = s3a.withFilerClientFailover(ctx, owner, false, update) + } if err != nil { return err } - if resp.Error != "" { - return fmt.Errorf("routed metadata replace %s/%s: %s", bucket, object, resp.Error) - } - return nil + return conditionErr } diff --git a/weed/s3api/s3api_object_upload_acl.go b/weed/s3api/s3api_object_upload_acl.go index a87977e3e..929b79165 100644 --- a/weed/s3api/s3api_object_upload_acl.go +++ b/weed/s3api/s3api_object_upload_acl.go @@ -14,7 +14,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" ) -// putObjectACLContextKey carries validated ACL metadata to every PutObject write +// putObjectACLContextKey carries validated ACL metadata to each object creation // path without exposing an internal header that a client could forge. type putObjectACLContextKey struct{} @@ -61,20 +61,11 @@ func putObjectACLValue(r *http.Request, query url.Values, header string) (string return value, s3err.ErrNone } -// putObjectACLPolicyRequest exposes effective PUT ACLs to policy conditions only +// putObjectACLPolicyRequest exposes upload and copy ACLs to policy conditions only // after authentication. Other operations keep their original request semantics. func putObjectACLPolicyRequest(r *http.Request, action Action, bucket, object string) (*http.Request, s3err.ErrorCode) { - // Copy routes match any repeated header value, so checking only the first line can misclassify a copy as a regular upload. - copyRequest := false - for _, copySource := range r.Header.Values("X-Amz-Copy-Source") { - if strings.Contains(copySource, "/") || strings.Contains(strings.ToLower(copySource), "%2f") { - copyRequest = true - break - } - } if (action != s3_constants.ACTION_WRITE && action != s3_constants.ACTION_WRITE_ACP) || r.Method != http.MethodPut || object == "" || object == "/" || - copyRequest || ResolveS3Action(r, string(s3_constants.ACTION_WRITE), bucket, object) != s3_constants.S3_ACTION_PUT_OBJECT { return r, s3err.ErrNone } @@ -128,8 +119,8 @@ func putObjectACLPolicyRequest(r *http.Request, action Action, bucket, object st return policyRequest, s3err.ErrNone } -// preparePutObjectACL validates and authorizes ACLs before the upload body is -// consumed. The resulting metadata is committed in the same entry as the object. +// preparePutObjectACL validates and authorizes ACLs before upload or copy data +// is written. The metadata is committed in the same entry as the new object. func (s3a *S3ApiServer) preparePutObjectACL(r *http.Request, bucket string) (*http.Request, s3err.ErrorCode) { metadata, code := s3a.getBucketConfig(bucket) if code != s3err.ErrNone { @@ -164,7 +155,7 @@ func (s3a *S3ApiServer) preparePutObjectACL(r *http.Request, bucket string) (*ht if !s3a.iam.isEnabled() { accountID = AccountAdmin.Id } else if explicit { - // Setting an ACL during PutObject also requires s3:PutObjectAcl. Use the + // Setting an ACL during upload or copy also requires s3:PutObjectAcl. Use the // unified authorization path so bucket-policy allows and explicit denies // retain the same semantics as standalone ACL requests. identity, authCode := s3a.iam.authRequest(r.Clone(r.Context()), s3_constants.ACTION_WRITE_ACP) diff --git a/weed/s3api/s3api_object_upload_acl_test.go b/weed/s3api/s3api_object_upload_acl_test.go index 2d6bcf93f..067145376 100644 --- a/weed/s3api/s3api_object_upload_acl_test.go +++ b/weed/s3api/s3api_object_upload_acl_test.go @@ -663,7 +663,7 @@ func TestPutObjectUploadACL(t *testing.T) { } // TestPutObjectACLPolicyScope ensures query normalization cannot alter other -// operations or the original signed request passed to the upload handler. +// operations or the original signed request passed to the upload/copy handler. func TestPutObjectACLPolicyScope(t *testing.T) { tests := []struct { name, method, object, subresource string @@ -676,8 +676,10 @@ func TestPutObjectACLPolicyScope(t *testing.T) { {name: "upload acl authorization", method: http.MethodPut, object: "key", action: s3_constants.ACTION_WRITE_ACP, wantACL: "public-read"}, {name: "bucket", method: http.MethodPut, action: s3_constants.ACTION_WRITE}, {name: "post form", method: http.MethodPost, object: "key", action: s3_constants.ACTION_WRITE}, - {name: "copy", method: http.MethodPut, object: "key", action: s3_constants.ACTION_WRITE, copy: true}, - {name: "repeated copy source", method: http.MethodPut, object: "key", action: s3_constants.ACTION_WRITE, repeatedCopy: true}, + {name: "copy", method: http.MethodPut, object: "key", action: s3_constants.ACTION_WRITE, copy: true, wantACL: "public-read"}, + {name: "copy acl authorization", method: http.MethodPut, object: "key", action: s3_constants.ACTION_WRITE_ACP, copy: true, wantACL: "public-read"}, + {name: "repeated copy source", method: http.MethodPut, object: "key", action: s3_constants.ACTION_WRITE, repeatedCopy: true, wantACL: "public-read"}, + {name: "multipart part copy", method: http.MethodPut, object: "key", subresource: "uploadId=upload&partNumber=1", action: s3_constants.ACTION_WRITE, copy: true}, {name: "multipart part", method: http.MethodPut, object: "key", subresource: "uploadId=upload&partNumber=1", action: s3_constants.ACTION_WRITE}, {name: "multipart leading zero", method: http.MethodPut, object: "key", subresource: "uploadId=upload&partNumber=01", action: s3_constants.ACTION_WRITE}, {name: "upload id only", method: http.MethodPut, object: "key", subresource: "uploadId=upload", action: s3_constants.ACTION_WRITE, wantACL: "public-read"}, diff --git a/weed/s3api/s3err/s3api_errors.go b/weed/s3api/s3err/s3api_errors.go index 302bb0665..a040a59b0 100644 --- a/weed/s3api/s3err/s3api_errors.go +++ b/weed/s3api/s3err/s3api_errors.go @@ -172,6 +172,8 @@ const ( ErrRenameDestinationSameAsSource ErrIdempotentParameterMismatch ErrAccessControlListNotSupported + // The request's snapshot changed during a conditional write; clients can retry. + ErrConditionalRequestConflict ) // Error message constants for checksum validation @@ -591,6 +593,11 @@ var errorCodeResponse = map[ErrorCode]APIError{ Description: "At least one of the pre-conditions you specified did not hold", HTTPStatusCode: http.StatusPreconditionFailed, }, + ErrConditionalRequestConflict: { + Code: "ConditionalRequestConflict", + Description: "A conflicting conditional operation is currently in progress against this resource. Please try again.", + HTTPStatusCode: http.StatusConflict, + }, ErrNotModified: { Code: "NotModified", Description: "The object was not modified since the specified time", diff --git a/weed/server/filer_grpc_server_self_copy_condition_test.go b/weed/server/filer_grpc_server_self_copy_condition_test.go new file mode 100644 index 000000000..dafd66d94 --- /dev/null +++ b/weed/server/filer_grpc_server_self_copy_condition_test.go @@ -0,0 +1,87 @@ +package weed_server + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +// Exercise the native condition and UpdateEntry commit, independently of the +// S3 handler's fake filer: same-content changes must invalidate a whole snapshot. +func TestSelfCopyEntryCondition(t *testing.T) { + tests := []struct { + name string + change func(*filer_pb.Entry) + }{ + {name: "unchanged control"}, + {name: "mode", change: func(e *filer_pb.Entry) { e.Attributes.FileMode = 0644 }}, + {name: "mime", change: func(e *filer_pb.Entry) { e.Attributes.Mime = "image/png" }}, + {name: "legal hold", change: func(e *filer_pb.Entry) { e.Extended[s3_constants.ExtLegalHoldKey] = []byte("ON") }}, + {name: "content", change: func(e *filer_pb.Entry) { + e.Content = []byte("next") + e.Extended[s3_constants.ExtETagKey] = []byte("next-etag") + }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fs, store := newTxnTestServer(t, map[string]*filer.Entry{ + "/buckets/b/obj": { + Attr: filer.Attr{Inode: 1, Mtime: time.Unix(1700000000, 0), Crtime: time.Unix(1700000000, 0), Ctime: time.Unix(1700000000, 0), Atime: time.Unix(1700000000, 0), Mode: 0660, Mime: "text/plain", FileSize: 4}, + Content: []byte("data"), + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("same-etag"), s3_constants.ExtAmzAclKey: []byte("original grants")}, + }, + }) + expected := store.entries["/buckets/b/obj"].ToProtoEntry() + updated := proto.Clone(expected).(*filer_pb.Entry) + updated.Extended[s3_constants.ExtAmzAclKey] = []byte("private grants") + req := &filer_pb.UpdateEntryRequest{ + Directory: "/buckets/b", Entry: updated, + Condition: one(&filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: expected}), + } + if tt.change != nil { + concurrent := proto.Clone(expected).(*filer_pb.Entry) + tt.change(concurrent) + if _, err := fs.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{Directory: "/buckets/b", Entry: concurrent}); err != nil { + t.Fatal(err) + } + } + _, err := fs.UpdateEntry(context.Background(), req) + if tt.change != nil { + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("stale self-copy: want FailedPrecondition, got %v", err) + } + current := store.entries["/buckets/b/obj"].ToProtoEntry() + if string(current.Extended[s3_constants.ExtAmzAclKey]) != "original grants" { + t.Fatal("failed condition changed grants") + } + // A fresh retry preserves the intervening content and unmanaged keys + // while setting the desired mode and grants in the same update. + req.Condition.Clauses[0].ExpectedEntry = current + req.Entry = proto.Clone(current).(*filer_pb.Entry) + req.Entry.Attributes.FileMode = 0660 + req.Entry.Extended[s3_constants.ExtAmzAclKey] = []byte("private grants") + _, err = fs.UpdateEntry(context.Background(), req) + } + if err != nil { + t.Fatalf("fresh self-copy: %v", err) + } + stored := store.entries["/buckets/b/obj"].ToProtoEntry() + if stored.Attributes.FileMode != 0660 || string(stored.Extended[s3_constants.ExtAmzAclKey]) != "private grants" { + t.Fatal("mode and grants did not commit together") + } + if tt.name == "legal hold" && string(stored.Extended[s3_constants.ExtLegalHoldKey]) != "ON" { + t.Fatal("retry lost legal hold") + } + if tt.name == "content" && string(stored.Content) != "next" { + t.Fatal("retry resurrected stale content") + } + }) + } +}