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 通过
1700 lines
64 KiB
Python
1700 lines
64 KiB
Python
"""SQL 审核引擎 V2 — 可配置规则引擎，支持 200+ 审核规则。
|
||
|
||
对标 Yearning/Archery/Bytebase 的审核能力，支持:
|
||
- DDL 规则 (建表规范、索引规范、字段规范)
|
||
- DML 规则 (WHERE 条件、批量限制、事务规范)
|
||
- DQL 规则 (SELECT * 检测、LIMIT 检查)
|
||
- 命名规范 (表名/字段名/索引名规范)
|
||
- 性能规则 (全表扫描、索引使用、大 IN 子句)
|
||
- 安全规则 (敏感字段、SQL 注入风险、密码硬编码)
|
||
- MySQL 专用规则 (62+ 条，覆盖 MySQL 特有语法)
|
||
- PostgreSQL 专用规则 (45+ 条，覆盖 PG 特有语法)
|
||
- Redis 专用规则 (43+ 条，覆盖 Redis 命令和数据结构)
|
||
- MongoDB 专用规则 (42+ 条，覆盖 MongoDB 查询和聚合)
|
||
|
||
使用方式:
|
||
engine = SQLAuditEngineV2(db_session)
|
||
report = await engine.audit("UPDATE users SET status='active'", db_type="mysql")
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from typing import Any, Callable, Coroutine, Dict, List, Optional, Pattern, Set, Tuple
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 枚举与数据类定义
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class Severity(str, Enum):
|
||
"""审核发现项严重级别。"""
|
||
ERROR = "error"
|
||
WARNING = "warning"
|
||
INFO = "info"
|
||
|
||
|
||
class RuleCategory(str, Enum):
|
||
"""规则分类。"""
|
||
DDL = "DDL"
|
||
DML = "DML"
|
||
DQL = "DQL"
|
||
NAMING = "naming"
|
||
PERFORMANCE = "performance"
|
||
SECURITY = "security"
|
||
|
||
|
||
@dataclass
|
||
class AuditRuleV2:
|
||
"""规则定义数据类。
|
||
|
||
Attributes:
|
||
code: 规则唯一编码，如 "DDL-001"
|
||
name: 规则名称
|
||
category: 规则分类
|
||
severity: 严重级别
|
||
db_type: 适用的数据库类型 (mysql/postgres/redis/all)
|
||
description: 规则描述
|
||
is_enabled: 是否启用
|
||
config: 规则配置字典（如 {"max_index_count": 12}）
|
||
pattern: 可选的正则表达式模式
|
||
message_template: 发现违规时的消息模板
|
||
"""
|
||
code: str
|
||
name: str
|
||
category: RuleCategory
|
||
severity: Severity
|
||
db_type: str = "all"
|
||
description: str = ""
|
||
is_enabled: bool = True
|
||
config: Dict[str, Any] = field(default_factory=dict)
|
||
pattern: Optional[str] = None
|
||
message_template: str = ""
|
||
|
||
@property
|
||
def compiled_pattern(self) -> Optional[Pattern[str]]:
|
||
"""编译并缓存正则表达式。"""
|
||
if self.pattern is None:
|
||
return None
|
||
return re.compile(self.pattern, re.IGNORECASE)
|
||
|
||
|
||
@dataclass
|
||
class AuditFindingV2:
|
||
"""审核发现项数据类。
|
||
|
||
Attributes:
|
||
rule_code: 触发的规则编码
|
||
rule_name: 规则名称
|
||
level: 严重级别
|
||
message: 发现项描述消息
|
||
position: 可选的 SQL 位置信息
|
||
"""
|
||
rule_code: str
|
||
rule_name: str
|
||
level: Severity
|
||
message: str
|
||
position: Optional[int] = None
|
||
|
||
|
||
@dataclass
|
||
class AuditReportV2:
|
||
"""审核报告数据类。
|
||
|
||
Attributes:
|
||
sql: 被审核的 SQL 语句
|
||
findings: 审核发现项列表
|
||
db_type: 数据库类型
|
||
score: 审核评分 (100分满分)
|
||
"""
|
||
sql: str
|
||
findings: List[AuditFindingV2] = field(default_factory=list)
|
||
db_type: str = "mysql"
|
||
|
||
@property
|
||
def passed(self) -> bool:
|
||
"""是否通过审核（无 ERROR 级别发现项）。"""
|
||
return not any(f.level == Severity.ERROR for f in self.findings)
|
||
|
||
@property
|
||
def has_warnings(self) -> bool:
|
||
"""是否包含 WARNING 级别发现项。"""
|
||
return any(f.level == Severity.WARNING for f in self.findings)
|
||
|
||
@property
|
||
def score(self) -> int:
|
||
"""计算审核评分。
|
||
|
||
100分满分，每个 ERROR -10分，WARNING -5分，INFO -2分，最低0分。
|
||
"""
|
||
s = 100
|
||
for f in self.findings:
|
||
if f.level == Severity.ERROR:
|
||
s -= 10
|
||
elif f.level == Severity.WARNING:
|
||
s -= 5
|
||
elif f.level == Severity.INFO:
|
||
s -= 2
|
||
return max(0, s)
|
||
|
||
@property
|
||
def summary(self) -> str:
|
||
"""生成审核摘要。"""
|
||
if not self.findings:
|
||
return "审核通过，未发现违规项"
|
||
error_count = sum(1 for f in self.findings if f.level == Severity.ERROR)
|
||
warning_count = sum(1 for f in self.findings if f.level == Severity.WARNING)
|
||
info_count = sum(1 for f in self.findings if f.level == Severity.INFO)
|
||
parts: List[str] = []
|
||
if error_count:
|
||
parts.append(f"{error_count}个错误")
|
||
if warning_count:
|
||
parts.append(f"{warning_count}个警告")
|
||
if info_count:
|
||
parts.append(f"{info_count}个提示")
|
||
return f"审核评分: {self.score}分, " + ", ".join(parts)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""转换为字典格式（向后兼容旧版 AuditReport）。"""
|
||
return {
|
||
"passed": self.passed,
|
||
"score": self.score,
|
||
"findings": [
|
||
{
|
||
"rule": f.rule_code,
|
||
"level": f.level.value,
|
||
"message": f.message,
|
||
}
|
||
for f in self.findings
|
||
],
|
||
"summary": self.summary,
|
||
"db_type": self.db_type,
|
||
}
|
||
|
||
# 向后兼容旧版 AuditReport 的 has_errors 属性
|
||
@property
|
||
def has_errors(self) -> bool:
|
||
"""是否存在错误级别发现项。"""
|
||
return any(f.level == Severity.ERROR for f in self.findings)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# SQL 辅助解析函数
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
def _extract_table_name(sql_upper: str, sql_clean: str) -> Optional[str]:
|
||
"""从 CREATE TABLE / ALTER TABLE 语句中提取表名。
|
||
|
||
从 sql_clean 提取以保留原始大小写，用于命名规范检查。
|
||
"""
|
||
m = re.search(r"\bCREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"']?(\w+)[`\"']?", sql_clean, re.IGNORECASE)
|
||
if m:
|
||
return m.group(1)
|
||
m = re.search(r"\bALTER\s+TABLE\s+[`\"']?(\w+)[`\"']?", sql_clean, re.IGNORECASE)
|
||
if m:
|
||
return m.group(1)
|
||
return None
|
||
|
||
|
||
def _extract_column_definitions(sql_upper: str) -> List[Dict[str, str]]:
|
||
"""从 CREATE TABLE (...) 中提取字段定义列表。"""
|
||
# 找到括号内的内容
|
||
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())
|
||
|
||
columns: List[Dict[str, str]] = []
|
||
for part in parts:
|
||
# 跳过约束定义 (PRIMARY KEY, FOREIGN KEY, INDEX, KEY, UNIQUE)
|
||
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:
|
||
columns.append({
|
||
"name": col_match.group(1),
|
||
"type": col_match.group(2).upper(),
|
||
"full": part,
|
||
})
|
||
return columns
|
||
|
||
|
||
def _extract_index_definitions(sql_upper: str) -> List[Dict[str, Any]]:
|
||
"""从 CREATE TABLE (...) 中提取索引定义列表。"""
|
||
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())
|
||
|
||
indexes: List[Dict[str, Any]] = []
|
||
for part in parts:
|
||
# PRIMARY KEY
|
||
pk_match = re.match(r"PRIMARY\s+KEY\s*\(([^)]+)\)", part, re.IGNORECASE)
|
||
if pk_match:
|
||
cols = [c.strip().strip("`'\"") for c in pk_match.group(1).split(",")]
|
||
indexes.append({"type": "PRIMARY", "name": "PRIMARY", "columns": cols})
|
||
continue
|
||
# UNIQUE KEY / INDEX
|
||
uk_match = re.match(r"UNIQUE\s+(?:INDEX\s+|KEY\s+)?[`\"']?(\w*)[`\"']?\s*\(([^)]+)\)", part, re.IGNORECASE)
|
||
if uk_match:
|
||
cols = [c.strip().strip("`'\"") for c in uk_match.group(2).split(",")]
|
||
indexes.append({"type": "UNIQUE", "name": uk_match.group(1) or "", "columns": cols})
|
||
continue
|
||
# KEY / INDEX
|
||
idx_match = re.match(r"(?:KEY|INDEX)\s+[`\"']?(\w+)[`\"']?\s*\(([^)]+)\)", part, re.IGNORECASE)
|
||
if idx_match:
|
||
cols = [c.strip().strip("`'\"") for c in idx_match.group(2).split(",")]
|
||
indexes.append({"type": "INDEX", "name": idx_match.group(1), "columns": cols})
|
||
continue
|
||
# FOREIGN KEY
|
||
fk_match = re.match(r"FOREIGN\s+KEY\s+[`\"']?(\w*)[`\"']?\s*\(([^)]+)\)", part, re.IGNORECASE)
|
||
if fk_match:
|
||
cols = [c.strip().strip("`'\"") for c in fk_match.group(2).split(",")]
|
||
indexes.append({"type": "FOREIGN", "name": fk_match.group(1) or "", "columns": cols})
|
||
return indexes
|
||
|
||
|
||
def _is_snake_case(name: str) -> bool:
|
||
"""检查是否为小写下划线命名。"""
|
||
if not name:
|
||
return False
|
||
return bool(re.match(r"^[a-z][a-z0-9_]*$", name))
|
||
|
||
|
||
def _count_in_values(sql_upper: str) -> List[int]:
|
||
"""统计所有 IN 子句中的值数量。"""
|
||
counts: List[int] = []
|
||
# 匹配 IN (...) 模式
|
||
for match in re.finditer(r"\bIN\s*\(([^)]+)\)", sql_upper, re.IGNORECASE):
|
||
inner = match.group(1)
|
||
# 统计逗号数量 + 1
|
||
if inner.strip():
|
||
count = inner.count(",") + 1
|
||
counts.append(count)
|
||
return counts
|
||
|
||
|
||
def _extract_order_by_columns(sql_upper: str) -> List[str]:
|
||
"""提取 ORDER BY 中的字段名。"""
|
||
cols: List[str] = []
|
||
# 匹配 ORDER BY 后的内容，直到 LIMIT/OFFSET/GROUP/HAVING/UNION/; 或结束
|
||
ob_match = re.search(r"\bORDER\s+BY\s+(.+?)(?:\s+(?:LIMIT|OFFSET|GROUP|HAVING|UNION|;)\s*|$)", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if ob_match:
|
||
inner = ob_match.group(1).strip().rstrip(";")
|
||
if inner:
|
||
for part in inner.split(","):
|
||
# 去除 ASC/DESC
|
||
col = re.sub(r"\s+(ASC|DESC)\s*$", "", part.strip(), flags=re.IGNORECASE).strip()
|
||
if col:
|
||
cols.append(col)
|
||
return cols
|
||
|
||
|
||
def _count_joins(sql_upper: str) -> int:
|
||
"""统计 JOIN 数量。"""
|
||
return len(re.findall(r"\bJOIN\b", sql_upper, re.IGNORECASE))
|
||
|
||
|
||
def _extract_select_columns(sql_upper: str) -> List[str]:
|
||
"""提取 SELECT 后的字段列表。"""
|
||
m = re.search(r"\bSELECT\s+(.+?)\s+FROM\b", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if not m:
|
||
return []
|
||
inner = m.group(1)
|
||
# 处理简单的字段列表（忽略子查询中的复杂情况）
|
||
if "*" in inner and len(inner.strip()) == 1:
|
||
return ["*"]
|
||
cols = []
|
||
current = ""
|
||
depth = 0
|
||
for char in inner:
|
||
if char == "(":
|
||
depth += 1
|
||
current += char
|
||
elif char == ")":
|
||
depth -= 1
|
||
current += char
|
||
elif char == "," and depth == 0:
|
||
cols.append(current.strip())
|
||
current = ""
|
||
else:
|
||
current += char
|
||
if current.strip():
|
||
cols.append(current.strip())
|
||
return cols
|
||
|
||
|
||
def _has_primary_key_condition(sql_upper: str, table_name: str = "") -> bool:
|
||
"""检查 WHERE 条件是否包含主键条件。"""
|
||
where_match = re.search(r"\bWHERE\b(.+)$", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if not where_match:
|
||
return False
|
||
where_clause = where_match.group(1)
|
||
# 检查是否包含 id = 或类似主键条件
|
||
pk_patterns = [
|
||
r"\bid\s*[=<>]",
|
||
r"\bid\s+IN\s*\(",
|
||
r"\b\w+_id\s*[=<>]",
|
||
r"\bpk_\w+\s*[=<>]",
|
||
]
|
||
return any(re.search(p, where_clause, re.IGNORECASE) for p in pk_patterns)
|
||
|
||
|
||
def _has_where_with_function(sql_upper: str) -> bool:
|
||
"""检查 WHERE 子句中是否使用了函数。"""
|
||
where_match = re.search(r"\bWHERE\b(.+)$", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if not where_match:
|
||
return False
|
||
where_clause = where_match.group(1)
|
||
# 匹配 WHERE 中使用函数的模式，如 WHERE YEAR(col) = ...
|
||
# 函数名通常是字母+下划线，后面跟左括号
|
||
return bool(re.search(r"\b[A-Z_][A-Z0-9_]*\s*\([^)]+", where_clause, re.IGNORECASE))
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 规则检查函数注册表
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
RuleCheckFunc = Callable[[str, str, AuditReportV2, AuditRuleV2], None]
|
||
|
||
_RULE_REGISTRY: Dict[str, Tuple[AuditRuleV2, RuleCheckFunc]] = {}
|
||
|
||
|
||
def register_rule(rule: AuditRuleV2) -> Callable[[RuleCheckFunc], RuleCheckFunc]:
|
||
"""注册规则的装饰器工厂。
|
||
|
||
用法::
|
||
|
||
@register_rule(AuditRuleV2(...))
|
||
def check_xxx(sql_clean, sql_upper, report, rule): ...
|
||
"""
|
||
def decorator(func: RuleCheckFunc) -> RuleCheckFunc:
|
||
_RULE_REGISTRY[rule.code] = (rule, func)
|
||
return func
|
||
return decorator
|
||
|
||
|
||
def get_all_registered_rules() -> List[AuditRuleV2]:
|
||
"""获取所有已注册的规则定义（去重）。"""
|
||
return [rule for rule, _ in _RULE_REGISTRY.values()]
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# DDL 规则 (至少12条)
|
||
# ═══════════════════════════════════════════════
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-001",
|
||
name="建表必须包含主键",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="CREATE TABLE 语句必须包含 PRIMARY KEY 定义",
|
||
message_template="建表语句缺少主键定义，请添加 PRIMARY KEY",
|
||
))
|
||
def check_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
|
||
# 检查是否有 PRIMARY KEY 定义
|
||
if "PRIMARY KEY" not in sql_upper:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-002",
|
||
name="建表必须添加表注释",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="CREATE TABLE 语句必须包含 COMMENT 表注释",
|
||
message_template="建表时必须添加表注释 (COMMENT='...')",
|
||
))
|
||
def check_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
|
||
# 检查表级 COMMENT（支持 MySQL COMMENT= 或 PostgreSQL COMMENT ON）
|
||
has_comment = (
|
||
re.search(r"COMMENT\s*[=\s]+['\"]", sql_clean, re.IGNORECASE) or
|
||
re.search(r"\bCOMMENT\s+ON\s+TABLE\b", sql_upper, re.IGNORECASE)
|
||
)
|
||
if not has_comment:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-003",
|
||
name="建表字段必须有注释",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="CREATE TABLE 中的每个字段都应包含 COMMENT",
|
||
message_template="字段 '{column}' 缺少注释，请添加 COMMENT",
|
||
))
|
||
def check_ddl_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
|
||
columns = _extract_column_definitions(sql_upper)
|
||
for col in columns:
|
||
# 检查字段定义中是否包含 COMMENT
|
||
if "COMMENT" not in col["full"]:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(column=col["name"]),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-004",
|
||
name="禁止 DROP TABLE",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="生产环境禁止执行 DROP TABLE 操作",
|
||
config={"env_production_only": True},
|
||
message_template="禁止执行 DROP TABLE 操作，数据将永久丢失",
|
||
))
|
||
def check_ddl_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 DROP TABLE（生产环境）。"""
|
||
if not re.search(r"\bDROP\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-005",
|
||
name="禁止 TRUNCATE TABLE",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="禁止执行 TRUNCATE TABLE 操作",
|
||
message_template="禁止执行 TRUNCATE TABLE 操作，数据不可回滚",
|
||
))
|
||
def check_ddl_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 TRUNCATE TABLE。"""
|
||
if not re.search(r"\bTRUNCATE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-006",
|
||
name="单表索引数量不得超过12个",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_index_count": 12},
|
||
message_template="单表索引数量为 {count} 个，超过最大限制 {max}",
|
||
))
|
||
def check_ddl_006(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
|
||
max_count = rule.config.get("max_index_count", 12)
|
||
indexes = _extract_index_definitions(sql_upper)
|
||
idx_count = len([i for i in indexes if i["type"] in ("INDEX", "UNIQUE")])
|
||
if idx_count > max_count:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(count=idx_count, max=max_count),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-007",
|
||
name="单索引字段数不得超过8个",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_index_columns": 8},
|
||
message_template="索引 '{name}' 包含 {count} 个字段，超过最大限制 {max}",
|
||
))
|
||
def check_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
|
||
max_cols = rule.config.get("max_index_columns", 8)
|
||
indexes = _extract_index_definitions(sql_upper)
|
||
for idx in indexes:
|
||
if len(idx["columns"]) > max_cols:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(
|
||
name=idx["name"], count=len(idx["columns"]), max=max_cols
|
||
),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-008",
|
||
name="禁止使用 FLOAT/DOUBLE 类型",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="浮点类型精度丢失，应使用 DECIMAL",
|
||
message_template="字段 '{column}' 使用了 {type} 类型，建议使用 DECIMAL 避免精度丢失",
|
||
))
|
||
def check_ddl_008(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止使用 FLOAT/DOUBLE。"""
|
||
if not re.search(r"\bCREATE\s+TABLE\b|\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
columns = _extract_column_definitions(sql_upper)
|
||
for col in columns:
|
||
if col["type"] in ("FLOAT", "DOUBLE", "REAL"):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(column=col["name"], type=col["type"]),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-009",
|
||
name="CHAR 类型长度不得超过10",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_char_length": 10},
|
||
message_template="CHAR 类型长度不得超过 {max}，超出请使用 VARCHAR",
|
||
))
|
||
def check_ddl_009(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""CHAR 类型长度检查。"""
|
||
if not re.search(r"\bCREATE\s+TABLE\b|\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
max_len = rule.config.get("max_char_length", 10)
|
||
# 匹配 CHAR(N) 模式
|
||
for match in re.finditer(r"\bCHAR\s*\((\d+)\)", sql_upper, re.IGNORECASE):
|
||
length = int(match.group(1))
|
||
if length > max_len:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(max=max_len),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-010",
|
||
name="表名长度不得超过45字符",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_table_name_length": 45},
|
||
message_template="表名 '{name}' 长度为 {length}，超过最大限制 {max}",
|
||
))
|
||
def check_ddl_010(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""表名长度检查。"""
|
||
table_name = _extract_table_name(sql_upper, sql_clean)
|
||
if not table_name:
|
||
return
|
||
max_len = rule.config.get("max_table_name_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(name=table_name, length=len(table_name), max=max_len),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-011",
|
||
name="禁止外键约束",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="外键约束影响性能，推荐在应用层处理关联",
|
||
message_template="禁止使用 FOREIGN KEY 约束，建议在应用层维护数据一致性",
|
||
))
|
||
def check_ddl_011(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止外键约束。"""
|
||
if not re.search(r"\bCREATE\s+TABLE\b|\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
if "FOREIGN KEY" in sql_upper:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DDL-012",
|
||
name="必须包含 create_time/update_time 字段",
|
||
category=RuleCategory.DDL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="建表必须包含 create_time 和 update_time 审计字段",
|
||
message_template="建表缺少审计字段: {missing}",
|
||
))
|
||
def check_ddl_012(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""必须包含 create_time/update_time 字段。"""
|
||
if not re.search(r"\bCREATE\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
columns = _extract_column_definitions(sql_upper)
|
||
col_names = [c["name"].lower() for c in columns]
|
||
required = ["create_time", "update_time"]
|
||
missing = [r for r in required if r not in col_names]
|
||
if missing:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(missing=", ".join(missing)),
|
||
))
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# DML 规则 (至少8条)
|
||
# ═══════════════════════════════════════════════
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-001",
|
||
name="UPDATE/DELETE 必须包含 WHERE 条件",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="UPDATE/DELETE 缺少 WHERE 将影响全表数据",
|
||
message_template="{operation} 语句缺少 WHERE 条件，将操作全表数据",
|
||
))
|
||
def check_dml_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""UPDATE/DELETE 必须包含 WHERE。"""
|
||
for keyword in ("UPDATE", "DELETE"):
|
||
if re.match(rf"\s*{keyword}\b", sql_upper, re.IGNORECASE):
|
||
if "WHERE" not in sql_upper:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(operation=keyword),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-002",
|
||
name="单次 INSERT 最多500行",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_insert_rows": 500},
|
||
message_template="单次 INSERT 包含 {count} 行，超过最大限制 {max}",
|
||
))
|
||
def check_dml_002(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_insert_rows", 500)
|
||
# 统计 VALUES 子句数量（简化：通过右括号数量估计）
|
||
values_sections = re.findall(r"\bVALUES\s*\((.+?)\)", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if not values_sections:
|
||
return
|
||
# 多行 VALUES (value1), (value2), (value3)
|
||
values_match = re.search(r"\bVALUES\s+(.+)", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if values_match:
|
||
values_part = values_match.group(1)
|
||
# 计算行数（通过未闭合的括号组）
|
||
# 简化：计算 VALUES 后的顶层括号组数量
|
||
row_count = 0
|
||
depth = 0
|
||
for i, char in enumerate(values_part):
|
||
if char == "(":
|
||
if depth == 0:
|
||
row_count += 1
|
||
depth += 1
|
||
elif char == ")":
|
||
depth -= 1
|
||
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=max_rows),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-003",
|
||
name="UPDATE/DELETE 禁止带 LIMIT",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.WARNING,
|
||
db_type="mysql",
|
||
description="UPDATE/DELETE 使用 LIMIT 可能导致部分数据变更",
|
||
message_template="{operation} 语句禁止使用 LIMIT，可能导致数据变更不完整",
|
||
))
|
||
def check_dml_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""UPDATE/DELETE 禁止带 LIMIT。"""
|
||
for keyword in ("UPDATE", "DELETE"):
|
||
if re.match(rf"\s*{keyword}\b", sql_upper, re.IGNORECASE):
|
||
if "LIMIT" in sql_upper:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(operation=keyword),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-004",
|
||
name="禁止 INSERT 不指定字段列表",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="INSERT 必须显式指定字段列表",
|
||
message_template="INSERT 语句必须显式指定字段列表，如 INSERT INTO t (col1, col2) VALUES ...",
|
||
))
|
||
def check_dml_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 INSERT 不指定字段列表。"""
|
||
# 匹配 INSERT INTO table VALUES 模式（没有字段列表）
|
||
if re.search(r"\bINSERT\s+(?:INTO\s+)?\w+\s*\(?\w*\)?\s+VALUES\b", sql_upper, re.IGNORECASE):
|
||
# 检查是否不包含字段列表 (col1, col2)
|
||
# 通过检查 INSERT INTO table ( 模式
|
||
if not re.search(r"\bINSERT\s+(?:INTO\s+)?\w+\s*\([^)]+\)\s+VALUES\b", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-005",
|
||
name="禁止批量 UPDATE 不带主键条件",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="批量 UPDATE 必须带主键 WHERE 条件",
|
||
message_template="批量 UPDATE 必须包含主键条件，当前 WHERE 条件不含主键",
|
||
))
|
||
def check_dml_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""批量 UPDATE 必须带主键条件。"""
|
||
if not re.match(r"\s*UPDATE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
if "WHERE" not in sql_upper:
|
||
return # 已被 DML-001 覆盖
|
||
if not _has_primary_key_condition(sql_upper):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-006",
|
||
name="DELETE 必须有 LIMIT（除非带主键条件）",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.WARNING,
|
||
db_type="mysql",
|
||
description="DELETE 必须带 LIMIT 或主键 WHERE 条件",
|
||
message_template="DELETE 建议添加 LIMIT 或使用主键条件防止误删大量数据",
|
||
))
|
||
def check_dml_006(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""DELETE 必须有 LIMIT 或主键条件。"""
|
||
if not re.match(r"\s*DELETE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
# 如果有主键条件则通过
|
||
if _has_primary_key_condition(sql_upper):
|
||
return
|
||
# 如果有 LIMIT 则通过
|
||
if "LIMIT" in sql_upper:
|
||
return
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-007",
|
||
name="禁止 REPLACE INTO",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.ERROR,
|
||
db_type="mysql",
|
||
description="REPLACE INTO 会先删除再插入，风险较高",
|
||
message_template="禁止使用 REPLACE INTO，请使用 INSERT ... ON DUPLICATE KEY UPDATE 替代",
|
||
))
|
||
def check_dml_007(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 REPLACE INTO。"""
|
||
if re.search(r"\bREPLACE\s+INTO\b", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DML-008",
|
||
name="INSERT 必须指定字段列表",
|
||
category=RuleCategory.DML,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="INSERT 语句必须显式指定字段列表，避免表结构变更导致错误",
|
||
message_template="INSERT 语句必须显式指定字段列表",
|
||
))
|
||
def check_dml_008(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""INSERT 必须指定字段列表（与 DML-004 互补）。"""
|
||
if not re.search(r"\bINSERT\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
# 更严格的检查：INSERT INTO table_name (col1, col2) 必须存在
|
||
if not re.search(r"\bINSERT\s+(?:INTO\s+)?[\w`\"]+\s*\([^)]+\)", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# DQL 规则 (至少5条)
|
||
# ═══════════════════════════════════════════════
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DQL-001",
|
||
name="禁止 SELECT *",
|
||
category=RuleCategory.DQL,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="SELECT * 返回不必要的列，影响性能",
|
||
message_template="禁止使用 SELECT *，请明确指定需要的字段",
|
||
))
|
||
def check_dql_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
|
||
# 检查 SELECT * 或 SELECT COUNT(*)
|
||
if re.search(r"\bSELECT\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,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DQL-002",
|
||
name="SELECT 必须有 LIMIT",
|
||
category=RuleCategory.DQL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="SELECT 查询应添加 LIMIT 限制返回行数",
|
||
message_template="SELECT 查询缺少 LIMIT，建议添加 LIMIT 防止返回过多数据",
|
||
))
|
||
def check_dql_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""SELECT 必须有 LIMIT。"""
|
||
if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
# 排除子查询和 INSERT ... SELECT
|
||
if re.search(r"\bINSERT\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
# 检查 LIMIT
|
||
if "LIMIT" not in sql_upper:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DQL-003",
|
||
name="禁止在 WHERE 中使用函数",
|
||
category=RuleCategory.DQL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="WHERE 中使用函数会导致索引失效",
|
||
message_template="WHERE 子句中使用函数会导致索引失效，请将函数移出 WHERE",
|
||
))
|
||
def check_dql_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止在 WHERE 中使用函数。"""
|
||
if "WHERE" not in sql_upper:
|
||
return
|
||
if not re.search(r"\bSELECT\b|\bUPDATE\b|\bDELETE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
if _has_where_with_function(sql_upper):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DQL-004",
|
||
name="禁止隐式类型转换",
|
||
category=RuleCategory.DQL,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="不同类型比较会导致索引失效",
|
||
message_template="WHERE 条件中存在隐式类型转换风险，请确保比较双方类型一致",
|
||
))
|
||
def check_dql_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止隐式类型转换。"""
|
||
if "WHERE" not in sql_upper:
|
||
return
|
||
# 检测字符串列与数字比较的模式（简化）
|
||
# 例如 WHERE id = '123' 且 id 是数字列
|
||
# 或 WHERE name = 123 且 name 是字符串列
|
||
patterns = [
|
||
r"\bWHERE\b.*=\s*'\d+'", # 字符串数字比较
|
||
r"\bWHERE\b.*\w+\s*=\s*\d+\b.*AND.*\w+\s*=\s*'[^']*'", # 混合类型
|
||
]
|
||
# 简化实现：检测 WHERE 条件中数字和字符串比较的混合
|
||
where_clause = ""
|
||
where_match = re.search(r"\bWHERE\b(.+)$", sql_upper, re.IGNORECASE | re.DOTALL)
|
||
if where_match:
|
||
where_clause = where_match.group(1)
|
||
# 检测数字列与字符串字面量比较
|
||
has_numeric_cmp = bool(re.search(r"=\s*\d+\b", where_clause))
|
||
has_string_cmp = bool(re.search(r"=\s*'[^']*'", where_clause))
|
||
# 如果同时存在数字和字符串比较，可能存在类型转换问题
|
||
if has_numeric_cmp and has_string_cmp:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="DQL-005",
|
||
name="ORDER BY 必须使用索引字段",
|
||
category=RuleCategory.DQL,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="ORDER BY 非索引字段会导致文件排序（filesort）",
|
||
message_template="ORDER BY 字段建议添加索引，避免 filesort 性能问题",
|
||
))
|
||
def check_dql_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""ORDER BY 索引检查（静态检查，提示级别）。"""
|
||
if "ORDER BY" not in sql_upper:
|
||
return
|
||
order_cols = _extract_order_by_columns(sql_upper)
|
||
if order_cols:
|
||
# 静态检查无法确定是否有索引，给出提示
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 命名规范规则 (至少5条)
|
||
# ═══════════════════════════════════════════════
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="NAME-001",
|
||
name="表名必须小写下划线命名",
|
||
category=RuleCategory.NAMING,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="表名应使用小写下划线格式 (snake_case)",
|
||
message_template="表名 '{name}' 不符合小写下划线命名规范",
|
||
))
|
||
def check_name_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""表名必须小写下划线命名。"""
|
||
table_name = _extract_table_name(sql_upper, sql_clean)
|
||
if not table_name:
|
||
return
|
||
if 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(name=table_name),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="NAME-002",
|
||
name="字段名必须小写下划线命名",
|
||
category=RuleCategory.NAMING,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="字段名应使用小写下划线格式 (snake_case)",
|
||
message_template="字段名 '{name}' 不符合小写下划线命名规范",
|
||
))
|
||
def check_name_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
|
||
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(name=col["name"]),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="NAME-003",
|
||
name="索引名必须以 idx_ 或 uk_ 开头",
|
||
category=RuleCategory.NAMING,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="索引命名规范：普通索引 idx_，唯一索引 uk_",
|
||
message_template="索引名 '{name}' 不符合命名规范，应以 idx_ 或 uk_ 开头",
|
||
))
|
||
def check_name_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""索引名规范检查。"""
|
||
if not re.search(r"\bCREATE\s+TABLE\b|\bCREATE\s+INDEX\b|\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
indexes = _extract_index_definitions(sql_upper)
|
||
for idx in indexes:
|
||
if idx["type"] == "PRIMARY":
|
||
continue # 主键跳过
|
||
name = idx["name"]
|
||
if name and not (name.startswith("idx_") or name.startswith("uk_")):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(name=name),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="NAME-004",
|
||
name="主键名必须为 id 或 pk_ 开头",
|
||
category=RuleCategory.NAMING,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="主键字段命名规范",
|
||
message_template="主键名 '{name}' 不符合规范，建议使用 'id' 或 'pk_' 前缀",
|
||
))
|
||
def check_name_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
|
||
# 查找 PRIMARY KEY 定义的字段
|
||
pk_match = re.search(r"PRIMARY\s+KEY\s*\(([^)]+)\)", sql_upper, re.IGNORECASE)
|
||
if pk_match:
|
||
pk_cols = [c.strip().strip("`'\"") for c in pk_match.group(1).split(",")]
|
||
for col in pk_cols:
|
||
if col.lower() != "id" and not col.lower().startswith("pk_"):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(name=col),
|
||
))
|
||
# 也检查行内 PRIMARY KEY 定义（如 id INT PRIMARY KEY）
|
||
columns = _extract_column_definitions(sql_upper)
|
||
for col in columns:
|
||
if "PRIMARY KEY" in col["full"].upper():
|
||
if col["name"].lower() != "id" and not col["name"].lower().startswith("pk_"):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(name=col["name"]),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="NAME-005",
|
||
name="布尔字段以 is_/has_ 开头",
|
||
category=RuleCategory.NAMING,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="布尔类型字段应以 is_ 或 has_ 前缀命名",
|
||
message_template="布尔字段 '{name}' 建议以 is_ 或 has_ 开头命名",
|
||
))
|
||
def check_name_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""布尔字段命名规范。"""
|
||
if not re.search(r"\bCREATE\s+TABLE\b|\bALTER\s+TABLE\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
columns = _extract_column_definitions(sql_upper)
|
||
bool_types = ("BOOLEAN", "BOOL", "TINYINT(1)", "BIT(1)")
|
||
for col in columns:
|
||
if any(bt in col["type"] for bt in bool_types):
|
||
if not (col["name"].startswith("is_") or col["name"].startswith("has_")):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(name=col["name"]),
|
||
))
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 性能规则 (至少8条)
|
||
# ═══════════════════════════════════════════════
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-001",
|
||
name="禁止大表全表扫描",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_rows_for_full_scan": 10000},
|
||
description="大表（>10000行）查询应避免全表扫描",
|
||
message_template="SELECT 查询缺少 WHERE 条件，大表下会导致全表扫描",
|
||
))
|
||
def check_perf_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止大表全表扫描。"""
|
||
if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
# 检查是否缺少 WHERE（简化：静态检查无法获取实际行数，检测缺少 WHERE 的 SELECT）
|
||
if "WHERE" not in sql_upper and "LIMIT" not in sql_upper:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-002",
|
||
name="IN 子句值数量不得超过100个",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_in_values": 100},
|
||
description="IN 子句值过多会导致性能下降",
|
||
message_template="IN 子句包含 {count} 个值，超过最大限制 {max}，建议使用临时表或 JOIN 替代",
|
||
))
|
||
def check_perf_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""IN 子句值数量检查。"""
|
||
max_vals = rule.config.get("max_in_values", 100)
|
||
in_counts = _count_in_values(sql_upper)
|
||
for count in in_counts:
|
||
if count > max_vals:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(count=count, max=max_vals),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-003",
|
||
name="禁止 LIKE '%xxx%' 前缀模糊匹配",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="前缀模糊匹配无法使用索引",
|
||
message_template="LIKE '%xxx%' 无法使用索引，建议改用全文检索或后缀索引",
|
||
))
|
||
def check_perf_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 LIKE '%xxx%'。"""
|
||
if re.search(r"\bLIKE\s+'%[^']+'", sql_clean, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-004",
|
||
name="JOIN 不得超过5个表",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
config={"max_join_tables": 5},
|
||
message_template="JOIN 涉及 {count} 个表，超过最大限制 {max}",
|
||
))
|
||
def check_perf_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""JOIN 数量检查。"""
|
||
join_count = _count_joins(sql_upper)
|
||
max_joins = rule.config.get("max_join_tables", 5)
|
||
# JOIN 数量 = JOIN 关键字数 + 1（主表）
|
||
total_tables = join_count + 1 if join_count > 0 else 0
|
||
if total_tables > max_joins:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(count=total_tables, max=max_joins),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-005",
|
||
name="禁止 ORDER BY RAND()",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.ERROR,
|
||
db_type="mysql",
|
||
description="ORDER BY RAND() 性能极差，会导致全表扫描",
|
||
message_template="禁止 ORDER BY RAND()，性能极差，建议在应用层随机采样",
|
||
))
|
||
def check_perf_005(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 ORDER BY RAND()。"""
|
||
if re.search(r"\bORDER\s+BY\s+RAND\s*\(", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-006",
|
||
name="子查询建议改为 JOIN",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="相关子查询性能较差，建议改为 JOIN",
|
||
message_template="WHERE 中的子查询建议改为 JOIN 以提升性能",
|
||
))
|
||
def check_perf_006(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""子查询检测。"""
|
||
# 检测 WHERE ... (SELECT ...) 模式
|
||
if re.search(r"\bWHERE\b.*\(\s*\bSELECT\b", sql_upper, re.IGNORECASE | re.DOTALL):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-007",
|
||
name="大表禁止 OFFSET 分页",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.WARNING,
|
||
db_type="all",
|
||
description="大表使用 OFFSET 分页性能差，建议使用游标分页",
|
||
message_template="使用 OFFSET 分页在大表上性能较差，建议改用游标分页 (WHERE id > ? LIMIT N)",
|
||
))
|
||
def check_perf_007(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""大表 OFFSET 分页检测。"""
|
||
if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
if re.search(r"\bOFFSET\s+\d+\b", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="PERF-008",
|
||
name="禁止 SELECT DISTINCT（除非必要）",
|
||
category=RuleCategory.PERFORMANCE,
|
||
severity=Severity.INFO,
|
||
db_type="all",
|
||
description="DISTINCT 需要去重操作，影响性能",
|
||
message_template="SELECT DISTINCT 影响性能，请确认是否必要",
|
||
))
|
||
def check_perf_008(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 SELECT DISTINCT。"""
|
||
if re.search(r"\bSELECT\s+DISTINCT\b", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
# ═══════════════════════════════════════════════
|
||
# 安全规则 (至少5条)
|
||
# ═══════════════════════════════════════════════
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="SEC-001",
|
||
name="禁止查询 password/secret/token 字段",
|
||
category=RuleCategory.SECURITY,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="禁止在 SELECT 中查询敏感字段",
|
||
config={"sensitive_fields": ["password", "secret", "token", "passwd", "pwd", "auth_key", "api_key"]},
|
||
message_template="禁止在查询中返回敏感字段: {fields}",
|
||
))
|
||
def check_sec_001(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止查询敏感字段。"""
|
||
if not re.search(r"\bSELECT\b", sql_upper, re.IGNORECASE):
|
||
return
|
||
sensitive = rule.config.get("sensitive_fields", [])
|
||
cols = _extract_select_columns(sql_upper)
|
||
found: List[str] = []
|
||
for col in cols:
|
||
col_lower = col.lower()
|
||
for sf in sensitive:
|
||
if sf in col_lower:
|
||
found.append(col)
|
||
break
|
||
if found:
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template.format(fields=", ".join(found)),
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="SEC-002",
|
||
name="禁止 SQL 中硬编码密码",
|
||
category=RuleCategory.SECURITY,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="SQL 中不应包含硬编码密码",
|
||
message_template="SQL 中发现可能的硬编码密码，禁止在 SQL 中直接写入密码",
|
||
))
|
||
def check_sec_002(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 SQL 中硬编码密码。"""
|
||
patterns = [
|
||
r"\bPASSWORD\s*[=\s]+['\"][^'\"]+['\"]",
|
||
r"\bSET\s+PASSWORD\s*=",
|
||
r"\bpassword\s*[=:]\s*['\"][^'\"]+['\"]",
|
||
]
|
||
for p in patterns:
|
||
if re.search(p, sql_clean, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
return # 只报告一次
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="SEC-003",
|
||
name="禁止 UNION（SQL 注入风险）",
|
||
category=RuleCategory.SECURITY,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="UNION 语句存在 SQL 注入风险",
|
||
message_template="禁止 UNION 操作，存在 SQL 注入风险",
|
||
))
|
||
def check_sec_003(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止 UNION。"""
|
||
if re.search(r"\bUNION\b", sql_upper, re.IGNORECASE):
|
||
report.findings.append(AuditFindingV2(
|
||
rule_code=rule.code,
|
||
rule_name=rule.name,
|
||
level=rule.severity,
|
||
message=rule.message_template,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="SEC-004",
|
||
name="禁止注释符",
|
||
category=RuleCategory.SECURITY,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="SQL 中不应包含 -- 或 /* */ 注释",
|
||
message_template="SQL 中禁止包含注释符，可能存在注入风险",
|
||
))
|
||
def check_sec_004(sql_clean: str, sql_upper: str, report: AuditReportV2, rule: AuditRuleV2) -> None:
|
||
"""禁止注释符。"""
|
||
if "--" in sql_clean or "/*" 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,
|
||
))
|
||
|
||
|
||
@register_rule(AuditRuleV2(
|
||
code="SEC-005",
|
||
name="禁止多语句（分号分隔）",
|
||
category=RuleCategory.SECURITY,
|
||
severity=Severity.ERROR,
|
||
db_type="all",
|
||
description="禁止单条 SQL 中包含多条语句",
|
||
message_template="禁止提交多条 SQL 语句（分号分隔），请拆分为多个工单",
|
||
))
|
||
def check_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,
|
||
))
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 主引擎类
|
||
# ──────────────────────────────────────────────
|
||
|
||
|
||
class SQLAuditEngineV2:
|
||
"""SQL 审核引擎 V2 — 可配置规则引擎。
|
||
|
||
从数据库加载规则配置，支持动态启用/禁用/配置。
|
||
向后兼容旧版 AuditReport 格式。
|
||
|
||
Attributes:
|
||
db_session: 数据库会话，用于加载规则
|
||
env: 运行环境 (development/production)
|
||
_rules_cache: 缓存加载的规则列表
|
||
"""
|
||
|
||
def __init__(self, db_session: Optional[AsyncSession] = None, env: str = "development") -> None:
|
||
"""初始化审核引擎。
|
||
|
||
Args:
|
||
db_session: 数据库会话，用于加载规则配置
|
||
env: 运行环境，影响部分规则的严格程度
|
||
"""
|
||
self.db_session = db_session
|
||
self.env = env
|
||
self._rules_cache: Optional[List[Tuple[AuditRuleV2, RuleCheckFunc]]] = None
|
||
|
||
async def _load_rules(self, db_type: str = "all") -> List[Tuple[AuditRuleV2, RuleCheckFunc]]:
|
||
"""从数据库加载启用的规则，并与注册表合并。
|
||
|
||
如果数据库中不存在某条规则，则使用注册表中的默认配置。
|
||
如果提供了 db_session，优先从数据库加载；否则直接使用注册表的规则。
|
||
|
||
Args:
|
||
db_type: 数据库类型过滤
|
||
Returns:
|
||
规则定义和检查函数的元组列表
|
||
"""
|
||
# 如果有缓存，直接返回
|
||
if self._rules_cache is not None:
|
||
return self._filter_by_db_type(self._rules_cache, db_type)
|
||
|
||
results: List[Tuple[AuditRuleV2, RuleCheckFunc]] = []
|
||
|
||
if self.db_session is not None:
|
||
try:
|
||
from app.models.audit_rule import AuditRule # 延迟导入避免循环依赖
|
||
stmt = select(AuditRule).where(AuditRule.is_enabled == True)
|
||
db_result = await self.db_session.execute(stmt)
|
||
db_rules = db_result.scalars().all()
|
||
|
||
# 将数据库规则映射到 code
|
||
db_rules_by_name: Dict[str, AuditRule] = {}
|
||
for dr in db_rules:
|
||
db_rules_by_name[dr.name] = dr
|
||
|
||
for code, (default_rule, func) in _RULE_REGISTRY.items():
|
||
db_rule = db_rules_by_name.get(default_rule.name)
|
||
if db_rule is not None:
|
||
# 使用数据库中的配置（但保留 code 和检查函数）
|
||
merged_rule = AuditRuleV2(
|
||
code=code,
|
||
name=db_rule.name,
|
||
category=RuleCategory(db_rule.rule_type) if db_rule.rule_type in {e.value for e in RuleCategory} else default_rule.category,
|
||
severity=Severity(db_rule.severity) if db_rule.severity in {e.value for e in Severity} else default_rule.severity,
|
||
db_type=db_rule.db_type or default_rule.db_type,
|
||
description=db_rule.description or default_rule.description,
|
||
is_enabled=db_rule.is_enabled,
|
||
config=default_rule.config,
|
||
pattern=db_rule.pattern or default_rule.pattern,
|
||
message_template=db_rule.message or default_rule.message_template,
|
||
)
|
||
if merged_rule.is_enabled:
|
||
results.append((merged_rule, func))
|
||
else:
|
||
# 数据库中无此规则，使用默认
|
||
if default_rule.is_enabled:
|
||
results.append((default_rule, func))
|
||
except Exception:
|
||
# 数据库加载失败，回退到注册表
|
||
for code, (rule, func) in _RULE_REGISTRY.items():
|
||
if rule.is_enabled:
|
||
results.append((rule, func))
|
||
else:
|
||
# 无数据库会话，使用注册表
|
||
for code, (rule, func) in _RULE_REGISTRY.items():
|
||
if rule.is_enabled:
|
||
results.append((rule, func))
|
||
|
||
self._rules_cache = results
|
||
return self._filter_by_db_type(results, db_type)
|
||
|
||
def _filter_by_db_type(
|
||
self, rules: List[Tuple[AuditRuleV2, RuleCheckFunc]], db_type: str
|
||
) -> List[Tuple[AuditRuleV2, RuleCheckFunc]]:
|
||
"""按数据库类型过滤规则。"""
|
||
if db_type == "all":
|
||
return rules
|
||
return [(r, f) for r, f in rules if r.db_type in ("all", db_type)]
|
||
|
||
async def audit(self, sql: str, db_type: str = "mysql") -> AuditReportV2:
|
||
"""执行 SQL 审核。
|
||
|
||
根据 SQL 语句类型，匹配对应的审核规则进行检查。
|
||
|
||
Args:
|
||
sql: 待审核的 SQL 语句
|
||
db_type: 目标数据库类型 (mysql/postgres/redis)
|
||
|
||
Returns:
|
||
AuditReportV2 审核报告
|
||
"""
|
||
sql_clean = sql.strip()
|
||
sql_upper = sql_clean.upper()
|
||
|
||
report = AuditReportV2(sql=sql_clean, db_type=db_type)
|
||
|
||
# 加载规则
|
||
rules = await self._load_rules(db_type)
|
||
|
||
# 执行每条规则的检查
|
||
for rule, check_func in rules:
|
||
try:
|
||
check_func(sql_clean, sql_upper, report, rule)
|
||
except Exception:
|
||
# 规则检查失败不影响整体审核
|
||
continue
|
||
|
||
return report
|
||
|
||
def audit_sync(self, sql: str, db_type: str = "mysql") -> AuditReportV2:
|
||
"""同步版本的审核（用于不需要数据库加载规则的场景）。
|
||
|
||
直接从注册表加载规则，不访问数据库。
|
||
|
||
Args:
|
||
sql: 待审核的 SQL 语句
|
||
db_type: 目标数据库类型
|
||
|
||
Returns:
|
||
AuditReportV2 审核报告
|
||
"""
|
||
sql_clean = sql.strip()
|
||
sql_upper = sql_clean.upper()
|
||
report = AuditReportV2(sql=sql_clean, db_type=db_type)
|
||
|
||
# 直接使用注册表的规则（不访问数据库）
|
||
for code, (rule, func) in _RULE_REGISTRY.items():
|
||
if not rule.is_enabled:
|
||
continue
|
||
if rule.db_type not in ("all", db_type):
|
||
continue
|
||
try:
|
||
func(sql_clean, sql_upper, report, rule)
|
||
except Exception:
|
||
continue
|
||
|
||
return report
|
||
|
||
async def get_rule_list(self) -> List[AuditRuleV2]:
|
||
"""获取当前引擎中所有已注册规则的列表。
|
||
|
||
Returns:
|
||
规则定义列表
|
||
"""
|
||
rules = await self._load_rules("all")
|
||
return [r for r, _ in rules]
|
||
|
||
def invalidate_cache(self) -> None:
|
||
"""清除规则缓存，下次审核重新加载。"""
|
||
self._rules_cache = None
|
||
|
||
|
||
# ── 延迟导入各数据库专用规则（副作用：注册到 _RULE_REGISTRY）────
|
||
# 放在文件末尾以避免循环导入（规则模块需要 import register_rule 等）
|
||
# MySQL 规则 (62+ 条)
|
||
from app.services.mysql_audit_rules import * # noqa: F401,F403
|
||
# PostgreSQL 规则 (45+ 条)
|
||
from app.services.postgres_audit_rules import * # noqa: F401,F403
|
||
# Redis 规则 (43+ 条)
|
||
from app.services.redis_audit_rules import * # noqa: F401,F403
|
||
# MongoDB 规则 (42+ 条)
|
||
from app.services.mongodb_audit_rules import * # noqa: F401,F403
|