some simple restrictions on clone privilege. (#22329)

```
1. 普通租户不能 clone 任一系统 db/table
如 mo_task, mo_catalog, system, mysql, system_mertric, information_schema, mo_debug

这个操作等同于在系统 db 下面创建表，为了和 create table 权限保持一致
2. sys租户和普通租户都不能 clone 数据到系统 db
不能将其他 table clone 到 系统 db 下


3. 普通租户不能 clone cluster table
cluster table 会聚合很多租户的数据，普通租户可以用 select 查看属于自己的数据。
如果允许 clone，那么 clone 就需要过滤 table 的数据，无法优化，与 select 功能无异。(rawlog, statement_info...)

4. 普通租户可以 clone 已经订阅的 db/table


==> 只有 sys 租户才能
1. clone 系统表
sys 租户和普通租户都不能
1. clone 数据到系统 db 下面

```

Approved by: @heni02, @iamlinjunhong, @ouyuanning, @daviszhen

Author: gouhongshen

pkg/frontend/clone.go

@@ -85,7 +85,9 @@ func handleCloneTableAcrossAccounts(
 		fromAccountId uint32
 	)
 
-	reqCtx = context.WithValue(reqCtx, tree.CloneLevelCtxKey{}, tree.CloneLevelTable)
+	if reqCtx.Value(tree.CloneLevelCtxKey{}) == nil {
+		reqCtx = context.WithValue(reqCtx, tree.CloneLevelCtxKey{}, tree.NormalCloneLevelTable)
+	}
 
 	bh = ses.GetBackgroundExec(reqCtx)
 	if err = bh.Exec(reqCtx, "begin"); err != nil {
@@ -191,7 +193,9 @@ func handleCloneDatabase(
 		snapshotTS int64
 	)
 
-	reqCtx = context.WithValue(reqCtx, tree.CloneLevelCtxKey{}, tree.CloneLevelDatabase)
+	if reqCtx.Value(tree.CloneLevelCtxKey{}) == nil {
+		reqCtx = context.WithValue(reqCtx, tree.CloneLevelCtxKey{}, tree.NormalCloneLevelDatabase)
+	}
 
 	bh = ses.GetBackgroundExec(reqCtx)
 	if err = bh.Exec(reqCtx, "begin"); err != nil {

pkg/frontend/pitr.go

@@ -1113,7 +1113,7 @@ func doRestorePitr(ctx context.Context, ses *Session, stmt *tree.RestorePitr) (s
 	// restore according the restore level
 	switch restoreLevel {
 	case tree.RESTORELEVELCLUSTER:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelCluster)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelCluster)
 		subDbToRestore := make(map[string]*subDbRestoreRecord)
 		if err = restoreToCluster(ctx, ses, bh, pitrName, ts, subDbToRestore); err != nil {
 			return
@@ -1129,17 +1129,17 @@ func doRestorePitr(ctx context.Context, ses *Session, stmt *tree.RestorePitr) (s
 		}
 		return
 	case tree.RESTORELEVELACCOUNT:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelAccount)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelAccount)
 		if err = restoreToAccountWithPitr(ctx, ses.GetService(), bh, pitrName, ts, fkTableMap, viewMap, tenantInfo.TenantID); err != nil {
 			return
 		}
 	case tree.RESTORELEVELDATABASE:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelDatabase)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelDatabase)
 		if err = restoreToDatabaseWithPitr(ctx, ses.GetService(), bh, pitrName, ts, dbName, fkTableMap, viewMap, tenantInfo.TenantID); err != nil {
 			return
 		}
 	case tree.RESTORELEVELTABLE:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelTable)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelTable)
 		if err = restoreToTableWithPitr(ctx, ses.service, bh, pitrName, ts, dbName, tblName, fkTableMap, viewMap, tenantInfo.TenantID); err != nil {
 			return
 		}

pkg/frontend/snapshot.go

@@ -557,7 +557,7 @@ func doRestoreSnapshot(ctx context.Context, ses *Session, stmt *tree.RestoreSnap
 
 	// restore cluster
 	if stmt.Level == tree.RESTORELEVELCLUSTER {
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelCluster)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelCluster)
 
 		// restore cluster
 		subDbToRestore := make(map[string]*subDbRestoreRecord)
@@ -580,7 +580,7 @@ func doRestoreSnapshot(ctx context.Context, ses *Session, stmt *tree.RestoreSnap
 
 	// restore account by cluster level snapshot
 	if snapshot.level == tree.RESTORELEVELCLUSTER.String() && len(srcAccountName) != 0 {
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelCluster)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelCluster)
 
 		err = restoreToAccountUsingCluster(ctx, ses, bh, stmt, *snapshot)
 		if err != nil {
@@ -616,7 +616,7 @@ func doRestoreSnapshot(ctx context.Context, ses *Session, stmt *tree.RestoreSnap
 
 	switch stmt.Level {
 	case tree.RESTORELEVELACCOUNT:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelAccount)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelAccount)
 
 		if err = restoreToAccount(ctx,
 			ses.GetService(),
@@ -631,7 +631,7 @@ func doRestoreSnapshot(ctx context.Context, ses *Session, stmt *tree.RestoreSnap
 			return stats, err
 		}
 	case tree.RESTORELEVELDATABASE:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelDatabase)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelDatabase)
 
 		if err = restoreToDatabase(ctx,
 			ses.GetService(),
@@ -648,7 +648,7 @@ func doRestoreSnapshot(ctx context.Context, ses *Session, stmt *tree.RestoreSnap
 			return stats, err
 		}
 	case tree.RESTORELEVELTABLE:
-		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.CloneLevelTable)
+		ctx = context.WithValue(ctx, tree.CloneLevelCtxKey{}, tree.RestoreCloneLevelTable)
 
 		if err = restoreToTable(ctx,
 			ses.GetService(),

pkg/sql/parsers/tree/clone.go

@@ -41,14 +41,19 @@ func init() {
 }
 
 type CloneLevelCtxKey struct{}
-type CloneLevelType int
+type CloneLevelType uint64
 type CloneStmtType int
 
 const (
-	CloneLevelTable CloneLevelType = iota
-	CloneLevelDatabase
-	CloneLevelAccount
-	CloneLevelCluster
+	NormalCloneLevelTable CloneLevelType = 1 << iota
+	NormalCloneLevelDatabase
+	NormalCloneLevelAccount
+	NormalCloneLevelCluster
+
+	RestoreCloneLevelTable
+	RestoreCloneLevelDatabase
+	RestoreCloneLevelAccount
+	RestoreCloneLevelCluster
 )
 
 const (
@@ -162,25 +167,25 @@ func DecideCloneStmtType(
 	}
 
 	var (
-		level = CloneLevelTable
+		level = NormalCloneLevelTable
 	)
 
 	if val := ctx.Value(CloneLevelCtxKey{}); val != nil {
 		level = val.(CloneLevelType)
 	}
 
 	switch level {
-	case CloneLevelCluster:
+	case NormalCloneLevelCluster, RestoreCloneLevelCluster:
 		return CloneCluster
-	case CloneLevelAccount:
+	case NormalCloneLevelAccount, RestoreCloneLevelAccount:
 		return CloneAccount
-	case CloneLevelDatabase:
+	case NormalCloneLevelDatabase, RestoreCloneLevelDatabase:
 		if srcAccount == toAccount {
 			return WithinAccCloneDB
 		}
 		return BetweenAccCloneDB
 
-	case CloneLevelTable:
+	case NormalCloneLevelTable, RestoreCloneLevelTable:
 		if srcAccount == toAccount {
 			if srcDbName == dstDbName {
 				return WithinDBCloneTable

pkg/sql/plan/build_table_clone.go

@@ -15,6 +15,11 @@
 package plan
 
 import (
+	"context"
+	"slices"
+	"strings"
+
+	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/sql/parsers/tree"
@@ -135,6 +140,12 @@ func buildCloneTable(
 		dstAccount, srcAccount,
 	)
 
+	if err = checkPrivilege(
+		ctx.GetContext(), stmt, opAccount, srcAccount, dstAccount, srcTblDef, dstTblDef,
+	); err != nil {
+		return nil, err
+	}
+
 	if createTablePlan, err = buildCreateTable(ctx, &stmt.CreateTable, stmt); err != nil {
 		return nil, err
 	}
@@ -148,3 +159,56 @@ func buildCloneTable(
 
 	return createTablePlan, nil
 }
+
+func checkPrivilege(
+	ctx context.Context,
+	stmt *tree.CloneTable,
+	opAccount uint32,
+	dstAccount uint32,
+	srcAccount uint32,
+	srcTblDef *TableDef,
+	dstTblDef *TableDef,
+) (err error) {
+
+	// 1. only sys can clone from system databases
+	// 2. sys and non-sys both cannot clone to system database
+	// 3. if this is a restore clone stmt, skip this check
+
+	if val := ctx.Value(tree.CloneLevelCtxKey{}); val != nil {
+		switch val.(tree.CloneLevelType) {
+		case tree.RestoreCloneLevelAccount,
+			tree.RestoreCloneLevelCluster,
+			tree.RestoreCloneLevelDatabase,
+			tree.RestoreCloneLevelTable:
+			// skip this check
+			return nil
+		default:
+		}
+	}
+
+	var (
+		typ int
+	)
+
+	if slices.Index(
+		catalog.SystemDatabases, strings.ToLower(srcTblDef.DbName),
+	) != -1 {
+		// clone from system databases
+		typ = 1
+	} else if slices.Index(
+		catalog.SystemDatabases, strings.ToLower(dstTblDef.DbName),
+	) != -1 {
+		// clone to a system database
+		typ = 2
+	}
+
+	if typ == 2 {
+		return moerr.NewInternalErrorNoCtx("cannot clone data into system database")
+	} else if typ == 1 {
+		if opAccount != catalog.System_Account {
+			return moerr.NewInternalErrorNoCtx("non-sys account cannot clone data from system database")
+		}
+	}
+
+	return nil
+}

test/distributed/cases/snapshot/clone/clone_privilege.result

@@ -0,0 +1,35 @@
+drop account if exists acc1;
+create account acc1 admin_name "root1" identified by "111";
+create database db1 clone mo_catalog;
+internal error: non-sys account cannot clone data from system database
+create database db3 clone system;
+internal error: non-sys account cannot clone data from system database
+create database db4 clone information_schema;
+internal error: non-sys account cannot clone data from system database
+create database db6;
+create table db6.t1 clone mo_catalog.mo_tables;
+internal error: non-sys account cannot clone data from system database
+create table db6.t2 clone system.statement_info;
+internal error: non-sys account cannot clone data from system database
+create database db7;
+create table db7.t1 (a int primary key);
+create table mo_catalog.t1 clone db7.t1;
+internal error: cannot clone data into system database
+create table system.t1 clone db7.t1;
+internal error: cannot clone data into system database
+create database db1 clone mo_catalog;
+create database db2 clone system;
+create database db3;
+create table db3.t1 clone mo_catalog.mo_tables;
+create table db3.t2 clone system.statement_info;
+create database db4;
+create table db4.t1 (a int primary key);
+create table mo_catalog.t1 clone db4.t1;
+internal error: cannot clone data into system database
+create table system.t1 clone db4.t1;
+internal error: cannot clone data into system database
+drop account if exists acc1;
+drop database if exists db1;
+drop database if exists db2;
+drop database if exists db3;
+drop database if exists db4;

test/distributed/cases/snapshot/clone/clone_privilege.sql

@@ -0,0 +1,37 @@
+drop account if exists acc1;
+create account acc1 admin_name "root1" identified by "111";
+
+-- @session:id=1&user=acc1:root1&password=111
+create database db1 clone mo_catalog;
+create database db3 clone system;
+create database db4 clone information_schema;
+
+create database db6;
+create table db6.t1 clone mo_catalog.mo_tables;
+create table db6.t2 clone system.statement_info;
+
+create database db7;
+create table db7.t1 (a int primary key);
+
+create table mo_catalog.t1 clone db7.t1;
+create table system.t1 clone db7.t1;
+-- @session
+
+create database db1 clone mo_catalog;
+create database db2 clone system;
+
+create database db3;
+create table db3.t1 clone mo_catalog.mo_tables;
+create table db3.t2 clone system.statement_info;
+
+create database db4;
+create table db4.t1 (a int primary key);
+create table mo_catalog.t1 clone db4.t1;
+create table system.t1 clone db4.t1;
+
+
+drop account if exists acc1;
+drop database if exists db1;
+drop database if exists db2;
+drop database if exists db3;
+drop database if exists db4;