Merge #142840

142840: crosscluster/logical: add table level auth for REPLICATIONDEST r=dt a=msbutler

Epic: [CRDB-47102](https://cockroachlabs.atlassian.net/browse/CRDB-47102)

Release note (sql change): this patch allows a user to begin LDR on an existing
table if the user has a table level REPLICATIONDEST priv. Furthermore, this
patch allows a user to begin LDR onto an automatically created table if the
user has the parent database level CREATE privilege. Finally, during
bidirectional replication, this patch allows the user in the OG source URI,
which will begin the reverse stream, to authorize via this table level
REPLICATIONDEST priv.

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

Author: craig[bot]

pkg/crosscluster/logical/create_logical_replication_stmt.go

@@ -102,14 +102,6 @@ func createLogicalReplicationStreamPlanHook(
 			return err
 		}
 
-		// TODO(dt): the global priv is a big hammer; should we be checking just on
-		// table(s) or database being replicated from and into?
-		if err := p.CheckPrivilege(
-			ctx, syntheticprivilege.GlobalPrivilegeObject, privilege.REPLICATIONDEST,
-		); err != nil {
-			return err
-		}
-
 		if stmt.From.Database != "" {
 			return errors.UnimplementedErrorf(errors.IssueLink{}, "logical replication streams on databases are unsupported")
 		}
@@ -158,6 +150,10 @@ func createLogicalReplicationStreamPlanHook(
 			return err
 		}
 
+		if err := checkReplicationPrivileges(ctx, p, stmt, resolvedDestObjects, options.BidirectionalURI()); err != nil {
+			return errors.Wrapf(err, "failed privilege check: table or system level REPLICATIONDEST privilege required")
+		}
+
 		if !p.ExtendedEvalContext().TxnIsSingleStmt {
 			return errors.New("cannot CREATE LOGICAL REPLICATION STREAM in a multi-statement transaction")
 		}
@@ -361,6 +357,14 @@ func (r *ResolvedDestObjects) TargetDescription() string {
 	return targetDescription
 }
 
+func (r *ResolvedDestObjects) TargetTableNames() []string {
+	var targetTableNames []string
+	for i := range r.TableNames {
+		targetTableNames = append(targetTableNames, r.TableNames[i].Table())
+	}
+	return targetTableNames
+}
+
 func resolveDestinationObjects(
 	ctx context.Context,
 	r resolver.SchemaResolver,
@@ -728,3 +732,36 @@ func (r *resolvedLogicalReplicationOptions) BidirectionalURI() string {
 	}
 	return r.bidirectionalURI
 }
+
+func checkReplicationPrivileges(
+	ctx context.Context,
+	p sql.PlanHookState,
+	stmt *tree.CreateLogicalReplicationStream,
+	resolvedDestObjects ResolvedDestObjects,
+	bidirectionalStream string,
+) error {
+	if err := p.CheckPrivilege(
+		ctx, syntheticprivilege.GlobalPrivilegeObject, privilege.REPLICATIONDEST,
+	); err != nil {
+		if !stmt.CreateTable {
+			return replicationutils.AuthorizeTableLevelPriv(ctx, p, p.ExtendedEvalContext().SessionAccessor, privilege.REPLICATIONDEST, resolvedDestObjects.TargetTableNames())
+		} else {
+			dbDesc, err := p.InternalSQLTxn().Descriptors().ByIDWithLeased(p.InternalSQLTxn().KV()).WithoutNonPublic().Get().Database(ctx, resolvedDestObjects.ParentDatabaseID)
+			if err != nil {
+				return err
+			}
+			if err := p.CheckPrivilege(ctx, dbDesc, privilege.CREATE); err != nil {
+				return err
+			}
+			if bidirectionalStream != "" {
+				// TODO(msbutler): how to validate that the user in the reverse stream
+				// URI has REPLICATIONSOURCE priv on a table that has yet to be created?
+				// We could assert it is the same user as the current user, then we
+				// could grant the user the REPLICATIONSOURCE priv on table creation?
+				// Or, we could make REPLICATIONSOURCE a db level priv, required for
+				// BIDI??
+			}
+		}
+	}
+	return nil
+}

pkg/crosscluster/logical/logical_replication_job_test.go

@@ -1954,6 +1954,7 @@ func TestShowLogicalReplicationJobs(t *testing.T) {
 func TestUserPrivileges(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	skip.UnderDeadlock(t)
+	skip.UnderRaceWithIssue(t, 142992)
 	defer log.Scope(t).Close(t)
 
 	ctx := context.Background()
@@ -1978,6 +1979,8 @@ func TestUserPrivileges(t *testing.T) {
 	dbA.Exec(t, fmt.Sprintf("GRANT SYSTEM REPLICATIONDEST TO %s", username.TestUser+"2"))
 	testuser := sqlutils.MakeSQLRunner(s.SQLConn(t, serverutils.User(username.TestUser), serverutils.DBName("a")))
 	testuser2 := sqlutils.MakeSQLRunner(s.SQLConn(t, serverutils.User(username.TestUser+"2"), serverutils.DBName("a")))
+	dbB.Exec(t, "CREATE USER testuser3")
+	dbBURL2 := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("b"), serverutils.User(username.TestUser+"3"))
 
 	var jobAID jobspb.JobID
 	createStmt := "CREATE LOGICAL REPLICATION STREAM FROM TABLE tab ON $1 INTO TABLE tab"
@@ -2022,21 +2025,58 @@ func TestUserPrivileges(t *testing.T) {
 	})
 
 	t.Run("replication-dest", func(t *testing.T) {
-		testuser.ExpectErr(t, "user testuser does not have REPLICATIONDEST system privilege", createStmt, dbBURL.String())
+		testuser.ExpectErr(t, "failed privilege check: table or system level REPLICATIONDEST privilege required: user testuser does not have REPLICATIONDEST privilege on relation tab", createStmt, dbBURL.String())
 		dbA.Exec(t, fmt.Sprintf("GRANT SYSTEM REPLICATIONDEST TO %s", username.TestUser))
 		testuser.QueryRow(t, createStmt, dbBURL.String()).Scan(&jobAID)
+		dbA.Exec(t, fmt.Sprintf("REVOKE SYSTEM REPLICATIONDEST FROM %s", username.TestUser))
 	})
 	t.Run("replication-src", func(t *testing.T) {
-		dbB.Exec(t, "CREATE USER testuser3")
-		dbBURL2 := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("b"), serverutils.User(username.TestUser+"3"))
-		testuser.ExpectErr(t, "user testuser3 does not have REPLICATIONSOURCE system privilege", createStmt, dbBURL2.String())
+		dbA.ExpectErr(t, "user testuser3 does not have REPLICATIONSOURCE system privilege", createStmt, dbBURL2.String())
 		sourcePriv := "REPLICATIONSOURCE"
 		if rng.Intn(3) == 0 {
 			// Test deprecated privilege name.
 			sourcePriv = "REPLICATION"
 		}
+
 		dbB.Exec(t, fmt.Sprintf("GRANT SYSTEM %s TO %s", sourcePriv, username.TestUser+"3"))
-		testuser.QueryRow(t, createStmt, dbBURL2.String()).Scan(&jobAID)
+		dbA.QueryRow(t, createStmt, dbBURL2.String()).Scan(&jobAID)
+		dbB.Exec(t, fmt.Sprintf("REVOKE SYSTEM %s FROM %s", sourcePriv, username.TestUser+"3"))
+	})
+	t.Run("table-level-replication-dest", func(t *testing.T) {
+
+		dbA.Exec(t, `CREATE TABLE tab2 (x INT PRIMARY KEY)`)
+		dbB.Exec(t, `CREATE TABLE tab2 (x INT PRIMARY KEY)`)
+
+		multiTableStmt := `CREATE LOGICAL REPLICATION STREAM FROM TABLES (tab, tab2) ON $1 INTO TABLES (tab, tab2)`
+
+		testuser.ExpectErr(t, "failed privilege check: table or system level REPLICATIONDEST privilege required: user testuser does not have REPLICATIONDEST privilege on relation tab", multiTableStmt, dbBURL.String())
+
+		dbA.Exec(t, fmt.Sprintf(`GRANT REPLICATIONDEST ON TABLE tab TO %s`, username.TestUser))
+		testuser.ExpectErr(t, "failed privilege check: table or system level REPLICATIONDEST privilege required: user testuser does not have REPLICATIONDEST privilege on relation tab2", multiTableStmt, dbBURL.String())
+
+		dbA.Exec(t, fmt.Sprintf(`GRANT REPLICATIONDEST ON TABLE tab2 TO %s`, username.TestUser))
+		testuser.Exec(t, multiTableStmt, dbBURL.String())
+		dbA.Exec(t, fmt.Sprintf(`REVOKE REPLICATIONDEST ON TABLE tab FROM %s`, username.TestUser))
+		dbA.Exec(t, fmt.Sprintf(`REVOKE REPLICATIONDEST ON TABLE tab2 FROM %s`, username.TestUser))
+	})
+	t.Run("db-create-replication-dest", func(t *testing.T) {
+		createStmt := "CREATE LOGICALLY REPLICATED TABLES (tab_clone, tab2_clone) FROM TABLES (tab, tab2) ON $1 WITH UNIDIRECTIONAL"
+		// First try without CREATE privilege on destination database - should fail
+		testuser.ExpectErr(t, "user testuser does not have CREATE privilege on database a", createStmt, dbBURL.String())
+
+		// Grant CREATE privilege on destination database - should now succeed
+		dbA.Exec(t, `GRANT CREATE ON DATABASE a TO testuser`)
+		testuser.Exec(t, createStmt, dbBURL.String())
+
+		dbAURL := replicationtestutils.GetExternalConnectionURI(t, s, s, serverutils.DBName("a"), serverutils.User(username.TestUser))
+
+		dbB.Exec(t, fmt.Sprintf("GRANT SYSTEM REPLICATION TO %s", username.TestUser+"3"))
+		createStmtBidi := "CREATE LOGICALLY REPLICATED TABLES (tab_clone_2, tab2_clone_2) FROM TABLES (tab, tab2) ON $1 WITH BIDIRECTIONAL ON $2"
+		testuser.ExpectErr(t, " uri requires REPLICATIONDEST privilege for bidirectional replication: user testuser3 does not have REPLICATIONDEST privilege on relation tab", createStmtBidi, dbBURL2.String(), dbAURL.String())
+
+		dbB.Exec(t, fmt.Sprintf("GRANT SYSTEM REPLICATIONDEST TO %s", username.TestUser+"3"))
+		testuser.QueryRow(t, createStmtBidi, dbBURL2.String(), dbAURL.String()).Scan(&jobAID)
+
 	})
 }
 

pkg/crosscluster/producer/replication_manager.go

@@ -84,7 +84,9 @@ func (r *replicationStreamManagerImpl) StartReplicationStreamForTables(
 	if execConfig.Codec.IsSystem() && !kvserver.RangefeedEnabled.Get(&execConfig.Settings.SV) {
 		return streampb.ReplicationProducerSpec{}, errors.Errorf("kv.rangefeed.enabled must be enabled on the source cluster for logical replication")
 	}
-
+	if err := maybeAuthorizeReverseStream(ctx, r, req); err != nil {
+		return streampb.ReplicationProducerSpec{}, errors.Wrap(err, "uri requires REPLICATIONDEST privilege for bidirectional replication")
+	}
 	if err := maybeValidateReverseURI(ctx, req.UnvalidatedReverseStreamURI, r.evalCtx.Planner.ExecutorConfig().(*sql.ExecutorConfig).InternalDB); err != nil {
 		return streampb.ReplicationProducerSpec{}, errors.Wrap(err, "reverse stream uri failed validation")
 	}
@@ -157,6 +159,20 @@ func (r *replicationStreamManagerImpl) StartReplicationStreamForTables(
 	}, nil
 }
 
+func maybeAuthorizeReverseStream(
+	ctx context.Context, r *replicationStreamManagerImpl, req streampb.ReplicationProducerRequest,
+) error {
+	if req.UnvalidatedReverseStreamURI == "" {
+		return nil
+	}
+	if err := r.evalCtx.SessionAccessor.CheckPrivilege(ctx,
+		syntheticprivilege.GlobalPrivilegeObject,
+		privilege.REPLICATIONDEST); err != nil {
+		return replicationutils.AuthorizeTableLevelPriv(ctx, r.resolver, r.evalCtx.SessionAccessor, privilege.REPLICATIONDEST, req.TableNames)
+	}
+	return nil
+}
+
 func maybeValidateReverseURI(ctx context.Context, reverseURI string, db *sql.InternalDB) error {
 	if reverseURI == "" {
 		return nil

pkg/crosscluster/replicationutils/BUILD.bazel

@@ -12,11 +12,17 @@ go_library(
         "//pkg/repstream/streampb",
         "//pkg/roachpb",
         "//pkg/sql",
+        "//pkg/sql/catalog",
         "//pkg/sql/catalog/catpb",
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/catalog/descs",
+        "//pkg/sql/catalog/resolver",
         "//pkg/sql/catalog/tabledesc",
         "//pkg/sql/isql",
+        "//pkg/sql/parser",
+        "//pkg/sql/privilege",
+        "//pkg/sql/sem/eval",
+        "//pkg/sql/sem/tree",
         "//pkg/storage",
         "//pkg/storage/mvccencoding",
         "//pkg/testutils/fingerprintutils",

pkg/crosscluster/replicationutils/utils.go

@@ -18,11 +18,17 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/repstream/streampb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descs"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/resolver"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/tabledesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/isql"
+	"github.com/cockroachdb/cockroach/pkg/sql/parser"
+	"github.com/cockroachdb/cockroach/pkg/sql/privilege"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/eval"
+	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/mvccencoding"
 	"github.com/cockroachdb/cockroach/pkg/testutils/fingerprintutils"
@@ -327,3 +333,35 @@ func UnlockLDRTables(
 		return nil
 	})
 }
+
+func AuthorizeTableLevelPriv(
+	ctx context.Context,
+	r resolver.SchemaResolver,
+	sessionAccessor eval.SessionAccessor,
+	priv privilege.Kind,
+	tableNames []string,
+) error {
+	for _, name := range tableNames {
+		uon, err := parser.ParseTableName(name)
+		if err != nil {
+			return err
+		}
+		lookupFlags := tree.ObjectLookupFlags{
+			Required:             true,
+			DesiredObjectKind:    tree.TableObject,
+			DesiredTableDescKind: tree.ResolveRequireTableDesc,
+		}
+		d, _, err := resolver.ResolveExistingObject(ctx, r, uon, lookupFlags)
+		if err != nil {
+			return err
+		}
+		td, ok := d.(catalog.TableDescriptor)
+		if !ok {
+			return errors.New("expected table descriptor")
+		}
+		if err := sessionAccessor.CheckPrivilege(ctx, td, priv); err != nil {
+			return err
+		}
+	}
+	return nil
+}