s3tables: scope management authorization to the caller's identity (#9961)

* s3tables: resolve account-less identities to a distinct principal

Static identities with no account block default to the shared admin
account, so getAccountID returned "admin" for every such user and the
permission checks treated them all as the admin principal. Only keep the
admin account when the identity actually carries an admin action;
otherwise fall back to the unique identity name.

* s3tables: limit the open-by-default fallback to anonymous access

The legacy permission path allowed any request that no policy explicitly
denied whenever default-allow was on, which is the zero-config default.
That let an authenticated identity without table permissions reach table
resources owned by others. Restrict the fallback to requests with no
identity or the anonymous identity; authenticated callers must pass an
explicit action or policy check. Zero-config and anonymous access are
unchanged.

* s3tables: drop the no-op ListTableBuckets account gate

The top-level check passed the principal as its own owner, so it always
allowed. Per-bucket filtering in the loop is the real authority; remove
the dead gate and the now-unused locals.

* s3tables: derive the Iceberg catalog's default-allow from auth state

The Iceberg catalog reuses the S3 Tables Manager, which hardcoded
default-allow on. Authenticated callers were enforced only because the
identity struct happens to propagate into the handler; if it were ever
dropped, a secured catalog would fall open. Mirror the S3 port and set
the Manager's default-allow from the authenticator, so an authenticated
caller is enforced regardless. Shell and admin keep their own trusted
Manager. Regression test covers the struct, name-only, and admin paths.

* s3tables: drop redundant ACTION_ADMIN string conversion

ACTION_ADMIN is an untyped string constant, so the conversion is a no-op.

* s3tables: enforce name-only authenticated callers, add trusted bypass

defaultAllowFor treated a request with no identity object as anonymous,
but the Manager path forwards only the identity name (not the struct).
A name-only authenticated caller could therefore be misclassified as
anonymous and allowed under the open default. Treat a server-set identity
name as authenticated too, and add an explicit trusted flag for the local
shell/admin tooling that legitimately bypasses authorization.

* s3tables: trim verbose comments

weed/s3api/iceberg/server.go

@@ -46,6 +46,13 @@ type Server struct {
 // NewServer creates a new Iceberg REST Catalog server.
 func NewServer(filerClient FilerClient, authenticator S3Authenticator) *Server {
 	manager := s3tables.NewManager()
+	// Mirror the S3 port: fall open by default only when the gateway itself is
+	// open. With auth configured, an authenticated catalog caller must pass the
+	// normal permission check instead of being allowed because no policy denied
+	// it — even if the full identity struct ever fails to reach the handler.
+	if authenticator != nil {
+		manager.SetDefaultAllow(authenticator.DefaultAllow())
+	}
 	return &Server{
 		filerClient:   filerClient,
 		tablesManager: manager,

weed/s3api/s3tables/handler.go

@@ -48,6 +48,7 @@ type S3TablesHandler struct {
 	region        string
 	accountID     string
 	defaultAllow  bool // Whether to allow access by default (for zero-config IAM)
+	trusted       bool // Trusted local tooling (shell/admin) bypasses authorization
 	iamAuthorizer IAMAuthorizer
 }
 
@@ -79,6 +80,12 @@ func (h *S3TablesHandler) SetDefaultAllow(allow bool) {
 	h.defaultAllow = allow
 }
 
+// SetTrusted lets local tooling that talks to the filer directly (shell, admin
+// console) bypass authorization. HTTP-facing callers must not set it.
+func (h *S3TablesHandler) SetTrusted(trusted bool) {
+	h.trusted = trusted
+}
+
 // FilerClient interface for filer operations
 type FilerClient interface {
 	WithFilerClient(streamingMode bool, fn func(client filer_pb.SeaweedFilerClient) error) error
@@ -212,7 +219,11 @@ func (h *S3TablesHandler) getAccountID(r *http.Request) string {
 					idField := accountVal.FieldByName("Id")
 					if idField.IsValid() && idField.Kind() == reflect.String {
 						if principal := normalizePrincipalID(idField.String()); principal != "" {
-							return principal
+							// Account-less identities default to the admin account; only
+							// keep it for real admins, else use the unique identity name.
+							if principal != s3_constants.AccountAdminId || hasAdminAction(getIdentityActions(r)) {
+								return principal
+							}
 						}
 					}
 				}

weed/s3api/s3tables/handler_bucket_create.go

@@ -53,7 +53,7 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http
 		}
 		if !CheckPermissionWithContext("CreateTableBucket", principal, owner, "", "", &PolicyContext{
 			IdentityActions: identityActions,
-			DefaultAllow:    h.defaultAllow,
+			DefaultAllow:    h.defaultAllowFor(r),
 		}) {
 			h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table buckets")
 			return NewAuthError("CreateTableBucket", principal, "not authorized to create table buckets")

weed/s3api/s3tables/handler_bucket_get_list_delete.go

@@ -72,7 +72,7 @@ func (h *S3TablesHandler) handleGetTableBucket(w http.ResponseWriter, r *http.Re
 	if !CheckPermissionWithContext("GetTableBucket", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to get table bucket details")
 		return ErrAccessDenied
@@ -97,16 +97,9 @@ func (h *S3TablesHandler) handleListTableBuckets(w http.ResponseWriter, r *http.
 		return err
 	}
 
-	principal := h.getAccountID(r)
+	// No account-level gate: visibility is enforced per bucket below, so an
+	// owner can always list its own buckets and others are filtered out.
 	accountID := h.getAccountID(r)
-	identityActions := getIdentityActions(r)
-	if !CheckPermissionWithContext("ListTableBuckets", principal, accountID, "", "", &PolicyContext{
-		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
-	}) {
-		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to list table buckets")
-		return NewAuthError("ListTableBuckets", principal, "not authorized to list table buckets")
-	}
 
 	maxBuckets := req.MaxBuckets
 	if maxBuckets <= 0 {
@@ -200,7 +193,7 @@ func (h *S3TablesHandler) handleListTableBuckets(w http.ResponseWriter, r *http.
 				if !CheckPermissionWithContext("GetTableBucket", accountID, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 					TableBucketName: entry.Entry.Name,
 					IdentityActions: identityActions,
-					DefaultAllow:    h.defaultAllow,
+					DefaultAllow:    h.defaultAllowFor(r),
 				}) {
 					continue
 				}
@@ -303,7 +296,7 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http
 		if !CheckPermissionWithContext("DeleteTableBucket", principal, metadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 			TableBucketName: bucketName,
 			IdentityActions: identityActions,
-			DefaultAllow:    h.defaultAllow,
+			DefaultAllow:    h.defaultAllowFor(r),
 		}) {
 			return NewAuthError("DeleteTableBucket", principal, fmt.Sprintf("not authorized to delete bucket %s", bucketName))
 		}

weed/s3api/s3tables/handler_identity_test.go

@@ -16,7 +16,9 @@ type testIdentityAccount struct {
 }
 
 type testIdentity struct {
+	Name    string
 	Account *testIdentityAccount
+	Actions []string
 	Claims  map[string]interface{}
 }
 
@@ -112,6 +114,53 @@ func TestGetAccountIDFallsBackToAccountID(t *testing.T) {
 	assert.Equal(t, "my-account-id", h.getAccountID(req), "expected Account.Id to be returned when claims are missing")
 }
 
+func TestGetAccountIDNonAdminDoesNotInheritAdminAccount(t *testing.T) {
+	h := NewS3TablesHandler()
+	id := &testIdentity{
+		Account: &testIdentityAccount{Id: s3_constants.AccountAdminId},
+		Actions: []string{"Read", "List"},
+	}
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), id))
+	req = req.WithContext(s3_constants.SetIdentityNameInContext(req.Context(), "readonly"))
+
+	assert.Equal(t, "readonly", h.getAccountID(req), "a non-admin identity must not inherit the shared admin account")
+}
+
+func TestGetAccountIDAdminActionKeepsAdminAccount(t *testing.T) {
+	h := NewS3TablesHandler()
+	id := &testIdentity{
+		Account: &testIdentityAccount{Id: s3_constants.AccountAdminId},
+		Actions: []string{s3_constants.ACTION_ADMIN},
+	}
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	req = req.WithContext(s3_constants.SetIdentityInContext(req.Context(), id))
+	req = req.WithContext(s3_constants.SetIdentityNameInContext(req.Context(), "root"))
+
+	assert.Equal(t, s3_constants.AccountAdminId, h.getAccountID(req), "an admin identity keeps the admin account as principal")
+}
+
+func TestDefaultAllowForOnlyAppliesToUnauthenticatedOrAnonymous(t *testing.T) {
+	h := NewS3TablesHandler()
+	h.SetDefaultAllow(true)
+
+	noIdentity := httptest.NewRequest(http.MethodGet, "/", nil)
+	assert.True(t, h.defaultAllowFor(noIdentity), "zero-config requests with no identity keep the open default")
+
+	anon := httptest.NewRequest(http.MethodGet, "/", nil)
+	anon = anon.WithContext(s3_constants.SetIdentityInContext(anon.Context(),
+		&testIdentity{Name: s3_constants.AccountAnonymousId}))
+	assert.True(t, h.defaultAllowFor(anon), "anonymous requests keep the open default")
+
+	authed := httptest.NewRequest(http.MethodGet, "/", nil)
+	authed = authed.WithContext(s3_constants.SetIdentityInContext(authed.Context(),
+		&testIdentity{Name: "readonly", Account: &testIdentityAccount{Id: s3_constants.AccountAdminId}, Actions: []string{"Read"}}))
+	assert.False(t, h.defaultAllowFor(authed), "an authenticated identity must not benefit from the open default")
+
+	h.SetDefaultAllow(false)
+	assert.False(t, h.defaultAllowFor(noIdentity), "default-allow disabled is never open")
+}
+
 func TestGetAccountIDNormalizesAccountIDARN(t *testing.T) {
 	h := NewS3TablesHandler()
 	id := &testIdentity{

weed/s3api/s3tables/handler_namespace.go

@@ -118,7 +118,7 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
 		Namespace:       namespaceName,
 		TableBucketTags: bucketTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		glog.Infof("S3Tables: Permission denied for CreateNamespace - principal=%s, owner=%s", principal, bucketMetadata.OwnerAccountID)
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create namespace in this bucket")
@@ -259,7 +259,7 @@ func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Requ
 		Namespace:       namespaceName,
 		TableBucketTags: bucketTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, "namespace not found")
 		return ErrAccessDenied
@@ -346,7 +346,7 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re
 		TableBucketName: bucketName,
 		TableBucketTags: bucketTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusNotFound, ErrCodeNoSuchBucket, fmt.Sprintf("table bucket %s not found", bucketName))
 		return ErrAccessDenied
@@ -531,7 +531,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
 		Namespace:       namespaceName,
 		TableBucketTags: bucketTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, "namespace not found")
 		return ErrAccessDenied

weed/s3api/s3tables/handler_policy.go

@@ -94,7 +94,7 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
 	if !CheckPermissionWithContext("PutTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, "", bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to put table bucket policy")
 		return NewAuthError("PutTableBucketPolicy", principal, "not authorized to put table bucket policy")
@@ -172,7 +172,7 @@ func (h *S3TablesHandler) handleGetTableBucketPolicy(w http.ResponseWriter, r *h
 	if !CheckPermissionWithContext("GetTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, string(policy), bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to get table bucket policy")
 		return NewAuthError("GetTableBucketPolicy", principal, "not authorized to get table bucket policy")
@@ -248,7 +248,7 @@ func (h *S3TablesHandler) handleDeleteTableBucketPolicy(w http.ResponseWriter, r
 	if !CheckPermissionWithContext("DeleteTableBucketPolicy", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete table bucket policy")
 		return NewAuthError("DeleteTableBucketPolicy", principal, "not authorized to delete table bucket policy")
@@ -349,7 +349,7 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
 		Namespace:       namespaceName,
 		TableName:       tableName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to put table policy")
 		return NewAuthError("PutTablePolicy", principal, "not authorized to put table policy")
@@ -457,7 +457,7 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
 		Namespace:       namespaceName,
 		TableName:       tableName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to get table policy")
 		return NewAuthError("GetTablePolicy", principal, "not authorized to get table policy")
@@ -547,7 +547,7 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
 		Namespace:       namespaceName,
 		TableName:       tableName,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	}) {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete table policy")
 		return NewAuthError("DeleteTablePolicy", principal, "not authorized to delete table policy")
@@ -646,7 +646,7 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
 			TagKeys:         requestTagKeys,
 			ResourceTags:    existingTags,
 			IdentityActions: identityActions,
-			DefaultAllow:    h.defaultAllow,
+			DefaultAllow:    h.defaultAllowFor(r),
 		}) {
 			return NewAuthError("TagResource", principal, "not authorized to tag resource")
 		}
@@ -764,7 +764,7 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
 			TableBucketTags: bucketTags,
 			ResourceTags:    tags,
 			IdentityActions: identityActions,
-			DefaultAllow:    h.defaultAllow,
+			DefaultAllow:    h.defaultAllowFor(r),
 		}) {
 			return NewAuthError("ListTagsForResource", principal, "not authorized to list tags for resource")
 		}
@@ -872,7 +872,7 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
 			TagKeys:         req.TagKeys,
 			ResourceTags:    tags,
 			IdentityActions: identityActions,
-			DefaultAllow:    h.defaultAllow,
+			DefaultAllow:    h.defaultAllowFor(r),
 		}) {
 			return NewAuthError("UntagResource", principal, "not authorized to untag resource")
 		}

weed/s3api/s3tables/handler_table.go

@@ -143,7 +143,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		TagKeys:         mapKeys(req.Tags),
 		TableBucketTags: bucketTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	bucketAllowed := CheckPermissionWithContext("CreateTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
@@ -153,7 +153,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		TagKeys:         mapKeys(req.Tags),
 		TableBucketTags: bucketTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 
 	if !nsAllowed && !bucketAllowed {
@@ -386,7 +386,7 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
 		TableBucketTags: bucketTags,
 		ResourceTags:    tableTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	bucketAllowed := CheckPermissionWithContext("GetTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
@@ -395,7 +395,7 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
 		TableBucketTags: bucketTags,
 		ResourceTags:    tableTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 
 	if !tableAllowed && !bucketAllowed {
@@ -525,14 +525,14 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 				Namespace:       namespaceName,
 				TableBucketTags: bucketTags,
 				IdentityActions: identityActions,
-				DefaultAllow:    h.defaultAllow,
+				DefaultAllow:    h.defaultAllowFor(r),
 			})
 			bucketAllowed := CheckPermissionWithContext("ListTables", accountID, bucketMeta.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 				TableBucketName: bucketName,
 				Namespace:       namespaceName,
 				TableBucketTags: bucketTags,
 				IdentityActions: identityActions,
-				DefaultAllow:    h.defaultAllow,
+				DefaultAllow:    h.defaultAllowFor(r),
 			})
 			if !nsAllowed && !bucketAllowed {
 				return ErrAccessDenied
@@ -577,7 +577,7 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
 				TableBucketName: bucketName,
 				TableBucketTags: bucketTags,
 				IdentityActions: identityActions,
-				DefaultAllow:    h.defaultAllow,
+				DefaultAllow:    h.defaultAllowFor(r),
 			}) {
 				return ErrAccessDenied
 			}
@@ -916,7 +916,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 		TableBucketTags: bucketTags,
 		ResourceTags:    tableTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	bucketAllowed := CheckPermissionWithContext("DeleteTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
@@ -925,7 +925,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
 		TableBucketTags: bucketTags,
 		ResourceTags:    tableTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	if !tableAllowed && !bucketAllowed {
 		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to delete table")
@@ -1058,7 +1058,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		TableBucketTags: bucketTags,
 		ResourceTags:    tableTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 	bucketAllowed := CheckPermissionWithContext("UpdateTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
 		TableBucketName: bucketName,
@@ -1067,7 +1067,7 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 		TableBucketTags: bucketTags,
 		ResourceTags:    tableTags,
 		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllow,
+		DefaultAllow:    h.defaultAllowFor(r),
 	})
 
 	if !tableAllowed && !bucketAllowed {

weed/s3api/s3tables/iam.go

@@ -48,6 +48,24 @@ func (h *S3TablesHandler) shouldUseIAM(r *http.Request, identityActions, identit
 	return len(identityPolicyNames) > 0
 }
 
+// defaultAllowFor reports whether the open-by-default fallback applies: only for
+// trusted tooling or unauthenticated/anonymous access. An authenticated principal
+// must pass an explicit check.
+func (h *S3TablesHandler) defaultAllowFor(r *http.Request) bool {
+	if h.trusted {
+		return true
+	}
+	if !h.defaultAllow {
+		return false
+	}
+	// The Manager path forwards only the identity name, so a name alone (no
+	// identity object) still counts as an authenticated principal.
+	if s3_constants.GetIdentityFromContext(r) == nil && s3_constants.GetIdentityNameFromContext(r) == "" {
+		return true
+	}
+	return isAnonymousIdentity(r)
+}
+
 func isAnonymousIdentity(r *http.Request) bool {
 	val, ok := getIdentityStructValue(r)
 	if !ok {

weed/s3api/s3tables/manager.go

@@ -41,6 +41,11 @@ func (m *Manager) SetDefaultAllow(allow bool) {
 	m.handler.SetDefaultAllow(allow)
 }
 
+// SetTrusted lets trusted local tooling (shell, admin console) bypass authorization.
+func (m *Manager) SetTrusted(trusted bool) {
+	m.handler.SetTrusted(trusted)
+}
+
 // 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)

weed/s3api/s3tables/manager_test.go

@@ -0,0 +1,99 @@
+package s3tables
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var errFilerReached = errors.New("filer reached")
+
+// recordingFilerClient reports whether the handler reached the filer. The
+// CreateTableBucket handler authorizes before touching the filer, so "filer
+// reached" is a proxy for "authorization passed".
+type recordingFilerClient struct {
+	called bool
+}
+
+func (c *recordingFilerClient) WithFilerClient(streamingMode bool, fn func(client filer_pb.SeaweedFilerClient) error) error {
+	c.called = true
+	return errFilerReached
+}
+
+// The Manager path (used by the Iceberg catalog) enforces authorization for
+// authenticated callers and only falls open for trusted/zero-config access.
+func TestManagerCreateTableBucketAuthorization(t *testing.T) {
+	lowPriv := &testIdentity{
+		Name:    "alice",
+		Account: &testIdentityAccount{Id: s3_constants.AccountAdminId},
+		Actions: []string{"Read"},
+	}
+
+	cases := []struct {
+		name         string
+		defaultAllow bool
+		trusted      bool
+		ctx          context.Context
+		identity     string
+		wantFiler    bool // true => authorization passed (filer reached)
+	}{
+		{
+			name:         "authenticated identity struct is enforced",
+			defaultAllow: true,
+			ctx:          s3_constants.SetIdentityInContext(context.Background(), lowPriv),
+			identity:     "alice",
+			wantFiler:    false,
+		},
+		{
+			name:         "secured manager denies a name without struct",
+			defaultAllow: false,
+			ctx:          context.Background(),
+			identity:     "alice",
+			wantFiler:    false,
+		},
+		{
+			name:         "untrusted name without struct is enforced",
+			defaultAllow: true,
+			ctx:          context.Background(),
+			identity:     "alice",
+			wantFiler:    false,
+		},
+		{
+			name:         "trusted manager allows a name without struct",
+			defaultAllow: true,
+			trusted:      true,
+			ctx:          context.Background(),
+			identity:     "alice",
+			wantFiler:    true,
+		},
+		{
+			name:         "admin principal is allowed",
+			defaultAllow: false,
+			ctx:          context.Background(),
+			identity:     s3_constants.AccountAdminId,
+			wantFiler:    true,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			m := NewManager()
+			m.SetDefaultAllow(tc.defaultAllow)
+			m.SetTrusted(tc.trusted)
+			fc := &recordingFilerClient{}
+			err := m.Execute(tc.ctx, fc, "CreateTableBucket", &CreateTableBucketRequest{Name: "testbucket"}, nil, tc.identity)
+			require.Error(t, err) // denied (403) or the filer sentinel (reached); never nil here
+			assert.Equal(t, tc.wantFiler, fc.called, "filer reached should equal authorization passed")
+			if !tc.wantFiler {
+				var s3Err *S3TablesError
+				require.ErrorAs(t, err, &s3Err)
+				assert.Equal(t, ErrCodeAccessDenied, s3Err.Type)
+			}
+		})
+	}
+}

weed/s3api/s3tables/permissions.go

@@ -193,16 +193,14 @@ func hasIdentityPermission(operation string, ctx *PolicyContext) bool {
 	if !strings.Contains(operation, ":") {
 		fullAction = "s3tables:" + operation
 	}
+	if hasAdminAction(ctx.IdentityActions) {
+		return true
+	}
 	candidates := []string{operation, fullAction}
 	if ctx.TableBucketName != "" {
 		candidates = append(candidates, operation+":"+ctx.TableBucketName, fullAction+":"+ctx.TableBucketName)
 	}
 	for _, action := range ctx.IdentityActions {
-		// Legacy static identities may still use broad admin markers or s3 wildcards.
-		// s3:* is treated as s3tables:* so shared admin policies still permit table access.
-		if action == "*" || action == string(s3_constants.ACTION_ADMIN) || action == "s3:*" || action == "s3tables:*" {
-			return true
-		}
 		for _, candidate := range candidates {
 			if action == candidate {
 				return true
@@ -215,6 +213,18 @@ func hasIdentityPermission(operation string, ctx *PolicyContext) bool {
 	return false
 }
 
+// hasAdminAction reports whether the action list grants blanket admin access.
+// Legacy static identities may use broad markers or s3 wildcards; s3:* is treated
+// as s3tables:* so shared admin policies still permit table access.
+func hasAdminAction(actions []string) bool {
+	for _, action := range actions {
+		if action == "*" || action == s3_constants.ACTION_ADMIN || action == "s3:*" || action == "s3tables:*" {
+			return true
+		}
+	}
+	return false
+}
+
 // matchesPrincipal checks if the principal matches the statement's principal
 func matchesPrincipal(principalSpec interface{}, principal string) bool {
 	if principalSpec == nil {

weed/shell/s3tables_helpers.go

@@ -28,6 +28,8 @@ func executeS3Tables(commandEnv *CommandEnv, operation string, req interface{},
 	defer cancel()
 	return withFilerClient(commandEnv, func(client filer_pb.SeaweedFilerClient) error {
 		manager := s3tables.NewManager()
+		// The shell talks to the filer directly with no S3 auth, so it is trusted.
+		manager.SetTrusted(true)
 		mgrClient := s3tables.NewManagerClient(client)
 		return manager.Execute(ctx, mgrClient, operation, req, resp, accountID)
 	})