sql: add CONTROLJOB system privilege

Release note (sql change): There is now a `CONTROLJOB` system privilege,
which is analogous to the existing `CONTROLJOB` role option, but can
also be inherited by role membership.

pkg/jobs/jobsauth/BUILD.bazel

@@ -32,6 +32,7 @@ go_test(
         "//pkg/sql/pgwire/pgerror",
         "//pkg/sql/privilege",
         "//pkg/sql/roleoption",
+        "//pkg/sql/syntheticprivilege",
         "//pkg/util/randutil",
         "@com_github_stretchr_testify//assert",
     ],

pkg/jobs/jobsauth/authorization.go

@@ -70,6 +70,9 @@ type AuthorizationAccessor interface {
 	// HasAdminRole mirrors sql.AuthorizationAccessor.
 	HasAdminRole(ctx context.Context) (bool, error)
 
+	// HasGlobalPrivilegeOrRoleOption mirrors sql.AuthorizationAccessor.
+	HasGlobalPrivilegeOrRoleOption(ctx context.Context, privilege privilege.Kind) (bool, error)
+
 	// User mirrors sql.PlanHookState.
 	User() username.SQLUsername
 }
@@ -101,7 +104,7 @@ func Authorize(
 		return nil
 	}
 
-	hasControlJob, err := a.HasRoleOption(ctx, roleoption.CONTROLJOB)
+	hasControlJob, err := a.HasGlobalPrivilegeOrRoleOption(ctx, privilege.CONTROLJOB)
 	if err != nil {
 		return err
 	}

pkg/jobs/jobsauth/authorization_test.go

@@ -26,6 +26,7 @@ import (
 	"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/syntheticprivilege"
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
 	"github.com/stretchr/testify/assert"
 )
@@ -37,6 +38,8 @@ type userPrivilege struct {
 
 var viewJobGlobalPrivilege = userPrivilege{privilege.VIEWJOB, privilege.Global}
 
+var controlJobGlobalPrivilege = userPrivilege{privilege.CONTROLJOB, privilege.Global}
+
 type testAuthAccessor struct {
 	user username.SQLUsername
 
@@ -112,6 +115,22 @@ func (a *testAuthAccessor) HasAdminRole(ctx context.Context) (bool, error) {
 	return ok, nil
 }
 
+func (a *testAuthAccessor) HasGlobalPrivilegeOrRoleOption(
+	ctx context.Context, privilege privilege.Kind,
+) (bool, error) {
+	ok, err := a.HasPrivilege(ctx, syntheticprivilege.GlobalPrivilegeObject, privilege, a.User())
+	if err != nil {
+		return false, err
+	}
+	if ok {
+		return true, nil
+	}
+	if roleOption, ok := roleoption.ByName[privilege.String()]; ok {
+		return a.HasRoleOption(ctx, roleOption)
+	}
+	return false, nil
+}
+
 func (a *testAuthAccessor) User() username.SQLUsername {
 	return a.user
 }
@@ -157,7 +176,7 @@ func TestAuthorization(t *testing.T) {
 		userErr     error
 	}{
 		{
-			name: "controljob-sufficient-for-non-admin-jobs",
+			name: "controljob-role-option-sufficient-for-non-admin-jobs",
 
 			user: username.MakeSQLUsernameFromPreNormalizedString("user1"),
 			roleOptions: map[roleoption.Option]struct{}{
@@ -168,7 +187,7 @@ func TestAuthorization(t *testing.T) {
 			accessLevel: jobsauth.ControlAccess,
 		},
 		{
-			name: "controljob-sufficient-to-view-admin-jobs",
+			name: "controljob-role-option-sufficient-to-view-admin-jobs",
 			user: username.MakeSQLUsernameFromPreNormalizedString("user1"),
 			roleOptions: map[roleoption.Option]struct{}{
 				roleoption.CONTROLJOB: {},
@@ -262,6 +281,27 @@ func TestAuthorization(t *testing.T) {
 			accessLevel: jobsauth.ControlAccess,
 			userErr:     pgerror.New(pgcode.InsufficientPrivilege, "foo"),
 		},
+		{
+			name: "controljob-system-privilege-sufficient-for-non-admin-jobs",
+
+			user: username.MakeSQLUsernameFromPreNormalizedString("user1"),
+			userPrivileges: map[userPrivilege]struct{}{
+				controlJobGlobalPrivilege: {},
+			},
+			admins:      map[string]struct{}{},
+			payload:     makeBackupPayload("user2"),
+			accessLevel: jobsauth.ControlAccess,
+		},
+		{
+			name: "controljob-system-privilege-sufficient-to-view-admin-jobs",
+			user: username.MakeSQLUsernameFromPreNormalizedString("user1"),
+			userPrivileges: map[userPrivilege]struct{}{
+				controlJobGlobalPrivilege: {},
+			},
+			admins:      map[string]struct{}{"user2": {}},
+			payload:     makeBackupPayload("user2"),
+			accessLevel: jobsauth.ViewAccess,
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			testAuth := &testAuthAccessor{

pkg/sql/logictest/testdata/logic_test/jobs

@@ -280,3 +280,77 @@ query I
 SELECT count(*) FROM [SHOW AUTOMATIC JOBS] WHERE job_type = 'AUTO CONFIG RUNNER' AND status = 'running'
 ----
 1
+
+subtest control_job_priv
+
+user testuser
+
+statement ok
+CREATE TABLE t_control_job_priv(x INT);
+DROP TABLE t_control_job_priv
+
+let $job_id
+SELECT job_id FROM [SHOW JOBS] WHERE user_name = 'testuser' AND job_type = 'SCHEMA CHANGE GC' AND description LIKE 'GC for DROP TABLE test.public.t_control_job_priv'
+
+statement error user testuser does not have privileges for job
+PAUSE JOB (SELECT $job_id)
+
+user root
+
+statement ok
+GRANT SYSTEM CONTROLJOB TO testuser
+
+user testuser
+
+statement ok
+PAUSE JOB (SELECT $job_id)
+
+user root
+
+statement ok
+REVOKE SYSTEM CONTROLJOB FROM testuser
+
+subtest end
+
+subtest control_job_priv_inherited
+
+user testuser
+
+statement ok
+CREATE TABLE t_control_job_priv_inherited(x INT);
+DROP TABLE t_control_job_priv_inherited
+
+let $job_id
+SELECT job_id FROM [SHOW JOBS] WHERE user_name = 'testuser' AND job_type = 'SCHEMA CHANGE GC' AND description LIKE 'GC for DROP TABLE test.public.t_control_job_priv_inherited'
+
+statement error user testuser does not have privileges for job
+PAUSE JOB (SELECT $job_id)
+
+user root
+
+statement ok
+CREATE ROLE jobcontroller
+
+statement ok
+GRANT SYSTEM CONTROLJOB TO jobcontroller
+
+statement ok
+GRANT jobcontroller TO testuser
+
+user testuser
+
+statement ok
+PAUSE JOB (SELECT $job_id)
+
+user root
+
+statement ok
+REVOKE SYSTEM CONTROLJOB FROM jobcontroller
+
+statement ok
+REVOKE jobcontroller FROM testuser
+
+statement ok
+DROP ROLE jobcontroller
+
+subtest end

pkg/sql/privilege/privilege.go

@@ -75,7 +75,8 @@ const (
 	CREATEROLE               Kind = 32
 	CREATELOGIN              Kind = 33
 	CREATEDB                 Kind = 34
-	largestKind                   = CREATEDB
+	CONTROLJOB               Kind = 35
+	largestKind                   = CONTROLJOB
 )
 
 var isDeprecatedKind = map[Kind]bool{
@@ -163,7 +164,7 @@ var (
 	GlobalPrivileges   = List{
 		ALL, BACKUP, RESTORE, MODIFYCLUSTERSETTING, EXTERNALCONNECTION, VIEWACTIVITY, VIEWACTIVITYREDACTED,
 		VIEWCLUSTERSETTING, CANCELQUERY, NOSQLLOGIN, VIEWCLUSTERMETADATA, VIEWDEBUG, EXTERNALIOIMPLICITACCESS, VIEWJOB,
-		MODIFYSQLCLUSTERSETTING, REPLICATION, MANAGETENANT, VIEWSYSTEMTABLE, CREATEROLE, CREATELOGIN, CREATEDB,
+		MODIFYSQLCLUSTERSETTING, REPLICATION, MANAGETENANT, VIEWSYSTEMTABLE, CREATEROLE, CREATELOGIN, CREATEDB, CONTROLJOB,
 	}
 	VirtualTablePrivileges       = List{ALL, SELECT}
 	ExternalConnectionPrivileges = List{ALL, USAGE, DROP}