Merge #52869

52869: sql: add CONTROLCHANGEFEED role option r=solongordon a=solongordon Fixes #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> NOCONTROLCHANGEFEED`. Co-authored-by: Solon Gordon <[email protected]>
cockroachdb · Aug 23, 2020 · 4a241d8 · 4a241d8
2 parents 7425e85 + 7af2dda
commit 4a241d8
Show file tree

Hide file tree

Showing 8 changed files with 127 additions and 79 deletions.
diff --git a/docs/generated/sql/bnf/stmt_block.bnf b/docs/generated/sql/bnf/stmt_block.bnf
@@ -746,6 +746,7 @@ unreserved_keyword ::=
 	| 'CONFIGURATIONS'
 	| 'CONFIGURE'
 	| 'CONSTRAINTS'
+	| 'CONTROLCHANGEFEED'
 	| 'CONTROLJOB'
 	| 'CONVERSION'
 	| 'CONVERT'
@@ -857,6 +858,7 @@ unreserved_keyword ::=
 	| 'NORMAL'
 	| 'NO_INDEX_JOIN'
 	| 'NOCREATEROLE'
+	| 'NOCONTROLCHANGEFEED'
 	| 'NOCONTROLJOB'
 	| 'NOLOGIN'
 	| 'NOWAIT'
@@ -1967,6 +1969,8 @@ role_option ::=
 	| 'NOLOGIN'
 	| 'CONTROLJOB'
 	| 'NOCONTROLJOB'
+	| 'CONTROLCHANGEFEED'
+	| 'NOCONTROLCHANGEFEED'
 	| password_clause
 	| valid_until_clause
 

diff --git a/pkg/ccl/changefeedccl/changefeed_stmt.go b/pkg/ccl/changefeedccl/changefeed_stmt.go
@@ -34,6 +34,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/flowinfra"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+	"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/types"
 	"github.com/cockroachdb/cockroach/pkg/storage/cloudimpl"
@@ -104,9 +108,19 @@ 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 {
+			ok, err := p.HasRoleOption(ctx, roleoption.CONTROLCHANGEFEED)
+			if err != nil {
+				return err
+			}
+			if !ok {
+				return pgerror.New(pgcode.InsufficientPrivilege, "permission denied to create changefeed")
+			}
+		}
 
 		sinkURI, err := sinkURIFn()
 		if err != nil {
@@ -164,6 +178,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.(catalog.TableDescriptor); isTable {

diff --git a/pkg/ccl/changefeedccl/changefeed_test.go b/pkg/ccl/changefeedccl/changefeed_test.go
@@ -1973,39 +1973,6 @@ func TestChangefeedErrors(t *testing.T) {
 	)
 }
 
-func TestChangefeedPermissions(t *testing.T) {
-	defer leaktest.AfterTest(t)()
-	defer log.Scope(t).Close(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`)
-
-		s := f.Server()
-		pgURL, cleanupFunc := sqlutils.PGUrl(
-			t, s.ServingSQLAddr(), "TestChangefeedPermissions-testuser", url.User("testuser"),
-		)
-		defer cleanupFunc()
-		testuser, err := gosql.Open("postgres", pgURL.String())
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer testuser.Close()
-
-		stmt := `EXPERIMENTAL CHANGEFEED FOR foo`
-		if strings.Contains(t.Name(), `enterprise`) {
-			stmt = `CREATE CHANGEFEED FOR foo`
-		}
-		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)
-		}
-	}
-
-	t.Run(`sinkless`, sinklessTest(testFn))
-	t.Run(`enterprise`, enterpriseTest(testFn))
-}
-
 func TestChangefeedDescription(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	defer log.Scope(t).Close(t)

diff --git a/pkg/ccl/logictestccl/testdata/logic_test/changefeed b/pkg/ccl/logictestccl/testdata/logic_test/changefeed
@@ -0,0 +1,34 @@
+# LogicTest: local
+
+statement ok
+CREATE TABLE t ()
+
+user testuser
+
+statement error permission denied to create changefeed
+CREATE CHANGEFEED FOR t
+
+user root
+
+# Test granting CONTROLCHANGEFEED.
+statement ok
+ALTER USER testuser CONTROLCHANGEFEED;
+GRANT SELECT ON DATABASE test TO testuser
+
+user testuser
+
+# Now we should pass the CONTROLCHANGEFEED permission check but error on missing
+# SELECT privileges.
+statement error user testuser does not have SELECT privilege on relation t
+CREATE CHANGEFEED FOR t
+
+# Test revoking CONTROLCHANGEFEED.
+user root
+
+statement ok
+ALTER USER testuser NOCONTROLCHANGEFEED
+
+user testuser
+
+statement error permission denied to create changefeed
+CREATE CHANGEFEED FOR t
diff --git a/pkg/sql/authorization.go b/pkg/sql/authorization.go
@@ -73,6 +73,15 @@ 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 option has value true. Requires a
+	// valid transaction to be open.
+	//
+	// This check should be done on the version of the privilege that is stored in
+	// the role options table. Example: CREATEROLE instead of NOCREATEROLE.
+	// NOLOGIN instead of LOGIN.
+	HasRoleOption(ctx context.Context, roleOption roleoption.Option) (bool, error)
 }
 
 var _ AuthorizationAccessor = &planner{}
@@ -390,12 +399,7 @@ func (p *planner) resolveMemberOfWithAdminOption(
 	return ret, nil
 }
 
-// HasRoleOption converts the roleoption to it's SQL column name and
-// checks if the user belongs to a role where the roleprivilege has value true.
-// Requires a valid transaction to be open.
-// This check should be done on the version of the privilege that is stored in
-// the role options table.
-// Example: CREATEROLE instead of NOCREATEROLE. NOLOGIN instead of LOGIN.
+// HasRoleOption implements the AuthorizationAccessor interface.
 func (p *planner) HasRoleOption(ctx context.Context, roleOption roleoption.Option) (bool, error) {
 	// Verify that the txn is valid in any case, so that
 	// we don't get the risk to say "OK" to root requests

diff --git a/pkg/sql/parser/sql.y b/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 CONTROLJOB CONVERSION CONVERT COPY COVERING CREATE CREATEROLE
+%token <str> CONFLICT CONSTRAINT CONSTRAINTS CONTAINS CONTROLCHANGEFEED CONTROLJOB CONVERSION CONVERT 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 NOCONTROLJOB NOCREATEROLE NOLOGIN NO_INDEX_JOIN
+%token <str> NAN NAME NAMES NATURAL NEVER NEXT NO NOCONTROLCHANGEFEED NOCONTROLJOB 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
@@ -6322,17 +6322,17 @@ role_option:
     $$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
   }
 | NOCREATEROLE
-	{
-		$$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
-	}
+  {
+    $$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
+  }
 | LOGIN
-	{
-		$$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
-	}
+  {
+    $$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
+  }
 | NOLOGIN
-	{
-		$$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
-	}
+  {
+    $$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
+  }
 | CONTROLJOB
   {
    $$.val = tree.KVOption{Key: tree.Name($1), Value: nil}
@@ -6341,6 +6341,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
 
@@ -6356,24 +6364,24 @@ role_options:
   }
 
 opt_role_options:
-	opt_with role_options
-	{
-		$$.val = $2.kvOptions()
-	}
+  opt_with role_options
+  {
+    $$.val = $2.kvOptions()
+  }
 | /* EMPTY */
-	{
-		$$.val = nil
-	}
+  {
+    $$.val = nil
+  }
 
 valid_until_clause:
   VALID UNTIL string_or_placeholder
   {
-		$$.val = tree.KVOption{Key: tree.Name(fmt.Sprintf("%s_%s",$1, $2)), Value: $3.expr()}
+    $$.val = tree.KVOption{Key: tree.Name(fmt.Sprintf("%s_%s",$1, $2)), Value: $3.expr()}
   }
 | VALID UNTIL NULL
   {
-		$$.val = tree.KVOption{Key: tree.Name(fmt.Sprintf("%s_%s",$1, $2)), Value: tree.DNull}
-	}
+    $$.val = tree.KVOption{Key: tree.Name(fmt.Sprintf("%s_%s",$1, $2)), Value: tree.DNull}
+  }
 
 opt_view_recursive:
   /* EMPTY */ { /* no error */ }
@@ -11189,6 +11197,7 @@ unreserved_keyword:
 | CONFIGURATIONS
 | CONFIGURE
 | CONSTRAINTS
+| CONTROLCHANGEFEED
 | CONTROLJOB
 | CONVERSION
 | CONVERT
@@ -11300,6 +11309,7 @@ unreserved_keyword:
 | NORMAL
 | NO_INDEX_JOIN
 | NOCREATEROLE
+| NOCONTROLCHANGEFEED
 | NOCONTROLJOB
 | NOLOGIN
 | NOWAIT

diff --git a/pkg/sql/roleoption/option_string.go b/pkg/sql/roleoption/option_string.go
diff --git a/pkg/sql/roleoption/role_option.go b/pkg/sql/roleoption/role_option.go
@@ -43,18 +43,22 @@ const (
 	VALIDUNTIL
 	CONTROLJOB
 	NOCONTROLJOB
+	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)`,
-	CONTROLJOB:   `UPSERT INTO system.role_options (username, option) VALUES ($1, 'CONTROLJOB')`,
-	NOCONTROLJOB: `DELETE FROM system.role_options WHERE username = $1 AND option = 'CONTROLJOB'`,
+	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)`,
+	CONTROLJOB:          `UPSERT INTO system.role_options (username, option) VALUES ($1, 'CONTROLJOB')`,
+	NOCONTROLJOB:        `DELETE FROM system.role_options WHERE username = $1 AND option = 'CONTROLJOB'`,
+	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.
@@ -64,14 +68,16 @@ 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,
-	"CONTROLJOB":   CONTROLJOB,
-	"NOCONTROLJOB": NOCONTROLJOB,
+	"CREATEROLE":          CREATEROLE,
+	"NOCREATEROLE":        NOCREATEROLE,
+	"PASSWORD":            PASSWORD,
+	"LOGIN":               LOGIN,
+	"NOLOGIN":             NOLOGIN,
+	"VALID_UNTIL":         VALIDUNTIL,
+	"CONTROLJOB":          CONTROLJOB,
+	"NOCONTROLJOB":        NOCONTROLJOB,
+	"CONTROLCHANGEFEED":   CONTROLCHANGEFEED,
+	"NOCONTROLCHANGEFEED": NOCONTROLCHANGEFEED,
 }
 
 // ToOption takes a string and returns the corresponding Option.
@@ -162,7 +168,9 @@ func (rol List) CheckRoleOptionConflicts() error {
 		(roleOptionBits&LOGIN.Mask() != 0 &&
 			roleOptionBits&NOLOGIN.Mask() != 0) ||
 		(roleOptionBits&CONTROLJOB.Mask() != 0 &&
-			roleOptionBits&NOCONTROLJOB.Mask() != 0) {
+			roleOptionBits&NOCONTROLJOB.Mask() != 0) ||
+		(roleOptionBits&CONTROLCHANGEFEED.Mask() != 0 &&
+			roleOptionBits&NOCONTROLCHANGEFEED.Mask() != 0) {
 		return pgerror.Newf(pgcode.Syntax, "conflicting role options")
 	}
 	return nil