iceberg: authorize a commit before staging its metadata file (#11891)

* iceberg: authorize a commit before staging its metadata file

Both commit paths (CommitTable and CommitTransaction) write the new
vN.metadata.json into the table's metadata/ directory and only then reach
UpdateTable, whose authorization refuses a caller that may read the table but
not commit to it. The refused commit then deletes the staged file on the way
out, so such a caller could make the server write and delete a file under the
table's prefix on every attempt. The same shape was closed for creates in
#10991.

Both paths now ask Manager.AuthorizeUpdateTable right after loading the table
and answer 403 ForbiddenException before anything is staged. UpdateTable still
decides under the policies in force when the pointer flips, so a policy changed
mid-commit is caught as before.

Test: TestCommitDeniedBeforeStagingMetadata records every file the handlers
create; on master both the single-table and the transaction commit by a
GetTable-only identity write v2.metadata.json before answering 403, with the
change nothing is written. TestCommitAllowedStagesMetadata is the positive
control (the owner, and an identity whose own actions include UpdateTable).

* iceberg: authorize a view commit before staging its metadata

handleUpdateView has the same write-before-authorize gap the table commit
paths had: it stages the new metadata file and only reaches the UpdateView
authorization at the pointer flip, so a caller who may only load the view
could still make the server write and delete files under it. Gate it on
AuthorizeUpdateView, mirroring authorizeCommit.

* iceberg: seed committed view metadata in the denied-commit test

Without stored view metadata the request fails on the missing-metadata
check before it could stage a file, so the zero-write assertion could
not catch a write-before-denial regression. Seed the same metadata the
allowed-path test uses so the denied request would otherwise stage.

---------

Co-authored-by: Javier de la Torre <jatorre@lakehousebox.com>
Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>

weed/s3api/iceberg/commit_helpers.go

@@ -252,3 +252,59 @@ func (s *Server) finalizeCreateOnCommit(ctx context.Context, input createOnCommi
 		Metadata:         newMetadata,
 	}, nil
 }
+
+// authorizeCommit asks the S3 Tables layer, before anything is written, whether
+// the caller may update the table. Both commit paths stage the new metadata
+// file in the table's directory before UpdateTable authorizes the pointer
+// flip, so without this gate a caller who may only read the table could still
+// make the server write and delete files under it. UpdateTable decides again
+// when the pointer flips.
+func (s *Server) authorizeCommit(ctx context.Context, bucketARN string, namespace []string, tableName, identityName string) *icebergRequestError {
+	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return s.tablesManager.AuthorizeUpdateTable(ctx, s3tables.NewManagerClient(client), &s3tables.UpdateTableRequest{
+			TableBucketARN: bucketARN,
+			Namespace:      namespace,
+			Name:           tableName,
+		}, identityName)
+	})
+	if err == nil {
+		return nil
+	}
+	var tableErr *s3tables.S3TablesError
+	if errors.As(err, &tableErr) {
+		switch tableErr.Type {
+		case s3tables.ErrCodeAccessDenied:
+			return &icebergRequestError{http.StatusForbidden, "ForbiddenException", fmt.Sprintf("Not authorized to commit to table %s", tableName)}
+		case s3tables.ErrCodeNoSuchTable, s3tables.ErrCodeNoSuchNamespace:
+			return &icebergRequestError{http.StatusNotFound, "NoSuchTableException", fmt.Sprintf("Table does not exist: %s", tableName)}
+		}
+	}
+	glog.V(1).Infof("Iceberg: commit authorization for %s failed: %v", tableName, err)
+	return &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to authorize commit: " + err.Error()}
+}
+
+// authorizeViewCommit is authorizeCommit for a view, whose update path stages
+// its metadata file before UpdateView authorizes the pointer flip.
+func (s *Server) authorizeViewCommit(ctx context.Context, bucketARN string, namespace []string, viewName, identityName string) *icebergRequestError {
+	err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		return s.tablesManager.AuthorizeUpdateView(ctx, s3tables.NewManagerClient(client), &s3tables.UpdateViewRequest{
+			TableBucketARN: bucketARN,
+			Namespace:      namespace,
+			Name:           viewName,
+		}, identityName)
+	})
+	if err == nil {
+		return nil
+	}
+	var tableErr *s3tables.S3TablesError
+	if errors.As(err, &tableErr) {
+		switch tableErr.Type {
+		case s3tables.ErrCodeAccessDenied:
+			return &icebergRequestError{http.StatusForbidden, "ForbiddenException", fmt.Sprintf("Not authorized to commit to view %s", viewName)}
+		case s3tables.ErrCodeNoSuchView, s3tables.ErrCodeNoSuchNamespace:
+			return &icebergRequestError{http.StatusNotFound, "NoSuchViewException", fmt.Sprintf("View does not exist: %s", viewName)}
+		}
+	}
+	glog.V(1).Infof("Iceberg: view commit authorization for %s failed: %v", viewName, err)
+	return &icebergRequestError{http.StatusInternalServerError, "InternalServerError", "Failed to authorize commit: " + err.Error()}
+}

weed/s3api/iceberg/handlers_commit.go

@@ -242,6 +242,10 @@ func (s *Server) handleUpdateTable(w http.ResponseWriter, r *http.Request) {
 			writeManagerError(w, err)
 			return
 		}
+		if reqErr := s.authorizeCommit(r.Context(), bucketARN, namespace, tableName, identityName); reqErr != nil {
+			writeError(w, reqErr.status, reqErr.errType, reqErr.message)
+			return
+		}
 
 		location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
 		if location == "" {

weed/s3api/iceberg/handlers_transaction.go

@@ -140,6 +140,9 @@ func (s *Server) prepareTableCommit(ctx context.Context, bucketName, bucketARN,
 		glog.V(1).Infof("Iceberg: CommitTransaction GetTable error: %v", err)
 		return nil, &icebergRequestError{http.StatusInternalServerError, "InternalServerError", err.Error()}
 	}
+	if reqErr := s.authorizeCommit(ctx, bucketARN, namespace, tableName, identityName); reqErr != nil {
+		return nil, reqErr
+	}
 
 	location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
 	if location == "" {

weed/s3api/iceberg/handlers_view_update.go

@@ -48,6 +48,10 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) {
 			writeManagerError(w, err)
 			return
 		}
+		if reqErr := s.authorizeViewCommit(r.Context(), bucketARN, namespace, viewName, identityName); reqErr != nil {
+			writeError(w, reqErr.status, reqErr.errType, reqErr.message)
+			return
+		}
 		if getResp.Metadata == nil || len(getResp.Metadata.FullMetadata) == 0 {
 			writeError(w, http.StatusInternalServerError, "InternalServerError", "view has no metadata")
 			return

weed/s3api/iceberg/iceberg_commit_authorize_test.go

@@ -0,0 +1,237 @@
+package iceberg
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"path"
+	"strings"
+	"testing"
+
+	"github.com/apache/iceberg-go"
+	"github.com/apache/iceberg-go/table"
+	"github.com/apache/iceberg-go/view"
+	"github.com/gorilla/mux"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
+	"google.golang.org/grpc"
+)
+
+// writeLogFiler records every file the handlers create, including files they
+// delete again before answering, which the final entry map cannot show.
+type writeLogFiler struct {
+	*memFiler
+	created []string
+}
+
+func (f *writeLogFiler) WithFilerClient(_ bool, fn func(client filer_pb.SeaweedFilerClient) error) error {
+	return fn(f)
+}
+
+func (f *writeLogFiler) CreateEntry(ctx context.Context, in *filer_pb.CreateEntryRequest, opts ...grpc.CallOption) (*filer_pb.CreateEntryResponse, error) {
+	if !in.Entry.IsDirectory {
+		f.created = append(f.created, path.Join(in.Directory, in.Entry.Name))
+	}
+	return f.memFiler.CreateEntry(ctx, in, opts...)
+}
+
+// seedOwnedTable registers an existing table owned by owner.
+func seedOwnedTable(fc *memFiler, bucket, namespace, tableName, owner string) {
+	internal := map[string]any{
+		"name":             tableName,
+		"namespace":        namespace,
+		"format":           "ICEBERG",
+		"ownerAccountId":   owner,
+		"versionToken":     "v1",
+		"metadataVersion":  1,
+		"metadataLocation": "s3://" + bucket + "/" + namespace + "/" + tableName + "/metadata/v1.metadata.json",
+		"metadata": s3tables.TableMetadata{
+			Iceberg: &s3tables.IcebergMetadata{TableUUID: "00000000-0000-0000-0000-000000000001"},
+		},
+	}
+	metaBytes, _ := json.Marshal(internal)
+	fc.seed(s3tables.GetTablePath(bucket, namespace, tableName), &filer_pb.Entry{
+		Name:        tableName,
+		IsDirectory: true,
+		Extended:    map[string][]byte{s3tables.ExtendedKeyMetadata: metaBytes},
+	})
+}
+
+// seedOwnedView registers an existing view owned by owner.
+func seedOwnedView(fc *memFiler, bucket, namespace, viewName, owner string, fullMetadata []byte) {
+	internal := map[string]any{
+		"name":             viewName,
+		"namespace":        namespace,
+		"format":           "ICEBERG",
+		"ownerAccountId":   owner,
+		"versionToken":     "v1",
+		"metadataVersion":  1,
+		"metadataLocation": "s3://" + bucket + "/" + namespace + "/" + viewName + "/metadata/v1.metadata.json",
+	}
+	if len(fullMetadata) > 0 {
+		internal["metadata"] = s3tables.TableMetadata{FullMetadata: fullMetadata}
+	}
+	metaBytes, _ := json.Marshal(internal)
+	fc.seed(s3tables.GetTablePath(bucket, namespace, viewName), &filer_pb.Entry{
+		Name:        viewName,
+		IsDirectory: true,
+		Extended: map[string][]byte{
+			s3tables.ExtendedKeyMetadata:  metaBytes,
+			s3tables.ExtendedKeyEntryType: []byte(s3tables.EntryTypeView),
+		},
+	})
+}
+
+// withActions gives the request an identity allowed only the listed actions.
+func withActions(r *http.Request, name string, actions ...string) *http.Request {
+	ctx := s3_constants.SetIdentityNameInContext(r.Context(), name)
+	return r.WithContext(s3_constants.SetIdentityInContext(ctx, &struct {
+		Name    string
+		Actions []string
+	}{Name: name, Actions: actions}))
+}
+
+func viewMetadataJSON(t *testing.T, location string) []byte {
+	t.Helper()
+	schema := iceberg.NewSchemaWithIdentifiers(0, nil,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true})
+	ver, err := view.NewVersionFromSQL(1, schema.ID, "SELECT 1", table.Identifier{"finance"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	md, err := view.NewMetadata(ver, schema, location, iceberg.Properties{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	fullMetadata, err := json.Marshal(md)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return fullMetadata
+}
+
+// A caller that may load a table but not commit to it must be refused before
+// the commit stages its metadata file in the table's directory. The commit
+// paths used to write that file, reach the UpdateTable authorization, and
+// delete the file again on the way out.
+func TestCommitDeniedBeforeStagingMetadata(t *testing.T) {
+	const bucket = "warehouse"
+	cases := map[string]func(s *Server) *httptest.ResponseRecorder{
+		"commit-table": func(s *Server) *httptest.ResponseRecorder {
+			r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/finance/tables/quarterly_reports",
+				strings.NewReader(`{"requirements":[],"updates":[{"action":"set-properties","updates":{"k":"v"}}]}`))
+			r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": "finance", "table": "quarterly_reports"})
+			w := httptest.NewRecorder()
+			s.handleUpdateTable(w, withActions(r, "reader", "s3tables:GetTable"))
+			return w
+		},
+		"commit-transaction": func(s *Server) *httptest.ResponseRecorder {
+			body := `{"table-changes":[{"identifier":{"namespace":["finance"],"name":"quarterly_reports"},"requirements":[],"updates":[{"action":"set-properties","updates":{"k":"v"}}]}]}`
+			r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/transactions/commit", strings.NewReader(body))
+			r = mux.SetURLVars(r, map[string]string{"prefix": bucket})
+			w := httptest.NewRecorder()
+			s.handleCommitTransaction(w, withActions(r, "reader", "s3tables:GetTable"))
+			return w
+		},
+		"commit-view": func(s *Server) *httptest.ResponseRecorder {
+			r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/finance/views/quarterly_view",
+				strings.NewReader(`{"requirements":[],"updates":[{"action":"set-properties","updates":{"k":"v"}}]}`))
+			r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": "finance", "view": "quarterly_view"})
+			w := httptest.NewRecorder()
+			s.handleUpdateView(w, withActions(r, "reader", "s3tables:GetView"))
+			return w
+		},
+	}
+	for name, run := range cases {
+		t.Run(name, func(t *testing.T) {
+			fc := &writeLogFiler{memFiler: newMemFiler()}
+			seedNamespace(fc.memFiler, bucket, "finance", "alice")
+			seedOwnedTable(fc.memFiler, bucket, "finance", "quarterly_reports", "alice")
+			seedOwnedView(fc.memFiler, bucket, "finance", "quarterly_view", "alice", viewMetadataJSON(t, "s3://"+bucket+"/finance/quarterly_view"))
+			s := NewServer(fc, nil)
+
+			w := run(s)
+
+			if w.Code != http.StatusForbidden {
+				t.Errorf("status = %d, want %d (body: %s)", w.Code, http.StatusForbidden, w.Body.String())
+			}
+			if len(fc.created) != 0 {
+				t.Errorf("refused commit wrote %v", fc.created)
+			}
+		})
+	}
+}
+
+// The owner, and a caller whose own identity carries UpdateTable, still
+// stage their metadata file and land the commit.
+func TestCommitAllowedStagesMetadata(t *testing.T) {
+	const bucket = "warehouse"
+	callers := map[string]func(r *http.Request) *http.Request{
+		"owner": func(r *http.Request) *http.Request {
+			return r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice"))
+		},
+		"identity-actions": func(r *http.Request) *http.Request {
+			return withActions(r, "writer", "s3tables:GetTable", "s3tables:UpdateTable")
+		},
+	}
+	for name, as := range callers {
+		t.Run(name, func(t *testing.T) {
+			fc := &writeLogFiler{memFiler: newMemFiler()}
+			seedNamespace(fc.memFiler, bucket, "finance", "alice")
+			seedOwnedTable(fc.memFiler, bucket, "finance", "quarterly_reports", "alice")
+			s := NewServer(fc, nil)
+
+			r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/finance/tables/quarterly_reports",
+				strings.NewReader(`{"requirements":[],"updates":[{"action":"set-properties","updates":{"k":"v"}}]}`))
+			r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": "finance", "table": "quarterly_reports"})
+			w := httptest.NewRecorder()
+			s.handleUpdateTable(w, as(r))
+
+			if w.Code != http.StatusOK {
+				t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusOK, w.Body.String())
+			}
+			if len(fc.created) != 1 || !strings.HasSuffix(fc.created[0], ".metadata.json") {
+				t.Fatalf("created = %v, want one metadata file", fc.created)
+			}
+		})
+	}
+}
+
+// Same staging gate on the view commit path: the owner, and a caller whose own
+// identity carries UpdateView, stage a metadata file and land the commit.
+func TestViewCommitAllowedStagesMetadata(t *testing.T) {
+	const bucket = "warehouse"
+	fullMetadata := viewMetadataJSON(t, "s3://"+bucket+"/finance/quarterly_view")
+
+	callers := map[string]func(r *http.Request) *http.Request{
+		"owner": func(r *http.Request) *http.Request {
+			return r.WithContext(s3_constants.SetIdentityNameInContext(r.Context(), "alice"))
+		},
+		"identity-actions": func(r *http.Request) *http.Request {
+			return withActions(r, "writer", "s3tables:GetView", "s3tables:UpdateView")
+		},
+	}
+	for name, as := range callers {
+		t.Run(name, func(t *testing.T) {
+			fc := &writeLogFiler{memFiler: newMemFiler()}
+			seedNamespace(fc.memFiler, bucket, "finance", "alice")
+			seedOwnedView(fc.memFiler, bucket, "finance", "quarterly_view", "alice", fullMetadata)
+			s := NewServer(fc, nil)
+
+			r := httptest.NewRequest(http.MethodPost, "/v1/"+bucket+"/namespaces/finance/views/quarterly_view",
+				strings.NewReader(`{"requirements":[],"updates":[{"action":"set-properties","updates":{"k":"v"}}]}`))
+			r = mux.SetURLVars(r, map[string]string{"prefix": bucket, "namespace": "finance", "view": "quarterly_view"})
+			w := httptest.NewRecorder()
+			s.handleUpdateView(w, as(r))
+
+			if w.Code != http.StatusOK {
+				t.Fatalf("status = %d, want %d (body: %s)", w.Code, http.StatusOK, w.Body.String())
+			}
+			if len(fc.created) != 1 || !strings.HasSuffix(fc.created[0], ".metadata.json") {
+				t.Fatalf("created = %v, want one metadata file", fc.created)
+			}
+		})
+	}
+}