checker(dm): support wildcard in privilege checking (#7739)

close #7645
pingcap · Nov 30, 2022 · a2d2a52 · a2d2a52
1 parent 519c6f0
commit a2d2a52
Show file tree

Hide file tree

Showing 2 changed files with 92 additions and 14 deletions.
diff --git a/dm/pkg/checker/privilege.go b/dm/pkg/checker/privilege.go
@@ -26,6 +26,7 @@ import (
 	_ "github.com/pingcap/tidb/types/parser_driver" // for parser driver
 	"github.com/pingcap/tidb/util/dbutil"
 	"github.com/pingcap/tidb/util/filter"
+	"github.com/pingcap/tidb/util/stringutil"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/pkg/container/sortmap"
 	"go.uber.org/zap"
@@ -309,7 +310,7 @@ func VerifyPrivileges(
 			return nil, errors.Errorf("grant has no user %s", grant)
 		}
 
-		dbName := grantStmt.Level.DBName
+		dbPatChar, dbPatType := stringutil.CompilePattern(grantStmt.Level.DBName, '\\')
 		tableName := grantStmt.Level.TableName
 		switch grantStmt.Level.Level {
 		case ast.GrantLevelGlobal:
@@ -341,28 +342,31 @@ func VerifyPrivileges(
 						if privs.needGlobal {
 							continue
 						}
-						if _, ok := privs.dbs[dbName]; !ok {
-							continue
+						for dbName := range privs.dbs {
+							if stringutil.DoMatch(dbName, dbPatChar, dbPatType) {
+								delete(privs.dbs, dbName)
+							}
 						}
-						delete(privs.dbs, dbName)
 					}
 					continue
 				}
 				privs, ok := lackPrivs[privElem.Priv]
 				if !ok || privs.needGlobal {
 					continue
 				}
-				if _, ok := privs.dbs[dbName]; !ok {
-					continue
-				}
 				// dumpling could report error if an allow-list table is lack of privilege.
 				// we only check that SELECT is granted on all columns, otherwise we can't SHOW CREATE TABLE
 				if privElem.Priv == mysql.SelectPriv && len(privElem.Cols) != 0 {
 					continue
 				}
-				delete(privs.dbs, dbName)
+				for dbName := range privs.dbs {
+					if stringutil.DoMatch(dbName, dbPatChar, dbPatType) {
+						delete(privs.dbs, dbName)
+					}
+				}
 			}
 		case ast.GrantLevelTable:
+			dbName := grantStmt.Level.DBName
 			for _, privElem := range grantStmt.Privs {
 				// all privileges available at a given privilege level (except GRANT OPTION)
 				// from https://dev.mysql.com/doc/refman/5.7/en/privileges-provided.html#priv_all
@@ -375,9 +379,6 @@ func VerifyPrivileges(
 						if !ok || dbPrivs.wholeDB {
 							continue
 						}
-						if _, ok := dbPrivs.tables[tableName]; !ok {
-							continue
-						}
 						delete(dbPrivs.tables, tableName)
 					}
 					continue
@@ -390,9 +391,6 @@ func VerifyPrivileges(
 				if !ok || dbPrivs.wholeDB {
 					continue
 				}
-				if _, ok := dbPrivs.tables[tableName]; !ok {
-					continue
-				}
 				// dumpling could report error if an allow-list table is lack of privilege.
 				// we only check that SELECT is granted on all columns, otherwise we can't SHOW CREATE TABLE
 				if privElem.Priv == mysql.SelectPriv && len(privElem.Cols) != 0 {

diff --git a/dm/pkg/checker/privilege_test.go b/dm/pkg/checker/privilege_test.go
@@ -335,6 +335,86 @@ func TestVerifyReplicationPrivileges(t *testing.T) {
 	}
 }
 
+func TestVerifyPrivilegesWildcard(t *testing.T) {
+	cases := []struct {
+		grants           []string
+		checkTables      []filter.Table
+		replicationState State
+		errStr           string
+	}{
+		{
+			grants: []string{
+				"GRANT SELECT ON `demo\\_foobar`.* TO `dmuser`@`%`",
+			},
+			checkTables: []filter.Table{
+				{Schema: "demo_foobar", Name: "t1"},
+			},
+			replicationState: StateSuccess,
+		},
+		{
+			grants: []string{
+				"GRANT SELECT ON `demo\\_foobar`.* TO `dmuser`@`%`",
+			},
+			checkTables: []filter.Table{
+				{Schema: "demo2foobar", Name: "t1"},
+			},
+			replicationState: StateFailure,
+			errStr:           "lack of Select privilege: {`demo2foobar`.`t1`}; ",
+		},
+		{
+			grants: []string{
+				"GRANT SELECT ON `demo_`.* TO `dmuser`@`%`",
+			},
+			checkTables: []filter.Table{
+				{Schema: "demo1", Name: "t1"},
+				{Schema: "demo2", Name: "t1"},
+			},
+			replicationState: StateSuccess,
+		},
+		{
+			grants: []string{
+				"GRANT SELECT ON `demo%`.* TO `dmuser`@`%`",
+			},
+			checkTables: []filter.Table{
+				{Schema: "demo_some", Name: "t1"},
+				{Schema: "block_db", Name: "t1"},
+			},
+			replicationState: StateFailure,
+			errStr:           "lack of Select privilege: {`block_db`.`t1`}; ",
+		},
+		{
+			grants: []string{
+				"GRANT SELECT ON `demo_db`.`t1` TO `dmuser`@`%`",
+			},
+			checkTables: []filter.Table{
+				{Schema: "demo_db", Name: "t1"},
+				{Schema: "demo2db", Name: "t1"},
+			},
+			replicationState: StateFailure,
+			errStr:           "lack of Select privilege: {`demo2db`.`t1`}; ",
+		},
+	}
+
+	for i, cs := range cases {
+		t.Logf("case %d", i)
+		result := &Result{
+			State: StateFailure,
+		}
+		requiredPrivs := map[mysql.PrivilegeType]priv{
+			mysql.SelectPriv: {
+				dbs: genTableLevelPrivs(cs.checkTables),
+			},
+		}
+		err := verifyPrivilegesWithResult(result, cs.grants, requiredPrivs)
+		if cs.replicationState == StateSuccess {
+			require.Nil(t, err, "grants: %v", cs.grants)
+		} else {
+			require.NotNil(t, err, "grants: %v", cs.grants)
+			require.Equal(t, cs.errStr, err.ShortErr, "grants: %v", cs.grants)
+		}
+	}
+}
+
 func TestVerifyTargetPrivilege(t *testing.T) {
 	cases := []struct {
 		grants     []string