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

dm/pkg/checker/privilege.go

@@ -25,9 +25,9 @@ import (
 	_ "github.com/pingcap/tidb/types/parser_driver" // for parser driver
 	"github.com/pingcap/tidb/util/dbutil"
 	"github.com/pingcap/tidb/util/filter"
-	"go.uber.org/zap"
-
+	"github.com/pingcap/tidb/util/stringutil"
 	"github.com/pingcap/tiflow/dm/pkg/log"
+	"go.uber.org/zap"
 )
 
 // some privileges are only effective on global level. in other words, GRANT ALL ON test.* is not enough for them
@@ -172,7 +172,7 @@ func verifyPrivileges(result *Result, grants []string, lackPriv map[mysql.Privil
 			return NewError("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:
@@ -203,11 +203,16 @@ func verifyPrivileges(result *Result, grants []string, lackPriv map[mysql.Privil
 						if priv == mysql.GrantPriv {
 							continue
 						}
-						if _, ok := lackPriv[priv][dbName]; !ok {
-							continue
+						// in this old release branch, we don't have a flag that mark a privilege is global level.
+						// we use `deleted` to avoid delete a global level privilege when processing AllPriv.
+						deleted := false
+						for dbName := range lackPriv[priv] {
+							if stringutil.DoMatch(dbName, dbPatChar, dbPatType) {
+								delete(lackPriv[priv], dbName)
+							}
+							deleted = true
 						}
-						delete(lackPriv[priv], dbName)
-						if len(lackPriv[priv]) == 0 {
+						if deleted && len(lackPriv[priv]) == 0 {
 							delete(lackPriv, priv)
 						}
 					}
@@ -216,20 +221,22 @@ func verifyPrivileges(result *Result, grants []string, lackPriv map[mysql.Privil
 				if _, ok := lackPriv[privElem.Priv]; !ok {
 					continue
 				}
-				if _, ok := lackPriv[privElem.Priv][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(lackPriv[privElem.Priv], dbName)
+				for dbName := range lackPriv[privElem.Priv] {
+					if stringutil.DoMatch(dbName, dbPatChar, dbPatType) {
+						delete(lackPriv[privElem.Priv], dbName)
+					}
+				}
 				if len(lackPriv[privElem.Priv]) == 0 {
 					delete(lackPriv, privElem.Priv)
 				}
 			}
 		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

dm/pkg/checker/privilege_test.go

@@ -19,6 +19,7 @@ import (
 	tc "github.com/pingcap/check"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/util/filter"
+	"github.com/stretchr/testify/require"
 )
 
 func TestClient(t *testing.T) {
@@ -299,3 +300,81 @@ func (t *testCheckSuite) TestVerifyReplicationPrivileges(c *tc.C) {
 		}
 	}
 }
+
+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 := genExpectPriv(map[mysql.PrivilegeType]struct{}{
+			mysql.SelectPriv: {},
+		}, cs.checkTables)
+		err := verifyPrivileges(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)
+		}
+	}
+}