diff --git a/weed/s3api/iceberg/handlers_commit.go b/weed/s3api/iceberg/handlers_commit.go index e6922465e..b8780860e 100644 --- a/weed/s3api/iceberg/handlers_commit.go +++ b/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) diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index c31d73151..855870ac4 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/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) diff --git a/weed/s3api/iceberg/iceberg_commit_updates_test.go b/weed/s3api/iceberg/iceberg_commit_updates_test.go index 96ed7e402..9811b5e73 100644 --- a/weed/s3api/iceberg/iceberg_commit_updates_test.go +++ b/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) + } + } +} diff --git a/weed/s3api/iceberg/iceberg_create_table_test.go b/weed/s3api/iceberg/iceberg_create_table_test.go index ea09e4252..bd083f811 100644 --- a/weed/s3api/iceberg/iceberg_create_table_test.go +++ b/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()) + } +} diff --git a/weed/s3api/iceberg/iceberg_create_view_test.go b/weed/s3api/iceberg/iceberg_create_view_test.go index bc7c6a74e..7f5e641f0 100644 --- a/weed/s3api/iceberg/iceberg_create_view_test.go +++ b/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) diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 987a97347..4d92a2ae8 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/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 +} diff --git a/weed/s3api/s3tables/manager.go b/weed/s3api/s3tables/manager.go index 10d7856ff..33dba70ae 100644 --- a/weed/s3api/s3tables/manager.go +++ b/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 {