diff --git a/.nextchanges/bundles/grants-before-dependents.md b/.nextchanges/bundles/grants-before-dependents.md new file mode 100644 index 00000000000..b3162ec8738 --- /dev/null +++ b/.nextchanges/bundles/grants-before-dependents.md @@ -0,0 +1 @@ +* Resources that depend on another resource are now deployed after that resource's `grants`, through an implicit `lifecycle.depends_on` entry, fixing intermittent `User does not have CREATE TABLE on Schema` errors when a pipeline's `run_as` principal is granted access to its schema in the same bundle. ([#7142](https://github.com/databricks/cli/pull/7142)) diff --git a/acceptance/bundle/migrate/command/grants/out.new_state.json b/acceptance/bundle/migrate/command/grants/out.new_state.json index 82b918f76d3..a0505879630 100644 --- a/acceptance/bundle/migrate/command/grants/out.new_state.json +++ b/acceptance/bundle/migrate/command/grants/out.new_state.json @@ -16,6 +16,10 @@ { "node": "resources.schemas.my_schema", "label": "${resources.schemas.my_schema.name}" + }, + { + "node": "resources.schemas.my_schema.grants", + "label": "depends_on" } ] }, @@ -81,6 +85,10 @@ { "node": "resources.schemas.my_schema", "label": "${resources.schemas.my_schema.name}" + }, + { + "node": "resources.schemas.my_schema.grants", + "label": "depends_on" } ] }, diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/databricks.yml.tmpl b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/databricks.yml.tmpl new file mode 100644 index 00000000000..f029e2d98a1 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/databricks.yml.tmpl @@ -0,0 +1,34 @@ +bundle: + name: schema-grants-pipeline-$UNIQUE_NAME + +resources: + catalogs: + pipeline_catalog: + name: catalog_grants_pipeline_$UNIQUE_NAME + grants: + - principal: $RUN_AS_SP + privileges: + - USE_CATALOG + + schemas: + src: + catalog_name: ${resources.catalogs.pipeline_catalog.name} + name: src + grants: + - principal: $RUN_AS_SP + privileges: + - USE_SCHEMA + - CREATE_TABLE + - CREATE_MATERIALIZED_VIEW + + pipelines: + load: + name: schema-grants-pipeline-$UNIQUE_NAME + catalog: ${resources.schemas.src.catalog_name} + schema: ${resources.schemas.src.name} + serverless: true + libraries: + - file: + path: ./transformation.py + run_as: + service_principal_name: $RUN_AS_SP diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml new file mode 100644 index 00000000000..a927a5fbc06 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/out.test.toml @@ -0,0 +1,2 @@ +Cloud = false +EnvMatrix.DMS = ["", "true"] diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/output.txt b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/output.txt new file mode 100644 index 00000000000..4e15f2629e2 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/output.txt @@ -0,0 +1,101 @@ + +=== The grants of a referenced catalog or schema become an implicit lifecycle.depends_on +>>> [CLI] bundle validate -o json +{ + "catalogs": { + "pipeline_catalog": null + }, + "pipelines": { + "load": [ + { + "resource": "schemas.src.grants" + } + ] + }, + "schemas": { + "src": [ + { + "resource": "catalogs.pipeline_catalog.grants" + } + ] + } +} + +=== The plan orders dependents after the grants +>>> [CLI] bundle plan -o json +{ + "resources.catalogs.pipeline_catalog": null, + "resources.catalogs.pipeline_catalog.grants": [ + { + "node": "resources.catalogs.pipeline_catalog", + "label": "${resources.catalogs.pipeline_catalog.id}" + } + ], + "resources.pipelines.load": [ + { + "node": "resources.schemas.src", + "label": "${resources.schemas.src.catalog_name}" + }, + { + "node": "resources.schemas.src", + "label": "${resources.schemas.src.name}" + }, + { + "node": "resources.schemas.src.grants", + "label": "depends_on" + } + ], + "resources.schemas.src": [ + { + "node": "resources.catalogs.pipeline_catalog", + "label": "${resources.catalogs.pipeline_catalog.name}" + }, + { + "node": "resources.catalogs.pipeline_catalog.grants", + "label": "depends_on" + } + ], + "resources.schemas.src.grants": [ + { + "node": "resources.schemas.src", + "label": "${resources.schemas.src.id}" + } + ] +} + +>>> [CLI] bundle deploy +Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/schema-grants-pipeline-[UNIQUE_NAME]/default/files... +Created catalogs.pipeline_catalog +Created catalogs.pipeline_catalog.grants +Created pipelines.load +Created schemas.src +Created schemas.src.grants +Files: 3 uploaded, 0 deleted +Resources: 5 created, 0 changed, 0 deleted, 0 unchanged + +>>> print_requests.py --summary //unity-catalog/catalogs //unity-catalog/schemas //unity-catalog/permissions //pipelines +POST /api/2.1/unity-catalog/catalogs +PATCH /api/2.1/unity-catalog/permissions/catalog/catalog_grants_pipeline_[UNIQUE_NAME] +POST /api/2.1/unity-catalog/schemas +PATCH /api/2.1/unity-catalog/permissions/schema/catalog_grants_pipeline_[UNIQUE_NAME].src +POST /api/2.0/pipelines + +>>> [CLI] bundle plan +Plan: 0 to add, 0 to change, 0 to delete, 5 unchanged + +>>> [CLI] bundle destroy --auto-approve +The following resources will be deleted: + delete resources.catalogs.pipeline_catalog + delete resources.pipelines.load + delete resources.schemas.src + +This action will result in the deletion of the following UC schemas. Any underlying data may be lost: + delete resources.schemas.src + +This action will result in the deletion of the following Lakeflow Spark Declarative Pipelines along with the +Streaming Tables (STs) and Materialized Views (MVs) managed by them. Set 'cascade_on_destroy: false' on a pipeline to retain datasets on pipeline deletion: + delete resources.pipelines.load + +All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/schema-grants-pipeline-[UNIQUE_NAME]/default + +Destroy: 3 deleted diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/script b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/script new file mode 100644 index 00000000000..392803e4258 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/script @@ -0,0 +1,22 @@ +# Creating the pipeline fails unless its run_as principal can already create tables in the +# schema, so the pipeline must be deployed after the grants it relies on. +# See https://github.com/databricks/cli/issues/4573 +export RUN_AS_SP="${TEST_SP_APPLICATION_ID:-6f1c2b3a-4d5e-4f60-8a7b-9c0d1e2f3a4b}" + +envsubst < databricks.yml.tmpl > databricks.yml + +cleanup() { + trace $CLI bundle destroy --auto-approve +} +trap cleanup EXIT + +title "The grants of a referenced catalog or schema become an implicit lifecycle.depends_on" +trace $CLI bundle validate -o json | jq '.resources | map_values(map_values(.lifecycle.depends_on))' + +title "The plan orders dependents after the grants" +trace $CLI bundle plan -o json | jq '.plan | map_values(.depends_on)' + +trace $CLI bundle deploy +trace print_requests.py --summary //unity-catalog/catalogs //unity-catalog/schemas //unity-catalog/permissions //pipelines + +trace $CLI bundle plan diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/test.toml b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/test.toml new file mode 100644 index 00000000000..10a7e8b41bb --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/test.toml @@ -0,0 +1,3 @@ +# Local only: on cloud the deploying identity lacks the servicePrincipal.user role on +# TEST_SP_APPLICATION_ID, so it cannot set that service principal as the pipeline's run_as. +Cloud = false diff --git a/acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py new file mode 100644 index 00000000000..7df869a15e7 --- /dev/null +++ b/acceptance/bundle/resources/grants/schemas/pipeline_run_as/transformation.py @@ -0,0 +1 @@ +print("Hello, World!") diff --git a/bundle/config/mutator/resourcemutator/capture_grants_dependencies.go b/bundle/config/mutator/resourcemutator/capture_grants_dependencies.go new file mode 100644 index 00000000000..00008db1360 --- /dev/null +++ b/bundle/config/mutator/resourcemutator/capture_grants_dependencies.go @@ -0,0 +1,172 @@ +package resourcemutator + +import ( + "context" + "fmt" + "maps" + "reflect" + "slices" + "strings" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/libs/diag" + "github.com/databricks/cli/libs/dyn" + "github.com/databricks/cli/libs/dyn/dynvar" +) + +type captureGrantsDependencies struct{} + +// CaptureGrantsDependencies makes a resource that depends on another resource also depend on +// that resource's grants, because creating some resources requires the grants to be in place: +// a pipeline cannot be created before its run_as principal has CREATE TABLE on the target +// schema. The dependency is added to lifecycle.depends_on, and skipped when it would create a +// cycle, e.g. when a schema grants privileges to the service principal of an app that refers +// to the schema. +func CaptureGrantsDependencies() bundle.Mutator { + return &captureGrantsDependencies{} +} + +func (m *captureGrantsDependencies) Name() string { + return "CaptureGrantsDependencies" +} + +func (m *captureGrantsDependencies) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics { + graph, err := dependencyGraph(b) + if err != nil { + return diag.FromErr(err) + } + + // Resources are visited in a fixed order because an added dependency can rule out a later + // one: with two of them forming a cycle together, the same one must be skipped on every run. + added := make(map[string][]string) + for _, node := range slices.Sorted(maps.Keys(graph)) { + // Grants and permissions have no lifecycle. + if strings.Contains(config.GetResourceTypeFromKey(node), ".") { + continue + } + + // The grants appended below are not visited: range evaluates the slice once. + for _, dep := range graph[node] { + grants := dep + ".grants" + if _, ok := graph[grants]; !ok || slices.Contains(graph[node], grants) || reaches(graph, grants, node) { + continue + } + graph[node] = append(graph[node], grants) + added[node] = append(added[node], grants) + } + } + + resources := dyn.NewPattern(dyn.Key("resources"), dyn.AnyKey(), dyn.AnyKey()) + err = b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) { + return dyn.MapByPattern(root, resources, func(p dyn.Path, v dyn.Value) (dyn.Value, error) { + grants := added[p.String()] + if len(grants) == 0 { + return v, nil + } + return appendDependsOn(v, grants) + }) + }) + return diag.FromErr(err) +} + +// dependencyGraph returns, for every resource and its grants or permissions, the nodes it depends +// on: those it refers to and those in its lifecycle.depends_on. Nodes are keyed as in the deployment +// plan, e.g. "resources.schemas.foo.grants"; a node without dependencies maps to nil. +func dependencyGraph(b *bundle.Bundle) (map[string][]string, error) { + graph := make(map[string][]string) + + err := dyn.WalkReadOnly(b.Config.Value(), func(p dyn.Path, v dyn.Value) error { + node, resourceType := config.GetNodeAndType(p) + if resourceType == "" { + return nil + } + + src := node.String() + if len(p) == len(node) { + // Grants and permissions without entries are not deployed, so nothing can depend on them. + if seq, ok := v.AsSequence(); len(node) == 4 && (!ok || len(seq) == 0) { + return nil + } + if _, ok := graph[src]; !ok { + graph[src] = nil + } + return nil + } + + ref, ok := dynvar.NewRef(v) + if !ok { + return nil + } + for _, r := range ref.References() { + path, err := dyn.NewPathFromString(r) + if err != nil { + return fmt.Errorf("parsing %q: %w", r, err) + } + if target, targetType := config.GetNodeAndType(path); targetType != "" { + graph[src] = append(graph[src], target.String()) + } + } + return nil + }) + if err != nil { + return nil, err + } + + for _, group := range b.Config.Resources.AllResources() { + for key, resource := range group.Resources { + // Python-generated configs can contain null resource entries. + if reflect.ValueOf(resource).IsNil() { + continue + } + src := "resources." + group.Description.PluralName + "." + key + for _, dep := range resource.GetLifecycle().GetDependsOn() { + // An entry that does not resolve is reported by validate.LifecycleDependsOn. + if target, err := b.Config.Resources.ResolveDependsOn(dep.Resource); err == nil { + graph[src] = append(graph[src], target) + } + } + } + } + return graph, nil +} + +// reaches reports whether from is to or depends on it, directly or through other nodes. +func reaches(graph map[string][]string, from, to string) bool { + visited := make(map[string]bool) + stack := []string{from} + for len(stack) > 0 { + n := stack[len(stack)-1] + stack = stack[:len(stack)-1] + if n == to { + return true + } + if visited[n] { + continue + } + visited[n] = true + stack = append(stack, graph[n]...) + } + return false +} + +// appendDependsOn appends lifecycle.depends_on entries for the given nodes to a resource. +func appendDependsOn(resource dyn.Value, nodes []string) (dyn.Value, error) { + lifecycle := dyn.GetValue(resource, "lifecycle") + if lifecycle.Kind() != dyn.KindMap { + lifecycle = dyn.V(dyn.NewMapping()) + } + + entries, _ := dyn.GetValue(lifecycle, "depends_on").AsSequence() + for _, node := range nodes { + entries = append(entries, dyn.V(map[string]dyn.Value{ + "resource": dyn.V(strings.TrimPrefix(node, "resources.")), + })) + } + + lifecycle, err := dyn.Set(lifecycle, "depends_on", dyn.V(entries)) + if err != nil { + return dyn.InvalidValue, err + } + return dyn.Set(resource, "lifecycle", lifecycle) +} diff --git a/bundle/config/mutator/resourcemutator/capture_grants_dependencies_test.go b/bundle/config/mutator/resourcemutator/capture_grants_dependencies_test.go new file mode 100644 index 00000000000..0cc50f327ad --- /dev/null +++ b/bundle/config/mutator/resourcemutator/capture_grants_dependencies_test.go @@ -0,0 +1,137 @@ +package resourcemutator + +import ( + "testing" + + "github.com/databricks/cli/bundle" + "github.com/databricks/cli/bundle/config" + "github.com/databricks/cli/bundle/config/resources" + "github.com/databricks/databricks-sdk-go/service/catalog" + "github.com/databricks/databricks-sdk-go/service/jobs" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestCaptureGrantsDependencies(t *testing.T) { + grants := func(principal string) []catalog.PrivilegeAssignment { + return []catalog.PrivilegeAssignment{{Principal: principal, Privileges: []catalog.Privilege{catalog.PrivilegeCreateTable}}} + } + schema := func(grants []catalog.PrivilegeAssignment) *resources.Schema { + return &resources.Schema{CatalogName: "main", Name: "src", Grants: grants} + } + // A pipeline that refers to the schema. + pipeline := &resources.Pipeline{ + Catalog: "${resources.schemas.src.catalog_name}", + Schema: "${resources.schemas.src.name}", + } + // An app that refers to the schema. + app := &resources.App{Name: "app", Description: "${resources.schemas.src.name}"} + + tests := []struct { + name string + resources config.Resources + expected map[string][]resources.DependsOn + }{ + { + name: "resource that refers to a resource with grants depends on the grants", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "resource without grants", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(nil)}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline}, + }, + expected: map[string][]resources.DependsOn{}, + }, + { + name: "each resource depends on the grants of the resource it refers to", + resources: config.Resources{ + Catalogs: map[string]*resources.Catalog{"main": {CreateCatalog: catalog.CreateCatalog{Name: "main"}, Grants: grants("sp")}}, + Schemas: map[string]*resources.Schema{"src": { + CreateSchema: catalog.CreateSchema{CatalogName: "${resources.catalogs.main.name}", Name: "src"}, + Grants: grants("sp"), + }}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline}, + }, + expected: map[string][]resources.DependsOn{ + "schemas.src": {{Resource: "catalogs.main.grants"}}, + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "grants that depend on the referring resource", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("${resources.apps.app.service_principal_client_id}"))}, + Apps: map[string]*resources.App{"app": app}, + Pipelines: map[string]*resources.Pipeline{"load": pipeline}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + { + name: "grants that transitively depend on the referring resource", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("${resources.jobs.job.run_as.service_principal_name}"))}, + Jobs: map[string]*resources.Job{"job": {JobSettings: jobs.JobSettings{Name: "${resources.apps.app.name}"}}}, + Apps: map[string]*resources.App{"app": app}, + }, + expected: map[string][]resources.DependsOn{}, + }, + { + name: "lifecycle.depends_on on a resource implies its grants", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Jobs: map[string]*resources.Job{"job": { + BaseResource: resources.BaseResource{Lifecycle: resources.Lifecycle{DependsOn: []resources.DependsOn{{Resource: "schemas.src"}}}}, + JobSettings: jobs.JobSettings{Name: "job"}, + }}, + }, + expected: map[string][]resources.DependsOn{ + "jobs.job": {{Resource: "schemas.src"}, {Resource: "schemas.src.grants"}}, + }, + }, + { + name: "grants already in lifecycle.depends_on", + resources: config.Resources{ + Schemas: map[string]*resources.Schema{"src": schema(grants("sp"))}, + Pipelines: map[string]*resources.Pipeline{"load": { + BaseResource: resources.BaseResource{Lifecycle: resources.Lifecycle{DependsOn: []resources.DependsOn{{Resource: "schemas.src.grants"}}}}, + CreatePipeline: pipeline.CreatePipeline, + }}, + }, + expected: map[string][]resources.DependsOn{ + "pipelines.load": {{Resource: "schemas.src.grants"}}, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + b := &bundle.Bundle{Config: config.Root{Resources: tc.resources}} + + // Normalization runs more than once, so applying the mutator again must not add anything. + for range 2 { + diags := bundle.Apply(t.Context(), b, CaptureGrantsDependencies()) + require.NoError(t, diags.Error()) + + actual := make(map[string][]resources.DependsOn) + for _, group := range b.Config.Resources.AllResources() { + for key, resource := range group.Resources { + if deps := resource.GetLifecycle().GetDependsOn(); len(deps) > 0 { + actual[group.Description.PluralName+"."+key] = deps + } + } + } + assert.Equal(t, tc.expected, actual) + } + }) + } +} diff --git a/bundle/config/mutator/resourcemutator/resource_mutator.go b/bundle/config/mutator/resourcemutator/resource_mutator.go index e00444cc773..35d8d9bbe8f 100644 --- a/bundle/config/mutator/resourcemutator/resource_mutator.go +++ b/bundle/config/mutator/resourcemutator/resource_mutator.go @@ -183,6 +183,10 @@ func applyNormalizeMutators(ctx context.Context, b *bundle.Bundle) { // Translates implicit schema and catalog references across all UC resources to explicit syntax to capture dependencies CaptureUCDependencies(), + // Reads (dynamic): resources.*.** (searches for ${resources.*} references), resources.*.*.grants, resources.*.*.lifecycle.depends_on + // Updates (dynamic): resources.*.*.lifecycle.depends_on (adds the grants of the resources a resource depends on) + CaptureGrantsDependencies(), + // Reads (dynamic): resources.dashboards.*.file_path // Updates (dynamic): resources.dashboards.*.serialized_dashboard // Drops (dynamic): resources.dashboards.*.file_path diff --git a/libs/testserver/pipelines.go b/libs/testserver/pipelines.go index 0aedc44c5d3..2cb479b849b 100644 --- a/libs/testserver/pipelines.go +++ b/libs/testserver/pipelines.go @@ -1,10 +1,13 @@ package testserver import ( + "cmp" "encoding/json" "fmt" + "slices" "strings" + "github.com/databricks/databricks-sdk-go/service/catalog" "github.com/databricks/databricks-sdk-go/service/pipelines" ) @@ -64,6 +67,18 @@ func (s *FakeWorkspace) PipelineCreate(req Request) Response { } r.Parameters = create.Parameters + // "schema" and "target" both name the target schema; only one of them can be set. + schemaName := spec.Catalog + "." + cmp.Or(spec.Schema, spec.Target) + if !s.runAsCanCreateTables(schemaName, create.RunAs) { + return Response{ + StatusCode: 403, + Body: map[string]string{ + "error_code": permissionDeniedErrorCode, + "message": fmt.Sprintf("User does not have CREATE TABLE on Schema '%s'.", schemaName), + }, + } + } + pipelineId := nextUUID() r.PipelineId = pipelineId r.CreatorUserName = "tester@databricks.com" @@ -88,6 +103,34 @@ func (s *FakeWorkspace) PipelineCreate(req Request) Response { } } +// runAsCanCreateTables reports whether the run_as principal of a new pipeline can create +// tables in its target schema, which the backend requires to create the pipeline. Only +// schemas created in this workspace are checked, because grants are tracked only for them. +func (s *FakeWorkspace) runAsCanCreateTables(schemaName string, runAs *pipelines.RunAs) bool { + var principal string + if runAs != nil { + principal = cmp.Or(runAs.ServicePrincipalName, runAs.UserName) + } + + schema, ok := s.Schemas[schemaName] + if principal == "" || !ok || principal == schema.Owner { + return true + } + + // Privileges granted on the catalog are inherited by its schemas. + for _, key := range []string{grantsKey("catalog", schema.CatalogName), grantsKey("schema", schemaName)} { + for _, assignment := range s.Grants[key] { + if assignment.Principal != principal { + continue + } + if slices.Contains(assignment.Privileges, catalog.PrivilegeCreateTable) || slices.Contains(assignment.Privileges, catalog.PrivilegeAllPrivileges) { + return true + } + } + } + return false +} + func setSpecDefaults(spec *pipelines.PipelineSpec, pipelineId string) { spec.Id = pipelineId // If the pipeline definition does not specify a catalog, it switches to Hive metastore mode diff --git a/libs/testserver/pipelines_test.go b/libs/testserver/pipelines_test.go index 39114714082..044f24d67b6 100644 --- a/libs/testserver/pipelines_test.go +++ b/libs/testserver/pipelines_test.go @@ -1,6 +1,7 @@ package testserver import ( + "strings" "testing" "github.com/databricks/databricks-sdk-go/service/pipelines" @@ -48,6 +49,99 @@ func TestPipelineCreate_AllowsSingleSegmentTargetSchemaName(t *testing.T) { assert.Equal(t, 0, response.StatusCode) } +func TestPipelineCreate_RequiresRunAsCreateTableOnSchema(t *testing.T) { + const pipeline = `{"name": "p", "catalog": "main", "schema": "src", "run_as": {"service_principal_name": "sp"}}` + + tests := []struct { + name string + createSchema bool + grants map[string]string + pipeline string + expectedStatusCode int + }{ + { + name: "no grants", + createSchema: true, + pipeline: pipeline, + expectedStatusCode: 403, + }, + { + name: "no grants with target", + createSchema: true, + pipeline: `{"name": "p", "catalog": "main", "target": "src", "run_as": {"service_principal_name": "sp"}}`, + expectedStatusCode: 403, + }, + { + name: "grants without CREATE_TABLE", + createSchema: true, + grants: map[string]string{"schema/main.src": "USE_SCHEMA"}, + pipeline: pipeline, + expectedStatusCode: 403, + }, + { + name: "CREATE_TABLE on schema", + createSchema: true, + grants: map[string]string{"schema/main.src": "CREATE_TABLE"}, + pipeline: pipeline, + }, + { + name: "ALL_PRIVILEGES on schema", + createSchema: true, + grants: map[string]string{"schema/main.src": "ALL_PRIVILEGES"}, + pipeline: pipeline, + }, + { + name: "CREATE_TABLE on catalog", + createSchema: true, + grants: map[string]string{"catalog/main": "CREATE_TABLE"}, + pipeline: pipeline, + }, + { + name: "run_as schema owner", + createSchema: true, + pipeline: `{"name": "p", "catalog": "main", "schema": "src", "run_as": {"user_name": "` + TestUser.UserName + `"}}`, + }, + { + name: "no run_as", + createSchema: true, + pipeline: `{"name": "p", "catalog": "main", "schema": "src"}`, + }, + { + name: "schema not created in workspace", + pipeline: pipeline, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // A user token, so that the schema owner is TestUser. + workspace := NewFakeWorkspace("http://test", UserNameTokenPrefix+"123") + + if tt.createSchema { + response := workspace.SchemasCreate(Request{Body: []byte(`{"catalog_name": "main", "name": "src"}`)}) + require.Equal(t, 0, response.StatusCode) + } + + for securable, privilege := range tt.grants { + securableType, fullName, _ := strings.Cut(securable, "/") + response := workspace.GrantsUpdate(Request{ + Body: []byte(`{"changes": [{"principal": "sp", "add": ["` + privilege + `"]}]}`), + }, securableType, fullName) + require.Equal(t, 0, response.StatusCode) + } + + response := workspace.PipelineCreate(Request{Body: []byte(tt.pipeline)}) + assert.Equal(t, tt.expectedStatusCode, response.StatusCode) + if tt.expectedStatusCode == 403 { + assert.Equal(t, map[string]string{ + "error_code": "PERMISSION_DENIED", + "message": "User does not have CREATE TABLE on Schema 'main.src'.", + }, response.Body) + } + }) + } +} + func TestPipelineStartUpdate_HandlesNonExistentPipeline(t *testing.T) { workspace := NewFakeWorkspace("http://test", "dbapi123")