From 9e4b24ced3a8985ef3c37186d807417b6d374a25 Mon Sep 17 00:00:00 2001 From: Mateo Hernandez Date: Tue, 3 Nov 2026 15:22:28 -0300 Subject: [PATCH 1/2] feat(databricks): add unity catalog schema sync CXH-2722. Schemas hang from the catalog, keyed {metastore_id}::{catalog}.{schema}, with the catalog privilege set minus BROWSE, CREATE_SCHEMA and USE_CATALOG, plus a read-only owner. information_schema is skipped. Grant and revoke use the same echo-confirmed idempotency as the catalog. The tree is unchanged from 433e33c5, the state reviewed and approved on #72. Co-Authored-By: Claude Opus 5 --- README.md | 1 + baton_capabilities.json | 79 +++ docs/connector.mdx | 20 +- docs/docs-info.md | 5 +- pkg/connector/catalog.go | 71 +-- pkg/connector/catalog_test.go | 2 +- pkg/connector/connector.go | 3 +- pkg/connector/connector_test.go | 2 +- pkg/connector/metastore_helpers.go | 120 +++++ pkg/connector/raw_id_test.go | 7 + pkg/connector/resource_types.go | 8 + pkg/connector/schema.go | 476 +++++++++++++++++ pkg/connector/schema_test.go | 796 +++++++++++++++++++++++++++++ pkg/connector/securable_helpers.go | 48 +- pkg/databricks/models.go | 45 +- pkg/databricks/unity_catalog.go | 96 +++- pkg/databricks/vars.go | 5 + 17 files changed, 1709 insertions(+), 75 deletions(-) create mode 100644 pkg/connector/schema.go create mode 100644 pkg/connector/schema_test.go diff --git a/README.md b/README.md index 12b80715..85bae015 100644 --- a/README.md +++ b/README.md @@ -222,6 +222,7 @@ account ├── service principal └── metastore └── catalog + └── schema ``` A metastore exposes its privileges as entitlements (`CREATE_CATALOG`, diff --git a/baton_capabilities.json b/baton_capabilities.json index 6a663099..9633f1b3 100644 --- a/baton_capabilities.json +++ b/baton_capabilities.json @@ -255,6 +255,85 @@ ] } }, + { + "resourceType": { + "id": "schema", + "displayName": "Schema", + "traits": [ + "TRAIT_APP" + ], + "annotations": [ + { + "@type": "type.googleapis.com/c1.connector.v2.OptInRequired" + }, + { + "@type": "type.googleapis.com/c1.connector.v2.CapabilityPermissions", + "permissions": [ + { + "permission": "unity-catalog" + }, + { + "permission": "scim" + }, + { + "permission": "access-management" + }, + { + "permission": "provisioning" + }, + { + "permission": "Account admin" + }, + { + "permission": "Workspace admin" + }, + { + "permission": "USE_CATALOG" + }, + { + "permission": "USE_SCHEMA" + } + ] + }, + { + "@type": "type.googleapis.com/c1.connector.v2.SkipEntitlements" + } + ] + }, + "capabilities": [ + "CAPABILITY_SYNC", + "CAPABILITY_PROVISION" + ], + "permissions": { + "permissions": [ + { + "permission": "unity-catalog" + }, + { + "permission": "scim" + }, + { + "permission": "access-management" + }, + { + "permission": "provisioning" + }, + { + "permission": "Account admin" + }, + { + "permission": "Workspace admin" + }, + { + "permission": "USE_CATALOG" + }, + { + "permission": "USE_SCHEMA" + } + ] + }, + "optInRequired": true + }, { "resourceType": { "id": "service_principal", diff --git a/docs/connector.mdx b/docs/connector.mdx index ef9be667..cfb7b05f 100644 --- a/docs/connector.mdx +++ b/docs/connector.mdx @@ -15,6 +15,7 @@ sidebarTitle: "Databricks" | Groups | | | | Metastores | | | | Roles | | | +| Schemas | | | | Service principals | | | | Users | | | | Workspaces | | | @@ -29,19 +30,19 @@ Revoking the workspace **member** entitlement removes the principal's entire wor [This connector syncs non-human identities](/product/admin/nhi) and displays them on the **Identities overview** dashboard. -## Metastore and catalog access in C1 +## Metastore, catalog and schema access in C1 The connector syncs Databricks [Unity Catalog](https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges) -metastores and the catalogs in them as C1 resources. Each privilege Databricks +metastores, the catalogs in them, and the schemas in those catalogs as C1 resources. Each privilege Databricks records on one of them (`CREATE_CATALOG` on a metastore, `USE_CATALOG` or -`SELECT` on a catalog, and so on) becomes an entitlement in C1, so you can +`SELECT` on a catalog, `USE_SCHEMA` on a schema, and so on) becomes an entitlement in C1, so you can review who holds it, request it, and revoke it for users, groups and service principals. A grant held by a group is expandable so C1 can resolve the group's members as holders. Ownership is shown as a read-only `owner` entitlement — it cannot be granted or revoked through C1. -The Metastores and Catalogs resource types are turned off by default. Turn on the ones you want from the connector's configuration page in C1. Catalogs appear under their metastore and use the same OAuth credentials; no additional configuration fields are needed. +The Metastores, Catalogs and Schemas resource types are turned off by default. Turn on the ones you want from the connector's configuration page in C1. Catalogs appear under their metastore and schemas under their catalog, and all of them use the same OAuth credentials; no additional configuration fields are needed. @@ -51,6 +52,7 @@ The Metastores and Catalogs resource types are turned off by default. Turn on th - **MANAGE** and **ALL_PRIVILEGES** are not offered on a metastore, because Databricks rejects both at that level. - A catalog is synced when any running workspace in scope attached to its metastore lists it, and the same catalog name is kept once per metastore. A catalog that only excluded workspaces can reach is not synced. - The built-in `system` catalog and catalogs Databricks identifies as `SYSTEM_CATALOG` are excluded regardless of ownership. This includes `samples` when Databricks reports that type. +- `information_schema` is skipped. - If a catalog becomes unreachable, access requests can fail until the connector can distinguish deletion from a binding or visibility change. Catalog ownership or **MANAGE** alone cannot prove deletion; that check requires the **Metastore Admin**. - Unity Catalog tags are not synced. @@ -202,6 +204,16 @@ For a discovered catalog that is not excluded by the system-catalog rule above, The connector asks every running workspace in scope attached to the metastore, so a catalog bound to one of those workspaces is included. A catalog the service principal cannot see on any of them is not, and ConductorOne removes it and its grants. +### Let the service principal list every schema + +Skip this section if you are not syncing schemas, or if the service principal owns the metastore or the catalog. + +Workspace **Admin** does not list every schema, except when the catalog is owned by the workspace admins group (`_workspace_admins__`) and the service principal is a direct Admin of that workspace. On any other catalog the service principal does not own, Databricks returns only the schemas it owns or can use, with HTTP 200. To list every schema, the service principal must own the metastore, own the catalog, be that direct Admin, or hold **MANAGE**, **USE SCHEMA**, or **ALL PRIVILEGES** on the catalog. + + +A catalog whose schemas cannot all be listed fails the sync. `system` and `samples` are not a special case of that listing: the catalog sync leaves them out, when their privilege list cannot be proven complete, before their schemas are read. + + **Done.** Here's the set of credentials you'll need when setting up the connector: - Account ID diff --git a/docs/docs-info.md b/docs/docs-info.md index 0d71529b..64160900 100644 --- a/docs/docs-info.md +++ b/docs/docs-info.md @@ -60,9 +60,10 @@ While developing the connector, please fill out this form. This information is n - **Ownership is read-only.** It is shown and cannot be granted or revoked. An owner that is not an account identity (`System user`, `_workspace_admins_…`) is not shown. - **No `MANAGE` or `ALL_PRIVILEGES` on a metastore.** Databricks rejects both, so they are not offered. - **Read through one running workspace.** Privileges are not served on the account plane. The connector reads each metastore through the first running workspace in scope where it can see every privilege. With no running workspace in scope, the metastore is not synced and ConductorOne removes it. A workspace that is still provisioning fails the sync until it is `RUNNING`. An unreadable assignment list fails the sync and ConductorOne keeps the metastore. -- **`workspaces` scopes metastores.** An allowlist that names no workspace on the account fails at credential validation, before any sync. +- **`workspaces` scopes metastores.** An allowlist that names no workspace on the account fails at credential validation, before any sync. A name that is also in `databricks-exclude-workspaces` is reported separately from a name the account does not have. - **Catalogs are opt-in.** The resource type is off until it is turned on from the connector's configuration page in C1. - **Catalog discovery visits every attached workspace in scope.** The connector asks each running workspace attached to the metastore, including for `ISOLATED` catalogs bound to only some workspaces. Each catalog name is kept once per metastore. A catalog bound only to workspaces the sync excludes is not listed, and ConductorOne removes it with its grants. - **System catalogs are excluded regardless of ownership.** The built-in `system` catalog and any catalog with `CatalogType == "SYSTEM_CATALOG"` are excluded, including `samples` when it has that type. - **Seeing a catalog needs more than `MANAGE`.** The list holds the catalogs the service principal owns or holds `USE_CATALOG` on, or every catalog when it is the metastore admin. `MANAGE` alone reads a catalog's privileges but does not make it appear. -- **Not synced yet:** schemas, tables, volumes, functions, registered models, external locations, storage credentials, connections, shares, recipients, and tags. +- **Built-in catalogs.** `system` and any `SYSTEM_CATALOG`, including `samples`, are synced while the service principal owns the metastore. When it does not, and the privilege list cannot be proven complete, those catalogs are left out. Databricks refuses `MANAGE` on them. Leaving them out after they were synced removes them in ConductorOne. +- **Not synced yet:** tables, volumes, functions, registered models, external locations, storage credentials, connections, shares, recipients, and tags. diff --git a/pkg/connector/catalog.go b/pkg/connector/catalog.go index f4597738..36eba554 100644 --- a/pkg/connector/catalog.go +++ b/pkg/connector/catalog.go @@ -41,7 +41,8 @@ const ( errorCodeMetastoreGone = "METASTORE_DOES_NOT_EXIST" // systemCatalog is the built-in catalog Databricks puts in every metastore. - // It is excluded independently of the connector's ownership. + // samples is the other one; both are SYSTEM_CATALOG. Both are excluded + // independently of the connector's ownership. systemCatalog = "system" // workspaceAdminsGroupPrefix starts the name of the system group Databricks @@ -186,7 +187,10 @@ func catalogResource( nil, rs.WithResourceProfile(profile), rs.WithParentResourceID(parent), - rs.WithAnnotation(&v2.RawId{Id: resourceID}), + rs.WithAnnotation( + &v2.ChildResourceType{ResourceTypeId: schemaResourceType.Id}, + &v2.RawId{Id: resourceID}, + ), ) } @@ -474,7 +478,13 @@ func catalogGrantsAreTrustworthy( } } - return holdsCatalogManage(ctx, client, trust, annos) + manages, err := holdsPrivilegeOn(ctx, client, trust, databricks.SecurableCatalog, trust.fullName, + managePrivilege, []string{errorCodeCatalogGone}, annos) + if errors.Is(err, errSecurableGone) { + return false, errCatalogGone + } + + return manages, err } func catalogAdminWorkspaceID(owner string) (string, bool) { @@ -493,61 +503,6 @@ func catalogAdminWorkspaceID(owner string) (string, bool) { return id, true } -// holdsCatalogManage asks the permissions endpoint for the connector's own -// effective privileges on the catalog. The principal filter is effective, not -// literal: a row can be the MANAGE a group holds and this principal inherits, so -// only the privileges are read and never the principal field. The read bypasses -// the cache, because a remembered page would decide this call with the access of -// an earlier one. -// -// ALL_PRIVILEGES is not enough. Databricks leaves MANAGE out of it, so its holder -// still reads a partial list. -// https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges -func holdsCatalogManage( - ctx context.Context, - client *databricks.Client, - trust securableTrust, - annos *annotations.Annotations, -) (bool, error) { - if trust.ownPrincipal == "" || trust.workspace == "" { - return false, nil - } - - var manages bool - rateLimit, err := client.ForEachUncachedPermissionsPage(ctx, trust.workspace, databricks.SecurableCatalog, trust.fullName, trust.ownPrincipal, - func(page []databricks.PrivilegeAssignment) (bool, error) { - for _, assignment := range page { - if slices.Contains(assignment.Privileges, privilegeManage) { - manages = true - - return true, nil - } - } - - return false, nil - }) - noteRateLimit(annos, rateLimit) - if err != nil { - if databricks.IsSecurableGone(err, errorCodeCatalogGone) { - return false, errCatalogGone - } - if apiStatus(err) == http.StatusNotFound { - return false, refuseUnreadable( - fmt.Sprintf("databricks-connector: failed to read the connector's own privileges on catalog %s", trust.fullName), err) - } - if !deniesTrust(err) { - return false, fmt.Errorf("databricks-connector: failed to read the connector's own privileges on catalog %s: %w", trust.fullName, err) - } - ctxzap.Extract(ctx).Debug("databricks-connector: the connector's own privileges on the catalog cannot be read, so MANAGE stays unproven", - zap.String("catalog", trust.fullName), - zap.Error(err)) - - return false, nil - } - - return manages, nil -} - func (b *catalogBuilder) Grants(ctx context.Context, resource *v2.Resource, attr rs.SyncOpAttrs) ([]*v2.Grant, *rs.SyncOpResults, error) { return b.syncGrants(ctx, resource, attr) } diff --git a/pkg/connector/catalog_test.go b/pkg/connector/catalog_test.go index ee332e36..0f148bba 100644 --- a/pkg/connector/catalog_test.go +++ b/pkg/connector/catalog_test.go @@ -415,7 +415,7 @@ func TestCatalogListSkipsBuiltInCatalogs(t *testing.T) { f := &fakeCatalogAccount{ attached: []int{1}, listed: map[string][]map[string]any{"wsa": { - catalogEntry(systemCatalog, "someone"), + catalogEntry(systemCatalog, "System user"), {"name": "samples", "metastore_id": "ms-1", "owner": "System user", "catalog_type": "SYSTEM_CATALOG"}, catalogEntry("owned", "app-9"), }}, diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 85d84aa4..f06c4efc 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -44,6 +44,7 @@ func (d *Databricks) ResourceSyncers(_ context.Context) []connectorbuilder.Resou newRoleBuilder(d.client), newMetastoreBuilder(d.client, d.clientID, workspaces, principals), newCatalogBuilder(d.client, d.clientID, workspaces, principals), + newSchemaBuilder(d.client, d.clientID, workspaces, principals), } } @@ -75,7 +76,7 @@ func (d *Databricks) Metadata(_ context.Context) (*v2.ConnectorMetadata, error) return &v2.ConnectorMetadata{ DisplayName: "Databricks", Description: "Connector syncing Databricks workspaces, users, groups, service principals and roles to Baton, " + - "along with Unity Catalog data assets — metastores and catalogs — and the privileges granted on them", + "along with Unity Catalog data assets — metastores, catalogs and schemas — and the privileges granted on them", AccountCreationSchema: &v2.ConnectorAccountCreationSchema{ FieldMap: map[string]*v2.ConnectorAccountCreationSchema_Field{ "email": { diff --git a/pkg/connector/connector_test.go b/pkg/connector/connector_test.go index c38751f9..b2d96dcb 100644 --- a/pkg/connector/connector_test.go +++ b/pkg/connector/connector_test.go @@ -87,7 +87,7 @@ func TestResourceSyncersAlwaysRegisterTheUnityCatalogBuilders(t *testing.T) { for _, syncer := range d.ResourceSyncers(ctx) { registered = append(registered, syncer.ResourceType(ctx).GetId()) } - for _, want := range []string{metastoreResourceType.Id, catalogResourceType.Id} { + for _, want := range []string{metastoreResourceType.Id, catalogResourceType.Id, schemaResourceType.Id} { if !slices.Contains(registered, want) { t.Errorf("registered syncers = %v, want them to include %s", registered, want) } diff --git a/pkg/connector/metastore_helpers.go b/pkg/connector/metastore_helpers.go index c9b72176..3a3854f3 100644 --- a/pkg/connector/metastore_helpers.go +++ b/pkg/connector/metastore_helpers.go @@ -467,6 +467,126 @@ func ownPrincipalScimID( return id, nil } +// managePrivilege is what a trust rule looks for when ownership did not settle +// it. ALL_PRIVILEGES is deliberately not here: Databricks leaves MANAGE out of +// it, so its holder still reads a partial privilege list. +// https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges +var managePrivilege = []string{privilegeManage} + +// errSecurableGone reports that the securable a trust read addressed no longer +// exists. Only a level that names its gone codes can raise it. +var errSecurableGone = errors.New("securable no longer exists") + +// holdsPrivilegeOn reports whether the connector holds one of wanted. A refusal leaves it unproven. The read is not cached. +// +// goneCodes are the error_code values that prove the securable is gone. A level +// that names them can tell an absence from a routing miss, so a 404 carrying one +// reports errSecurableGone and any other 404 refuses the read instead of passing +// as unproven. A level that names none keeps every 404 unproven, because it has +// no way to tell the two apart. +func holdsPrivilegeOn( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + securableType, fullName string, + wanted []string, + goneCodes []string, + annos *annotations.Annotations, +) (bool, error) { + if trust.ownPrincipal == "" || trust.workspace == "" { + return false, nil + } + + var holds bool + rateLimit, err := client.ForEachUncachedPermissionsPage(ctx, trust.workspace, securableType, fullName, trust.ownPrincipal, + func(page []databricks.PrivilegeAssignment) (bool, error) { + for _, assignment := range page { + for _, privilege := range assignment.Privileges { + if slices.Contains(wanted, privilege) { + holds = true + + return true, nil + } + } + } + + return false, nil + }) + noteRateLimit(annos, rateLimit) + if err != nil { + if databricks.IsSecurableGone(err, goneCodes...) { + return false, errSecurableGone + } + if len(goneCodes) > 0 && apiStatus(err) == http.StatusNotFound { + return false, refuseUnreadable(fmt.Sprintf( + "databricks-connector: failed to read the connector's own privileges on %s %s", securableType, fullName), err) + } + if !deniesTrust(err) { + return false, fmt.Errorf("databricks-connector: failed to read the connector's own privileges on %s %s: %w", securableType, fullName, err) + } + ctxzap.Extract(ctx).Debug("databricks-connector: the connector's own privileges on the securable cannot be read, so they stay unproven", + zap.String("securable_type", securableType), + zap.String("securable", fullName), + zap.Error(err)) + + return false, nil + } + + return holds, nil +} + +// errTrustRead404 marks a trust read that answered 404 and still has to be +// classified by the level, which is the only holder of the codes that prove +// the securable is gone. +var errTrustRead404 = errors.New("trust read answered 404") + +// holdsEffectivePrivilegeOn reports whether the connector holds one of wanted on +// the securable or any parent, which one uncached read answers. A refusal leaves +// it unproven, and a 404 is left to the level, which owns the gone codes. +func holdsEffectivePrivilegeOn( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + securableType, fullName string, + wanted []string, + annos *annotations.Annotations, +) (bool, error) { + if trust.ownPrincipal == "" || trust.workspace == "" { + return false, nil + } + + var holds bool + rateLimit, err := client.ForEachUncachedEffectivePermissionsPage(ctx, trust.workspace, securableType, fullName, trust.ownPrincipal, + func(page []databricks.EffectivePrivilegeAssignment) (bool, error) { + // The walk stops on the first page that holds one, so this never + // overwrites a hit with a later miss. + holds = slices.ContainsFunc(page, func(assignment databricks.EffectivePrivilegeAssignment) bool { + return assignment.HoldsAny(wanted) + }) + + return holds, nil + }) + noteRateLimit(annos, rateLimit) + if err != nil { + if apiStatus(err) == http.StatusNotFound { + return false, fmt.Errorf("%w: %w", errTrustRead404, err) + } + if !deniesTrust(err) { + return false, fmt.Errorf( + "databricks-connector: failed to read the connector's own effective privileges on %s %s: %w", securableType, fullName, err) + } + ctxzap.Extract(ctx).Debug( + "databricks-connector: the connector's own effective privileges on the securable cannot be read, so they stay unproven", + zap.String("securable_type", securableType), + zap.String("securable", fullName), + zap.Error(err)) + + return false, nil + } + + return holds, nil +} + // assignmentListError explains a failed read of the workspaces a metastore is // assigned to, which decides which workspaces its securables are read through. // A 404 is not an empty assignment list: the SDK would skip the child List and diff --git a/pkg/connector/raw_id_test.go b/pkg/connector/raw_id_test.go index 536e10f0..d0552e9d 100644 --- a/pkg/connector/raw_id_test.go +++ b/pkg/connector/raw_id_test.go @@ -77,6 +77,13 @@ func TestResourcesHaveRawId(t *testing.T) { &v2.ResourceId{ResourceType: metastoreResourceType.Id, Resource: "ms1"}, catalogRoute{})), "ms1::cat", }, + { + // The schema extends its catalog's id with the dotted full name. + "schema", + build(schemaResource(&databricks.Schema{Name: "sch"}, "ms1", "cat", + &v2.ResourceId{ResourceType: catalogResourceType.Id, Resource: "ms1::cat"}, schemaRoute{})), + "ms1::cat.sch", + }, } for _, tc := range cases { diff --git a/pkg/connector/resource_types.go b/pkg/connector/resource_types.go index 22c94445..5f192937 100644 --- a/pkg/connector/resource_types.go +++ b/pkg/connector/resource_types.go @@ -17,6 +17,7 @@ const ( permissionAccountAdmin = "Account admin" permissionWorkspaceAdmin = "Workspace admin" permissionUseCatalog = "USE_CATALOG" + permissionUseSchema = "USE_SCHEMA" ) var securableScopes = []string{ @@ -93,4 +94,11 @@ var ( Traits: []v2.ResourceType_Trait{v2.ResourceType_TRAIT_APP}, Annotations: annotationsForStaticSecurableResourceType(securablePermissions(permissionUseCatalog)...), } + + schemaResourceType = &v2.ResourceType{ + Id: "schema", + DisplayName: "Schema", + Traits: []v2.ResourceType_Trait{v2.ResourceType_TRAIT_APP}, + Annotations: annotationsForStaticSecurableResourceType(securablePermissions(permissionUseCatalog, permissionUseSchema)...), + } ) diff --git a/pkg/connector/schema.go b/pkg/connector/schema.go new file mode 100644 index 00000000..76103b87 --- /dev/null +++ b/pkg/connector/schema.go @@ -0,0 +1,476 @@ +package connector + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "slices" + "strconv" + "strings" + + "github.com/conductorone/baton-databricks/pkg/databricks" + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/annotations" + "github.com/conductorone/baton-sdk/pkg/connectorbuilder" + "github.com/conductorone/baton-sdk/pkg/pagination" + ent "github.com/conductorone/baton-sdk/pkg/types/entitlement" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "github.com/conductorone/baton-sdk/pkg/uhttp" + "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap" + "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +var _ connectorbuilder.StaticEntitlementSyncerV2 = (*schemaBuilder)(nil) + +const ( + securableTypeSchema = "SCHEMA" + + // informationSchema is the read-only schema Databricks puts in every catalog. + // Its privileges are not grantable, so it is left out of the sync. + informationSchema = "information_schema" + + // schemaNameSeparator joins the catalog and the schema into the dotted full + // name the permissions endpoint addresses a schema by. Databricks rejects a + // name holding a period, so splitting on it is unambiguous. + schemaNameSeparator = "." + + profileKeySchemaName = "schema_name" + + // profileKeyCatalogOwner carries the owner of the parent catalog. Owning the + // catalog is one of the things that makes a schema's privilege read complete, + // and Grants is handed only the schema, so List writes the owner down instead + // of listing the catalogs again on every call. + profileKeyCatalogOwner = "catalog_owner" + + // schemaWriteRemedy is what a 403 on the PATCH tells the operator. + schemaWriteRemedy = "the connector must own the metastore, the parent catalog or the schema, or hold MANAGE on the schema or the parent catalog" + + // schemaTrustRemedy is what a refusal to report a partial privilege list + // tells the operator. + schemaTrustRemedy = "the connector's service principal must own the metastore, the parent catalog or the schema, hold MANAGE on the schema or " + + "on the parent catalog, or be assigned Admin directly on the workspace whose admins group owns the parent catalog" + + // schemaListRemedy is what a refusal to sync a catalog whose schema listing + // would be partial tells the operator. + schemaListRemedy = "the connector's service principal must own the metastore or the catalog, or hold MANAGE, USE_SCHEMA or ALL_PRIVILEGES on the catalog" + + // errorCodeSchemaGone is what Databricks names in error_code when the schema + // a request addresses does not exist. + errorCodeSchemaGone = "SCHEMA_DOES_NOT_EXIST" +) + +// schemaPrivileges is the catalog set minus the three that stop at the catalog. +// https://docs.databricks.com/aws/en/data-governance/unity-catalog/manage-privileges/privileges +var schemaPrivileges = slices.DeleteFunc(slices.Clone(catalogPrivileges), func(privilege string) bool { + return slices.Contains([]string{"BROWSE", "CREATE_SCHEMA", "USE_CATALOG"}, privilege) +}) + +// schemaListingPrivileges each make ListSchemas return every schema in the +// catalog. Unlike a privilege read, a complete listing does not need MANAGE: +// USE_SCHEMA and ALL_PRIVILEGES carry it too, which was measured on a tenant +// where neither made the privilege list complete. +var schemaListingPrivileges = []string{privilegeManage, "USE_SCHEMA", "ALL_PRIVILEGES"} + +var schemaLevel = securableLevel{ + securableType: databricks.SecurableSchema, + noun: "schema", + privileges: schemaPrivileges, + writeRemedy: schemaWriteRemedy, + trustRemedy: schemaTrustRemedy, + permissionsName: schemaPermissionsName, + trusts: schemaGrantsAreTrustworthy, + // SCHEMA_DOES_NOT_EXIST is the schema itself. CATALOG_DOES_NOT_EXIST is the + // parent: dropping the catalog makes the schema read 404 with that code. + goneErrorCodes: []string{errorCodeSchemaGone, errorCodeCatalogGone}, +} + +type schemaBuilder struct { + securableOps + // workspaces is the configured allowlist, empty when none was given. + workspaces map[string]struct{} +} + +func newSchemaBuilder(client *databricks.Client, clientID string, workspaces map[string]struct{}, sync syncPrincipals) *schemaBuilder { + return &schemaBuilder{ + securableOps: securableOps{client: client, clientID: clientID, sync: sync, level: schemaLevel}, + workspaces: workspaces, + } +} + +func (b *schemaBuilder) ResourceType(_ context.Context) *v2.ResourceType { + return schemaResourceType +} + +// schemaResourceID keys a schema by its metastore and its dotted full name, which +// extends the catalog's "{metastore_id}::{catalog_name}". +func schemaResourceID(metastoreID, catalogName, name string) string { + return metastoreID + securableKeySeparator + catalogName + schemaNameSeparator + name +} + +// parseSchemaID reads a schema resource id, "{metastore_id}::{catalog}.{schema}". +func parseSchemaID(resourceID string) (string, string, string, error) { + metastoreID, fullName, ok := strings.Cut(resourceID, securableKeySeparator) + if !ok { + return "", "", "", status.Errorf(codes.InvalidArgument, "databricks-connector: invalid schema resource id %q", resourceID) + } + + catalogName, name, ok := strings.Cut(fullName, schemaNameSeparator) + if !ok || metastoreID == "" || catalogName == "" || name == "" { + return "", "", "", status.Errorf(codes.InvalidArgument, "databricks-connector: invalid schema resource id %q", resourceID) + } + + return metastoreID, catalogName, name, nil +} + +// schemaPermissionsName: the permissions endpoint addresses a schema by its +// dotted full name. The metastore is carried by the host. +func schemaPermissionsName(resourceID string) (string, error) { + _, catalogName, name, err := parseSchemaID(resourceID) + if err != nil { + return "", err + } + + return catalogName + schemaNameSeparator + name, nil +} + +func schemaResource( + schema *databricks.Schema, + metastoreID, catalogName string, + parent *v2.ResourceId, + route schemaRoute, +) (*v2.Resource, error) { + fullName := catalogName + schemaNameSeparator + schema.Name + profile := map[string]any{ + profileKeyMetastoreID: metastoreID, + profileKeyMetastoreOwner: route.MetastoreOwner, + profileKeyFullName: fullName, + profileKeyCatalogName: catalogName, + profileKeyCatalogOwner: route.CatalogOwner, + profileKeySchemaName: schema.Name, + profileKeyOwner: schema.Owner, + profileKeySecurableType: securableTypeSchema, + profileKeyWorkspace: route.Workspace, + profileKeyWorkspaceID: route.WorkspaceID, + profileKeyWorkspaceName: route.WorkspaceName, + } + + resourceID := schemaResourceID(metastoreID, catalogName, schema.Name) + + // The display name is the full name rather than the leaf: main.sales and + // finance.sales would otherwise be two rows both reading "sales" in C1. + return rs.NewAppResource( + fullName, + schemaResourceType, + resourceID, + nil, + rs.WithResourceProfile(profile), + rs.WithParentResourceID(parent), + rs.WithAnnotation(&v2.RawId{Id: resourceID}), + ) +} + +// schemaRoute is the workspace a catalog's schemas are listed through, carried in +// the page token so later pages neither re-resolve the route nor prove again that +// the listing is complete. +type schemaRoute struct { + Workspace string `json:"w"` + WorkspaceID string `json:"i"` + WorkspaceName string `json:"n"` + // MetastoreOwner and CatalogOwner ride along so each schema's profile can + // carry them. See profileKeyCatalogOwner. + MetastoreOwner string `json:"o"` + CatalogOwner string `json:"c"` +} + +// List pages a catalog's schemas through a workspace that can see every schema. +// Databricks answers a caller without catalog-wide visibility with only the +// schemas it owns or can use, at 200. For the built-in system catalog that +// short list is left out. For every other catalog the sync stops, so +// ConductorOne keeps the schemas it already has. +func (b *schemaBuilder) List(ctx context.Context, parentResourceID *v2.ResourceId, attr rs.SyncOpAttrs) ([]*v2.Resource, *rs.SyncOpResults, error) { + if parentResourceID.GetResourceType() != catalogResourceType.Id { + return nil, nil, nil + } + + metastoreID, catalogName, err := parseCatalogID(parentResourceID.GetResource()) + if err != nil { + return nil, nil, err + } + + annos := annotations.Annotations{} + bag := &pagination.Bag{} + if err := bag.Unmarshal(attr.PageToken.Token); err != nil { + return nil, nil, status.Errorf(codes.InvalidArgument, "databricks-connector: invalid schema page token: %v", err) + } + + if bag.Current() == nil { + if err := b.pushRoute(ctx, metastoreID, catalogName, bag, &annos); err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, err + } + if bag.Current() == nil { + ctxzap.Extract(ctx).Debug("databricks-connector: no workspace in sync scope lists the catalog; listing no schemas", + zap.String("metastore", metastoreID), + zap.String("catalog", catalogName)) + + return nil, &rs.SyncOpResults{Annotations: annos}, nil + } + } + + var route schemaRoute + if err := json.Unmarshal([]byte(bag.ResourceID()), &route); err != nil { + return nil, nil, status.Errorf(codes.Internal, "databricks-connector: invalid schema page token: %v", err) + } + + schemas, next, rateLimit, err := b.client.ListSchemas(ctx, route.Workspace, catalogName, bag.PageToken(), ResourcesPageSize) + noteRateLimit(&annos, rateLimit) + if err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, schemaListingError(catalogName, route.Workspace, err) + } + + var rv []*v2.Resource + for _, schema := range schemas { + if schema.Name == "" || schema.Name == informationSchema { + continue + } + + built, err := schemaResource(&schema, metastoreID, catalogName, parentResourceID, route) + if err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, fmt.Errorf( + "databricks-connector: failed to build the resource for schema %s.%s: %w", catalogName, schema.Name, err) + } + rv = append(rv, built) + } + + nextToken, err := bag.NextToken(next) + if err != nil { + return nil, &rs.SyncOpResults{Annotations: annos}, err + } + + return rv, &rs.SyncOpResults{Annotations: annos, NextPageToken: nextToken}, nil +} + +// pushRoute records a workspace that can list every schema of the catalog. It +// walks the running in-scope workspaces until one both lists the catalog and +// can read every schema. An empty stack means no workspace this sync can use +// serves the metastore. A catalog none of them can list completely fails the +// sync: returning no schemas would make ConductorOne delete them. system and +// samples are not special here. The catalog sync leaves them out, when it +// leaves them out, before this List runs. A workspace answering that the catalog +// is gone has only said it does not serve it, since an unbound workspace answers +// the same way; only every workspace saying so, with no refusal among them, means +// there is nothing left to list. +func (b *schemaBuilder) pushRoute( + ctx context.Context, + metastoreID, catalogName string, + bag *pagination.Bag, + annos *annotations.Annotations, +) error { + scope, err := scopedWorkspaces(ctx, b.client, b.workspaces, annos) + if err != nil { + return err + } + + usable, inScope, err := scope.attached(ctx, b.client, metastoreID, annos) + if err != nil { + return assignmentListError(metastoreID, err) + } + if len(usable) == 0 { + if inScope { + return status.Errorf(codes.FailedPrecondition, + "databricks-connector: metastore %s has no running workspace in sync scope", metastoreID) + } + + return nil + } + + owner, err := metastoreOwner(ctx, b.client, metastoreID, annos) + if err != nil { + return err + } + + var sawCatalog, sawGone bool + var lastDeny error + for _, workspace := range usable { + catalog, err := catalogThrough(ctx, b.client, workspace.DeploymentName, metastoreID, catalogName, annos) + if err != nil { + if errors.Is(err, errCatalogGone) { + sawGone = true + continue + } + if deniesTrust(err) { + lastDeny = err + continue + } + return err + } + if catalog == nil { + continue + } + sawCatalog = true + + workspaceID := strconv.Itoa(workspace.ID) + route := schemaRoute{ + Workspace: workspace.DeploymentName, + WorkspaceID: workspaceID, + WorkspaceName: workspace.Name, + MetastoreOwner: owner, + CatalogOwner: catalog.Owner, + } + complete, err := schemaListingIsComplete(ctx, b.client, securableTrust{ + ownPrincipal: b.clientID, + metastoreOwner: owner, + parentOwner: catalog.Owner, + fullName: catalogName, + workspace: route.Workspace, + workspaceID: workspaceID, + }, annos) + if err != nil { + return err + } + if !complete { + continue + } + + encoded, err := json.Marshal(route) + if err != nil { + return err + } + bag.Push(pagination.PageState{ResourceTypeID: schemaResourceType.Id, ResourceID: string(encoded)}) + + return nil + } + if !sawCatalog { + // The catalog syncer drops the parent in the same sync, so this empties + // what ConductorOne is already removing. + if sawGone && lastDeny == nil { + return nil + } + msg := fmt.Sprintf("databricks-connector: no workspace in sync scope lists catalog %s of metastore %s", catalogName, metastoreID) + if lastDeny != nil { + msg = fmt.Sprintf("%s: %s", msg, lastDeny.Error()) + } + return status.Errorf(codes.FailedPrecondition, "%s", msg) + } + + return status.Errorf(codes.PermissionDenied, + "databricks-connector: cannot list every schema in catalog %s; %s", + catalogName, schemaListRemedy) +} + +// schemaListingError returns NotFound only when Databricks names the missing +// catalog; any other refusal leaves the schema listing unknown. +func schemaListingError(catalogName, workspace string, err error) error { + if databricks.IsSecurableGone(err, errorCodeCatalogGone) { + return uhttp.WrapErrors(codes.NotFound, + fmt.Sprintf("databricks-connector: catalog %s no longer exists", catalogName), err) + } + + switch apiStatus(err) { + case http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound: + return uhttp.WrapErrors(codes.FailedPrecondition, + fmt.Sprintf("databricks-connector: the schemas of catalog %s cannot be listed through workspace %s, "+ + "so which schemas are in scope is unknown; %s", catalogName, workspace, schemaListRemedy), + err) + } + + return fmt.Errorf("databricks-connector: failed to list the schemas of catalog %s through workspace %s: %w", + catalogName, workspace, err) +} + +// schemaListingIsComplete is true when ListSchemas returns every schema. USE_SCHEMA and ALL_PRIVILEGES count here. They do not make a privilege read complete. +// It reports errCatalogGone when Databricks names the catalog missing, which the +// gone codes are what tell apart from a privilege this caller cannot read. +func schemaListingIsComplete( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + annos *annotations.Annotations, +) (bool, error) { + // An owner sees everything below it, and the metastore's owner is its admin. + if ownsPrincipal(trust.ownPrincipal, trust.metastoreOwner) || ownsPrincipal(trust.ownPrincipal, trust.parentOwner) { + return true, nil + } + + // The workspace admins group acts as the owner for a direct Admin of its workspace. + if ownerWorkspaceID, ok := catalogAdminWorkspaceID(trust.parentOwner); ok { + admin, err := isWorkspaceAdmin(ctx, client, trust.ownPrincipal, ownerWorkspaceID, annos) + if err != nil || admin { + return admin, err + } + } + + complete, err := holdsPrivilegeOn(ctx, client, trust, databricks.SecurableCatalog, trust.fullName, + schemaListingPrivileges, []string{errorCodeCatalogGone}, annos) + if errors.Is(err, errSecurableGone) { + return false, errCatalogGone + } + + return complete, err +} + +// schemaGrantsAreTrustworthy is true for an owner of the schema, catalog, or +// metastore, a holder of MANAGE on the schema or catalog, or a direct Admin +// when the workspace admins group owns the catalog. +func schemaGrantsAreTrustworthy( + ctx context.Context, + client *databricks.Client, + trust securableTrust, + annos *annotations.Annotations, +) (bool, error) { + // An owner holds every privilege implicitly, which is not reported as a grant. + // On the live tenant the connector owns the metastore, so this settles the + // whole rule with no extra call at all. + if ownsPrincipal(trust.ownPrincipal, trust.metastoreOwner) || + ownsPrincipal(trust.ownPrincipal, trust.parentOwner) || + ownsPrincipal(trust.ownPrincipal, trust.owner) { + return true, nil + } + + if ownerWorkspaceID, ok := catalogAdminWorkspaceID(trust.parentOwner); ok { + admin, err := isWorkspaceAdmin(ctx, client, trust.ownPrincipal, ownerWorkspaceID, annos) + if err != nil || admin { + return admin, err + } + } + + // MANAGE on the schema or on the catalog above it makes the privilege list + // complete, and one effective read reports both. + return holdsEffectivePrivilegeOn(ctx, client, trust, databricks.SecurableSchema, trust.fullName, managePrivilege, annos) +} + +func (b *schemaBuilder) StaticEntitlements(_ context.Context, _ rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { + grantable := ent.WithGrantableTo(userResourceType, groupResourceType, servicePrincipalResourceType) + rv := make([]*v2.Entitlement, 0, len(schemaPrivileges)+1) + for _, privilege := range schemaPrivileges { + rv = append(rv, ent.NewPermissionEntitlement(nil, privilege, + grantable, + ent.WithDescription(fmt.Sprintf("%s privilege on a schema in Databricks", privilege)), + )) + } + + rv = append(rv, ent.NewOwnershipEntitlement(nil, ownerEntitlement, + ent.WithDescription("Owns a schema in Databricks"), + )) + + return rv, nil, nil +} + +func (b *schemaBuilder) Entitlements(context.Context, *v2.Resource, rs.SyncOpAttrs) ([]*v2.Entitlement, *rs.SyncOpResults, error) { + return nil, nil, nil +} + +func (b *schemaBuilder) Grants(ctx context.Context, resource *v2.Resource, attr rs.SyncOpAttrs) ([]*v2.Grant, *rs.SyncOpResults, error) { + return b.syncGrants(ctx, resource, attr) +} + +func (b *schemaBuilder) Grant(ctx context.Context, principal *v2.Resource, entitlement *v2.Entitlement) (annotations.Annotations, error) { + return b.grant(ctx, principal.GetId(), entitlement) +} + +func (b *schemaBuilder) Revoke(ctx context.Context, revoked *v2.Grant) (annotations.Annotations, error) { + return b.revoke(ctx, revoked.GetPrincipal().GetId(), revoked.GetEntitlement()) +} diff --git a/pkg/connector/schema_test.go b/pkg/connector/schema_test.go new file mode 100644 index 00000000..ecef53d4 --- /dev/null +++ b/pkg/connector/schema_test.go @@ -0,0 +1,796 @@ +package connector + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "slices" + "strconv" + "strings" + "testing" + + v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" + "github.com/conductorone/baton-sdk/pkg/pagination" + "github.com/conductorone/baton-sdk/pkg/types/grant" + rs "github.com/conductorone/baton-sdk/pkg/types/resource" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/conductorone/baton-databricks/pkg/databricks" +) + +// fakeSchemaAccount is one metastore (ms-1) holding the catalog main, served by +// the RUNNING workspace wsa (id 1, named prod) and, when second is set, by wsb +// (id 2, named other). Each field shapes one answer; the counters record what the +// connector asked. +type fakeSchemaAccount struct { + metastoreOwner string + catalogOwner string + second bool + + // schemas is what the schemas listing returns, paged by max_results. + schemas []map[string]any + // schemaStatus fails the schemas listing with that status. + schemaStatus int + // assignments maps a securable's full name to the privilege assignments on it. + assignments map[string][]map[string]any + // ownPrivileges answers a read filtered by principal, keyed by + // "{securable_type}/{full_name}", which is how the connector asks for its own + // effective privileges on a securable. + ownPrivileges map[string][]map[string]any + // permStatus fails the permissions endpoint on that deployment name. + permStatus map[string]int + // permErrorCode is the vendor error_code that rides along with permStatus. + permErrorCode string + // catalogStatus fails the catalog read on that deployment name. + catalogStatus map[string]int + // catalogErrorCode is the vendor error_code that rides along with + // catalogStatus. CATALOG_DOES_NOT_EXIST is what an unbound workspace answers, + // which is the same code a deleted catalog answers. + catalogErrorCode string + // workspaceMembers maps a workspace id to its permission assignments, or to + // the HTTP status the account API answers the list with. + workspaceMembers map[string]any + patchStatus int + users map[string]string + groups map[string]string + + routed int + schemaPages int + catalogPages int + patches []map[string]any + permHosts []string + memberLists []string + filteredReads []string + catalogReads []string +} + +func (f *fakeSchemaAccount) handler(t *testing.T) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + path := r.URL.Path + encode := func(v any) { _ = json.NewEncoder(w).Encode(v) } + switch { + case strings.Contains(path, "/scim/v2/"): + serveSCIM(w, r, f.users, f.groups, nil, nil) + case strings.HasSuffix(path, "/metastores"): + encode(map[string]any{"metastores": []map[string]any{ + {"metastore_id": "ms-1", "name": "main", "owner": f.metastoreOwner}, + }}) + case strings.Contains(path, "/metastores/") && strings.HasSuffix(path, "/workspaces"): + ids := []int{1} + if f.second { + ids = append(ids, 2) + } + encode(map[string]any{"workspace_ids": ids}) + case strings.HasSuffix(path, "/permissionassignments"): + parts := strings.Split(path, "/") + workspaceID := parts[len(parts)-2] + f.memberLists = append(f.memberLists, workspaceID) + switch v := f.workspaceMembers[workspaceID].(type) { + case int: + http.Error(w, `{"message":"denied"}`, v) + default: + encode(map[string]any{"permission_assignments": v}) + } + case strings.HasSuffix(path, "/workspaces"): + f.routed++ + listed := []map[string]any{ + {"workspace_id": 1, "workspace_name": "prod", "deployment_name": "wsa", "workspace_status": "RUNNING"}, + } + if f.second { + listed = append(listed, map[string]any{ + "workspace_id": 2, "workspace_name": "other", "deployment_name": "wsb", "workspace_status": "RUNNING", + }) + } + encode(listed) + case strings.HasSuffix(path, "/unity-catalog/catalogs"): + f.catalogPages++ + encode(map[string]any{"catalogs": []map[string]any{ + {"name": "main", "metastore_id": "ms-1", "owner": f.catalogOwner}, + }}) + case strings.Contains(path, "/unity-catalog/catalogs/"): + f.serveCatalog(w, r) + case strings.HasSuffix(path, "/unity-catalog/schemas"): + f.serveSchemas(w, r) + case strings.Contains(path, "/unity-catalog/effective-permissions/"): + _, securable, _ := strings.Cut(path, "/unity-catalog/effective-permissions/") + f.serveEffectivePermissions(w, r, securable) + case strings.Contains(path, "/unity-catalog/permissions/"): + _, securable, _ := strings.Cut(path, "/unity-catalog/permissions/") + f.servePermissions(w, r, securable) + default: + t.Errorf("unexpected request %s %s", r.Method, path) + http.Error(w, `{"message":"unexpected"}`, http.StatusInternalServerError) + } + } +} + +// serveCatalog answers GET /catalogs/{name}. catalogStatus makes a workspace +// refuse it, which is how a catalog bound elsewhere reads from a workspace that +// does not serve it. +func (f *fakeSchemaAccount) serveCatalog(w http.ResponseWriter, r *http.Request) { + workspace := r.Header.Get("X-Test-Workspace") + f.catalogReads = append(f.catalogReads, workspace) + if code, ok := f.catalogStatus[workspace]; ok { + body := `{"message":"refused"}` + if f.catalogErrorCode != "" { + body = fmt.Sprintf(`{"error_code":%q,"message":"refused"}`, f.catalogErrorCode) + } + http.Error(w, body, code) + + return + } + + _ = json.NewEncoder(w).Encode(map[string]any{"name": "main", "metastore_id": "ms-1", "owner": f.catalogOwner}) +} + +func (f *fakeSchemaAccount) serveSchemas(w http.ResponseWriter, r *http.Request) { + if f.schemaStatus != 0 { + http.Error(w, `{"message":"denied"}`, f.schemaStatus) + + return + } + if got := r.URL.Query().Get("catalog_name"); got != "main" { + http.Error(w, `{"message":"catalog_name is required"}`, http.StatusBadRequest) + + return + } + + f.schemaPages++ + all := f.schemas + start, _ := strconv.Atoi(r.URL.Query().Get("page_token")) + size, _ := strconv.Atoi(r.URL.Query().Get("max_results")) + if size <= 0 || size > len(all)-start { + size = len(all) - start + } + next := "" + if start+size < len(all) { + next = strconv.Itoa(start + size) + } + _ = json.NewEncoder(w).Encode(map[string]any{"schemas": all[start : start+size], "next_page_token": next}) +} + +func (f *fakeSchemaAccount) servePermissions(w http.ResponseWriter, r *http.Request, securable string) { + workspace := r.Header.Get("X-Test-Workspace") + f.permHosts = append(f.permHosts, workspace) + if code, ok := f.permStatus[workspace]; ok { + body := `{"message":"refused"}` + if f.permErrorCode != "" { + body = fmt.Sprintf(`{"error_code":%q,"message":"refused"}`, f.permErrorCode) + } + http.Error(w, body, code) + + return + } + if principal := r.URL.Query().Get("principal"); principal != "" { + f.filteredReads = append(f.filteredReads, securable) + _ = json.NewEncoder(w).Encode(map[string]any{"privilege_assignments": f.ownPrivileges[securable]}) + + return + } + + _, fullName, _ := strings.Cut(securable, "/") + if r.Method == http.MethodPatch { + if f.patchStatus != 0 { + http.Error(w, `{"message":"denied"}`, f.patchStatus) + + return + } + var body struct { + Changes []map[string]any `json:"changes"` + } + decoder := json.NewDecoder(r.Body) + decoder.UseNumber() + _ = decoder.Decode(&body) + for _, change := range body.Changes { + f.patches = append(f.patches, change) + f.apply(fullName, change) + } + } + _ = json.NewEncoder(w).Encode(map[string]any{"privilege_assignments": f.assignments[fullName]}) +} + +// serveEffectivePermissions answers from the same grants the permissions +// endpoint serves, applying inheritance the way Databricks does: a parent's +// privilege is reported on the securable below it, naming the parent. +func (f *fakeSchemaAccount) serveEffectivePermissions(w http.ResponseWriter, r *http.Request, securable string) { + workspace := r.Header.Get("X-Test-Workspace") + f.permHosts = append(f.permHosts, workspace) + if code, ok := f.permStatus[workspace]; ok { + body := `{"message":"refused"}` + if f.permErrorCode != "" { + body = fmt.Sprintf(`{"error_code":%q,"message":"refused"}`, f.permErrorCode) + } + http.Error(w, body, code) + + return + } + + f.filteredReads = append(f.filteredReads, securable) + _ = json.NewEncoder(w).Encode(map[string]any{"privilege_assignments": f.effectivePrivileges(securable)}) +} + +// effectivePrivileges flattens the securable's own grants with every parent's. +// An inherited row names the parent and a direct one does not, which is the +// only thing that tells them apart. +func (f *fakeSchemaAccount) effectivePrivileges(securable string) []map[string]any { + rows := []map[string]any{} + collect := func(key, inheritedFrom string) { + for _, assignment := range f.ownPrivileges[key] { + held, ok := assignment["privileges"].([]string) + if !ok { + continue + } + privileges := make([]map[string]any, 0, len(held)) + for _, privilege := range held { + row := map[string]any{"privilege": privilege} + if inheritedFrom != "" { + securableType, fullName, _ := strings.Cut(inheritedFrom, "/") + row["inherited_from_type"] = strings.ToUpper(securableType) + row["inherited_from_name"] = fullName + } + privileges = append(privileges, row) + } + rows = append(rows, map[string]any{"principal": assignment["principal"], "privileges": privileges}) + } + } + + collect(securable, "") + for _, parent := range securableParents(securable) { + collect(parent, parent) + } + + return rows +} + +// securableParents names what a privilege can be inherited from, innermost +// first. A catalog has none: its parent is the metastore, which rejects MANAGE. +func securableParents(securable string) []string { + securableType, fullName, _ := strings.Cut(securable, "/") + parts := strings.Split(fullName, ".") + switch securableType { + case "schema": + if len(parts) == 2 { + return []string{"catalog/" + parts[0]} + } + case "table", "volume": + if len(parts) == 3 { + return []string{"schema/" + parts[0] + "." + parts[1], "catalog/" + parts[0]} + } + } + + return nil +} + +func (f *fakeSchemaAccount) apply(fullName string, change map[string]any) { + added, _ := change["add"].([]any) + for _, privilege := range added { + f.assignments[fullName] = append(f.assignments[fullName], map[string]any{ + "principal": change["principal"], "principal_id": 1, "privileges": []string{privilege.(string)}, + }) + } + remove, ok := change["remove"].([]any) + if !ok { + return + } + kept := f.assignments[fullName][:0] + for _, assignment := range f.assignments[fullName] { + if fmt.Sprint(assignment["principal_id"]) != fmt.Sprint(change["principal_id"]) || + !slices.Contains(assignment["privileges"].([]string), remove[0].(string)) { + kept = append(kept, assignment) + } + } + f.assignments[fullName] = kept +} + +// newTestSchemaBuilder authenticates as the owner of the metastore, which is the +// live shape: ownership settles every trust rule with no probe call at all. +func newTestSchemaBuilder(t *testing.T, f *fakeSchemaAccount) *schemaBuilder { + return newTestSchemaBuilderAs(t, f, testOwnPrincipal) +} + +func newTestSchemaBuilderAs(t *testing.T, f *fakeSchemaAccount, clientID string) *schemaBuilder { + t.Setenv("BATON_HTTP_CACHE_TTL", "0") + if f.metastoreOwner == "" { + f.metastoreOwner = testOwnPrincipal + } + if f.assignments == nil { + f.assignments = map[string][]map[string]any{} + } + client, _ := newWorkspaceRoutedClient(t, f.handler(t)) + + return newSchemaBuilder(client, clientID, nil, syncPrincipals{groups: true}) +} + +var testCatalogResourceID = &v2.ResourceId{ResourceType: catalogResourceType.Id, Resource: "ms-1::main"} + +// testSchema is a schema resource as List writes it, routed through wsa. +func testSchema(t *testing.T, name, owner, catalogOwner, metastoreOwner, workspace string) *v2.Resource { + t.Helper() + r, err := schemaResource(&databricks.Schema{Name: name, Owner: owner}, "ms-1", "main", testCatalogResourceID, + schemaRoute{Workspace: workspace, WorkspaceID: "1", WorkspaceName: "prod", MetastoreOwner: metastoreOwner, CatalogOwner: catalogOwner}) + if err != nil { + t.Fatalf("schemaResource: %v", err) + } + + return r +} + +// listAllSchemas drives List the way the SDK does, page token by page token. +func listAllSchemas(t *testing.T, b *schemaBuilder) ([]*v2.Resource, error) { + t.Helper() + + var all []*v2.Resource + token := "" + for range 1000 { + resources, results, err := b.List(context.Background(), testCatalogResourceID, rs.SyncOpAttrs{PageToken: pagination.Token{Token: token}}) + if err != nil { + return all, err + } + all = append(all, resources...) + if results == nil || results.NextPageToken == "" { + return all, nil + } + token = results.NextPageToken + } + t.Fatal("List never returned an empty page token") + + return nil, nil +} + +// Databricks answers a caller without catalog-wide visibility with only the +// schemas it owns or can use, at 200, which is indistinguishable from a catalog +// that holds just those. A short list read as the whole one makes ConductorOne +// delete the schemas it did not see and every grant under them, so the sync +// fails with PermissionDenied and keeps them. +func TestSchemaListRequiresACompleteListing(t *testing.T) { + adminOfTheWorkspace := []map[string]any{ + {"principal": map[string]any{"service_principal_name": "app-9"}, "permissions": []string{"ADMIN"}}, + } + // The principal filter answers with the group the privilege is inherited + // through, so the row names the group and not the connector's principal. + heldThroughAGroup := func(privilege string) map[string][]map[string]any { + return map[string][]map[string]any{ + "catalog/main": {{"principal": "data-eng", "principal_id": 11, "privileges": []string{privilege}}}, + } + } + + tests := []struct { + name string + clientID string + metastoreOwner string + catalogOwner string + members any + ownPrivileges map[string][]map[string]any + wantSchemas bool + wantMemberLists int + wantFilteredReads int + }{ + { + name: "owner of the metastore", clientID: "eng", + metastoreOwner: "eng", catalogOwner: "someone", + wantSchemas: true, + }, + { + name: "owner of the catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "app-9", + wantSchemas: true, + }, + { + name: "manage on the catalog, inherited through a group", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", ownPrivileges: heldThroughAGroup("MANAGE"), + wantSchemas: true, wantFilteredReads: 1, + }, + { + // Measured: a catalog grant of USE_SCHEMA does list a schema owned by + // someone else, even though it does not make the privilege list complete. + name: "use schema on the catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", ownPrivileges: heldThroughAGroup("USE_SCHEMA"), + wantSchemas: true, wantFilteredReads: 1, + }, + { + name: "all privileges on the catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", ownPrivileges: heldThroughAGroup("ALL_PRIVILEGES"), + wantSchemas: true, wantFilteredReads: 1, + }, + { + name: "admin of the workspace whose admins group owns the catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "_workspace_admins_prod_1", members: adminOfTheWorkspace, + wantSchemas: true, wantMemberLists: 1, + }, + { + // Workspace Admin on its own is not an authority: the system catalog + // came back with 2 schemas out of 15 for one. + name: "admin of a workspace that does not own the catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", members: adminOfTheWorkspace, + wantSchemas: false, wantFilteredReads: 1, + }, + { + name: "none of them", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", + wantSchemas: false, wantFilteredReads: 1, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + f := &fakeSchemaAccount{ + metastoreOwner: test.metastoreOwner, + catalogOwner: test.catalogOwner, + workspaceMembers: map[string]any{"1": test.members}, + ownPrivileges: test.ownPrivileges, + schemas: []map[string]any{{"name": "sales", "owner": "someone"}}, + } + b := newTestSchemaBuilderAs(t, f, test.clientID) + + resources, err := listAllSchemas(t, b) + switch { + case test.wantSchemas && err != nil: + t.Fatalf("List: %v", err) + case test.wantSchemas && len(resources) != 1: + t.Fatalf("listed %d schemas, want 1", len(resources)) + case !test.wantSchemas: + if status.Code(err) != codes.PermissionDenied { + t.Fatalf("List = %v, want PermissionDenied", err) + } + if len(resources) != 0 { + t.Errorf("listed %d schemas, want none", len(resources)) + } + } + if len(f.memberLists) != test.wantMemberLists { + t.Errorf("workspace assignments listed %d times, want %d", len(f.memberLists), test.wantMemberLists) + } + if len(f.filteredReads) != test.wantFilteredReads { + t.Errorf("own privileges read %d times, want %d", len(f.filteredReads), test.wantFilteredReads) + } + }) + } +} + +// The catalog's schemas are read through a workspace that serves the catalog, +// not through whichever workspace the metastore's assignment list names first. +// An ISOLATED catalog is bound to a subset of them, so stopping at the first one +// would emit no schemas and ConductorOne would delete every schema it had. +// A workspace that does not serve the catalog is skipped whichever way it says +// so. A bare 404 is a refusal; a 404 carrying CATALOG_DOES_NOT_EXIST is what an +// ISOLATED catalog answers from every workspace it is not bound to, and it must +// not be read as a deletion that ends the walk. +func TestSchemaListTriesTheNextWorkspace(t *testing.T) { + for _, errorCode := range []string{"", errorCodeCatalogGone} { + name := "bare 404" + if errorCode != "" { + name = errorCode + } + t.Run(name, func(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "eng", + catalogStatus: map[string]int{"wsa": http.StatusNotFound}, + catalogErrorCode: errorCode, + schemas: []map[string]any{{"name": "sales", "owner": "someone"}}, + } + b := newTestSchemaBuilderAs(t, f, "eng") + + resources, err := listAllSchemas(t, b) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(resources) != 1 { + t.Fatalf("listed %d schemas, want 1 through the second workspace", len(resources)) + } + if workspace, _ := rs.GetProfileStringValue(rs.GetProfile(resources[0]), profileKeyWorkspace); workspace != "wsb" { + t.Errorf("profile[%s] = %q, want wsb", profileKeyWorkspace, workspace) + } + if want := []string{"wsa", "wsb"}; !slices.Equal(f.catalogReads, want) { + t.Errorf("catalog reads = %v, want %v (the first workspace is tried before the second)", f.catalogReads, want) + } + }) + } +} + +// No workspace in scope serving the catalog is not the catalog being empty. +// Returning no schemas would make ConductorOne delete the ones it already has, +// so the sync fails and keeps them. +func TestSchemaListFailsWhenNoWorkspaceServesTheCatalog(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "eng", + catalogStatus: map[string]int{"wsa": http.StatusNotFound, "wsb": http.StatusForbidden}, + schemas: []map[string]any{{"name": "sales", "owner": "someone"}}, + } + b := newTestSchemaBuilderAs(t, f, "eng") + + resources, err := listAllSchemas(t, b) + if status.Code(err) != codes.FailedPrecondition { + t.Fatalf("List = %v, want FailedPrecondition", err) + } + if len(resources) != 0 { + t.Errorf("listed %d schemas, want none", len(resources)) + } +} + +// A listing the workspace refuses leaves which schemas exist unknown. Emitting +// nothing would read as deleting them all, and a NotFound is what the SDK reads +// as a securable with no children. +func TestSchemaListFailsInsteadOfEmittingAShortList(t *testing.T) { + for code, want := range map[int]codes.Code{ + http.StatusUnauthorized: codes.FailedPrecondition, + http.StatusForbidden: codes.FailedPrecondition, + http.StatusNotFound: codes.FailedPrecondition, + http.StatusInternalServerError: codes.Unavailable, + } { + f := &fakeSchemaAccount{catalogOwner: "someone", schemaStatus: code} + b := newTestSchemaBuilder(t, f) + + resources, err := listAllSchemas(t, b) + if err == nil { + t.Errorf("List over a schemas listing answering %d returned %d resources and no error", code, len(resources)) + + continue + } + if got := status.Code(err); got != want { + t.Errorf("List over a schemas listing answering %d = %v, want %s", code, err, want) + } + } +} + +// A privilege read a schema's own access cannot account for is refused rather +// than reported: Databricks answers it with 200 and only the rows the caller may +// see, and reporting that subset would revoke the access ConductorOne cannot see. +// Dropping the schema instead would delete it and its whole subtree. +func TestSchemaGrantsRequireFullVisibility(t *testing.T) { + adminOfTheWorkspace := []map[string]any{ + {"principal": map[string]any{"service_principal_name": "app-9"}, "permissions": []string{"ADMIN"}}, + } + manage := func(securable string) map[string][]map[string]any { + return map[string][]map[string]any{ + securable: {{"principal": "data-eng", "principal_id": 11, "privileges": []string{"MANAGE"}}}, + } + } + + tests := []struct { + name string + clientID string + metastoreOwner string + catalogOwner string + schemaOwner string + members any + ownPrivileges map[string][]map[string]any + wantGrants bool + wantMemberLists int + wantFilteredReads []string + }{ + { + name: "owner of the metastore", clientID: "eng", + metastoreOwner: "eng", catalogOwner: "someone", schemaOwner: "someone", + wantGrants: true, + }, + { + name: "owner of the parent catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "app-9", schemaOwner: "someone", + wantGrants: true, + }, + { + name: "owner of the schema", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", schemaOwner: "app-9", + wantGrants: true, + }, + { + name: "manage on the schema, inherited through a group", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", schemaOwner: "someone", + ownPrivileges: manage("schema/main.sales"), + wantGrants: true, wantFilteredReads: []string{"schema/main.sales"}, + }, + { + name: "manage on the parent catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", schemaOwner: "someone", + ownPrivileges: manage("catalog/main"), + wantGrants: true, wantFilteredReads: []string{"schema/main.sales"}, + }, + { + name: "admin of the workspace whose admins group owns the parent catalog", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "_workspace_admins_prod_1", schemaOwner: "someone", + members: adminOfTheWorkspace, + wantGrants: true, wantMemberLists: 1, + }, + { + // Databricks leaves MANAGE out of ALL_PRIVILEGES, so its holder still + // reads a partial list. + name: "all privileges on the schema", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", schemaOwner: "someone", + ownPrivileges: map[string][]map[string]any{ + "schema/main.sales": {{"principal": "app-9", "principal_id": 3, "privileges": []string{"ALL_PRIVILEGES"}}}, + }, + wantGrants: false, wantFilteredReads: []string{"schema/main.sales"}, + }, + { + name: "none of them", clientID: "app-9", + metastoreOwner: "eng", catalogOwner: "someone", schemaOwner: "someone", + wantGrants: false, wantFilteredReads: []string{"schema/main.sales"}, + }, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + f := &fakeSchemaAccount{ + metastoreOwner: test.metastoreOwner, + catalogOwner: test.catalogOwner, + users: map[string]string{"1": "alice@example.com"}, + workspaceMembers: map[string]any{"1": test.members}, + ownPrivileges: test.ownPrivileges, + assignments: map[string][]map[string]any{"main.sales": { + {"principal": "alice@example.com", "principal_id": 1, "privileges": []string{"SELECT"}}, + }}, + } + b := newTestSchemaBuilderAs(t, f, test.clientID) + resource := testSchema(t, "sales", test.schemaOwner, test.catalogOwner, test.metastoreOwner, "wsa") + + grants, _, err := b.Grants(context.Background(), resource, rs.SyncOpAttrs{}) + switch { + case test.wantGrants && err != nil: + t.Fatalf("Grants: %v", err) + case test.wantGrants && len(grants) == 0: + t.Fatal("Grants returned nothing, want the privileges on the schema") + case !test.wantGrants: + if status.Code(err) != codes.PermissionDenied { + t.Fatalf("Grants = %v, want PermissionDenied", err) + } + if len(grants) != 0 { + t.Errorf("grants = %v, want none", grantKeys(grants)) + } + if !strings.Contains(err.Error(), "main.sales") || !strings.Contains(err.Error(), schemaTrustRemedy) { + t.Errorf("error %q does not name the schema and the remedy", err) + } + } + if len(f.memberLists) != test.wantMemberLists { + t.Errorf("workspace assignments listed %d times, want %d", len(f.memberLists), test.wantMemberLists) + } + if !slices.Equal(f.filteredReads, test.wantFilteredReads) { + t.Errorf("own privileges read on %v, want %v", f.filteredReads, test.wantFilteredReads) + } + }) + } +} + +// Databricks no-ops a privilege already held or already absent, so the pre-read +// is the only signal for the idempotency annotations and the echo is the only +// proof the PATCH landed. A principal gone from SCIM still carries its numeric id +// on the row, so the removal goes by principal_id. +func TestSchemaGrantAndRevokeAreIdempotent(t *testing.T) { + f := &fakeSchemaAccount{catalogOwner: "someone", users: map[string]string{"1": "alice@example.com"}} + b := newTestSchemaBuilder(t, f) + g := grant.NewGrant(testSchema(t, "sales", "someone", "someone", testOwnPrincipal, "wsa"), "SELECT", alice.GetId()) + + annos, err := b.Grant(context.Background(), alice, g.GetEntitlement()) + if err != nil { + t.Fatalf("Grant: %v", err) + } + if annos.Contains(&v2.GrantAlreadyExists{}) { + t.Error("a fresh grant reported GrantAlreadyExists") + } + if len(f.patches) != 1 || f.patches[0]["principal"] != "alice@example.com" { + t.Fatalf("patches = %v, want one add for alice@example.com", f.patches) + } + + annos, err = b.Grant(context.Background(), alice, g.GetEntitlement()) + if err != nil || !annos.Contains(&v2.GrantAlreadyExists{}) { + t.Fatalf("second Grant = %v, %v; want GrantAlreadyExists", annos, err) + } + + if _, err := b.Revoke(context.Background(), g); err != nil { + t.Fatalf("Revoke: %v", err) + } + if len(f.patches) != 2 || fmt.Sprint(f.patches[1]["principal_id"]) != "1" { + t.Fatalf("patches = %v, want a removal by principal_id 1", f.patches) + } + + annos, err = b.Revoke(context.Background(), g) + if err != nil || !annos.Contains(&v2.GrantAlreadyRevoked{}) { + t.Fatalf("second Revoke = %v, %v; want GrantAlreadyRevoked", annos, err) + } +} + +// The first workspace can see the catalog and still be unable to list every +// schema. The walk continues, and a later workspace that can list them is the +// one the schemas are read through. +func TestSchemaListContinuesWhenTheFirstWorkspaceCannotListCompletely(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "someone", + // wsa refuses the privilege read, so trust stays unproven there. + permStatus: map[string]int{"wsa": http.StatusForbidden}, + ownPrivileges: map[string][]map[string]any{ + "catalog/main": {{"principal": "app-9", "privileges": []string{"USE_SCHEMA"}}}, + }, + schemas: []map[string]any{{"name": "sales", "owner": "someone"}}, + } + b := newTestSchemaBuilderAs(t, f, "app-9") + + resources, err := listAllSchemas(t, b) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(resources) != 1 { + t.Fatalf("listed %d schemas, want 1 through the second workspace", len(resources)) + } + if workspace, _ := rs.GetProfileStringValue(rs.GetProfile(resources[0]), profileKeyWorkspace); workspace != "wsb" { + t.Errorf("profile[%s] = %q, want wsb", profileKeyWorkspace, workspace) + } +} + +// The admins group that owns the catalog names its own workspace. A direct Admin +// of that workspace reads every privilege, whatever route the schema is listed +// through, so the first workspace already settles trust. +func TestSchemaTrustChecksTheWorkspaceNamedByItsAdminsGroupOwner(t *testing.T) { + f := &fakeSchemaAccount{ + second: true, + metastoreOwner: "eng", + catalogOwner: "_workspace_admins_other_2", + workspaceMembers: map[string]any{"2": []map[string]any{{ + "principal": map[string]any{"service_principal_name": "app-9"}, + "permissions": []string{"ADMIN"}, + }}}, + schemas: []map[string]any{{"name": "sales", "owner": "someone"}}, + } + b := newTestSchemaBuilderAs(t, f, "app-9") + + resources, err := listAllSchemas(t, b) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(resources) != 1 { + t.Fatalf("listed %d schemas, want 1", len(resources)) + } + if workspace, _ := rs.GetProfileStringValue(rs.GetProfile(resources[0]), profileKeyWorkspace); workspace != "wsa" { + t.Errorf("profile[%s] = %q, want wsa", profileKeyWorkspace, workspace) + } +} + +// A dropped schema, or a dropped parent catalog, leaves nothing to revoke. Only +// the error_code proves it, and without it on schemaLevel the revoke task fails +// and the grant is never marked revoked. +func TestSchemaRevokeOnADroppedSchemaIsAlreadyRevoked(t *testing.T) { + for _, errorCode := range []string{errorCodeSchemaGone, errorCodeCatalogGone} { + t.Run(errorCode, func(t *testing.T) { + f := &fakeSchemaAccount{ + catalogOwner: "someone", + users: map[string]string{"1": "alice@example.com"}, + permStatus: map[string]int{"wsa": http.StatusNotFound}, + permErrorCode: errorCode, + } + b := newTestSchemaBuilder(t, f) + g := grant.NewGrant(testSchema(t, "sales", "someone", "someone", testOwnPrincipal, "wsa"), "SELECT", alice.GetId()) + + annos, err := b.Revoke(context.Background(), g) + if err != nil || !annos.Contains(&v2.GrantAlreadyRevoked{}) { + t.Fatalf("Revoke = %v, %v; want GrantAlreadyRevoked", annos, err) + } + if len(f.patches) != 0 { + t.Errorf("patches = %v, want none", f.patches) + } + }) + } +} diff --git a/pkg/connector/securable_helpers.go b/pkg/connector/securable_helpers.go index 9e4e1bad..bec88075 100644 --- a/pkg/connector/securable_helpers.go +++ b/pkg/connector/securable_helpers.go @@ -83,6 +83,10 @@ type securableTrust struct { // metastoreOwner is empty on a metastore, whose own owner is in owner. metastoreOwner string + // parentOwner is the owner of the securable one level up, which only a + // schema has: the levels above it hang from the metastore directly. + parentOwner string + fullName string workspace string workspaceID string @@ -104,17 +108,21 @@ func ensureGrantsAreTrustworthy( owner, _ := rs.GetProfileStringValue(profile, profileKeyOwner) metastoreOwner, _ := rs.GetProfileStringValue(profile, profileKeyMetastoreOwner) workspaceID, _ := rs.GetProfileStringValue(profile, profileKeyWorkspaceID) + // Only a schema hangs from a securable other than the metastore, so at the + // other levels this key is absent and the owner above is the metastore's. + parentOwner, _ := rs.GetProfileStringValue(profile, profileKeyCatalogOwner) trustworthy, err := level.trusts(ctx, client, securableTrust{ ownPrincipal: clientID, owner: owner, metastoreOwner: metastoreOwner, + parentOwner: parentOwner, fullName: fullName, workspace: workspace, workspaceID: workspaceID, }, annos) if err != nil { - return err + return trustReadError(level, fullName, workspace, err) } if !trustworthy { return status.Errorf(codes.PermissionDenied, @@ -125,6 +133,22 @@ func ensureGrantsAreTrustworthy( return nil } +// trustReadError classifies the 404 a trust read left for the level. A proven +// absence reports errSecurableGone; an unproven one is refused, because read as +// an empty result ConductorOne deletes everything the securable held. +func trustReadError(level securableLevel, fullName, workspace string, err error) error { + if !errors.Is(err, errTrustRead404) { + return err + } + if databricks.IsSecurableGone(err, level.goneErrorCodes...) { + return errSecurableGone + } + + return refuseUnreadable(fmt.Sprintf( + "databricks-connector: cannot read whether every privilege on %s %s is visible through workspace %s", + level.noun, fullName, workspace), err) +} + // syncSecurableGrants emits the privileges held on one securable, plus its owner. // The workspace comes off the profile the last List wrote: every Unity Catalog // read is addressed to a workspace host, and resolving one here would be an @@ -331,7 +355,7 @@ func (c change) updateError(verb string, err error) error { // grantSecurable adds one privilege. Databricks no-ops a PATCH it cannot apply, // so the pre-read decides GrantAlreadyExists and the echo is the only proof the -// change landed. +// change landed. A securable the trust check proves gone answers NotFound. func (o securableOps) grant( ctx context.Context, principal *v2.ResourceId, @@ -341,6 +365,11 @@ func (o securableOps) grant( annos := annotations.Annotations{} change, err := resolveChange(ctx, client, clientID, level, principal, entitlement, true, &annos) if err != nil { + if errors.Is(err, errSecurableGone) || isSecurableGoneSentinel(err) { + return annos, status.Errorf(codes.NotFound, "databricks-connector: %s %s no longer exists", + level.noun, entitlement.GetResource().GetId().GetResource()) + } + return annos, err } @@ -378,7 +407,8 @@ func (o securableOps) grant( // revokeSecurable removes one privilege. Databricks no-ops a privilege already // absent, so the pre-read decides GrantAlreadyRevoked and the echo confirms the -// removal. +// removal. A securable the trust check proves gone holds nothing to revoke, so +// it reports the grant already revoked rather than an unreadable securable. func (o securableOps) revoke( ctx context.Context, principal *v2.ResourceId, @@ -388,6 +418,12 @@ func (o securableOps) revoke( annos := annotations.Annotations{} change, err := resolveChange(ctx, client, clientID, level, principal, entitlement, false, &annos) if err != nil { + if errors.Is(err, errSecurableGone) || isSecurableGoneSentinel(err) { + annos.Update(&v2.GrantAlreadyRevoked{}) + + return annos, nil + } + return annos, err } @@ -439,3 +475,9 @@ func (o securableOps) revoke( return annos, nil } + +// isSecurableGoneSentinel reports whether err is one of the per-level sentinels +// for a securable Databricks named as missing. +func isSecurableGoneSentinel(err error) bool { + return errors.Is(err, errCatalogGone) || errors.Is(err, errMetastoreGone) +} diff --git a/pkg/databricks/models.go b/pkg/databricks/models.go index fc8edf26..e20a9c07 100644 --- a/pkg/databricks/models.go +++ b/pkg/databricks/models.go @@ -1,6 +1,9 @@ package databricks -import "encoding/json" +import ( + "encoding/json" + "slices" +) type BaseResponse struct { ID string `json:"id"` @@ -182,6 +185,11 @@ type Catalog struct { IsolationMode string `json:"isolation_mode"` } +type Schema struct { + Name string `json:"name"` + Owner string `json:"owner"` +} + type PrivilegeAssignment struct { Principal string `json:"principal"` PrincipalID json.Number `json:"principal_id"` @@ -223,7 +231,42 @@ type catalogsResponse struct { NextPageToken string `json:"next_page_token"` } +type schemasResponse struct { + Schemas []Schema `json:"schemas"` + NextPageToken string `json:"next_page_token"` +} + type permissionsResponse struct { PrivilegeAssignments []PrivilegeAssignment `json:"privilege_assignments"` NextPageToken string `json:"next_page_token"` } + +// EffectivePrivilege is one privilege an effective-permissions answer reports. +// The inherited fields are empty when it is held on the securable itself, and +// otherwise name the parent that grants it. +type EffectivePrivilege struct { + Privilege string `json:"privilege"` + InheritedFromType string `json:"inherited_from_type,omitempty"` + InheritedFromName string `json:"inherited_from_name,omitempty"` +} + +// EffectivePrivilegeAssignment is what one principal holds on a securable once +// inheritance is applied. It carries no principal_id: this endpoint names a +// principal only. +type EffectivePrivilegeAssignment struct { + Principal string `json:"principal"` + Privileges []EffectivePrivilege `json:"privileges"` +} + +// HoldsAny reports whether the assignment carries one of wanted, counting a +// privilege a parent grants the same as one held directly. +func (a EffectivePrivilegeAssignment) HoldsAny(wanted []string) bool { + return slices.ContainsFunc(a.Privileges, func(held EffectivePrivilege) bool { + return slices.Contains(wanted, held.Privilege) + }) +} + +type effectivePermissionsResponse struct { + PrivilegeAssignments []EffectivePrivilegeAssignment `json:"privilege_assignments"` + NextPageToken string `json:"next_page_token"` +} diff --git a/pkg/databricks/unity_catalog.go b/pkg/databricks/unity_catalog.go index ea8703dc..339a8737 100644 --- a/pkg/databricks/unity_catalog.go +++ b/pkg/databricks/unity_catalog.go @@ -23,15 +23,24 @@ const ( // https://docs.databricks.com/api/workspace/catalogs/list unityCatalogCatalogsEndpoint = "/api/2.1/unity-catalog/catalogs" + // https://docs.databricks.com/api/workspace/schemas/list + unityCatalogSchemasEndpoint = "/api/2.1/unity-catalog/schemas" + // GET https://docs.databricks.com/api/workspace/grants/get // PATCH https://docs.databricks.com/api/workspace/grants/update unityCatalogPermissionsEndpoint = "/api/2.1/unity-catalog/permissions" + // GET https://docs.databricks.com/api/workspace/grants/geteffective + // Same securable paths and same error codes as the permissions endpoint, + // plus the privileges a parent grants, naming where each comes from. + unityCatalogEffectivePermissionsEndpoint = "/api/2.1/unity-catalog/effective-permissions" + // Path segments for the permissions endpoint, not interchangeable with the // uppercase securable_type a listed securable carries. A metastore is // addressed by its UUID: its name is rejected. SecurableMetastore = "metastore" SecurableCatalog = "catalog" + SecurableSchema = "schema" // The permissions endpoint rejects any max_results from 1 to 149 with a 400. // Zero selects the paginated response that replaces the unpaginated form. @@ -154,6 +163,35 @@ func (c *Client) GetCatalog( return &res, ratelimitData, nil } +// ListSchemas (GET /api/2.1/unity-catalog/schemas?catalog_name=) reads one page of +// the schemas of catalogName, through workspaceId, which is a deployment name. +// Unlike the catalogs listing, this one pages, so the cursor is passed through. +func (c *Client) ListSchemas( + ctx context.Context, + workspaceId, catalogName, pageToken string, + maxResults uint, +) ( + []Schema, + string, + *v2.RateLimitDescription, + error, +) { + var res schemasResponse + + u := c.workspaceUrl(workspaceId).JoinPath(unityCatalogSchemasEndpoint) + ratelimitData, err := c.Get(ctx, u, &res, &unityCatalogVars{ + catalogName: catalogName, + maxResults: &maxResults, + pageToken: pageToken, + }) + if err != nil { + return nil, "", ratelimitData, nameWorkspace403Remedy(workspaceId, + fmt.Errorf("failed to list schemas in catalog %s: %w", catalogName, err)) + } + + return res.Schemas, res.NextPageToken, ratelimitData, nil +} + // ListPermissions (GET /api/2.1/unity-catalog/permissions/{securable_type}/{full_name}). // The principal query filters effectively, not literally: an inherited assignment // comes back naming the group it is inherited through, so a caller must read every @@ -180,7 +218,7 @@ func (c *Client) listPermissions( *v2.RateLimitDescription, error, ) { - u, err := c.permissionsUrl(workspaceId, securableType, fullName) + u, err := c.permissionsUrl(unityCatalogPermissionsEndpoint, workspaceId, securableType, fullName) if err != nil { return nil, "", nil, err } @@ -238,6 +276,56 @@ func (c *Client) ForEachUncachedPermissionsPage( }, visit) } +// ForEachUncachedEffectivePermissionsPage walks the effective-permissions +// endpoint past the cache, without retaining the pages. One read covers the +// securable and every parent, so a caller does not walk the hierarchy itself. +func (c *Client) ForEachUncachedEffectivePermissionsPage( + ctx context.Context, + workspaceId, securableType, fullName, principal string, + visit func([]EffectivePrivilegeAssignment) (bool, error), +) (*v2.RateLimitDescription, error) { + return forEachPageFrom(ctx, "", func(pageToken string) ([]EffectivePrivilegeAssignment, string, *v2.RateLimitDescription, error) { + return c.listEffectivePermissions(ctx, workspaceId, securableType, fullName, principal, pageToken) + }, visit) +} + +// listEffectivePermissions (GET /api/2.1/unity-catalog/effective-permissions/{type}/{full_name}). +// The principal query filters effectively, not literally, so a caller reads every +// row's privileges. Never cached: a stale page would answer for a changed grant. +func (c *Client) listEffectivePermissions( + ctx context.Context, + workspaceId, securableType, fullName, principal, pageToken string, +) ( + []EffectivePrivilegeAssignment, + string, + *v2.RateLimitDescription, + error, +) { + u, err := c.permissionsUrl(unityCatalogEffectivePermissionsEndpoint, workspaceId, securableType, fullName) + if err != nil { + return nil, "", nil, err + } + + var res effectivePermissionsResponse + + maxResults := permissionsMaxResults + ratelimitData, err := c.GetUncached(ctx, u, &res, &unityCatalogVars{ + principal: principal, + maxResults: &maxResults, + pageToken: pageToken, + }) + if err != nil { + if principal != "" { + return nil, "", ratelimitData, fmt.Errorf("failed to list effective permissions for principal %s on %s %s: %w", + principal, securableType, fullName, err) + } + + return nil, "", ratelimitData, fmt.Errorf("failed to list effective permissions on %s %s: %w", securableType, fullName, err) + } + + return res.PrivilegeAssignments, res.NextPageToken, ratelimitData, nil +} + // UpdatePermissions (PATCH /api/2.1/unity-catalog/permissions/{securable_type}/{full_name}). // Databricks no-ops a privilege already held or already absent, so the echoed // assignments are the only confirmation the change landed — and that echo is @@ -290,7 +378,7 @@ func (c *Client) updatePermissions( } } - u, err := c.permissionsUrl(workspaceId, securableType, fullName) + u, err := c.permissionsUrl(unityCatalogPermissionsEndpoint, workspaceId, securableType, fullName) if err != nil { return nil, err } @@ -349,12 +437,12 @@ func (c *Client) updatePermissions( // A name that does not survive URL assembly addresses a different securable, // which on a PATCH writes someone else's permissions. -func (c *Client) permissionsUrl(workspaceId, securableType, fullName string) (*url.URL, error) { +func (c *Client) permissionsUrl(endpoint, workspaceId, securableType, fullName string) (*url.URL, error) { if securableType == "" || fullName == "" { return nil, invalidArgument("securable type and full name are required, got %q and %q", securableType, fullName) } - u := c.workspaceUrl(workspaceId).JoinPath(unityCatalogPermissionsEndpoint, securableType, fullName) + u := c.workspaceUrl(workspaceId).JoinPath(endpoint, securableType, fullName) if !addressesSecurable(u, "/"+securableType+"/"+fullName) { return nil, invalidArgument("cannot address %s %q: the name does not survive URL assembly", securableType, fullName) } diff --git a/pkg/databricks/vars.go b/pkg/databricks/vars.go index 7525e62a..6fac93ad 100644 --- a/pkg/databricks/vars.go +++ b/pkg/databricks/vars.go @@ -173,6 +173,7 @@ func NewNameVars(name string, etag string) *NameVars { // while an absent max_results asks for the deprecated unpaginated response, so // maxResults is a pointer: its presence is the signal. type unityCatalogVars struct { + catalogName string principal string maxResults *uint pageToken string @@ -180,6 +181,10 @@ type unityCatalogVars struct { } func (v *unityCatalogVars) Apply(params *url.Values) { + if v.catalogName != "" { + params.Add("catalog_name", v.catalogName) + } + if v.principal != "" { params.Add("principal", v.principal) } From 068e05dc744530c84cb752021fb720580261e36c Mon Sep 17 00:00:00 2001 From: Mateo Hernandez Date: Tue, 3 Nov 2026 17:10:16 -0300 Subject: [PATCH 2/2] fix(databricks): treat a securable proven gone as an empty grant list The trust read is the first call that can prove a securable gone, and it runs before the permissions read. syncGrants returned its error as is, so a schema deleted between List and Grants failed the sync with the bare errSecurableGone sentinel: codes.Unknown, no securable name. Twelve lines below, the permissions read already treats that same deletion as an empty result rather than a failure. The two reads see the same fact and now answer the same way. A dropped parent catalog reaches the same place: the trust probe reads the schema's effective permissions, a dropped catalog answers 404 with CATALOG_DOES_NOT_EXIST, and trustReadError classifies it with the level's full gone-code set. The tests use a principal that owns neither the metastore nor the catalog. With ownership the trust check short-circuits and never reaches the probe, which is why the existing dropped-schema test did not cover this. Also removes a docs bullet that said the built-in system catalogs are synced while the connector owns the metastore. Catalog List skips them unconditionally, as the bullet three lines above already says. Co-Authored-By: Claude Opus 5 --- docs/docs-info.md | 1 - pkg/connector/schema_test.go | 51 ++++++++++++++++++++++++++++++ pkg/connector/securable_helpers.go | 9 ++++++ 3 files changed, 60 insertions(+), 1 deletion(-) diff --git a/docs/docs-info.md b/docs/docs-info.md index 64160900..31b6f469 100644 --- a/docs/docs-info.md +++ b/docs/docs-info.md @@ -65,5 +65,4 @@ While developing the connector, please fill out this form. This information is n - **Catalog discovery visits every attached workspace in scope.** The connector asks each running workspace attached to the metastore, including for `ISOLATED` catalogs bound to only some workspaces. Each catalog name is kept once per metastore. A catalog bound only to workspaces the sync excludes is not listed, and ConductorOne removes it with its grants. - **System catalogs are excluded regardless of ownership.** The built-in `system` catalog and any catalog with `CatalogType == "SYSTEM_CATALOG"` are excluded, including `samples` when it has that type. - **Seeing a catalog needs more than `MANAGE`.** The list holds the catalogs the service principal owns or holds `USE_CATALOG` on, or every catalog when it is the metastore admin. `MANAGE` alone reads a catalog's privileges but does not make it appear. -- **Built-in catalogs.** `system` and any `SYSTEM_CATALOG`, including `samples`, are synced while the service principal owns the metastore. When it does not, and the privilege list cannot be proven complete, those catalogs are left out. Databricks refuses `MANAGE` on them. Leaving them out after they were synced removes them in ConductorOne. - **Not synced yet:** tables, volumes, functions, registered models, external locations, storage credentials, connections, shares, recipients, and tags. diff --git a/pkg/connector/schema_test.go b/pkg/connector/schema_test.go index ecef53d4..84bc5ca5 100644 --- a/pkg/connector/schema_test.go +++ b/pkg/connector/schema_test.go @@ -794,3 +794,54 @@ func TestSchemaRevokeOnADroppedSchemaIsAlreadyRevoked(t *testing.T) { }) } } + +// The trust probe is the first read that can prove a schema gone, and it runs +// before the permissions read. A schema deleted between List and Grants must +// not fail the sync at the earlier of the two reads that see the same deletion. +// The principal owns nothing here, so the probe actually runs: with ownership +// the trust check short-circuits and never reaches it. +func TestSchemaGrantsOnADroppedSecurableDoesNotFailTheSync(t *testing.T) { + for _, errorCode := range []string{errorCodeSchemaGone, errorCodeCatalogGone} { + t.Run(errorCode, func(t *testing.T) { + f := &fakeSchemaAccount{ + metastoreOwner: "other-team", + catalogOwner: "someone", + permStatus: map[string]int{"wsa": http.StatusNotFound}, + permErrorCode: errorCode, + } + b := newTestSchemaBuilderAs(t, f, testOwnPrincipal) + + grants, _, err := b.Grants(context.Background(), + testSchema(t, "sales", "someone", "someone", "other-team", "wsa"), rs.SyncOpAttrs{}) + if err != nil { + t.Fatalf("Grants = %v, want no error for a securable proven gone", err) + } + if len(grants) != 0 { + t.Errorf("emitted %d grants, want none", len(grants)) + } + }) + } +} + +// Revoke on a schema whose parent catalog was dropped reports the grant already +// revoked for a principal that owns neither the metastore nor the catalog, so +// the trust probe is what answers rather than the ownership short-circuit. +func TestSchemaRevokeOnADroppedParentIsAlreadyRevokedForANonOwner(t *testing.T) { + f := &fakeSchemaAccount{ + metastoreOwner: "other-team", + catalogOwner: "someone", + users: map[string]string{"1": "alice@example.com"}, + permStatus: map[string]int{"wsa": http.StatusNotFound}, + permErrorCode: errorCodeCatalogGone, + } + b := newTestSchemaBuilderAs(t, f, testOwnPrincipal) + g := grant.NewGrant(testSchema(t, "sales", "someone", "someone", "other-team", "wsa"), "SELECT", alice.GetId()) + + annos, err := b.Revoke(context.Background(), g) + if err != nil || !annos.Contains(&v2.GrantAlreadyRevoked{}) { + t.Fatalf("Revoke = %v, %v; want GrantAlreadyRevoked", annos, err) + } + if len(f.patches) != 0 { + t.Errorf("patches = %v, want none", f.patches) + } +} diff --git a/pkg/connector/securable_helpers.go b/pkg/connector/securable_helpers.go index bec88075..f9969364 100644 --- a/pkg/connector/securable_helpers.go +++ b/pkg/connector/securable_helpers.go @@ -179,6 +179,15 @@ func (o securableOps) syncGrants( // partial, so Grants does not prove it again. if pageToken == "" && level.securableType != databricks.SecurableMetastore && level.securableType != databricks.SecurableCatalog { if err := ensureGrantsAreTrustworthy(ctx, client, clientID, level, resource, fullName, workspace, &annos); err != nil { + // The trust read is the first call that can prove the securable gone, + // and one deleted between List and this read is not a failed sync. + // The permissions read below answers the same way, so answering + // differently here would fail the sync on the earlier of two reads + // that see the same deletion. + if errors.Is(err, errSecurableGone) || isSecurableGoneSentinel(err) { + return nil, &rs.SyncOpResults{Annotations: annos}, nil + } + return nil, &rs.SyncOpResults{Annotations: annos}, err } }