mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-03 02:32:35 +01:00
fix(s3tables,s3vectors): bind table and vector bucket writes to the bucket they were authorized against (#12329)
* s3tables: bind table bucket policy writes to the authorized bucket PutTableBucketPolicy compared only the policy bytes, and DeleteTableBucketPolicy removed the policy unconditionally. A bucket deleted and created again under the same name between authorization and the write took a policy change nobody authorized for it. Both writes now require the bucket's CreatedAt and the policy they were authorized against, answer 409 otherwise and 404 for a bucket that is gone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3tables: bind bucket tag writes to the authorized bucket TagResource and UntagResource checked the creation time only for a table ARN. On a bucket ARN the merged tags landed on a bucket deleted and created again since authorization. extractResourceOwnerAndBucket now returns the resource's CreatedAt for both kinds, and the write requires it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3tables: bind bucket maintenance writes to the authorized bucket PutTableBucketMaintenanceConfiguration wrote onto whatever bucket held the name at write time. authorizeMaintenanceBucket now returns the bucket's CreatedAt like authorizeMaintenanceTable, and putMaintenanceConfiguration takes the check and creation time from both callers instead of an optional closure. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3vectors: bind vector bucket policy writes to the authorized bucket Put/DeleteVectorBucketPolicy authorized inside the compare-and-swap with the owner from an earlier bucket read, so a bucket deleted and created again by another account in between had its policy rewritten by the old owner. The mutation now requires the entry's createdAt to match the record it authorized against. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3vectors: bind DeleteVectorBucket to the authorized bucket DeleteVectorBucket authorized one bucket record, then ran the trusted catalog DeleteTableBucket on whatever bucket held the name by then, so an empty bucket deleted and created again by another account in between was deleted by the old owner. The catalog delete now accepts the authorized bucket's createdAt through an in-process context value and treats any other bucket as gone. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3vectors: bind CreateIndex and DeleteIndex to the authorized bucket Both authorized the bucket, then ran trusted CreateNamespace, CreateTable or DeleteTable on whatever bucket held the name by then, so an index could be created in, or deleted from, a bucket another account created again in between. Those catalog operations now honor the bucket binding DeleteTableBucket already takes, and the vector handlers pass the authorized bucket's createdAt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
4850def831
commit
66a6598fdd
10 files changed
+464
-85
No files matched your search
@@ -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
|
||||
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user