"""SQL Audit Engine — rule-based SQL review for tickets. Checks SQL against configurable rules and returns structured results: - level: "error" (must fix), "warning" (should review), "info" (FYI) - rule: rule identifier - message: human-readable description """ import re from dataclasses import dataclass, field from typing import List, Optional from enum import Enum class Severity(str, Enum): ERROR = "error" WARNING = "warning" INFO = "info" @dataclass class AuditFinding: rule: str level: Severity message: str @dataclass class AuditReport: sql: str findings: List[AuditFinding] = field(default_factory=list) @property def passed(self) -> bool: return not any(f.level == Severity.ERROR for f in self.findings) @property def has_warnings(self) -> bool: return any(f.level == Severity.WARNING for f in self.findings) @property def score(self) -> int: """100 = clean, -10 per error, -5 per warning.""" s = 100 for f in self.findings: if f.level == Severity.ERROR: s -= 10 elif f.level == Severity.WARNING: s -= 5 return max(0, s) def to_dict(self) -> dict: return { "passed": self.passed, "score": self.score, "findings": [ {"rule": f.rule, "level": f.level.value, "message": f.message} for f in self.findings ], } class SQLAuditEngine: """Rule-based SQL audit engine.""" def __init__(self, env: str = "development"): self.env = env def audit(self, sql: str) -> AuditReport: report = AuditReport(sql=sql) clean = sql.strip() upper = clean.upper() # Rule 1: DDL without IF EXISTS self._check_ddl_if_exists(clean, upper, report) # Rule 2: DROP TABLE danger self._check_drop_table(upper, report) # Rule 3: UPDATE/DELETE without WHERE self._check_missing_where(clean, upper, report) # Rule 4: SELECT * usage self._check_select_star(clean, upper, report) # Rule 5: LIMIT on DML self._check_limit_dml(clean, upper, report) # Rule 6: Production env extra checks if self.env == "production": self._check_production_rules(clean, upper, report) # Rule 7: Multiple statements (semicolon) self._check_multiple_statements(clean, report) # Rule 8: Large IN clause self._check_large_in(clean, upper, report) # Rule 9: No WHERE with LIKE '%...' self._check_full_like(clean, upper, report) # Rule 10: JOIN without ON self._check_join_without_on(upper, report) return report def _check_ddl_if_exists(self, clean: str, upper: str, report: AuditReport): """ALTER/DROP statements should use IF EXISTS where applicable.""" if re.search(r"\bDROP\s+TABLE\b", upper) and "IF EXISTS" not in upper: report.findings.append(AuditFinding( rule="DDL_IF_EXISTS", level=Severity.WARNING, message="DROP TABLE 建议添加 IF EXISTS 防止表不存在时报错", )) def _check_drop_table(self, upper: str, report: AuditReport): """DROP TABLE is always dangerous in production.""" if re.search(r"\bDROP\s+TABLE\b", upper): if self.env == "production": report.findings.append(AuditFinding( rule="DROP_TABLE_PRODUCTION", level=Severity.ERROR, message="生产环境禁止 DROP TABLE 操作", )) else: report.findings.append(AuditFinding( rule="DROP_TABLE", level=Severity.WARNING, message="DROP TABLE 操作不可逆，请确认已备份", )) if re.search(r"\bTRUNCATE\b", upper): report.findings.append(AuditFinding( rule="TRUNCATE", level=Severity.ERROR if self.env == "production" else Severity.WARNING, message="TRUNCATE 会清空表数据且不可回滚", )) def _check_missing_where(self, clean: str, upper: str, report: AuditReport): """UPDATE/DELETE without WHERE clause affects all rows.""" if re.match(r"\s*UPDATE\s+", upper, re.IGNORECASE): if "WHERE" not in upper: report.findings.append(AuditFinding( rule="UPDATE_NO_WHERE", level=Severity.ERROR, message="UPDATE 缺少 WHERE 条件，将更新全表", )) if re.match(r"\s*DELETE\s+FROM\s+", upper, re.IGNORECASE): if "WHERE" not in upper: report.findings.append(AuditFinding( rule="DELETE_NO_WHERE", level=Severity.ERROR, message="DELETE 缺少 WHERE 条件，将删除全表数据", )) def _check_select_star(self, clean: str, upper: str, report: AuditReport): """SELECT * is inefficient for large tables.""" if re.search(r"SELECT\s+\*(?:\s|,|\))", upper): report.findings.append(AuditFinding( rule="SELECT_STAR", level=Severity.INFO, message="SELECT * 可能返回过多列，建议明确指定字段", )) def _check_limit_dml(self, clean: str, upper: str, report: AuditReport): """UPDATE/DELETE with LIMIT is MySQL-specific and risky.""" for kw in ("UPDATE", "DELETE"): if re.match(rf"\s*{kw}\b", upper): if "LIMIT" in upper: report.findings.append(AuditFinding( rule="DML_WITH_LIMIT", level=Severity.WARNING, message=f"{kw} 使用 LIMIT 可能导致部分数据变更，请确认意图", )) def _check_production_rules(self, clean: str, upper: str, report: AuditReport): """Extra strict rules for production.""" if re.search(r"\bALTER\s+TABLE\b", upper): report.findings.append(AuditFinding( rule="ALTER_TABLE_PRODUCTION", level=Severity.ERROR, message="生产环境 ALTER TABLE 需 DBA 特别审批", )) if re.search(r"\bCREATE\s+(TABLE|INDEX)\b", upper): report.findings.append(AuditFinding( rule="CREATE_DDL_PRODUCTION", level=Severity.WARNING, message="生产环境创建对象需确认影响范围", )) def _check_multiple_statements(self, clean: str, report: AuditReport): """Multiple SQL statements separated by semicolons.""" # Count semicolons not at the end stripped = clean.rstrip(";").strip() if ";" in stripped: report.findings.append(AuditFinding( rule="MULTI_STATEMENT", level=Severity.ERROR, message="单次提交不支持多条 SQL 语句（分号分隔），请拆分为多个工单", )) def _check_large_in(self, clean: str, upper: str, report: AuditReport): """IN clause with many values is slow.""" in_matches = re.findall(r"\bIN\s*\(([^)]+)\)", upper) for m in in_matches: count = m.count(",") + 1 if count > 100: report.findings.append(AuditFinding( rule="LARGE_IN", level=Severity.WARNING, message=f"IN 子句包含 {count} 个值，建议使用临时表或 JOIN 替代", )) def _check_full_like(self, clean: str, upper: str, report: AuditReport): """LIKE '%...' cannot use index.""" if re.search(r"\bLIKE\s+'%[^']*'", clean, re.IGNORECASE): report.findings.append(AuditFinding( rule="FULL_LIKE", level=Severity.INFO, message="LIKE '%...' 无法使用索引，大数据量下性能较差", )) def _check_join_without_on(self, upper: str, report: AuditReport): """JOIN without ON produces cartesian product.""" # Match JOIN but not followed by ON (within reasonable distance) if re.search(r"\bJOIN\b", upper) and "ON" not in upper: # Could be NATURAL JOIN, allow that if not re.search(r"\bNATURAL\s+JOIN\b", upper): report.findings.append(AuditFinding( rule="JOIN_NO_ON", level=Severity.WARNING, message="JOIN 缺少 ON 条件，可能产生笛卡尔积", ))