diff --git a/weed/s3api/iceberg/commit_helpers.go b/weed/s3api/iceberg/commit_helpers.go index 4caaaeb1a6..116ed85862 100644 --- a/weed/s3api/iceberg/commit_helpers.go +++ b/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()} +} diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index c40ee6b9e7..c31a6d5d5d 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/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 == "" { diff --git a/weed/s3api/iceberg/handlers_transaction.go b/weed/s3api/iceberg/handlers_transaction.go index 5eed534a87..6279cc4cc7 100644 --- a/weed/s3api/iceberg/handlers_transaction.go +++ b/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 == "" { diff --git a/weed/s3api/iceberg/handlers_view_update.go b/weed/s3api/iceberg/handlers_view_update.go index e7edc7db2f..60971a05fb 100644 --- a/weed/s3api/iceberg/handlers_view_update.go +++ b/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 diff --git a/weed/s3api/iceberg/iceberg_commit_authorize_test.go b/weed/s3api/iceberg/iceberg_commit_authorize_test.go new file mode 100644 index 0000000000..fc13c64dc9 --- /dev/null +++ b/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) + } + }) + } +}