From 4850def831f14365461b77ac942b05a387f10b97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20L=E1=BB=99c=20Ph=C3=BAc?= <112325531+omninguyen@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:50:45 +0700 Subject: [PATCH] fix(s3tables): bind table and view writes to the record they were authorized against (#12328) * s3tables: bind table and view writes to the record they were authorized against * s3tables: bind UpdateNamespace to the namespace it was authorized against UpdateNamespace writes back the whole record it read, owner and creation time included. A namespace deleted and created again before the write took the first one's owner and CreatedAt, which also undoes the CreatedAt binding creates now rely on. Refuse the write with 409 when the namespace at the path has another CreatedAt. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Chris Lu Co-authored-by: Claude Opus 5.5 --- weed/s3api/s3tables/filer_ops.go | 31 + .../s3tables/handler_bind_authorized_test.go | 585 ++++++++++++++++++ .../handler_delete_create_race_test.go | 19 + weed/s3api/s3tables/handler_maintenance.go | 35 +- weed/s3api/s3tables/handler_namespace.go | 5 + weed/s3api/s3tables/handler_policy.go | 49 +- weed/s3api/s3tables/handler_table.go | 54 +- .../s3tables/handler_update_conflict_test.go | 20 + weed/s3api/s3tables/handler_view.go | 50 +- 9 files changed, 796 insertions(+), 52 deletions(-) create mode 100644 weed/s3api/s3tables/handler_bind_authorized_test.go diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 774743dd69..3775621a6a 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -643,6 +643,23 @@ func isTableBucketCreatedAt(createdAt time.Time) func(*filer_pb.Entry) bool { } } +// isNamespaceCreatedAt is isTableBucketCreatedAt for a namespace its create has +// confirmed: one deleted and created again under the same name reads as the +// authorized one gone. A bucket created again holds only namespaces of its own, +// so a create bound to its namespace this way is bound to the bucket too. +func isNamespaceCreatedAt(createdAt time.Time) func(*filer_pb.Entry) bool { + return func(entry *filer_pb.Entry) bool { + if !isFinishedNamespaceEntry(entry) { + return false + } + var metadata namespaceMetadata + if json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata) != nil { + return false + } + return metadata.CreatedAt.Equal(createdAt) + } +} + // isNamespaceEntry reports whether entry carries a namespace's catalog record. func isNamespaceEntry(entry *filer_pb.Entry) bool { _, ok := entry.Extended[ExtendedKeyMetadata] @@ -741,6 +758,20 @@ func (h *S3TablesHandler) mutateCatalogRecord( }) } +// requireTableCreatedAt refuses a write to the record at path unless it is the +// table created at createdAt, the one the caller was authorized against: a +// table deleted and created again under the name has a CreatedAt of its own. +// Only the creation time is compared, so the commits an Iceberg writer lands +// meanwhile do not turn a write of the table's policy, tags or maintenance +// settings into a conflict. +func requireTableCreatedAt(extended map[string][]byte, path string, createdAt time.Time) error { + var current tableMetadataInternal + if json.Unmarshal(extended[ExtendedKeyMetadata], ¤t) != nil || !current.CreatedAt.Equal(createdAt) { + return fmt.Errorf("%w: %s holds another table", ErrConcurrentUpdate, path) + } + return nil +} + // confirmParent re-checks the parent after a create wrote written onto // childPath. If the parent is open, it clears the record's unfinished mark — // then checks the parent once more, since a delete claiming it in between diff --git a/weed/s3api/s3tables/handler_bind_authorized_test.go b/weed/s3api/s3tables/handler_bind_authorized_test.go new file mode 100644 index 0000000000..ce48bdf3e1 --- /dev/null +++ b/weed/s3api/s3tables/handler_bind_authorized_test.go @@ -0,0 +1,585 @@ +package s3tables + +import ( + "context" + "encoding/json" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" +) + +// recreatedNamespace is the record of a namespace deleted and created again +// under name: another owner, and a creation time of its own. +func recreatedNamespace(t *testing.T, name string) map[string][]byte { + t.Helper() + meta, err := json.Marshal(namespaceMetadata{ + Namespace: []string{name}, + OwnerAccountID: "111122223333", + CreatedAt: time.Now().Add(time.Hour), + }) + require.NoError(t, err) + return map[string][]byte{ExtendedKeyMetadata: meta} +} + +// replaceNamespaceAtLookup replaces namespace with a recreated one at the first +// lookup of namespace/child, the create's look for its own name, which comes +// after the create was authorized against the first namespace. +func replaceNamespaceAtLookup(t *testing.T, fs *s3tablestest.MemFiler, namespace, child string) map[string][]byte { + t.Helper() + bucketPath := GetTableBucketPath(renameTestBucket) + recreated := recreatedNamespace(t, namespace) + fs.OnLookup = func(dir, name string) { + if dir != GetNamespacePath(renameTestBucket, namespace) || name != child { + return + } + fs.OnLookup = nil + fs.Put(bucketPath, namespace, recreated) + } + return recreated +} + +// A create is authorized against the namespace it read. If that namespace is +// deleted and another is created under the same name before the create writes, +// the entry would land in a namespace nobody authorized it for, carrying the +// first one's owner. The create checks the namespace by its creation time, +// reads the new one as the old one gone, and backs out what it wrote. +func TestCreateInNamespaceTreatsARecreatedNamespaceAsGone(t *testing.T) { + for _, tc := range []struct { + operation string + input func(t *testing.T) interface{} + }{ + {"CreateTable", func(t *testing.T) interface{} { return createTableRequest(t, "ns", "t") }}, + {"RegisterTable", func(t *testing.T) interface{} { + return &RegisterTableRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + MetadataLocation: "s3://" + renameTestBucket + "/ns/t/metadata/v1.metadata.json", + } + }}, + {"CreateView", func(t *testing.T) interface{} { return createViewRequest(t, "t") }}, + } { + t.Run(tc.operation, func(t *testing.T) { + fs, m := startEmptyBucket(t) + bucketPath := GetTableBucketPath(renameTestBucket) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + recreated := replaceNamespaceAtLookup(t, fs, "ns", "t") + + code, resp := runHandler(t, m, fs.Client, tc.operation, tc.input(t)) + + assert.Equal(t, http.StatusNotFound, code, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp) + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns"), "t"), "nothing is left in the new namespace") + require.NotNil(t, fs.Get(bucketPath, "ns"), "the new namespace stays") + assert.Equal(t, recreated, fs.Get(bucketPath, "ns").Extended, "and is left as it was created") + }) + } +} + +// A rename is authorized to create in the destination namespace it read. If +// that namespace is deleted and created again before the rename publishes, the +// renamed table would land in a namespace nobody authorized it for. The rename +// reads the new one as the old one gone, backs out, and keeps the source. +func TestRenameTableTreatsARecreatedDestinationAsGone(t *testing.T) { + fs, m := startRenameManager(t) + bucketPath := GetTableBucketPath(renameTestBucket) + nsPath := GetNamespacePath(renameTestBucket, "ns") + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns2"}, + }) + source := fs.Get(nsPath, "t").Extended + // The rename looks the destination name up right after it read the + // destination namespace. + recreated := replaceNamespaceAtLookup(t, fs, "ns2", "t") + + code, resp := runHandler(t, m, fs.Client, "RenameTable", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "t", + DestNamespace: []string{"ns2"}, + DestName: "t", + }) + + assert.Equal(t, http.StatusNotFound, code, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp) + assert.Nil(t, fs.Get(GetNamespacePath(renameTestBucket, "ns2"), "t"), "nothing is left in the new namespace") + require.NotNil(t, fs.Get(bucketPath, "ns2"), "the new namespace stays") + assert.Equal(t, recreated, fs.Get(bucketPath, "ns2").Extended, "and is left as it was created") + require.NotNil(t, fs.Get(nsPath, "t"), "the source stays") + assert.Equal(t, source, fs.Get(nsPath, "t").Extended, "with its record and no claim") +} + +// recreatedRecord is the record of a table or view deleted and created again +// under ns/name: another owner, and a creation time of its own. +func recreatedRecord(t *testing.T, name, entryType string) map[string][]byte { + t.Helper() + now := time.Now().Add(time.Hour) + meta, err := json.Marshal(tableMetadataInternal{ + Name: name, + Namespace: "ns", + Format: FormatIceberg, + CreatedAt: now, + ModifiedAt: now, + OwnerAccountID: "111122223333", + VersionToken: generateVersionToken(), + MetadataVersion: 1, + MetadataLocation: "s3://" + renameTestBucket + "/ns/" + name + "/metadata/v1-recreated.metadata.json", + }) + require.NoError(t, err) + return map[string][]byte{ + ExtendedKeyMetadata: meta, + ExtendedKeyEntryType: []byte(entryType), + } +} + +// startViewV seeds the namespace ns holding the view v. +func startViewV(t *testing.T) (*s3tablestest.MemFiler, *Manager) { + t.Helper() + fs, m := startEmptyBucket(t) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + requireOK(t, m, fs, "CreateView", createViewRequest(t, "v")) + return fs, m +} + +func updateViewRequest(t *testing.T, token, version string) *UpdateViewRequest { + return &UpdateViewRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "v", + VersionToken: token, + MetadataLocation: "s3://" + renameTestBucket + "/ns/v/metadata/" + version + ".metadata.json", + } +} + +// An update is authorized against the view it read. If that view is deleted +// and another is created under the same name before the update writes, the +// write would land on a view nobody authorized it for. The write is conditional +// on the record the update read, so it is refused and the new view kept. +func TestUpdateViewSparesARecreatedView(t *testing.T) { + fs, m := startViewV(t) + nsPath := GetNamespacePath(renameTestBucket, "ns") + recreated := recreatedRecord(t, "v", EntryTypeView) + // The update's write is its first; the view is replaced as it lands, so the + // write loses and its retry reads the new view. + fs.BeforeUpdate = func() { fs.Put(nsPath, "v", recreated) } + + code, resp := runHandler(t, m, fs.Client, "UpdateView", updateViewRequest(t, "", "v2")) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(nsPath, "v"), "the new view stays") + assert.Equal(t, recreated, fs.Get(nsPath, "v").Extended, "and is left as it was created") +} + +// Two updates that read the same version both pass the version-token check. +// The second to write must not overwrite the first one's metadata pointer, or +// the version the first published is lost. +func TestUpdateViewRejectsLostUpdate(t *testing.T) { + fs, m := startViewV(t) + token := versionTokenOf(t, fs, "v") + var winnerStatus int + var winnerResp S3TablesError + // The winner runs whole between the loser's read and its write. + fs.BeforeUpdate = func() { + winnerStatus, winnerResp = runHandler(t, m, fs.Client, "UpdateView", updateViewRequest(t, token, "v2-winner")) + } + + code, resp := runHandler(t, m, fs.Client, "UpdateView", updateViewRequest(t, token, "v2-loser")) + + require.Equal(t, http.StatusOK, winnerStatus, "%+v", winnerResp) + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + got := &GetViewResponse{} + require.NoError(t, m.Execute(context.Background(), NewManagerClient(fs.Client), "GetView", getViewRequest(t, "v"), got, "")) + assert.Equal(t, updateViewRequest(t, token, "v2-winner").MetadataLocation, got.MetadataLocation, "the first update must survive") +} + +// A policy written between the authorization check and the write decides +// whether this caller may still update the view, so the update is refused +// rather than applied on a decision that has been overtaken. +func TestUpdateViewRejectsWhenTheAuthorizingPolicyChanges(t *testing.T) { + fs, m := startViewV(t) + nsPath := GetNamespacePath(renameTestBucket, "ns") + policy := []byte(`{"Version":"2012-10-17","Statement":[]}`) + fs.BeforeUpdate = func() { + extended := map[string][]byte{} + for key, value := range fs.Get(nsPath, "v").Extended { + extended[key] = value + } + extended[ExtendedKeyPolicy] = policy + fs.Put(nsPath, "v", extended) + } + + code, resp := runHandler(t, m, fs.Client, "UpdateView", updateViewRequest(t, "", "v2")) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + assert.Equal(t, policy, fs.Get(nsPath, "v").Extended[ExtendedKeyPolicy], "the concurrent policy write must survive") +} + +const bindTestPolicy = `{"Statement":[{"Effect":"Allow","Principal":"writer","Action":"s3tables:*","Resource":"*"}]}` + +// tableWriteOp is a write to the table ns/t, with the attributes the table +// carries for it besides its record. +type tableWriteOp struct { + operation string + input func(t *testing.T) interface{} + extra map[string][]byte +} + +func tableWriteOps() []tableWriteOp { + tableARN := func(t *testing.T) string { + arn, err := BuildTableARN(DefaultRegion, DefaultAccountID, renameTestBucket, "ns", "t") + require.NoError(t, err) + return arn + } + return []tableWriteOp{ + {operation: "PutTablePolicy", input: func(t *testing.T) interface{} { + return &PutTablePolicyRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t", ResourcePolicy: bindTestPolicy} + }}, + {operation: "DeleteTablePolicy", input: func(t *testing.T) interface{} { + return &DeleteTablePolicyRequest{TableBucketARN: mustBucketARN(t), Namespace: []string{"ns"}, Name: "t"} + }, extra: map[string][]byte{ExtendedKeyPolicy: []byte(bindTestPolicy)}}, + {operation: "TagResource", input: func(t *testing.T) interface{} { + return &TagResourceRequest{ResourceARN: tableARN(t), Tags: map[string]string{"k": "v"}} + }}, + {operation: "UntagResource", input: func(t *testing.T) interface{} { + return &UntagResourceRequest{ResourceARN: tableARN(t), TagKeys: []string{"k"}} + }, extra: map[string][]byte{ExtendedKeyTags: []byte(`{"k":"v","keep":"1"}`)}}, + {operation: "PutTableMaintenanceConfiguration", input: func(t *testing.T) interface{} { + return &PutTableMaintenanceConfigurationRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + Type: MaintenanceTypeIcebergCompaction, + Value: &MaintenanceConfigurationValue{Status: MaintenanceStatusEnabled}, + } + }}, + } +} + +// startTableT seeds the namespace ns holding the table t, carrying extra +// besides its record. +func startTableT(t *testing.T, extra map[string][]byte) (*s3tablestest.MemFiler, *Manager) { + t.Helper() + fs, m := startRenameManager(t) + nsPath := GetNamespacePath(renameTestBucket, "ns") + extended := extendedOf(t, fs, nsPath, "t") + for key, value := range extra { + extended[key] = value + } + fs.Put(nsPath, "t", extended) + return fs, m +} + +// A write to a table's policy, tags or maintenance settings is authorized +// against the table it read. If that table is deleted and another is created +// under the same name before the write, it would land on a table nobody +// authorized it for: the policy or tag values it compares can match the new +// table's just as well. The write checks the table by its creation time and is +// refused, and the new table is kept as it was created. +func TestTableWritesSpareARecreatedTable(t *testing.T) { + for _, op := range tableWriteOps() { + t.Run(op.operation, func(t *testing.T) { + fs, m := startTableT(t, op.extra) + nsPath := GetNamespacePath(renameTestBucket, "ns") + recreated := recreatedRecord(t, "t", EntryTypeTable) + for key, value := range op.extra { + recreated[key] = value + } + // The write is the operation's first; the table is replaced as it + // lands, so the write loses and its retry reads the new table. + fs.BeforeUpdate = func() { fs.Put(nsPath, "t", recreated) } + + code, resp := runHandler(t, m, fs.Client, op.operation, op.input(t)) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + require.NotNil(t, fs.Get(nsPath, "t"), "the new table stays") + assert.Equal(t, recreated, fs.Get(nsPath, "t").Extended, "and is left as it was created") + }) + } +} + +// The same writes compare only the table's creation time, not its metadata: +// an Iceberg commit to the table between the read and the write is the same +// table, and must not turn the write into a conflict. +func TestTableWritesTolerateACommitInBetween(t *testing.T) { + for _, op := range tableWriteOps() { + t.Run(op.operation, func(t *testing.T) { + fs, m := startTableT(t, op.extra) + nsPath := GetNamespacePath(renameTestBucket, "ns") + committedLocation := "s3://" + renameTestBucket + "/ns/t/metadata/v4.metadata.json" + fs.BeforeUpdate = func() { + extended := extendedOf(t, fs, nsPath, "t") + var metadata tableMetadataInternal + require.NoError(t, json.Unmarshal(extended[ExtendedKeyMetadata], &metadata)) + metadata.MetadataVersion = 4 + metadata.MetadataLocation = committedLocation + metadata.VersionToken = generateVersionToken() + metadata.ModifiedAt = time.Now() + data, err := json.Marshal(metadata) + require.NoError(t, err) + extended[ExtendedKeyMetadata] = data + fs.Put(nsPath, "t", extended) + } + + requireOK(t, m, fs, op.operation, op.input(t)) + + got, err := runGetTable(t, m, fs, "ns", "t") + require.NoError(t, err) + assert.Equal(t, committedLocation, got.MetadataLocation, "the commit stays") + }) + } +} + +// namespaceMetadataOf reads the record of a namespace in the test bucket. +func namespaceMetadataOf(t *testing.T, fs *s3tablestest.MemFiler, namespace string) namespaceMetadata { + t.Helper() + entry := fs.Get(GetTableBucketPath(renameTestBucket), namespace) + require.NotNil(t, entry) + var metadata namespaceMetadata + require.NoError(t, json.Unmarshal(entry.Extended[ExtendedKeyMetadata], &metadata)) + return metadata +} + +// updateNamespaceAtLookup runs a real UpdateNamespace the first time child is +// looked up under namespace: after the request was authorized against the +// namespace and before it writes. The answer fills in once the update has run. +func updateNamespaceAtLookup(t *testing.T, fs *s3tablestest.MemFiler, m *Manager, namespace, child string) *handlerResult { + t.Helper() + update := &handlerResult{} + fs.OnLookup = func(dir, name string) { + if dir != GetNamespacePath(renameTestBucket, namespace) || name != child { + return + } + fs.OnLookup = nil + update.status, update.resp = runHandler(t, m, fs.Client, "UpdateNamespace", &UpdateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{namespace}, + Properties: map[string]string{"edited": "true"}, + }) + } + return update +} + +// requireNamespaceUpdatedInPlace checks that the update ran, took effect, and +// kept the namespace the record it was: same owner, same creation time. +func requireNamespaceUpdatedInPlace(t *testing.T, fs *s3tablestest.MemFiler, namespace string, before namespaceMetadata, update *handlerResult) { + t.Helper() + require.NotZero(t, update.status, "the update has to land between the authorization and the write") + require.Equal(t, http.StatusOK, update.status, "%+v", update.resp) + after := namespaceMetadataOf(t, fs, namespace) + assert.Equal(t, map[string]string{"edited": "true"}, after.Properties) + assert.True(t, after.CreatedAt.Equal(before.CreatedAt), "an update must keep the creation time: %v became %v", before.CreatedAt, after.CreatedAt) + assert.Equal(t, before.OwnerAccountID, after.OwnerAccountID) +} + +// A namespace update edits the record a request was authorized against; it +// does not replace it. The creation time survives, so the request still writes: +// reading the edit as a delete and create would back the entry out and answer +// 404 for a namespace that never went away. Unlike +// TestCreateConfirmingAcrossANamespaceUpdateSucceeds, the update here is the +// real UpdateNamespace and lands before the first check against the authorized +// record, not after it. +func TestCreateAuthorizedBeforeANamespaceUpdateSucceeds(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startEmptyBucket(t) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + before := namespaceMetadataOf(t, fs, "ns") + update := updateNamespaceAtLookup(t, fs, m, "ns", lateEntry) + + code, resp := runHandler(t, m, fs.Client, op.operation, op.input) + + assert.Equal(t, http.StatusOK, code, "%+v", resp) + requireNamespaceUpdatedInPlace(t, fs, "ns", before, update) + entry := fs.Get(GetNamespacePath(renameTestBucket, "ns"), lateEntry) + require.NotNil(t, entry, "the entry the request answered for must exist") + assert.Contains(t, entry.Extended, ExtendedKeyMetadata) + assert.NotContains(t, entry.Extended, ExtendedKeyDeleting, "and carry no unfinished mark") + }) + } + + // A rename is bound to its destination namespace the same way. + for _, tc := range []struct { + operation string + name string + seed func(t *testing.T, fs *s3tablestest.MemFiler, name string) + }{ + {"RenameTable", "t", nil}, + {"RenameView", "v", seedView}, + } { + t.Run(tc.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + if tc.seed != nil { + tc.seed(t, fs, tc.name) + } + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns2"}, + }) + before := namespaceMetadataOf(t, fs, "ns2") + update := updateNamespaceAtLookup(t, fs, m, "ns2", tc.name) + + code, resp := runHandler(t, m, fs.Client, tc.operation, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: tc.name, + DestNamespace: []string{"ns2"}, + DestName: tc.name, + }) + + assert.Equal(t, http.StatusOK, code, "%+v", resp) + requireNamespaceUpdatedInPlace(t, fs, "ns2", before, update) + dest := fs.Get(GetNamespacePath(renameTestBucket, "ns2"), tc.name) + require.NotNil(t, dest, "the renamed entry must exist") + assert.Contains(t, dest.Extended, ExtendedKeyMetadata) + assert.NotContains(t, dest.Extended, ExtendedKeyDeleting, "and carry no unfinished mark") + }) + } +} + +// writeRecorder remembers the writes that reach a directory tree, so a test can +// tell a request that never wrote there from one that wrote and backed out. +type writeRecorder struct { + filer_pb.SeaweedFilerClient + tree string + mu sync.Mutex + writes []string +} + +func (c *writeRecorder) note(op, dir, name string) { + if dir != c.tree && !strings.HasPrefix(dir, c.tree+"/") { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.writes = append(c.writes, op+" "+dir+"/"+name) +} + +func (c *writeRecorder) CreateEntry(ctx context.Context, in *filer_pb.CreateEntryRequest, opts ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + c.note("create", in.Directory, in.Entry.GetName()) + return c.SeaweedFilerClient.CreateEntry(ctx, in, opts...) +} + +func (c *writeRecorder) UpdateEntry(ctx context.Context, in *filer_pb.UpdateEntryRequest, opts ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + c.note("update", in.Directory, in.Entry.GetName()) + return c.SeaweedFilerClient.UpdateEntry(ctx, in, opts...) +} + +func (c *writeRecorder) DeleteEntry(ctx context.Context, in *filer_pb.DeleteEntryRequest, opts ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + c.note("delete", in.Directory, in.Name) + return c.SeaweedFilerClient.DeleteEntry(ctx, in, opts...) +} + +// A namespace deleted and created again between the authorization and the +// early check is not the one the caller was authorized against, so the request +// is refused before it writes anything: the check after the write only backs +// the record out, and a back-out the filer turns away would leave it in a +// namespace nobody authorized it for. +func TestCreateRefusedBeforeWritingIntoAReplacedNamespace(t *testing.T) { + for _, op := range createOps(t) { + t.Run(op.operation, func(t *testing.T) { + fs, m := startEmptyBucket(t) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + bucketPath := GetTableBucketPath(renameTestBucket) + recreated := recreatedNamespace(t, "ns") + // The create reads the namespace record three times before it writes: for + // its metadata, for its policy, and for the early check. The namespace is + // replaced as the third read starts. + const earlyCheck = 3 + reads := 0 + fs.OnLookup = func(dir, name string) { + if dir != bucketPath || name != "ns" { + return + } + if reads++; reads == earlyCheck { + fs.OnLookup = nil + fs.Put(bucketPath, "ns", recreated) + } + } + recorder := &writeRecorder{SeaweedFilerClient: fs.Client, tree: GetNamespacePath(renameTestBucket, "ns")} + + code, resp := runHandler(t, m, recorder, op.operation, op.input) + + require.GreaterOrEqual(t, reads, earlyCheck, "the namespace has to be replaced before the early check") + assert.Equal(t, http.StatusNotFound, code, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp) + assert.Empty(t, recorder.writes, "nothing is written into the namespace that replaced the authorized one") + assert.Equal(t, recreated, fs.Get(bucketPath, "ns").Extended, "and it is left as it was created") + }) + } + + // A rename reads its destination namespace and then the destination name, + // both before the early check. + for _, tc := range []struct { + operation string + name string + seed func(t *testing.T, fs *s3tablestest.MemFiler, name string) + }{ + {"RenameTable", "t", nil}, + {"RenameView", "v", seedView}, + } { + t.Run(tc.operation, func(t *testing.T) { + fs, m := startRenameManager(t) + if tc.seed != nil { + tc.seed(t, fs, tc.name) + } + requireOK(t, m, fs, "CreateNamespace", &CreateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns2"}, + }) + recreated := replaceNamespaceAtLookup(t, fs, "ns2", tc.name) + recorder := &writeRecorder{SeaweedFilerClient: fs.Client, tree: GetNamespacePath(renameTestBucket, "ns2")} + + code, resp := runHandler(t, m, recorder, tc.operation, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: tc.name, + DestNamespace: []string{"ns2"}, + DestName: tc.name, + }) + + assert.Equal(t, http.StatusNotFound, code, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchNamespace, resp.Type, "%+v", resp) + assert.Empty(t, recorder.writes, "nothing is written into the namespace that replaced the authorized one") + assert.Equal(t, recreated, fs.Get(GetTableBucketPath(renameTestBucket), "ns2").Extended, "and it is left as it was created") + }) + } +} + +// UpdateNamespace writes the record it was authorized against, owner and +// creation time included. A namespace deleted and created again before that +// write is refused, not handed the first one's owner. +func TestUpdateNamespaceSparesARecreatedNamespace(t *testing.T) { + fs, m := startEmptyBucket(t) + requireOK(t, m, fs, "CreateNamespace", createNamespaceNsRequest(t)) + bucketPath := GetTableBucketPath(renameTestBucket) + recreated := recreatedNamespace(t, "ns") + fs.OnUpdate = func(dir, name string) { + if dir != bucketPath || name != "ns" { + return + } + fs.OnUpdate = nil + fs.Put(bucketPath, "ns", recreated) + } + + code, resp := runHandler(t, m, fs.Client, "UpdateNamespace", &UpdateNamespaceRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Properties: map[string]string{"edited": "true"}, + }) + + assert.Equal(t, http.StatusConflict, code, "%+v", resp) + assert.Equal(t, recreated, fs.Get(bucketPath, "ns").Extended, "the new namespace is untouched") +} diff --git a/weed/s3api/s3tables/handler_delete_create_race_test.go b/weed/s3api/s3tables/handler_delete_create_race_test.go index b46b9fb8d0..7f51145d10 100644 --- a/weed/s3api/s3tables/handler_delete_create_race_test.go +++ b/weed/s3api/s3tables/handler_delete_create_race_test.go @@ -512,6 +512,25 @@ func TestUpdateRefusedOnAMarkedRecord(t *testing.T) { assert.Equal(t, ErrCodeNoSuchTable, resp.Type) } +// DeleteTablePolicy refuses a marked record the same way: the policy it would +// remove sits on an entry every reader reports absent. +func TestDeleteTablePolicyRefusedOnAMarkedRecord(t *testing.T) { + fs, m := startTableT(t, map[string][]byte{ExtendedKeyPolicy: []byte(bindTestPolicy)}) + nsPath := GetNamespacePath(renameTestBucket, "ns") + fs.BeforeUpdate = func() { plantCreateMark(fs, nsPath, "t") } + + status, resp := runHandler(t, m, fs.Client, "DeleteTablePolicy", &DeleteTablePolicyRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + }) + assert.Equal(t, http.StatusNotFound, status, "%+v", resp) + assert.Equal(t, ErrCodeNoSuchTable, resp.Type) + extended := fs.Get(nsPath, "t").Extended + assert.Equal(t, []byte(bindTestPolicy), extended[ExtendedKeyPolicy], "the refused delete keeps the policy") + assert.Equal(t, []byte(createClaim), extended[ExtendedKeyDeleting], "and leaves the create's mark for its back-out") +} + // A rename into a namespace a delete claimed is refused before it touches the // source: the destination cannot be confirmed, so the name must stay where it // was. diff --git a/weed/s3api/s3tables/handler_maintenance.go b/weed/s3api/s3tables/handler_maintenance.go index 02a2764a25..f359f0a38b 100644 --- a/weed/s3api/s3tables/handler_maintenance.go +++ b/weed/s3api/s3tables/handler_maintenance.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net/http" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) @@ -50,7 +51,7 @@ func (h *S3TablesHandler) handlePutTableBucketMaintenanceConfiguration(w http.Re return err } - if err := h.putMaintenanceConfiguration(r, filerClient, GetTableBucketPath(bucketName), req.Type, req.Value); err != nil { + if err := h.putMaintenanceConfiguration(r, filerClient, GetTableBucketPath(bucketName), req.Type, req.Value, nil); err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) return err } @@ -116,13 +117,17 @@ func (h *S3TablesHandler) handlePutTableMaintenanceConfiguration(w http.Response return err } - if _, err := h.authorizeMaintenanceTable(r, filerClient, "PutTableMaintenanceConfiguration", bucketName, namespaceName, tableName); err != nil { + _, createdAt, err := h.authorizeMaintenanceTable(r, filerClient, "PutTableMaintenanceConfiguration", bucketName, namespaceName, tableName) + if err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) return err } tablePath := GetTablePath(bucketName, namespaceName, tableName) - if err := h.putMaintenanceConfiguration(r, filerClient, tablePath, req.Type, req.Value); err != nil { + sameTable := func(extended map[string][]byte) error { + return requireTableCreatedAt(extended, tablePath, createdAt) + } + if err := h.putMaintenanceConfiguration(r, filerClient, tablePath, req.Type, req.Value, sameTable); err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) return err } @@ -145,7 +150,7 @@ func (h *S3TablesHandler) handleGetTableMaintenanceConfiguration(w http.Response return err } - tableARN, err := h.authorizeMaintenanceTable(r, filerClient, "GetTableMaintenanceConfiguration", bucketName, namespaceName, tableName) + tableARN, _, err := h.authorizeMaintenanceTable(r, filerClient, "GetTableMaintenanceConfiguration", bucketName, namespaceName, tableName) if err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) return err @@ -181,7 +186,7 @@ func (h *S3TablesHandler) handleGetTableMaintenanceJobStatus(w http.ResponseWrit return err } - tableARN, err := h.authorizeMaintenanceTable(r, filerClient, "GetTableMaintenanceJobStatus", bucketName, namespaceName, tableName) + tableARN, _, err := h.authorizeMaintenanceTable(r, filerClient, "GetTableMaintenanceJobStatus", bucketName, namespaceName, tableName) if err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) return err @@ -253,18 +258,25 @@ func buildJobStatusResponse(recorded MaintenanceJobStatus, config MaintenanceCon // putMaintenanceConfiguration merges one maintenance type into the stored // configuration, leaving the other types alone. The write checks the catalog // identity in the same conditional mutation, so it cannot land on a name a -// concurrent rename or delete has already soft-deleted. +// concurrent rename or delete has already soft-deleted. sameRecord, when set, +// refuses a record other than the one the caller was authorized against. func (h *S3TablesHandler) putMaintenanceConfiguration( r *http.Request, filerClient FilerClient, resourcePath, maintenanceType string, value *MaintenanceConfigurationValue, + sameRecord func(extended map[string][]byte) error, ) error { return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { if len(extended[ExtendedKeyMetadata]) == 0 { return fmt.Errorf("%w: %s", filer_pb.ErrNotFound, resourcePath) } + if sameRecord != nil { + if err := sameRecord(extended); err != nil { + return err + } + } config := MaintenanceConfiguration{} if current := extended[ExtendedKeyMaintenance]; len(current) > 0 { @@ -362,8 +374,9 @@ func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClien } // authorizeMaintenanceTable checks the caller may perform operation on the -// table, returning the table ARN. -func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient FilerClient, operation, bucketName, namespaceName, tableName string) (string, error) { +// table, returning the table ARN and the creation time of the table record the +// check read. +func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient FilerClient, operation, bucketName, namespaceName, tableName string) (string, time.Time, error) { tablePath := GetTablePath(bucketName, namespaceName, tableName) bucketPath := GetTableBucketPath(bucketName) @@ -391,7 +404,7 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient return err }) if err != nil { - return "", err + return "", time.Time{}, err } tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName) @@ -409,10 +422,10 @@ func (h *S3TablesHandler) authorizeMaintenanceTable(r *http.Request, filerClient } if deniedByPolicy(operation, principal, tablePolicy, tableARN, ctx) || !CheckPermissionWithContext(operation, principal, metadata.OwnerAccountID, bucketPolicy, tableARN, ctx) { - return "", NewAuthError(operation, principal, "not authorized to "+operation) + return "", time.Time{}, NewAuthError(operation, principal, "not authorized to "+operation) } - return tableARN, nil + return tableARN, metadata.CreatedAt, nil } // readPolicy returns the policy stored on path, or an empty string when the diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index 2683e06fd0..ccd00298ba 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/weed/s3api/s3tables/handler_namespace.go @@ -495,6 +495,11 @@ func (h *S3TablesHandler) handleUpdateNamespace(w http.ResponseWriter, r *http.R err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, namespacePath, func(extended map[string][]byte) error { + // The write carries the authorized record's owner and creation + // time, so it must not land on a namespace created again since. + if !isNamespaceCreatedAt(metadata.CreatedAt)(&filer_pb.Entry{Extended: extended}) { + return fmt.Errorf("%w: %s holds another namespace", ErrConcurrentUpdate, namespacePath) + } extended[ExtendedKeyMetadata] = metadataBytes return nil }) diff --git a/weed/s3api/s3tables/handler_policy.go b/weed/s3api/s3tables/handler_policy.go index f6c89f6061..25cf2fcd12 100644 --- a/weed/s3api/s3tables/handler_policy.go +++ b/weed/s3api/s3tables/handler_policy.go @@ -7,10 +7,21 @@ import ( "fmt" "net/http" "strings" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) +// tableCreatedAtOf returns the creation time a table's metadata records, which +// tells the table apart from one created again under its name. +func tableCreatedAtOf(data []byte) (time.Time, error) { + var meta tableMetadataInternal + if err := json.Unmarshal(data, &meta); err != nil { + return time.Time{}, err + } + return meta.CreatedAt, nil +} + // extractResourceOwnerAndBucket extracts ownership info and bucket name from resource metadata. // This helper consolidates the repeated pattern used in handleTagResource, handleListTagsForResource, // and handleUntagResource. @@ -401,6 +412,9 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, tablePath, func(extended map[string][]byte) error { + if err := requireTableCreatedAt(extended, tablePath, metadata.CreatedAt); err != nil { + return err + } if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(tablePolicy)) { return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) } @@ -617,15 +631,26 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http return NewAuthError("DeleteTablePolicy", principal, "not authorized to delete table policy") } + // The removal is conditional on the table and the policy this request was + // authorized against, as PutTablePolicy's write is. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.removeExtendedAttributesIf(r.Context(), client, tablePath, - map[string][]byte{ExtendedKeyPolicy: []byte(tablePolicy)}, - ExtendedKeyPolicy) + return h.mutateCatalogRecord(r.Context(), client, tablePath, func(extended map[string][]byte) error { + if err := requireTableCreatedAt(extended, tablePath, metadata.CreatedAt); err != nil { + return err + } + if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(tablePolicy)) { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) + } + delete(extended, ExtendedKeyPolicy) + return nil + }) }) if err != nil && !errors.Is(err, ErrAttributeNotFound) { if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "table policy changed concurrently, retry the request") + } else if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) } else { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table policy") } @@ -666,6 +691,7 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque var bucketPolicy, tablePolicy string var bucketTags map[string]string var tableNamespace, tableName string + var tableCreatedAt time.Time requestTagKeys := mapKeys(req.Tags) err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // Read metadata for ownership check @@ -695,6 +721,9 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque } if rType == ResourceTypeTable { + if tableCreatedAt, err = tableCreatedAtOf(data); err != nil { + return err + } if _, tableNamespace, tableName, err = parseTableFromARN(req.ResourceARN); err != nil { return err } @@ -769,6 +798,11 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { + if rType == ResourceTypeTable { + if err := requireTableCreatedAt(extended, resourcePath, tableCreatedAt); err != nil { + return err + } + } extended[extendedKey] = tagsBytes return nil }) @@ -942,6 +976,7 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req var bucketPolicy, tablePolicy string var bucketTags map[string]string var tableNamespace, tableName string + var tableCreatedAt time.Time err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // Read metadata for ownership check data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, ExtendedKeyMetadata) @@ -970,6 +1005,9 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req } if rType == ResourceTypeTable { + if tableCreatedAt, err = tableCreatedAtOf(data); err != nil { + return err + } if _, tableNamespace, tableName, err = parseTableFromARN(req.ResourceARN); err != nil { return err } @@ -1042,6 +1080,11 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { + if rType == ResourceTypeTable { + if err := requireTableCreatedAt(extended, resourcePath, tableCreatedAt); err != nil { + return err + } + } extended[extendedKey] = tagsBytes return nil }) diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 10edb2e94d..afa5a2ddc9 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -166,7 +166,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque if err := h.claimCatalogEntry(r.Context(), client, tablePath, attrs); err != nil { return err } - return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceEntry, tablePath, attrs, dataPath, tablePath) + return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, dataPath, tablePath) }) if err != nil { @@ -313,7 +313,7 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req if err := h.claimCatalogEntry(r.Context(), client, tablePath, attrs); err != nil { return err } - return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceEntry, tablePath, attrs, tablePath) + return h.confirmParent(r.Context(), client, GetNamespacePath(bucketName, namespaceName), isNamespaceCreatedAt(target.namespaceCreatedAt), tablePath, attrs, tablePath) }) if err != nil { if errors.Is(err, ErrNameTaken) { @@ -1636,7 +1636,10 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ h.writeError(w, http.StatusConflict, kind.existsCode, fmt.Sprintf("%s %s already exists", kind.noun, destName)) return ErrTableAlreadyExists } - if err := h.refuseClaimedNamespace(w, r, filerClient, destNamespacePath, destNamespace); err != nil { + // The destination is bound to the namespace the create check ran against: + // one deleted and created again under its name reads as gone. + isDestNamespace := isNamespaceCreatedAt(destNamespaceMetadata.CreatedAt) + if err := h.refuseClaimedNamespace(w, r, filerClient, destNamespacePath, destNamespace, isDestNamespace); err != nil { return err } @@ -1718,7 +1721,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ h.removeLeftoverDirectory(ctx, client, destPath) return releaseClaim(err) } - destParent, err := h.openParent(r.Context(), client, destNamespacePath, isNamespaceEntry) + destParent, err := h.openParent(r.Context(), client, destNamespacePath, isDestNamespace) if err != nil { return backOutDest(err) } @@ -1744,7 +1747,7 @@ func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Requ destConfirmed := false if err = h.removeExtendedAttributesIf(commitCtx, client, destPath, written, ExtendedKeyDeleting); err == nil { destConfirmed = true - err = h.parentStillOpen(commitCtx, client, destNamespacePath, isNamespaceEntry, destParent) + err = h.parentStillOpen(commitCtx, client, destNamespacePath, isDestNamespace, destParent) } if err != nil { if destConfirmed { @@ -2116,13 +2119,15 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque } // createTableTarget is the namespace a create or register resolved to, once the -// caller has been authorized to put a table there. +// caller has been authorized to put a table there. namespaceCreatedAt names the +// namespace record the authorization read, so the write lands in no other. type createTableTarget struct { - bucketName string - namespaceName string - tableName string - bucketFormat string - ownerAccountID string + bucketName string + namespaceName string + tableName string + bucketFormat string + ownerAccountID string + namespaceCreatedAt time.Time } // authorizeCreateTable validates the names a create names and checks the caller @@ -2256,17 +2261,18 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re } if namespaceExists { - if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName); err != nil { + if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName, isNamespaceCreatedAt(nsMetadata.CreatedAt)); err != nil { return nil, err } } return &createTableTarget{ - bucketName: bucketName, - namespaceName: namespaceName, - tableName: tableName, - bucketFormat: bucketMetadata.Format, - ownerAccountID: nsMetadata.OwnerAccountID, + bucketName: bucketName, + namespaceName: namespaceName, + tableName: tableName, + bucketFormat: bucketMetadata.Format, + ownerAccountID: nsMetadata.OwnerAccountID, + namespaceCreatedAt: nsMetadata.CreatedAt, }, nil } @@ -2289,14 +2295,16 @@ func (h *S3TablesHandler) writeNamespaceClosedError(w http.ResponseWriter, err e return true } -// refuseClaimedNamespace answers 409 when a delete has claimed the namespace or -// its create is not confirmed yet. Call it after the authorization check, so -// denied callers never see the delete. -func (h *S3TablesHandler) refuseClaimedNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient, namespacePath, namespaceName string) error { +// refuseClaimedNamespace refuses a request before it writes anything when the +// namespace is not the open record the caller was authorized against: 409 when a +// delete has claimed it, 404 when it is gone, was replaced by another namespace, +// or its create is not confirmed yet. isNamespace accepts that record. Call it +// after the authorization check, so denied callers never see the delete. +func (h *S3TablesHandler) refuseClaimedNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient, namespacePath, namespaceName string, isNamespace func(*filer_pb.Entry) bool) error { err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.checkParentOpen(r.Context(), client, namespacePath, isNamespaceEntry) + return h.checkParentOpen(r.Context(), client, namespacePath, isNamespace) }) - if errors.Is(err, errParentDeleting) || errors.Is(err, errParentCreating) { + if errors.Is(err, errParentDeleting) || errors.Is(err, errParentCreating) || errors.Is(err, errParentGone) { h.writeNamespaceClosedError(w, err, namespaceName) return err } diff --git a/weed/s3api/s3tables/handler_update_conflict_test.go b/weed/s3api/s3tables/handler_update_conflict_test.go index 0117f736dc..5b4066b772 100644 --- a/weed/s3api/s3tables/handler_update_conflict_test.go +++ b/weed/s3api/s3tables/handler_update_conflict_test.go @@ -67,3 +67,23 @@ func TestUntagResourceClaimedRecordConflicts(t *testing.T) { assert.Equal(t, http.StatusConflict, status, "%+v", resp) assert.Equal(t, ErrCodeConflict, resp.Type) } + +// DeleteTablePolicy writes like every other mutation of a table record, so a +// record another operation claimed refuses it with 409 instead of losing the +// policy under the claim's owner. +func TestDeleteTablePolicyClaimedRecordConflicts(t *testing.T) { + fs, m := startTableT(t, map[string][]byte{ExtendedKeyPolicy: []byte(bindTestPolicy)}) + nsPath := GetNamespacePath(renameTestBucket, "ns") + claimRecord(t, fs, nsPath, "t") + + status, resp := runHandler(t, m, fs.Client, "DeleteTablePolicy", &DeleteTablePolicyRequest{ + TableBucketARN: mustBucketARN(t), + Namespace: []string{"ns"}, + Name: "t", + }) + assert.Equal(t, http.StatusConflict, status, "%+v", resp) + assert.Equal(t, ErrCodeConflict, resp.Type, "%+v", resp) + extended := fs.Get(nsPath, "t").Extended + assert.Equal(t, []byte(bindTestPolicy), extended[ExtendedKeyPolicy], "the refused delete keeps the policy") + assert.Equal(t, []byte("delete"), extended[ExtendedKeyDeleting], "and leaves the claim to its owner") +} diff --git a/weed/s3api/s3tables/handler_view.go b/weed/s3api/s3tables/handler_view.go index cb63a3b97c..e8f6dadb61 100644 --- a/weed/s3api/s3tables/handler_view.go +++ b/weed/s3api/s3tables/handler_view.go @@ -1,6 +1,7 @@ package s3tables import ( + "bytes" "context" "encoding/json" "errors" @@ -85,7 +86,9 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create view in this namespace") return ErrAccessDenied } - if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName); err != nil { + // The view is bound to the namespace record it was authorized against. + isNamespace := isNamespaceCreatedAt(namespaceMetadata.CreatedAt) + if err := h.refuseClaimedNamespace(w, r, filerClient, namespacePath, namespaceName, isNamespace); err != nil { return err } @@ -169,7 +172,7 @@ func (h *S3TablesHandler) handleCreateView(w http.ResponseWriter, r *http.Reques if err := h.claimCatalogEntry(r.Context(), client, viewPath, attrs); err != nil { return err } - return h.confirmParent(r.Context(), client, namespacePath, isNamespaceEntry, viewPath, attrs, viewPath) + return h.confirmParent(r.Context(), client, namespacePath, isNamespace, viewPath, attrs, viewPath) }) if err != nil { if errors.Is(err, ErrNameTaken) { @@ -432,14 +435,17 @@ func (h *S3TablesHandler) listViewsInNamespace(r *http.Request, client filer_pb. } // updateViewTarget is the view state an update was authorized against: the -// metadata the version-token check reads and the fields the response echoes. +// metadata the version-token check reads, the stored values the +// optimistic-concurrency check compares and the fields the response echoes. type updateViewTarget struct { - bucketName string - namespaceName string - viewName string - viewPath string - viewARN string - metadata tableMetadataInternal + bucketName string + namespaceName string + viewName string + viewPath string + viewARN string + metadata tableMetadataInternal + storedMetadata []byte + storedPolicy []byte } // authorizeUpdateView validates the names an update names, loads the view and @@ -460,6 +466,7 @@ func (h *S3TablesHandler) authorizeUpdateView(w http.ResponseWriter, r *http.Req viewPath := GetTablePath(bucketName, namespaceName, viewName) var metadata tableMetadataInternal + var storedMetadata []byte err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { entry, err := h.lookupEntry(r.Context(), client, viewPath) if err != nil { @@ -472,6 +479,7 @@ func (h *S3TablesHandler) authorizeUpdateView(w http.ResponseWriter, r *http.Req if !ok { return fmt.Errorf("%w: %w: %s", filer_pb.ErrNotFound, ErrAttributeNotFound, ExtendedKeyMetadata) } + storedMetadata = data return json.Unmarshal(data, &metadata) }) if err != nil { @@ -496,12 +504,14 @@ func (h *S3TablesHandler) authorizeUpdateView(w http.ResponseWriter, r *http.Req } return &updateViewTarget{ - bucketName: bucketName, - namespaceName: namespaceName, - viewName: viewName, - viewPath: viewPath, - viewARN: viewARN, - metadata: metadata, + bucketName: bucketName, + namespaceName: namespaceName, + viewName: viewName, + viewPath: viewPath, + viewARN: viewARN, + metadata: metadata, + storedMetadata: storedMetadata, + storedPolicy: []byte(viewPolicy), }, nil } @@ -558,8 +568,18 @@ func (h *S3TablesHandler) handleUpdateView(w http.ResponseWriter, r *http.Reques return err } + // Conditional on the record this request read and authorized against, as + // UpdateTable's write is: a concurrent update would otherwise be dropped, + // and a view deleted and created again under the name would take a write + // nobody authorized for it. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, viewPath, func(extended map[string][]byte) error { + if !bytes.Equal(extended[ExtendedKeyMetadata], target.storedMetadata) { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyMetadata) + } + if !bytes.Equal(extended[ExtendedKeyPolicy], target.storedPolicy) { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) + } extended[ExtendedKeyMetadata] = metadataBytes return nil })