logan
58eb78c987
后端新增 (Step 1): - mysql_audit_rules.py: 63条 MySQL审核规则 (DDL 36 + DML 9 + 命名 5 + 性能 5 + 安全 5) - redis_audit_rules.py: 43条 Redis审核规则 (KEY管理 10 + 命令安全 12 + 数据结构 8 + 性能 8 + 安全 5) - mongodb_audit_rules.py: 42条 MongoDB审核规则 (集合管理 + 查询安全 + 聚合管道 + 写入安全 + 性能) - postgres_audit_rules.py: 45条 PostgreSQL审核规则 (DDL 18 + DML 8 + DQL 6 + 安全 6 + 性能 7) 后端合并 (Step 2): - audit_engine_v2.py: 新增4个DB规则模块import (100+→200+规则) - audit_rule_seeds.py: 新增30条MongoDB种子规则 + 别名映射 - audit_rule.py: 新增数据库类型别名 (postgresql→postgres, pg→postgres, mongo→mongodb) 前端优化 (Step 4) — 16页面全覆盖: - InstancePage, WorkflowPage, TicketsPage, TicketCreatePage, TicketDetailPage - UsersPage, SlowQueryPage, DataDictionaryPage, Dashboard - InstanceGroupPage, AuditLogsPage, QueriesPage, AuditRulesPage, RolesPage - useAuth.tsx, SqlEditor.tsx 后端测试 895/895 通过, 前端 tsc --noEmit 通过
2102 lines
86 KiB
Python
2102 lines
86 KiB
Python
"""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)
|
||
# ═══════════════════════════════════════════════════
|