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 })