保存MySQL事件备份与恢复安全实现检查点（回归待修正）
This commit is contained in:
parent
863202fd15
commit
8cba0d5fd2
@ -15,7 +15,7 @@ import (
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
)
|
||||
|
||||
// backupMySQL 将 MySQL 的基础表、数据、例程和触发器导出为可在 MySQL 5.7/8.0 导入的 SQL 文件。
|
||||
// backupMySQL 将 MySQL 的基础表、数据、例程、触发器和事件导出为可在 MySQL 5.7/8.0 导入的 SQL 文件。
|
||||
func backupMySQL(ctx context.Context, cfg config.Config, output string) error {
|
||||
resolved, err := resolveMaintenanceDatabase(ctx, cfg.Database, cfg.Paths.Config)
|
||||
if err != nil {
|
||||
@ -59,6 +59,9 @@ func backupMySQLDatabase(ctx context.Context, cfg config.DatabaseConfig, output
|
||||
if _, err := fmt.Fprintf(file, "-- Scert MySQL 数据备份\n-- 生成时间：%s\n\nSET FOREIGN_KEY_CHECKS=0;\n\n", time.Now().UTC().Format(time.RFC3339)); err != nil {
|
||||
return fmt.Errorf("写入 MySQL 备份头失败：%w", err)
|
||||
}
|
||||
if _, err := fmt.Fprint(file, "-- 恢复前须进入维护窗口，停止事件调度并等待工作线程退出；恢复后先核验对象及警告，再由管理员恢复调度。\n\n"); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
rows, err := tx.QueryContext(ctx, "SELECT TABLE_NAME, ENGINE, TABLE_TYPE FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() ORDER BY TABLE_NAME")
|
||||
if err != nil {
|
||||
@ -92,6 +95,10 @@ func backupMySQLDatabase(ctx context.Context, cfg config.DatabaseConfig, output
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
events, err := listMySQLEvents(ctx, tx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, tableName := range tableNames {
|
||||
if err := dumpMySQLTable(ctx, tx, file, tableName); err != nil {
|
||||
@ -109,6 +116,10 @@ func backupMySQLDatabase(ctx context.Context, cfg config.DatabaseConfig, output
|
||||
if _, err := fmt.Fprint(file, "SET FOREIGN_KEY_CHECKS=1;\n"); err != nil {
|
||||
return fmt.Errorf("写入 MySQL 备份尾失败：%w", err)
|
||||
}
|
||||
// 事件必须排在基础数据、例程、触发器和外键恢复之后。
|
||||
if err := dumpMySQLEvents(ctx, tx, file, events); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("结束 MySQL 一致性备份事务失败：%w", err)
|
||||
}
|
||||
@ -342,10 +353,16 @@ func restoreMySQL(ctx context.Context, cfg config.DatabaseConfig, input string)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := preflightMySQLEventRestore(ctx, connection, statements); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, statement := range statements {
|
||||
if _, err := connection.ExecContext(ctx, statement); err != nil {
|
||||
return fmt.Errorf("执行 MySQL 恢复语句失败：%w", err)
|
||||
}
|
||||
if err := checkMySQLEventRestoreWarnings(ctx, connection, statement); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
96
backend/internal/cli/mysql_event_expiry_integration_test.go
Normal file
96
backend/internal/cli/mysql_event_expiry_integration_test.go
Normal file
@ -0,0 +1,96 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/testsupport/mysqltest"
|
||||
)
|
||||
|
||||
func TestMySQLRealEventExpiry(t *testing.T) {
|
||||
server := mysqltest.New(t)
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = OFF"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, scenario := range []string{"not_preserved", "enabled_preserved", "disabled_preserved"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
cfg := server.Database(t)
|
||||
db, err := database.Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
options := "NOT PRESERVE ENABLE"
|
||||
if scenario == "enabled_preserved" {
|
||||
options = "PRESERVE ENABLE"
|
||||
} else if scenario == "disabled_preserved" {
|
||||
options = "PRESERVE DISABLE"
|
||||
}
|
||||
if _, err := db.DB.ExecContext(t.Context(), "CREATE EVENT expiry ON SCHEDULE AT CURRENT_TIMESTAMP + INTERVAL 1 SECOND ON COMPLETION "+options+" DO SELECT 1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var expired bool
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if err := db.DB.QueryRowContext(t.Context(), "SELECT EXECUTE_AT < CURRENT_TIMESTAMP FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() AND EVENT_NAME = 'expiry'").Scan(&expired); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if expired {
|
||||
break
|
||||
}
|
||||
time.Sleep(25 * time.Millisecond)
|
||||
}
|
||||
if !expired {
|
||||
t.Fatal("未实际观测到事件到期")
|
||||
}
|
||||
output := filepath.Join(server.Root(), scenario+"-expiry.sql")
|
||||
if err := os.WriteFile(output, []byte("existing backup"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = backupMySQLDatabase(t.Context(), cfg, output)
|
||||
if scenario == "disabled_preserved" {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := restoreMySQL(t.Context(), cfg, output); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
checkMySQLInteropValue(t, cfg, "SELECT STATUS FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() AND EVENT_NAME = 'expiry'", "DISABLED")
|
||||
return
|
||||
}
|
||||
if err == nil || !strings.Contains(err.Error(), "已到期") {
|
||||
t.Fatalf("过期事件无法原样重建时应拒绝备份：%v", err)
|
||||
}
|
||||
data, err := os.ReadFile(output)
|
||||
if err != nil || string(data) != "existing backup" {
|
||||
t.Fatal("事件导出失败破坏了已有备份")
|
||||
}
|
||||
if _, err := os.Stat(output + ".tmp"); !os.IsNotExist(err) {
|
||||
t.Fatal("失败后遗留临时备份")
|
||||
}
|
||||
checkMySQLInteropValue(t, cfg, "SELECT CAST(COUNT(*) AS CHAR) FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE()", "1")
|
||||
})
|
||||
}
|
||||
for _, scenario := range []string{"dropped_after_create", "disabled_after_create"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
cfg := server.Database(t)
|
||||
completion := "NOT PRESERVE"
|
||||
if scenario == "disabled_after_create" {
|
||||
completion = "PRESERVE"
|
||||
}
|
||||
input := filepath.Join(server.Root(), scenario+".sql")
|
||||
script := "CREATE EVENT expired ON SCHEDULE AT '2001-01-01 00:00:00' ON COMPLETION " + completion + " ENABLE DO SELECT 1;\nCREATE TABLE must_not_execute (id INT);"
|
||||
if err := os.WriteFile(input, []byte(script), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := restoreMySQL(t.Context(), cfg, input); err == nil || !strings.Contains(err.Error(), "未保留原定义或状态") {
|
||||
t.Fatalf("事件被服务器删除或禁用时不能报告成功：%v", err)
|
||||
}
|
||||
checkMySQLInteropValue(t, cfg, "SELECT CAST(COUNT(*) AS CHAR) FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE()", "0")
|
||||
})
|
||||
}
|
||||
}
|
||||
229
backend/internal/cli/mysql_event_restore.go
Normal file
229
backend/internal/cli/mysql_event_restore.go
Normal file
@ -0,0 +1,229 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// 预检必须先于任何恢复语句；不会擅自关闭实例调度器或修改已有事件状态。
|
||||
// 管理员仍须维持维护窗口，禁止外部会话在预检后开启调度或修改事件。
|
||||
func preflightMySQLEventRestore(ctx context.Context, conn *sql.Conn, statements []string) error {
|
||||
active := false
|
||||
for _, statement := range statements {
|
||||
// 两种字符串转义模式中任一种无法证明安全，就要求停止调度。
|
||||
for _, noBackslash := range []bool{false, true} {
|
||||
tokens := mysqlEventTokens(statement, noBackslash)
|
||||
if len(tokens) > 0 && tokens[0] == "SET" {
|
||||
for _, token := range tokens[1:] {
|
||||
if token == "EVENT_SCHEDULER" {
|
||||
return fmt.Errorf("MySQL 恢复文件不能修改事件调度器，请由管理员在维护窗口单独管理")
|
||||
}
|
||||
}
|
||||
}
|
||||
active = active || mysqlStatementMayEnableEvent(tokens)
|
||||
}
|
||||
}
|
||||
tx, err := conn.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
var version, scheduler string
|
||||
if err := tx.QueryRowContext(ctx, "SELECT VERSION(), @@GLOBAL.event_scheduler").Scan(&version, &scheduler); err != nil {
|
||||
return fmt.Errorf("读取 MySQL 事件调度状态失败：%w", err)
|
||||
}
|
||||
grants, err := readMySQLEffectiveGrants(ctx, tx, version)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
process := false
|
||||
for _, grant := range grants {
|
||||
match := mysqlPrivilegeGrant.FindStringSubmatch(grant)
|
||||
if len(match) != 0 && match[1] == "GRANT" && match[3] == "*" && match[4] == "*" && mysqlHasPrivilege(match[2], "PROCESS") {
|
||||
process = true
|
||||
}
|
||||
}
|
||||
if !process {
|
||||
return fmt.Errorf("MySQL 恢复需要 PROCESS 权限以确认事件工作线程已退出，不能把不可见线程当成没有运行")
|
||||
}
|
||||
var workers int
|
||||
// 事件工作线程以定义者身份运行，不能按 event_scheduler 用户名筛选。
|
||||
// Connect 也可能包含正在建立的普通连接；维护窗口内保守拒绝，不冒险误判为空。
|
||||
if err := tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE COMMAND = 'Connect'").Scan(&workers); err != nil {
|
||||
return fmt.Errorf("读取 MySQL 事件工作线程失败：%w", err)
|
||||
}
|
||||
if workers != 0 {
|
||||
return fmt.Errorf("MySQL 仍有事件工作线程运行，请等待其退出后再恢复；不会自动终止事件")
|
||||
}
|
||||
if strings.EqualFold(scheduler, "ON") {
|
||||
if err := requireMySQLEventVisibility(ctx, tx); err != nil {
|
||||
return err
|
||||
}
|
||||
var enabled int
|
||||
if err := tx.QueryRowContext(ctx, "SELECT COUNT(*) FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() AND STATUS = 'ENABLED'").Scan(&enabled); err != nil {
|
||||
return err
|
||||
}
|
||||
if active || enabled != 0 {
|
||||
return fmt.Errorf("MySQL 事件调度器仍为 ON，备份或目标库含启用事件；请管理员在维护窗口停止调度并等待事件工作线程退出后重试")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func mysqlStatementMayEnableEvent(tokens []string) bool {
|
||||
_, active := mysqlEventStatement(tokens)
|
||||
return active
|
||||
}
|
||||
|
||||
func mysqlEventStatement(tokens []string) (bool, bool) {
|
||||
if len(tokens) < 2 || (tokens[0] != "CREATE" && tokens[0] != "ALTER") {
|
||||
return false, false
|
||||
}
|
||||
i := 1
|
||||
// 只识别语句开头的对象类型，不把例程正文、注释或标识符中的 EVENT 当成事件。
|
||||
if tokens[i] == "DEFINER" {
|
||||
i += 3 // 跳过 DEFINER、等号及账户名称。
|
||||
if i < len(tokens) && tokens[i] == "@" {
|
||||
i += 2
|
||||
} else if i+1 < len(tokens) && tokens[i] == "(" && tokens[i+1] == ")" {
|
||||
i += 2
|
||||
}
|
||||
}
|
||||
if i >= len(tokens) || tokens[i] != "EVENT" {
|
||||
return false, false
|
||||
}
|
||||
i++
|
||||
if i+2 < len(tokens) && tokens[i] == "IF" && tokens[i+1] == "NOT" && tokens[i+2] == "EXISTS" {
|
||||
i += 3
|
||||
}
|
||||
// EVENT、ENABLE、DISABLE 等可能作为未引用的名称，不能把名称当成状态。
|
||||
i = mysqlSkipEventName(tokens, i)
|
||||
disabled, depth := false, 0
|
||||
for ; i < len(tokens); i++ {
|
||||
token := tokens[i]
|
||||
switch token {
|
||||
case "(":
|
||||
depth++
|
||||
case ")":
|
||||
depth--
|
||||
default:
|
||||
if depth != 0 {
|
||||
continue
|
||||
}
|
||||
switch token {
|
||||
case "RENAME":
|
||||
if i+1 < len(tokens) && tokens[i+1] == "TO" {
|
||||
i = mysqlSkipEventName(tokens, i+2) - 1
|
||||
}
|
||||
case "DO":
|
||||
return true, !disabled
|
||||
case "DISABLE":
|
||||
disabled = true
|
||||
case "ENABLE":
|
||||
disabled = false
|
||||
}
|
||||
}
|
||||
}
|
||||
return true, !disabled
|
||||
}
|
||||
|
||||
func checkMySQLEventRestoreWarnings(ctx context.Context, conn *sql.Conn, statement string) error {
|
||||
event, active := mysqlEventStatement(mysqlEventTokens(statement, false))
|
||||
otherEvent, otherActive := mysqlEventStatement(mysqlEventTokens(statement, true))
|
||||
if !event && !otherEvent {
|
||||
return nil
|
||||
}
|
||||
rows, err := conn.QueryContext(ctx, "SHOW WARNINGS")
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取 MySQL 事件恢复警告失败：%w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
for rows.Next() {
|
||||
var level, message string
|
||||
var code int
|
||||
if err := rows.Scan(&level, &code, &message); err != nil {
|
||||
return err
|
||||
}
|
||||
// 到期自动删除、拒绝修改或改变启停状态，不能算作成功恢复。
|
||||
if code == 1588 || code == 1589 || code == 1605 || (code == 1544 && (active || otherActive)) {
|
||||
return fmt.Errorf("MySQL 事件恢复未保留原定义或状态（警告 %d）：%s", code, message)
|
||||
}
|
||||
}
|
||||
return rows.Err()
|
||||
}
|
||||
|
||||
func mysqlSkipEventName(tokens []string, start int) int {
|
||||
if start >= len(tokens) {
|
||||
return start
|
||||
}
|
||||
start++
|
||||
if start+1 < len(tokens) && tokens[start] == "." {
|
||||
start += 2
|
||||
}
|
||||
return start
|
||||
}
|
||||
|
||||
// 只用于调度预检，不执行或改写 SQL。版本化可执行注释按可能执行处理。
|
||||
func mysqlEventTokens(statement string, noBackslash bool) []string {
|
||||
var tokens []string
|
||||
for i := 0; i < len(statement); {
|
||||
c := statement[i]
|
||||
if c <= ' ' {
|
||||
i++
|
||||
continue
|
||||
}
|
||||
if c == '#' || (c == '-' && i+2 < len(statement) && statement[i+1] == '-' && statement[i+2] <= ' ') {
|
||||
for i < len(statement) && statement[i] != '\n' {
|
||||
i++
|
||||
}
|
||||
continue
|
||||
}
|
||||
if c == '/' && i+1 < len(statement) && statement[i+1] == '*' {
|
||||
end := strings.Index(statement[i+2:], "*/")
|
||||
if end < 0 {
|
||||
return append(tokens, "?")
|
||||
}
|
||||
end += i + 2
|
||||
if i+2 < end && statement[i+2] == '!' {
|
||||
start := i + 3
|
||||
for start < end && statement[start] >= '0' && statement[start] <= '9' {
|
||||
start++
|
||||
}
|
||||
tokens = append(tokens, mysqlEventTokens(statement[start:end], noBackslash)...)
|
||||
}
|
||||
i = end + 2
|
||||
continue
|
||||
}
|
||||
if c == '\'' || c == '"' || c == '`' {
|
||||
i++
|
||||
for i < len(statement) {
|
||||
if statement[i] == '\\' && c != '`' && !noBackslash {
|
||||
i += 2
|
||||
continue
|
||||
}
|
||||
if statement[i] == c {
|
||||
i++
|
||||
if i < len(statement) && statement[i] == c {
|
||||
i++
|
||||
continue
|
||||
}
|
||||
break
|
||||
}
|
||||
i++
|
||||
}
|
||||
tokens = append(tokens, "?")
|
||||
continue
|
||||
}
|
||||
start := i
|
||||
for i < len(statement) && ((statement[i] >= 'a' && statement[i] <= 'z') || (statement[i] >= 'A' && statement[i] <= 'Z') || (statement[i] >= '0' && statement[i] <= '9') || statement[i] == '_' || statement[i] >= 128) {
|
||||
i++
|
||||
}
|
||||
if i == start {
|
||||
i++
|
||||
}
|
||||
tokens = append(tokens, strings.ToUpper(statement[start:i]))
|
||||
}
|
||||
return tokens
|
||||
}
|
||||
41
backend/internal/cli/mysql_event_restore_test.go
Normal file
41
backend/internal/cli/mysql_event_restore_test.go
Normal file
@ -0,0 +1,41 @@
|
||||
package cli
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestMySQLEventStatementSafety(t *testing.T) {
|
||||
for _, item := range []struct {
|
||||
name, statement string
|
||||
active bool
|
||||
}{
|
||||
{"default_enabled", "CREATE EVENT task ON SCHEDULE EVERY 1 DAY DO SELECT 1", true},
|
||||
{"explicit_enabled", "CREATE DEFINER=`owner`@`host` EVENT task ON SCHEDULE EVERY 1 DAY ENABLE DO SELECT 1", true},
|
||||
{"disabled", "CREATE EVENT task ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 'ENABLE'", false},
|
||||
{"replica_disabled", "CREATE EVENT task ON SCHEDULE EVERY 1 DAY DISABLE ON SLAVE DO SELECT 1", false},
|
||||
{"body_not_status", "CREATE EVENT task ON SCHEDULE EVERY 1 DAY DO SELECT 'DISABLE'", true},
|
||||
{"comment_not_status", "CREATE EVENT task ON SCHEDULE EVERY 1 DAY COMMENT 'DISABLE' DO SELECT 1", true},
|
||||
{"name_not_status", "CREATE EVENT disable ON SCHEDULE EVERY 1 DAY DO SELECT 1", true},
|
||||
{"schema_not_status", "CREATE EVENT disable.task ON SCHEDULE EVERY 1 DAY DO SELECT 1", true},
|
||||
{"quoted_name", "CREATE EVENT `DO``DISABLE` ON SCHEDULE EVERY 1 DAY DO SELECT 1", true},
|
||||
{"if_not_exists", "CREATE EVENT IF NOT EXISTS disable ON SCHEDULE EVERY 1 DAY DO SELECT 1", true},
|
||||
{"definer_keywords", "CREATE DEFINER=event@disable EVENT task ON SCHEDULE EVERY 1 DAY DO SELECT 1", true},
|
||||
{"current_user", "CREATE DEFINER=CURRENT_USER() EVENT task ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 1", false},
|
||||
{"versioned", "/*!50106 CREATE*/ /*!50117 DEFINER=`owner`@`host`*/ /*!50106 EVENT task ON SCHEDULE EVERY 1 DAY DO SELECT 1 */", true},
|
||||
{"versioned_disabled", "/*!50106 CREATE*/ /*!50117 DEFINER=`owner`@`host`*/ /*!50106 EVENT task ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 1 */", false},
|
||||
{"alter_enable", "ALTER EVENT task ENABLE", true},
|
||||
{"alter_disable", "ALTER EVENT task DISABLE", false},
|
||||
{"alter_rename", "ALTER EVENT task RENAME TO disable", true},
|
||||
{"alter_preserves_unknown_state", "ALTER EVENT task DO SELECT 'DISABLE'", true},
|
||||
{"drop", "DROP EVENT IF EXISTS task", false},
|
||||
{"ordinary_comment", "/* CREATE EVENT task ENABLE */ SELECT 1", false},
|
||||
{"line_comment", "-- CREATE EVENT task ENABLE\nCREATE EVENT task ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 1", false},
|
||||
{"routine", "CREATE DEFINER=`owner`@`host` PROCEDURE task() SELECT 'EVENT DISABLE'", false},
|
||||
} {
|
||||
t.Run(item.name, func(t *testing.T) {
|
||||
for _, mode := range []bool{false, true} {
|
||||
if actual := mysqlStatementMayEnableEvent(mysqlEventTokens(item.statement, mode)); actual != item.active {
|
||||
t.Fatalf("事件调度识别错误：得到 %v，期望 %v，语句 %s", actual, item.active, item.statement)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
147
backend/internal/cli/mysql_event_safety_integration_test.go
Normal file
147
backend/internal/cli/mysql_event_safety_integration_test.go
Normal file
@ -0,0 +1,147 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-sql-driver/mysql"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/testsupport/mysqltest"
|
||||
)
|
||||
|
||||
func TestMySQLRealEventRestoreSafety(t *testing.T) {
|
||||
server := mysqltest.New(t)
|
||||
for _, scenario := range []string{"incoming_enabled", "incoming_versioned", "incoming_default", "target_enabled", "target_hidden", "disabled_allowed", "scheduler_off", "missing_process", "script_changes_scheduler"} {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
cfg := server.Database(t)
|
||||
parsed, err := mysql.ParseDSN(cfg.DSN)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
account := mysqlStringLiteral(parsed.User) + "@'127.0.0.1'"
|
||||
db, err := database.Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if _, err := db.DB.ExecContext(t.Context(), "CREATE TABLE restore_marker (id INT PRIMARY KEY) ENGINE=InnoDB"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
event := "CREATE EVENT scheduled ON SCHEDULE AT '2036-01-01 00:00:00' ON COMPLETION PRESERVE ENABLE DO SELECT 'DISABLE'"
|
||||
script := "INSERT INTO restore_marker VALUES (1);\n" + event + ";"
|
||||
state, expected := "ON", "事件调度器仍为 ON"
|
||||
switch scenario {
|
||||
case "incoming_versioned":
|
||||
script = "INSERT INTO restore_marker VALUES (1);\n/*!50106 " + event + " */;"
|
||||
case "incoming_default":
|
||||
script = "INSERT INTO restore_marker VALUES (1);\nCREATE EVENT disable ON SCHEDULE EVERY 1 DAY DO SELECT 1;"
|
||||
case "target_enabled", "target_hidden":
|
||||
if _, err := db.DB.ExecContext(t.Context(), event); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
script = "INSERT INTO restore_marker VALUES (1);"
|
||||
if scenario == "target_hidden" {
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "REVOKE EVENT ON "+quoteMySQLIdentifier(parsed.DBName)+".* FROM "+account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
expected = "EVENT"
|
||||
}
|
||||
case "disabled_allowed":
|
||||
script = "INSERT INTO restore_marker VALUES (1);\nCREATE EVENT scheduled ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 1;"
|
||||
expected = ""
|
||||
case "scheduler_off":
|
||||
state, expected = "OFF", ""
|
||||
case "missing_process":
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "REVOKE PROCESS ON *.* FROM "+account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
expected = "PROCESS"
|
||||
case "script_changes_scheduler":
|
||||
script = "INSERT INTO restore_marker VALUES (1);\nSET GLOBAL event_scheduler = ON;"
|
||||
expected = "恢复文件不能修改"
|
||||
}
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = "+state); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
input := filepath.Join(server.Root(), scenario+"-restore.sql")
|
||||
if err := os.WriteFile(input, []byte(script), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
err = restoreMySQL(t.Context(), cfg, input)
|
||||
count := "0"
|
||||
if expected == "" {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
count = "1"
|
||||
} else if err == nil || !strings.Contains(err.Error(), expected) {
|
||||
t.Fatalf("恢复前应拒绝且不执行首条语句，期望 %s，得到 %v", expected, err)
|
||||
}
|
||||
checkMySQLInteropValue(t, cfg, "SELECT CAST(COUNT(*) AS CHAR) FROM restore_marker", count)
|
||||
checkMySQLInteropValue(t, cfg, "SELECT @@GLOBAL.event_scheduler", state)
|
||||
})
|
||||
}
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = OFF"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// OFF 不代表已经执行的事件被终止；实库确认工作线程存在时必须拒绝恢复。
|
||||
t.Run("running_worker_after_scheduler_off", func(t *testing.T) {
|
||||
cfg := server.Database(t)
|
||||
db, err := database.Open(t.Context(), cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
if _, err := db.DB.ExecContext(t.Context(), "CREATE EVENT waiting_event ON SCHEDULE AT CURRENT_TIMESTAMP + INTERVAL 1 SECOND ON COMPLETION PRESERVE ENABLE DO SELECT SLEEP(4)"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = ON"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
deadline := time.Now().Add(8 * time.Second)
|
||||
var workers int
|
||||
for time.Now().Before(deadline) {
|
||||
if err := server.Admin().QueryRowContext(t.Context(), "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE COMMAND = 'Connect' AND STATE = 'User sleep'").Scan(&workers); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if workers > 0 {
|
||||
break
|
||||
}
|
||||
time.Sleep(25 * time.Millisecond)
|
||||
}
|
||||
if workers == 0 {
|
||||
t.Fatal("没有观测到事件工作线程，不能把未覆盖的场景算作通过")
|
||||
}
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = OFF"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
input := filepath.Join(server.Root(), "running-worker.sql")
|
||||
if err := os.WriteFile(input, []byte("CREATE TABLE must_not_exist (id INT);"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := restoreMySQL(t.Context(), cfg, input); err == nil || !strings.Contains(err.Error(), "工作线程") {
|
||||
t.Fatalf("事件尚未退出时应拒绝：%v", err)
|
||||
}
|
||||
checkMySQLInteropValue(t, cfg, "SELECT CAST(COUNT(*) AS CHAR) FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE()", "0")
|
||||
deadline = time.Now().Add(8 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if err := server.Admin().QueryRowContext(t.Context(), "SELECT COUNT(*) FROM information_schema.PROCESSLIST WHERE COMMAND = 'Connect'").Scan(&workers); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if workers == 0 {
|
||||
break
|
||||
}
|
||||
time.Sleep(25 * time.Millisecond)
|
||||
}
|
||||
if workers != 0 {
|
||||
t.Fatal("事件线程未在预期时间自然退出")
|
||||
}
|
||||
if err := restoreMySQL(t.Context(), cfg, input); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
113
backend/internal/cli/mysql_events.go
Normal file
113
backend/internal/cli/mysql_events.go
Normal file
@ -0,0 +1,113 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"unicode/utf8"
|
||||
)
|
||||
|
||||
func listMySQLEvents(ctx context.Context, tx *sql.Tx) ([]string, error) {
|
||||
if err := requireMySQLEventVisibility(ctx, tx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
rows, err := tx.QueryContext(ctx, "SELECT EVENT_NAME FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() ORDER BY EVENT_NAME")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 MySQL 事件列表失败：%w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var names []string
|
||||
for rows.Next() {
|
||||
var name string
|
||||
if err := rows.Scan(&name); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
names = append(names, name)
|
||||
}
|
||||
return names, rows.Err()
|
||||
}
|
||||
|
||||
func dumpMySQLEvents(ctx context.Context, tx *sql.Tx, output io.Writer, names []string) error {
|
||||
if len(names) == 0 {
|
||||
return nil
|
||||
}
|
||||
var databaseCollation string
|
||||
if err := tx.QueryRowContext(ctx, "SELECT @@collation_database").Scan(&databaseCollation); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := fmt.Fprintln(output, `SET @scert_event_sql_mode = @@SESSION.sql_mode;
|
||||
SET @scert_event_charset = @@SESSION.character_set_client;
|
||||
SET @scert_event_collation = @@SESSION.collation_connection;
|
||||
SET @scert_event_time_zone = @@SESSION.time_zone;`); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, name := range names {
|
||||
var expired sql.NullBool
|
||||
var status, completion string
|
||||
query := `SELECT CASE
|
||||
WHEN EXECUTE_AT IS NOT NULL THEN EXECUTE_AT <= CONVERT_TZ(UTC_TIMESTAMP(), '+00:00', TIME_ZONE)
|
||||
WHEN ENDS IS NOT NULL THEN ENDS <= CONVERT_TZ(UTC_TIMESTAMP(), '+00:00', TIME_ZONE)
|
||||
ELSE FALSE END, STATUS, ON_COMPLETION
|
||||
FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() AND EVENT_NAME = ?`
|
||||
if err := tx.QueryRowContext(ctx, query, name).Scan(&expired, &status, &completion); err != nil {
|
||||
return fmt.Errorf("读取 MySQL 事件到期状态失败：%w", err)
|
||||
}
|
||||
if !expired.Valid || (expired.Bool && (status == "ENABLED" || completion == "NOT PRESERVE")) {
|
||||
return fmt.Errorf("MySQL 事件已到期或时区不可验证，重新创建将改变状态或删除对象，拒绝生成不完整备份：%s", name)
|
||||
}
|
||||
definition, err := readMySQLStoredDefinition(ctx, tx, "EVENT", name, "Create Event")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
zone, exists := definition["time_zone"]
|
||||
if !exists || zone == "" {
|
||||
return fmt.Errorf("MySQL 事件时区字段不完整：%s", name)
|
||||
}
|
||||
if err := normalizeMySQLEventMetadata(ctx, tx, name, definition); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := fmt.Fprintf(output, "SET SESSION time_zone = %s;\n", mysqlStringLiteral(zone)); err != nil {
|
||||
return err
|
||||
}
|
||||
drop := "DROP EVENT IF EXISTS " + quoteMySQLIdentifier(name)
|
||||
if err := writeMySQLStoredDefinition(ctx, tx, output, definition, "Create Event", drop); err != nil {
|
||||
return fmt.Errorf("写入 MySQL 事件失败（%s）：%w", name, err)
|
||||
}
|
||||
}
|
||||
_, err := fmt.Fprintf(output, "ALTER DATABASE COLLATE %s;\nSET SESSION sql_mode = @scert_event_sql_mode;\nSET SESSION character_set_client = @scert_event_charset;\nSET SESSION collation_connection = @scert_event_collation;\nSET SESSION time_zone = @scert_event_time_zone;\n\n", quoteMySQLIdentifier(databaseCollation))
|
||||
return err
|
||||
}
|
||||
|
||||
// 两个版本的事件声明头均来自 UTF-8 目录，正文仍使用原客户端编码。
|
||||
// 先与目录正文比对边界，再单独反解声明头；失败不写回部分转换结果。
|
||||
func normalizeMySQLEventMetadata(ctx context.Context, tx *sql.Tx, name string, definition map[string]string) error {
|
||||
charset := definition["character_set_client"]
|
||||
if charset == "utf8" || charset == "utf8mb3" || charset == "utf8mb4" {
|
||||
return nil
|
||||
}
|
||||
var body string
|
||||
if err := tx.QueryRowContext(ctx, "SELECT EVENT_DEFINITION FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() AND EVENT_NAME = ?", name).Scan(&body); err != nil {
|
||||
return fmt.Errorf("读取 MySQL 事件编码元数据失败：%w", err)
|
||||
}
|
||||
original := definition["Create Event"]
|
||||
if body == "" || !strings.HasSuffix(original, body) {
|
||||
return fmt.Errorf("MySQL 事件正文与目录元数据不一致，拒绝猜测编码：%s", name)
|
||||
}
|
||||
header := original[:len(original)-len(body)]
|
||||
if !strings.HasSuffix(header, " DO ") {
|
||||
return fmt.Errorf("MySQL 事件声明头不完整：%s", name)
|
||||
}
|
||||
identifier := quoteMySQLIdentifier(charset)
|
||||
var encoded, roundTrip []byte
|
||||
query := "SELECT CAST(CONVERT(? USING " + identifier + ") AS BINARY), CAST(CONVERT(CONVERT(? USING " + identifier + ") USING utf8mb4) AS BINARY)"
|
||||
if err := tx.QueryRowContext(ctx, query, header, header).Scan(&encoded, &roundTrip); err != nil {
|
||||
return fmt.Errorf("MySQL 事件目录元数据无法无损还原为 UTF-8（%s）：%w", name, err)
|
||||
}
|
||||
if string(roundTrip) != header || !utf8.Valid(encoded) {
|
||||
return fmt.Errorf("MySQL 事件目录元数据无法无损还原为 UTF-8：%s", name)
|
||||
}
|
||||
definition["Create Event"] = string(encoded) + body
|
||||
return nil
|
||||
}
|
||||
273
backend/internal/cli/mysql_events_integration_test.go
Normal file
273
backend/internal/cli/mysql_events_integration_test.go
Normal file
@ -0,0 +1,273 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-sql-driver/mysql"
|
||||
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/database"
|
||||
"git.pcux.cn/xuans/Scert/backend/internal/testsupport/mysqltest"
|
||||
)
|
||||
|
||||
func TestMySQLRealNativeEvents(t *testing.T) {
|
||||
server := mysqltest.New(t)
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = OFF"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := server.Database(t)
|
||||
parsed, err := mysql.ParseDSN(source.DSN)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
grantMySQLProgramAccess(t, server, parsed.User, parsed.DBName)
|
||||
db, err := database.Open(t.Context(), source)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
conn, err := db.DB.Conn(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
for _, statement := range []string{
|
||||
"CREATE TABLE event_values (id INT PRIMARY KEY, value INT NOT NULL) ENGINE=InnoDB",
|
||||
"INSERT INTO event_values VALUES (1, 10)",
|
||||
"CREATE PROCEDURE event_action() UPDATE event_values SET value = value + 7 WHERE id = 1",
|
||||
"SET SESSION time_zone = '+08:00'",
|
||||
"SET SESSION sql_mode = 'STRICT_TRANS_TABLES,NO_ENGINE_SUBSTITUTION'",
|
||||
"CREATE EVENT `中文``事件` ON SCHEDULE AT '2036-01-02 03:04:05' ON COMPLETION PRESERVE DISABLE COMMENT '说明;;不要改写 DISABLE' DO BEGIN CALL event_action(); UPDATE event_values SET value = value + 3 WHERE id = 1; END",
|
||||
"SET SESSION time_zone = '-05:00'",
|
||||
"CREATE EVENT repeating_event ON SCHEDULE EVERY 2 DAY STARTS '2036-02-03 04:05:06' ENDS '2037-03-04 05:06:07' ON COMPLETION PRESERVE ENABLE COMMENT '周期事件' DO CALL event_action()",
|
||||
"CREATE EVENT replica_event ON SCHEDULE EVERY 3 HOUR STARTS '2036-04-05 06:07:08' ON COMPLETION NOT PRESERVE DISABLE ON SLAVE DO CALL event_action()",
|
||||
"ALTER DATABASE COLLATE latin1_swedish_ci",
|
||||
"SET NAMES latin1 COLLATE latin1_bin",
|
||||
"CREATE EVENT `legacy_caf" + string([]byte{0xe9}) + "` ON SCHEDULE EVERY 1 DAY DISABLE COMMENT 'caf" + string([]byte{0xe9}) + ";;' DO SELECT 'caf" + string([]byte{0xe9}) + ";;'",
|
||||
"SET NAMES utf8mb4 COLLATE utf8mb4_bin",
|
||||
"ALTER DATABASE COLLATE utf8mb4_unicode_ci",
|
||||
} {
|
||||
if _, err := conn.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
definition := readMySQLEventDefinitions(t, db.DB)
|
||||
t.Run("metadata_rejection", func(t *testing.T) {
|
||||
tx, err := db.DB.BeginTx(t.Context(), nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
original, err := readMySQLStoredDefinition(t.Context(), tx, "EVENT", "legacy_café", "Create Event")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := "SELECT 'café;;'"
|
||||
for _, item := range []struct{ name, statement, expected string }{
|
||||
{"wrong_body", "CREATE EVENT x DO SELECT 'different'", "正文与目录"},
|
||||
{"missing_header", "CREATE EVENT x " + body, "声明头不完整"},
|
||||
{"lossy_header", "CREATE EVENT `中文` DO " + body, "无法无损"},
|
||||
{"invalid_utf8_header", "CREATE EVENT `ÿ` DO " + body, "无法无损"},
|
||||
} {
|
||||
t.Run(item.name, func(t *testing.T) {
|
||||
definition := make(map[string]string, len(original))
|
||||
for key, value := range original {
|
||||
definition[key] = value
|
||||
}
|
||||
definition["Create Event"] = item.statement
|
||||
err := normalizeMySQLEventMetadata(t.Context(), tx, "legacy_café", definition)
|
||||
if err == nil || !strings.Contains(err.Error(), item.expected) {
|
||||
t.Fatalf("无效编码元数据没有明确拒绝：%v", err)
|
||||
}
|
||||
if definition["Create Event"] != item.statement {
|
||||
t.Fatal("失败后留下部分转换定义")
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
output := filepath.Join(server.Root(), "native-events.sql")
|
||||
if err := backupMySQLDatabase(t.Context(), source, output); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, direction := range []string{"scert_to_scert", "scert_to_official"} {
|
||||
t.Run(direction, func(t *testing.T) {
|
||||
target := server.Database(t)
|
||||
targetDSN, err := mysql.ParseDSN(target.DSN)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
grantMySQLProgramAccess(t, server, parsed.User, targetDSN.DBName)
|
||||
targetDSN.User = parsed.User
|
||||
target.DSN = targetDSN.FormatDSN()
|
||||
verify, err := database.Open(t.Context(), target)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer verify.Close()
|
||||
if _, err := verify.DB.ExecContext(t.Context(), "CREATE EVENT `中文``事件` ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for attempt := 0; attempt < 2; attempt++ {
|
||||
if direction == "scert_to_official" {
|
||||
server.Import(t, target, output)
|
||||
} else if err := restoreMySQL(t.Context(), target, output); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if actual := readMySQLEventDefinitions(t, verify.DB); actual != definition {
|
||||
t.Fatalf("事件定义、计划、时区或状态不一致：\n源：%s\n恢复：%s", definition, actual)
|
||||
}
|
||||
checkMySQLInteropValue(t, target, "SELECT CAST(value AS CHAR) FROM event_values WHERE id = 1", "10")
|
||||
}
|
||||
// 只在数据、例程、事件全部恢复并核验后，在专用实例显式开启调度。
|
||||
if _, err := verify.DB.ExecContext(t.Context(), "ALTER EVENT `中文``事件` ON SCHEDULE AT CURRENT_TIMESTAMP + INTERVAL 2 SECOND ENABLE"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = ON"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer func() {
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "SET GLOBAL event_scheduler = OFF"); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}()
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
var value int
|
||||
if err := verify.DB.QueryRowContext(t.Context(), "SELECT value FROM event_values WHERE id = 1").Scan(&value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if value == 20 {
|
||||
break
|
||||
}
|
||||
time.Sleep(25 * time.Millisecond)
|
||||
}
|
||||
checkMySQLInteropValue(t, target, "SELECT CAST(value AS CHAR) FROM event_values WHERE id = 1", "20")
|
||||
checkMySQLInteropValue(t, source, "SELECT CAST(value AS CHAR) FROM event_values WHERE id = 1", "10")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func readMySQLEventDefinitions(t *testing.T, db *sql.DB) string {
|
||||
t.Helper()
|
||||
// 8.0 的 JSON_ARRAY 可能把相同周期文本显示成不透明 Base64。
|
||||
// 同时比较显式文本、原始字节、字符集及排序规则，不忽略实际周期值差异。
|
||||
rows, err := db.QueryContext(t.Context(), `SELECT JSON_ARRAY(EVENT_NAME, DEFINER, TIME_ZONE,
|
||||
EVENT_BODY, EVENT_DEFINITION, EVENT_TYPE, EXECUTE_AT, CONVERT(INTERVAL_VALUE USING utf8mb4),
|
||||
HEX(INTERVAL_VALUE), CHARSET(INTERVAL_VALUE), COLLATION(INTERVAL_VALUE), INTERVAL_FIELD,
|
||||
STARTS, ENDS, SQL_MODE, STATUS, ON_COMPLETION, EVENT_COMMENT,
|
||||
CHARACTER_SET_CLIENT, COLLATION_CONNECTION, DATABASE_COLLATION)
|
||||
FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE() ORDER BY EVENT_NAME`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var definitions []string
|
||||
for rows.Next() {
|
||||
var value string
|
||||
if err := rows.Scan(&value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
definitions = append(definitions, value)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return strings.Join(definitions, "\n")
|
||||
}
|
||||
|
||||
func TestMySQLRealEventVisibility(t *testing.T) {
|
||||
server := mysqltest.New(t)
|
||||
scenarios := []string{"visible", "hidden", "empty_without_permission", "global", "unrelated_schema", "escaped_schema", "shadowed_schema", "shadowed_exact"}
|
||||
if server.IsMySQL8() {
|
||||
scenarios = append(scenarios, "active_role", "nested_role", "inactive_role", "partial_revoke", "partial_revoke_with_schema")
|
||||
}
|
||||
for _, scenario := range scenarios {
|
||||
t.Run(scenario, func(t *testing.T) {
|
||||
cfg := server.Database(t)
|
||||
parsed, err := mysql.ParseDSN(cfg.DSN)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
account := mysqlStringLiteral(parsed.User) + "@'127.0.0.1'"
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "GRANT EVENT ON "+quoteMySQLIdentifier(parsed.DBName)+".* TO "+account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if scenario != "empty_without_permission" {
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "CREATE EVENT "+quoteMySQLIdentifier(parsed.DBName)+".hidden_event ON SCHEDULE EVERY 1 DAY DISABLE DO SELECT 1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if scenario != "visible" {
|
||||
if _, err := server.Admin().ExecContext(t.Context(), "REVOKE EVENT ON "+quoteMySQLIdentifier(parsed.DBName)+".* FROM "+account); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
schema := quoteMySQLIdentifier(parsed.DBName)
|
||||
var grants []string
|
||||
switch scenario {
|
||||
case "global":
|
||||
grants = append(grants, "GRANT EVENT ON *.* TO "+account)
|
||||
case "unrelated_schema":
|
||||
grants = append(grants, "GRANT EVENT ON `unrelated`.* TO "+account)
|
||||
case "escaped_schema":
|
||||
escaped := strings.ReplaceAll(parsed.DBName, "_", string(rune(92))+"_")
|
||||
grants = append(grants, "GRANT SELECT, EVENT ON "+quoteMySQLIdentifier(escaped)+".* TO "+account)
|
||||
case "shadowed_schema":
|
||||
grants = append(grants, "GRANT EVENT ON `scert%`.* TO "+account)
|
||||
case "shadowed_exact":
|
||||
escaped := strings.ReplaceAll(parsed.DBName, "_", string(rune(92))+"_")
|
||||
grants = append(grants, "GRANT EVENT ON "+schema+".* TO "+account, "GRANT SELECT ON "+quoteMySQLIdentifier(escaped)+".* TO "+account)
|
||||
case "active_role", "nested_role", "inactive_role":
|
||||
role := quoteMySQLIdentifier("event_"+scenario) + "@'%'"
|
||||
grants = append(grants, "CREATE ROLE "+role, "GRANT EVENT ON "+schema+".* TO "+role, "GRANT "+role+" TO "+account)
|
||||
if scenario == "active_role" {
|
||||
grants = append(grants, "SET DEFAULT ROLE "+role+" TO "+account)
|
||||
}
|
||||
if scenario == "nested_role" {
|
||||
parent := "'event_parent'@'%'"
|
||||
grants = append(grants, "REVOKE "+role+" FROM "+account, "CREATE ROLE "+parent, "GRANT "+role+" TO "+parent, "GRANT "+parent+" TO "+account, "SET DEFAULT ROLE "+parent+" TO "+account)
|
||||
}
|
||||
case "partial_revoke", "partial_revoke_with_schema":
|
||||
grants = append(grants, "SET GLOBAL partial_revokes = ON", "GRANT EVENT ON *.* TO "+account, "REVOKE EVENT ON "+schema+".* FROM "+account)
|
||||
if scenario == "partial_revoke_with_schema" {
|
||||
grants = append(grants, "GRANT EVENT ON "+schema+".* TO "+account)
|
||||
}
|
||||
}
|
||||
for _, grant := range grants {
|
||||
if _, err := server.Admin().ExecContext(t.Context(), grant); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
allowed := scenario == "visible" || scenario == "global" || scenario == "escaped_schema" || scenario == "active_role" || scenario == "nested_role" || scenario == "partial_revoke_with_schema"
|
||||
count := "0"
|
||||
if allowed {
|
||||
count = "1"
|
||||
}
|
||||
checkMySQLInteropValue(t, cfg, "SELECT CAST(COUNT(*) AS CHAR) FROM information_schema.EVENTS WHERE EVENT_SCHEMA = DATABASE()", count)
|
||||
output := filepath.Join(server.Root(), scenario+"-events.sql")
|
||||
err = backupMySQLDatabase(t.Context(), cfg, output)
|
||||
if allowed {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := os.ReadFile(output)
|
||||
if err != nil || !strings.Contains(string(data), "hidden_event") {
|
||||
t.Fatal("可见事件未导出")
|
||||
}
|
||||
return
|
||||
}
|
||||
if err == nil || !strings.Contains(err.Error(), "EVENT") {
|
||||
t.Fatalf("缺少事件可见权限必须拒绝，不能生成不完整备份：%v", err)
|
||||
}
|
||||
for _, path := range []string{output, output + ".tmp"} {
|
||||
if _, err := os.Stat(path); !os.IsNotExist(err) {
|
||||
t.Fatalf("权限失败后留下了备份文件：%s", path)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@ -85,6 +85,30 @@ func readMySQLEffectiveGrants(ctx context.Context, tx *sql.Tx, version string) (
|
||||
}
|
||||
|
||||
func requireMySQLTableTriggerGrants(ctx context.Context, tx *sql.Tx, name string, tables, grants []string, normalize func(string) string) error {
|
||||
return requireMySQLObjectGrants(ctx, tx, name, tables, grants, normalize, "TRIGGER", false)
|
||||
}
|
||||
|
||||
func requireMySQLEventVisibility(ctx context.Context, tx *sql.Tx) error {
|
||||
var name, version string
|
||||
var lowerCaseNames int
|
||||
if err := tx.QueryRowContext(ctx, "SELECT DATABASE(), VERSION(), @@lower_case_table_names").Scan(&name, &version, &lowerCaseNames); err != nil {
|
||||
return err
|
||||
}
|
||||
grants, err := readMySQLEffectiveGrants(ctx, tx, version)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
normalize := func(value string) string {
|
||||
if lowerCaseNames != 0 {
|
||||
return strings.ToLower(value)
|
||||
}
|
||||
return value
|
||||
}
|
||||
// EVENT 只有全局或库级授权，不能用某一张表的权限证明完整事件列表可见。
|
||||
return requireMySQLObjectGrants(ctx, tx, normalize(name), []string{name}, grants, normalize, "EVENT", true)
|
||||
}
|
||||
|
||||
func requireMySQLObjectGrants(ctx context.Context, tx *sql.Tx, name string, objects, grants []string, normalize func(string) string, privilege string, schemaOnly bool) error {
|
||||
// 5.7 返回空列表；8.0 部分撤销模式下，数据库授权中的百分号和下划线为字面量。
|
||||
partialRevokes := false
|
||||
rows, err := tx.QueryContext(ctx, "SHOW GLOBAL VARIABLES LIKE 'partial_revokes'")
|
||||
@ -115,9 +139,9 @@ func requireMySQLTableTriggerGrants(ctx context.Context, tx *sql.Tx, name string
|
||||
if len(match) == 0 {
|
||||
continue
|
||||
}
|
||||
hasTrigger := mysqlHasTriggerPrivilege(match[2])
|
||||
hasPrivilege := mysqlHasPrivilege(match[2], privilege)
|
||||
if match[3] == "*" && match[4] == "*" {
|
||||
global = global || (match[1] == "GRANT" && hasTrigger)
|
||||
global = global || (match[1] == "GRANT" && hasPrivilege)
|
||||
continue
|
||||
}
|
||||
databaseName := normalize(unquoteMySQLGrantIdentifier(match[3]))
|
||||
@ -131,16 +155,16 @@ func requireMySQLTableTriggerGrants(ctx context.Context, tx *sql.Tx, name string
|
||||
continue
|
||||
}
|
||||
if match[1] == "REVOKE" {
|
||||
revoked = revoked || hasTrigger
|
||||
revoked = revoked || hasPrivilege
|
||||
} else if match[4] == "*" {
|
||||
if partialRevokes || !mysqlGrantHasWildcard(databaseName) {
|
||||
exactSeen = true
|
||||
exactAllowed = exactAllowed || hasTrigger
|
||||
exactAllowed = exactAllowed || hasPrivilege
|
||||
} else {
|
||||
wildcardSeen = true
|
||||
wildcardAllowed = wildcardAllowed && hasTrigger
|
||||
wildcardAllowed = wildcardAllowed && hasPrivilege
|
||||
}
|
||||
} else if hasTrigger {
|
||||
} else if hasPrivilege && !schemaOnly {
|
||||
covered[normalize(unquoteMySQLGrantIdentifier(match[4]))] = true
|
||||
}
|
||||
}
|
||||
@ -150,9 +174,9 @@ func requireMySQLTableTriggerGrants(ctx context.Context, tx *sql.Tx, name string
|
||||
if !exactSeen {
|
||||
schema = wildcardSeen && wildcardAllowed
|
||||
}
|
||||
for _, table := range tables {
|
||||
if !((global && !revoked) || schema || covered[normalize(table)]) {
|
||||
return fmt.Errorf("MySQL 备份无法确认表 %s 的 TRIGGER 可见权限，不能把隐藏的触发器当成不存在", table)
|
||||
for _, object := range objects {
|
||||
if !((global && !revoked) || schema || covered[normalize(object)]) {
|
||||
return fmt.Errorf("MySQL 无法确认对象 %s 的 %s 可见权限，不能把隐藏的对象当成不存在", object, privilege)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
@ -173,8 +197,12 @@ func mysqlGrantHasWildcard(value string) bool {
|
||||
}
|
||||
|
||||
func mysqlHasTriggerPrivilege(privileges string) bool {
|
||||
return mysqlHasPrivilege(privileges, "TRIGGER")
|
||||
}
|
||||
|
||||
func mysqlHasPrivilege(privileges, required string) bool {
|
||||
for _, privilege := range strings.Split(privileges, ",") {
|
||||
if value := strings.TrimSpace(privilege); value == "TRIGGER" || value == "ALL PRIVILEGES" {
|
||||
if value := strings.TrimSpace(privilege); value == required || value == "ALL PRIVILEGES" {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
@ -212,7 +212,8 @@ func (f *Server) Database(t *testing.T) config.DatabaseConfig {
|
||||
for _, statement := range []string{
|
||||
"CREATE DATABASE " + name + " CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci",
|
||||
"CREATE USER '" + user + "'@'127.0.0.1' IDENTIFIED BY '" + f.password + "' REQUIRE SSL",
|
||||
"GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, INDEX, DROP, REFERENCES, TRIGGER ON " + name + ".* TO '" + user + "'@'127.0.0.1'",
|
||||
"GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER, INDEX, DROP, REFERENCES, TRIGGER, EVENT ON " + name + ".* TO '" + user + "'@'127.0.0.1'",
|
||||
"GRANT PROCESS ON *.* TO '" + user + "'@'127.0.0.1'",
|
||||
} {
|
||||
if _, err := f.admin.ExecContext(t.Context(), statement); err != nil {
|
||||
t.Fatal("创建隔离数据库及库级账户失败")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user