changefeedccl: add new fine-grained permissions

This change updates permissions semantics related to creating and managing changefeeds.

The `CONTROLCHANGEFEED` role option will be deprecated in the future (see cockroachdb#94757). With this change, usages of `CONTROLCHANGEFEED` will come with a deprecation warning. Its existing behavior (see rules for creating changefeeds below) remains the same.

The `SELECT` and `CHANGEFEED` privileges will be used for changefeeds henceforth:

The `SELECT` privilege on a set of tables allows a user to run core changefeeds against them.

The `CHANGEFEED` privilege on a set of tables allows a user to run enterprise changefeeds on them, and also manage the underlying changefeed job (ie. view, pause, cancel, and resume the job). Notably, a new cluster setting `changefeed.permissions.enforce_external_connections` is added and set to `false` by default. Enabling this setting restricts users with `CHANGEFEED` on a set of tables to create enterprise changefeeds into external connections only. To use a given external connection, a user typically needs the `USAGE` privilege on it.

Note `ALTER DEFAULT PRIVILEGES` can be used with both the `CHANGEFEED` and `SELECT` privileges to assign course-grained permissions (ie. assign permissions to all tables in a schema rather than manually assign them for each table).

Before this change, to create a changefeed, these checks were made in order:
(a) If the user has the `CONTROLCHANGEFEED` role, then they require the `SELECT`privilege
     on all targeted tables
(b) Otherwise, the user requires  the `CHANGEFEED` privilege on all targeted tables
With this change, these checks are updated:
(a) If the user has the `CONTROLCHANGEFEED` role, then they require the `SELECT`
     privilege on all targeted tables. Note: creating a changefeed this way will now produce a
     deprecation notice.
(b) If the changefeed is a core changefeed, they require the `SELECT` privilege on all targeted
     tables
(c) Otherwise, the user requires the `CHANGEFEED` privilege on all targeted tables. Note: If
     `changefeed.permissions.enforce_external_connections` (disabled by default) is set to true,
     then the user will only be able to create a changefeed into an external connection which they
     have the `USAGE` privilege on.

Before this change, to manage a changefeed job `J` (defined a viewing, pausing, resuming, and canceling), a user `U` could do so if they met at least one of the following conditions:
(a) `U` is an admin
(b) `U` is not an admin and `J` is owned by `U`  (only for SHOW JOBS)
(c) `U` is not an admin, `J` is not owned by an admin, and `U` has the `CONTROLJOB` role
With this change, the conditions are updated:
(a) `U` is an admin
(b) `U` is not an admin and `J` is owned by `U`  (only for `SHOW JOBS` or `SHOW
     CHANGEFEED JOBS`)
(c) `U` is not an admin, `J` is not owned by an admin, and `U` has the `CONTROLJOB` role
(d) `U` is not an admin, `J` is not owned by an admin, `J` is a changefeed job, and `U` has
     the `CHANGEFEED` privilege on targeted tables

Before this change, permissions related to altering changefeeds with `ALTER CHANGEFEED` were not explicitly defined (we did not have tests to assert its behavior, but there were some permissions checks regardless). Basically, a user needed access to view a job (ie. look up it’s job ID via `SHOW JOBS`) and they needed to be able to create a new job. After all, `ALTER CHANGEFEED` is essentially the same as creating a new job after stopping the old one.

With this change, the same rules apply: the user needs to be able to access the existing job and to be able to create a new changefeed with the new rules introduced in this change respectively.

Fixes: cockroachdb#94756
Fixes: cockroachdb#92261
Fixes: cockroachdb#87884
Fixes: cockroachdb#85082
Informs: cockroachdb#94759
Informs: cockroachdb#94757
Epic: CRDB-21508
Epic: CRDB-19709

Release note (enterprise change):

The `CONTROLCHANGEFEED` role option will be deprecated in the future (see cockroachdb#94757). With this change, usages of `CONTROLCHANGEFEED` will come with a deprecation warning. Its existing behavior (see rules for creating changefeeds above) remains the same.

The `SELECT` and `CHANGEFEED` privileges will be used for changefeeds henceforth:

The `SELECT` privilege on a set of tables allows a user to run core changefeeds against them.

The `CHANGEFEED` privilege on a set of tables allows a user to run enterprise changefeeds on them, and also manage the underlying changefeed job (ie. view, pause, cancel, and resume the job). Notably, a new cluster setting `changefeed.permissions.enforce_external_connections` is added and set to `false` by default. Enabling this setting restricts users with `CHANGEFEED` on a set of tables to create enterprise changefeeds into external connections only. To use a given external connection, a user typically needs the `USAGE` privilege on it.

Note `ALTER DEFAULT PRIVILEGES` can be used with both both the `CHANGEFEED` and `SELECT` privileges to assign coarse-grained permissions (ie. assign permissions to all tables in a schema rather than manually assign them for each table).

pkg/BUILD.bazel

@@ -169,6 +169,7 @@ ALL_TESTS = [
     "//pkg/internal/team:team_test",
     "//pkg/jobs/joberror:joberror_test",
     "//pkg/jobs/jobsprotectedts:jobsprotectedts_test",
+    "//pkg/jobs/privilege:privilege_test",
     "//pkg/jobs:jobs_test",
     "//pkg/keys:keys_test",
     "//pkg/kv/bulk:bulk_test",
@@ -1114,6 +1115,8 @@ GO_TARGETS = [
     "//pkg/jobs/jobsprotectedts:jobsprotectedts",
     "//pkg/jobs/jobsprotectedts:jobsprotectedts_test",
     "//pkg/jobs/jobstest:jobstest",
+    "//pkg/jobs/privilege:privilege",
+    "//pkg/jobs/privilege:privilege_test",
     "//pkg/jobs:jobs",
     "//pkg/jobs:jobs_test",
     "//pkg/keys:keys",
@@ -2513,6 +2516,7 @@ GET_X_DATA_TARGETS = [
     "//pkg/jobs/jobspb:get_x_data",
     "//pkg/jobs/jobsprotectedts:get_x_data",
     "//pkg/jobs/jobstest:get_x_data",
+    "//pkg/jobs/privilege:get_x_data",
     "//pkg/keys:get_x_data",
     "//pkg/keysbase:get_x_data",
     "//pkg/kv:get_x_data",

pkg/ccl/changefeedccl/BUILD.bazel

@@ -62,6 +62,7 @@ go_library(
         "//pkg/jobs",
         "//pkg/jobs/jobspb",
         "//pkg/jobs/jobsprotectedts",
+        "//pkg/jobs/privilege",
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvclient/kvcoord",
@@ -105,6 +106,7 @@ go_library(
         "//pkg/sql/sem/volatility",
         "//pkg/sql/sessiondatapb",
         "//pkg/sql/sqlutil",
+        "//pkg/sql/syntheticprivilege",
         "//pkg/sql/types",
         "//pkg/util",
         "//pkg/util/admission",
@@ -234,6 +236,7 @@ go_test(
         "//pkg/sql",
         "//pkg/sql/catalog",
         "//pkg/sql/catalog/bootstrap",
+        "//pkg/sql/catalog/catpb",
         "//pkg/sql/catalog/colinfo",
         "//pkg/sql/catalog/descbuilder",
         "//pkg/sql/catalog/descpb",

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -17,6 +17,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/ccl/changefeedccl/changefeedvalidators"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	jobsprivilege "github.com/cockroachdb/cockroach/pkg/jobs/privilege"
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
@@ -100,6 +101,12 @@ func alterChangefeedPlanHook(
 			return err
 		}
 
+		jobPayload := job.Payload()
+
+		if err := jobsprivilege.Authorize(ctx, p, jobID, &jobPayload, jobsprivilege.NoFlag); err != nil {
+			return err
+		}
+
 		prevDetails, ok := job.Details().(jobspb.ChangefeedDetails)
 		if !ok {
 			return errors.Errorf(`job %d is not changefeed job`, jobID)
@@ -123,11 +130,12 @@ func alterChangefeedPlanHook(
 			return err
 		}
 
-		newTargets, newProgress, newStatementTime, originalSpecs, err := generateNewTargets(
+		newTargets, newProgress, newStatementTime, originalSpecs, err := generateAndValidateNewTargets(
 			ctx, exprEval, p,
 			alterChangefeedStmt.Cmds,
 			newOptions.AsMap(), // TODO: Remove .AsMap()
 			prevDetails, job.Progress(),
+			newSinkURI,
 		)
 		if err != nil {
 			return err
@@ -320,14 +328,15 @@ func generateNewOpts(
 	return changefeedbase.MakeStatementOptions(newOptions), sinkURI, nil
 }
 
-func generateNewTargets(
+func generateAndValidateNewTargets(
 	ctx context.Context,
 	exprEval exprutil.Evaluator,
 	p sql.PlanHookState,
 	alterCmds tree.AlterChangefeedCmds,
 	opts map[string]string,
 	prevDetails jobspb.ChangefeedDetails,
 	prevProgress jobspb.Progress,
+	sinkURI string,
 ) (
 	tree.ChangefeedTargets,
 	*jobspb.Progress,
@@ -490,6 +499,7 @@ func generateNewTargets(
 			existingTargetSpans := fetchSpansForDescs(p, existingTargetDescs)
 			var newTargetDescs []catalog.Descriptor
 			for _, target := range v.Targets {
+
 				desc, found, err := getTargetDesc(ctx, p, descResolver, target.TableName)
 				if err != nil {
 					return nil, nil, hlc.Timestamp{}, nil, err
@@ -501,6 +511,7 @@ func generateNewTargets(
 						tree.ErrString(&target),
 					)
 				}
+
 				k := targetKey{TableID: desc.GetID(), FamilyName: target.FamilyName}
 				newTargets[k] = target
 				newTableDescs[desc.GetID()] = desc
@@ -546,6 +557,7 @@ func generateNewTargets(
 						tree.ErrString(&target),
 					)
 				}
+				newTableDescs[desc.GetID()] = desc
 				delete(newTargets, k)
 			}
 			telemetry.CountBucketed(telemetryPath+`.dropped_targets`, int64(len(v.Targets)))
@@ -580,6 +592,20 @@ func generateNewTargets(
 		newTargetList = append(newTargetList, target)
 	}
 
+	hasSelectPrivOnAllTables := true
+	hasChangefeedPrivOnAllTables := true
+	for _, desc := range newTableDescs {
+		hasSelect, hasChangefeed, err := checkPrivilegesForDescriptor(ctx, p, desc)
+		if err != nil {
+			return nil, nil, hlc.Timestamp{}, nil, err
+		}
+		hasSelectPrivOnAllTables = hasSelectPrivOnAllTables && hasSelect
+		hasChangefeedPrivOnAllTables = hasChangefeedPrivOnAllTables && hasChangefeed
+	}
+	if err := verifyUserCanCreateChangefeed(ctx, p, sinkURI, hasSelectPrivOnAllTables, hasChangefeedPrivOnAllTables); err != nil {
+		return nil, nil, hlc.Timestamp{}, nil, err
+	}
+
 	if err := validateNewTargets(ctx, p, newTargetList, newJobProgress, newJobStatementTime); err != nil {
 		return nil, nil, hlc.Timestamp{}, nil, err
 	}