sql,backupccl: allow owners to also control schedules

This change relaxes the admin only check that was enforced
when resuming, pasuing, dropping and altering a backup schedule
to also allow non-admin owners of the schedules to perform
these control operations.

Release note (sql change): Owners of a backup schedule can now
control their schedule using the supported pause, resume, drop
and alter queries.

Release justification: high impact change to introduce finer grained
permission control to disaster recovery operations

pkg/ccl/backupccl/alter_backup_schedule.go

@@ -53,7 +53,6 @@ func loadSchedules(
 	if err != nil {
 		return s, err
 	}
-
 	args := &backuppb.ScheduledBackupExecutionArgs{}
 	if err := pbtypes.UnmarshalAny(schedule.ExecutionArgs().Args, args); err != nil {
 		return s, errors.Wrap(err, "un-marshaling args")
@@ -109,8 +108,7 @@ func doAlterBackupSchedules(
 	eval *alterBackupScheduleEval,
 	resultsCh chan<- tree.Datums,
 ) error {
-	s, err := loadSchedules(
-		ctx, p, eval)
+	s, err := loadSchedules(ctx, p, eval)
 	if err != nil {
 		return err
 	}
@@ -126,14 +124,16 @@ func doAlterBackupSchedules(
 			s.incJob.ScheduleID())
 	}
 
-	// Note that even ADMIN is subject to these restrictions. We expect to
-	// add a finer-grained permissions model soon.
-	if s.fullJob.Owner() != p.User() {
-		return pgerror.Newf(pgcode.InsufficientPrivilege, "only the OWNER of a schedule may alter it")
+	// Check that the user is admin or the owner of the schedules being altered.
+	isAdmin, err := p.UserHasAdminRole(ctx, p.User())
+	if err != nil {
+		return err
 	}
-
-	if s.incJob != nil && s.incJob.Owner() != p.User() {
-		return pgerror.Newf(pgcode.InsufficientPrivilege, "only the OWNER of a schedule may alter it")
+	isOwnerOfFullJob := s.fullJob == nil || s.fullJob.Owner() == p.User()
+	isOwnerOfIncJob := s.incJob == nil || s.incJob.Owner() == p.User()
+	if !isAdmin && !(isOwnerOfFullJob && isOwnerOfIncJob) {
+		return pgerror.New(pgcode.InsufficientPrivilege, "must be admin or owner of the "+
+			"schedules being altered.")
 	}
 
 	if s, err = processFullBackupRecurrence(

pkg/ccl/backupccl/testdata/backup-restore/schedule-privileges

@@ -1,6 +1,8 @@
 new-server name=s1
 ----
 
+subtest create-scheduled-privileges
+
 exec-sql
 CREATE DATABASE foo;
 ----
@@ -15,15 +17,15 @@ CREATE EXTERNAL CONNECTION foo AS 'userfile:///foo';
 
 # Admin can create all schedules.
 exec-sql
-CREATE SCHEDULE foocluster FOR BACKUP INTO 'external://foo/cluster' RECURRING '@hourly';
+CREATE SCHEDULE foocluster_admin FOR BACKUP INTO 'external://foo/cluster' RECURRING '@hourly';
 ----
 
 exec-sql
-CREATE SCHEDULE foodb FOR BACKUP DATABASE foo INTO 'external://foo/database' RECURRING '@hourly';
+CREATE SCHEDULE foodb_admin FOR BACKUP DATABASE foo INTO 'external://foo/database' RECURRING '@hourly';
 ----
 
 exec-sql
-CREATE SCHEDULE footable FOR BACKUP TABLE foo.foo INTO 'external://foo/table' RECURRING '@hourly';
+CREATE SCHEDULE footable_admin FOR BACKUP TABLE foo.foo INTO 'external://foo/table' RECURRING '@hourly';
 ----
 
 # Non-root admin can create all schedules.
@@ -33,15 +35,15 @@ GRANT ADMIN TO testuser;
 ----
 
 exec-sql user=testuser
-CREATE SCHEDULE foocluster FOR BACKUP INTO 'external://foo/cluster' RECURRING '@hourly';
+CREATE SCHEDULE foocluster_admintestuser FOR BACKUP INTO 'external://foo/cluster' RECURRING '@hourly';
 ----
 
 exec-sql user=testuser
-CREATE SCHEDULE foodb FOR BACKUP DATABASE foo INTO 'external://foo/database' RECURRING '@hourly';
+CREATE SCHEDULE foodb_admintestuser FOR BACKUP DATABASE foo INTO 'external://foo/database' RECURRING '@hourly';
 ----
 
 exec-sql user=testuser
-CREATE SCHEDULE footable FOR BACKUP TABLE foo.foo INTO 'external://foo/table' RECURRING '@hourly';
+CREATE SCHEDULE footable_admintestuser FOR BACKUP TABLE foo.foo INTO 'external://foo/table' RECURRING '@hourly';
 ----
 
 # Non-root non-admin cannot create any schedules.
@@ -78,13 +80,115 @@ GRANT USAGE ON EXTERNAL CONNECTION foo TO testuser;
 ----
 
 exec-sql user=testuser
-CREATE SCHEDULE foocluster FOR BACKUP INTO 'external://foo/cluster' RECURRING '@hourly';
+CREATE SCHEDULE foocluster_testuser FOR BACKUP INTO 'external://foo/cluster' RECURRING '@hourly';
 ----
 
 exec-sql user=testuser
-CREATE SCHEDULE foodb FOR BACKUP DATABASE foo INTO 'external://foo/database' RECURRING '@hourly';
+CREATE SCHEDULE foodb_testuser FOR BACKUP DATABASE foo INTO 'external://foo/database' RECURRING '@hourly';
 ----
 
 exec-sql user=testuser
-CREATE SCHEDULE footable FOR BACKUP TABLE foo.foo INTO 'external://foo/table' RECURRING '@hourly';
+CREATE SCHEDULE footable_testuser FOR BACKUP TABLE foo.foo INTO 'external://foo/table' RECURRING '@hourly';
+----
+
+subtest end
+
+subtest control-schedules-privileges
+
+# Save the schedule IDs for the schedules created by admin root.
+let $fullID $incID
+with schedules as (show schedules) select id from schedules where label='foocluster_admin' order by command->>'backup_type' asc;
+----
+
+query-sql
+with schedules as (SHOW SCHEDULES FOR BACKUP) SELECT label, command FROM schedules WHERE id IN
+($fullID, $incID) ORDER BY next_run;
+----
+foocluster_admin BACKUP INTO LATEST IN 'external://foo/cluster' WITH detached
+foocluster_admin BACKUP INTO 'external://foo/cluster' WITH detached
+
+# nonadmin testuser is not allowed to drop a schedule they do not own.
+exec-sql expect-error-regex=(must be admin or owner of the schedule [0-9]+ to DROP it) user=testuser
+DROP SCHEDULE $fullID
+----
+regex matches error
+
+exec-sql
+GRANT ADMIN TO testuser
+----
+
+# testuser has been granted admin so the drop should succeed.
+exec-sql user=testuser
+DROP SCHEDULE $fullID
+----
+
+query-sql
+with schedules as (SHOW SCHEDULES FOR BACKUP) SELECT label,command,owner FROM schedules WHERE id IN
+($fullID, $incID) ORDER BY next_run;
+----
+foocluster_admin BACKUP INTO LATEST IN 'external://foo/cluster' WITH detached root
+
+exec-sql
+REVOKE ADMIN FROM testuser
+----
+
+exec-sql expect-error-regex=(must be admin or owner of the schedule [0-9]+ to DROP it) user=testuser
+DROP SCHEDULE $incID
+----
+regex matches error
+
+# Save the schedule IDs for the schedules that were created by non-admin testuser.
+let $testuserFullID $testuserIncID
+with schedules as (show schedules) select id from schedules where label='foocluster_testuser' order by command->>'backup_type' asc;
+----
+
+query-sql
+with schedules as (SHOW SCHEDULES FOR BACKUP) SELECT label, command, owner FROM schedules WHERE id
+IN ($testuserFullID, $testuserIncID) ORDER BY next_run;
+----
+foocluster_testuser BACKUP INTO LATEST IN 'external://foo/cluster' WITH detached testuser
+foocluster_testuser BACKUP INTO 'external://foo/cluster' WITH detached testuser
+
+# testuser owns these schedules so should be able to pause, resume, drop, alter
+# them without admin.
+exec-sql user=testuser
+PAUSE SCHEDULE $testuserFullID;
+PAUSE SCHEDULE $testuserIncID;
+----
+
+exec-sql user=testuser
+RESUME SCHEDULE $testuserFullID;
+RESUME SCHEDULE $testuserIncID;
+----
+
+exec-sql user=testuser
+ALTER BACKUP SCHEDULE $testuserFullID SET WITH revision_history = false;
+----
+
+exec-sql user=testuser
+DROP SCHEDULE $testuserFullID;
+DROP SCHEDULE $testuserIncID;
+----
+
+# But testuser can't drop, alter, resume or pause the root owned schedules.
+exec-sql expect-error-regex=(must be admin or owner of the schedule [0-9]+ to PAUSE it) user=testuser
+PAUSE SCHEDULE $incID
 ----
+regex matches error
+
+exec-sql expect-error-regex=(must be admin or owner of the schedule [0-9]+ to RESUME it) user=testuser
+RESUME SCHEDULE $incID
+----
+regex matches error
+
+exec-sql user=testuser expect-error-regex=(incremental backup schedule [0-9]+ has no corresponding full backup schedule)
+ALTER BACKUP SCHEDULE $incID SET WITH revision_history = false;
+----
+regex matches error
+
+exec-sql expect-error-regex=(must be admin or owner of the schedule [0-9]+ to DROP it) user=testuser
+DROP SCHEDULE $incID;
+----
+regex matches error
+
+subtest end

pkg/jobs/delegate_control_test.go

@@ -12,9 +12,7 @@ package jobs
 
 import (
 	"context"
-	gosql "database/sql"
 	"fmt"
-	"net/url"
 	"strings"
 	"testing"
 	"time"
@@ -25,7 +23,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
-	"github.com/cockroachdb/cockroach/pkg/testutils/sqlutils"
 	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/stretchr/testify/require"
@@ -138,23 +135,6 @@ func TestScheduleControl(t *testing.T) {
 		th.sqlDB.Exec(t, "DROP SCHEDULES "+querySchedules)
 		require.Equal(t, 0, len(th.sqlDB.QueryStr(t, querySchedules)))
 	})
-
-	t.Run("pause-non-privileged-user", func(t *testing.T) {
-		scheduleID := makeSchedule("one-schedule", "@daily")
-
-		th.sqlDB.Exec(t, `CREATE USER testuser`)
-		pgURL, cleanupFunc := sqlutils.PGUrl(
-			t, th.server.ServingSQLAddr(), "NonPrivileged-testuser",
-			url.User("testuser"),
-		)
-		defer cleanupFunc()
-		testuser, err := gosql.Open("postgres", pgURL.String())
-		require.NoError(t, err)
-		defer testuser.Close()
-
-		_, err = testuser.Exec("PAUSE SCHEDULE $1", scheduleID)
-		require.EqualError(t, err, "pq: only users with the admin role are allowed to PAUSE SCHEDULES")
-	})
 }
 
 func TestJobsControlForSchedules(t *testing.T) {

pkg/sql/control_schedules.go

@@ -19,6 +19,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/scheduledjobs"
 	"github.com/cockroachdb/cockroach/pkg/security/username"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/sessiondata"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqltelemetry"
@@ -57,19 +59,22 @@ func JobSchedulerEnv(execCfg *ExecutorConfig) scheduledjobs.JobSchedulerEnv {
 	return scheduledjobs.ProdJobSchedulerEnv
 }
 
-// loadSchedule loads schedule information.
+// loadSchedule loads schedule information as the node user.
 func loadSchedule(params runParams, scheduleID tree.Datum) (*jobs.ScheduledJob, error) {
 	env := JobSchedulerEnv(params.ExecCfg())
 	schedule := jobs.NewScheduledJob(env)
 
 	// Load schedule expression.  This is needed for resume command, but we
 	// also use this query to check for the schedule existence.
+	//
+	// Run the query as the node user since we perform our own privilege checks
+	// before using the returned schedule.
 	datums, cols, err := params.ExecCfg().InternalExecutor.QueryRowExWithCols(
 		params.ctx,
 		"load-schedule",
-		params.p.Txn(), sessiondata.InternalExecutorOverride{User: username.RootUserName()},
+		params.p.Txn(), sessiondata.InternalExecutorOverride{User: username.NodeUserName()},
 		fmt.Sprintf(
-			"SELECT schedule_id, next_run, schedule_expr, executor_type, execution_args FROM %s WHERE schedule_id = $1",
+			"SELECT schedule_id, next_run, schedule_expr, executor_type, execution_args, owner FROM %s WHERE schedule_id = $1",
 			env.ScheduledJobsTableName(),
 		),
 		scheduleID)
@@ -136,6 +141,16 @@ func (n *controlSchedulesNode) startExec(params runParams) error {
 			continue // not an error if schedule does not exist
 		}
 
+		isAdmin, err := params.p.UserHasAdminRole(params.ctx, params.p.User())
+		if err != nil {
+			return err
+		}
+		isOwner := schedule.Owner() == params.p.User()
+		if !isAdmin && !isOwner {
+			return pgerror.Newf(pgcode.InsufficientPrivilege, "must be admin or owner of the "+
+				"schedule %d to %s it", schedule.ScheduleID(), n.command.String())
+		}
+
 		switch n.command {
 		case tree.PauseSchedule:
 			schedule.Pause()

pkg/sql/opt/optbuilder/misc_statements.go

@@ -108,10 +108,6 @@ func (b *Builder) buildCancelSessions(n *tree.CancelSessions, inScope *scope) (o
 func (b *Builder) buildControlSchedules(
 	n *tree.ControlSchedules, inScope *scope,
 ) (outScope *scope) {
-	if err := b.catalog.RequireAdminRole(b.ctx, n.StatementTag()); err != nil {
-		panic(err)
-	}
-
 	// We don't allow the input statement to reference outer columns, so we
 	// pass a "blank" scope rather than inScope.
 	emptyScope := b.allocScope()