diff --git a/weed/s3api/lance/handlers_table.go b/weed/s3api/lance/handlers_table.go index 7c63fd6ce8..65c0e1a11d 100644 --- a/weed/s3api/lance/handlers_table.go +++ b/weed/s3api/lance/handlers_table.go @@ -339,6 +339,10 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) { writeError(w, r, http.StatusConflict, codeTableAlreadyExists, "table already exists") return } + if err := s.tablesManager.CheckVectorNamespaceWrite(r.Context(), s.filerClient, bucket, ns); err != nil { + writeStorageError(w, r, err) + return + } if existing.location != location { // Repointing the name at another dataset is an update. Dropping and // recreating the entry would take the old dataset's files with it, @@ -394,6 +398,10 @@ func (s *Server) handleDeregisterTable(w http.ResponseWriter, r *http.Request) { writeStorageError(w, r, err) return } + if err := s.tablesManager.CheckVectorNamespaceWrite(r.Context(), s.filerClient, bucket, ns); err != nil { + writeStorageError(w, r, err) + return + } // Deregistering is a state, not a deletion. Dropping the catalog entry would // take the dataset with it, because the entry is the dataset directory and diff --git a/weed/s3api/lance/vector_namespace_test.go b/weed/s3api/lance/vector_namespace_test.go new file mode 100644 index 0000000000..01ec1770a8 --- /dev/null +++ b/weed/s3api/lance/vector_namespace_test.go @@ -0,0 +1,77 @@ +package lance + +import ( + "net/http" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// seedVectorIndex creates bucket/vector/idx through the trusted path, the way +// the S3 Vectors gateway does before it asks the engine to build the index. +func (h *testHarness) seedVectorIndex(t *testing.T, bucket, format string) { + t.Helper() + h.createBucket(t, bucket, format) + client := s3tables.NewManagerClient(h.filer.Client) + for _, op := range []struct { + name string + req interface{} + }{ + {"CreateNamespace", &s3tables.CreateNamespaceRequest{TableBucketARN: h.bucketARN(t, bucket), Namespace: []string{s3tables.VectorIndexNamespace}}}, + {"CreateNamespace", &s3tables.CreateNamespaceRequest{TableBucketARN: h.bucketARN(t, bucket), Namespace: []string{"other"}}}, + {"CreateTable", &s3tables.CreateTableRequest{TableBucketARN: h.bucketARN(t, bucket), Namespace: []string{s3tables.VectorIndexNamespace}, + Name: "idx", Format: s3tables.FormatLance}}, + } { + if err := h.admin.Execute(t.Context(), client, op.name, op.req, nil, ""); err != nil { + t.Fatalf("%s: %v", op.name, err) + } + } +} + +// Through the Lance namespace the vector namespace of a vector bucket is +// read-only too, while the engine's declare of an index the gateway already +// created keeps working. +func TestVectorNamespaceIsReadOnlyOverLance(t *testing.T) { + h := newTestHarness(t) + h.seedVectorIndex(t, "vec", s3tables.FormatLance) + idx := "/v1/table/vec$vector$idx" + + h.mustDo(t, http.MethodPost, idx+"/declare", `{"id":["vec","vector","idx"]}`, http.StatusOK) + h.mustDo(t, http.MethodPost, idx+"/describe", `{}`, http.StatusOK) + + for _, tc := range []struct{ name, target, body string }{ + {"declare a new index", "/v1/table/vec$vector$fresh/declare", `{}`}, + {"register a new index", "/v1/table/vec$vector$fresh/register", `{"location":"s3://vec/vector/fresh"}`}, + {"register over the index", idx + "/register", `{"location":"s3://vec/vector/idx","mode":"overwrite"}`}, + {"repoint the index", idx + "/register", `{"location":"s3://vec/vector/elsewhere","mode":"overwrite"}`}, + {"deregister", idx + "/deregister", `{}`}, + {"drop", idx + "/drop", `{}`}, + {"rename out", idx + "/rename", `{"new_id":["vec","other","moved"]}`}, + {"drop the namespace", "/v1/namespace/vec$vector/drop", `{}`}, + {"overwrite the namespace", "/v1/namespace/vec$vector/create", `{"mode":"overwrite"}`}, + } { + t.Run(tc.name, func(t *testing.T) { + h.mustDo(t, http.MethodPost, tc.target, tc.body, http.StatusForbidden) + }) + } + + if h.filer.Get("/buckets/vec/vector/idx", deregisteredMarker) != nil { + t.Fatal("a refused deregister left its marker") + } + h.mustDo(t, http.MethodPost, idx+"/exists", `{}`, http.StatusOK) + + h.createBucket(t, "fresh", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/namespace/fresh$vector/create", `{}`, http.StatusForbidden) +} + +// A bucket from before formats existed is not a vector bucket here, so its +// namespace named vector keeps the ordinary Lance lifecycle. +func TestVectorNamespaceOfUndeclaredBucketIsWritableOverLance(t *testing.T) { + h := newTestHarness(t) + h.seedVectorIndex(t, "old", "") + idx := "/v1/table/old$vector$idx" + + h.mustDo(t, http.MethodPost, idx+"/deregister", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, idx+"/register", `{"location":"s3://old/vector/idx"}`, http.StatusOK) + h.mustDo(t, http.MethodPost, idx+"/drop", `{}`, http.StatusOK) +} diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 3350ef1f43..2666480e40 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -143,6 +143,10 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request, // the status the handler answers decides how its error is logged (logRequestError) sw := &statusRecorder{ResponseWriter: w} w = sw + fence := h.fenceVectorNamespace(w, r, filerClient, operation) + if fence != nil { + w, filerClient = fence, fence + } var err error switch operation { @@ -239,6 +243,9 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request, return } + if fence != nil { + fence.settle() + } if err != nil { logRequestError(operation, sw.status, err) } diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index 6986109993..c051a5ecb9 100644 --- a/weed/s3api/s3tables/manager.go +++ b/weed/s3api/s3tables/manager.go @@ -95,9 +95,9 @@ func (m *Manager) AuthorizeCreateTable(ctx context.Context, filerClient FilerCli return err } recorder := httptest.NewRecorder() - _, authErr := m.handler.authorizeCreateTable(recorder, httpReq, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags, req.AllowMissingNamespace) + target, authErr := m.handler.authorizeCreateTable(recorder, httpReq, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags, req.AllowMissingNamespace) if authErr == nil { - return nil + return m.handler.checkVectorNamespaceWrite(ctx, filerClient, target.bucketName, req.Namespace) } if decoded := decodeS3TablesHTTPResponse(recorder, nil); decoded != nil { return decoded diff --git a/weed/s3api/s3tables/vector_namespace.go b/weed/s3api/s3tables/vector_namespace.go new file mode 100644 index 0000000000..028518683b --- /dev/null +++ b/weed/s3api/s3tables/vector_namespace.go @@ -0,0 +1,208 @@ +package s3tables + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" + "strings" + "sync/atomic" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/grpc" +) + +// VectorIndexNamespace holds a vector bucket's indexes. Only the S3 Vectors +// gateway, through a trusted Manager, may change it in a LANCE bucket. +const VectorIndexNamespace = "vector" + +const vectorNamespaceReadOnly = "the vector namespace of a vector bucket is read-only; manage its indexes through S3 Vectors" + +var errVectorNamespaceReadOnly = errors.New(vectorNamespaceReadOnly) + +func isVectorIndexNamespace(namespace []string) bool { + name, err := validateNamespace(namespace) + return err == nil && name == VectorIndexNamespace +} + +func (h *S3TablesHandler) checkVectorNamespaceWrite(ctx context.Context, filerClient FilerClient, bucketName string, namespace []string) error { + if h.trusted || !isVectorIndexNamespace(namespace) { + return nil + } + var metadata tableBucketMetadata + err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + data, err := h.getExtendedAttribute(ctx, client, GetTableBucketPath(bucketName), ExtendedKeyMetadata) + if err != nil { + return err + } + return json.Unmarshal(data, &metadata) + }) + if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) { + return nil + } + if err != nil { + return err + } + if metadata.Format == FormatLance { + return &S3TablesError{Type: ErrCodeAccessDenied, Message: vectorNamespaceReadOnly} + } + return nil +} + +// fenceVectorNamespace returns a fence for an untrusted write naming the +// vector namespace of a LANCE bucket. The request still runs, so the +// operation's own checks answer everything short of a write and an ungranted +// caller gets its usual denial. A CreateTable of an index that already exists +// is the vector engine's declare and passes unfenced. +func (h *S3TablesHandler) fenceVectorNamespace(w http.ResponseWriter, r *http.Request, filerClient FilerClient, operation string) *vectorNamespaceFence { + if h.trusted || strings.HasPrefix(operation, "Get") || strings.HasPrefix(operation, "List") { + return nil + } + body, _ := io.ReadAll(io.LimitReader(r.Body, MaxRequestBodySize+1)) + r.Body = io.NopCloser(bytes.NewReader(body)) + + var req struct { + TableBucketARN string `json:"tableBucketARN"` + Namespace []string `json:"namespace"` + SourceNamespace []string `json:"sourceNamespace"` + DestNamespace []string `json:"destNamespace"` + ResourceARN string `json:"resourceArn"` + Name string `json:"name"` + } + // Like the handler's own decode, this keeps going past a mistyped field. + _ = json.Unmarshal(body, &req) + type target struct { + bucket string + namespace []string + } + bucketName, _ := parseBucketNameFromARN(req.TableBucketARN) + targets := []target{{bucketName, req.Namespace}, {bucketName, req.SourceNamespace}, {bucketName, req.DestNamespace}} + if bucket, namespace, _, err := parseTableFromARN(req.ResourceARN); err == nil { + targets = append(targets, target{bucket, []string{namespace}}) + } + for _, t := range targets { + if t.bucket == "" || h.checkVectorNamespaceWrite(r.Context(), filerClient, t.bucket, t.namespace) == nil { + continue + } + if operation == "CreateTable" && isVectorIndexNamespace(req.Namespace) && h.vectorIndexExists(r.Context(), filerClient, bucketName, req.Name) { + return nil + } + return &vectorNamespaceFence{ResponseWriter: w, h: h, filer: filerClient} + } + return nil +} + +func (h *S3TablesHandler) vectorIndexExists(ctx context.Context, filerClient FilerClient, bucketName, name string) bool { + if _, err := validateTableName(name); err != nil { + return false + } + exists := false + _ = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + entry, err := h.lookupEntry(ctx, client, GetTablePath(bucketName, VectorIndexNamespace, name)) + if err != nil { + return err + } + _, hasMetadata := entry.Extended[ExtendedKeyMetadata] + exists = hasMetadata && !isUnfinishedCreate(entry.Extended) && EntryType(entry.Extended) == EntryTypeTable + return nil + }) + return exists +} + +// vectorNamespaceFence passes reads, refuses filer writes, and answers the +// request as read-only once a write was refused. Handlers write only after +// their permission check, so an ungranted caller never sees that answer. +type vectorNamespaceFence struct { + http.ResponseWriter + h *S3TablesHandler + filer FilerClient + tripped atomic.Bool + decided bool + refusing bool +} + +func (f *vectorNamespaceFence) WithFilerClient(streamingMode bool, fn func(client filer_pb.SeaweedFilerClient) error) error { + return f.filer.WithFilerClient(streamingMode, func(client filer_pb.SeaweedFilerClient) error { + return fn(readOnlyFilerClient{SeaweedFilerClient: client, tripped: &f.tripped}) + }) +} + +func (f *vectorNamespaceFence) WriteHeader(status int) { + if !f.settle() { + f.ResponseWriter.WriteHeader(status) + } +} + +func (f *vectorNamespaceFence) Write(b []byte) (int, error) { + if f.settle() { + return len(b), nil + } + return f.ResponseWriter.Write(b) +} + +// settle decides, at the first response write, whether the handler's answer +// is replaced by the read-only refusal. +func (f *vectorNamespaceFence) settle() bool { + if !f.decided { + f.decided = true + f.refusing = f.tripped.Load() + if f.refusing { + f.h.writeError(f.ResponseWriter, http.StatusForbidden, ErrCodeAccessDenied, vectorNamespaceReadOnly) + } + } + return f.refusing +} + +type readOnlyFilerClient struct { + filer_pb.SeaweedFilerClient + tripped *atomic.Bool +} + +func (c readOnlyFilerClient) refuse() error { + c.tripped.Store(true) + return errVectorNamespaceReadOnly +} + +func (c readOnlyFilerClient) CreateEntry(context.Context, *filer_pb.CreateEntryRequest, ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) UpdateEntry(context.Context, *filer_pb.UpdateEntryRequest, ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) AppendToEntry(context.Context, *filer_pb.AppendToEntryRequest, ...grpc.CallOption) (*filer_pb.AppendToEntryResponse, error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) DeleteEntry(context.Context, *filer_pb.DeleteEntryRequest, ...grpc.CallOption) (*filer_pb.DeleteEntryResponse, error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) AtomicRenameEntry(context.Context, *filer_pb.AtomicRenameEntryRequest, ...grpc.CallOption) (*filer_pb.AtomicRenameEntryResponse, error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) StreamRenameEntry(context.Context, *filer_pb.StreamRenameEntryRequest, ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.StreamRenameEntryResponse], error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) StreamMutateEntry(context.Context, ...grpc.CallOption) (grpc.BidiStreamingClient[filer_pb.StreamMutateEntryRequest, filer_pb.StreamMutateEntryResponse], error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) ObjectTransaction(context.Context, *filer_pb.ObjectTransactionRequest, ...grpc.CallOption) (*filer_pb.ObjectTransactionResponse, error) { + return nil, c.refuse() +} + +func (c readOnlyFilerClient) ObjectTransactionBatch(context.Context, *filer_pb.ObjectTransactionBatchRequest, ...grpc.CallOption) (*filer_pb.ObjectTransactionBatchResponse, error) { + return nil, c.refuse() +} + +// CheckVectorNamespaceWrite refuses an untrusted write into the vector +// namespace of a LANCE bucket, for callers that write the filer directly. +func (m *Manager) CheckVectorNamespaceWrite(ctx context.Context, filerClient FilerClient, bucketName string, namespace []string) error { + return m.handler.checkVectorNamespaceWrite(ctx, filerClient, bucketName, namespace) +} diff --git a/weed/s3api/s3tables/vector_namespace_test.go b/weed/s3api/s3tables/vector_namespace_test.go new file mode 100644 index 0000000000..4f43aada37 --- /dev/null +++ b/weed/s3api/s3tables/vector_namespace_test.go @@ -0,0 +1,287 @@ +package s3tables + +import ( + "context" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const vectorTestBucket = "vecns" + +func vectorTestARN(t *testing.T) string { + t.Helper() + arn, err := BuildBucketARN(DefaultRegion, DefaultAccountID, vectorTestBucket) + require.NoError(t, err) + return arn +} + +// seedVectorBucket lays down, through a trusted Manager, a bucket of the given +// format holding vector/idx (with a policy and a tag) and other/t, the shape +// the S3 Vectors gateway leaves behind. +func seedVectorBucket(t *testing.T, fs *s3tablestest.MemFiler, format string) *Manager { + t.Helper() + trusted := NewManager() + trusted.SetTrusted(true) + run := func(operation string, req interface{}) { + t.Helper() + require.NoError(t, trusted.Execute(context.Background(), NewManagerClient(fs.Client), operation, req, nil, "")) + } + arn := vectorTestARN(t) + run("CreateTableBucket", &CreateTableBucketRequest{Name: vectorTestBucket, Format: format}) + for _, ns := range []string{VectorIndexNamespace, "other"} { + run("CreateNamespace", &CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{ns}}) + } + run("CreateTable", &CreateTableRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}, Name: "idx", Format: format}) + run("CreateTable", &CreateTableRequest{TableBucketARN: arn, Namespace: []string{"other"}, Name: "t", Format: format}) + run("PutTablePolicy", &PutTablePolicyRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}, Name: "idx", + ResourcePolicy: `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:GetTable","Resource":"*"}]}`}) + idx, err := getTestTable(t, trusted, fs, VectorIndexNamespace, "idx") + require.NoError(t, err) + run("TagResource", &TagResourceRequest{ResourceARN: idx.TableARN, Tags: map[string]string{"k": "v"}}) + return trusted +} + +func getTestTable(t *testing.T, m *Manager, fs *s3tablestest.MemFiler, namespace, name string) (*GetTableResponse, error) { + t.Helper() + var resp GetTableResponse + err := m.Execute(context.Background(), NewManagerClient(fs.Client), "GetTable", &GetTableRequest{ + TableBucketARN: vectorTestARN(t), Namespace: []string{namespace}, Name: name}, &resp, "") + return &resp, err +} + +// vectorWrite labels a case "Operation detail" when an operation appears twice. +type vectorWrite struct { + operation string + request interface{} +} + +func (w vectorWrite) op() string { + op, _, _ := strings.Cut(w.operation, " ") + return op +} + +// vectorNamespaceWrites lists the catalog writes that reach vector/idx or move +// a table across the namespace boundary. +func vectorNamespaceWrites(t *testing.T, indexARN, versionToken string) []vectorWrite { + arn := vectorTestARN(t) + vector := []string{VectorIndexNamespace} + return []vectorWrite{ + {"UpdateNamespace", &UpdateNamespaceRequest{TableBucketARN: arn, Namespace: vector, Properties: map[string]string{"a": "b"}}}, + {"CreateTable", &CreateTableRequest{TableBucketARN: arn, Namespace: vector, Name: "fresh", Format: FormatLance}}, + {"RegisterTable", &RegisterTableRequest{TableBucketARN: arn, Namespace: vector, Name: "registered", + MetadataLocation: "s3://" + vectorTestBucket + "/vector/registered/metadata/v1.metadata.json"}}, + {"UpdateTable", &UpdateTableRequest{TableBucketARN: arn, Namespace: vector, Name: "idx", VersionToken: versionToken, + MetadataLocation: "s3://" + vectorTestBucket + "/vector/elsewhere"}}, + {"DeleteTable", &DeleteTableRequest{TableBucketARN: arn, Namespace: vector, Name: "idx"}}, + {"RenameTable out", &RenameTableRequest{TableBucketARN: arn, SourceNamespace: vector, SourceName: "idx", + DestNamespace: []string{"other"}, DestName: "moved"}}, + {"RenameTable in", &RenameTableRequest{TableBucketARN: arn, SourceNamespace: []string{"other"}, SourceName: "t", + DestNamespace: vector, DestName: "t"}}, + {"PutTablePolicy", &PutTablePolicyRequest{TableBucketARN: arn, Namespace: vector, Name: "idx", + ResourcePolicy: `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:*","Resource":"*"}]}`}}, + {"DeleteTablePolicy", &DeleteTablePolicyRequest{TableBucketARN: arn, Namespace: vector, Name: "idx"}}, + {"TagResource", &TagResourceRequest{ResourceARN: indexARN, Tags: map[string]string{"x": "y"}}}, + {"UntagResource", &UntagResourceRequest{ResourceARN: indexARN, TagKeys: []string{"k"}}}, + {"PutTableMaintenanceConfiguration", &PutTableMaintenanceConfigurationRequest{TableBucketARN: arn, Namespace: vector, Name: "idx", + Type: MaintenanceTypeIcebergCompaction, Value: &MaintenanceConfigurationValue{Status: "enabled"}}}, + } +} + +func untrustedManager() *Manager { + m := NewManager() + m.SetDefaultAllow(false) + return m +} + +// A caller holding s3tables:* still cannot change the index namespace of a +// vector bucket, and nothing it tried is left behind. +func TestVectorNamespaceRefusesUntrustedWrites(t *testing.T) { + fs := s3tablestest.Start(t) + trusted := seedVectorBucket(t, fs, FormatLance) + before, err := getTestTable(t, trusted, fs, VectorIndexNamespace, "idx") + require.NoError(t, err) + m := untrustedManager() + granted := []string{"s3tables:*"} + + for _, tc := range vectorNamespaceWrites(t, before.TableARN, before.VersionToken) { + t.Run(tc.operation, func(t *testing.T) { + got := runRequestAs(t, m, fs, tc.op(), tc.request, granted) + assert.Equal(t, 403, got.status) + assert.Equal(t, ErrCodeAccessDenied, got.body.Type) + assert.Equal(t, vectorNamespaceReadOnly, got.body.Message) + }) + } + + after, err := getTestTable(t, trusted, fs, VectorIndexNamespace, "idx") + require.NoError(t, err) + assert.Equal(t, before.VersionToken, after.VersionToken) + assert.Equal(t, before.MetadataLocation, after.MetadataLocation) + _, err = getTestTable(t, trusted, fs, "other", "t") + assert.NoError(t, err) + for _, absent := range [][2]string{{VectorIndexNamespace, "fresh"}, {VectorIndexNamespace, "registered"}, {VectorIndexNamespace, "t"}, {"other", "moved"}} { + _, err := getTestTable(t, trusted, fs, absent[0], absent[1]) + assert.Error(t, err, "%s/%s should not exist", absent[0], absent[1]) + } + var tags ListTagsForResourceResponse + require.NoError(t, trusted.Execute(context.Background(), NewManagerClient(fs.Client), "ListTagsForResource", + &ListTagsForResourceRequest{ResourceARN: before.TableARN}, &tags, "")) + assert.Equal(t, map[string]string{"k": "v"}, tags.Tags) +} + +func TestVectorNamespaceCannotBeCreatedOrDeleted(t *testing.T) { + fs := s3tablestest.Start(t) + trusted := NewManager() + trusted.SetTrusted(true) + m := untrustedManager() + granted := []string{"s3tables:*"} + arn := vectorTestARN(t) + ctx := context.Background() + client := NewManagerClient(fs.Client) + require.NoError(t, trusted.Execute(ctx, client, "CreateTableBucket", &CreateTableBucketRequest{Name: vectorTestBucket, Format: FormatLance}, nil, "")) + + create := &CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}} + got := runRequestAs(t, m, fs, "CreateNamespace", create, granted) + assert.Equal(t, 403, got.status) + assert.Equal(t, vectorNamespaceReadOnly, got.body.Message) + + require.NoError(t, trusted.Execute(ctx, client, "CreateNamespace", create, nil, "")) + got = runRequestAs(t, m, fs, "DeleteNamespace", &DeleteNamespaceRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}}, granted) + assert.Equal(t, 403, got.status) + assert.Equal(t, vectorNamespaceReadOnly, got.body.Message) + require.NoError(t, trusted.Execute(ctx, client, "GetNamespace", &GetNamespaceRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}}, nil, "")) + + other := &CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{"other"}} + assert.Equal(t, 200, runRequestAs(t, m, fs, "CreateNamespace", other, granted).status) +} + +// The vector engine declares an index the gateway already created; that +// CreateTable writes nothing and keeps succeeding, as do reads. +func TestVectorNamespaceAllowsReadsAndExistingCreate(t *testing.T) { + fs := s3tablestest.Start(t) + trusted := seedVectorBucket(t, fs, FormatLance) + before, err := getTestTable(t, trusted, fs, VectorIndexNamespace, "idx") + require.NoError(t, err) + m := untrustedManager() + granted := []string{"s3tables:*"} + arn := vectorTestARN(t) + vector := []string{VectorIndexNamespace} + + for _, tc := range []vectorWrite{ + {"CreateTable", &CreateTableRequest{TableBucketARN: arn, Namespace: vector, Name: "idx", Format: FormatLance}}, + {"GetTable", &GetTableRequest{TableBucketARN: arn, Namespace: vector, Name: "idx"}}, + {"ListTables", &ListTablesRequest{TableBucketARN: arn, Namespace: vector}}, + {"GetNamespace", &GetNamespaceRequest{TableBucketARN: arn, Namespace: vector}}, + {"GetTablePolicy", &GetTablePolicyRequest{TableBucketARN: arn, Namespace: vector, Name: "idx"}}, + {"ListTagsForResource", &ListTagsForResourceRequest{ResourceARN: before.TableARN}}, + } { + t.Run(tc.operation, func(t *testing.T) { + assert.Equal(t, 200, runRequestAs(t, m, fs, tc.op(), tc.request, granted).status) + }) + } + after, err := getTestTable(t, trusted, fs, VectorIndexNamespace, "idx") + require.NoError(t, err) + assert.Equal(t, before.VersionToken, after.VersionToken) + + require.NoError(t, trusted.Execute(context.Background(), NewManagerClient(fs.Client), "DeleteTable", + &DeleteTableRequest{TableBucketARN: arn, Namespace: vector, Name: "idx"}, nil, ""), "the trusted path stays writable") +} + +// An Iceberg bucket's namespace named vector is an ordinary namespace. +func TestVectorNamespaceOfIcebergBucketIsWritable(t *testing.T) { + fs := s3tablestest.Start(t) + trusted := seedVectorBucket(t, fs, FormatIceberg) + before, err := getTestTable(t, trusted, fs, VectorIndexNamespace, "idx") + require.NoError(t, err) + m := untrustedManager() + granted := []string{"s3tables:*"} + + for _, tc := range vectorNamespaceWrites(t, before.TableARN, before.VersionToken) { + switch tc.operation { + case "CreateTable", "UpdateTable", "DeleteTable", "RenameTable out": + continue // covered below: they need an Iceberg request or would consume idx + } + t.Run(tc.operation, func(t *testing.T) { + got := runRequestAs(t, m, fs, tc.op(), tc.request, granted) + assert.Equal(t, 200, got.status, "%+v", got.body) + }) + } + arn := vectorTestARN(t) + vector := []string{VectorIndexNamespace} + assert.Equal(t, 200, runRequestAs(t, m, fs, "CreateTable", + &CreateTableRequest{TableBucketARN: arn, Namespace: vector, Name: "fresh", Format: FormatIceberg}, granted).status) + assert.Equal(t, 200, runRequestAs(t, m, fs, "DeleteTable", + &DeleteTableRequest{TableBucketARN: arn, Namespace: vector, Name: "idx"}, granted).status) +} + +// A caller without a grant gets the answer an Iceberg bucket of the same name +// and shape gives: the read-only rule never tells it the bucket exists or +// holds vectors. +func TestVectorNamespaceDenialHidesBucketFormat(t *testing.T) { + lanceFS := s3tablestest.Start(t) + lance := seedVectorBucket(t, lanceFS, FormatLance) + icebergFS := s3tablestest.Start(t) + seedVectorBucket(t, icebergFS, FormatIceberg) + idx, err := getTestTable(t, lance, lanceFS, VectorIndexNamespace, "idx") + require.NoError(t, err) + m := untrustedManager() + + arn := vectorTestARN(t) + writes := append(vectorNamespaceWrites(t, idx.TableARN, idx.VersionToken), + vectorWrite{"CreateNamespace", &CreateNamespaceRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}}}, + vectorWrite{"DeleteNamespace", &DeleteNamespaceRequest{TableBucketARN: arn, Namespace: []string{VectorIndexNamespace}}}) + for _, tc := range writes { + t.Run(tc.operation, func(t *testing.T) { + got := runUnauthorizedRequest(t, m, lanceFS, tc.op(), tc.request) + want := runUnauthorizedRequest(t, m, icebergFS, tc.op(), tc.request) + assert.Equal(t, want, got) + assert.NotEqual(t, vectorNamespaceReadOnly, got.body.Message) + }) + } +} + +// A deferred create writes into the bucket before it registers anything, so +// its gate refuses the vector namespace as the create itself would. +func TestAuthorizeCreateTableRefusesVectorNamespace(t *testing.T) { + identity := &testIdentity{Name: "reader", Account: &testIdentityAccount{Id: "reader"}, Actions: []string{"s3tables:*"}} + ctx := s3_constants.SetIdentityInContext(context.Background(), identity) + req := &CreateTableRequest{TableBucketARN: vectorTestARN(t), Namespace: []string{VectorIndexNamespace}, Name: "fresh"} + + lanceFS := s3tablestest.Start(t) + seedVectorBucket(t, lanceFS, FormatLance) + err := untrustedManager().AuthorizeCreateTable(ctx, NewManagerClient(lanceFS.Client), req, "reader") + var apiErr *S3TablesError + require.ErrorAs(t, err, &apiErr) + assert.Equal(t, ErrCodeAccessDenied, apiErr.Type) + assert.Equal(t, vectorNamespaceReadOnly, apiErr.Message) + + icebergFS := s3tablestest.Start(t) + seedVectorBucket(t, icebergFS, FormatIceberg) + assert.NoError(t, untrustedManager().AuthorizeCreateTable(ctx, NewManagerClient(icebergFS.Client), req, "reader")) +} + +// Only a CreateTable that finds a finished index passes unfenced, so the +// declare keeps working even where the idempotent path writes. +func TestVectorNamespaceFenceSparesOnlyExistingIndexCreate(t *testing.T) { + fs := s3tablestest.Start(t) + seedVectorBucket(t, fs, FormatLance) + fs.Put(GetNamespacePath(vectorTestBucket, VectorIndexNamespace), "pending", map[string][]byte{ + ExtendedKeyMetadata: []byte(`{"name":"pending","format":"LANCE"}`), + ExtendedKeyDeleting: []byte(createClaim), + }) + h := untrustedManager().handler + + for name, fenced := range map[string]bool{"idx": false, "fresh": true, "pending": true} { + t.Run(name, func(t *testing.T) { + body := `{"tableBucketARN":"` + vectorTestARN(t) + `","namespace":["vector"],"name":"` + name + `","format":"LANCE"}` + req, err := newManagerRequest(context.Background(), "CreateTable", []byte(body), "reader") + require.NoError(t, err) + got := h.fenceVectorNamespace(nil, req, NewManagerClient(fs.Client), "CreateTable") + assert.Equal(t, fenced, got != nil) + }) + } +} diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 0c20fa228d..983de36233 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -41,7 +41,7 @@ var handlers = map[string]func(*Server, http.ResponseWriter, *http.Request){ } // lance table ids put indexes under a "vector" namespace inside their bucket. -const indexNamespace = "vector" +const indexNamespace = s3tables.VectorIndexNamespace // Vector bucket records are s3tables table buckets; the S3 Vectors ARN form is // arn:aws:s3vectors:{region}:{account}:bucket/{name}[/index/{name}]. diff --git a/weed/s3api/s3vectors/vector_namespace_test.go b/weed/s3api/s3vectors/vector_namespace_test.go new file mode 100644 index 0000000000..f3f474d6ba --- /dev/null +++ b/weed/s3api/s3vectors/vector_namespace_test.go @@ -0,0 +1,63 @@ +package s3vectors + +import ( + "context" + "fmt" + "net/http" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +// An identity the vector policy denies DeleteIndex must not delete the index +// through the S3 Tables API instead, while the gateway's own index lifecycle +// keeps working. +func TestCatalogCannotBypassVectorPolicy(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()) + } + for _, name := range []string{"idx", "spare"} { + if w := postJSONAs(t, "", s.handleCreateIndex, &CreateIndexRequest{VectorBucketName: "vec-bucket", IndexName: name, + DataType: "float32", Dimension: 3, DistanceMetric: "euclidean"}); w.Code != http.StatusOK { + t.Fatalf("create index %s: %d %s", name, w.Code, w.Body.String()) + } + } + policy := fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":"s3vectors:DeleteIndex","Resource":"%s"}]}`, + s.indexArn("vec-bucket", "idx")) + if w := postJSONAs(t, "", s.handlePutVectorBucketPolicy, &PutVectorBucketPolicyRequest{VectorBucketName: "vec-bucket", Policy: policy}); w.Code != http.StatusOK { + t.Fatalf("put policy: %d %s", w.Code, w.Body.String()) + } + + bob := &s3api.Identity{Name: "bob", Actions: []s3api.Action{"s3tables:*"}} + ctx := s3_constants.SetIdentityInContext(s3_constants.SetIdentityNameInContext(context.Background(), bob.Name), bob) + deleteReq := &DeleteIndexRequest{VectorBucketName: "vec-bucket", IndexName: "idx"} + if w := postJSONAs(t, "bob", s.handleDeleteIndex, deleteReq); w.Code != http.StatusForbidden { + t.Fatalf("gateway DeleteIndex under deny: %d %s, want 403", w.Code, w.Body.String()) + } + err := s.tables.Execute(ctx, s.filerClient, "DeleteTable", &s3tables.DeleteTableRequest{ + TableBucketARN: s.s3tablesArn("vec-bucket"), Namespace: []string{indexNamespace}, Name: "idx"}, nil, bob.Name) + if catalogErrCode(err) != s3tables.ErrCodeAccessDenied { + t.Fatalf("catalog DeleteTable of a vector index: %v, want AccessDenied", err) + } + if _, err := s.indexRecord(context.Background(), "vec-bucket", "idx"); err != nil { + t.Fatalf("index gone after a refused catalog delete: %v", err) + } + + if w := postJSONAs(t, "", s.handleDeleteIndex, &DeleteIndexRequest{VectorBucketName: "vec-bucket", IndexName: "spare"}); w.Code != http.StatusOK { + t.Fatalf("gateway DeleteIndex: %d %s", w.Code, w.Body.String()) + } + if w := postJSONAs(t, "", s.handleDeleteVectorBucketPolicy, &DeleteVectorBucketPolicyRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("delete policy: %d %s", w.Code, w.Body.String()) + } + if w := postJSONAs(t, "", s.handleDeleteIndex, deleteReq); w.Code != http.StatusOK { + t.Fatalf("gateway DeleteIndex: %d %s", w.Code, w.Body.String()) + } + if w := postJSONAs(t, "", s.handleDeleteVectorBucket, &DeleteVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("gateway DeleteVectorBucket: %d %s", w.Code, w.Body.String()) + } +}