executor: fix case-insensitive grant and revoke on schema names

pkg/executor/grant.go

@@ -81,6 +81,9 @@ func (e *GrantExec) Next(ctx context.Context, _ *chunk.Chunk) error {
 	if len(dbName) == 0 {
 		dbName = e.Ctx().GetSessionVars().CurrentDB
 	}
+	if e.Level.Level == ast.GrantLevelDB {
+		dbName = getTargetSchemaName(e.Ctx(), dbName, e.is)
+	}
 
 	// For table & column level, check whether table exists and privilege is valid
 	if e.Level.Level == ast.GrantLevelTable {
@@ -493,7 +496,7 @@ func (e *GrantExec) grantLevelPriv(ctx context.Context, priv *ast.PrivElem, user
 		return e.grantDBLevel(priv, user, internalSession)
 	case ast.GrantLevelTable:
 		if len(priv.Cols) == 0 {
-			return e.grantTableLevel(priv, user, internalSession)
+			return e.grantTableLevel(ctx, priv, user, internalSession)
 		}
 		return e.grantColumnLevel(ctx, priv, user, internalSession)
 	default:
@@ -547,6 +550,7 @@ func (e *GrantExec) grantDBLevel(priv *ast.PrivElem, user *ast.UserSpec, interna
 	if len(dbName) == 0 {
 		dbName = e.Ctx().GetSessionVars().CurrentDB
 	}
+	dbName = getTargetSchemaName(e.Ctx(), dbName, e.is)
 
 	sql := new(strings.Builder)
 	sqlescape.MustFormatSQL(sql, "UPDATE %n.%n SET ", mysql.SystemDB, mysql.DBTable)
@@ -562,23 +566,28 @@ func (e *GrantExec) grantDBLevel(priv *ast.PrivElem, user *ast.UserSpec, interna
 }
 
 // grantTableLevel manipulates mysql.tables_priv table.
-func (e *GrantExec) grantTableLevel(priv *ast.PrivElem, user *ast.UserSpec, internalSession sessionctx.Context) error {
-	dbName := e.Level.DBName
-	if len(dbName) == 0 {
-		dbName = e.Ctx().GetSessionVars().CurrentDB
+func (e *GrantExec) grantTableLevel(ctx context.Context, priv *ast.PrivElem, user *ast.UserSpec, internalSession sessionctx.Context) error {
+	dbName, tbl, err := getTargetSchemaAndTable(ctx, e.Ctx(), e.Level.DBName, e.Level.TableName, e.is)
+	// Earlier validation may allow GRANT on a missing table, for example CREATE/ALL.
+	// Keep that behavior here: ignore ErrTableNotExists and continue with the input table name.
+	if err != nil && !terror.ErrorEqual(err, infoschema.ErrTableNotExists) {
+		return err
 	}
 	tblName := e.Level.TableName
+	if tbl != nil {
+		tblName = tbl.Meta().Name.O
+	}
 
 	sql := new(strings.Builder)
 	sqlescape.MustFormatSQL(sql, "UPDATE %n.%n SET ", mysql.SystemDB, mysql.TablePrivTable)
-	err := composeTablePrivUpdateForGrant(internalSession, sql, priv.Priv, user.User.Username, user.User.Hostname, dbName, tblName)
+	err = composeTablePrivUpdateForGrant(internalSession, sql, priv.Priv, user.User.Username, user.User.Hostname, dbName, tblName)
 	if err != nil {
 		return err
 	}
 	sqlescape.MustFormatSQL(sql, " WHERE User=%? AND Host=%? AND DB=%? AND Table_name=%?", user.User.Username, user.User.Hostname, dbName, tblName)
 
-	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnPrivilege)
-	_, err = internalSession.GetSQLExecutor().ExecuteInternal(ctx, sql.String())
+	internalCtx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnPrivilege)
+	_, err = internalSession.GetSQLExecutor().ExecuteInternal(internalCtx, sql.String())
 	return err
 }
 
@@ -783,6 +792,22 @@ func getColumnPriv(sctx sessionctx.Context, name string, host string, db string,
 	return cPriv, nil
 }
 
+// getTargetSchemaName returns the real schema name from infoschema when it exists.
+// Keep the original input for non-existent schemas so GRANT/REVOKE on them preserves
+// the current behavior.
+func getTargetSchemaName(sctx sessionctx.Context, dbName string, is infoschema.InfoSchema) string {
+	if len(dbName) == 0 {
+		dbName = sctx.GetSessionVars().CurrentDB
+	}
+	if len(dbName) == 0 {
+		return ""
+	}
+	if db, ok := is.SchemaByName(ast.NewCIStr(dbName)); ok {
+		return db.Name.O
+	}
+	return dbName
+}
+
 // getTargetSchemaAndTable finds the schema and table by dbName and tableName.
 func getTargetSchemaAndTable(ctx context.Context, sctx sessionctx.Context, dbName, tableName string, is infoschema.InfoSchema) (string, table.Table, error) {
 	if len(dbName) == 0 {
@@ -791,6 +816,7 @@ func getTargetSchemaAndTable(ctx context.Context, sctx sessionctx.Context, dbNam
 			return "", nil, errors.New("miss DB name for grant privilege")
 		}
 	}
+	dbName = getTargetSchemaName(sctx, dbName, is)
 	name := ast.NewCIStr(tableName)
 	tbl, err := is.TableByName(ctx, ast.NewCIStr(dbName), name)
 	if terror.ErrorEqual(err, infoschema.ErrTableNotExists) {

pkg/executor/grant_test.go

@@ -19,6 +19,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/parser/auth"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/testkit"
@@ -27,6 +29,34 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func newCollationDisabledBootstrapTestKit(t *testing.T) *testkit.TestKit {
+	t.Helper()
+
+	originCfg := *config.GetGlobalConfig()
+	t.Cleanup(func() {
+		config.StoreGlobalConfig(&originCfg)
+	})
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.NewCollationsEnabledOnFirstBootstrap = false
+	})
+
+	oldNewCollationEnabled := collate.NewCollationEnabled()
+	t.Cleanup(func() {
+		collate.SetNewCollationEnabledForTest(oldNewCollationEnabled)
+	})
+	// Keep the in-process flag aligned before bootstrap; bootstrap will reload
+	// and enforce the persisted cluster value again.
+	collate.SetNewCollationEnabledForTest(false)
+
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	require.NoError(t, tk.Session().Auth(&auth.UserIdentity{Username: "root", Hostname: "localhost"}, nil, nil, nil))
+	tk.MustQuery(`SELECT VARIABLE_VALUE FROM mysql.tidb WHERE VARIABLE_NAME='new_collation_enabled'`).
+		Check(testkit.Rows("False"))
+	require.False(t, collate.NewCollationEnabled())
+	return tk
+}
+
 func TestGrantGlobal(t *testing.T) {
 	store := testkit.CreateMockStore(t)
 
@@ -105,12 +135,7 @@ func TestGrantDBScope(t *testing.T) {
 }
 
 func TestGrantDBScopeCaseInsensitiveWithNewCollationDisabled(t *testing.T) {
-	oldNewCollationEnabled := collate.NewCollationEnabled()
-	collate.SetNewCollationEnabledForTest(false)
-	defer collate.SetNewCollationEnabledForTest(oldNewCollationEnabled)
-
-	store := testkit.CreateMockStore(t)
-	tk := testkit.NewTestKit(t, store)
+	tk := newCollationDisabledBootstrapTestKit(t)
 
 	tk.MustExec(`DROP USER IF EXISTS 'testDBCase'@'%'`)
 	tk.MustExec(`CREATE USER 'testDBCase'@'%' IDENTIFIED BY '123'`)
@@ -165,65 +190,122 @@ func TestGrantTableScope(t *testing.T) {
 }
 
 func TestGrantTableScopeCaseInsensitiveWithNewCollationDisabled(t *testing.T) {
-	oldNewCollationEnabled := collate.NewCollationEnabled()
-	collate.SetNewCollationEnabledForTest(false)
-	defer collate.SetNewCollationEnabledForTest(oldNewCollationEnabled)
+	t.Run("table-name", func(t *testing.T) {
+		tk := newCollationDisabledBootstrapTestKit(t)
 
-	store := testkit.CreateMockStore(t)
-	tk := testkit.NewTestKit(t, store)
+		tk.MustExec(`DROP USER IF EXISTS 'testTblCase'@'%'`)
+		tk.MustExec(`CREATE USER 'testTblCase'@'%' IDENTIFIED BY '123'`)
+		tk.MustExec(`DROP TABLE IF EXISTS test.issue66867_grant`)
+		tk.MustExec(`CREATE TABLE test.issue66867_grant(c1 int)`)
+
+		tk.MustExec(`GRANT SELECT ON test.issue66867_grant TO 'testTblCase'@'%'`)
+		tk.MustExec(`GRANT SELECT ON test.ISSUE66867_GRANT TO 'testTblCase'@'%'`)
+		tk.MustQuery(`SELECT Table_name, Table_priv FROM mysql.tables_priv WHERE User='testTblCase' AND Host='%' AND DB='test' ORDER BY Table_name`).
+			Check(testkit.Rows("issue66867_grant Select"))
+	})
+
+	t.Run("schema-name", func(t *testing.T) {
+		tk := newCollationDisabledBootstrapTestKit(t)
+
+		tk.MustExec(`DROP USER IF EXISTS 'testTblCaseSchema'@'%'`)
+		tk.MustExec(`CREATE USER 'testTblCaseSchema'@'%'`)
+		tk.MustExec(`DROP TABLE IF EXISTS test.issue68406_grant`)
+		tk.MustExec(`CREATE TABLE test.issue68406_grant(id int, name int)`)
+
+		tk.MustExec(`GRANT SELECT ON TEST.issue68406_grant TO 'testTblCaseSchema'@'%'`)
+		tk.MustExec(`GRANT SELECT,INSERT,UPDATE,DELETE ON test.issue68406_grant TO 'testTblCaseSchema'@'%'`)
+		tk.MustQuery(`SELECT DB, Table_name, Table_priv FROM mysql.tables_priv WHERE User='testTblCaseSchema' AND Host='%' ORDER BY DB, Table_name`).
+			Check(testkit.Rows("test issue68406_grant Select,Insert,Update,Delete"))
+		tk.MustQuery(`SHOW GRANTS FOR 'testTblCaseSchema'@'%'`).
+			Check(testkit.Rows(
+				"GRANT USAGE ON *.* TO 'testTblCaseSchema'@'%'",
+				"GRANT SELECT,INSERT,UPDATE,DELETE ON `test`.`issue68406_grant` TO 'testTblCaseSchema'@'%'",
+			))
+
+		tkUser := testkit.NewTestKit(t, tk.Session().GetStore())
+		require.NoError(t, tkUser.Session().Auth(&auth.UserIdentity{Username: "testTblCaseSchema", Hostname: "%"}, nil, nil, nil))
+		tkUser.MustExec(`INSERT INTO test.issue68406_grant VALUES (1, 2)`)
+		tkUser.MustQuery(`SELECT * FROM test.issue68406_grant`).Check(testkit.Rows("1 2"))
+	})
+
+	t.Run("missing-table fallback", func(t *testing.T) {
+		tk := newCollationDisabledBootstrapTestKit(t)
 
-	tk.MustExec(`DROP USER IF EXISTS 'testTblCase'@'%'`)
-	tk.MustExec(`CREATE USER 'testTblCase'@'%' IDENTIFIED BY '123'`)
-	tk.MustExec(`DROP TABLE IF EXISTS test.issue66867_grant`)
-	tk.MustExec(`CREATE TABLE test.issue66867_grant(c1 int)`)
+		tk.MustExec(`DROP USER IF EXISTS 'testTblCaseMissing'@'%'`)
+		tk.MustExec(`CREATE USER 'testTblCaseMissing'@'%'`)
+		tk.MustExec(`DROP TABLE IF EXISTS test.missing_tbl`)
 
-	tk.MustExec(`GRANT SELECT ON test.issue66867_grant TO 'testTblCase'@'%'`)
-	tk.MustExec(`GRANT SELECT ON test.ISSUE66867_GRANT TO 'testTblCase'@'%'`)
-	tk.MustQuery(`SELECT Table_name, Table_priv FROM mysql.tables_priv WHERE User='testTblCase' AND Host='%' AND DB='test' ORDER BY Table_name`).
-		Check(testkit.Rows("issue66867_grant Select"))
+		tk.MustExec(`GRANT CREATE ON TEST.missing_tbl TO 'testTblCaseMissing'@'%'`)
+		tk.MustQuery(`SELECT DB, Table_name, Table_priv FROM mysql.tables_priv WHERE User='testTblCaseMissing' AND Host='%'`).
+			Check(testkit.Rows("test missing_tbl Create"))
+		tk.MustQuery(`SHOW GRANTS FOR 'testTblCaseMissing'@'%'`).
+			Check(testkit.Rows(
+				"GRANT USAGE ON *.* TO 'testTblCaseMissing'@'%'",
+				"GRANT CREATE ON `test`.`missing_tbl` TO 'testTblCaseMissing'@'%'",
+			))
+	})
 }
 
 func TestGrantColumnScope(t *testing.T) {
-	store := testkit.CreateMockStore(t)
+	t.Run("basic", func(t *testing.T) {
+		store := testkit.CreateMockStore(t)
 
-	tk := testkit.NewTestKit(t, store)
-	// Create a new user.
-	createUserSQL := `CREATE USER 'testCol'@'localhost' IDENTIFIED BY '123';`
-	tk.MustExec(createUserSQL)
-	tk.MustExec(`CREATE TABLE test.test3(c1 int, c2 int);`)
+		tk := testkit.NewTestKit(t, store)
+		// Create a new user.
+		createUserSQL := `CREATE USER 'testCol'@'localhost' IDENTIFIED BY '123';`
+		tk.MustExec(createUserSQL)
+		tk.MustExec(`CREATE TABLE test.test3(c1 int, c2 int);`)
 
-	// Make sure all the column privs for new user is empty.
-	tk.MustQuery(`SELECT * FROM mysql.Columns_priv WHERE User="testCol" and host="localhost" and db="test" and Table_name="test3" and Column_name="c1"`).Check(testkit.Rows())
-	tk.MustQuery(`SELECT * FROM mysql.Columns_priv WHERE User="testCol" and host="localhost" and db="test" and Table_name="test3" and Column_name="c2"`).Check(testkit.Rows())
+		// Make sure all the column privs for new user is empty.
+		tk.MustQuery(`SELECT * FROM mysql.Columns_priv WHERE User="testCol" and host="localhost" and db="test" and Table_name="test3" and Column_name="c1"`).Check(testkit.Rows())
+		tk.MustQuery(`SELECT * FROM mysql.Columns_priv WHERE User="testCol" and host="localhost" and db="test" and Table_name="test3" and Column_name="c2"`).Check(testkit.Rows())
 
-	// Grant each priv to the user.
-	for _, v := range mysql.AllColumnPrivs {
-		sql := fmt.Sprintf("GRANT %s(c1) ON test.test3 TO 'testCol'@'localhost';", mysql.Priv2Str[v])
-		tk.MustExec(sql)
-		rows := tk.MustQuery(`SELECT Column_priv FROM mysql.Columns_priv WHERE User="testCol" and host="localhost" and db="test" and Table_name="test3" and Column_name="c1";`).Rows()
-		require.Len(t, rows, 1)
-		row := rows[0]
-		require.Len(t, rows, 1)
-		p := fmt.Sprintf("%v", row[0])
-		require.Greater(t, strings.Index(p, mysql.Priv2SetStr[v]), -1)
-	}
+		// Grant each priv to the user.
+		for _, v := range mysql.AllColumnPrivs {
+			sql := fmt.Sprintf("GRANT %s(c1) ON test.test3 TO 'testCol'@'localhost';", mysql.Priv2Str[v])
+			tk.MustExec(sql)
+			rows := tk.MustQuery(`SELECT Column_priv FROM mysql.Columns_priv WHERE User="testCol" and host="localhost" and db="test" and Table_name="test3" and Column_name="c1";`).Rows()
+			require.Len(t, rows, 1)
+			row := rows[0]
+			require.Len(t, rows, 1)
+			p := fmt.Sprintf("%v", row[0])
+			require.Greater(t, strings.Index(p, mysql.Priv2SetStr[v]), -1)
+		}
 
-	// Create a new user.
-	createUserSQL = `CREATE USER 'testCol1'@'localhost' IDENTIFIED BY '123';`
-	tk.MustExec(createUserSQL)
-	tk.MustExec("USE test;")
-	// Grant all column scope privs.
-	tk.MustExec("GRANT ALL(c2) ON test3 TO 'testCol1'@'localhost';")
-	// Make sure all the column privs for granted user are in the Column_priv set.
-	for _, v := range mysql.AllColumnPrivs {
-		rows := tk.MustQuery(`SELECT Column_priv FROM mysql.Columns_priv WHERE User="testCol1" and host="localhost" and db="test" and Table_name="test3" and Column_name="c2";`).Rows()
-		require.Len(t, rows, 1)
-		row := rows[0]
-		require.Len(t, rows, 1)
-		p := fmt.Sprintf("%v", row[0])
-		require.Greater(t, strings.Index(p, mysql.Priv2SetStr[v]), -1)
-	}
+		// Create a new user.
+		createUserSQL = `CREATE USER 'testCol1'@'localhost' IDENTIFIED BY '123';`
+		tk.MustExec(createUserSQL)
+		tk.MustExec("USE test;")
+		// Grant all column scope privs.
+		tk.MustExec("GRANT ALL(c2) ON test3 TO 'testCol1'@'localhost';")
+		// Make sure all the column privs for granted user are in the Column_priv set.
+		for _, v := range mysql.AllColumnPrivs {
+			rows := tk.MustQuery(`SELECT Column_priv FROM mysql.Columns_priv WHERE User="testCol1" and host="localhost" and db="test" and Table_name="test3" and Column_name="c2";`).Rows()
+			require.Len(t, rows, 1)
+			row := rows[0]
+			require.Len(t, rows, 1)
+			p := fmt.Sprintf("%v", row[0])
+			require.Greater(t, strings.Index(p, mysql.Priv2SetStr[v]), -1)
+		}
+
+		tk.MustGetErrMsg("GRANT SUPER(c2) ON test3 TO 'testCol1'@'localhost';",
+			"[executor:1221]Incorrect usage of COLUMN GRANT and NON-COLUMN PRIVILEGES")
+	})
+
+	t.Run("case-insensitive schema-name with new collation disabled", func(t *testing.T) {
+		tk := newCollationDisabledBootstrapTestKit(t)
+
+		tk.MustExec(`DROP USER IF EXISTS 'testColCaseSchema'@'%'`)
+		tk.MustExec(`CREATE USER 'testColCaseSchema'@'%' IDENTIFIED BY '123'`)
+		tk.MustExec(`DROP TABLE IF EXISTS test.issue68406_grant_col`)
+		tk.MustExec(`CREATE TABLE test.issue68406_grant_col(id int, name int)`)
 
-	tk.MustGetErrMsg("GRANT SUPER(c2) ON test3 TO 'testCol1'@'localhost';",
-		"[executor:1221]Incorrect usage of COLUMN GRANT and NON-COLUMN PRIVILEGES")
+		tk.MustExec(`GRANT SELECT(id) ON TEST.issue68406_grant_col TO 'testColCaseSchema'@'%'`)
+		tk.MustExec(`GRANT INSERT(id), UPDATE(name) ON test.issue68406_grant_col TO 'testColCaseSchema'@'%'`)
+		tk.MustQuery(`SELECT DB, Table_name, Column_name, Column_priv FROM mysql.columns_priv WHERE User='testColCaseSchema' AND Host='%' ORDER BY DB, Table_name, Column_name`).
+			Check(testkit.Rows(
+				"test issue68406_grant_col id Select,Insert",
+				"test issue68406_grant_col name Update",
+			))
+	})
 }

pkg/executor/revoke.go

@@ -157,44 +157,38 @@ func (e *RevokeExec) revokeOneUser(ctx context.Context, internalSession sessionc
 	if len(dbName) == 0 {
 		dbName = e.Ctx().GetSessionVars().CurrentDB
 	}
+	requestedDBName := dbName
 
 	// If there is no privilege entry in corresponding table, insert a new one.
 	// DB scope:		mysql.DB
 	// Table scope:		mysql.Tables_priv
 	// Column scope:	mysql.Columns_priv
 	switch e.Level.Level {
 	case ast.GrantLevelDB:
+		dbName = getTargetSchemaName(e.Ctx(), dbName, e.is)
 		ok, err := dbUserExists(internalSession, user, host, dbName)
 		if err != nil {
 			return err
 		}
 		if !ok {
-			return errors.Errorf("There is no such grant defined for user '%s' on host '%s' on database %s", user, host, dbName)
+			return errors.Errorf("There is no such grant defined for user '%s' on host '%s' on database %s", user, host, requestedDBName)
 		}
 	case ast.GrantLevelTable:
-		tblName := e.Level.TableName
-		if len(dbName) > 0 {
-			// Normalize db/table names before checking mysql.tables_priv.
-			// Different input case should still map to the same object.
-			dbNameCI := ast.NewCIStr(dbName)
-			if db, succ := e.is.SchemaByName(dbNameCI); succ {
-				dbName = db.Name.O
-			}
-			tbl, err := e.is.TableByName(ctx, dbNameCI, ast.NewCIStr(e.Level.TableName))
-			if err != nil && !terror.ErrorEqual(err, infoschema.ErrTableNotExists) {
-				return err
-			}
-			if err == nil {
-				// Use the real table name from schema metadata.
-				tblName = tbl.Meta().Name.O
-			}
+		requestedTblName := e.Level.TableName
+		dbName, tbl, err := getTargetSchemaAndTable(ctx, e.Ctx(), dbName, e.Level.TableName, e.is)
+		if err != nil && !terror.ErrorEqual(err, infoschema.ErrTableNotExists) {
+			return err
+		}
+		tblName := requestedTblName
+		if tbl != nil {
+			tblName = tbl.Meta().Name.O
 		}
 		ok, err := tableUserExists(internalSession, user, host, dbName, tblName)
 		if err != nil {
 			return err
 		}
 		if !ok {
-			return errors.Errorf("There is no such grant defined for user '%s' on host '%s' on table %s.%s", user, host, dbName, e.Level.TableName)
+			return errors.Errorf("There is no such grant defined for user '%s' on host '%s' on table %s.%s", user, host, requestedDBName, requestedTblName)
 		}
 	}
 
@@ -264,6 +258,7 @@ func (e *RevokeExec) revokeDBPriv(internalSession sessionctx.Context, priv *ast.
 	if len(dbName) == 0 {
 		dbName = e.Ctx().GetSessionVars().CurrentDB
 	}
+	dbName = getTargetSchemaName(e.Ctx(), dbName, e.is)
 
 	sql := new(strings.Builder)
 	sqlescape.MustFormatSQL(sql, "UPDATE %n.%n SET ", mysql.SystemDB, mysql.DBTable)