sql: add CONTROLCHANGEFEED role option

Fixes cockroachdb#52869

Release note (sql change): Introduced a new CONTROLCHANGEFEED role
option. This grants non-admin roles the ability to create new
changefeeds, as long as they have SELECT privileges on the target table.
It can be conferred via `ALTER ROLE <role> CONTROLCHANGEFEED` and
revoked via `ALTER ROLE <role> CONTROLCHANGEFEED`.

docs/generated/sql/bnf/stmt_block.bnf

@@ -729,6 +729,7 @@ unreserved_keyword ::=
 	| 'CONFIGURATIONS'
 	| 'CONFIGURE'
 	| 'CONSTRAINTS'
+	| 'CONTROLCHANGEFEED'
 	| 'CONVERSION'
 	| 'COPY'
 	| 'COVERING'
@@ -838,6 +839,7 @@ unreserved_keyword ::=
 	| 'NORMAL'
 	| 'NO_INDEX_JOIN'
 	| 'NOCREATEROLE'
+	| 'NOCONTROLCHANGEFEED'
 	| 'NOLOGIN'
 	| 'NOWAIT'
 	| 'NULLS'
@@ -1931,6 +1933,8 @@ role_option ::=
 	| 'NOCREATEROLE'
 	| 'LOGIN'
 	| 'NOLOGIN'
+	| 'CONTROLCHANGEFEED'
+	| 'NOCONTROLCHANGEFEED'
 	| password_clause
 	| valid_until_clause
 

pkg/ccl/changefeedccl/changefeed_stmt.go

@@ -32,6 +32,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/flowinfra"
+	"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/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
@@ -103,9 +105,15 @@ func changefeedPlanHook(
 		ctx, span := tracing.ChildSpan(ctx, stmt.StatementTag())
 		defer tracing.FinishSpan(span)
 
-		if err := p.RequireAdminRole(ctx, "CREATE CHANGEFEED"); err != nil {
+		hasAdmin, err := p.HasAdminRole(ctx)
+		if err != nil {
 			return err
 		}
+		if !hasAdmin {
+			if err := p.HasRoleOption(ctx, roleoption.CONTROLCHANGEFEED); err != nil {
+				return err
+			}
+		}
 
 		sinkURI, err := sinkURIFn()
 		if err != nil {
@@ -163,6 +171,11 @@ func changefeedPlanHook(
 		if err != nil {
 			return err
 		}
+		for _, desc := range targetDescs {
+			if err := p.CheckPrivilege(ctx, desc, privilege.SELECT); err != nil {
+				return err
+			}
+		}
 		targets := make(jobspb.ChangefeedTargets, len(targetDescs))
 		for _, desc := range targetDescs {
 			if table, isTable := desc.(sqlbase.TableDescriptor); isTable {

pkg/ccl/changefeedccl/changefeed_test.go

@@ -1980,8 +1980,9 @@ func TestChangefeedPermissions(t *testing.T) {
 
 	testFn := func(t *testing.T, db *gosql.DB, f cdctest.TestFeedFactory) {
 		sqlDB := sqlutils.MakeSQLRunner(db)
-		sqlDB.Exec(t, `CREATE TABLE foo (a INT PRIMARY KEY, b STRING)`)
 		sqlDB.Exec(t, `CREATE USER testuser`)
+		sqlDB.Exec(t, `GRANT SELECT ON DATABASE d TO testuser`)
+		sqlDB.Exec(t, `CREATE TABLE d.foo ()`)
 
 		s := f.Server()
 		pgURL, cleanupFunc := sqlutils.PGUrl(
@@ -1994,12 +1995,28 @@ func TestChangefeedPermissions(t *testing.T) {
 		}
 		defer testuser.Close()
 
-		stmt := `EXPERIMENTAL CHANGEFEED FOR foo`
+		stmt := `EXPERIMENTAL CHANGEFEED FOR d.foo`
 		if strings.Contains(t.Name(), `enterprise`) {
-			stmt = `CREATE CHANGEFEED FOR foo`
+			stmt = `CREATE CHANGEFEED FOR d.foo`
+		}
+		privErr := `user testuser does not have CONTROLCHANGEFEED privilege`
+		if _, err := testuser.Exec(stmt); !testutils.IsError(err, privErr) {
+			t.Errorf(`expected '%s' error got: %+v`, privErr, err)
 		}
-		if _, err := testuser.Exec(stmt); !testutils.IsError(err, `only users with the admin role`) {
-			t.Errorf(`expected 'only users with the admin role' error got: %+v`, err)
+
+		// Grant CONTROLCHANGEFEED.
+		sqlDB.Exec(t, `ALTER USER testuser CONTROLCHANGEFEED`)
+		// Now that we pass the CONTROLCHANGEFEED permission check but error on
+		// missing SELECT privileges.
+		selectErr := `user testuser does not have SELECT privilege on relation foo`
+		if _, err := testuser.Exec(stmt); !testutils.IsError(err, selectErr) {
+			t.Errorf(`expected '%s' error got: %+v`, selectErr, err)
+		}
+
+		// Revoke CONTROLCHANGEFEED.
+		sqlDB.Exec(t, `ALTER USER testuser NOCONTROLCHANGEFEED`)
+		if _, err := testuser.Exec(stmt); !testutils.IsError(err, privErr) {
+			t.Errorf(`expected '%s' error got: %+v`, privErr, err)
 		}
 	}
 

pkg/sql/authorization.go

@@ -64,6 +64,13 @@ type AuthorizationAccessor interface {
 	// MemberOfWithAdminOption looks up all the roles (direct and indirect) that 'member' is a member
 	// of and returns a map of role -> isAdmin.
 	MemberOfWithAdminOption(ctx context.Context, member string) (map[string]bool, error)
+
+	// HasRoleOption converts the roleoption to its SQL column name and checks if
+	// the user belongs to a role where the roleprivilege has value true. Only
+	// works on checking the "positive version" of the privilege. Requires a valid
+	// transaction to be open.
+	// Example: CREATEROLE instead of NOCREATEROLE.
+	HasRoleOption(ctx context.Context, roleOption roleoption.Option) error
 }
 
 var _ AuthorizationAccessor = &planner{}
@@ -375,7 +382,6 @@ func (p *planner) resolveMemberOfWithAdminOption(
 // checks if the user belongs to a role where the roleprivilege has value true.
 // Only works on checking the "positive version" of the privilege.
 // Requires a valid transaction to be open.
-// Example: CREATEROLE instead of NOCREATEROLE.
 func (p *planner) HasRoleOption(ctx context.Context, roleOption roleoption.Option) error {
 	// Verify that the txn is valid in any case, so that
 	// we don't get the risk to say "OK" to root requests

pkg/sql/parser/sql.y

@@ -576,7 +576,7 @@ func (u *sqlSymUnion) executorType() tree.ScheduledJobExecutorType {
 %token <str> CHARACTER CHARACTERISTICS CHECK CLOSE
 %token <str> CLUSTER COALESCE COLLATE COLLATION COLUMN COLUMNS COMMENT COMMENTS COMMIT
 %token <str> COMMITTED COMPACT COMPLETE CONCAT CONCURRENTLY CONFIGURATION CONFIGURATIONS CONFIGURE
-%token <str> CONFLICT CONSTRAINT CONSTRAINTS CONTAINS CONVERSION COPY COVERING CREATE CREATEROLE
+%token <str> CONFLICT CONSTRAINT CONSTRAINTS CONTAINS CONTROLCHANGEFEED CONVERSION COPY COVERING CREATE CREATEROLE
 %token <str> CROSS CUBE CURRENT CURRENT_CATALOG CURRENT_DATE CURRENT_SCHEMA
 %token <str> CURRENT_ROLE CURRENT_TIME CURRENT_TIMESTAMP
 %token <str> CURRENT_USER CYCLE
@@ -618,7 +618,7 @@ func (u *sqlSymUnion) executorType() tree.ScheduledJobExecutorType {
 %token <str> MATCH MATERIALIZED MERGE MINVALUE MAXVALUE MINUTE MONTH
 %token <str> MULTILINESTRING MULTIPOINT MULTIPOLYGON
 
-%token <str> NAN NAME NAMES NATURAL NEVER NEXT NO NOCREATEROLE NOLOGIN NO_INDEX_JOIN
+%token <str> NAN NAME NAMES NATURAL NEVER NEXT NO NOCONTROLCHANGEFEED NOCREATEROLE NOLOGIN NO_INDEX_JOIN
 %token <str> NONE NORMAL NOT NOTHING NOTNULL NOWAIT NULL NULLIF NULLS NUMERIC
 
 %token <str> OF OFF OFFSET OID OIDS OIDVECTOR ON ONLY OPT OPTION OPTIONS OR
@@ -6243,6 +6243,14 @@ role_option:
 	{
 		$$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
 	}
+| CONTROLCHANGEFEED
+  {
+    $$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
+  }
+| NOCONTROLCHANGEFEED
+	{
+		$$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
+	}
 | password_clause
 | valid_until_clause
 
@@ -11038,6 +11046,7 @@ unreserved_keyword:
 | CONFIGURATIONS
 | CONFIGURE
 | CONSTRAINTS
+| CONTROLCHANGEFEED
 | CONVERSION
 | COPY
 | COVERING
@@ -11147,6 +11156,7 @@ unreserved_keyword:
 | NORMAL
 | NO_INDEX_JOIN
 | NOCREATEROLE
+| NOCONTROLCHANGEFEED
 | NOLOGIN
 | NOWAIT
 | NULLS

pkg/sql/roleoption/role_option.go

@@ -41,16 +41,20 @@ const (
 	LOGIN
 	NOLOGIN
 	VALIDUNTIL
+	CONTROLCHANGEFEED
+	NOCONTROLCHANGEFEED
 )
 
 // toSQLStmts is a map of Kind -> SQL statement string for applying the
 // option to the role.
 var toSQLStmts = map[Option]string{
-	CREATEROLE:   `UPSERT INTO system.role_options (username, option) VALUES ($1, 'CREATEROLE')`,
-	NOCREATEROLE: `DELETE FROM system.role_options WHERE username = $1 AND option = 'CREATEROLE'`,
-	LOGIN:        `DELETE FROM system.role_options WHERE username = $1 AND option = 'NOLOGIN'`,
-	NOLOGIN:      `UPSERT INTO system.role_options (username, option) VALUES ($1, 'NOLOGIN')`,
-	VALIDUNTIL:   `UPSERT INTO system.role_options (username, option, value) VALUES ($1, 'VALID UNTIL', $2::timestamptz::string)`,
+	CREATEROLE:          `UPSERT INTO system.role_options (username, option) VALUES ($1, 'CREATEROLE')`,
+	NOCREATEROLE:        `DELETE FROM system.role_options WHERE username = $1 AND option = 'CREATEROLE'`,
+	LOGIN:               `DELETE FROM system.role_options WHERE username = $1 AND option = 'NOLOGIN'`,
+	NOLOGIN:             `UPSERT INTO system.role_options (username, option) VALUES ($1, 'NOLOGIN')`,
+	VALIDUNTIL:          `UPSERT INTO system.role_options (username, option, value) VALUES ($1, 'VALID UNTIL', $2::timestamptz::string)`,
+	CONTROLCHANGEFEED:   `UPSERT INTO system.role_options (username, option) VALUES ($1, 'CONTROLCHANGEFEED')`,
+	NOCONTROLCHANGEFEED: `DELETE FROM system.role_options WHERE username = $1 AND option = 'CONTROLCHANGEFEED'`,
 }
 
 // Mask returns the bitmask for a given role option.
@@ -60,12 +64,14 @@ func (o Option) Mask() uint32 {
 
 // ByName is a map of string -> kind value.
 var ByName = map[string]Option{
-	"CREATEROLE":   CREATEROLE,
-	"NOCREATEROLE": NOCREATEROLE,
-	"PASSWORD":     PASSWORD,
-	"LOGIN":        LOGIN,
-	"NOLOGIN":      NOLOGIN,
-	"VALID_UNTIL":  VALIDUNTIL,
+	"CREATEROLE":          CREATEROLE,
+	"NOCREATEROLE":        NOCREATEROLE,
+	"PASSWORD":            PASSWORD,
+	"LOGIN":               LOGIN,
+	"NOLOGIN":             NOLOGIN,
+	"VALID_UNTIL":         VALIDUNTIL,
+	"CONTROLCHANGEFEED":   CONTROLCHANGEFEED,
+	"NOCONTROLCHANGEFEED": NOCONTROLCHANGEFEED,
 }
 
 // ToOption takes a string and returns the corresponding Option.
@@ -151,10 +157,12 @@ func (rol List) CheckRoleOptionConflicts() error {
 		return err
 	}
 
-	if (roleOptionBits&CREATEROLE.Mask() != 0 &&
-		roleOptionBits&NOCREATEROLE.Mask() != 0) ||
-		(roleOptionBits&LOGIN.Mask() != 0 &&
-			roleOptionBits&NOLOGIN.Mask() != 0) {
+	if roleOptionBits&CREATEROLE.Mask() != 0 &&
+		roleOptionBits&NOCREATEROLE.Mask() != 0 ||
+		roleOptionBits&LOGIN.Mask() != 0 &&
+			roleOptionBits&NOLOGIN.Mask() != 0 ||
+		roleOptionBits&CONTROLCHANGEFEED.Mask() != 0 &&
+			roleOptionBits&NOCONTROLCHANGEFEED.Mask() != 0 {
 		return pgerror.Newf(pgcode.Syntax, "conflicting role options")
 	}
 	return nil