diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go index 3775621a6a..80ad13ef25 100644 --- a/weed/s3api/s3tables/filer_ops.go +++ b/weed/s3api/s3tables/filer_ops.go @@ -483,14 +483,6 @@ func (h *S3TablesHandler) lookupEntry(ctx context.Context, client filer_pb.Seawe return resp.Entry, nil } -// deleteExtendedAttribute deletes an extended attribute from an entry -func (h *S3TablesHandler) deleteExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string) error { - return h.mutateEntryExtended(ctx, client, path, func(extended map[string][]byte) error { - delete(extended, key) - return nil - }) -} - // deleteDirectory deletes a directory and all its contents func (h *S3TablesHandler) deleteDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error { dir, name := splitPath(path) @@ -772,6 +764,14 @@ func requireTableCreatedAt(extended map[string][]byte, path string, createdAt ti return nil } +// requireTableBucketCreatedAt is requireTableCreatedAt for a table bucket. +func requireTableBucketCreatedAt(extended map[string][]byte, path string, createdAt time.Time) error { + if !isTableBucketCreatedAt(createdAt)(&filer_pb.Entry{Extended: extended}) { + return fmt.Errorf("%w: %s holds another table bucket", 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_bucket_test.go b/weed/s3api/s3tables/handler_bind_bucket_test.go new file mode 100644 index 0000000000..e36229a897 --- /dev/null +++ b/weed/s3api/s3tables/handler_bind_bucket_test.go @@ -0,0 +1,104 @@ +package s3tables + +import ( + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// recreatedBucket is the record of the test bucket deleted and created again: +// another owner, and a creation time of its own. +func recreatedBucket(t *testing.T) map[string][]byte { + t.Helper() + meta, err := json.Marshal(tableBucketMetadata{ + Name: renameTestBucket, + OwnerAccountID: "111122223333", + CreatedAt: time.Now().Add(time.Hour), + }) + require.NoError(t, err) + return map[string][]byte{ + ExtendedKeyTableBucket: []byte("{}"), + ExtendedKeyMetadata: meta, + } +} + +// bucketWriteOp is a write to the test bucket, with the attributes the bucket +// carries for it besides its record. +type bucketWriteOp struct { + operation string + input func(t *testing.T) interface{} + extra map[string][]byte +} + +func bucketWriteOps() []bucketWriteOp { + return []bucketWriteOp{ + {operation: "PutTableBucketPolicy", input: func(t *testing.T) interface{} { + return &PutTableBucketPolicyRequest{TableBucketARN: mustBucketARN(t), ResourcePolicy: bindTestPolicy} + }}, + {operation: "DeleteTableBucketPolicy", input: func(t *testing.T) interface{} { + return &DeleteTableBucketPolicyRequest{TableBucketARN: mustBucketARN(t)} + }, extra: map[string][]byte{ExtendedKeyPolicy: []byte(bindTestPolicy)}}, + {operation: "TagResource", input: func(t *testing.T) interface{} { + return &TagResourceRequest{ResourceARN: mustBucketARN(t), Tags: map[string]string{"k": "v"}} + }}, + {operation: "UntagResource", input: func(t *testing.T) interface{} { + return &UntagResourceRequest{ResourceARN: mustBucketARN(t), TagKeys: []string{"k"}} + }, extra: map[string][]byte{ExtendedKeyTags: []byte(`{"k":"v","keep":"1"}`)}}, + {operation: "PutTableBucketMaintenanceConfiguration", input: func(t *testing.T) interface{} { + return &PutTableBucketMaintenanceConfigurationRequest{ + TableBucketARN: mustBucketARN(t), + Type: MaintenanceTypeIcebergUnreferencedFileRemoval, + Value: &MaintenanceConfigurationValue{Status: MaintenanceStatusEnabled}, + } + }}, + } +} + +// A write to a bucket's policy, tags or maintenance settings is authorized +// against the bucket it read. If that bucket is deleted and another is created under the same name +// before the write, the values it compares can match the new bucket's just as +// well. The write checks the bucket by its creation time and is refused, and +// the new bucket is kept as it was created. +func TestBucketWritesSpareARecreatedBucket(t *testing.T) { + for _, op := range bucketWriteOps() { + t.Run(op.operation, func(t *testing.T) { + fs, m := startEmptyBucket(t) + extended := extendedOf(t, fs, TablesPath, renameTestBucket) + recreated := recreatedBucket(t) + for key, value := range op.extra { + extended[key] = value + recreated[key] = value + } + fs.Put(TablesPath, renameTestBucket, extended) + fs.BeforeUpdate = func() { fs.Put(TablesPath, renameTestBucket, 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) + assert.Equal(t, recreated, fs.Get(TablesPath, renameTestBucket).Extended, "the new bucket is left as it was created") + }) + } +} + +// The same writes still land on the bucket they were authorized against. +func TestBucketWritesLandOnTheAuthorizedBucket(t *testing.T) { + for _, op := range bucketWriteOps() { + t.Run(op.operation, func(t *testing.T) { + fs, m := startEmptyBucket(t) + extended := extendedOf(t, fs, TablesPath, renameTestBucket) + for key, value := range op.extra { + extended[key] = value + } + fs.Put(TablesPath, renameTestBucket, extended) + + code, resp := runHandler(t, m, fs.Client, op.operation, op.input(t)) + + assert.Equal(t, http.StatusOK, code, "%+v", resp) + }) + } +} diff --git a/weed/s3api/s3tables/handler_bucket_get_list_delete.go b/weed/s3api/s3tables/handler_bucket_get_list_delete.go index 2e7c604988..34b8b8ab04 100644 --- a/weed/s3api/s3tables/handler_bucket_get_list_delete.go +++ b/weed/s3api/s3tables/handler_bucket_get_list_delete.go @@ -311,6 +311,9 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http if err := json.Unmarshal(data, &metadata); err != nil { return fmt.Errorf("failed to unmarshal metadata: %w", err) } + if otherTableBucket(r.Context(), metadata.CreatedAt) { + return filer_pb.ErrNotFound + } // Fetch bucket policy if it exists policyData, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy) diff --git a/weed/s3api/s3tables/handler_maintenance.go b/weed/s3api/s3tables/handler_maintenance.go index f359f0a38b..bd971e64cf 100644 --- a/weed/s3api/s3tables/handler_maintenance.go +++ b/weed/s3api/s3tables/handler_maintenance.go @@ -46,12 +46,13 @@ func (h *S3TablesHandler) handlePutTableBucketMaintenanceConfiguration(w http.Re return err } - if _, err := h.authorizeMaintenanceBucket(r, filerClient, "PutTableBucketMaintenanceConfiguration", bucketName); err != nil { + _, createdAt, err := h.authorizeMaintenanceBucket(r, filerClient, "PutTableBucketMaintenanceConfiguration", bucketName) + if err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) return err } - if err := h.putMaintenanceConfiguration(r, filerClient, GetTableBucketPath(bucketName), req.Type, req.Value, nil); err != nil { + if err := h.putMaintenanceConfiguration(r, filerClient, GetTableBucketPath(bucketName), req.Type, req.Value, requireTableBucketCreatedAt, createdAt); err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) return err } @@ -79,7 +80,7 @@ func (h *S3TablesHandler) handleGetTableBucketMaintenanceConfiguration(w http.Re return err } - bucketARN, err := h.authorizeMaintenanceBucket(r, filerClient, "GetTableBucketMaintenanceConfiguration", bucketName) + bucketARN, _, err := h.authorizeMaintenanceBucket(r, filerClient, "GetTableBucketMaintenanceConfiguration", bucketName) if err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) return err @@ -124,10 +125,7 @@ func (h *S3TablesHandler) handlePutTableMaintenanceConfiguration(w http.Response } tablePath := GetTablePath(bucketName, namespaceName, tableName) - 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 { + if err := h.putMaintenanceConfiguration(r, filerClient, tablePath, req.Type, req.Value, requireTableCreatedAt, createdAt); err != nil { h.writeMaintenanceError(w, err, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", tableName)) return err } @@ -258,24 +256,24 @@ 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. sameRecord, when set, -// refuses a record other than the one the caller was authorized against. +// concurrent rename or delete has already soft-deleted. sameRecord refuses a +// record other than the one created at createdAt, which 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, + sameRecord func(extended map[string][]byte, path string, createdAt time.Time) error, + createdAt time.Time, ) 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 - } + if err := sameRecord(extended, resourcePath, createdAt); err != nil { + return err } config := MaintenanceConfiguration{} @@ -337,8 +335,9 @@ func (h *S3TablesHandler) parseTableTarget(tableBucketARN string, namespace []st } // authorizeMaintenanceBucket checks the caller may perform operation on the -// table bucket, returning the bucket ARN. -func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClient FilerClient, operation, bucketName string) (string, error) { +// table bucket, returning the bucket ARN and the creation time of the bucket +// record the check read. +func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClient FilerClient, operation, bucketName string) (string, time.Time, error) { bucketPath := GetTableBucketPath(bucketName) var bucketMetadata tableBucketMetadata @@ -355,7 +354,7 @@ func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClien return err }) if err != nil { - return "", err + return "", time.Time{}, err } bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName) @@ -367,10 +366,10 @@ func (h *S3TablesHandler) authorizeMaintenanceBucket(r *http.Request, filerClien IdentityActions: getIdentityActions(r), DefaultAllow: h.defaultAllowFor(r), }) { - return "", NewAuthError(operation, principal, "not authorized to "+operation) + return "", time.Time{}, NewAuthError(operation, principal, "not authorized to "+operation) } - return bucketARN, nil + return bucketARN, bucketMetadata.CreatedAt, nil } // authorizeMaintenanceTable checks the caller may perform operation on the diff --git a/weed/s3api/s3tables/handler_namespace.go b/weed/s3api/s3tables/handler_namespace.go index ccd00298ba..4327fc2f56 100644 --- a/weed/s3api/s3tables/handler_namespace.go +++ b/weed/s3api/s3tables/handler_namespace.go @@ -151,6 +151,11 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R return err } + if otherTableBucket(r.Context(), bucketMetadata.CreatedAt) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) + return filer_pb.ErrNotFound + } + bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName) principal := h.getAccountID(r) identityActions := getIdentityActions(r) diff --git a/weed/s3api/s3tables/handler_policy.go b/weed/s3api/s3tables/handler_policy.go index 25cf2fcd12..8483900637 100644 --- a/weed/s3api/s3tables/handler_policy.go +++ b/weed/s3api/s3tables/handler_policy.go @@ -12,24 +12,15 @@ import ( "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. +// extractResourceOwnerAndBucket extracts ownership info, bucket name and the +// creation time that tells the resource apart from one created again under its +// name. It consolidates the repeated pattern used in handleTagResource, +// handleListTagsForResource, and handleUntagResource. func (h *S3TablesHandler) extractResourceOwnerAndBucket( data []byte, resourcePath string, rType ResourceType, -) (ownerAccountID, bucketName string, err error) { +) (ownerAccountID, bucketName string, createdAt time.Time, err error) { // Extract bucket name from resource path (format: /buckets/{bucket}/... for both tables and buckets) parts := strings.Split(strings.Trim(resourcePath, "/"), "/") if len(parts) >= 2 { @@ -39,17 +30,17 @@ func (h *S3TablesHandler) extractResourceOwnerAndBucket( if rType == ResourceTypeTable { var meta tableMetadataInternal if err := json.Unmarshal(data, &meta); err != nil { - return "", "", err + return "", "", time.Time{}, err } - ownerAccountID = meta.OwnerAccountID + ownerAccountID, createdAt = meta.OwnerAccountID, meta.CreatedAt } else { var meta tableBucketMetadata if err := json.Unmarshal(data, &meta); err != nil { - return "", "", err + return "", "", time.Time{}, err } - ownerAccountID = meta.OwnerAccountID + ownerAccountID, createdAt = meta.OwnerAccountID, meta.CreatedAt } - return ownerAccountID, bucketName, nil + return ownerAccountID, bucketName, createdAt, nil } // handlePutTableBucketPolicy puts a policy on a table bucket @@ -133,6 +124,9 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h // landing mid-request sends the caller back through authorization. err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateEntryExtended(r.Context(), client, bucketPath, func(extended map[string][]byte) error { + if err := requireTableBucketCreatedAt(extended, bucketPath, bucketMetadata.CreatedAt); err != nil { + return err + } if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(bucketPolicy)) { return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) } @@ -144,6 +138,8 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h if err != nil { if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "table bucket policy changed concurrently, retry the request") + } else if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) } else { h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to put table bucket policy") } @@ -299,11 +295,26 @@ func (h *S3TablesHandler) handleDeleteTableBucketPolicy(w http.ResponseWriter, r } err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return h.deleteExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy) + return h.mutateEntryExtended(r.Context(), client, bucketPath, func(extended map[string][]byte) error { + if err := requireTableBucketCreatedAt(extended, bucketPath, bucketMetadata.CreatedAt); err != nil { + return err + } + if !bytes.Equal(extended[ExtendedKeyPolicy], []byte(bucketPolicy)) { + return fmt.Errorf("%w: %s", ErrConcurrentUpdate, ExtendedKeyPolicy) + } + delete(extended, ExtendedKeyPolicy) + return nil + }) }) - if err != nil && !errors.Is(err, ErrAttributeNotFound) { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table bucket policy") + if err != nil { + if errors.Is(err, ErrConcurrentUpdate) { + h.writeError(w, http.StatusConflict, ErrCodeConflict, "table bucket policy changed concurrently, retry the request") + } else if errors.Is(err, filer_pb.ErrNotFound) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) + } else { + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table bucket policy") + } return err } @@ -691,7 +702,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 + var createdAt time.Time requestTagKeys := mapKeys(req.Tags) err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { // Read metadata for ownership check @@ -703,10 +714,11 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque return err } - ownerAccountID, bucketName, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType) + ownerAccountID, bucketName, resourceCreatedAt, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType) if err != nil { return err } + createdAt = resourceCreatedAt // Fetch bucket policy if we have a bucket name if bucketName != "" { @@ -721,9 +733,6 @@ 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 } @@ -796,12 +805,14 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal tags") return fmt.Errorf("failed to marshal tags: %w", err) } + sameResource := requireTableCreatedAt + if rType != ResourceTypeTable { + sameResource = requireTableBucketCreatedAt + } 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 - } + if err := sameResource(extended, resourcePath, createdAt); err != nil { + return err } extended[extendedKey] = tagsBytes return nil @@ -863,7 +874,7 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht return err } - ownerAccountID, bucketName, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType) + ownerAccountID, bucketName, _, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType) if err != nil { return err } @@ -976,7 +987,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 + var createdAt 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) @@ -987,10 +998,11 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req return err } - ownerAccountID, bucketName, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType) + ownerAccountID, bucketName, resourceCreatedAt, err := h.extractResourceOwnerAndBucket(data, resourcePath, rType) if err != nil { return err } + createdAt = resourceCreatedAt // Fetch bucket policy if we have a bucket name if bucketName != "" { @@ -1005,9 +1017,6 @@ 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 } @@ -1078,12 +1087,14 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal tags") return fmt.Errorf("failed to marshal tags: %w", err) } + sameResource := requireTableCreatedAt + if rType != ResourceTypeTable { + sameResource = requireTableBucketCreatedAt + } 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 - } + if err := sameResource(extended, resourcePath, createdAt); 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 afa5a2ddc9..6db8fab282 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -1009,6 +1009,10 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque } return err } + if otherTableBucket(r.Context(), bucketMetadata.CreatedAt) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) + return filer_pb.ErrNotFound + } tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, namespaceName+"/"+tableName) bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName) @@ -2233,6 +2237,10 @@ func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Re h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to fetch policies: %v", err)) return nil, err } + if otherTableBucket(r.Context(), bucketMetadata.CreatedAt) { + h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName)) + return nil, filer_pb.ErrNotFound + } if !namespaceExists { // The implicit namespace has no record yet: authorize it as the diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index bffe3c9c5f..0334f80fec 100644 --- a/weed/s3api/s3tables/manager.go +++ b/weed/s3api/s3tables/manager.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "net/http/httptest" + "time" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -76,6 +77,24 @@ func (m *Manager) PolicyContext(r *http.Request, bucket string) *PolicyContext { return ctx } +type tableBucketCreatedAtKey struct{} + +// WithTableBucketCreatedAt binds a DeleteTableBucket, CreateNamespace, +// CreateTable or DeleteTable run through Execute to the bucket created at +// createdAt, so a +// caller that authorized the bucket itself never writes into one created +// again under its name. In-process only: no wire request can carry it. +func WithTableBucketCreatedAt(ctx context.Context, createdAt time.Time) context.Context { + return context.WithValue(ctx, tableBucketCreatedAtKey{}, createdAt) +} + +// otherTableBucket reports whether ctx binds the request to a bucket other +// than the one created at createdAt. +func otherTableBucket(ctx context.Context, createdAt time.Time) bool { + want, ok := ctx.Value(tableBucketCreatedAtKey{}).(time.Time) + return ok && !createdAt.Equal(want) +} + // Execute runs an S3 Tables operation and decodes the response into resp (if provided). func (m *Manager) Execute(ctx context.Context, filerClient FilerClient, operation string, req interface{}, resp interface{}, identity string) error { body, err := json.Marshal(req) diff --git a/weed/s3api/s3vectors/bind_bucket_test.go b/weed/s3api/s3vectors/bind_bucket_test.go new file mode 100644 index 0000000000..3a73326874 --- /dev/null +++ b/weed/s3api/s3vectors/bind_bucket_test.go @@ -0,0 +1,209 @@ +package s3vectors + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +// A vector bucket policy write authorizes against the bucket it read: when the +// bucket is deleted and created again by another account before the write's +// compare-and-swap reads the entry, the write must not land on the new bucket. +func TestVectorBucketPolicyWritesKeepToAuthorizedBucket(t *testing.T) { + const policy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectorBucket","Resource":"*"}]}` + cases := []struct { + name string + call func(s *Server) *httptest.ResponseRecorder + }{ + {"put", func(s *Server) *httptest.ResponseRecorder { + return postJSONAs(t, "", s.handlePutVectorBucketPolicy, &PutVectorBucketPolicyRequest{VectorBucketName: "vec-bucket", Policy: policy}) + }}, + {"delete", func(s *Server) *httptest.ResponseRecorder { + return postJSONAs(t, "", s.handleDeleteVectorBucketPolicy, &DeleteVectorBucketPolicyRequest{VectorBucketName: "vec-bucket"}) + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + fs := s3tablestest.Start(t) + s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0)) + if w := postJSONAs(t, "", s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("create bucket: %d %s", w.Code, w.Body.String()) + } + if w := postJSONAs(t, "", s.handlePutVectorBucketPolicy, &PutVectorBucketPolicyRequest{VectorBucketName: "vec-bucket", Policy: policy}); w.Code != http.StatusOK { + t.Fatalf("seed policy: %d %s", w.Code, w.Body.String()) + } + + dir, name := splitEntryPath(s3tables.GetTableBucketPath("vec-bucket")) + lookups := 0 + fs.OnLookup = func(d, n string) { + if d == dir && n == name { + lookups++ + } + } + if w := tc.call(s); w.Code != http.StatusOK { + t.Fatalf("baseline %s: %d %s", tc.name, w.Code, w.Body.String()) + } + fs.OnLookup = nil + if w := postJSONAs(t, "", s.handlePutVectorBucketPolicy, &PutVectorBucketPolicyRequest{VectorBucketName: "vec-bucket", Policy: policy}); w.Code != http.StatusOK { + t.Fatalf("reseed policy: %d %s", w.Code, w.Body.String()) + } + // The last bucket lookup is the compare-and-swap read; replace the + // bucket with another account's right before it. + last, seen := lookups, 0 + fs.OnLookup = func(d, n string) { + if d != dir || n != name { + return + } + if seen++; seen != last { + return + } + entry := fs.Get(dir, name) + var meta map[string]interface{} + if err := json.Unmarshal(entry.Extended[s3tables.ExtendedKeyMetadata], &meta); err != nil { + t.Errorf("decode bucket metadata: %v", err) + return + } + meta["ownerAccountId"] = "bob" + meta["createdAt"] = time.Now().Add(time.Hour).UTC() + data, _ := json.Marshal(meta) + ext := make(map[string][]byte, len(entry.Extended)) + for k, v := range entry.Extended { + ext[k] = v + } + ext[s3tables.ExtendedKeyMetadata] = data + ext[vectorPolicyKey] = []byte("bob-policy") + fs.Put(dir, name, ext) + } + w := tc.call(s) + fs.OnLookup = nil + if w.Code != http.StatusConflict { + t.Fatalf("%s over a replaced bucket = %d %s, want 409", tc.name, w.Code, w.Body.String()) + } + if got := string(fs.Get(dir, name).Extended[vectorPolicyKey]); got != "bob-policy" { + t.Fatalf("%s rewrote the replacing bucket's policy: %q", tc.name, got) + } + }) + } +} + +// replaceBucket swaps the bucket's record for another account's bucket of the +// same name, keeping its contents. +func replaceBucket(t *testing.T, fs *s3tablestest.MemFiler, bucket string) { + dir, name := splitEntryPath(s3tables.GetTableBucketPath(bucket)) + entry := fs.Get(dir, name) + var meta map[string]interface{} + if err := json.Unmarshal(entry.Extended[s3tables.ExtendedKeyMetadata], &meta); err != nil { + t.Errorf("decode bucket metadata: %v", err) + return + } + meta["ownerAccountId"] = "bob" + meta["createdAt"] = time.Now().Add(time.Hour).UTC() + data, _ := json.Marshal(meta) + ext := make(map[string][]byte, len(entry.Extended)) + for k, v := range entry.Extended { + ext[k] = v + } + ext[s3tables.ExtendedKeyMetadata] = data + fs.Put(dir, name, ext) +} + +// replaceBucketAfterAuthorization replaces bucket right after the lookups +// authorizeVector makes, counted on a probe bucket of the same shape. +func replaceBucketAfterAuthorization(t *testing.T, fs *s3tablestest.MemFiler, s *Server, bucket string) { + if w := postJSONAs(t, "", s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-probe"}); w.Code != http.StatusOK { + t.Fatalf("create probe bucket: %d %s", w.Code, w.Body.String()) + } + probeDir, probeName := splitEntryPath(s3tables.GetTableBucketPath("vec-probe")) + authLookups := 0 + fs.OnLookup = func(d, n string) { + if d == probeDir && n == probeName { + authLookups++ + } + } + probe := httptest.NewRequest(http.MethodPost, "/", nil) + probe = probe.WithContext(contextWithIdentity(probe.Context(), "")) + if _, err := s.authorizeVector(probe.Context(), probe, "s3vectors:DeleteVectorBucket", "vec-probe", ""); err != nil { + t.Fatalf("authorize probe: %v", err) + } + + dir, name := splitEntryPath(s3tables.GetTableBucketPath(bucket)) + seen := 0 + fs.OnLookup = func(d, n string) { + if d == dir && n == name { + if seen++; seen == authLookups+1 { + replaceBucket(t, fs, bucket) + } + } + } +} + +// DeleteVectorBucket deletes only the bucket it authorized: one deleted and +// created again by another account in between survives. +func TestDeleteVectorBucketKeepsReplacingBucket(t *testing.T) { + fs := s3tablestest.Start(t) + s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0)) + if w := postJSONAs(t, "", s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("create bucket: %d %s", w.Code, w.Body.String()) + } + replaceBucketAfterAuthorization(t, fs, s, "vec-bucket") + w := postJSONAs(t, "", s.handleDeleteVectorBucket, &DeleteVectorBucketRequest{VectorBucketName: "vec-bucket"}) + fs.OnLookup = nil + if w.Code != http.StatusNotFound { + t.Fatalf("delete over a replaced bucket = %d %s, want 404", w.Code, w.Body.String()) + } + dir, name := splitEntryPath(s3tables.GetTableBucketPath("vec-bucket")) + if fs.Get(dir, name) == nil { + t.Fatal("DeleteVectorBucket removed the bucket that replaced the authorized one") + } +} + +// CreateIndex and DeleteIndex write only into the bucket they authorized. +func TestIndexWritesKeepToAuthorizedBucket(t *testing.T) { + newServer := func(t *testing.T) (*s3tablestest.MemFiler, *Server) { + fs := s3tablestest.Start(t) + s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0)) + if w := postJSONAs(t, "", s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("create bucket: %d %s", w.Code, w.Body.String()) + } + return fs, s + } + createIndex := func(s *Server) *httptest.ResponseRecorder { + return postJSONAs(t, "", s.handleCreateIndex, &CreateIndexRequest{ + VectorBucketName: "vec-bucket", IndexName: "idx", DataType: "float32", Dimension: 3, DistanceMetric: "euclidean", + }) + } + indexDir, indexName := splitEntryPath(s3tables.GetTablePath("vec-bucket", indexNamespace, "idx")) + + t.Run("create", func(t *testing.T) { + fs, s := newServer(t) + replaceBucketAfterAuthorization(t, fs, s, "vec-bucket") + w := createIndex(s) + fs.OnLookup = nil + if w.Code != http.StatusNotFound { + t.Fatalf("create over a replaced bucket = %d %s, want 404", w.Code, w.Body.String()) + } + if fs.Get(indexDir, indexName) != nil { + t.Fatal("CreateIndex wrote an index into the bucket that replaced the authorized one") + } + }) + t.Run("delete", func(t *testing.T) { + fs, s := newServer(t) + if w := createIndex(s); w.Code != http.StatusOK { + t.Fatalf("create index: %d %s", w.Code, w.Body.String()) + } + replaceBucketAfterAuthorization(t, fs, s, "vec-bucket") + w := postJSONAs(t, "", s.handleDeleteIndex, &DeleteIndexRequest{VectorBucketName: "vec-bucket", IndexName: "idx"}) + fs.OnLookup = nil + if w.Code != http.StatusNotFound { + t.Fatalf("delete over a replaced bucket = %d %s, want 404", w.Code, w.Body.String()) + } + if fs.Get(indexDir, indexName) == nil { + t.Fatal("DeleteIndex removed an index of the bucket that replaced the authorized one") + } + }) +} diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 7e7bd12e83..af762d7097 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -405,12 +405,16 @@ func (s *Server) handleDeleteVectorBucket(w http.ResponseWriter, r *http.Request writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } - if _, err := s.authorizeVector(r.Context(), r, "s3vectors:DeleteVectorBucket", arns.bucket, ""); err != nil { + record, err := s.authorizeVector(r.Context(), r, "s3vectors:DeleteVectorBucket", arns.bucket, "") + if err != nil { catalogErr(w, err) return } + // Only the bucket authorized above is deleted, not one created again + // under its name since. + ctx := s3tables.WithTableBucketCreatedAt(r.Context(), record.CreatedAt) var resp map[string]interface{} - err = s.internalCatalog(r.Context(), "DeleteTableBucket", + err = s.internalCatalog(ctx, "DeleteTableBucket", &s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp) if err != nil && catalogErrCode(err) == s3tables.ErrCodeBucketNotEmpty { // The implicit index namespace counts as bucket content even when it @@ -423,7 +427,7 @@ func (s *Server) handleDeleteVectorBucket(w http.ResponseWriter, r *http.Request Namespace: []string{indexNamespace}, }, &nsResp) if nsErr == nil { - err = s.internalCatalog(r.Context(), "DeleteTableBucket", + err = s.internalCatalog(ctx, "DeleteTableBucket", &s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp) } else if code := catalogErrCode(nsErr); code != s3tables.ErrCodeNoSuchNamespace { glog.V(1).Infof("S3 Vectors: drop internal namespace in %s: %v", arns.bucket, nsErr) @@ -752,14 +756,26 @@ func isVectorBucket(record *s3tables.GetTableBucketResponse) bool { // mutation; catalogErr renders it as AccessDenied. var errVectorDenied = &s3tables.S3TablesError{Type: s3tables.ErrCodeAccessDenied, Message: "access denied"} +var errVectorBucketReplaced = &s3tables.S3TablesError{Type: s3tables.ErrCodeConflict, + Message: "vector bucket changed concurrently, retry the request"} + // checkVectorPolicyDoc evaluates the caller's vector action against the // document the entry currently holds. Called inside a policy mutation's // compare-and-swap, so a grant revoked between reads never authorizes the -// pending write. -func (s *Server) checkVectorPolicyDoc(r *http.Request, action, bucket, owner string, ext map[string][]byte) bool { +// pending write, and the entry must still be the bucket record was read from. +func (s *Server) checkVectorPolicyDoc(r *http.Request, action, bucket string, record *s3tables.GetTableBucketResponse, ext map[string][]byte) error { + var meta struct { + CreatedAt time.Time `json:"createdAt"` + } + if json.Unmarshal(ext[s3tables.ExtendedKeyMetadata], &meta) != nil || !meta.CreatedAt.Equal(record.CreatedAt) { + return errVectorBucketReplaced + } principal := s.tables.PrincipalFor(r) - return s3tables.CheckPermissionWithContext(action, principal, owner, - vectorPolicyFromExt(ext), s.vectorBucketArn(bucket), s.tables.PolicyContext(r, bucket)) + if !s3tables.CheckPermissionWithContext(action, principal, record.OwnerAccountID, + vectorPolicyFromExt(ext), s.vectorBucketArn(bucket), s.tables.PolicyContext(r, bucket)) { + return errVectorDenied + } + return nil } // authorizeVector grants the request's own vector operation — not a catalog @@ -818,8 +834,8 @@ func (s *Server) handlePutVectorBucketPolicy(w http.ResponseWriter, r *http.Requ return } err = s.mutateVectorPolicy(r.Context(), arns.bucket, func(ext map[string][]byte) error { - if !s.checkVectorPolicyDoc(r, "s3vectors:PutVectorBucketPolicy", arns.bucket, record.OwnerAccountID, ext) { - return errVectorDenied + if err := s.checkVectorPolicyDoc(r, "s3vectors:PutVectorBucketPolicy", arns.bucket, record, ext); err != nil { + return err } // A raw vector document still in the shared slot migrates out with // this write; catalog rules in it stay untouched. @@ -889,8 +905,8 @@ func (s *Server) handleDeleteVectorBucketPolicy(w http.ResponseWriter, r *http.R return } err = s.mutateVectorPolicy(r.Context(), arns.bucket, func(ext map[string][]byte) error { - if !s.checkVectorPolicyDoc(r, "s3vectors:DeleteVectorBucketPolicy", arns.bucket, record.OwnerAccountID, ext) { - return errVectorDenied + if err := s.checkVectorPolicyDoc(r, "s3vectors:DeleteVectorBucketPolicy", arns.bucket, record, ext); err != nil { + return err } removed := false if _, ok := ext[vectorPolicyKey]; ok { @@ -949,12 +965,15 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { } // The create grant is exercised before any write: a denied caller must not // leave even the implicit namespace behind. - if _, err := s.authorizeVector(r.Context(), r, "s3vectors:CreateIndex", arns.bucket, arns.index); err != nil { + bucketRecord, err := s.authorizeVector(r.Context(), r, "s3vectors:CreateIndex", arns.bucket, arns.index) + if err != nil { catalogErr(w, err) return } + // The catalog writes land only in the bucket authorized above. + bound := s3tables.WithTableBucketCreatedAt(r.Context(), bucketRecord.CreatedAt) // The namespace is implicit on AWS and internal bookkeeping here. - if err := s.ensureIndexNamespace(r.Context(), arns.bucket); err != nil { + if err := s.ensureIndexNamespace(bound, arns.bucket); err != nil { catalogErr(w, err) return } @@ -989,7 +1008,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { } } metadataJSON, _ := json.Marshal(metadata) - err = s.internalCatalog(r.Context(), "CreateTable", &s3tables.CreateTableRequest{ + err = s.internalCatalog(bound, "CreateTable", &s3tables.CreateTableRequest{ TableBucketARN: s.s3tablesArn(arns.bucket), Namespace: []string{indexNamespace}, Name: arns.index, @@ -1189,7 +1208,8 @@ func (s *Server) handleDeleteIndex(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } - if _, err := s.authorizeVector(r.Context(), r, "s3vectors:DeleteIndex", arns.bucket, arns.index); err != nil { + bucketRecord, err := s.authorizeVector(r.Context(), r, "s3vectors:DeleteIndex", arns.bucket, arns.index) + if err != nil { catalogErr(w, err) return } @@ -1213,8 +1233,9 @@ func (s *Server) handleDeleteIndex(w http.ResponseWriter, r *http.Request) { // Deleting the catalog record is the delete: the catalog purges the // dataset under the index's location along with the entry. var resp map[string]interface{} - // The token limits the delete to the record checked above. - err = s.internalCatalog(r.Context(), "DeleteTable", &s3tables.DeleteTableRequest{ + // The token limits the delete to the record checked above, and the bound + // bucket to the one authorized. + err = s.internalCatalog(s3tables.WithTableBucketCreatedAt(r.Context(), bucketRecord.CreatedAt), "DeleteTable", &s3tables.DeleteTableRequest{ TableBucketARN: s.s3tablesArn(arns.bucket), Namespace: []string{indexNamespace}, Name: arns.index,