Merge #149794

149794: changefeedccl: add privilege check for database-level changefeed r=andyyang890,aerfrei a=KeithCh

Only a user with changefeed privileges on a database can create an
enterprise changefeed on the database. This also allows them to create
an enterprise changefeed on any table in that database. Creation of core
changefeeds at the table-level will still require the select privilege 
on all tables being watched.

resolves #149470

Release note: None

Co-authored-by: Keith Chow <[email protected]>

Author: craig[bot]

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -683,7 +683,8 @@ func generateAndValidateNewTargets(
 		hasSelectPrivOnAllTables = hasSelectPrivOnAllTables && hasSelect
 		hasChangefeedPrivOnAllTables = hasChangefeedPrivOnAllTables && hasChangefeed
 	}
-	if err := authorizeUserToCreateChangefeed(ctx, p, sinkURI, hasSelectPrivOnAllTables, hasChangefeedPrivOnAllTables); err != nil {
+	if err := authorizeUserToCreateChangefeed(
+		ctx, p, sinkURI, hasSelectPrivOnAllTables, hasChangefeedPrivOnAllTables, tree.ChangefeedLevelTable); err != nil {
 		return nil, nil, hlc.Timestamp{}, nil, err
 	}
 

pkg/ccl/changefeedccl/alter_changefeed_test.go

@@ -1882,7 +1882,7 @@ func TestAlterChangefeedAccessControl(t *testing.T) {
 		})
 		// jobController can access the job, but will hit an error re-creating the changefeed.
 		asUser(t, f, `jobController`, func(userDB *sqlutils.SQLRunner) {
-			userDB.ExpectErr(t, "pq: user jobcontroller requires the CHANGEFEED privilege on all target tables to be able to run an enterprise changefeed", fmt.Sprintf(`ALTER CHANGEFEED %d DROP table_b`, currentFeed.JobID()))
+			userDB.ExpectErr(t, `pq: user "jobcontroller" requires the CHANGEFEED privilege on all target tables to be able to run an enterprise changefeed`, fmt.Sprintf(`ALTER CHANGEFEED %d DROP table_b`, currentFeed.JobID()))
 		})
 		asUser(t, f, `userWithSomeGrants`, func(userDB *sqlutils.SQLRunner) {
 			userDB.ExpectErr(t, "does not have privileges for job", fmt.Sprintf(`ALTER CHANGEFEED %d ADD table_b`, currentFeed.JobID()))

pkg/ccl/changefeedccl/authorization.go

@@ -18,15 +18,25 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgnotice"
 	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/roleoption"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilege"
 	"github.com/cockroachdb/errors"
 )
 
 func checkPrivilegesForDescriptor(
 	ctx context.Context, p sql.PlanHookState, desc catalog.Descriptor,
 ) (hasSelect bool, hasChangefeed bool, err error) {
-	if desc.GetObjectType() != privilege.Table {
-		return false, false, errors.AssertionFailedf("expected descriptor %d to be a table descriptor, found: %s ", desc.GetID(), desc.GetObjectType())
+	if desc == nil {
+		return false, false, errors.AssertionFailedf("expected descriptor to be non-nil")
+	}
+	switch desc.GetObjectType() {
+	case privilege.Database, privilege.Table:
+	default:
+		return false, false, errors.AssertionFailedf(
+			"expected descriptor for %q to be a table or database descriptor, found: %s ",
+			desc.GetName(),
+			desc.GetObjectType(),
+		)
 	}
 
 	hasSelect, err = p.HasPrivilege(ctx, desc, privilege.SELECT, p.User())
@@ -60,6 +70,7 @@ func authorizeUserToCreateChangefeed(
 	sinkURI string,
 	hasSelectPrivOnAllTables bool,
 	hasChangefeedPrivOnAllTables bool,
+	changefeedLevel tree.ChangefeedLevel,
 	otherExternalURIs ...string,
 ) error {
 	isAdmin, err := p.HasAdminRole(ctx)
@@ -94,10 +105,17 @@ func authorizeUserToCreateChangefeed(
 		return nil
 	}
 
+	requiredPrivilegeTarget := func() string {
+		if changefeedLevel == tree.ChangefeedLevelDatabase {
+			return "the target database"
+		}
+		return "all target tables"
+	}()
+
 	if !hasChangefeedPrivOnAllTables {
 		return pgerror.Newf(pgcode.InsufficientPrivilege,
-			`user %s requires the %s privilege on all target tables to be able to run an enterprise changefeed`,
-			p.User(), privilege.CHANGEFEED)
+			`user %q requires the %s privilege on %s to be able to run an enterprise changefeed`,
+			p.User(), privilege.CHANGEFEED, requiredPrivilegeTarget)
 	}
 
 	enforceExternalConnections := changefeedbase.RequireExternalConnectionSink.Get(&p.ExecCfg().Settings.SV)
@@ -124,8 +142,8 @@ func authorizeUserToCreateChangefeed(
 			} else {
 				return pgerror.Newf(
 					pgcode.InsufficientPrivilege,
-					`the %s privilege on all tables can only be used with external connection sinks. see cluster setting %s`,
-					privilege.CHANGEFEED, changefeedbase.RequireExternalConnectionSink.Name(),
+					`the %s privilege on %s can only be used with external connection sinks. see cluster setting %s`,
+					privilege.CHANGEFEED, requiredPrivilegeTarget, changefeedbase.RequireExternalConnectionSink.Name(),
 				)
 			}
 		}

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -484,13 +484,14 @@ func evalCursor(
 
 func getTargetList(changefeedStmt *annotatedChangefeedStatement) (*tree.BackupTargetList, error) {
 	targetList := tree.BackupTargetList{}
-	if changefeedStmt.Level == tree.ChangefeedLevelTable {
+	switch changefeedStmt.Level {
+	case tree.ChangefeedLevelTable:
 		for _, t := range changefeedStmt.TableTargets {
 			targetList.Tables.TablePatterns = append(targetList.Tables.TablePatterns, t.TableName)
 		}
-	} else if changefeedStmt.Level == tree.ChangefeedLevelDatabase {
+	case tree.ChangefeedLevelDatabase:
 		targetList.Databases = tree.NameList{tree.Name(changefeedStmt.DatabaseTarget)}
-	} else {
+	default:
 		return nil, errors.AssertionFailedf("unknown changefeed level: %s", changefeedStmt.Level.String())
 	}
 	return &targetList, nil
@@ -570,11 +571,23 @@ func createChangefeedJobRecord(
 
 	var details jobspb.ChangefeedDetails
 
-	tableNameToDescriptor, targetDatabaseDescs, err := getTargetDescriptors(ctx, p, targetList, statementTime, initialHighWater)
+	tableNameToDescriptor, targetDatabaseDescs, tableAndParentDescs, err := getTargetDescriptors(
+		ctx,
+		p,
+		targetList,
+		statementTime,
+		initialHighWater,
+	)
 
 	if err != nil {
 		return nil, changefeedbase.Targets{}, err
 	}
+	if len(tableAndParentDescs) == 0 {
+		return nil, changefeedbase.Targets{}, errors.AssertionFailedf("expected at least one descriptor")
+	}
+	if len(targetDatabaseDescs) > 1 {
+		return nil, changefeedbase.Targets{}, errors.AssertionFailedf("expected at most one database descriptor, got %d", len(targetDatabaseDescs))
+	}
 
 	for _, t := range tableNameToDescriptor {
 		if tbl, ok := t.(catalog.TableDescriptor); ok && tbl.ExternalRowData() != nil {
@@ -640,25 +653,65 @@ func createChangefeedJobRecord(
 	hasSelectPrivOnAllTables := true
 	hasChangefeedPrivOnAllTables := true
 	tolerances := opts.GetCanHandle()
-	for _, desc := range tableNameToDescriptor {
-		if table, isTable := desc.(catalog.TableDescriptor); isTable {
-			if err := changefeedvalidators.ValidateTable(targets, table, tolerances); err != nil {
-				return nil, changefeedbase.Targets{}, err
-			}
-			for _, warning := range changefeedvalidators.WarningsForTable(table, tolerances) {
-				p.BufferClientNotice(ctx, pgnotice.Newf("%s", warning))
-			}
+	// Core changefeed:
+	//	- Table-level changefeeds require the user to have SELECT privileges
+	// 	  on all target tables
+	//	- DB-level feeds are not supported
+	// Enterprise changefeed:
+	//  - Table-level feeds require the CHANGEFEED privilege on all target tables
+	//  - DB-level feeds require the CHANGEFEED privilege on the target database
+	switch changefeedStmt.Level {
+	case tree.ChangefeedLevelTable:
+		_, tableToDatabaseLookup := buildTableToDatabaseAndSchemaLookup(tableAndParentDescs)
+		for _, desc := range tableNameToDescriptor {
+			if table, isTable := desc.(catalog.TableDescriptor); isTable {
+				if err := changefeedvalidators.ValidateTable(targets, table, tolerances); err != nil {
+					return nil, changefeedbase.Targets{}, err
+				}
+				for _, warning := range changefeedvalidators.WarningsForTable(table, tolerances) {
+					p.BufferClientNotice(ctx, pgnotice.Newf("%s", warning))
+				}
 
-			hasSelect, hasChangefeed, err := checkPrivilegesForDescriptor(ctx, p, desc)
-			if err != nil {
-				return nil, changefeedbase.Targets{}, err
+				hasSelect, hasChangefeed, err := checkPrivilegesForDescriptor(ctx, p, desc)
+				if err != nil {
+					return nil, changefeedbase.Targets{}, err
+				}
+
+				databaseDesc, ok := tableToDatabaseLookup[table.GetID()]
+				if !ok {
+					return nil, changefeedbase.Targets{}, errors.AssertionFailedf("expected to find a database descriptor for table %s", table.GetName())
+				}
+				if !hasChangefeed {
+					_, hasChangefeed, err = checkPrivilegesForDescriptor(ctx, p, databaseDesc)
+					if err != nil {
+						return nil, changefeedbase.Targets{}, err
+					}
+				}
+
+				hasSelectPrivOnAllTables = hasSelectPrivOnAllTables && hasSelect
+				hasChangefeedPrivOnAllTables = hasChangefeedPrivOnAllTables && hasChangefeed
 			}
-			hasSelectPrivOnAllTables = hasSelectPrivOnAllTables && hasSelect
-			hasChangefeedPrivOnAllTables = hasChangefeedPrivOnAllTables && hasChangefeed
 		}
+	case tree.ChangefeedLevelDatabase:
+		_, hasDatabaseChangefeedPriv, err := checkPrivilegesForDescriptor(ctx, p, targetDatabaseDescs[0])
+		if err != nil {
+			return nil, changefeedbase.Targets{}, err
+		}
+		hasChangefeedPrivOnAllTables = hasDatabaseChangefeedPriv
+	default:
+		return nil, changefeedbase.Targets{}, errors.AssertionFailedf("unknown changefeed level: %s", changefeedStmt.Level)
 	}
+
 	if checkPrivs {
-		if err := authorizeUserToCreateChangefeed(ctx, p, sinkURI, hasSelectPrivOnAllTables, hasChangefeedPrivOnAllTables, opts.GetConfluentSchemaRegistry()); err != nil {
+		if err := authorizeUserToCreateChangefeed(
+			ctx,
+			p,
+			sinkURI,
+			hasSelectPrivOnAllTables,
+			hasChangefeedPrivOnAllTables,
+			changefeedStmt.Level,
+			opts.GetConfluentSchemaRegistry(),
+		); err != nil {
 			return nil, changefeedbase.Targets{}, err
 		}
 	}
@@ -984,22 +1037,24 @@ func getTargetDescriptors(
 ) (
 	tableNameToDescriptor map[tree.TablePattern]catalog.Descriptor,
 	databaseDescs []catalog.DatabaseDescriptor,
+	tableAndParentDescs []catalog.Descriptor,
 	err error,
 ) {
 	if len(targets.Databases) > 0 && len(targets.Tables.TablePatterns) > 0 {
-		return nil, nil, errors.Errorf(`CHANGEFEED cannot target both databases and tables`)
+		return nil, nil, nil, errors.Errorf(`CHANGEFEED cannot target both databases and tables`)
 	}
+
 	for _, t := range targets.Tables.TablePatterns {
 		p, err := t.NormalizeTablePattern()
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 		if _, ok := p.(*tree.TableName); !ok {
-			return nil, nil, errors.Errorf(`CHANGEFEED cannot target %s`, tree.AsString(t))
+			return nil, nil, nil, errors.Errorf(`CHANGEFEED cannot target %s`, tree.AsString(t))
 		}
 	}
-
-	_, _, targetDatabaseDescs, targetTableDescs, err := backupresolver.ResolveTargetsToDescriptors(ctx, p, statementTime, targets)
+	// targetTableDescs is empty if the targets are not tables, targetDatabaseDescs is empty if the targets are not databases
+	targetAndParentDescs, _, targetDatabaseDescs, targetTableDescs, err := backupresolver.ResolveTargetsToDescriptors(ctx, p, statementTime, targets)
 	if err != nil {
 		var m *backupresolver.MissingTableErr
 		if errors.As(err, &m) {
@@ -1013,7 +1068,7 @@ func getTargetDescriptors(
 				"do the targets exist at the specified cursor time %s?", initialHighWater)
 		}
 	}
-	return targetTableDescs, targetDatabaseDescs, err
+	return targetTableDescs, targetDatabaseDescs, targetAndParentDescs, err
 }
 
 func getTargetsAndTables(
@@ -1217,8 +1272,13 @@ func makeChangefeedDescription(
 	opts changefeedbase.StatementOptions,
 ) (string, error) {
 	c := &tree.CreateChangefeed{
-		TableTargets: changefeed.TableTargets,
-		Select:       changefeed.Select,
+		Select: changefeed.Select,
+		Level:  changefeed.Level,
+	}
+	if changefeed.Level == tree.ChangefeedLevelDatabase {
+		c.DatabaseTarget = changefeed.DatabaseTarget
+	} else {
+		c.TableTargets = changefeed.TableTargets
 	}
 
 	if sinkURI != "" {
@@ -2124,3 +2184,34 @@ func getChangefeedEventMigrator(migrateEvent bool) log.StructuredEventMigrator {
 		channel.CHANGEFEED,
 	)
 }
+
+func buildTableToDatabaseAndSchemaLookup(
+	targetAndParentDescs []catalog.Descriptor,
+) (
+	tableToSchema map[descpb.ID]catalog.SchemaDescriptor,
+	tableToDatabase map[descpb.ID]catalog.DatabaseDescriptor,
+) {
+	// getSchema maps schema IDs to its schema descriptor.
+	getSchema := make(map[descpb.ID]catalog.SchemaDescriptor)
+	// getDatabase maps database IDs to its database descriptor.
+	getDatabase := make(map[descpb.ID]catalog.DatabaseDescriptor)
+	for _, desc := range targetAndParentDescs {
+		switch desc.DescriptorType() {
+		case catalog.Schema:
+			getSchema[desc.GetID()] = desc.(catalog.SchemaDescriptor)
+		case catalog.Database:
+			getDatabase[desc.GetID()] = desc.(catalog.DatabaseDescriptor)
+		}
+	}
+	// tableToSchema maps table IDs to its parent schema descriptor.
+	tableToSchema = make(map[descpb.ID]catalog.SchemaDescriptor)
+	// tableToDatabase maps table IDs to its parent database descriptor.
+	tableToDatabase = make(map[descpb.ID]catalog.DatabaseDescriptor)
+	for _, desc := range targetAndParentDescs {
+		if desc.DescriptorType() == catalog.Table {
+			tableToDatabase[desc.GetID()] = getDatabase[desc.GetParentID()]
+			tableToSchema[desc.GetID()] = getSchema[desc.GetParentSchemaID()]
+		}
+	}
+	return tableToSchema, tableToDatabase
+}