"""MySQL 审核规则引擎 — 63条 MySQL 专属审核规则。 本模块包含 MySQL 数据库的完整审核规则集，涵盖: - DDL 规则 (36条): 建表规范、索引规范、字段规范、字符集规范 - DML 规则 (9条): WHERE条件、批量限制、事务规范 - Online-DDL 规则 (3条): 在线变更工具配置 - 命名规则 (5条): 表名/字段名/索引名规范 - 性能规则 (5条): 全表扫描、索引使用 - 安全规则 (5条): 敏感字段、SQL注入风险 使用方式: 规则通过 @register_rule 装饰器自动注册到审核引擎， 由 SQLAuditEngineV2 在审核时自动调用。 依赖: - audit_engine_v2.py: register_rule, AuditRuleV2, AuditFindingV2, AuditReportV2 - Severity, RuleCategory 枚举 """ from __future__ import annotations import re from typing import Any, Dict, List, Optional from app.services.audit_engine_v2 import register_rule, AuditRuleV2, AuditFindingV2, AuditReportV2 from app.services.audit_engine_v2 import Severity, RuleCategory from app.services.audit_engine_v2 import ( _extract_table_name, _extract_column_definitions, _extract_index_definitions, _is_snake_case, ) # ────────────────────────────────────────────── # 辅助解析函数 # ────────────────────────────────────────────────── def _extract_create_table_comment(sql_upper: str) -> Optional[str]: """从 CREATE TABLE 语句中提取表注释。""" comment_match = re.search(r"COMMENT\s*=\s*['\"]([^'\"]*)['\"]", sql_upper, re.IGNORECASE) if comment_match: return comment_match.group(1).strip() # 也匹配 COMMENT 'xxx' 格式 comment_match = re.search(r"COMMENT\s+['\"]([^'\"]+)['\"]", sql_upper, re.IGNORECASE) if comment_match: return comment_match.group(1).strip() return None def _extract_charset_info(sql_upper: str) -> Dict[str, Optional[str]]: """从 CREATE TABLE 语句中提取字符集和排序规则信息。""" result: Dict[str, Optional[str]] = {"charset": None, "collation": None} cs_match = re.search(r"DEFAULT\s+CHARSET\s*=\s*(\w+)", sql_upper, re.IGNORECASE) if not cs_match: cs_match = re.search(r"CHARSET\s*=\s*(\w+)", sql_upper, re.IGNORECASE) if cs_match: result["charset"] = cs_match.group(1).lower() col_match = re.search(r"COLLATE\s*=\s*(\w+)", sql_upper, re.IGNORECASE) if col_match: result["collation"] = col_match.group(1).lower() return result def _extract_alter_operations(sql_upper: str) -> List[Dict[str, str]]: """从 ALTER TABLE 语句中提取操作列表。""" operations: List[Dict[str, str]] = [] # 匹配 ADD COLUMN, DROP COLUMN, MODIFY COLUMN, CHANGE COLUMN 等 patterns = [ (r"ADD\s+COLUMN\s+[`\"']?(\w+)", "ADD_COLUMN"), (r"ADD\s+(?:INDEX|KEY)\s+[`\"']?(\w*)", "ADD_INDEX"), (r"DROP\s+COLUMN\s+[`\"']?(\w+)", "DROP_COLUMN"), (r"MODIFY\s+COLUMN\s+[`\"']?(\w+)", "MODIFY_COLUMN"), (r"CHANGE\s+COLUMN\s+[`\"']?(\w+)", "CHANGE_COLUMN"), (r"DROP\s+(?:INDEX|KEY)\s+[`\"']?(\w+)", "DROP_INDEX"), (r"ADD\s+PRIMARY\s+KEY", "ADD_PK"), (r"DROP\s+PRIMARY\s+KEY", "DROP_PK"), ] for pattern, op_type in patterns: for match in re.finditer(pattern, sql_upper, re.IGNORECASE): name = match.group(1) if match.lastindex and match.group(1) else "" operations.append({"type": op_type, "name": name}) return operations def _has_where_clause(sql_upper: str) -> bool: """检查SQL语句是否包含WHERE子句。""" return bool(re.search(r"\bWHERE\b", sql_upper, re.IGNORECASE)) def _has_limit_clause(sql_upper: str) -> bool: """检查SQL语句是否包含LIMIT子句。""" return bool(re.search(r"\bLIMIT\s+\d+", sql_upper, re.IGNORECASE)) def _has_order_by(sql_upper: str) -> bool: """检查SQL语句是否包含ORDER BY子句。""" return bool(re.search(r"\bORDER\s+BY\b", sql_upper, re.IGNORECASE)) def _extract_insert_columns(sql_upper: str) -> List[str]: """从 INSERT 语句中提取插入的字段名列表。""" cols: List[str] = [] match = re.search(r"\bINSERT\b.+?\(([^)]+)\)", sql_upper, re.IGNORECASE | re.DOTALL) if match: inner = match.group(1) for col in inner.split(","): cols.append(col.strip().strip("`'\"")) return cols def _count_insert_rows(sql_clean: str) -> int: """统计 INSERT 语句插入的行数（通过 VALUES 子句数量）。""" values_count = len(re.findall(r"\bVALUES\s*\(", sql_clean, re.IGNORECASE)) if values_count == 0: # 尝试 INSERT ... SELECT 形式 if re.search(r"\bINSERT\b.*\bSELECT\b", sql_clean, re.IGNORECASE | re.DOTALL): return -1 # 表示无法确定的行数 return max(values_count, 1) def _is_ddl_statement(sql_upper: str) -> bool: """判断是否为DDL语句。""" return bool(re.search(r"\b(CREATE|ALTER|DROP|TRUNCATE)\b", sql_upper, re.IGNORECASE)) def _is_dml_statement(sql_upper: str) -> bool: """判断是否为DML语句。""" return bool(re.search(r"\b(INSERT|UPDATE|DELETE|REPLACE)\b", sql_upper, re.IGNORECASE)) def _extract_column_defaults(sql_upper: str) -> List[Dict[str, Any]]: """从 CREATE TABLE 中提取字段默认值信息。""" columns = _extract_column_definitions(sql_upper) result: List[Dict[str, Any]] = [] for col in columns: has_default = "DEFAULT" in col["full"].upper() is_nullable = "NOT NULL" not in col["full"].upper() is_timestamp = col["type"].upper() in ("TIMESTAMP", "DATETIME") result.append({ "name": col["name"], "type": col["type"], "has_default": has_default, "is_nullable": is_nullable, "is_timestamp": is_timestamp, "full": col["full"], }) return result def _extract_primary_key_info(sql_upper: str) -> Optional[Dict[str, Any]]: """从 CREATE TABLE 中提取主键信息。""" # 内联主键定义 pk_match = re.search(r"PRIMARY\s+KEY\s*\(([^)]+)\)", sql_upper, re.IGNORECASE) if pk_match: cols = [c.strip().strip("`'\"") for c in pk_match.group(1).split(",")] return {"type": "inline", "columns": cols} # 单列主键如 `id` INT PRIMARY KEY for col in _extract_column_definitions(sql_upper): if "PRIMARY KEY" in col["full"].upper(): return {"type": "column", "columns": [col["name"]], "column_def": col} return None def _count_alter_statements(sql_upper: str) -> int: """统计 ALTER TABLE 语句数量。""" return len(re.findall(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE)) def _has_auto_increment(sql_upper: str) -> bool: """检查是否包含 AUTO_INCREMENT。""" return bool(re.search(r"\bAUTO_INCREMENT\b", sql_upper, re.IGNORECASE)) def _extract_auto_increment_init(sql_upper: str) -> Optional[int]: """提取 AUTO_INCREMENT 初始值。""" match = re.search(r"AUTO_INCREMENT\s*=\s*(\d+)", sql_upper, re.IGNORECASE) if match: return int(match.group(1)) return None # ────────────────────────────────────────────── # ══════════════════════════════════════════════ # DDL 规则 (36条) — MYSQL-DDL-001 ~ MYSQL-DDL-036 # ══════════════════════════════════════════════ # ────────────────────────────────────────────── # ── DDL-001: 强制表必须拥有表注释 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-001", name="DDLCheckTableComment", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="CREATE TABLE 语句必须包含 COMMENT 表注释", config={"enabled": True}, message_template="表 '{table}' 缺少表注释，请添加 COMMENT='...'", )) def check_mysql_ddl_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制表必须拥有表注释。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return comment = _extract_create_table_comment(sql_upper) table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" if not comment: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name), )) # ── DDL-002: 强制表字段必须拥有列注释 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-002", name="DDLCheckColumnComment", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="CREATE TABLE 中每个字段定义必须包含 COMMENT 注释", config={"enabled": True}, message_template="表 '{table}' 的字段 '{column}' 缺少列注释，请添加 COMMENT '...'", )) def check_mysql_ddl_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制表字段必须拥有列注释。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" paren_match = re.search(r"\((.*)\)", sql_upper, re.DOTALL) if not paren_match: return inner = paren_match.group(1) parts = [] depth = 0 current = "" for char in inner: if char == "(": depth += 1 current += char elif char == ")": depth -= 1 current += char elif char == "," and depth == 0: parts.append(current.strip()) current = "" else: current += char if current.strip(): parts.append(current.strip()) for part in parts: # 跳过约束定义 if re.match(r"^(PRIMARY\s+KEY|FOREIGN\s+KEY|INDEX|KEY|UNIQUE|CONSTRAINT|CHECK)\b", part, re.IGNORECASE): continue col_match = re.match(r"[`\"']?(\w+)[`\"']?\s+(\w+)", part, re.IGNORECASE) if col_match: col_name = col_match.group(1) if "COMMENT" not in part.upper(): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=col_name), )) # ── DDL-003: 强制非timestamp类型字段必须为NOT NULL ── @register_rule(AuditRuleV2( code="MYSQL-DDL-003", name="DDLCheckColumnNullable", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="非 TIMESTAMP/DATETIME 类型字段必须定义为 NOT NULL", config={"enabled": True}, message_template="表 '{table}' 的字段 '{column}' (类型: {ctype}) 必须定义为 NOT NULL", )) def check_mysql_ddl_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制非timestamp类型字段必须为NOT NULL。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" paren_match = re.search(r"\((.*)\)", sql_upper, re.DOTALL) if not paren_match: return inner = paren_match.group(1) parts = [] depth = 0 current = "" for char in inner: if char == "(": depth += 1 current += char elif char == ")": depth -= 1 current += char elif char == "," and depth == 0: parts.append(current.strip()) current = "" else: current += char if current.strip(): parts.append(current.strip()) for part in parts: if re.match(r"^(PRIMARY\s+KEY|FOREIGN\s+KEY|INDEX|KEY|UNIQUE|CONSTRAINT|CHECK)\b", part, re.IGNORECASE): continue col_match = re.match(r"[`\"']?(\w+)[`\"']?\s+(\w+)", part, re.IGNORECASE) if col_match: col_name = col_match.group(1) col_type = col_match.group(2).upper() if col_type in ("TIMESTAMP", "DATETIME"): continue if "NOT NULL" not in part.upper(): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=col_name, ctype=col_type), )) # ── DDL-004: 强制非text,blob,json,timestamp类型字段必须拥有默认值 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-004", name="DDLCheckColumnDefault", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="非 TEXT/BLOB/JSON/TIMESTAMP/DATETIME 类型字段必须指定 DEFAULT 值", config={"enabled": True}, message_template="表 '{table}' 的字段 '{column}' (类型: {ctype}) 缺少 DEFAULT 值", )) def check_mysql_ddl_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制非text,blob,json,timestamp类型字段必须拥有默认值。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" excluded_types = {"TEXT", "BLOB", "JSON", "TIMESTAMP", "DATETIME", "TINYTEXT", "MEDIUMTEXT", "LONGTEXT", "TINYBLOB", "MEDIUMBLOB", "LONGBLOB"} paren_match = re.search(r"\((.*)\)", sql_upper, re.DOTALL) if not paren_match: return inner = paren_match.group(1) parts = [] depth = 0 current = "" for char in inner: if char == "(": depth += 1 current += char elif char == ")": depth -= 1 current += char elif char == "," and depth == 0: parts.append(current.strip()) current = "" else: current += char if current.strip(): parts.append(current.strip()) for part in parts: if re.match(r"^(PRIMARY\s+KEY|FOREIGN\s+KEY|INDEX|KEY|UNIQUE|CONSTRAINT|CHECK)\b", part, re.IGNORECASE): continue col_match = re.match(r"[`\"']?(\w+)[`\"']?\s+(\w+)", part, re.IGNORECASE) if col_match: col_name = col_match.group(1) col_type = col_match.group(2).upper() if col_type in excluded_types: continue if "DEFAULT" not in part.upper() and "NOT NULL" in part.upper(): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=col_name, ctype=col_type), )) # ── DDL-005: 强制float/double类型变更为decimal类型 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-005", name="DDLCheckFloatDouble", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="不建议使用 FLOAT/DOUBLE 类型，建议使用 DECIMAL 替代", config={"enabled": True}, message_template="表 '{table}' 的字段 '{column}' 使用了 {ctype} 类型，建议改为 DECIMAL", )) def check_mysql_ddl_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制float/double类型变更为decimal类型。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" columns = _extract_column_definitions(sql_upper) for col in columns: if col["type"].upper() in ("FLOAT", "DOUBLE"): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=col["name"], ctype=col["type"]), )) # ── DDL-006: 强制自增列初始值为1 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-006", name="DDLEnableAutoincrementInit", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="AUTO_INCREMENT 初始值必须为 1", config={"enabled": True, "init_value": 1}, message_template="表 '{table}' 的 AUTO_INCREMENT 初始值不为 {init}，建议设置为 AUTO_INCREMENT=1", )) def check_mysql_ddl_006(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制自增列初始值为1。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return if not _has_auto_increment(sql_upper): return init_val = _extract_auto_increment_init(sql_upper) expected = rule.config.get("init_value", 1) table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" if init_val is not None and init_val != expected: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, init=expected), )) # ── DDL-007/008: 强制必须拥有主键 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-007", name="DDLEnablePrimaryKey", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="CREATE TABLE 语句必须定义主键", config={"enabled": True}, message_template="表 '{table}' 缺少主键定义，必须添加 PRIMARY KEY", )) def check_mysql_ddl_007(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制必须拥有主键。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return pk_info = _extract_primary_key_info(sql_upper) table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" if pk_info is None: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name), )) # ── DDL-009: 强制主键名称为ID ── @register_rule(AuditRuleV2( code="MYSQL-DDL-009", name="DDLPrimaryKeyMust", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="主键列名必须为 'id'（不区分大小写）", config={"enabled": True}, message_template="表 '{table}' 的主键列名为 '{pk_name}'，建议命名为 'id'", )) def check_mysql_ddl_009(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制主键名称为ID。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return pk_info = _extract_primary_key_info(sql_upper) table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" if pk_info and len(pk_info["columns"]) == 1: pk_name = pk_info["columns"][0] if pk_name.lower() != "id": report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, pk_name=pk_name), )) # ── DDL-010: 强制主键为自增列 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-010", name="DDLEnableAutoIncrement", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="主键列必须使用 AUTO_INCREMENT", config={"enabled": True}, message_template="表 '{table}' 的主键列 '{pk_name}' 缺少 AUTO_INCREMENT", )) def check_mysql_ddl_010(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制主键为自增列。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" pk_info = _extract_primary_key_info(sql_upper) if pk_info is None: return if pk_info["type"] == "column": col_def = pk_info.get("column_def", {}) pk_name = pk_info["columns"][0] if "AUTO_INCREMENT" not in col_def.get("full", "").upper(): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, pk_name=pk_name), )) # ── DDL-011: 强制主键必须使用无符号标志unsigned ── @register_rule(AuditRuleV2( code="MYSQL-DDL-011", name="DDLEnableAutoincrementUnsigned", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="主键整数类型必须使用 UNSIGNED 属性", config={"enabled": True}, message_template="表 '{table}' 的主键列 '{pk_name}' 建议使用 UNSIGNED 属性", )) def check_mysql_ddl_011(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制主键必须使用无符号标志unsigned。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" pk_info = _extract_primary_key_info(sql_upper) if pk_info is None: return if pk_info["type"] == "column": col_def = pk_info.get("column_def", {}) pk_name = pk_info["columns"][0] col_type = col_def.get("type", "").upper() if col_type in ("INT", "BIGINT", "TINYINT", "SMALLINT", "MEDIUMINT"): if "UNSIGNED" not in col_def.get("full", "").upper(): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, pk_name=pk_name), )) # ── DDL-012: 基于主键回滚 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-012", name="PRIRollBack", category=RuleCategory.DDL, severity=Severity.INFO, db_type="mysql", description="建议基于主键进行数据变更回滚操作", config={"enabled": True}, message_template="建议在回滚操作中使用主键进行定位", )) def check_mysql_ddl_012(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """基于主键回滚建议。""" # 这是一个建议性规则，在 DELETE/UPDATE 语句中提示 if not _is_dml_statement(sql_upper): return pk_info = _extract_primary_key_info(sql_upper) if pk_info is None and _is_dml_statement(sql_upper): # 简单提示：DML操作建议基于主键 if re.search(r"\b(DELETE|UPDATE)\b", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── DDL-013: 开启索引名称规范(索引名必须以idx_为开头) ── @register_rule(AuditRuleV2( code="MYSQL-DDL-013", name="DDLIndexNameSpec", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="普通索引名称必须以 'idx_' 为前缀，唯一索引以 'uniq_' 为前缀", config={"enabled": True, "index_prefix": "idx_", "unique_prefix": "uniq_"}, message_template="表 '{table}' 的索引 '{idx_name}' 命名不规范，普通索引应以 '{prefix}' 开头", )) def check_mysql_ddl_013(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """开启索引名称规范。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" idx_prefix = rule.config.get("index_prefix", "idx_") uniq_prefix = rule.config.get("unique_prefix", "uniq_") indexes = _extract_index_definitions(sql_upper) for idx in indexes: if idx["type"] == "INDEX" and not idx["name"].lower().startswith(idx_prefix.lower()): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format( table=table_name, idx_name=idx["name"] or "(unnamed)", prefix=idx_prefix, ), )) # ── DDL-014: 开启MySQL关键词检查 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-014", name="CheckIdentifier", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="表名和字段名不能是MySQL保留关键字", config={"enabled": True}, message_template="表 '{table}' 中的标识符 '{identifier}' 是MySQL保留关键字，请修改", )) def check_mysql_ddl_014(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """开启MySQL关键词检查。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" # MySQL保留关键字列表（常用部分） mysql_keywords = { "SELECT", "INSERT", "UPDATE", "DELETE", "FROM", "WHERE", "ORDER", "GROUP", "BY", "HAVING", "LIMIT", "OFFSET", "JOIN", "LEFT", "RIGHT", "INNER", "OUTER", "ON", "AS", "AND", "OR", "NOT", "NULL", "TRUE", "FALSE", "CREATE", "DROP", "ALTER", "TABLE", "INDEX", "VIEW", "DATABASE", "SCHEMA", "PRIMARY", "KEY", "FOREIGN", "REFERENCES", "UNIQUE", "CHECK", "DEFAULT", "AUTO_INCREMENT", "UNSIGNED", "CHARSET", "COLLATE", "ENGINE", "COMMENT", "INT", "VARCHAR", "TEXT", "BLOB", "DATE", "TIME", "DATETIME", "TIMESTAMP", "FLOAT", "DOUBLE", "DECIMAL", "SHOW", "USE", "DESC", "EXPLAIN", "SET", "COMMIT", "ROLLBACK", } columns = _extract_column_definitions(sql_upper) if table_name.upper() in mysql_keywords: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, identifier=table_name), )) for col in columns: if col["name"].upper() in mysql_keywords: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, identifier=col["name"]), )) # ── DDL-015: 允许跨库表迁移 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-015", name="DDLEnableAcrossDBRename", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许跨库RENAME TABLE操作", config={"enabled": False}, message_template="禁止跨库RENAME TABLE操作: {detail}", )) def check_mysql_ddl_015(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许跨库表迁移（默认关闭）。""" if rule.config.get("enabled", False): return if not re.search(r"\bRENAME\s+TABLE\b", sql_upper, re.IGNORECASE): return # 检查是否包含数据库名前缀 (db.table) if re.search(r"`?\w+`?\.`?\w+`?\s+TO\s+`?\w+`?\.`?\w+`?", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(detail=sql_clean[:80]), )) # ── DDL-016: 允许删除表 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-016", name="DDLEnableDropTable", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许 DROP TABLE 操作", config={"enabled": False}, message_template="禁止 DROP TABLE 操作，表 '{table}' 将被删除", )) def check_mysql_ddl_016(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许删除表（默认关闭）。""" if rule.config.get("enabled", False): return match = re.search(r"\bDROP\s+TABLE\s+(?:IF\s+EXISTS\s+)?[`\"']?(\w+)[`\"']?", sql_upper, re.IGNORECASE) if match: table_name = match.group(1) report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name), )) # ── DDL-017: 允许删除库 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-017", name="DDLEnableDropDatabase", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许 DROP DATABASE 操作", config={"enabled": False}, message_template="禁止 DROP DATABASE 操作，数据库 '{db}' 将被删除", )) def check_mysql_ddl_017(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许删除库（默认关闭）。""" if rule.config.get("enabled", False): return match = re.search(r"\bDROP\s+DATABASE\s+(?:IF\s+EXISTS\s+)?[`\"']?(\w+)[`\"']?", sql_upper, re.IGNORECASE) if match: db_name = match.group(1) report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(db=db_name), )) # ── DDL-018: 允许主键类型为非int/bigint ── @register_rule(AuditRuleV2( code="MYSQL-DDL-018", name="DDLAllowPRINotInt", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="默认要求主键类型必须为 INT 或 BIGINT", config={"enabled": False}, message_template="表 '{table}' 的主键类型为 '{pk_type}'，建议使用 INT 或 BIGINT", )) def check_mysql_ddl_018(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许主键类型为非int/bigint（默认不允许）。""" if rule.config.get("enabled", False): return if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" pk_info = _extract_primary_key_info(sql_upper) if pk_info is None: return if pk_info["type"] == "column": col_def = pk_info.get("column_def", {}) pk_type = col_def.get("type", "").upper() if pk_type not in ("INT", "BIGINT", "TINYINT", "SMALLINT", "MEDIUMINT"): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, pk_type=pk_type), )) # ── DDL-019: 允许索引名为空 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-019", name="DDLEnableNullIndexName", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="默认要求索引必须命名", config={"enabled": False}, message_template="表 '{table}' 存在未命名的索引，建议为所有索引指定名称", )) def check_mysql_ddl_019(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许索引名为空（默认不允许）。""" if rule.config.get("enabled", False): return if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" indexes = _extract_index_definitions(sql_upper) for idx in indexes: if not idx["name"] and idx["type"] not in ("PRIMARY", "FOREIGN"): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name), )) # ── DDL-020: 允许单个工单提交多条DDL语句 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-020", name="DDLMultiToCommit", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许单个工单提交多条DDL语句", config={"enabled": False, "max_statements": 1}, message_template="单个工单包含 {count} 条DDL语句，建议每条工单只提交1条DDL", )) def check_mysql_ddl_020(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许单个工单提交多条DDL语句（默认不允许）。""" if rule.config.get("enabled", False): return ddl_count = 0 # 按分号分割统计DDL语句数 statements = [s.strip() for s in sql_clean.split(";") if s.strip()] for stmt in statements: if re.search(r"\b(CREATE|ALTER|DROP|TRUNCATE)\b", stmt, re.IGNORECASE): ddl_count += 1 max_stmt = rule.config.get("max_statements", 1) if ddl_count > max_stmt: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(count=ddl_count), )) # ── DDL-021: DDL单个工单允许多alter语句提交 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-021", name="DDLAllowMultiAlter", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许单个工单包含多条 ALTER TABLE 语句", config={"enabled": False, "max_alter": 1}, message_template="单个工单包含 {count} 条 ALTER TABLE 语句，建议拆分为多个工单", )) def check_mysql_ddl_021(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """DDL单个工单允许多alter语句提交（默认不允许）。""" if rule.config.get("enabled", False): return alter_count = _count_alter_statements(sql_upper) max_alter = rule.config.get("max_alter", 1) if alter_count > max_alter: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(count=alter_count), )) # ── DDL-022: 允许字段进行类型转换 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-022", name="DDLAllowColumnType", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="默认不允许 ALTER TABLE MODIFY/CHANGE 进行字段类型转换", config={"enabled": False}, message_template="ALTER TABLE 包含字段类型变更操作，请确认变更的必要性和兼容性", )) def check_mysql_ddl_022(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许字段进行类型转换（默认不允许）。""" if rule.config.get("enabled", False): return if not re.search(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE): return operations = _extract_alter_operations(sql_upper) for op in operations: if op["type"] in ("MODIFY_COLUMN", "CHANGE_COLUMN"): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── DDL-023: 允许使用after/first ── @register_rule(AuditRuleV2( code="MYSQL-DDL-023", name="DDLAllowChangeColumnPosition", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="默认不建议使用 AFTER/FIRST 调整字段位置（引起全表重建）", config={"enabled": False}, message_template="ALTER TABLE 使用了 AFTER/FIRST 调整字段位置，可能引起全表重建", )) def check_mysql_ddl_023(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许使用after/first（默认不建议）。""" if rule.config.get("enabled", False): return if not re.search(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE): return if re.search(r"\b(AFTER|FIRST)\b", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── DDL-024: 允许创建视图 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-024", name="AllowCreateView", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许创建视图", config={"enabled": False}, message_template="禁止创建视图: '{view_name}'", )) def check_mysql_ddl_024(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许创建视图（默认不允许）。""" if rule.config.get("enabled", False): return match = re.search(r"\bCREATE\s+(?:OR\s+REPLACE\s+)?VIEW\s+[`\"']?(\w+)[`\"']?", sql_upper, re.IGNORECASE) if match: view_name = match.group(1) report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(view_name=view_name), )) # ── DDL-025: 允许创建外键 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-025", name="DDLEnableForeignKey", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="默认不允许创建外键约束", config={"enabled": False}, message_template="禁止创建外键约束: '{fk_name}'，建议在应用层实现外键逻辑", )) def check_mysql_ddl_025(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许创建外键（默认不允许）。""" if rule.config.get("enabled", False): return if re.search(r"\bFOREIGN\s+KEY\b", sql_upper, re.IGNORECASE): fk_match = re.search(r"FOREIGN\s+KEY\s+[`\"']?(\w*)[`\"']?", sql_upper, re.IGNORECASE) fk_name = fk_match.group(1) if fk_match and fk_match.group(1) else "(unnamed)" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(fk_name=fk_name), )) # ── DDL-026: 允许创建分区表 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-026", name="AllowCreatePartition", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="默认不建议创建分区表", config={"enabled": False}, message_template="创建分区表需要谨慎评估，请确认分区策略的合理性", )) def check_mysql_ddl_026(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许创建分区表（默认不建议）。""" if rule.config.get("enabled", False): return if re.search(r"\bPARTITION\s+BY\b", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── DDL-027: 允许添加bit,enum,set类型字段 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-027", name="AllowSpecialType", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="默认不建议使用 BIT/ENUM/SET 类型", config={"enabled": False}, message_template="表 '{table}' 使用了特殊类型 '{sp_type}'，建议评估替代方案", )) def check_mysql_ddl_027(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许添加bit,enum,set类型字段（默认不建议）。""" if rule.config.get("enabled", False): return if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" columns = _extract_column_definitions(sql_upper) special_types = {"BIT", "ENUM", "SET"} for col in columns: if col["type"].upper() in special_types: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, sp_type=col["type"]), )) # ── DDL-028: 指定表名前缀 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-028", name="DDLTablePrefix", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="表名必须以指定前缀开头", config={"enabled": True, "prefix": "t_"}, message_template="表名 '{table}' 不符合前缀规范，必须以 '{prefix}' 开头", )) def check_mysql_ddl_028(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """指定表名前缀。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return prefix = rule.config.get("prefix", "t_") table_name = _extract_table_name(sql_upper, sql_clean) if table_name and not table_name.lower().startswith(prefix.lower()): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, prefix=prefix), )) # ── DDL-029: 允许的Collate范围 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-029", name="SupportCollation", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="表字符集排序规则必须在允许的范围内", config={"enabled": True, "allowed_collations": ["utf8mb4_general_ci", "utf8mb4_unicode_ci"]}, message_template="表 '{table}' 使用了不允许的排序规则 '{collation}'，允许的规则: {allowed}", )) def check_mysql_ddl_029(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许的Collate范围。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" charset_info = _extract_charset_info(sql_upper) collation = charset_info.get("collation") allowed = rule.config.get("allowed_collations", []) if collation and allowed and collation.lower() not in [a.lower() for a in allowed]: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format( table=table_name, collation=collation, allowed=", ".join(allowed), ), )) # ── DDL-030: 允许的Charset范围 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-030", name="SupportCharset", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="表字符集必须在允许的范围内", config={"enabled": True, "allowed_charsets": ["utf8mb4"]}, message_template="表 '{table}' 使用了不允许的字符集 '{charset}'，允许的字符集: {allowed}", )) def check_mysql_ddl_030(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许的Charset范围。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" charset_info = _extract_charset_info(sql_upper) charset = charset_info.get("charset") allowed = rule.config.get("allowed_charsets", ["utf8mb4"]) if charset and allowed and charset.lower() not in [a.lower() for a in allowed]: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format( table=table_name, charset=charset, allowed=", ".join(allowed), ), )) # ── DDL-031: 建表必须拥有的字段 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-031", name="MustHaveColumns", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="建表时必须包含指定的字段", config={"enabled": True, "required_columns": ["id", "created_at", "updated_at"]}, message_template="表 '{table}' 缺少必须字段 '{column}'", )) def check_mysql_ddl_031(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """建表必须拥有的字段。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" required = rule.config.get("required_columns", []) columns = _extract_column_definitions(sql_upper) column_names = {c["name"].lower() for c in columns} for req_col in required: if req_col.lower() not in column_names: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=req_col), )) # ── DDL-032: 单个索引指定字段上限 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-032", name="DDLMaxKeyParts", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="单个索引最多允许包含的字段数", config={"enabled": True, "max_parts": 8}, message_template="表 '{table}' 的索引 '{idx_name}' 包含 {count} 个字段，超过上限 {max_parts}", )) def check_mysql_ddl_032(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """单个索引指定字段上限。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" max_parts = rule.config.get("max_parts", 8) indexes = _extract_index_definitions(sql_upper) for idx in indexes: col_count = len(idx.get("columns", [])) if col_count > max_parts: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format( table=table_name, idx_name=idx["name"] or "(unnamed)", count=col_count, max_parts=max_parts, ), )) # ── DDL-033: 单个表最多允许几个索引 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-033", name="DDLMaxKey", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="单个表最多允许创建的索引数量", config={"enabled": True, "max_index": 12}, message_template="表 '{table}' 包含 {count} 个索引，超过上限 {max_idx}", )) def check_mysql_ddl_033(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """单个表最多允许几个索引。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" max_idx = rule.config.get("max_index", 12) indexes = _extract_index_definitions(sql_upper) idx_count = len([i for i in indexes if i["type"] != "FOREIGN"]) if idx_count > max_idx: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, count=idx_count, max_idx=max_idx), )) # ── DDL-034: DDL最大影响行数 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-034", name="MaxDDLAffectRows", category=RuleCategory.DDL, severity=Severity.ERROR, db_type="mysql", description="DDL操作影响的最大行数限制（0表示不限制）", config={"enabled": False, "max_rows": 0}, message_template="DDL操作可能影响 {rows} 行，超过上限 {max_rows}", )) def check_mysql_ddl_034(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """DDL最大影响行数（此规则作为配置项存在，实际行数需运行时检查）。""" max_rows = rule.config.get("max_rows", 0) if max_rows == 0: return # 此规则的实际检查需要数据库连接获取表行数 # 这里仅做语句级别的标记提示 if re.search(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE): table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=f"ALTER TABLE '{table_name}' 需要确认影响行数不超过 {max_rows}", )) # ── DDL-035: char字段最大长度 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-035", name="DDLMaxCharLength", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="CHAR 类型字段的最大长度限制", config={"enabled": True, "max_length": 10}, message_template="表 '{table}' 的字段 '{column}' CHAR({length}) 超过最大长度限制 {max_len}", )) def check_mysql_ddl_035(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """char字段最大长度。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" max_len = rule.config.get("max_length", 10) paren_match = re.search(r"\((.*)\)", sql_upper, re.DOTALL) if not paren_match: return inner = paren_match.group(1) parts = [] depth = 0 current = "" for char in inner: if char == "(": depth += 1 current += char elif char == ")": depth -= 1 current += char elif char == "," and depth == 0: parts.append(current.strip()) current = "" else: current += char if current.strip(): parts.append(current.strip()) for part in parts: if re.match(r"^(PRIMARY\s+KEY|FOREIGN\s+KEY|INDEX|KEY|UNIQUE|CONSTRAINT|CHECK)\b", part, re.IGNORECASE): continue char_match = re.search(r"[`\"']?(\w+)[`\"']?\s+CHAR\s*\((\d+)\)", part, re.IGNORECASE) if char_match: col_name = char_match.group(1) length = int(char_match.group(2)) if length > max_len: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format( table=table_name, column=col_name, length=length, max_len=max_len, ), )) # ── DDL-036: 表名最大长度限制 ── @register_rule(AuditRuleV2( code="MYSQL-DDL-036", name="MaxTableNameLen", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="表名最大长度限制", config={"enabled": True, "max_length": 45}, message_template="表名 '{table}' 长度为 {length}，超过最大长度限制 {max_len}", )) def check_mysql_ddl_036(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """表名最大长度限制。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) if not table_name: return max_len = rule.config.get("max_length", 45) if len(table_name) > max_len: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, length=len(table_name), max_len=max_len), )) # ────────────────────────────────────────────── # ══════════════════════════════════════════════ # DML 规则 (9条) — MYSQL-DML-001 ~ MYSQL-DML-009 # ══════════════════════════════════════════════ # ────────────────────────────────────────────── # ── DML-001: Insert最大插入行数上限 ── @register_rule(AuditRuleV2( code="MYSQL-DML-001", name="DMLMaxInsertRows", category=RuleCategory.DML, severity=Severity.ERROR, db_type="mysql", description="INSERT 语句单次插入的最大行数限制", config={"enabled": True, "max_rows": 500}, message_template="INSERT 语句插入 {count} 行，超过最大限制 {max_rows} 行，请分批插入", )) def check_mysql_dml_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """Insert最大插入行数上限。""" if not re.search(r"\bINSERT\b", sql_upper, re.IGNORECASE): return max_rows = rule.config.get("max_rows", 500) row_count = _count_insert_rows(sql_clean) if row_count == -1: # INSERT ... SELECT 无法确定行数 return if row_count > max_rows: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(count=row_count, max_rows=max_rows), )) # ── DML-002: 允许update/insert语句使用limit关键字 ── @register_rule(AuditRuleV2( code="MYSQL-DML-002", name="DMLAllowLimitSTMT", category=RuleCategory.DML, severity=Severity.ERROR, db_type="mysql", description="默认不允许 UPDATE/DELETE 语句使用 LIMIT 关键字", config={"enabled": False}, message_template="{stmt_type} 语句禁止使用 LIMIT 关键字", )) def check_mysql_dml_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许update/insert语句使用limit关键字（默认不允许）。""" if rule.config.get("enabled", False): return if not re.search(r"\b(UPDATE|DELETE)\b", sql_upper, re.IGNORECASE): return if _has_limit_clause(sql_upper): stmt_type = "UPDATE" if re.search(r"\bUPDATE\b", sql_upper, re.IGNORECASE) else "DELETE" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(stmt_type=stmt_type), )) # ── DML-003: 允许insert语句插入Null值 ── @register_rule(AuditRuleV2( code="MYSQL-DML-003", name="DMLAllowInsertNull", category=RuleCategory.DML, severity=Severity.WARNING, db_type="mysql", description="默认不建议 INSERT 语句显式插入 NULL 值", config={"enabled": False}, message_template="INSERT 语句中包含 NULL 值插入，建议确认必要性", )) def check_mysql_dml_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """允许insert语句插入Null值（默认不建议）。""" if rule.config.get("enabled", False): return if not re.search(r"\bINSERT\b", sql_upper, re.IGNORECASE): return # 匹配 VALUES (...NULL...) 中的 NULL values_section = re.search(r"\bVALUES\s+\((.*?)\)", sql_upper, re.IGNORECASE | re.DOTALL) if values_section and "NULL" in values_section.group(1).upper(): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── DML-004: 不允许隐式转换 ── @register_rule(AuditRuleV2( code="MYSQL-DML-004", name="DDLImplicitTypeConversion", category=RuleCategory.DML, severity=Severity.WARNING, db_type="mysql", description="WHERE 条件中避免隐式类型转换", config={"enabled": True}, message_template="WHERE 条件中字段 '{column}' 可能存在隐式类型转换，请确保类型一致", )) def check_mysql_dml_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """不允许隐式转换（检测WHERE中可能的类型不匹配）。""" if not _has_where_clause(sql_upper): return # 简单检测：数字字段与字符串值比较，或函数字段使用 where_match = re.search(r"\bWHERE\b(.+?)(?:\s+ORDER\s+BY|\s+GROUP\s+BY|\s+LIMIT|\s+HAVING|$)", sql_upper, re.IGNORECASE | re.DOTALL) if not where_match: return where_clause = where_match.group(1) # 检测函数包裹字段（常见隐式转换场景） func_patterns = [ r"\b(CONVERT|CAST)\s*\(", r"\bDATE\s*\(", r"\bYEAR\s*\(", r"\bMONTH\s*\(", ] for pattern in func_patterns: if re.search(pattern, where_clause, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(column="(函数包装字段)"), )) # ── DML-005: 检查Insert语句中插入的字段名是否存在 ── @register_rule(AuditRuleV2( code="MYSQL-DML-005", name="DMLInsertColumns", category=RuleCategory.DML, severity=Severity.ERROR, db_type="mysql", description="INSERT 语句必须显式指定字段名", config={"enabled": True}, message_template="INSERT 语句必须显式指定插入的字段名，禁止使用 INSERT INTO table VALUES (...)", )) def check_mysql_dml_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """检查Insert语句中插入的字段名是否存在。""" if not re.search(r"\bINSERT\b", sql_upper, re.IGNORECASE): return insert_cols = _extract_insert_columns(sql_upper) if not insert_cols: # 检查是否是 INSERT INTO table VALUES (没有指定字段) if re.search(r"\bINSERT\b.+?\bVALUES\b", sql_upper, re.IGNORECASE) and \ not re.search(r"\bINSERT\b.+?\([^)]+\)\s*\bVALUES\b", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── DML-006: 强制DML语句必须拥有where条件 ── @register_rule(AuditRuleV2( code="MYSQL-DML-006", name="DMLWhere", category=RuleCategory.DML, severity=Severity.ERROR, db_type="mysql", description="UPDATE/DELETE 语句必须包含 WHERE 条件", config={"enabled": True}, message_template="{stmt_type} 语句必须包含 WHERE 条件，禁止全表操作", )) def check_mysql_dml_006(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """强制DML语句必须拥有where条件。""" if not re.search(r"\b(UPDATE|DELETE)\b", sql_upper, re.IGNORECASE): return if not _has_where_clause(sql_upper): stmt_type = "UPDATE" if re.search(r"\bUPDATE\b", sql_upper, re.IGNORECASE) else "DELETE" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(stmt_type=stmt_type), )) # ── DML-007: 禁止DML语句使用Order by子句 ── @register_rule(AuditRuleV2( code="MYSQL-DML-007", name="DMLOrder", category=RuleCategory.DML, severity=Severity.WARNING, db_type="mysql", description="UPDATE/DELETE 语句禁止使用 ORDER BY 子句", config={"enabled": True}, message_template="{stmt_type} 语句禁止使用 ORDER BY 子句", )) def check_mysql_dml_007(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """禁止DML语句使用Order by子句。""" if not re.search(r"\b(UPDATE|DELETE)\b", sql_upper, re.IGNORECASE): return if _has_order_by(sql_upper): stmt_type = "UPDATE" if re.search(r"\bUPDATE\b", sql_upper, re.IGNORECASE) else "DELETE" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(stmt_type=stmt_type), )) # ── DML-008: 禁止DML语句使用Select子句 ── @register_rule(AuditRuleV2( code="MYSQL-DML-008", name="DMLSelect", category=RuleCategory.DML, severity=Severity.ERROR, db_type="mysql", description="UPDATE/DELETE 语句禁止使用子查询（SELECT 子句）", config={"enabled": True}, message_template="{stmt_type} 语句禁止使用 SELECT 子查询", )) def check_mysql_dml_008(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """禁止DML语句使用Select子句。""" if not re.search(r"\b(UPDATE|DELETE)\b", sql_upper, re.IGNORECASE): return if re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE): stmt_type = "UPDATE" if re.search(r"\bUPDATE\b", sql_upper, re.IGNORECASE) else "DELETE" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(stmt_type=stmt_type), )) # ── DML-009: DML最大影响行数 ── @register_rule(AuditRuleV2( code="MYSQL-DML-009", name="MaxAffectRows", category=RuleCategory.DML, severity=Severity.ERROR, db_type="mysql", description="DML 操作影响的最大行数限制（0表示不限制）", config={"enabled": False, "max_rows": 0}, message_template="DML操作可能影响 {rows} 行，超过上限 {max_rows}", )) def check_mysql_dml_009(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """DML最大影响行数（此规则作为配置项存在，实际行数需运行时检查）。""" max_rows = rule.config.get("max_rows", 0) if max_rows == 0: return # 仅作标记提示，实际行数需要在执行时检查 if _is_dml_statement(sql_upper): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=f"DML语句需要确认影响行数不超过 {max_rows}", )) # ────────────────────────────────────────────── # ══════════════════════════════════════════════ # Online-DDL 规则 (3条) — MYSQL-OSC-001 ~ MYSQL-OSC-003 # ══════════════════════════════════════════════ # ────────────────────────────────────────────── # ── OSC-001: 开启在线表变更工具 ── @register_rule(AuditRuleV2( code="MYSQL-OSC-001", name="IsOSC", category=RuleCategory.DDL, severity=Severity.INFO, db_type="mysql", description="建议使用在线表变更工具（pt-osc/gh-ost）执行大表DDL", config={"enabled": True, "tool": "gh-ost"}, message_template="表 '{table}' 的DDL操作建议使用 {tool} 在线变更工具执行", )) def check_mysql_osc_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """开启在线表变更工具建议。""" if not re.search(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" tool = rule.config.get("tool", "gh-ost") report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, tool=tool), )) # ── OSC-002: 表体积大于该值后触发在线表变更同步 ── @register_rule(AuditRuleV2( code="MYSQL-OSC-002", name="OscSize", category=RuleCategory.DDL, severity=Severity.WARNING, db_type="mysql", description="表体积超过阈值（MB）后必须使用在线变更工具", config={"enabled": True, "size_mb": 100}, message_template="表 '{table}' 预估体积 {size}MB，超过阈值 {threshold}MB，必须使用在线变更工具", )) def check_mysql_osc_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """表体积大于阈值触发在线表变更建议。""" if not re.search(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" threshold = rule.config.get("size_mb", 100) # 此规则实际需要在数据库中查询表大小，这里仅作提示 report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=f"ALTER TABLE '{table_name}' 请确认表体积是否超过 {threshold}MB，大表请使用在线变更工具", )) # ── OSC-003: 同步工具参数配置 ── @register_rule(AuditRuleV2( code="MYSQL-OSC-003", name="OSCExpr", category=RuleCategory.DDL, severity=Severity.INFO, db_type="mysql", description="在线变更工具参数配置检查", config={"enabled": True, "chunk_size": 1000, "max_load": "Threads_running=80"}, message_template="请确认在线变更工具参数: chunk_size={chunk_size}, max_load={max_load}", )) def check_mysql_osc_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """同步工具参数配置检查。""" if not re.search(r"\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE): return chunk_size = rule.config.get("chunk_size", 1000) max_load = rule.config.get("max_load", "Threads_running=80") report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(chunk_size=chunk_size, max_load=max_load), )) # ────────────────────────────────────────────── # ══════════════════════════════════════════════ # 命名规则 (5条) — MYSQL-NAM-001 ~ MYSQL-NAM-005 # ══════════════════════════════════════════════ # ────────────────────────────────────────────── # ── NAM-001: 表名小写下划线规范 ── @register_rule(AuditRuleV2( code="MYSQL-NAM-001", name="TableNamingCase", category=RuleCategory.NAMING, severity=Severity.ERROR, db_type="mysql", description="表名必须使用小写下划线命名法（snake_case）", config={"enabled": True}, message_template="表名 '{table}' 不符合命名规范，必须使用小写下划线命名（snake_case）", )) def check_mysql_nam_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """表名小写下划线规范。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) if table_name and not _is_snake_case(table_name): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name), )) # ── NAM-002: 字段名小写下划线规范 ── @register_rule(AuditRuleV2( code="MYSQL-NAM-002", name="ColumnNamingCase", category=RuleCategory.NAMING, severity=Severity.ERROR, db_type="mysql", description="字段名必须使用小写下划线命名法（snake_case）", config={"enabled": True}, message_template="表 '{table}' 的字段 '{column}' 不符合命名规范，必须使用小写下划线命名（snake_case）", )) def check_mysql_nam_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """字段名小写下划线规范。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" columns = _extract_column_definitions(sql_upper) for col in columns: if not _is_snake_case(col["name"]): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=col["name"]), )) # ── NAM-003: 索引名小写下划线规范 ── @register_rule(AuditRuleV2( code="MYSQL-NAM-003", name="IndexNamingCase", category=RuleCategory.NAMING, severity=Severity.WARNING, db_type="mysql", description="索引名必须使用小写下划线命名法", config={"enabled": True}, message_template="表 '{table}' 的索引 '{idx_name}' 不符合命名规范，必须使用小写下划线命名", )) def check_mysql_nam_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """索引名小写下划线规范。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" indexes = _extract_index_definitions(sql_upper) for idx in indexes: if idx["name"] and not _is_snake_case(idx["name"]): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, idx_name=idx["name"]), )) # ── NAM-004: 表名数字开头检查 ── @register_rule(AuditRuleV2( code="MYSQL-NAM-004", name="TableNamingNoDigitPrefix", category=RuleCategory.NAMING, severity=Severity.ERROR, db_type="mysql", description="表名不能以数字开头", config={"enabled": True}, message_template="表名 '{table}' 不能以数字开头", )) def check_mysql_nam_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """表名数字开头检查。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) if table_name and re.match(r"^\d", table_name): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name), )) # ── NAM-005: 字段名长度限制 ── @register_rule(AuditRuleV2( code="MYSQL-NAM-005", name="MaxColumnNameLength", category=RuleCategory.NAMING, severity=Severity.WARNING, db_type="mysql", description="字段名最大长度限制", config={"enabled": True, "max_length": 30}, message_template="表 '{table}' 的字段 '{column}' 名长度为 {length}，超过限制 {max_len}", )) def check_mysql_nam_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """字段名长度限制。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" max_len = rule.config.get("max_length", 30) columns = _extract_column_definitions(sql_upper) for col in columns: if len(col["name"]) > max_len: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format( table=table_name, column=col["name"], length=len(col["name"]), max_len=max_len, ), )) # ────────────────────────────────────────────── # ══════════════════════════════════════════════ # 性能规则 (5条) — MYSQL-PERF-001 ~ MYSQL-PERF-005 # ══════════════════════════════════════════════ # ────────────────────────────────────────────── # ── PERF-001: 禁止SELECT * ── @register_rule(AuditRuleV2( code="MYSQL-PERF-001", name="NoSelectAll", category=RuleCategory.PERFORMANCE, severity=Severity.ERROR, db_type="mysql", description="禁止 SELECT * 查询，必须显式指定字段", config={"enabled": True}, message_template="禁止使用 SELECT *，请显式指定查询字段", )) def check_mysql_perf_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """禁止SELECT *。""" if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE): return if re.search(r"\bSELECT\s+\*\b", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── PERF-002: 缺少LIMIT的大查询检查 ── @register_rule(AuditRuleV2( code="MYSQL-PERF-002", name="SelectLimitRequired", category=RuleCategory.PERFORMANCE, severity=Severity.WARNING, db_type="mysql", description="SELECT 查询建议添加 LIMIT 子句", config={"enabled": True, "max_limit": 1000}, message_template="SELECT 查询缺少 LIMIT 子句，建议添加 LIMIT {max_limit}", )) def check_mysql_perf_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """缺少LIMIT的大查询检查。""" if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE): return if _is_dml_statement(sql_upper): return if not _has_limit_clause(sql_upper): max_limit = rule.config.get("max_limit", 1000) report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(max_limit=max_limit), )) # ── PERF-003: 全表扫描风险 — WHERE条件缺少索引字段 ── @register_rule(AuditRuleV2( code="MYSQL-PERF-003", name="FullTableScanRisk", category=RuleCategory.PERFORMANCE, severity=Severity.WARNING, db_type="mysql", description="UPDATE/DELETE 的 WHERE 条件应尽量使用索引字段", config={"enabled": True}, message_template="{stmt_type} 语句 WHERE 条件 '{where_clause}' 可能无法使用索引，存在全表扫描风险", )) def check_mysql_perf_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """全表扫描风险检查。""" if not re.search(r"\b(UPDATE|DELETE|SELECT)\b", sql_upper, re.IGNORECASE): return if not _has_where_clause(sql_upper): return # 检测 WHERE 中使用函数或 LIKE '%xxx%' 等无法使用索引的模式 where_match = re.search(r"\bWHERE\b(.+?)(?:\s+ORDER\s+BY|\s+GROUP\s+BY|\s+LIMIT|\s+HAVING|$)", sql_upper, re.IGNORECASE | re.DOTALL) if not where_match: return where_clause = where_match.group(1).strip() # 检测 LIKE '%xxx' 或 LIKE '%xxx%' if re.search(r"\bLIKE\s+['\"]%", where_clause, re.IGNORECASE): stmt_type_match = re.search(r"\b(UPDATE|DELETE|SELECT)\b", sql_upper, re.IGNORECASE) stmt_type = stmt_type_match.group(1).upper() if stmt_type_match else "SQL" report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(stmt_type=stmt_type, where_clause=where_clause[:50]), )) # ── PERF-004: 大IN子句检查 ── @register_rule(AuditRuleV2( code="MYSQL-PERF-004", name="LargeInClause", category=RuleCategory.PERFORMANCE, severity=Severity.WARNING, db_type="mysql", description="IN 子句中的值数量不应超过阈值", config={"enabled": True, "max_in_values": 100}, message_template="IN 子句包含 {count} 个值，超过建议上限 {max_in}", )) def check_mysql_perf_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """大IN子句检查。""" max_in = rule.config.get("max_in_values", 100) in_counts = [] for match in re.finditer(r"\bIN\s*\(([^)]+)\)", sql_upper, re.IGNORECASE): inner = match.group(1) if inner.strip(): count = inner.count(",") + 1 in_counts.append(count) for count in in_counts: if count > max_in: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(count=count, max_in=max_in), )) # ── PERF-005: 多表JOIN检查 ── @register_rule(AuditRuleV2( code="MYSQL-PERF-005", name="ExcessiveJoins", category=RuleCategory.PERFORMANCE, severity=Severity.WARNING, db_type="mysql", description="JOIN 表数量不应超过阈值", config={"enabled": True, "max_joins": 5}, message_template="查询包含 {count} 个 JOIN，超过建议上限 {max_joins}", )) def check_mysql_perf_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """多表JOIN检查。""" if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE): return join_count = len(re.findall(r"\bJOIN\b", sql_upper, re.IGNORECASE)) max_joins = rule.config.get("max_joins", 5) if join_count > max_joins: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(count=join_count, max_joins=max_joins), )) # ────────────────────────────────────────────── # ══════════════════════════════════════════════ # 安全规则 (5条) — MYSQL-SEC-001 ~ MYSQL-SEC-005 # ══════════════════════════════════════════════ # ────────────────────────────────────────────── # ── SEC-001: 敏感字段检测 ── @register_rule(AuditRuleV2( code="MYSQL-SEC-001", name="SensitiveColumnDetection", category=RuleCategory.SECURITY, severity=Severity.ERROR, db_type="mysql", description="禁止在表定义中使用敏感字段名（如password, secret等）", config={"enabled": True}, message_template="表 '{table}' 包含敏感字段名 '{column}'，请确认是否符合安全规范", )) def check_mysql_sec_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """敏感字段检测。""" if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE): return table_name = _extract_table_name(sql_upper, sql_clean) or "unknown" sensitive_patterns = ["password", "passwd", "secret", "token", "api_key", "apikey", "credential", "auth"] columns = _extract_column_definitions(sql_upper) for col in columns: col_lower = col["name"].lower() for pattern in sensitive_patterns: if pattern in col_lower: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(table=table_name, column=col["name"]), )) # ── SEC-002: SQL注入风险 — 字符串拼接检查 ── @register_rule(AuditRuleV2( code="MYSQL-SEC-002", name="SQLInjectionRisk", category=RuleCategory.SECURITY, severity=Severity.ERROR, db_type="mysql", description="检测SQL中可能的注入风险（如字符串拼接占位符）", config={"enabled": True}, message_template="SQL语句可能存在注入风险，请使用参数化查询", )) def check_mysql_sec_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """SQL注入风险检查。""" # 检测 ${xxx} 或 #{xxx} 等模板占位符（不应出现在最终SQL中） if re.search(r"\$\{[^}]+\}", sql_clean) or re.search(r"#\{[^}]+\}", sql_clean): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # 检测字符串拼接模式 if "' || '" in sql_clean or "' + '" in sql_clean or "CONCAT(" in sql_upper: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── SEC-003: 明文密码检测 ── @register_rule(AuditRuleV2( code="MYSQL-SEC-003", name="PlaintextPassword", category=RuleCategory.SECURITY, severity=Severity.ERROR, db_type="mysql", description="禁止 INSERT/UPDATE 中设置明文密码值", config={"enabled": True}, message_template="禁止在SQL中写入明文密码，请使用加密后的值", )) def check_mysql_sec_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """明文密码检测。""" if not _is_dml_statement(sql_upper): return # 检测 password/secret 相关的值设置 if re.search(r"\b(password|secret|passwd)\b\s*[=)]\s*['\"][^'\"]+['\"]", sql_upper, re.IGNORECASE): report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ── SEC-004: 注释符检查 ── @register_rule(AuditRuleV2( code="MYSQL-SEC-004", name="NoCommentInSQL", category=RuleCategory.SECURITY, severity=Severity.WARNING, db_type="mysql", description="SQL中不应包含注释（可能是SQL注入手法）", config={"enabled": True}, message_template="SQL中包含注释符，请移除: {comment}", )) def check_mysql_sec_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """注释符检查。""" if "--" in sql_clean: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(comment="--"), )) if "/*" in sql_clean or "*/" in sql_clean: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template.format(comment="/* */"), )) # ── SEC-005: 多语句检查 ── @register_rule(AuditRuleV2( code="MYSQL-SEC-005", name="NoMultiStatement", category=RuleCategory.SECURITY, severity=Severity.ERROR, db_type="mysql", description="禁止单条输入中包含多条SQL语句（分号分隔）", config={"enabled": True}, message_template="禁止提交多条SQL语句，请拆分为多个工单", )) def check_mysql_sec_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None: """多语句检查。""" stripped = sql_clean.rstrip(";").strip() if ";" in stripped: report.findings.append(AuditFindingV2( rule_code=rule.code, rule_name=rule.name, level=rule.severity, message=rule.message_template, )) # ═══════════════════════════════════════════════════ # 模块总结 # ═══════════════════════════════════════════════════ # 总计 63 条规则: # - DDL 规则: 36 条 (MYSQL-DDL-001 ~ MYSQL-DDL-036) # - DML 规则: 9 条 (MYSQL-DML-001 ~ MYSQL-DML-009) # - Online-DDL 规则: 3 条 (MYSQL-OSC-001 ~ MYSQL-OSC-003) # - 命名规则: 5 条 (MYSQL-NAM-001 ~ MYSQL-NAM-005) # - 性能规则: 5 条 (MYSQL-PERF-001 ~ MYSQL-PERF-005) # - 安全规则: 5 条 (MYSQL-SEC-001 ~ MYSQL-SEC-005) # ═══════════════════════════════════════════════════