Merge #142772

142772: crosscluster/physical: update ALTER REPLICATION JOB auth r=jeffswenson a=msbutler

Epic CRDB-47102

Release note (ops change): all ALTER VIRTUAL CLUSTER REPLICATION JOB cmds,
except for ALTER VIRTUAL CLUSTER SET REPLICATION SOURCE, will require the
REPLICATIONDEST priv, in addition to MANAGEVIRTUALCLUSTER. ALTER VIRTUAL
CLUSTER SET REPLICATION SOURCE now requires REPLICATIONSOURCE. Note, if the
ingestion job was created before 25.1, the user can still alter the replication
job without REPLICATIONDEST priv-- we'd like to leave this undocumented though.

Co-authored-by: Michael Butler <[email protected]>

Author: craig[bot]

pkg/crosscluster/physical/BUILD.bazel

@@ -27,6 +27,7 @@ go_library(
         "//pkg/cloud",
         "//pkg/cloud/externalconn",
         "//pkg/cloud/externalconn/connectionpb",
+        "//pkg/clusterversion",
         "//pkg/crosscluster",
         "//pkg/crosscluster/producer",
         "//pkg/crosscluster/replicationutils",
@@ -63,6 +64,7 @@ go_library(
         "//pkg/sql/isql",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
+        "//pkg/sql/pgwire/pgnotice",
         "//pkg/sql/physicalplan",
         "//pkg/sql/privilege",
         "//pkg/sql/rowenc",
@@ -132,6 +134,7 @@ go_test(
         "//pkg/ccl/kvccl/kvtenantccl",
         "//pkg/ccl/storageccl",
         "//pkg/cloud/impl:cloudimpl",
+        "//pkg/clusterversion",
         "//pkg/crosscluster",
         "//pkg/crosscluster/producer",
         "//pkg/crosscluster/replicationtestutils",

pkg/crosscluster/physical/alter_replication_job.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/ccl/utilccl"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationutils"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/streamclient"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
@@ -26,9 +27,12 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgnotice"
+	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/asof"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/sql/syntheticprivilege"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
@@ -224,9 +228,22 @@ func alterReplicationJobHook(
 		}
 		jobRegistry := p.ExecCfg().JobRegistry
 		if alterTenantStmt.Producer {
+			if err := p.CheckPrivilege(
+				ctx, syntheticprivilege.GlobalPrivilegeObject, privilege.REPLICATIONSOURCE,
+			); err != nil {
+				return err
+			}
 			return alterTenantSetReplicationSource(ctx, p.InternalSQLTxn(), jobRegistry, options, tenInfo)
 		}
-
+		if err := p.CheckPrivilege(
+			ctx, syntheticprivilege.GlobalPrivilegeObject, privilege.REPLICATIONDEST,
+		); err != nil {
+			if !p.ExecCfg().Settings.Version.IsActive(ctx, clusterversion.V25_2) {
+				p.BufferClientNotice(ctx, pgnotice.Newf("this command will require the REPLICATIONDEST privilege on a fully upgraded 25.2+ cluster"))
+			} else {
+				return err
+			}
+		}
 		// If a source uri is being provided, we're enabling replication into an
 		// existing virtual cluster. It must be inactive, and we'll verify that it
 		// was the cluster from which the one it will replicate was replicated, i.e.

pkg/crosscluster/physical/alter_replication_job_test.go

@@ -12,9 +12,12 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/clusterversion"
 	"github.com/cockroachdb/cockroach/pkg/crosscluster/replicationtestutils"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/security/username"
+	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
 	"github.com/cockroachdb/cockroach/pkg/testutils/jobutils"
@@ -544,10 +547,51 @@ func TestAlterReplicationJobErrors(t *testing.T) {
 	defer srv.Stopper().Stop(ctx)
 
 	db := sqlutils.MakeSQLRunner(sqlDB)
+	db.Exec(t, "CREATE TENANT t1")
+
+	db.Exec(t, fmt.Sprintf("CREATE USER %s", username.TestUser))
+	testuser := sqlutils.MakeSQLRunner(srv.SQLConn(t, serverutils.User(username.TestUser)))
 
 	t.Run("alter tenant subqueries", func(t *testing.T) {
 		// Regression test for #136339
 		db.ExpectErr(t, "subqueries are not allowed", "ALTER TENANT (select 't2') START REPLICATION OF t1 ON 'foo'")
 	})
+	t.Run("alter replication dest privs", func(t *testing.T) {
+		cmd := "ALTER TENANT t1 SET REPLICATION RETENTION ='100ms'"
+		testuser.ExpectErr(t, "user testuser does not have MANAGEVIRTUALCLUSTER system privilege", cmd)
+		db.Exec(t, fmt.Sprintf("GRANT SYSTEM MANAGEVIRTUALCLUSTER TO %s", username.TestUser))
+		testuser.ExpectErr(t, "user testuser does not have REPLICATIONDEST system privilege", cmd)
+		db.Exec(t, fmt.Sprintf("GRANT SYSTEM REPLICATIONDEST TO %s", username.TestUser))
+		// Implies we got past the priv checks.
+		testuser.ExpectErr(t, `does not have an active replication consumer job`, cmd)
+		db.Exec(t, fmt.Sprintf("REVOKE SYSTEM MANAGEVIRTUALCLUSTER FROM %s", username.TestUser))
+	})
+	t.Run("alter replication source privs", func(t *testing.T) {
+		cmd := "ALTER TENANT t1 SET REPLICATION SOURCE EXPIRATION WINDOW ='100ms'"
+		testuser.ExpectErr(t, "user testuser does not have MANAGEVIRTUALCLUSTER system privilege", cmd)
+		db.Exec(t, fmt.Sprintf("GRANT SYSTEM MANAGEVIRTUALCLUSTER TO %s", username.TestUser))
+		testuser.ExpectErr(t, "user testuser does not have REPLICATIONSOURCE system privilege", cmd)
+		db.Exec(t, fmt.Sprintf("GRANT SYSTEM REPLICATIONSOURCE TO %s", username.TestUser))
+		testuser.Exec(t, cmd)
+	})
+	t.Run("alter replication dest priv 24.3", func(t *testing.T) {
+		params := base.TestServerArgs{
+			DefaultTestTenant: base.TestControlsTenantsExplicitly,
+		}
+		params.Knobs.Server = &server.TestingKnobs{
+			ClusterVersionOverride:         clusterversion.V24_3.Version(),
+			DisableAutomaticVersionUpgrade: make(chan struct{}),
+		}
 
+		srv, sqlDB, _ := serverutils.StartServer(t, params)
+		defer srv.Stopper().Stop(ctx)
+		db := sqlutils.MakeSQLRunner(sqlDB)
+		db.Exec(t, "CREATE TENANT t1")
+		db.Exec(t, fmt.Sprintf("CREATE USER %s", username.TestUser))
+		testuser := sqlutils.MakeSQLRunner(srv.SQLConn(t, serverutils.User(username.TestUser)))
+		db.Exec(t, fmt.Sprintf("GRANT SYSTEM MANAGEVIRTUALCLUSTER TO %s", username.TestUser))
+		// Implies we got past the priv checks, without REPLICATIONDEST.
+		cmd := "ALTER TENANT t1 SET REPLICATION RETENTION ='100ms'"
+		testuser.ExpectErr(t, `does not have an active replication consumer job`, cmd)
+	})
 }