mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-01 09:50:39 +01:00
s3vectors: authorize CreateVectorBucket as a vector action, close DeleteIndex race, fold policy service case (#12287)
* s3vectors: authorize CreateVectorBucket as an s3vectors action CreateVectorBucket went through the catalog's CreateTableBucket check, so an identity granted s3vectors:* was denied the create while s3tables:* was allowed it. Check s3vectors:CreateVectorBucket in the gateway, the same way the catalog checks a bucket create, and create the record on the internal manager under the caller's name so the caller stays the owner. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3vectors: delete only the index record DeleteIndex checked DeleteIndex read the record, checked it is a Lance index, then deleted by name. A table that replaced the index in between was deleted too. Send the checked record's version token, so a replaced record fails the delete with a 409 ConflictException. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3tables: reject s3vectors rules in any letter case validatePolicyDocument kept s3vectors actions and resources out of the shared policy slot with case-sensitive checks, so "S3Vectors:GetIndex" was stored as a rule that never takes effect. IAM service prefixes are case-insensitive; fold case in both checks. 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
e7f5c6873a
commit
21fc46692c
4 files changed
+147
-9
No files matched your search
@@ -152,12 +152,12 @@ func validatePolicyDocument(raw string) error {
|
||||
}
|
||||
for _, stmt := range doc.Statement {
|
||||
for _, action := range statementActions(stmt.Action) {
|
||||
if strings.HasPrefix(action, "s3vectors:") {
|
||||
if strings.HasPrefix(strings.ToLower(action), "s3vectors:") {
|
||||
return fmt.Errorf("action %q is not enforceable in a resource policy", action)
|
||||
}
|
||||
}
|
||||
for _, resource := range statementActions(stmt.Resource) {
|
||||
if arnService(resource) == "s3vectors" {
|
||||
if strings.EqualFold(arnService(resource), "s3vectors") {
|
||||
return fmt.Errorf("resource %q is not enforceable in a resource policy", resource)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
package s3tables
|
||||
|
||||
import "testing"
|
||||
|
||||
// Service prefixes are case-insensitive in IAM, so a vector rule in any
|
||||
// letter case is kept out of the shared policy slot.
|
||||
func TestValidatePolicyDocumentRejectsVectorServiceInAnyCase(t *testing.T) {
|
||||
for _, doc := range []string{
|
||||
`{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":"S3Vectors:GetIndex","Resource":"*"}]}`,
|
||||
`{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":["s3tables:*","S3VECTORS:*"],"Resource":"*"}]}`,
|
||||
`{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":"s3tables:*","Resource":"arn:aws:S3Vectors:r:a:bucket/b"}]}`,
|
||||
} {
|
||||
if err := validatePolicyDocument(doc); err == nil {
|
||||
t.Errorf("vector rule accepted into the shared slot: %s", doc)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
package s3vectors
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"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"
|
||||
)
|
||||
|
||||
// postJSONAsIdentity runs the handler with the identity object in the context,
|
||||
// the way Auth leaves it.
|
||||
func postJSONAsIdentity(t *testing.T, id *s3api.Identity, handler http.HandlerFunc, payload interface{}) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal request: %v", err)
|
||||
}
|
||||
w := httptest.NewRecorder()
|
||||
r := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body))
|
||||
ctx := s3_constants.SetIdentityInContext(s3_constants.SetIdentityNameInContext(r.Context(), id.Name), id)
|
||||
handler(w, r.WithContext(ctx))
|
||||
return w
|
||||
}
|
||||
|
||||
// CreateVectorBucket is authorized as an s3vectors action, and the creator
|
||||
// still owns the bucket.
|
||||
func TestCreateVectorBucketChecksVectorAction(t *testing.T) {
|
||||
fs := s3tablestest.Start(t)
|
||||
s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0))
|
||||
create := &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}
|
||||
|
||||
tablesOnly := &s3api.Identity{Name: "bob", Actions: []s3api.Action{"s3tables:*"}}
|
||||
if w := postJSONAsIdentity(t, tablesOnly, s.handleCreateVectorBucket, create); w.Code != http.StatusForbidden {
|
||||
t.Fatalf("create with s3tables:* = %d %s, want 403", w.Code, w.Body.String())
|
||||
}
|
||||
vectors := &s3api.Identity{Name: "alice", Actions: []s3api.Action{"s3vectors:*"}}
|
||||
if w := postJSONAsIdentity(t, vectors, s.handleCreateVectorBucket, create); w.Code != http.StatusOK {
|
||||
t.Fatalf("create with s3vectors:* = %d %s, want 200", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
record, err := s.vectorBucket(context.Background(), "vec-bucket")
|
||||
if err != nil {
|
||||
t.Fatalf("read bucket record: %v", err)
|
||||
}
|
||||
if record.OwnerAccountID != "alice" {
|
||||
t.Fatalf("owner = %q, want alice", record.OwnerAccountID)
|
||||
}
|
||||
get := &GetVectorBucketRequest{VectorBucketName: "vec-bucket"}
|
||||
if w := postJSONAsIdentity(t, &s3api.Identity{Name: "alice"}, s.handleGetVectorBucket, get); w.Code != http.StatusOK {
|
||||
t.Fatalf("owner get without actions = %d %s, want 200", w.Code, w.Body.String())
|
||||
}
|
||||
if w := postJSONAsIdentity(t, &s3api.Identity{Name: "carol"}, s.handleGetVectorBucket, get); w.Code != http.StatusForbidden {
|
||||
t.Fatalf("non-owner get without actions = %d %s, want 403", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// DeleteIndex deletes only the record its format check read: a table that
|
||||
// replaced the index in between survives, and the caller gets a conflict.
|
||||
func TestDeleteIndexKeepsReplacedRecord(t *testing.T) {
|
||||
fs := s3tablestest.Start(t)
|
||||
s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0))
|
||||
if w := postJSON(t, s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK {
|
||||
t.Fatalf("create bucket: %d %s", w.Code, w.Body.String())
|
||||
}
|
||||
if w := postJSON(t, s.handleCreateIndex, &CreateIndexRequest{
|
||||
VectorBucketName: "vec-bucket", IndexName: "idx", DataType: "float32", Dimension: 3, DistanceMetric: "euclidean",
|
||||
}); w.Code != http.StatusOK {
|
||||
t.Fatalf("create index: %d %s", w.Code, w.Body.String())
|
||||
}
|
||||
|
||||
dir, name := splitEntryPath(s3tables.GetTablePath("vec-bucket", indexNamespace, "idx"))
|
||||
lookups := 0
|
||||
fs.OnLookup = func(d, n string) {
|
||||
if d != dir || n != name {
|
||||
return
|
||||
}
|
||||
// The first lookup is the handler's format check; swap the record
|
||||
// for an Iceberg table right after it.
|
||||
if lookups++; lookups != 2 {
|
||||
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 index metadata: %v", err)
|
||||
return
|
||||
}
|
||||
meta["format"] = s3tables.FormatIceberg
|
||||
meta["versionToken"] = "replaced"
|
||||
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)
|
||||
}
|
||||
|
||||
w := postJSON(t, s.handleDeleteIndex, &DeleteIndexRequest{VectorBucketName: "vec-bucket", IndexName: "idx"})
|
||||
fs.OnLookup = nil
|
||||
if w.Code != http.StatusConflict || !strings.Contains(w.Body.String(), "ConflictException") {
|
||||
t.Fatalf("delete over a replaced record = %d %s, want 409 ConflictException", w.Code, w.Body.String())
|
||||
}
|
||||
if fs.Get(dir, name) == nil {
|
||||
t.Fatal("DeleteIndex removed the table that replaced the index")
|
||||
}
|
||||
}
|
||||
@@ -159,10 +159,6 @@ func (s *Server) indexRecord(ctx context.Context, bucket, index string) (map[str
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func (s *Server) catalog(ctx context.Context, operation string, req, resp interface{}, r *http.Request) error {
|
||||
return s.tables.Execute(ctx, s.filerClient, operation, req, resp, identityOf(r))
|
||||
}
|
||||
|
||||
// internalCatalog runs catalog bookkeeping under the trusted identity. It is
|
||||
// for reads and writes that accompany — never replace — the operation's own
|
||||
// permission check: bucket format lookups, the implicit index namespace, and
|
||||
@@ -265,13 +261,21 @@ func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request
|
||||
if len(enc) != 0 {
|
||||
enc = map[string]interface{}{"sseAlgorithm": enc["sseType"]}
|
||||
}
|
||||
// No bucket exists yet, so check against the handler account as the
|
||||
// catalog does for CreateTableBucket.
|
||||
principal := s.tables.PrincipalFor(identityOf(r))
|
||||
if !s3tables.CheckPermissionWithContext("s3vectors:CreateVectorBucket", principal, s.tables.PrincipalFor(""), "", "", s.tables.PolicyContext(r)) {
|
||||
catalogErr(w, errVectorDenied)
|
||||
return
|
||||
}
|
||||
// The caller's name still makes it the owner of the record.
|
||||
var resp map[string]interface{}
|
||||
err := s.catalog(r.Context(), "CreateTableBucket",
|
||||
err := s.internal.Execute(r.Context(), s.filerClient, "CreateTableBucket",
|
||||
&s3tables.CreateTableBucketRequest{
|
||||
Name: req.VectorBucketName,
|
||||
Format: s3tables.FormatLance,
|
||||
EncryptionConfiguration: enc,
|
||||
}, &resp, r)
|
||||
}, &resp, identityOf(r))
|
||||
if err != nil {
|
||||
catalogErr(w, err)
|
||||
return
|
||||
@@ -1015,7 +1019,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// recordNonce and recordVersionToken read the catalog record's internal
|
||||
// bookkeeping for the create-index rollback.
|
||||
// bookkeeping for the create-index rollback and DeleteIndex.
|
||||
func recordNonce(record map[string]interface{}) string {
|
||||
if meta, ok := record["metadata"].(map[string]interface{}); ok {
|
||||
if fm, ok := meta["fullMetadata"].(map[string]interface{}); ok {
|
||||
@@ -1191,10 +1195,12 @@ 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{
|
||||
TableBucketARN: s.s3tablesArn(arns.bucket),
|
||||
Namespace: []string{indexNamespace},
|
||||
Name: arns.index,
|
||||
VersionToken: recordVersionToken(record),
|
||||
}, &resp)
|
||||
if err != nil {
|
||||
catalogErr(w, err)
|
||||
|
||||
Reference in new issue
Block a user