diff --git a/test/s3vectors/s3vectors_test.py b/test/s3vectors/s3vectors_test.py index 51ebc4b5b4..05f83943ee 100644 --- a/test/s3vectors/s3vectors_test.py +++ b/test/s3vectors/s3vectors_test.py @@ -417,6 +417,40 @@ def test_index_arn_validation(vectors, index): assert vectors.get_index(**index)["index"]["indexArn"] == arn +def test_resource_tags(vectors, vector_clients, index): + reader = vector_clients["reader"] + bucket_arn = vectors.get_vector_bucket(vectorBucketName=index["vectorBucketName"])["vectorBucket"]["vectorBucketArn"] + index_arn = vectors.get_index(**index)["index"]["indexArn"] + assert vectors.list_tags_for_resource(resourceArn=bucket_arn)["tags"] == {} + vectors.tag_resource(resourceArn=bucket_arn, tags={"team": "search", "env": "dev"}) + vectors.tag_resource(resourceArn=index_arn, tags={"team": "ranking"}) + assert vectors.list_tags_for_resource(resourceArn=bucket_arn)["tags"] == {"team": "search", "env": "dev"} + assert vectors.list_tags_for_resource(resourceArn=index_arn)["tags"] == {"team": "ranking"} + vectors.untag_resource(resourceArn=bucket_arn, tagKeys=["env", "absent"]) + assert vectors.list_tags_for_resource(resourceArn=bucket_arn)["tags"] == {"team": "search"} + expect_error("ValidationException", vectors.tag_resource, resourceArn=bucket_arn, + tags={f"k{i}": "v" for i in range(50)}) + expect_error("NotFoundException", vectors.list_tags_for_resource, resourceArn=bucket_arn + "/index/missing") + expect_error("AccessDeniedException", reader.list_tags_for_resource, resourceArn=bucket_arn) + expect_error("AccessDeniedException", reader.tag_resource, resourceArn=index_arn, tags={"team": "other"}) + assert vectors.list_tags_for_resource(resourceArn=index_arn)["tags"] == {"team": "ranking"} + + +def test_create_with_tags(vectors): + bucket = "tagged-" + uuid.uuid4().hex[:16] + bucket_arn = vectors.create_vector_bucket(vectorBucketName=bucket, tags={"team": "search"})["vectorBucketArn"] + try: + assert vectors.list_tags_for_resource(resourceArn=bucket_arn)["tags"] == {"team": "search"} + index_arn = vectors.create_index(vectorBucketName=bucket, indexName=INDEX, dataType="float32", dimension=4, + distanceMetric="cosine", tags={"tier": "hot"})["indexArn"] + try: + assert vectors.list_tags_for_resource(resourceArn=index_arn)["tags"] == {"tier": "hot"} + finally: + vectors.delete_index(vectorBucketName=bucket, indexName=INDEX) + finally: + vectors.delete_vector_bucket(vectorBucketName=bucket) + + def test_invalid_credentials(cluster): client = boto3.client("s3vectors", endpoint_url=cluster["endpoint"], aws_access_key_id="unknown", aws_secret_access_key="unknown", region_name="us-east-1") @@ -458,3 +492,4 @@ def test_engine_without_credentials(open_cluster): {"key": "b", "data": {"float32": [0, 1]}}]) hits = client.query_vectors(**index, queryVector={"float32": [1, 0]}, topK=1)["vectors"] assert [h["key"] for h in hits] == ["a"] + diff --git a/weed/s3api/s3tables/handler_bind_tags_test.go b/weed/s3api/s3tables/handler_bind_tags_test.go new file mode 100644 index 0000000000..cce363229e --- /dev/null +++ b/weed/s3api/s3tables/handler_bind_tags_test.go @@ -0,0 +1,96 @@ +package s3tables + +import ( + "context" + "encoding/json" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Tag operations bound to a table bucket answer not found once its name +// holds another bucket, for the bucket and for a table in it, and leave the +// tags as they were. +func TestTagOperationsKeepToTheBoundBucket(t *testing.T) { + fs, m := startRenameManager(t) + client := NewManagerClient(fs.Client) + tableARN, err := BuildTableARN(DefaultRegion, DefaultAccountID, renameTestBucket, "ns", "t") + require.NoError(t, err) + var bucket GetTableBucketResponse + require.NoError(t, m.Execute(context.Background(), client, "GetTableBucket", + &GetTableBucketRequest{TableBucketARN: mustBucketARN(t)}, &bucket, "")) + bound := WithTableBucketCreatedAt(context.Background(), bucket.CreatedAt) + stale := WithTableBucketCreatedAt(context.Background(), bucket.CreatedAt.Add(-time.Hour)) + + for _, arn := range []string{mustBucketARN(t), tableARN} { + require.NoError(t, m.Execute(bound, client, "TagResource", + &TagResourceRequest{ResourceARN: arn, Tags: map[string]string{"k": "v"}}, nil, ""), arn) + for operation, req := range map[string]interface{}{ + "TagResource": &TagResourceRequest{ResourceARN: arn, Tags: map[string]string{"k": "other"}}, + "UntagResource": &UntagResourceRequest{ResourceARN: arn, TagKeys: []string{"k"}}, + "ListTagsForResource": &ListTagsForResourceRequest{ResourceARN: arn}, + } { + err := m.Execute(stale, client, operation, req, nil, "") + var apiErr *S3TablesError + require.ErrorAs(t, err, &apiErr, "%s on %s", operation, arn) + assert.Contains(t, []string{ErrCodeNoSuchBucket, ErrCodeNoSuchTable}, apiErr.Type, "%s on %s", operation, arn) + } + var tags ListTagsForResourceResponse + require.NoError(t, m.Execute(bound, client, "ListTagsForResource", + &ListTagsForResourceRequest{ResourceARN: arn}, &tags, ""), arn) + assert.Equal(t, map[string]string{"k": "v"}, tags.Tags, arn) + } +} + +// A tag write merges into the tags the record holds when it is written, so a +// tag write that lands between its read and its write is kept. +func TestTagWritesKeepAConcurrentTagWrite(t *testing.T) { + for _, tc := range []struct { + operation string + input func(t *testing.T) interface{} + want map[string]string + }{ + {"TagResource", func(t *testing.T) interface{} { + return &TagResourceRequest{ResourceARN: mustBucketARN(t), Tags: map[string]string{"k": "v"}} + }, map[string]string{"keep": "1", "other": "1", "k": "v"}}, + {"UntagResource", func(t *testing.T) interface{} { + return &UntagResourceRequest{ResourceARN: mustBucketARN(t), TagKeys: []string{"keep"}} + }, map[string]string{"other": "1"}}, + } { + t.Run(tc.operation, func(t *testing.T) { + fs, m := startEmptyBucket(t) + extended := extendedOf(t, fs, TablesPath, renameTestBucket) + extended[ExtendedKeyTags] = []byte(`{"keep":"1"}`) + fs.Put(TablesPath, renameTestBucket, extended) + fs.BeforeUpdate = func() { + current := extendedOf(t, fs, TablesPath, renameTestBucket) + current[ExtendedKeyTags] = []byte(`{"keep":"1","other":"1"}`) + fs.Put(TablesPath, renameTestBucket, current) + } + code, resp := runHandler(t, m, fs.Client, tc.operation, tc.input(t)) + require.Equal(t, http.StatusOK, code, resp.Message) + var got map[string]string + require.NoError(t, json.Unmarshal(extendedOf(t, fs, TablesPath, renameTestBucket)[ExtendedKeyTags], &got)) + assert.Equal(t, tc.want, got) + }) + } +} + +func TestTagResourceKeepsToItsLimit(t *testing.T) { + fs, m := startEmptyBucket(t) + client := NewManagerClient(fs.Client) + ctx := WithMaxTags(context.Background(), 2) + tag := func(tags map[string]string) error { + return m.Execute(ctx, client, "TagResource", &TagResourceRequest{ResourceARN: mustBucketARN(t), Tags: tags}, nil, "") + } + require.NoError(t, tag(map[string]string{"a": "1", "b": "1"})) + require.NoError(t, tag(map[string]string{"a": "2"})) + err := tag(map[string]string{"c": "1"}) + var apiErr *S3TablesError + require.ErrorAs(t, err, &apiErr) + assert.Equal(t, ErrCodeInvalidRequest, apiErr.Type) + assert.JSONEq(t, `{"a":"2","b":"1"}`, string(extendedOf(t, fs, TablesPath, renameTestBucket)[ExtendedKeyTags])) +} diff --git a/weed/s3api/s3tables/handler_policy.go b/weed/s3api/s3tables/handler_policy.go index 8483900637..cf455df912 100644 --- a/weed/s3api/s3tables/handler_policy.go +++ b/weed/s3api/s3tables/handler_policy.go @@ -2,6 +2,7 @@ package s3tables import ( "bytes" + "context" "encoding/json" "errors" "fmt" @@ -43,6 +44,65 @@ func (h *S3TablesHandler) extractResourceOwnerAndBucket( return ownerAccountID, bucketName, createdAt, nil } +type maxTagsKey struct{} + +var errTooManyTags = errors.New("too many tags") + +// WithMaxTags caps how many tags a TagResource run through Execute may leave +// on its resource. In-process only, like WithTableBucketCreatedAt. +func WithMaxTags(ctx context.Context, n int) context.Context { + return context.WithValue(ctx, maxTagsKey{}, n) +} + +// updateTags applies set and remove to raw, the tags the record holds at the +// write, so concurrent tag writes do not undo each other. +func updateTags(raw []byte, set map[string]string, remove []string) (map[string]string, error) { + tags := make(map[string]string) + if len(raw) > 0 { + if err := json.Unmarshal(raw, &tags); err != nil { + return nil, err + } + } + for k, v := range set { + tags[k] = v + } + for _, k := range remove { + delete(tags, k) + } + return tags, nil +} + +func writeTags(extended map[string][]byte, key string, tags map[string]string) error { + data, err := json.Marshal(tags) + if err != nil { + return err + } + extended[key] = data + return nil +} + +// requireBoundTableBucket answers not found when ctx binds the request to a +// table bucket other than the one now named bucketName. +func (h *S3TablesHandler) requireBoundTableBucket(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketName string) error { + if _, bound := ctx.Value(tableBucketCreatedAtKey{}).(time.Time); !bound { + return nil + } + data, err := h.getExtendedAttribute(ctx, client, GetTableBucketPath(bucketName), ExtendedKeyMetadata) + if errors.Is(err, ErrAttributeNotFound) { + return filer_pb.ErrNotFound + } else if err != nil { + return err + } + var meta tableBucketMetadata + if err := json.Unmarshal(data, &meta); err != nil { + return err + } + if otherTableBucket(ctx, meta.CreatedAt) { + return filer_pb.ErrNotFound + } + return nil +} + // handlePutTableBucketPolicy puts a policy on a table bucket func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { @@ -751,6 +811,10 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque return err } + // Checked after the reads, so they all came from the bound bucket. + if err := h.requireBoundTableBucket(r.Context(), client, bucketName); err != nil { + return err + } var resourceARN string if rType == ResourceTypeTable { resourceARN = h.generateTableARN(ownerAccountID, bucketName, tableNamespace+"/"+tableName) @@ -794,28 +858,24 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque return err } - // Merge new tags - for k, v := range req.Tags { - existingTags[k] = v - } - - // Write merged tags - tagsBytes, err := json.Marshal(existingTags) - if err != nil { - 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 } + maxTags, limited := r.Context().Value(maxTagsKey{}).(int) err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return h.mutateCatalogRecord(r.Context(), client, resourcePath, func(extended map[string][]byte) error { if err := sameResource(extended, resourcePath, createdAt); err != nil { return err } - extended[extendedKey] = tagsBytes - return nil + tags, err := updateTags(extended[extendedKey], req.Tags, nil) + if err != nil { + return err + } + if limited && len(tags) > maxTags { + return errTooManyTags + } + return writeTags(extended, extendedKey, tags) }) }) @@ -828,6 +888,11 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque h.writeError(w, http.StatusNotFound, errorCode, "resource not found") return err } + if errors.Is(err, errTooManyTags) { + h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, + fmt.Sprintf("a resource can have at most %d tags", maxTags)) + return err + } if errors.Is(err, ErrConcurrentUpdate) { h.writeError(w, http.StatusConflict, ErrCodeConflict, "resource tags changed concurrently, retry the request") return err @@ -909,6 +974,10 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht return err } + // Checked after the reads, so they all came from the bound bucket. + if err := h.requireBoundTableBucket(r.Context(), client, bucketName); err != nil { + return err + } var resourceARN string if rType == ResourceTypeTable { resourceARN = h.generateTableARN(ownerAccountID, bucketName, tableNamespace+"/"+tableName) @@ -1034,6 +1103,10 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req return err } + // Checked after the reads, so they all came from the bound bucket. + if err := h.requireBoundTableBucket(r.Context(), client, bucketName); err != nil { + return err + } var resourceARN string if rType == ResourceTypeTable { resourceARN = h.generateTableARN(ownerAccountID, bucketName, tableNamespace+"/"+tableName) @@ -1076,17 +1149,6 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req return err } - // Remove specified tags - for _, key := range req.TagKeys { - delete(tags, key) - } - - // Write updated tags - tagsBytes, err := json.Marshal(tags) - if err != nil { - 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 @@ -1096,8 +1158,11 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req if err := sameResource(extended, resourcePath, createdAt); err != nil { return err } - extended[extendedKey] = tagsBytes - return nil + tags, err := updateTags(extended[extendedKey], nil, req.TagKeys) + if err != nil { + return err + } + return writeTags(extended, extendedKey, tags) }) }) diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index e3be966476..0572d74a36 100644 --- a/weed/s3api/s3tables/manager.go +++ b/weed/s3api/s3tables/manager.go @@ -80,10 +80,10 @@ func (m *Manager) PolicyContext(r *http.Request, bucket string) *PolicyContext { type tableBucketCreatedAtKey struct{} // WithTableBucketCreatedAt binds a DeleteTableBucket, CreateNamespace, -// DeleteNamespace, 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. +// DeleteNamespace, CreateTable, DeleteTable or tag operation 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) } diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 9f9be68b51..83018dca6b 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -39,6 +39,9 @@ var handlers = map[string]func(*Server, http.ResponseWriter, *http.Request){ "ListVectors": (*Server).handleListVectors, "DeleteVectors": (*Server).handleDeleteVectors, "QueryVectors": (*Server).handleQueryVectors, + "TagResource": (*Server).handleTagResource, + "UntagResource": (*Server).handleUntagResource, + "ListTagsForResource": (*Server).handleListTagsForResource, } // lance table ids put indexes under a "vector" namespace inside their bucket. @@ -275,6 +278,10 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) return } + if err := s3tables.ValidateTagsWithLimit(req.Tags, maxResourceTags); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } // The catalog keeps the s3tables field name; GetVectorBucket renames it back. enc := req.EncryptionConfiguration if len(enc) != 0 { @@ -283,7 +290,12 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request // No bucket exists yet, so check against the handler account as the // catalog does for CreateTableBucket. principal := s.tables.PrincipalFor(r) - if !s3tables.CheckPermissionWithContext("s3vectors:CreateVectorBucket", principal, s.tables.PrincipalFor(nil), "", "", s.tables.PolicyContext(r, "")) { + allowed := func(action, bucket string) bool { + return s3tables.CheckPermissionWithContext(action, principal, s.tables.PrincipalFor(nil), "", "", s.tables.PolicyContext(r, bucket)) + } + // Tagging names the new bucket, so a grant scoped to it applies. + if !allowed("s3vectors:CreateVectorBucket", "") || + (len(req.Tags) > 0 && !allowed("s3vectors:TagResource", req.VectorBucketName)) { catalogErr(w, errVectorDenied) return } @@ -293,6 +305,7 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request Name: req.VectorBucketName, Format: s3tables.FormatLance, EncryptionConfiguration: enc, + Tags: req.Tags, }, &resp, identityOf(r)) if err != nil { catalogErr(w, err) @@ -963,6 +976,10 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "ValidationException", "dimension must be 1..=4096") return } + if err := s3tables.ValidateTagsWithLimit(req.Tags, maxResourceTags); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } // The create grant is exercised before any write: a denied caller must not // leave even the implicit namespace behind. bucketRecord, err := s.authorizeVector(r.Context(), r, "s3vectors:CreateIndex", arns.bucket, arns.index) @@ -970,6 +987,12 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } + if len(req.Tags) > 0 { + if _, err := s.authorizeVector(r.Context(), r, "s3vectors:TagResource", arns.bucket, arns.index); 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. @@ -1014,6 +1037,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { Name: arns.index, Format: s3tables.FormatLance, Metadata: &s3tables.TableMetadata{FullMetadata: metadataJSON}, + Tags: req.Tags, }, nil) if err != nil { catalogErr(w, err) diff --git a/weed/s3api/s3vectors/server.go b/weed/s3api/s3vectors/server.go index c2bb46a607..b901881eaf 100644 --- a/weed/s3api/s3vectors/server.go +++ b/weed/s3api/s3vectors/server.go @@ -69,17 +69,24 @@ func (s *Server) SetAccountID(id string) { s.accountID = id } // apply the AES256 a bucket declares. func (s *Server) SetEncryptionChecker(fn func() bool) { s.internal.SetEncryptionChecker(fn) } -// RegisterRoutes mounts POST / dispatching on the X-Amz-Target header and -// POST /{Operation} for clients that route per-path. Metrics wrap Auth like the -// S3 track middleware so rejected requests still count toward the operation. +// RegisterRoutes mounts POST / dispatching on the X-Amz-Target header, +// POST /{Operation} for clients that route per-path, and the tagging calls on +// /tags/{resourceArn}. Metrics wrap Auth like the S3 track middleware so +// rejected requests still count toward the operation. func (s *Server) RegisterRoutes(router *mux.Router) { - router.HandleFunc("/", s.instrument(s.Auth(s.dispatch))).Methods(http.MethodPost) - router.HandleFunc("/{op}", s.instrument(s.Auth(s.dispatch))).Methods(http.MethodPost) + handler := s.instrument(s.Auth(s.dispatch)) + router.HandleFunc("/", handler).Methods(http.MethodPost) + router.HandleFunc(tagsPathPrefix+"{resourceArn:.+}", handler).Methods(http.MethodGet, http.MethodPost, http.MethodDelete) + router.HandleFunc("/{op}", handler).Methods(http.MethodPost) } -// operationFromRequest resolves the operation the way dispatch does: the last -// path segment, or the X-Amz-Target suffix when the path is bare. +// operationFromRequest resolves the operation the way dispatch does: the +// method on a /tags/ path, else the last path segment, or the X-Amz-Target +// suffix when the path is bare. func operationFromRequest(r *http.Request) string { + if strings.HasPrefix(r.URL.Path, tagsPathPrefix) { + return tagOperation(r.Method) + } op := strings.Trim(r.URL.Path, "/") if i := strings.LastIndex(op, "/"); i >= 0 { op = op[i+1:] diff --git a/weed/s3api/s3vectors/tagging.go b/weed/s3api/s3vectors/tagging.go new file mode 100644 index 0000000000..c7370a67f1 --- /dev/null +++ b/weed/s3api/s3vectors/tagging.go @@ -0,0 +1,191 @@ +package s3vectors + +import ( + "context" + "fmt" + "net/http" + "strings" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// maxResourceTags is how many tags a vector bucket or index may carry. +const maxResourceTags = 50 + +const tagsPathPrefix = "/tags/" + +// tagOperation names the tagging call a /tags/{resourceArn} request makes. +func tagOperation(method string) string { + switch method { + case http.MethodGet: + return "ListTagsForResource" + case http.MethodDelete: + return "UntagResource" + } + return "TagResource" +} + +// readTagRequest decodes the body of a POST; GET and DELETE carry everything +// in the path and query. +func readTagRequest(r *http.Request, req interface{}) error { + if r.Method != http.MethodPost { + return nil + } + return readJSON(r, req) +} + +// tagTarget resolves a tagging call's resourceArn, taken from the path when +// the request was routed there, to the vector bucket or index and the ARN the +// catalog keys that record by. +func (s *Server) tagTarget(r *http.Request, bodyArn string) (vectorBucketARNs, string, error) { + arn := bodyArn + if v := mux.Vars(r)["resourceArn"]; v != "" { + arn = v + } + var v vectorBucketARNs + if arn == "" { + return v, "", fmt.Errorf("resourceArn is required") + } + if !strings.HasPrefix(arn, "arn:") { + return v, "", fmt.Errorf("resourceArn must be an ARN, got %q", arn) + } + if err := s.arnParts(&v, "", arn); err != nil { + return v, "", err + } + catalogArn := s.s3tablesArn(v.bucket) + if v.index != "" { + catalogArn = fmt.Sprintf("%s/table/%s/%s", catalogArn, indexNamespace, v.index) + } + return v, catalogArn, nil +} + +// authorizeTagging grants action on the tagged bucket or index and returns a +// context binding the catalog calls to the bucket it authorized. An index +// target must be a vector index, not another table in its namespace. +func (s *Server) authorizeTagging(r *http.Request, action string, arns vectorBucketARNs) (context.Context, error) { + record, err := s.authorizeVector(r.Context(), r, action, arns.bucket, arns.index) + if err != nil { + return nil, err + } + ctx := s3tables.WithTableBucketCreatedAt(r.Context(), record.CreatedAt) + if arns.index != "" { + index, err := s.indexRecord(ctx, arns.bucket, arns.index) + if err != nil { + return nil, err + } + if format, _ := index["format"].(string); format != s3tables.FormatLance { + return nil, &s3tables.S3TablesError{Type: s3tables.ErrCodeInvalidRequest, + Message: fmt.Sprintf("index %q is not a vector index", arns.index)} + } + } + return ctx, nil +} + +func (s *Server) resourceTags(ctx context.Context, catalogArn string) (map[string]string, error) { + var resp s3tables.ListTagsForResourceResponse + if err := s.internalCatalog(ctx, "ListTagsForResource", + &s3tables.ListTagsForResourceRequest{ResourceARN: catalogArn}, &resp); err != nil { + return nil, err + } + if resp.Tags == nil { + resp.Tags = map[string]string{} + } + return resp.Tags, nil +} + +func (s *Server) handleTagResource(w http.ResponseWriter, r *http.Request) { + var req TagResourceRequest + if err := readTagRequest(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + arns, catalogArn, err := s.tagTarget(r, req.ResourceArn) + if err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + if len(req.Tags) == 0 { + writeError(w, http.StatusBadRequest, "ValidationException", "tags are required") + return + } + if err := s3tables.ValidateTagsWithLimit(req.Tags, maxResourceTags); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + ctx, err := s.authorizeTagging(r, "s3vectors:TagResource", arns) + if err != nil { + catalogErr(w, err) + return + } + if err := s.internalCatalog(s3tables.WithMaxTags(ctx, maxResourceTags), "TagResource", + &s3tables.TagResourceRequest{ResourceARN: catalogArn, Tags: req.Tags}, nil); err != nil { + catalogErr(w, err) + return + } + writeJSON(w, http.StatusOK, &TagResourceResponse{}) +} + +func (s *Server) handleUntagResource(w http.ResponseWriter, r *http.Request) { + var req UntagResourceRequest + if err := readTagRequest(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + if keys, ok := r.URL.Query()["tagKeys"]; ok { + req.TagKeys = keys + } + arns, catalogArn, err := s.tagTarget(r, req.ResourceArn) + if err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + if len(req.TagKeys) == 0 { + writeError(w, http.StatusBadRequest, "ValidationException", "tagKeys are required") + return + } + ctx, err := s.authorizeTagging(r, "s3vectors:UntagResource", arns) + if err != nil { + catalogErr(w, err) + return + } + key := make(map[string]string, 1) + for _, k := range req.TagKeys { + clear(key) + key[k] = "" + if err := s3tables.ValidateTagsWithLimit(key, 1); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + } + if err := s.internalCatalog(ctx, "UntagResource", + &s3tables.UntagResourceRequest{ResourceARN: catalogArn, TagKeys: req.TagKeys}, nil); err != nil { + catalogErr(w, err) + return + } + writeJSON(w, http.StatusOK, &UntagResourceResponse{}) +} + +func (s *Server) handleListTagsForResource(w http.ResponseWriter, r *http.Request) { + var req ListTagsForResourceRequest + if err := readTagRequest(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + arns, catalogArn, err := s.tagTarget(r, req.ResourceArn) + if err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", err.Error()) + return + } + ctx, err := s.authorizeTagging(r, "s3vectors:ListTagsForResource", arns) + if err != nil { + catalogErr(w, err) + return + } + tags, err := s.resourceTags(ctx, catalogArn) + if err != nil { + catalogErr(w, err) + return + } + writeJSON(w, http.StatusOK, &ListTagsForResourceResponse{Tags: tags}) +} diff --git a/weed/s3api/s3vectors/tagging_test.go b/weed/s3api/s3vectors/tagging_test.go new file mode 100644 index 0000000000..d68eec4896 --- /dev/null +++ b/weed/s3api/s3vectors/tagging_test.go @@ -0,0 +1,355 @@ +package s3vectors + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "net/url" + "reflect" + "runtime" + "strings" + "testing" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/s3api" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +type tagClient struct { + t *testing.T + ts *httptest.Server + fs *s3tablestest.MemFiler +} + +func newTagClient(t *testing.T) (*Server, *tagClient) { + fs := s3tablestest.Start(t) + s := NewServer(s3tables.NewManagerClient(fs.Client), headerAuth{}, newFakeVectors(0)) + router := mux.NewRouter().SkipClean(true) + s.RegisterRoutes(router) + ts := httptest.NewServer(router) + t.Cleanup(ts.Close) + return s, &tagClient{t: t, ts: ts, fs: fs} +} + +func (c *tagClient) do(identity, method, path string, header http.Header, payload interface{}) (int, string) { + c.t.Helper() + var body io.Reader + if payload != nil { + raw, err := json.Marshal(payload) + if err != nil { + c.t.Fatalf("marshal: %v", err) + } + body = bytes.NewReader(raw) + } + req, err := http.NewRequest(method, c.ts.URL+path, body) + if err != nil { + c.t.Fatalf("build %s %s: %v", method, path, err) + } + for k, v := range header { + req.Header[k] = v + } + if identity != "" { + req.Header.Set("X-Test-Identity", identity) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + c.t.Fatalf("%s %s: %v", method, path, err) + } + defer resp.Body.Close() + out, _ := io.ReadAll(resp.Body) + return resp.StatusCode, string(out) +} + +func (c *tagClient) op(identity, op string, payload interface{}) (int, string) { + return c.do(identity, http.MethodPost, "/"+op, nil, payload) +} + +func (c *tagClient) tag(identity, arn string, tags map[string]string) (int, string) { + return c.do(identity, http.MethodPost, tagsPathPrefix+url.PathEscape(arn), nil, map[string]interface{}{"tags": tags}) +} + +func (c *tagClient) untag(identity, arn string, keys ...string) (int, string) { + q := url.Values{"tagKeys": keys} + return c.do(identity, http.MethodDelete, tagsPathPrefix+url.PathEscape(arn)+"?"+q.Encode(), nil, nil) +} + +func (c *tagClient) list(identity, arn string) (int, map[string]string) { + c.t.Helper() + code, body := c.do(identity, http.MethodGet, tagsPathPrefix+url.PathEscape(arn), nil, nil) + if code != http.StatusOK { + return code, nil + } + var resp struct { + Tags map[string]string `json:"tags"` + } + if err := json.Unmarshal([]byte(body), &resp); err != nil || resp.Tags == nil { + c.t.Fatalf("decode ListTagsForResource %q: %v", body, err) + } + return code, resp.Tags +} + +func mustStatus(t *testing.T, what string, code int, body string, want int) { + t.Helper() + if code != want { + t.Fatalf("%s = %d %s, want %d", what, code, body, want) + } +} + +func TestTagVectorBucketsAndIndexes(t *testing.T) { + s, c := newTagClient(t) + code, body := c.op("", "CreateVectorBucket", &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}) + mustStatus(t, "create bucket", code, body, http.StatusOK) + code, body = c.op("", "CreateIndex", &CreateIndexRequest{ + VectorBucketName: "vec-bucket", IndexName: "idx", DataType: "float32", Dimension: 3, + }) + mustStatus(t, "create index", code, body, http.StatusOK) + bucketArn, indexArn := s.vectorBucketArn("vec-bucket"), s.indexArn("vec-bucket", "idx") + + if _, tags := c.list("", bucketArn); len(tags) != 0 { + t.Fatalf("new bucket tags = %v, want none", tags) + } + code, body = c.tag("", bucketArn, map[string]string{"team": "search", "env": "dev"}) + mustStatus(t, "tag bucket", code, body, http.StatusOK) + code, body = c.tag("", indexArn, map[string]string{"team": "ranking"}) + mustStatus(t, "tag index", code, body, http.StatusOK) + code, body = c.tag("", bucketArn, map[string]string{"env": "prod"}) + mustStatus(t, "retag bucket", code, body, http.StatusOK) + + if _, tags := c.list("", bucketArn); !reflect.DeepEqual(tags, map[string]string{"team": "search", "env": "prod"}) { + t.Fatalf("bucket tags = %v", tags) + } + if _, tags := c.list("", indexArn); !reflect.DeepEqual(tags, map[string]string{"team": "ranking"}) { + t.Fatalf("index tags = %v", tags) + } + + code, body = c.untag("", bucketArn, "env", "absent") + mustStatus(t, "untag bucket", code, body, http.StatusOK) + if _, tags := c.list("", bucketArn); !reflect.DeepEqual(tags, map[string]string{"team": "search"}) { + t.Fatalf("bucket tags after untag = %v", tags) + } + + // Clients that post the operation by name carry resourceArn in the body. + code, body = c.do("", http.MethodPost, "/", http.Header{"X-Amz-Target": {"AmazonS3Vectors.ListTagsForResource"}}, + &ListTagsForResourceRequest{ResourceArn: indexArn}) + mustStatus(t, "list by target header", code, body, http.StatusOK) + var resp ListTagsForResourceResponse + if err := json.Unmarshal([]byte(body), &resp); err != nil || resp.Tags["team"] != "ranking" { + t.Fatalf("list by target header = %s (%v)", body, err) + } +} + +func TestTagResourceValidation(t *testing.T) { + s, c := newTagClient(t) + code, body := c.op("", "CreateVectorBucket", &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}) + mustStatus(t, "create bucket", code, body, http.StatusOK) + bucketArn := s.vectorBucketArn("vec-bucket") + + code, body = c.tag("", s.indexArn("vec-bucket", "missing"), map[string]string{"a": "b"}) + mustStatus(t, "tag missing index", code, body, http.StatusNotFound) + code, _ = c.list("", s.vectorBucketArn("no-such-bucket")) + mustStatus(t, "list missing bucket", code, "", http.StatusNotFound) + code, body = c.tag("", "vec-bucket", map[string]string{"a": "b"}) + mustStatus(t, "tag by name", code, body, http.StatusBadRequest) + code, body = c.tag("", "arn:aws:s3vectors:elsewhere:1:bucket/vec-bucket", map[string]string{"a": "b"}) + mustStatus(t, "tag foreign ARN", code, body, http.StatusBadRequest) + code, body = c.tag("", bucketArn, map[string]string{}) + mustStatus(t, "tag with no tags", code, body, http.StatusBadRequest) + code, body = c.tag("", bucketArn, map[string]string{"bad key!": "v"}) + mustStatus(t, "tag with bad key", code, body, http.StatusBadRequest) + code, body = c.untag("", bucketArn) + mustStatus(t, "untag with no keys", code, body, http.StatusBadRequest) + code, body = c.untag("", bucketArn, "") + mustStatus(t, "untag an empty key", code, body, http.StatusBadRequest) + code, body = c.untag("", bucketArn, strings.Repeat("k", 129)) + mustStatus(t, "untag a key past 128 characters", code, body, http.StatusBadRequest) + + tags := map[string]string{} + for i := 0; i < maxResourceTags; i++ { + tags[fmt.Sprintf("k%d", i)] = "v" + } + code, body = c.tag("", bucketArn, tags) + mustStatus(t, "tag up to the limit", code, body, http.StatusOK) + code, body = c.tag("", bucketArn, map[string]string{"k0": "changed"}) + mustStatus(t, "overwrite at the limit", code, body, http.StatusOK) + code, body = c.tag("", bucketArn, map[string]string{"one-more": "v"}) + mustStatus(t, "tag past the limit", code, body, http.StatusBadRequest) + tags["one-more"] = "v" + code, body = c.tag("", s.vectorBucketArn("vec-bucket"), tags) + mustStatus(t, "request past the limit", code, body, http.StatusBadRequest) + if _, got := c.list("", bucketArn); len(got) != maxResourceTags || got["k0"] != "changed" { + t.Fatalf("tags after refused requests: %d tags, k0=%q", len(got), got["k0"]) + } +} + +func TestTaggingNeedsItsOwnGrant(t *testing.T) { + s, c := newTagClient(t) + code, body := c.op("", "CreateVectorBucket", &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}) + mustStatus(t, "create bucket", code, body, http.StatusOK) + bucketArn := s.vectorBucketArn("vec-bucket") + code, body = c.tag("", bucketArn, map[string]string{"team": "search"}) + mustStatus(t, "owner tags", code, body, http.StatusOK) + + code, _ = c.list("reader", bucketArn) + mustStatus(t, "list without a grant", code, "", http.StatusForbidden) + code, body = c.op("", "PutVectorBucketPolicy", &PutVectorBucketPolicyRequest{ + VectorBucketName: "vec-bucket", + Policy: `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*",` + + `"Action":["s3vectors:ListTagsForResource","s3vectors:GetVectorBucket"],"Resource":"` + bucketArn + `"}]}`, + }) + mustStatus(t, "put policy", code, body, http.StatusOK) + if code, tags := c.list("reader", bucketArn); code != http.StatusOK || tags["team"] != "search" { + t.Fatalf("list under grant = %d %v", code, tags) + } + code, body = c.tag("reader", bucketArn, map[string]string{"team": "other"}) + mustStatus(t, "tag under list grant", code, body, http.StatusForbidden) + code, body = c.untag("reader", bucketArn, "team") + mustStatus(t, "untag under list grant", code, body, http.StatusForbidden) + code, body = c.untag("reader", bucketArn, "") + mustStatus(t, "untag an empty key under list grant", code, body, http.StatusForbidden) + if _, tags := c.list("", bucketArn); tags["team"] != "search" { + t.Fatalf("refused writes changed tags: %v", tags) + } +} + +func TestCreateWithTagsNeedsTagResource(t *testing.T) { + e := newAuthTestEnv(t) + tags := map[string]string{"team": "search"} + listed := func(arn string) map[string]string { + t.Helper() + w := e.vectorResp(adminIdentity(), e.s.handleListTagsForResource, &ListTagsForResourceRequest{ResourceArn: arn}) + var resp ListTagsForResourceResponse + if w.Code != http.StatusOK || json.Unmarshal(w.Body.Bytes(), &resp) != nil { + t.Fatalf("list tags of %s: %d %s", arn, w.Code, w.Body.String()) + } + return resp.Tags + } + creator := func(actions ...s3api.Action) *s3api.Identity { + return &s3api.Identity{Name: "creator", Account: &s3api.Account{Id: "creator"}, Actions: actions} + } + createOnly := creator("s3vectors:CreateVectorBucket", "s3vectors:CreateIndex") + withTag := creator("s3vectors:CreateVectorBucket", "s3vectors:CreateIndex", "s3vectors:TagResource") + + if code := e.vector(createOnly, e.s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "denied", Tags: tags}); code != http.StatusForbidden { + t.Fatalf("tagged bucket create without TagResource: %d, want 403", code) + } + if code := e.vector(adminIdentity(), e.s.handleGetVectorBucket, &GetVectorBucketRequest{VectorBucketName: "denied"}); code != http.StatusNotFound { + t.Fatalf("refused create left a bucket: %d", code) + } + if code := e.vector(withTag, e.s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket", Tags: tags}); code != http.StatusOK { + t.Fatalf("tagged bucket create: %d", code) + } + if got := listed(e.s.vectorBucketArn("vec-bucket")); !reflect.DeepEqual(got, tags) { + t.Fatalf("bucket tags = %v, want %v", got, tags) + } + + e.createBuckets(adminIdentity(), "shared") + index := func(ident *s3api.Identity, name string, tags map[string]string) int { + return e.vector(ident, e.s.handleCreateIndex, &CreateIndexRequest{ + VectorBucketName: "v-shared", IndexName: name, DataType: "float32", Dimension: 3, Tags: tags, + }) + } + if code := index(createOnly, "denied", tags); code != http.StatusForbidden { + t.Fatalf("tagged index create without TagResource: %d, want 403", code) + } + if code := index(createOnly, "plain", nil); code != http.StatusOK { + t.Fatalf("untagged index create: %d", code) + } + if code := index(withTag, "tagged", tags); code != http.StatusOK { + t.Fatalf("tagged index create: %d", code) + } + if got := listed(e.s.indexArn("v-shared", "tagged")); !reflect.DeepEqual(got, tags) { + t.Fatalf("index tags = %v, want %v", got, tags) + } + if got := listed(e.s.indexArn("v-shared", "plain")); len(got) != 0 { + t.Fatalf("untagged index tags = %v", got) + } + + scoped := creator("s3vectors:CreateVectorBucket", "s3vectors:TagResource:team-bucket") + if code := e.vector(scoped, e.s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "other-bucket", Tags: tags}); code != http.StatusForbidden { + t.Fatalf("tagged create outside the scoped TagResource grant: %d, want 403", code) + } + if code := e.vector(scoped, e.s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "team-bucket", Tags: tags}); code != http.StatusOK { + t.Fatalf("tagged create under a TagResource grant scoped to the bucket: %d", code) + } + + tooMany := map[string]string{} + for i := 0; i <= maxResourceTags; i++ { + tooMany[fmt.Sprintf("k%d", i)] = "v" + } + if code := e.vector(withTag, e.s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "too-many", Tags: tooMany}); code != http.StatusBadRequest { + t.Fatalf("bucket create past the tag limit: %d, want 400", code) + } + if code := index(withTag, "too-many", tooMany); code != http.StatusBadRequest { + t.Fatalf("index create past the tag limit: %d, want 400", code) + } +} + +func TestTaggingRefusesTablesThatAreNotIndexes(t *testing.T) { + s, c := newTagClient(t) + code, body := c.op("", "CreateVectorBucket", &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}) + mustStatus(t, "create bucket", code, body, http.StatusOK) + code, body = c.op("", "CreateIndex", &CreateIndexRequest{ + VectorBucketName: "vec-bucket", IndexName: "idx", DataType: "float32", Dimension: 3, + }) + mustStatus(t, "create index", code, body, http.StatusOK) + indexArn := s.indexArn("vec-bucket", "idx") + code, body = c.tag("", indexArn, map[string]string{"team": "search"}) + mustStatus(t, "tag index", code, body, http.StatusOK) + + dir, name := splitEntryPath(s3tables.GetTablePath("vec-bucket", indexNamespace, "idx")) + entry := c.fs.Get(dir, name) + var meta map[string]interface{} + if err := json.Unmarshal(entry.Extended[s3tables.ExtendedKeyMetadata], &meta); err != nil { + t.Fatalf("decode index metadata: %v", err) + } + meta["format"] = s3tables.FormatIceberg + data, _ := json.Marshal(meta) + entry.Extended[s3tables.ExtendedKeyMetadata] = data + c.fs.Put(dir, name, entry.Extended) + + code, _ = c.list("", indexArn) + mustStatus(t, "list tags of an Iceberg table", code, "", http.StatusBadRequest) + code, body = c.tag("", indexArn, map[string]string{"team": "other"}) + mustStatus(t, "tag an Iceberg table", code, body, http.StatusBadRequest) + code, body = c.untag("", indexArn, "team") + mustStatus(t, "untag an Iceberg table", code, body, http.StatusBadRequest) + if got := string(c.fs.Get(dir, name).Extended[s3tables.ExtendedKeyTags]); got != `{"team":"search"}` { + t.Fatalf("refused calls changed the table's tags: %s", got) + } +} + +func TestUntagManyKeysAllocatesLikeDecoding(t *testing.T) { + s, c := newTagClient(t) + code, out := c.op("", "CreateVectorBucket", &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}) + mustStatus(t, "create bucket", code, out, http.StatusOK) + body, err := json.Marshal(&UntagResourceRequest{ResourceArn: s.vectorBucketArn("vec-bucket"), TagKeys: make([]string, 1<<18)}) + if err != nil { + t.Fatal(err) + } + allocated := func(f func()) uint64 { + var before, after runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&before) + f() + runtime.ReadMemStats(&after) + return after.TotalAlloc - before.TotalAlloc + } + decode := allocated(func() { + var req UntagResourceRequest + if err := json.Unmarshal(body, &req); err != nil { + t.Fatal(err) + } + }) + w := httptest.NewRecorder() + handle := allocated(func() { + s.handleUntagResource(w, httptest.NewRequest(http.MethodPost, "/UntagResource", bytes.NewReader(body))) + }) + mustStatus(t, "untag empty keys", w.Code, w.Body.String(), http.StatusBadRequest) + if handle > decode*3/2 { + t.Fatalf("untag allocated %d bytes, decoding the request takes %d", handle, decode) + } +} diff --git a/weed/s3api/s3vectors/types.go b/weed/s3api/s3vectors/types.go index 0bbbaff6f6..4f5f1496ec 100644 --- a/weed/s3api/s3vectors/types.go +++ b/weed/s3api/s3vectors/types.go @@ -62,7 +62,7 @@ type vectorBucketARNs struct { type CreateVectorBucketRequest struct { VectorBucketName string `json:"vectorBucketName"` EncryptionConfiguration map[string]interface{} `json:"encryptionConfiguration,omitempty"` - MetadataResourceTags map[string]string `json:"-"` + Tags map[string]string `json:"tags,omitempty"` } type CreateVectorBucketResponse struct { @@ -136,6 +136,7 @@ type CreateIndexRequest struct { MetadataConfiguration *MetadataConfiguration `json:"metadataConfiguration,omitempty"` VectorBucketName string `json:"vectorBucketName"` VectorBucketArn string `json:"vectorBucketArn"` + Tags map[string]string `json:"tags,omitempty"` } type MetadataConfiguration struct { @@ -295,3 +296,25 @@ type QueryVectorsResponse struct { DistanceMetric string `json:"distanceMetric"` Vectors []*QueryOutputVector `json:"vectors"` } + +type TagResourceRequest struct { + ResourceArn string `json:"resourceArn"` + Tags map[string]string `json:"tags"` +} + +type TagResourceResponse struct{} + +type UntagResourceRequest struct { + ResourceArn string `json:"resourceArn"` + TagKeys []string `json:"tagKeys"` +} + +type UntagResourceResponse struct{} + +type ListTagsForResourceRequest struct { + ResourceArn string `json:"resourceArn"` +} + +type ListTagsForResourceResponse struct { + Tags map[string]string `json:"tags"` +}