logan
66b09e4d85
Phase 3.1: Instance password AES-256-GCM encryption Phase 3.2: SQL audit rule engine (10 rules) Phase 3.3: Auto-audit on ticket creation Phase 3.4: Operation audit logs + GET /audit/logs Tests: 44 -> 85 (41 new), Frontend: AuditLogsPage
235 lines
8.6 KiB
Python
235 lines
8.6 KiB
Python
"""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 条件，可能产生笛卡尔积",
|
|
))
|