diff --git a/weed/s3api/s3tables/filer_ops.go b/weed/s3api/s3tables/filer_ops.go
index aec770ccc9..273c7989d4 100644
--- a/weed/s3api/s3tables/filer_ops.go
+++ b/weed/s3api/s3tables/filer_ops.go
@@ -282,6 +282,7 @@ var s3tablesExtendedKeys = []string{
 	ExtendedKeyMaintenanceStatus,
 	ExtendedKeyEntryType,
 	ExtendedKeyDeleting,
+	ExtendedKeyVectorPolicy,
 }
 
 // SnapshotExtended captures an entry's attributes as an UpdateEntry
diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go
index 521fcb868a..4e33146e32 100644
--- a/weed/s3api/s3tables/handler.go
+++ b/weed/s3api/s3tables/handler.go
@@ -37,6 +37,12 @@ const (
 	// or releases it.
 	ExtendedKeyDeleting = s3_constants.ExtS3TablesPrefix + "deleting"
 
+	// ExtendedKeyVectorPolicy is written by the S3 Vectors service on table
+	// buckets. It lives here rather than in the vectors package so the catalog
+	// can assert it in its update preconditions: a whole-entry write that does
+	// not know the key could silently erase a concurrently saved vector policy.
+	ExtendedKeyVectorPolicy = "s3vectors-policy"
+
 	// Entry-type marker values for ExtendedKeyEntryType. Absent or "table" means
 	// a table; views are stored like tables but tagged "view".
 	EntryTypeTable = "table"
diff --git a/weed/s3api/s3tables/handler_principal_test.go b/weed/s3api/s3tables/handler_principal_test.go
index 28ac9c9ee8..93014a78c3 100644
--- a/weed/s3api/s3tables/handler_principal_test.go
+++ b/weed/s3api/s3tables/handler_principal_test.go
@@ -26,6 +26,19 @@ func TestGetAccountIDHeaderBranchResolvesCaller(t *testing.T) {
 	assert.Equal(t, "alice", h.getAccountID(req))
 }
 
+func TestPrincipalForMirrorsGetAccountID(t *testing.T) {
+	m := NewManager()
+	m.SetAccountID("000000000000")
+
+	assert.Equal(t, "000000000000", m.PrincipalFor(""))
+	assert.Equal(t, "alice", m.PrincipalFor("alice"))
+	assert.Equal(t, s3_constants.AccountAdminId, m.PrincipalFor(s3_constants.AccountAdminId),
+		"the name branch of getAccountID keeps the admin name")
+	assert.Equal(t, "000000000000", m.PrincipalFor("   "))
+	assert.Equal(t, "bob", m.PrincipalFor("arn:aws:iam::123456789012:user/bob"),
+		"ARN identities collapse to the trailing segment like getAccountID")
+}
+
 func TestCheckPermissionDenyPolicyBindsNonAdmin(t *testing.T) {
 	denyAll := `{"Statement":[{"Effect":"Deny","Principal":"*","Action":"s3tables:DeleteTableBucket"}]}`
 	assert.False(t, CheckPermissionWithContext("s3tables:DeleteTableBucket", "mallory", "owner123", denyAll, "arn:aws:s3tables:us-east-1:000000000000:bucket/victim",
diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go
index da75b5d65f..6986109993 100644
--- a/weed/s3api/s3tables/manager.go
+++ b/weed/s3api/s3tables/manager.go
@@ -46,6 +46,30 @@ func (m *Manager) SetTrusted(trusted bool) {
 	m.handler.SetTrusted(trusted)
 }
 
+// PrincipalFor maps a forwarded identity name onto the account the catalog
+// would charge the request to: a synthesized request resolves its identity
+// name through getAccountID's name branch — normalized like any principal —
+// and an empty name falls back to the handler's account ID.
+func (m *Manager) PrincipalFor(identity string) string {
+	if principal := normalizePrincipalID(identity); principal != "" {
+		return principal
+	}
+	return m.handler.accountID
+}
+
+// PolicyContext builds the evaluation context a Manager call's synthesized
+// request produces: the configured local-account predicate, the caller's own
+// identity actions so admin and action grants resolve on privileges rather
+// than the forwarded name, plus the open-by-default fallback only for a
+// trusted caller or no resolved identity.
+func (m *Manager) PolicyContext(r *http.Request) *PolicyContext {
+	ctx := &PolicyContext{IsLocalAccountID: m.handler.localAccountIDs, IdentityActions: getIdentityActions(r)}
+	if m.handler.trusted || (m.handler.defaultAllow && s3_constants.GetIdentityNameFromContext(r) == "") {
+		ctx.DefaultAllow = true
+	}
+	return ctx
+}
+
 // Execute runs an S3 Tables operation and decodes the response into resp (if provided).
 func (m *Manager) Execute(ctx context.Context, filerClient FilerClient, operation string, req interface{}, resp interface{}, identity string) error {
 	body, err := json.Marshal(req)
diff --git a/weed/s3api/s3tables/permissions.go b/weed/s3api/s3tables/permissions.go
index caf0f31c1b..2991bcfe18 100644
--- a/weed/s3api/s3tables/permissions.go
+++ b/weed/s3api/s3tables/permissions.go
@@ -109,7 +109,7 @@ func CheckPermissionWithContext(operation, principal, owner, resourcePolicy, res
 	}
 
 	// Admin always has permission.
-	if principal == s3_constants.AccountAdminId {
+	if principal == s3_constants.AccountAdminId && adminCaller(operation, ctx) {
 		return true
 	}
 
@@ -125,7 +125,7 @@ func CheckTableOrBucketPermission(operation, principal, resourceOwner, resourceP
 	if principal == "" {
 		return false
 	}
-	if principal == s3_constants.AccountAdminId {
+	if principal == s3_constants.AccountAdminId && adminCaller(operation, ctx) {
 		return true
 	}
 	if matchesDenyStatement(operation, principal, resourcePolicy, resourceARN, ctx) ||
@@ -184,10 +184,18 @@ func statementActions(spec interface{}) []string {
 // deniedByPolicy reports whether another applicable resource policy vetoes the
 // request. The admin account is exempt, matching CheckPermissionWithContext.
 func deniedByPolicy(operation, principal, resourcePolicy, resourceARN string, ctx *PolicyContext) bool {
-	return principal != s3_constants.AccountAdminId &&
+	return (principal != s3_constants.AccountAdminId || !adminCaller(operation, ctx)) &&
 		matchesDenyStatement(operation, principal, resourcePolicy, resourceARN, ctx)
 }
 
+// adminCaller reports whether the admin principal bypass applies. Trusted
+// transports carry no action list, and an identified caller holds it only
+// when the identity administers the operation's service — a principal that
+// merely normalizes to the admin id does not bypass policy denies.
+func adminCaller(operation string, ctx *PolicyContext) bool {
+	return ctx == nil || len(ctx.IdentityActions) == 0 || hasAdminActionFor(operation, ctx.IdentityActions)
+}
+
 // matchesDenyStatement reports whether the JSON policy carries a statement
 // matching this request with Effect Deny. Malformed policies return false;
 // the caller's normal evaluation still denies them.
@@ -205,7 +213,7 @@ func matchesDenyStatement(operation, principal, resourcePolicy, resourceARN stri
 	}
 	for _, stmt := range policy.Statement {
 		if stmt.Effect != "Deny" ||
-			!matchesCaller(stmt.Principal, principal, ctx) ||
+			!matchesDenyCaller(stmt.Principal, principal, ctx) ||
 			!matchesAction(stmt.Action, fullAction) ||
 			(resourceARN != "" && !matchesResource(stmt.Resource, resourceARN)) ||
 			!matchesConditions(stmt.Condition, ctx) {
@@ -308,7 +316,7 @@ func hasIdentityPermission(operation string, ctx *PolicyContext) bool {
 	if !strings.Contains(operation, ":") {
 		fullAction = "s3tables:" + operation
 	}
-	if hasAdminAction(ctx.IdentityActions) {
+	if hasAdminActionFor(operation, ctx.IdentityActions) {
 		return true
 	}
 	candidates := []string{operation, fullAction}
@@ -340,11 +348,40 @@ func hasAdminAction(actions []string) bool {
 	return false
 }
 
+// hasAdminActionFor reports whether the action list administers the
+// operation's own service: the global admin markers always apply, while a
+// service wildcard covers only that service — s3tables:* never reaches an
+// s3vectors operation.
+func hasAdminActionFor(operation string, actions []string) bool {
+	if !strings.Contains(operation, ":") {
+		operation = "s3tables:" + operation
+	}
+	serviceWildcard := operation[:strings.Index(operation, ":")] + ":*"
+	for _, action := range actions {
+		if action == "*" || action == s3_constants.ACTION_ADMIN || action == "s3:*" || action == serviceWildcard {
+			return true
+		}
+	}
+	return false
+}
+
 // matchesCaller matches a statement's Principal against the caller's scoped
 // principal or, for policies written before issuer scoping, a bare subject
 // that resolves to it under the deployment's single trusted issuer.
 func matchesCaller(principalSpec interface{}, principal string, ctx *PolicyContext) bool {
-	if matchesPrincipal(principalSpec, principal) {
+	return matchCaller(principalSpec, principal, ctx, false)
+}
+
+// matchesDenyCaller is matchesCaller for deny evaluation: an ARN-scoped
+// statement also reaches a caller stored under the ARN's trailing name, so a
+// deny cannot be ducked through normalization the way an allow must not grant
+// through it.
+func matchesDenyCaller(principalSpec interface{}, principal string, ctx *PolicyContext) bool {
+	return matchCaller(principalSpec, principal, ctx, true)
+}
+
+func matchCaller(principalSpec interface{}, principal string, ctx *PolicyContext, forDeny bool) bool {
+	if matchPrincipal(principalSpec, principal, forDeny) {
 		return true
 	}
 	switch p := principalSpec.(type) {
@@ -358,7 +395,7 @@ func matchesCaller(principalSpec interface{}, principal string, ctx *PolicyConte
 		}
 	case map[string]interface{}:
 		if aws, ok := p["AWS"]; ok {
-			return matchesCaller(aws, principal, ctx)
+			return matchCaller(aws, principal, ctx, forDeny)
 		}
 	}
 	return false
@@ -390,42 +427,52 @@ func matchesLegacySubject(legacy, principal string, ctx *PolicyContext) bool {
 
 // matchesPrincipal checks if the principal matches the statement's principal
 func matchesPrincipal(principalSpec interface{}, principal string) bool {
+	return matchPrincipal(principalSpec, principal, false)
+}
+
+func matchPrincipal(principalSpec interface{}, principal string, forDeny bool) bool {
 	if principalSpec == nil {
 		return false
 	}
 
 	switch p := principalSpec.(type) {
 	case string:
-		// Direct string match or wildcard
-		if p == "*" || p == principal {
-			return true
-		}
-		// Support wildcard matching for principals (e.g., "arn:aws:iam::*:user/admin")
-		return wildcard.MatchesWildcard(p, principal)
+		return matchPrincipalString(p, principal, forDeny)
 	case []interface{}:
 		// Array of principals
 		for _, item := range p {
-			if str, ok := item.(string); ok {
-				if str == "*" || str == principal {
-					return true
-				}
-				// Support wildcard matching
-				if wildcard.MatchesWildcard(str, principal) {
-					return true
-				}
+			if str, ok := item.(string); ok && matchPrincipalString(str, principal, forDeny) {
+				return true
 			}
 		}
 	case map[string]interface{}:
 		// AWS-style principal with service prefix, e.g., {"AWS": "arn:aws:iam::..."}
 		// For S3 Tables, we primarily care about the AWS key
 		if aws, ok := p["AWS"]; ok {
-			return matchesPrincipal(aws, principal)
+			return matchPrincipal(aws, principal, forDeny)
 		}
 	}
 
 	return false
 }
 
+// matchPrincipalString reports whether a single principal spec matches the
+// caller. An ARN-scoped statement reaches a caller recorded under the ARN's
+// trailing name only for denies: normalization drops the ARN's account and
+// path, so an allow matched that way could grant a same-named caller the
+// statement never verified, while a deny must still land so it cannot be
+// ducked through the same collapse.
+func matchPrincipalString(spec, principal string, forDeny bool) bool {
+	if spec == "*" || spec == principal {
+		return true
+	}
+	if wildcard.MatchesWildcard(spec, principal) {
+		return true
+	}
+	return forDeny && strings.HasPrefix(spec, "arn:") && !strings.HasPrefix(principal, "arn:") &&
+		wildcard.MatchesWildcard(normalizePrincipalID(spec), principal)
+}
+
 // matchesAction checks if the action matches the statement's action
 func matchesAction(actionSpec interface{}, action string) bool {
 	if actionSpec == nil {
diff --git a/weed/s3api/s3tables/permissions_test.go b/weed/s3api/s3tables/permissions_test.go
index 8f91a22a32..bdab3c7f73 100644
--- a/weed/s3api/s3tables/permissions_test.go
+++ b/weed/s3api/s3tables/permissions_test.go
@@ -80,6 +80,13 @@ func TestMatchesPrincipal(t *testing.T) {
 		{"AWS map wildcard", map[string]interface{}{"AWS": "arn:aws:iam::*:user/admin"}, "arn:aws:iam::123:user/admin", true},
 		{"AWS map array", map[string]interface{}{"AWS": []interface{}{"user1", "user2"}}, "user1", true},
 
+		// ARN statements do not reach normalized-name callers on the allow
+		// path: normalization drops the ARN's account and path, so the grant
+		// could land on a same-named caller the statement never verified.
+		{"arn spec exact caller", "arn:aws:iam::123456789012:user/bob", "bob", false},
+		{"arn spec wildcard caller", "arn:aws:iam::*:user/alice", "alice", false},
+		{"arn spec no match", "arn:aws:iam::123456789012:user/carol", "dan", false},
+
 		// Nil/empty cases
 		{"nil principal", nil, "user123", false},
 	}
@@ -298,6 +305,26 @@ func TestCheckTableOrBucketPermissionDenyVeto(t *testing.T) {
 	}
 }
 
+// A principal that merely normalizes to the admin id must not bypass policy
+// denies: when the caller's actions are known, the admin bypass requires an
+// admin action; transports carrying no action list keep it.
+func TestCheckPermissionAdminRequiresAdminAction(t *testing.T) {
+	denyAll := `{"Statement":[{"Effect":"Deny","Principal":"*","Action":"s3tables:DeleteTable","Resource":"*"}]}`
+	arn := "arn:aws:s3tables:us-east-1:acct:bucket/b/table/ns/t"
+
+	reader := &PolicyContext{IdentityActions: []string{"Read"}}
+	if CheckPermissionWithContext("s3tables:DeleteTable", "admin", "admin", denyAll, arn, reader) {
+		t.Error("a name-only admin must not bypass an explicit deny")
+	}
+	admin := &PolicyContext{IdentityActions: []string{"Admin"}}
+	if !CheckPermissionWithContext("s3tables:DeleteTable", "admin", "admin", denyAll, arn, admin) {
+		t.Error("an identity holding the admin action keeps the bypass")
+	}
+	if !CheckPermissionWithContext("s3tables:DeleteTable", "admin", "admin", denyAll, arn, nil) {
+		t.Error("a caller with no action list keeps the trusted bypass")
+	}
+}
+
 func TestCheckTableOrBucketPermissionOwnerlessFailsClosed(t *testing.T) {
 	ctx := &PolicyContext{
 		IdentityActions: []string{"s3tables:*"},
@@ -450,3 +477,48 @@ func TestValidatePolicyDocument(t *testing.T) {
 		t.Fatal("s3vectors resource should be rejected")
 	}
 }
+
+// An ARN-scoped statement reaches a caller recorded under the ARN's trailing
+// name only for denies: normalization drops the account and path, so an allow
+// matched that way could grant a same-named caller the statement never
+// verified, while a deny that missed would let the caller duck the rule.
+func TestArnScopedPrincipalMatchesOnlyForDeny(t *testing.T) {
+	arn := "arn:aws:s3vectors:us-east-1:acct:bucket/b"
+	denyPolicy := `{"Statement":[{"Effect":"Deny","Principal":"arn:aws:iam::111:user/bob","Action":"s3vectors:DeleteVectors","Resource":"*"}]}`
+	allowPolicy := `{"Statement":[{"Effect":"Allow","Principal":"arn:aws:iam::111:user/bob","Action":"s3vectors:DeleteVectors","Resource":"*"}]}`
+
+	if CheckPermissionWithContext("s3vectors:DeleteVectors", "bob", "owner", denyPolicy, arn, &PolicyContext{}) {
+		t.Error("an ARN-scoped deny must still reach the normalized caller")
+	}
+	if CheckPermissionWithContext("s3vectors:DeleteVectors", "bob", "owner", allowPolicy, arn, &PolicyContext{}) {
+		t.Error("an ARN-scoped allow must not grant on the normalized name alone")
+	}
+}
+
+// A service wildcard administers only its own service: s3tables:* satisfies
+// the admin and identity grants for table operations but must not reach an
+// s3vectors operation.
+func TestServiceWildcardScopesToOwnService(t *testing.T) {
+	vectorOp := "s3vectors:DeleteVectors"
+	vectorArn := "arn:aws:s3vectors:us-east-1:acct:bucket/b"
+	tableOnly := &PolicyContext{IdentityActions: []string{"s3tables:*"}}
+
+	if CheckPermissionWithContext(vectorOp, "bob", "owner", "", vectorArn, tableOnly) {
+		t.Error("s3tables:* must not grant a vector operation")
+	}
+	if CheckPermissionWithContext(vectorOp, "admin", "owner", "", vectorArn, tableOnly) {
+		t.Error("s3tables:* must not fire the admin bypass for a vector operation")
+	}
+	if !CheckPermissionWithContext("GetTable", "bob", "owner", "", "arn:aws:s3tables:b", tableOnly) {
+		t.Error("s3tables:* keeps administering table operations")
+	}
+
+	vectorAdmin := &PolicyContext{IdentityActions: []string{"s3vectors:*"}}
+	if !CheckPermissionWithContext(vectorOp, "bob", "owner", "", vectorArn, vectorAdmin) {
+		t.Error("s3vectors:* administers vector operations")
+	}
+	global := &PolicyContext{IdentityActions: []string{"Admin"}}
+	if !CheckPermissionWithContext(vectorOp, "admin", "owner", "", vectorArn, global) {
+		t.Error("the global admin action keeps the bypass on vector operations")
+	}
+}
diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go
index 0145952cca..19c4aa51df 100644
--- a/weed/s3api/s3vectors/handlers.go
+++ b/weed/s3api/s3vectors/handlers.go
@@ -12,6 +12,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/vector_pb"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
 	"google.golang.org/grpc/codes"
@@ -132,15 +133,16 @@ func resolveIndexARNs(bucketName, bucketArn, indexName, indexArn string) (vector
 	return v, nil
 }
 
-// indexRecord fetches the index's catalog record; the call also authorizes
-// the request identity against the index resource.
-func (s *Server) indexRecord(ctx context.Context, r *http.Request, bucket, index string) (map[string]interface{}, error) {
+// indexRecord fetches the index's catalog record on the trusted path; the
+// operation's own vector permission — checked in authorizeVector before any
+// handler reaches this — is what gates access.
+func (s *Server) indexRecord(ctx context.Context, bucket, index string) (map[string]interface{}, error) {
 	var resp map[string]interface{}
-	err := s.catalog(ctx, "GetTable", &s3tables.GetTableRequest{
+	err := s.internalCatalog(ctx, "GetTable", &s3tables.GetTableRequest{
 		TableBucketARN: s.s3tablesArn(bucket),
 		Namespace:      []string{indexNamespace},
 		Name:           index,
-	}, &resp, r)
+	}, &resp)
 	return resp, err
 }
 
@@ -205,29 +207,6 @@ func vectorErr(w http.ResponseWriter, err error) {
 	}
 }
 
-// lanceBucket reports whether bucket is a table bucket that can hold vector
-// indexes: a declared Lance bucket, or one made before formats existed. An
-// explicit ICEBERG bucket is not a vector bucket and reads as absent. The
-// lookup is internal so the call needs no read grant — the requested
-// operation's own permission still gates it.
-func (s *Server) lanceBucket(ctx context.Context, bucket string) error {
-	var resp struct {
-		Format string `json:"format"`
-	}
-	err := s.internalCatalog(ctx, "GetTableBucket",
-		&s3tables.GetTableBucketRequest{TableBucketARN: s.s3tablesArn(bucket)}, &resp)
-	if err != nil {
-		return err
-	}
-	if resp.Format != "" && resp.Format != s3tables.FormatLance {
-		return &s3tables.S3TablesError{
-			Type:    s3tables.ErrCodeNoSuchBucket,
-			Message: fmt.Sprintf("vector bucket %s not found", bucket),
-		}
-	}
-	return nil
-}
-
 // ensureIndexNamespace declares the implicit index namespace when absent. The
 // calls are internal: a caller's CreateTable grant is what authorizes the
 // index, so it must not also need namespace reads or writes.
@@ -254,17 +233,6 @@ func (s *Server) ensureIndexNamespace(ctx context.Context, bucket string) error
 	return nil
 }
 
-// authorizeIndexWrite requires the update grant on the index's catalog table
-// — the same check credential vending uses — so a read-only GetTable grant
-// never reaches a mutation.
-func (s *Server) authorizeIndexWrite(ctx context.Context, r *http.Request, bucket, index string) error {
-	return s.tables.AuthorizeUpdateTable(ctx, s.filerClient, &s3tables.UpdateTableRequest{
-		TableBucketARN: s.s3tablesArn(bucket),
-		Namespace:      []string{indexNamespace},
-		Name:           index,
-	}, identityOf(r))
-}
-
 func (s *Server) handleCreateVectorBucket(w http.ResponseWriter, r *http.Request) {
 	var req CreateVectorBucketRequest
 	if err := readJSON(r, &req); err != nil {
@@ -309,30 +277,20 @@ func (s *Server) handleGetVectorBucket(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	var resp map[string]interface{}
-	err = s.catalog(r.Context(), "GetTableBucket",
-		&s3tables.GetTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp, r)
+	record, err := s.authorizeVector(r.Context(), r, "s3vectors:GetVectorBucket", arns.bucket, "")
 	if err != nil {
 		catalogErr(w, err)
 		return
 	}
-	// An explicit ICEBERG bucket is not a vector bucket.
-	if format, _ := resp["format"].(string); format != "" && format != s3tables.FormatLance {
-		writeError(w, http.StatusNotFound, "NotFoundException",
-			fmt.Sprintf("vector bucket %s not found", arns.bucket))
-		return
-	}
-	created, _ := time.Parse(time.RFC3339, fmt.Sprint(resp["createdAt"]))
 	bucket := &VectorBucket{
 		VectorBucketArn:  s.vectorBucketArn(arns.bucket),
 		VectorBucketName: arns.bucket,
-		CreationTime:     created,
+		CreationTime:     record.CreatedAt,
 	}
-	if cfg, ok := resp["encryptionConfiguration"].(map[string]interface{}); ok {
-		if v, ok := cfg["sseAlgorithm"]; ok {
-			cfg = map[string]interface{}{"sseType": v}
-		}
-		bucket.EncryptionConfiguration = cfg
+	if cfg, ok := record.EncryptionConfiguration["sseAlgorithm"]; ok {
+		bucket.EncryptionConfiguration = map[string]interface{}{"sseType": cfg}
+	} else if len(record.EncryptionConfiguration) > 0 {
+		bucket.EncryptionConfiguration = record.EncryptionConfiguration
 	}
 	writeJSON(w, http.StatusOK, &GetVectorBucketResponse{VectorBucket: bucket})
 }
@@ -352,14 +310,19 @@ func (s *Server) handleListVectorBuckets(w http.ResponseWriter, r *http.Request)
 		} `json:"tableBuckets"`
 		ContinuationToken string `json:"continuationToken"`
 	}
-	err := s.catalog(r.Context(), "ListTableBuckets",
+	// The listing runs internal and filters on the vector policy itself: a
+	// shared-slot check would hide buckets the caller was granted under
+	// s3vectors actions, while showing ones they were not.
+	err := s.internalCatalog(r.Context(), "ListTableBuckets",
 		&s3tables.ListTableBucketsRequest{
 			MaxBuckets: req.MaxResults, ContinuationToken: req.NextToken, Prefix: req.Prefix,
-		}, &resp, r)
+		}, &resp)
 	if err != nil {
 		catalogErr(w, err)
 		return
 	}
+	principal := s.tables.PrincipalFor(identityOf(r))
+	pctx := s.tables.PolicyContext(r)
 	// The catalog token continues the filtered page: ICEBERG buckets drop out
 	// of the listing, Lance and format-less legacy buckets stay.
 	out := &ListVectorBucketsResponse{NextToken: resp.ContinuationToken, VectorBuckets: []*VectorBucket{}}
@@ -367,6 +330,17 @@ func (s *Server) handleListVectorBuckets(w http.ResponseWriter, r *http.Request)
 		if b.Format != "" && b.Format != s3tables.FormatLance {
 			continue
 		}
+		owner, policy, err := s.vectorPolicyView(r.Context(), b.Name)
+		if err != nil {
+			// A bucket that vanished mid-list drops out of the page rather
+			// than failing the whole listing.
+			continue
+		}
+		arn := s.vectorBucketArn(b.Name)
+		if !s3tables.CheckPermissionWithContext("s3vectors:ListVectorBuckets", principal, owner, policy, arn, pctx) &&
+			!s3tables.CheckPermissionWithContext("s3vectors:GetVectorBucket", principal, owner, policy, arn, pctx) {
+			continue
+		}
 		created, _ := time.Parse(time.RFC3339, b.CreatedAt)
 		out.VectorBuckets = append(out.VectorBuckets, &VectorBucket{
 			VectorBucketArn:  s.vectorBucketArn(b.Name),
@@ -388,15 +362,13 @@ func (s *Server) handleDeleteVectorBucket(w http.ResponseWriter, r *http.Request
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	// The format check needs no read grant — the delete below is the
-	// authorization boundary — but an ICEBERG bucket is not a vector bucket.
-	if err := s.lanceBucket(r.Context(), arns.bucket); err != nil {
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:DeleteVectorBucket", arns.bucket, ""); err != nil {
 		catalogErr(w, err)
 		return
 	}
 	var resp map[string]interface{}
-	err = s.catalog(r.Context(), "DeleteTableBucket",
-		&s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp, r)
+	err = s.internalCatalog(r.Context(), "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
 		// holds no indexes; drop it under the delete's own authorization and
@@ -408,8 +380,8 @@ func (s *Server) handleDeleteVectorBucket(w http.ResponseWriter, r *http.Request
 				Namespace:      []string{indexNamespace},
 			}, &nsResp)
 		if nsErr == nil {
-			err = s.catalog(r.Context(), "DeleteTableBucket",
-				&s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp, r)
+			err = s.internalCatalog(r.Context(), "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)
 		}
@@ -437,8 +409,346 @@ func tableID(bucket, index string) *vector_pb.Table {
 	return &vector_pb.Table{Id: []string{bucket, indexNamespace, index}}
 }
 
-// Vector bucket policies are stored on the shared table bucket, the same
-// resource s3tables' PutTableBucketPolicy guards.
+// vectorPolicyKey names the table-bucket attribute holding the bucket's
+// s3vectors policy. The document stays in s3vectors form: the shared
+// evaluator only honors catalog actions, so vector operations evaluate it
+// themselves — against the s3vectors action and ARN the request names. A
+// separate key keeps a vector grant from ever authorizing a catalog call.
+// The constant lives in s3tables so the catalog asserts it in the update
+// preconditions of its own whole-entry writes.
+const vectorPolicyKey = s3tables.ExtendedKeyVectorPolicy
+
+// errNoVectorPolicy reports a missing policy document; handlers map it to a
+// 404 the same way the catalog maps a missing resource policy.
+var errNoVectorPolicy = errors.New("vector bucket policy not found")
+
+// validateVectorPolicy rejects a document the evaluator cannot honor: the
+// shared Statement shape has no negated fields, so a rule relying on one would
+// silently lose its exclusion; conditions are unsupported since the vector
+// evaluation supplies no condition keys; and a rule naming another service is
+// a dead rule the writer believes is enforced. The document must decode into
+// the same typed form the evaluator uses.
+func validateVectorPolicy(raw string) error {
+	var typed s3tables.PolicyDocument
+	if err := json.Unmarshal([]byte(raw), &typed); err != nil {
+		return fmt.Errorf("invalid policy document: %w", err)
+	}
+	var doc map[string]interface{}
+	if err := json.Unmarshal([]byte(raw), &doc); err != nil {
+		return fmt.Errorf("invalid policy document: %w", err)
+	}
+	for _, stmt := range policyStatements(doc) {
+		for key := range stmt {
+			for _, field := range []string{"NotAction", "NotResource", "NotPrincipal", "Condition"} {
+				if strings.EqualFold(key, field) {
+					return fmt.Errorf("invalid policy document: %s is not supported", field)
+				}
+			}
+		}
+	}
+	for _, stmt := range typed.Statement {
+		for _, action := range statementActions(stmt.Action) {
+			if service, _, found := strings.Cut(action, ":"); found && service != "s3vectors" {
+				return fmt.Errorf("invalid policy document: action %q is not enforceable in a vector bucket policy", action)
+			}
+		}
+		for _, resource := range statementActions(stmt.Resource) {
+			if service := arnService(resource); service != "" && service != "s3vectors" {
+				return fmt.Errorf("invalid policy document: resource %q is not enforceable in a vector bucket policy", resource)
+			}
+		}
+	}
+	return nil
+}
+
+// arnService returns the service field of an ARN in any partition.
+func arnService(resource string) string {
+	parts := strings.SplitN(resource, ":", 4)
+	if len(parts) == 4 && parts[0] == "arn" {
+		return parts[2]
+	}
+	return ""
+}
+
+func policyStatements(doc map[string]interface{}) []map[string]interface{} {
+	var stmts []map[string]interface{}
+	switch s := doc["Statement"].(type) {
+	case []interface{}:
+		for _, item := range s {
+			if stmt, ok := item.(map[string]interface{}); ok {
+				stmts = append(stmts, stmt)
+			}
+		}
+	case map[string]interface{}:
+		stmts = append(stmts, s)
+	}
+	return stmts
+}
+
+// policyHasVectorAction reports whether any statement names an s3vectors
+// action — how a raw vector document in the shared policy slot is recognized.
+// It reads the typed document so detection matches what the evaluator sees.
+func policyHasVectorAction(raw string) bool {
+	var doc s3tables.PolicyDocument
+	if json.Unmarshal([]byte(raw), &doc) != nil {
+		return false
+	}
+	for _, stmt := range doc.Statement {
+		for _, action := range statementActions(stmt.Action) {
+			if strings.HasPrefix(action, "s3vectors:") {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func statementActions(spec interface{}) []string {
+	switch a := spec.(type) {
+	case string:
+		return []string{a}
+	case []interface{}:
+		var actions []string
+		for _, item := range a {
+			if str, ok := item.(string); ok {
+				actions = append(actions, str)
+			}
+		}
+		return actions
+	}
+	return nil
+}
+
+func splitEntryPath(p string) (dir, name string) {
+	i := strings.LastIndex(p, "/")
+	return p[:i], p[i+1:]
+}
+
+// vectorPolicyFromExt resolves the effective s3vectors policy from an entry's
+// attributes: the dedicated key wins, else a shared-slot document that still
+// names vector actions.
+func vectorPolicyFromExt(ext map[string][]byte) string {
+	if doc := ext[vectorPolicyKey]; len(doc) > 0 {
+		return string(doc)
+	}
+	if doc := string(ext[s3tables.ExtendedKeyPolicy]); policyHasVectorAction(doc) {
+		return doc
+	}
+	return ""
+}
+
+// stripVectorActions removes s3vectors statements from a shared-slot
+// document, keeping its catalog rules: a statement mixing both kinds keeps
+// its catalog actions, and a now-empty statement drops out. It reports the
+// document to write back — empty when nothing remains — and whether anything
+// changed.
+func stripVectorActions(raw string) (string, bool) {
+	var doc map[string]interface{}
+	if err := json.Unmarshal([]byte(raw), &doc); err != nil {
+		return "", false
+	}
+	changed := false
+	var kept []map[string]interface{}
+	for _, stmt := range policyStatements(doc) {
+		dropped := false
+		for key, value := range stmt {
+			if !strings.EqualFold(key, "Action") {
+				continue
+			}
+			var actions []string
+			keyDropped := false
+			for _, a := range statementActions(value) {
+				if strings.HasPrefix(a, "s3vectors:") {
+					keyDropped = true
+					continue
+				}
+				actions = append(actions, a)
+			}
+			if !keyDropped {
+				continue
+			}
+			dropped = true
+			if len(actions) > 0 {
+				stmt[key] = actions
+			} else {
+				delete(stmt, key)
+			}
+		}
+		if !dropped {
+			kept = append(kept, stmt)
+			continue
+		}
+		changed = true
+		for key := range stmt {
+			if strings.EqualFold(key, "Action") {
+				kept = append(kept, stmt)
+				break
+			}
+		}
+	}
+	if !changed {
+		return "", false
+	}
+	if len(kept) == 0 {
+		return "", true
+	}
+	doc["Statement"] = kept
+	out, err := json.Marshal(doc)
+	if err != nil {
+		return "", false
+	}
+	return string(out), true
+}
+
+// vectorBucket fetches the bucket's catalog record under the trusted
+// identity for the authorization lookup; nothing reaches the caller until
+// the permission check below it passes.
+func (s *Server) vectorBucket(ctx context.Context, bucket string) (*s3tables.GetTableBucketResponse, error) {
+	var resp s3tables.GetTableBucketResponse
+	if err := s.internalCatalog(ctx, "GetTableBucket",
+		&s3tables.GetTableBucketRequest{TableBucketARN: s.s3tablesArn(bucket)}, &resp); err != nil {
+		return nil, err
+	}
+	return &resp, nil
+}
+
+// vectorPolicy reads the bucket's stored s3vectors policy. A bucket written
+// before the dedicated key carried the raw document in the shared policy
+// slot; one that still names vector actions keeps working until the next
+// vector policy write migrates it out.
+func (s *Server) vectorPolicy(ctx context.Context, bucket string) (string, error) {
+	dir, name := splitEntryPath(s3tables.GetTableBucketPath(bucket))
+	var policy string
+	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
+			Directory: dir,
+			Name:      name,
+		})
+		if err != nil {
+			return err
+		}
+		policy = vectorPolicyFromExt(resp.Entry.Extended)
+		return nil
+	})
+	return policy, err
+}
+
+// vectorPolicyView resolves a bucket's owner and stored s3vectors policy in
+// one filer lookup — the per-bucket check the vector listing runs.
+func (s *Server) vectorPolicyView(ctx context.Context, bucket string) (owner, policy string, err error) {
+	dir, name := splitEntryPath(s3tables.GetTableBucketPath(bucket))
+	err = s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
+			Directory: dir,
+			Name:      name,
+		})
+		if err != nil {
+			return err
+		}
+		var meta struct {
+			OwnerAccountID string `json:"ownerAccountId"`
+		}
+		if raw := resp.Entry.Extended[s3tables.ExtendedKeyMetadata]; len(raw) > 0 {
+			_ = json.Unmarshal(raw, &meta)
+		}
+		owner = meta.OwnerAccountID
+		policy = vectorPolicyFromExt(resp.Entry.Extended)
+		return nil
+	})
+	return owner, policy, err
+}
+
+// mutateVectorPolicy applies a change to the bucket's vector-policy
+// attribute under the entry's extended-attribute compare-and-swap, retrying
+// a racing writer the way the catalog's own policy updates do.
+func (s *Server) mutateVectorPolicy(ctx context.Context, bucket string, mutate func(ext map[string][]byte) error) error {
+	dir, name := splitEntryPath(s3tables.GetTableBucketPath(bucket))
+	return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		for attempt := 0; attempt < 3; attempt++ {
+			resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{
+				Directory: dir,
+				Name:      name,
+			})
+			if err != nil {
+				return err
+			}
+			entry := resp.Entry
+			// Assert the vector policy key stays absent too, so a concurrent
+			// policy write trips the compare-and-swap instead of being
+			// silently reverted by the whole-entry update.
+			expected := s3tables.SnapshotExtended(entry.Extended, vectorPolicyKey)
+			if entry.Extended == nil {
+				entry.Extended = make(map[string][]byte)
+			}
+			if err := mutate(entry.Extended); err != nil {
+				return err
+			}
+			if _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
+				Directory:        dir,
+				Entry:            entry,
+				ExpectedExtended: expected,
+			}); err == nil {
+				return nil
+			} else if status.Code(err) != codes.FailedPrecondition {
+				return err
+			}
+		}
+		return &s3tables.S3TablesError{Type: s3tables.ErrCodeConflict,
+			Message: "vector bucket policy changed concurrently, retry the request"}
+	})
+}
+
+// isVectorBucket reports whether the catalog record describes a vector
+// bucket: an explicit ICEBERG bucket is not one, while a bucket declared
+// before formats existed reads as Lance.
+func isVectorBucket(record *s3tables.GetTableBucketResponse) bool {
+	return record.Format == "" || record.Format == s3tables.FormatLance
+}
+
+// errVectorDenied marks a failed vector permission check inside a policy
+// mutation; catalogErr renders it as AccessDenied.
+var errVectorDenied = &s3tables.S3TablesError{Type: s3tables.ErrCodeAccessDenied, Message: "access denied"}
+
+// 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 {
+	principal := s.tables.PrincipalFor(identityOf(r))
+	return s3tables.CheckPermissionWithContext(action, principal, owner,
+		vectorPolicyFromExt(ext), s.vectorBucketArn(bucket), s.tables.PolicyContext(r))
+}
+
+// authorizeVector grants the request's own vector operation — not a catalog
+// stand-in — against the stored s3vectors policy and the ARN it actually
+// targets, so an index-scoped rule reaches the index it names and a read
+// grant never carries a write. Catalog bookkeeping below rides on the
+// trusted internal path, the same pattern the implicit namespace uses.
+// The returned record saves the handler a second bucket lookup.
+func (s *Server) authorizeVector(ctx context.Context, r *http.Request, action, bucket, index string) (*s3tables.GetTableBucketResponse, error) {
+	record, err := s.vectorBucket(ctx, bucket)
+	if err != nil {
+		return nil, err
+	}
+	policy, err := s.vectorPolicy(ctx, bucket)
+	if err != nil {
+		return nil, err
+	}
+	resourceARN := s.vectorBucketArn(bucket)
+	if index != "" {
+		resourceARN = s.indexArn(bucket, index)
+	}
+	principal := s.tables.PrincipalFor(identityOf(r))
+	if !s3tables.CheckPermissionWithContext(action, principal, record.OwnerAccountID, policy, resourceARN, s.tables.PolicyContext(r)) {
+		return nil, &s3tables.S3TablesError{Type: s3tables.ErrCodeAccessDenied, Message: "access denied"}
+	}
+	if !isVectorBucket(record) {
+		return nil, &s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchBucket,
+			Message: fmt.Sprintf("vector bucket %s not found", bucket)}
+	}
+	return record, nil
+}
+
 func (s *Server) handlePutVectorBucketPolicy(w http.ResponseWriter, r *http.Request) {
 	var req PutVectorBucketPolicyRequest
 	if err := readJSON(r, &req); err != nil {
@@ -450,15 +760,36 @@ func (s *Server) handlePutVectorBucketPolicy(w http.ResponseWriter, r *http.Requ
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	if err := s.lanceBucket(r.Context(), arns.bucket); err != nil {
+	if err := validateVectorPolicy(req.Policy); err != nil {
+		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
+		return
+	}
+	record, err := s.vectorBucket(r.Context(), arns.bucket)
+	if err != nil {
 		catalogErr(w, err)
 		return
 	}
-	var resp map[string]interface{}
-	err = s.catalog(r.Context(), "PutTableBucketPolicy", &s3tables.PutTableBucketPolicyRequest{
-		TableBucketARN: s.s3tablesArn(arns.bucket),
-		ResourcePolicy: req.Policy,
-	}, &resp, r)
+	if !isVectorBucket(record) {
+		writeError(w, http.StatusNotFound, "NotFoundException",
+			fmt.Sprintf("vector bucket %s not found", arns.bucket))
+		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
+		}
+		// A raw vector document still in the shared slot migrates out with
+		// this write; catalog rules in it stay untouched.
+		if stripped, changed := stripVectorActions(string(ext[s3tables.ExtendedKeyPolicy])); changed {
+			if stripped == "" {
+				delete(ext, s3tables.ExtendedKeyPolicy)
+			} else {
+				ext[s3tables.ExtendedKeyPolicy] = []byte(stripped)
+			}
+		}
+		ext[vectorPolicyKey] = []byte(req.Policy)
+		return nil
+	})
 	if err != nil {
 		catalogErr(w, err)
 		return
@@ -477,19 +808,20 @@ func (s *Server) handleGetVectorBucketPolicy(w http.ResponseWriter, r *http.Requ
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	if err := s.lanceBucket(r.Context(), arns.bucket); err != nil {
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:GetVectorBucketPolicy", arns.bucket, ""); err != nil {
 		catalogErr(w, err)
 		return
 	}
-	var resp s3tables.GetTableBucketPolicyResponse
-	err = s.catalog(r.Context(), "GetTableBucketPolicy", &s3tables.GetTableBucketPolicyRequest{
-		TableBucketARN: s.s3tablesArn(arns.bucket),
-	}, &resp, r)
+	policy, err := s.vectorPolicy(r.Context(), arns.bucket)
 	if err != nil {
 		catalogErr(w, err)
 		return
 	}
-	writeJSON(w, http.StatusOK, &GetVectorBucketPolicyResponse{Policy: resp.ResourcePolicy})
+	if policy == "" {
+		writeError(w, http.StatusNotFound, "NotFoundException", errNoVectorPolicy.Error())
+		return
+	}
+	writeJSON(w, http.StatusOK, &GetVectorBucketPolicyResponse{Policy: policy})
 }
 
 func (s *Server) handleDeleteVectorBucketPolicy(w http.ResponseWriter, r *http.Request) {
@@ -503,14 +835,44 @@ func (s *Server) handleDeleteVectorBucketPolicy(w http.ResponseWriter, r *http.R
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	if err := s.lanceBucket(r.Context(), arns.bucket); err != nil {
+	record, err := s.vectorBucket(r.Context(), arns.bucket)
+	if err != nil {
 		catalogErr(w, err)
 		return
 	}
-	var resp map[string]interface{}
-	err = s.catalog(r.Context(), "DeleteTableBucketPolicy", &s3tables.DeleteTableBucketPolicyRequest{
-		TableBucketARN: s.s3tablesArn(arns.bucket),
-	}, &resp, r)
+	if !isVectorBucket(record) {
+		writeError(w, http.StatusNotFound, "NotFoundException",
+			fmt.Sprintf("vector bucket %s not found", arns.bucket))
+		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
+		}
+		removed := false
+		if _, ok := ext[vectorPolicyKey]; ok {
+			delete(ext, vectorPolicyKey)
+			removed = true
+		}
+		// A legacy document in the shared slot loses its vector rules with
+		// the delete; catalog rules in it stay untouched.
+		if stripped, changed := stripVectorActions(string(ext[s3tables.ExtendedKeyPolicy])); changed {
+			removed = true
+			if stripped == "" {
+				delete(ext, s3tables.ExtendedKeyPolicy)
+			} else {
+				ext[s3tables.ExtendedKeyPolicy] = []byte(stripped)
+			}
+		}
+		if !removed {
+			return errNoVectorPolicy
+		}
+		return nil
+	})
+	if errors.Is(err, errNoVectorPolicy) {
+		writeError(w, http.StatusNotFound, "NotFoundException", errNoVectorPolicy.Error())
+		return
+	}
 	if err != nil {
 		catalogErr(w, err)
 		return
@@ -542,19 +904,9 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", "dimension must be 1..=4096")
 		return
 	}
-	if err := s.lanceBucket(r.Context(), arns.bucket); err != nil {
-		catalogErr(w, err)
-		return
-	}
 	// The create grant is exercised before any write: a denied caller must not
-	// leave even the implicit namespace behind. The namespace may not exist
-	// yet, so the check tolerates its absence.
-	if err := s.tables.AuthorizeCreateTable(r.Context(), s.filerClient, &s3tables.CreateTableRequest{
-		TableBucketARN:        s.s3tablesArn(arns.bucket),
-		Namespace:             []string{indexNamespace},
-		Name:                  arns.index,
-		AllowMissingNamespace: true,
-	}, identityOf(r)); err != nil {
+	// leave even the implicit namespace behind.
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:CreateIndex", arns.bucket, arns.index); err != nil {
 		catalogErr(w, err)
 		return
 	}
@@ -594,13 +946,13 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 	metadataJSON, _ := json.Marshal(metadata)
-	err = s.catalog(r.Context(), "CreateTable", &s3tables.CreateTableRequest{
+	err = s.internalCatalog(r.Context(), "CreateTable", &s3tables.CreateTableRequest{
 		TableBucketARN: s.s3tablesArn(arns.bucket),
 		Namespace:      []string{indexNamespace},
 		Name:           arns.index,
 		Format:         s3tables.FormatLance,
 		Metadata:       &s3tables.TableMetadata{FullMetadata: metadataJSON},
-	}, nil, r)
+	}, nil)
 	if err != nil {
 		catalogErr(w, err)
 		return
@@ -705,12 +1057,16 @@ func (s *Server) handleGetIndex(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:GetIndex", arns.bucket, arns.index); err != nil {
+		catalogErr(w, err)
+		return
+	}
 	var resp map[string]interface{}
-	err = s.catalog(r.Context(), "GetTable", &s3tables.GetTableRequest{
+	err = s.internalCatalog(r.Context(), "GetTable", &s3tables.GetTableRequest{
 		TableBucketARN: s.s3tablesArn(arns.bucket),
 		Namespace:      []string{indexNamespace},
 		Name:           arns.index,
-	}, &resp, r)
+	}, &resp)
 	if err != nil {
 		catalogErr(w, err)
 		return
@@ -729,6 +1085,10 @@ func (s *Server) handleListIndexes(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:ListIndexes", arns.bucket, ""); err != nil {
+		catalogErr(w, err)
+		return
+	}
 	var resp struct {
 		TableSummaries []struct {
 			Name      string `json:"name"`
@@ -737,13 +1097,13 @@ func (s *Server) handleListIndexes(w http.ResponseWriter, r *http.Request) {
 		} `json:"tables"`
 		ContinuationToken string `json:"continuationToken"`
 	}
-	err = s.catalog(r.Context(), "ListTables", &s3tables.ListTablesRequest{
+	err = s.internalCatalog(r.Context(), "ListTables", &s3tables.ListTablesRequest{
 		TableBucketARN:    s.s3tablesArn(arns.bucket),
 		Namespace:         []string{indexNamespace},
 		MaxTables:         req.MaxResults,
 		ContinuationToken: req.NextToken,
 		Prefix:            req.Prefix,
-	}, &resp, r)
+	}, &resp)
 	if err != nil {
 		catalogErr(w, err)
 		return
@@ -778,12 +1138,13 @@ 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 {
+		catalogErr(w, err)
+		return
+	}
 	// A named table is an index only while it carries the Lance format —
 	// an Iceberg or format-less table under the same namespace is not an
 	// index and must not be deleted by this operation.
-	// The format check needs no read grant — the delete below is the
-	// authorization boundary — and only a Lance-formatted record is a
-	// vector index.
 	var record map[string]interface{}
 	if err := s.internalCatalog(r.Context(), "GetTable", &s3tables.GetTableRequest{
 		TableBucketARN: s.s3tablesArn(arns.bucket),
@@ -801,11 +1162,11 @@ 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{}
-	err = s.catalog(r.Context(), "DeleteTable", &s3tables.DeleteTableRequest{
+	err = s.internalCatalog(r.Context(), "DeleteTable", &s3tables.DeleteTableRequest{
 		TableBucketARN: s.s3tablesArn(arns.bucket),
 		Namespace:      []string{indexNamespace},
 		Name:           arns.index,
-	}, &resp, r)
+	}, &resp)
 	if err != nil {
 		catalogErr(w, err)
 		return
@@ -843,14 +1204,14 @@ func (s *Server) handlePutVectors(w http.ResponseWriter, r *http.Request) {
 			fmt.Sprintf("vectors must contain between 1 and %d items", vectorsMaxPutBatch))
 		return
 	}
-	record, err := s.indexRecord(r.Context(), r, arns.bucket, arns.index)
-	if err != nil {
+	// A read grant on the index does not carry a write: mutations require the
+	// vector write grant, the same check credential vending applies.
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:PutVectors", arns.bucket, arns.index); err != nil {
 		catalogErr(w, err)
 		return
 	}
-	// A read grant on the index does not carry a write: mutations require the
-	// catalog's update grant, the same check credential vending applies.
-	if err := s.authorizeIndexWrite(r.Context(), r, arns.bucket, arns.index); err != nil {
+	record, err := s.indexRecord(r.Context(), arns.bucket, arns.index)
+	if err != nil {
 		catalogErr(w, err)
 		return
 	}
@@ -922,7 +1283,11 @@ func (s *Server) handleGetVectors(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	if _, err := s.indexRecord(r.Context(), r, arns.bucket, arns.index); err != nil {
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:GetVectors", arns.bucket, arns.index); err != nil {
+		catalogErr(w, err)
+		return
+	}
+	if _, err := s.indexRecord(r.Context(), arns.bucket, arns.index); err != nil {
 		catalogErr(w, err)
 		return
 	}
@@ -952,7 +1317,11 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	if _, err := s.indexRecord(r.Context(), r, arns.bucket, arns.index); err != nil {
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:ListVectors", arns.bucket, arns.index); err != nil {
+		catalogErr(w, err)
+		return
+	}
+	if _, err := s.indexRecord(r.Context(), arns.bucket, arns.index); err != nil {
 		catalogErr(w, err)
 		return
 	}
@@ -1075,11 +1444,11 @@ func (s *Server) handleDeleteVectors(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusBadRequest, "ValidationException", err.Error())
 		return
 	}
-	if _, err := s.indexRecord(r.Context(), r, arns.bucket, arns.index); err != nil {
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:DeleteVectors", arns.bucket, arns.index); err != nil {
 		catalogErr(w, err)
 		return
 	}
-	if err := s.authorizeIndexWrite(r.Context(), r, arns.bucket, arns.index); err != nil {
+	if _, err := s.indexRecord(r.Context(), arns.bucket, arns.index); err != nil {
 		catalogErr(w, err)
 		return
 	}
@@ -1378,7 +1747,11 @@ func (s *Server) handleQueryVectors(w http.ResponseWriter, r *http.Request) {
 			fmt.Sprintf("topK must be between 1 and %d", vectorsMaxTopK))
 		return
 	}
-	record, err := s.indexRecord(r.Context(), r, arns.bucket, arns.index)
+	if _, err := s.authorizeVector(r.Context(), r, "s3vectors:QueryVectors", arns.bucket, arns.index); err != nil {
+		catalogErr(w, err)
+		return
+	}
+	record, err := s.indexRecord(r.Context(), arns.bucket, arns.index)
 	if err != nil {
 		catalogErr(w, err)
 		return
diff --git a/weed/s3api/s3vectors/handlers_test.go b/weed/s3api/s3vectors/handlers_test.go
index 9b45d9c2d4..f39c678876 100644
--- a/weed/s3api/s3vectors/handlers_test.go
+++ b/weed/s3api/s3vectors/handlers_test.go
@@ -72,6 +72,10 @@ func (f *fakeVectors) CreateIndex(ctx context.Context, req *vector_pb.CreateInde
 	return &vector_pb.CreateIndexResponse{}, nil
 }
 
+func (f *fakeVectors) GetVectors(ctx context.Context, req *vector_pb.GetVectorsRequest, opts ...grpc.CallOption) (*vector_pb.GetVectorsResponse, error) {
+	return &vector_pb.GetVectorsResponse{}, nil
+}
+
 func (f *fakeVectors) PutVectors(ctx context.Context, req *vector_pb.PutVectorsRequest, opts ...grpc.CallOption) (*vector_pb.PutVectorsResponse, error) {
 	f.puts = append(f.puts, req)
 	return &vector_pb.PutVectorsResponse{}, nil
@@ -524,6 +528,7 @@ func TestQueryVectorsTopKBounds(t *testing.T) {
 	s := &Server{
 		vectors:     newFakeVectors(0),
 		tables:      s3tables.NewManager(),
+		internal:    s3tables.NewManager(),
 		filerClient: errFiler{},
 	}
 	for _, tc := range []struct {
@@ -572,6 +577,7 @@ func TestValidateEncryptionConfigurationRejectsCustomerKeys(t *testing.T) {
 		})
 	}
 }
+
 // A vector index is colocated: its catalog entry sits at the dataset root, so
 // the catalog delete removes entry and data together.
 func TestDeleteIndexRemovesDatasetDirectory(t *testing.T) {
@@ -617,3 +623,197 @@ func TestDeleteIndexRemovesDatasetDirectory(t *testing.T) {
 		t.Fatal("index dataset survived DeleteIndex")
 	}
 }
+
+func TestValidateVectorPolicy(t *testing.T) {
+	valid := `{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","Action":"s3vectors:PutVectors","Resource":"arn:aws:s3vectors:r:a:bucket/b"}]}`
+	if err := validateVectorPolicy(valid); err != nil {
+		t.Fatalf("valid policy rejected: %v", err)
+	}
+	if err := validateVectorPolicy("not-json"); err == nil {
+		t.Fatal("malformed policy should be rejected")
+	}
+	for _, field := range []string{"NotAction", "NotResource", "NotPrincipal", "Condition"} {
+		bad := fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","%s":"s3vectors:GetIndex","Action":"s3vectors:PutVectors","Resource":"*"}]}`, field)
+		if err := validateVectorPolicy(bad); err == nil {
+			t.Fatalf("%s should be rejected", field)
+		}
+		lower := fmt.Sprintf(`{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Principal":"*","%s":"s3vectors:GetIndex","Action":"s3vectors:PutVectors","Resource":"*"}]}`, strings.ToLower(field))
+		if err := validateVectorPolicy(lower); err == nil {
+			t.Fatalf("case-variant %s should be rejected", strings.ToLower(field))
+		}
+	}
+	for _, dead := range []string{
+		`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:GetTable","Resource":"*"}]}`,
+		`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","action":"s3tables:GetTable","Resource":"*"}]}`,
+		`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectors","Resource":"arn:aws:s3:r:a:bucket/b"}]}`,
+		`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectors","resource":"arn:aws:s3tables:r:a:bucket/b"}]}`,
+		`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectors","Resource":"arn:aws:s3tables:r:a:bucket/b"}]}`,
+		`{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectors","Resource":"arn:aws-cn:s3tables:r:a:bucket/b"}]}`,
+	} {
+		if err := validateVectorPolicy(dead); err == nil {
+			t.Fatalf("unenforceable rule should be rejected: %s", dead)
+		}
+	}
+}
+
+// postJSONAs runs the handler under a named identity: empty resolves to the
+// account the catalog would charge, a name is a non-owner principal.
+func postJSONAs(t *testing.T, identity string, 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))
+	handler(w, r.WithContext(contextWithIdentity(r.Context(), identity)))
+	return w
+}
+
+// A grant names the exact vector action and resource it allows: a put grant
+// carries no delete, an index-scoped rule reaches only that index, and a
+// read grant never authorizes a write.
+func TestVectorBucketPolicyGatesVectorActions(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())
+	}
+	newIndex := func(name string) {
+		t.Helper()
+		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())
+		}
+	}
+	newIndex("idx")
+	newIndex("other")
+
+	indexARN := s.indexArn("vec-bucket", "idx")
+	policy := fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+		`{"Sid":"put","Effect":"Allow","Principal":"*","Action":"s3vectors:PutVectors","Resource":"%s"},`+
+		`{"Sid":"reads","Effect":"Allow","Principal":"*","Action":"s3vectors:Get*","Resource":"%s"},`+
+		`{"Sid":"noDelete","Effect":"Deny","Principal":"*","Action":"s3vectors:DeleteVectors","Resource":"%s"}]}`,
+		indexARN, indexARN, indexARN)
+	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())
+	}
+	if w := postJSONAs(t, "", s.handleGetVectorBucketPolicy, &GetVectorBucketPolicyRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK {
+		t.Fatalf("get policy: %d %s", w.Code, w.Body.String())
+	} else {
+		var resp GetVectorBucketPolicyResponse
+		if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
+			t.Fatalf("decode policy response: %v", err)
+		}
+		if resp.Policy != policy {
+			t.Fatalf("policy round-trip changed the document:\nsaved: %s\ngot:   %s", policy, resp.Policy)
+		}
+	}
+
+	put := func(index string) *httptest.ResponseRecorder {
+		return postJSONAs(t, "alice", s.handlePutVectors, &PutVectorsRequest{
+			VectorBucketName: "vec-bucket",
+			IndexName:        index,
+			Vectors:          []*PutInputVector{{Key: "a", Data: &VectorData{Float32: []float32{1, 2, 3}}}},
+		})
+	}
+	if w := put("idx"); w.Code != http.StatusOK {
+		t.Fatalf("put on granted index: %d %s", w.Code, w.Body.String())
+	}
+	if w := put("other"); w.Code != http.StatusForbidden {
+		t.Fatalf("put on ungranted index: %d %s, want 403", w.Code, w.Body.String())
+	}
+	if w := postJSONAs(t, "alice", s.handleDeleteVectors, &DeleteVectorsRequest{
+		VectorBucketName: "vec-bucket", IndexName: "idx", Keys: []string{"a"},
+	}); w.Code != http.StatusForbidden {
+		t.Fatalf("delete under deny: %d %s, want 403", w.Code, w.Body.String())
+	}
+	if w := postJSONAs(t, "alice", s.handleGetVectors, &GetVectorsRequest{
+		VectorBucketName: "vec-bucket", IndexName: "idx", Keys: []string{"a"},
+	}); w.Code != http.StatusOK {
+		t.Fatalf("get under wildcard read grant: %d %s", w.Code, w.Body.String())
+	}
+	if w := postJSONAs(t, "alice", s.handleListIndexes, &ListIndexesRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusForbidden {
+		t.Fatalf("bucket-scoped list under index grant: %d %s, want 403", w.Code, w.Body.String())
+	}
+}
+
+// A vector document still living in the shared policy slot is a legacy
+// policy: deleting through the vector API strips its s3vectors statements,
+// keeps catalog rules, and the stripped grant stops authorizing.
+func TestDeleteVectorBucketPolicyRemovesLegacyDoc(t *testing.T) {
+	fs := s3tablestest.Start(t)
+	s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0))
+
+	lookupEntry := func(bucket string) *filer_pb.Entry {
+		t.Helper()
+		dir, name := splitEntryPath(s3tables.GetTableBucketPath(bucket))
+		resp, err := filer_pb.LookupEntry(context.Background(), fs.Client, &filer_pb.LookupDirectoryEntryRequest{Directory: dir, Name: name})
+		if err != nil {
+			t.Fatalf("lookup bucket %s: %v", bucket, err)
+		}
+		return resp.Entry
+	}
+	writeSharedPolicy := func(bucket, doc string) {
+		t.Helper()
+		entry := lookupEntry(bucket)
+		if entry.Extended == nil {
+			entry.Extended = map[string][]byte{}
+		}
+		entry.Extended[s3tables.ExtendedKeyPolicy] = []byte(doc)
+		dir, _ := splitEntryPath(s3tables.GetTableBucketPath(bucket))
+		if _, err := fs.Client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{Directory: dir, Entry: entry}); err != nil {
+			t.Fatalf("write shared policy on %s: %v", bucket, err)
+		}
+	}
+	create := func(bucket string) {
+		t.Helper()
+		if w := postJSONAs(t, "", s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: bucket}); w.Code != http.StatusOK {
+			t.Fatalf("create bucket %s: %d %s", bucket, w.Code, w.Body.String())
+		}
+	}
+	deletePolicy := func(bucket string) *httptest.ResponseRecorder {
+		return postJSONAs(t, "", s.handleDeleteVectorBucketPolicy, &DeleteVectorBucketPolicyRequest{VectorBucketName: bucket})
+	}
+	getAsAlice := func(bucket string) *httptest.ResponseRecorder {
+		return postJSONAs(t, "alice", s.handleGetVectorBucket, &GetVectorBucketRequest{VectorBucketName: bucket})
+	}
+
+	create("vec-mixed")
+	writeSharedPolicy("vec-mixed", fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+		`{"Effect":"Allow","Principal":"*","Action":"s3tables:GetTableBucket","Resource":"%s"},`+
+		`{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectorBucket","Resource":"%s"}]}`,
+		s.s3tablesArn("vec-mixed"), s.vectorBucketArn("vec-mixed")))
+	if w := getAsAlice("vec-mixed"); w.Code != http.StatusOK {
+		t.Fatalf("legacy grant should authorize: %d %s", w.Code, w.Body.String())
+	}
+	if w := deletePolicy("vec-mixed"); w.Code != http.StatusOK {
+		t.Fatalf("delete mixed policy: %d %s", w.Code, w.Body.String())
+	}
+	if w := getAsAlice("vec-mixed"); w.Code != http.StatusForbidden {
+		t.Fatalf("stripped grant should deny: %d %s, want 403", w.Code, w.Body.String())
+	}
+	if doc := string(lookupEntry("vec-mixed").Extended[s3tables.ExtendedKeyPolicy]); !strings.Contains(doc, "s3tables:") || strings.Contains(doc, "s3vectors:") {
+		t.Fatalf("catalog statements should survive the delete: %s", doc)
+	}
+
+	create("vec-only")
+	writeSharedPolicy("vec-only", fmt.Sprintf(`{"Version":"2012-10-17","Statement":[`+
+		`{"Effect":"Allow","Principal":"*","Action":"s3vectors:GetVectorBucket","Resource":"%s"}]}`,
+		s.vectorBucketArn("vec-only")))
+	if w := deletePolicy("vec-only"); w.Code != http.StatusOK {
+		t.Fatalf("delete vector-only policy: %d %s", w.Code, w.Body.String())
+	}
+	if doc := string(lookupEntry("vec-only").Extended[s3tables.ExtendedKeyPolicy]); doc != "" {
+		t.Fatalf("empty shared policy should be removed: %s", doc)
+	}
+}
diff --git a/weed/s3api/s3vectors/owner_check_test.go b/weed/s3api/s3vectors/owner_check_test.go
new file mode 100644
index 0000000000..9e6372b8b5
--- /dev/null
+++ b/weed/s3api/s3vectors/owner_check_test.go
@@ -0,0 +1,28 @@
+package s3vectors
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
+)
+
+// The zero-config deployment authenticates every request as "admin"; the
+// principal the vector checks compute must match the owner the catalog
+// records at creation.
+func TestAdminNamedIdentityReachesOwnBucket(t *testing.T) {
+	fs := s3tablestest.Start(t)
+	s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, newFakeVectors(0))
+	if w := postJSONAs(t, "admin", s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK {
+		t.Fatalf("create: %d %s", w.Code, w.Body.String())
+	}
+	if w := postJSONAs(t, "admin", s.handleGetVectorBucket, &GetVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK {
+		t.Fatalf("get denied after admin-named create: %d %s", w.Code, w.Body.String())
+	}
+	if w := postJSONAs(t, "admin", s.handleCreateIndex, &CreateIndexRequest{
+		VectorBucketName: "vec-bucket", IndexName: "idx", DataType: "float32", Dimension: 3, DistanceMetric: "euclidean",
+	}); w.Code != http.StatusOK {
+		t.Fatalf("create index denied after admin-named create: %d %s", w.Code, w.Body.String())
+	}
+}