Authorize an Iceberg table create before it writes (#10991)

* s3tables: share one CreateTable authorization gate

CreateTable and RegisterTable each carried their own copy of the name
validation, policy load and permission check. Fold them into
authorizeCreateTable, and expose it on the Manager for callers that write
into a table bucket before the table itself is registered.

Claude-Session: https://claude.ai/code/session_01QiJkka1T2NAWDWq4JQ8Vuy

* iceberg: authorize a table create before it writes

Stage-create returns before the S3Tables registration that authorizes a
create, and the plain create writes its metadata file before reaching it,
so a caller who may not create the table could still leave a staged
template, a marker and a v1.metadata.json in the target bucket - and get
vended credentials for a location of their choosing. Run the CreateTable
gate as soon as the table is known to be absent.

Claude-Session: https://claude.ai/code/session_01QiJkka1T2NAWDWq4JQ8Vuy

* iceberg: authorize a create-on-commit the same way

A commit against a table that does not exist creates it, writing the
metadata file first and only then reaching the registration that checks
the caller may create it. Denied callers saw a 500 for what is a 403.

Claude-Session: https://claude.ai/code/session_01QiJkka1T2NAWDWq4JQ8Vuy

* iceberg: pin that identity actions reach the create gate

The manager request is built from the caller's own context, so an identity
whose actions carry the permission still passes. Worth a test: a fresh
context here would silently deny every such caller.

Claude-Session: https://claude.ai/code/session_01QiJkka1T2NAWDWq4JQ8Vuy

weed/s3api/iceberg/handlers_commit.go

@@ -169,6 +169,12 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 					writeError(w, http.StatusNotFound, "NoSuchTableException", fmt.Sprintf("Table does not exist: %s", tableName))
 					return
 				}
+				// From here the commit creates the table, writing its metadata
+				// file before the create that authorizes it.
+				if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, tableName, identityName); authErr != nil {
+					writeManagerError(w, authErr)
+					return
+				}
 
 				for _, requirement := range req.Requirements {
 					validateAgainst := table.Metadata(nil)

weed/s3api/iceberg/handlers_table.go

@@ -119,6 +119,18 @@ func (s *Server) tablePathOccupied(ctx context.Context, bucketName, tablePath st
 	return occupied, err
 }
 
+// authorizeCreateTable checks the caller may create the table, for the create
+// paths that write to the bucket before the catalog registers it.
+func (s *Server) authorizeCreateTable(ctx context.Context, bucketARN string, namespace []string, tableName, identityName string) error {
+	return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return s.tablesManager.AuthorizeCreateTable(ctx, s3tables.NewManagerClient(client), &s3tables.CreateTableRequest{
+			TableBucketARN: bucketARN,
+			Namespace:      namespace,
+			Name:           tableName,
+		}, identityName)
+	})
+}
+
 // handleCreateTable creates a new table.
 func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 	vars := mux.Vars(r)
@@ -263,6 +275,14 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Both branches below write into the table bucket, and stage-create never
+	// reaches the registration that carries the authorization, so the caller has
+	// to pass the CreateTable gate here.
+	if authErr := s.authorizeCreateTable(r.Context(), bucketARN, namespace, tableName, identityName); authErr != nil {
+		writeManagerError(w, authErr)
+		return
+	}
+
 	// Stage-create persists metadata in the internal staged area and skips S3Tables registration.
 	if req.StageCreate {
 		stagedTablePath := stageCreateStagedTablePath(namespace, tableName, tableUUID)

weed/s3api/iceberg/iceberg_commit_updates_test.go

@@ -3,9 +3,14 @@ package iceberg
 import (
 	"encoding/json"
 	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"testing"
 
 	"github.com/apache/iceberg-go/table"
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
 )
 
@@ -101,3 +106,29 @@ func TestIsS3TablesConflict(t *testing.T) {
 		t.Fatalf("unexpected conflict for non-conflict error")
 	}
 }
+
+// A create-on-commit is a create, and it writes the new table's metadata file
+// before the registration that authorizes it.
+func TestCreateOnCommitDeniedBeforeAnyWrite(t *testing.T) {
+	const bucket = "warehouse"
+	fc := newMemFiler()
+	seedNamespace(fc, bucket, "finance", "alice")
+	s := NewServer(fc, nil)
+
+	r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/finance/tables/quarterly_reports",
+		strings.NewReader(`{"requirements":[{"type":"assert-create"}],"updates":[]}`))
+	r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": "finance", "table": "quarterly_reports"})
+	r = r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "mallory"))
+
+	w := httptest.NewRecorder()
+	s.handleUpdateTable(w, r)
+
+	if w.Code != http.StatusForbidden {
+		t.Errorf("status = %d, want %d (body: %s)", w.Code, http.StatusForbidden, w.Body.String())
+	}
+	for p, entry := range fc.entries {
+		if !entry.IsDirectory && strings.Contains(p, "quarterly_reports") {
+			t.Errorf("unauthorized caller wrote %s", p)
+		}
+	}
+}

weed/s3api/iceberg/iceberg_create_table_test.go

@@ -10,6 +10,8 @@ import (
 
 	"github.com/apache/iceberg-go"
 	"github.com/google/uuid"
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
 	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
 )
 
@@ -109,3 +111,90 @@ func TestBuildLoadTableResultNeverReturnsNilMetadata(t *testing.T) {
 		})
 	}
 }
+
+func newCreateTableRequest(t *testing.T, bucket, namespace, body, identity string) *http.Request {
+	t.Helper()
+	r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/"+namespace+"/tables", strings.NewReader(body))
+	r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": namespace})
+	return r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), identity))
+}
+
+// A caller who may not create the table must be refused before anything is
+// written: both branches of CreateTable write to the bucket, and stage-create
+// never reaches the registration that carries the authorization.
+func TestCreateTableDeniedBeforeAnyWrite(t *testing.T) {
+	const bucket = "warehouse"
+	for name, body := range map[string]string{
+		"stage-create": `{"name":"quarterly_reports","stage-create":true}`,
+		"create":       `{"name":"quarterly_reports"}`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			fc := newMemFiler()
+			seedNamespace(fc, bucket, "finance", "alice")
+			s := NewServer(fc, nil)
+
+			w := httptest.NewRecorder()
+			s.handleCreateTable(w, newCreateTableRequest(t, bucket, "finance", body, "mallory"))
+
+			if w.Code != http.StatusForbidden {
+				t.Errorf("status = %d, want %d (body: %s)", w.Code, http.StatusForbidden, w.Body.String())
+			}
+			for p, entry := range fc.entries {
+				if !entry.IsDirectory && strings.Contains(p, "quarterly_reports") {
+					t.Errorf("unauthorized caller wrote %s", p)
+				}
+			}
+		})
+	}
+}
+
+// The namespace owner still gets the staged metadata and marker stage-create
+// exists to leave behind.
+func TestStageCreateWritesStagedFilesForOwner(t *testing.T) {
+	const bucket = "warehouse"
+	fc := newMemFiler()
+	seedNamespace(fc, bucket, "finance", "alice")
+	s := NewServer(fc, nil)
+
+	w := httptest.NewRecorder()
+	s.handleCreateTable(w, newCreateTableRequest(t, bucket, "finance", `{"name":"quarterly_reports","stage-create":true}`, "alice"))
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusOK, w.Body.String())
+	}
+
+	staged := 0
+	for p, entry := range fc.entries {
+		if !entry.IsDirectory && strings.Contains(p, stageCreateMarkerDirName) {
+			staged++
+		}
+	}
+	if staged != 2 {
+		t.Fatalf("staged files = %d, want the metadata file and its marker", staged)
+	}
+	if _, ok := fc.entries[s3tables.GetTablePath(bucket, "finance", "quarterly_reports")]; ok {
+		t.Error("stage-create registered the table")
+	}
+}
+
+// A caller allowed by the actions on its own identity, rather than by ownership
+// or a resource policy, has to survive the hop into the s3tables manager: the
+// gate must not refuse what the create behind it would allow.
+func TestCreateTableAllowedByIdentityActions(t *testing.T) {
+	const bucket = "warehouse"
+	fc := newMemFiler()
+	seedNamespace(fc, bucket, "finance", "alice")
+	s := NewServer(fc, nil)
+
+	r := newCreateTableRequest(t, bucket, "finance", `{"name":"quarterly_reports","stage-create":true}`, "mallory")
+	r = r.WithContext(s3_constants.SetIdentityInContext(r.Context(), &struct {
+		Name    string
+		Actions []string
+	}{Name: "mallory", Actions: []string{"s3tables:CreateTable"}}))
+
+	w := httptest.NewRecorder()
+	s.handleCreateTable(w, r)
+
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusOK, w.Body.String())
+	}
+}

weed/s3api/iceberg/iceberg_create_view_test.go

@@ -5,9 +5,11 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"path"
+	"sort"
 	"strings"
 	"testing"
 
@@ -76,6 +78,35 @@ func (m *memFiler) UpdateEntry(_ context.Context, in *filer_pb.UpdateEntryReques
 	return &filer_pb.UpdateEntryResponse{}, nil
 }
 
+func (m *memFiler) ListEntries(_ context.Context, in *filer_pb.ListEntriesRequest, _ ...grpc.CallOption) (filer_pb.SeaweedFiler_ListEntriesClient, error) {
+	var names []string
+	for p := range m.entries {
+		if path.Dir(p) == in.Directory {
+			names = append(names, p)
+		}
+	}
+	sort.Strings(names)
+	stream := &memListEntries{}
+	for _, name := range names {
+		stream.entries = append(stream.entries, m.entries[name])
+	}
+	return stream, nil
+}
+
+type memListEntries struct {
+	grpc.ClientStream
+	entries []*filer_pb.Entry
+}
+
+func (m *memListEntries) Recv() (*filer_pb.ListEntriesResponse, error) {
+	if len(m.entries) == 0 {
+		return nil, io.EOF
+	}
+	entry := m.entries[0]
+	m.entries = m.entries[1:]
+	return &filer_pb.ListEntriesResponse{Entry: entry}, nil
+}
+
 func newCreateViewRequest(t *testing.T, namespace, name, sql string) *http.Request {
 	t.Helper()
 	schema := iceberg.NewSchemaWithIdentifiers(0, nil,
@@ -95,12 +126,11 @@ func newCreateViewRequest(t *testing.T, namespace, name, sql string) *http.Reque
 	return r
 }
 
-// seedNamespace registers a bucket and namespace so the s3tables existence and
-// auth-context lookups pass; ownership is irrelevant since the admin principal
-// is always allowed.
-func seedNamespace(fc *memFiler, bucket, namespace string) {
+// seedNamespace registers a bucket and namespace owned by owner so the s3tables
+// existence and auth-context lookups pass.
+func seedNamespace(fc *memFiler, bucket, namespace, owner string) {
 	fc.seed(s3tables.GetTableBucketPath(bucket), &filer_pb.Entry{Name: bucket, IsDirectory: true})
-	meta, _ := json.Marshal(map[string]any{"namespace": []string{namespace}, "ownerAccountId": s3_constants.AccountAdminId})
+	meta, _ := json.Marshal(map[string]any{"namespace": []string{namespace}, "ownerAccountId": owner})
 	fc.seed(s3tables.GetNamespacePath(bucket, namespace), &filer_pb.Entry{
 		Name:        namespace,
 		IsDirectory: true,
@@ -130,7 +160,7 @@ func TestCreateViewMissingNamespaceReturns404(t *testing.T) {
 func TestCreateViewTagsEntryAsView(t *testing.T) {
 	const bucket = "warehouse"
 	fc := newMemFiler()
-	seedNamespace(fc, bucket, "ns")
+	seedNamespace(fc, bucket, "ns", s3_constants.AccountAdminId)
 	s := NewServer(fc, nil)
 
 	w := httptest.NewRecorder()
@@ -154,7 +184,7 @@ func TestCreateViewTagsEntryAsView(t *testing.T) {
 func TestCreateViewDuplicateDoesNotClobberMetadata(t *testing.T) {
 	const bucket = "warehouse"
 	fc := newMemFiler()
-	seedNamespace(fc, bucket, "ns")
+	seedNamespace(fc, bucket, "ns", s3_constants.AccountAdminId)
 	s := NewServer(fc, nil)
 
 	w := httptest.NewRecorder()
@@ -185,7 +215,7 @@ func TestCreateViewDuplicateDoesNotClobberMetadata(t *testing.T) {
 func TestCreateViewRollsBackEntryWhenMetadataWriteFails(t *testing.T) {
 	const bucket = "warehouse"
 	fc := newMemFiler()
-	seedNamespace(fc, bucket, "ns")
+	seedNamespace(fc, bucket, "ns", s3_constants.AccountAdminId)
 	fc.failFileCreate = errors.New("disk full")
 	s := NewServer(fc, nil)
 

weed/s3api/s3tables/handler_table.go

@@ -23,22 +23,6 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		return err
 	}
 
-	if req.TableBucketARN == "" {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN is required")
-		return fmt.Errorf("tableBucketARN is required")
-	}
-
-	namespaceName, err := validateNamespace(req.Namespace)
-	if err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
-
-	if req.Name == "" {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "name is required")
-		return fmt.Errorf("name is required")
-	}
-
 	if req.Format == "" {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "format is required")
 		return fmt.Errorf("format is required")
@@ -50,117 +34,21 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		return fmt.Errorf("invalid format")
 	}
 
-	bucketName, err := parseBucketNameFromARN(req.TableBucketARN)
+	target, err := h.authorizeCreateTable(w, r, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags)
 	if err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
-
-	// Validate table name
-	tableName, err := validateTableName(req.Name)
-	if err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
-
-	// Check if namespace exists
-	namespacePath := GetNamespacePath(bucketName, namespaceName)
-	namespaceMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName)
-	if err != nil {
-		if errors.Is(err, filer_pb.ErrNotFound) {
-			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName))
-		} else {
-			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err))
-		}
-		return err
-	}
-
-	// Authorize table creation using policy framework (namespace + bucket policies)
-	accountID := h.getAccountID(r)
-	bucketPath := GetTableBucketPath(bucketName)
-	namespacePolicy := ""
-	bucketPolicy := ""
-	bucketTags := map[string]string{}
-	var data []byte
-	var bucketMetadata tableBucketMetadata
-
-	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		// Fetch bucket metadata to use correct owner for bucket policy evaluation
-		data, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
-		if err == nil {
-			if err := json.Unmarshal(data, &bucketMetadata); err != nil {
-				return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
-			}
-		} else if !errors.Is(err, ErrAttributeNotFound) {
-			return fmt.Errorf("failed to fetch bucket metadata: %v", err)
-		}
-
-		// Fetch namespace policy if it exists
-		policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
-		if err == nil {
-			namespacePolicy = string(policyData)
-		} else if !errors.Is(err, ErrAttributeNotFound) {
-			return fmt.Errorf("failed to fetch namespace policy: %v", err)
-		}
-
-		// Fetch bucket policy if it exists
-		policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-		if err == nil {
-			bucketPolicy = string(policyData)
-		} else if !errors.Is(err, ErrAttributeNotFound) {
-			return fmt.Errorf("failed to fetch bucket policy: %v", err)
-		}
-		if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
-			return err
-		} else if tags != nil {
-			bucketTags = tags
-		}
-
-		return nil
-	})
-
-	if err != nil {
-		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to fetch policies: %v", err))
 		return err
 	}
+	bucketName, namespaceName, tableName := target.bucketName, target.namespaceName, target.tableName
 
 	// A bucket declares the format it holds, and a table of another format would
 	// be invisible to the catalog serving it. A bucket made before the
 	// declaration existed has none, and keeps taking anything.
-	if bucketMetadata.Format != "" && bucketMetadata.Format != req.Format {
-		message := fmt.Sprintf("table bucket %s holds %s tables", bucketName, bucketMetadata.Format)
+	if target.bucketFormat != "" && target.bucketFormat != req.Format {
+		message := fmt.Sprintf("table bucket %s holds %s tables", bucketName, target.bucketFormat)
 		h.writeError(w, http.StatusConflict, ErrCodeConflict, message)
 		return fmt.Errorf("%s", message)
 	}
 
-	bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
-	identityActions := getIdentityActions(r)
-	nsAllowed := CheckPermissionWithContext("CreateTable", accountID, namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN, &PolicyContext{
-		TableBucketName: bucketName,
-		Namespace:       namespaceName,
-		TableName:       tableName,
-		RequestTags:     req.Tags,
-		TagKeys:         mapKeys(req.Tags),
-		TableBucketTags: bucketTags,
-		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllowFor(r),
-	})
-	bucketAllowed := CheckPermissionWithContext("CreateTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
-		TableBucketName: bucketName,
-		Namespace:       namespaceName,
-		TableName:       tableName,
-		RequestTags:     req.Tags,
-		TagKeys:         mapKeys(req.Tags),
-		TableBucketTags: bucketTags,
-		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllowFor(r),
-	})
-
-	if !nsAllowed && !bucketAllowed {
-		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in this namespace")
-		return ErrAccessDenied
-	}
-
 	tablePath := GetTablePath(bucketName, namespaceName, tableName)
 
 	// Check if a table or view already exists at this name. Names are unique
@@ -221,7 +109,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
 		Format:           req.Format,
 		CreatedAt:        now,
 		ModifiedAt:       now,
-		OwnerAccountID:   namespaceMetadata.OwnerAccountID, // Inherit namespace owner for consistency
+		OwnerAccountID:   target.ownerAccountID, // Inherit namespace owner for consistency
 		VersionToken:     versionToken,
 		MetadataVersion:  max(req.MetadataVersion, 1),
 		MetadataLocation: req.MetadataLocation,
@@ -328,117 +216,16 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req
 		return err
 	}
 
-	if req.TableBucketARN == "" {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN is required")
-		return fmt.Errorf("tableBucketARN is required")
-	}
-
-	namespaceName, err := validateNamespace(req.Namespace)
-	if err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
-
-	if req.Name == "" {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "name is required")
-		return fmt.Errorf("name is required")
-	}
-
 	if req.MetadataLocation == "" {
 		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "metadataLocation is required")
 		return fmt.Errorf("metadataLocation is required")
 	}
 
-	bucketName, err := parseBucketNameFromARN(req.TableBucketARN)
+	target, err := h.authorizeCreateTable(w, r, filerClient, req.TableBucketARN, req.Namespace, req.Name, nil)
 	if err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
 		return err
 	}
-
-	tableName, err := validateTableName(req.Name)
-	if err != nil {
-		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
-		return err
-	}
-
-	// Namespace must exist.
-	namespacePath := GetNamespacePath(bucketName, namespaceName)
-	namespaceMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName)
-	if err != nil {
-		if errors.Is(err, filer_pb.ErrNotFound) {
-			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName))
-		} else {
-			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err))
-		}
-		return err
-	}
-
-	// Authorize using policy framework (namespace + bucket policies).
-	accountID := h.getAccountID(r)
-	bucketPath := GetTableBucketPath(bucketName)
-	namespacePolicy := ""
-	bucketPolicy := ""
-	bucketTags := map[string]string{}
-	var bucketMetadata tableBucketMetadata
-
-	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
-		if err == nil {
-			if err := json.Unmarshal(data, &bucketMetadata); err != nil {
-				return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
-			}
-		} else if !errors.Is(err, ErrAttributeNotFound) {
-			return fmt.Errorf("failed to fetch bucket metadata: %v", err)
-		}
-
-		policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
-		if err == nil {
-			namespacePolicy = string(policyData)
-		} else if !errors.Is(err, ErrAttributeNotFound) {
-			return fmt.Errorf("failed to fetch namespace policy: %v", err)
-		}
-
-		policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
-		if err == nil {
-			bucketPolicy = string(policyData)
-		} else if !errors.Is(err, ErrAttributeNotFound) {
-			return fmt.Errorf("failed to fetch bucket policy: %v", err)
-		}
-		if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
-			return err
-		} else if tags != nil {
-			bucketTags = tags
-		}
-
-		return nil
-	})
-	if err != nil {
-		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to fetch policies: %v", err))
-		return err
-	}
-
-	bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
-	identityActions := getIdentityActions(r)
-	nsAllowed := CheckPermissionWithContext("CreateTable", accountID, namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN, &PolicyContext{
-		TableBucketName: bucketName,
-		Namespace:       namespaceName,
-		TableName:       tableName,
-		TableBucketTags: bucketTags,
-		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllowFor(r),
-	})
-	bucketAllowed := CheckPermissionWithContext("CreateTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{
-		TableBucketName: bucketName,
-		Namespace:       namespaceName,
-		TableName:       tableName,
-		TableBucketTags: bucketTags,
-		IdentityActions: identityActions,
-		DefaultAllow:    h.defaultAllowFor(r),
-	})
-	if !nsAllowed && !bucketAllowed {
-		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to register table in this namespace")
-		return ErrAccessDenied
-	}
+	bucketName, namespaceName, tableName := target.bucketName, target.namespaceName, target.tableName
 
 	tablePath := GetTablePath(bucketName, namespaceName, tableName)
 
@@ -467,7 +254,7 @@ func (h *S3TablesHandler) handleRegisterTable(w http.ResponseWriter, r *http.Req
 		Format:           FormatIceberg,
 		CreatedAt:        now,
 		ModifiedAt:       now,
-		OwnerAccountID:   namespaceMetadata.OwnerAccountID,
+		OwnerAccountID:   target.ownerAccountID,
 		VersionToken:     versionToken,
 		MetadataVersion:  metadataVersionFromLocation(req.MetadataLocation),
 		MetadataLocation: req.MetadataLocation,
@@ -1808,3 +1595,135 @@ func (h *S3TablesHandler) handleUpdateTable(w http.ResponseWriter, r *http.Reque
 	})
 	return nil
 }
+
+// createTableTarget is the namespace a create or register resolved to, once the
+// caller has been authorized to put a table there.
+type createTableTarget struct {
+	bucketName     string
+	namespaceName  string
+	tableName      string
+	bucketFormat   string
+	ownerAccountID string
+}
+
+// authorizeCreateTable validates the names a create names and checks the caller
+// may create a table in that namespace, writing the error response itself.
+// Deferred creates (Iceberg stage-create) write into the table bucket before any
+// table is registered, so they run this same gate first through
+// Manager.AuthorizeCreateTable.
+func (h *S3TablesHandler) authorizeCreateTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient, tableBucketARN string, namespace []string, name string, requestTags map[string]string) (*createTableTarget, error) {
+	if tableBucketARN == "" {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "tableBucketARN is required")
+		return nil, fmt.Errorf("tableBucketARN is required")
+	}
+
+	namespaceName, err := validateNamespace(namespace)
+	if err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return nil, err
+	}
+
+	if name == "" {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, "name is required")
+		return nil, fmt.Errorf("name is required")
+	}
+
+	bucketName, err := parseBucketNameFromARN(tableBucketARN)
+	if err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return nil, err
+	}
+
+	tableName, err := validateTableName(name)
+	if err != nil {
+		h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error())
+		return nil, err
+	}
+
+	// Check if namespace exists
+	namespacePath := GetNamespacePath(bucketName, namespaceName)
+	namespaceMetadata, err := h.loadNamespaceMetadata(r.Context(), filerClient, bucketName, namespaceName)
+	if err != nil {
+		if errors.Is(err, filer_pb.ErrNotFound) {
+			h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", namespaceName))
+		} else {
+			h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check namespace: %v", err))
+		}
+		return nil, err
+	}
+
+	// Authorize table creation using policy framework (namespace + bucket policies)
+	accountID := h.getAccountID(r)
+	bucketPath := GetTableBucketPath(bucketName)
+	namespacePolicy := ""
+	bucketPolicy := ""
+	bucketTags := map[string]string{}
+	var bucketMetadata tableBucketMetadata
+
+	err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		// Fetch bucket metadata to use correct owner for bucket policy evaluation
+		data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
+		if err == nil {
+			if err := json.Unmarshal(data, &bucketMetadata); err != nil {
+				return fmt.Errorf("failed to unmarshal bucket metadata: %w", err)
+			}
+		} else if !errors.Is(err, ErrAttributeNotFound) {
+			return fmt.Errorf("failed to fetch bucket metadata: %v", err)
+		}
+
+		// Fetch namespace policy if it exists
+		policyData, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyPolicy)
+		if err == nil {
+			namespacePolicy = string(policyData)
+		} else if !errors.Is(err, ErrAttributeNotFound) {
+			return fmt.Errorf("failed to fetch namespace policy: %v", err)
+		}
+
+		// Fetch bucket policy if it exists
+		policyData, err = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
+		if err == nil {
+			bucketPolicy = string(policyData)
+		} else if !errors.Is(err, ErrAttributeNotFound) {
+			return fmt.Errorf("failed to fetch bucket policy: %v", err)
+		}
+		if tags, err := h.readTags(r.Context(), client, bucketPath); err != nil {
+			return err
+		} else if tags != nil {
+			bucketTags = tags
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to fetch policies: %v", err))
+		return nil, err
+	}
+
+	bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName)
+	identityActions := getIdentityActions(r)
+	policyContext := &PolicyContext{
+		TableBucketName: bucketName,
+		Namespace:       namespaceName,
+		TableName:       tableName,
+		RequestTags:     requestTags,
+		TagKeys:         mapKeys(requestTags),
+		TableBucketTags: bucketTags,
+		IdentityActions: identityActions,
+		DefaultAllow:    h.defaultAllowFor(r),
+	}
+	nsAllowed := CheckPermissionWithContext("CreateTable", accountID, namespaceMetadata.OwnerAccountID, namespacePolicy, bucketARN, policyContext)
+	bucketAllowed := CheckPermissionWithContext("CreateTable", accountID, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, policyContext)
+	if !nsAllowed && !bucketAllowed {
+		h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in this namespace")
+		return nil, ErrAccessDenied
+	}
+
+	return &createTableTarget{
+		bucketName:     bucketName,
+		namespaceName:  namespaceName,
+		tableName:      tableName,
+		bucketFormat:   bucketMetadata.Format,
+		ownerAccountID: namespaceMetadata.OwnerAccountID,
+	}, nil
+}

weed/s3api/s3tables/manager.go

@@ -52,19 +52,47 @@ func (m *Manager) Execute(ctx context.Context, filerClient FilerClient, operatio
 	if err != nil {
 		return err
 	}
-	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, "/", bytes.NewReader(body))
+	httpReq, err := newManagerRequest(ctx, operation, body, identity)
 	if err != nil {
 		return err
 	}
+	recorder := httptest.NewRecorder()
+	m.handler.HandleRequest(recorder, httpReq, filerClient)
+	return decodeS3TablesHTTPResponse(recorder, resp)
+}
+
+// AuthorizeCreateTable checks that identity may create the table the request
+// describes, without creating anything. A deferred create (Iceberg
+// stage-create) writes into the table bucket long before it registers the
+// table, so it passes this gate first.
+func (m *Manager) AuthorizeCreateTable(ctx context.Context, filerClient FilerClient, req *CreateTableRequest, identity string) error {
+	httpReq, err := newManagerRequest(ctx, "CreateTable", nil, identity)
+	if err != nil {
+		return err
+	}
+	recorder := httptest.NewRecorder()
+	_, authErr := m.handler.authorizeCreateTable(recorder, httpReq, filerClient, req.TableBucketARN, req.Namespace, req.Name, req.Tags)
+	if authErr == nil {
+		return nil
+	}
+	if decoded := decodeS3TablesHTTPResponse(recorder, nil); decoded != nil {
+		return decoded
+	}
+	return authErr
+}
+
+func newManagerRequest(ctx context.Context, operation string, body []byte, identity string) (*http.Request, error) {
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, "/", bytes.NewReader(body))
+	if err != nil {
+		return nil, err
+	}
 	httpReq.Header.Set("Content-Type", "application/x-amz-json-1.1")
 	httpReq.Header.Set("X-Amz-Target", "S3Tables."+operation)
 	if identity != "" {
 		httpReq.Header.Set(s3_constants.AmzAccountId, identity)
 		httpReq = httpReq.WithContext(s3_constants.SetIdentityNameInContext(httpReq.Context(), identity))
 	}
-	recorder := httptest.NewRecorder()
-	m.handler.HandleRequest(recorder, httpReq, filerClient)
-	return decodeS3TablesHTTPResponse(recorder, resp)
+	return httpReq, nil
 }
 
 func decodeS3TablesHTTPResponse(recorder *httptest.ResponseRecorder, resp interface{}) error {