"""SQL 审核引擎 V2 规则种子数据。 包含 40+ 条覆盖 DDL/DML/DQL/命名规范/性能/安全等多个维度的审核规则。 提供 seed_all_rules(db) 函数用于一次性初始化所有规则到数据库。 """ from typing import Any, Dict, List from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.models.audit_rule import AuditRule # ── DDL 规则种子 ────────────────────────────────────────────── DDL_RULES: List[Dict[str, Any]] = [ { "name": "建表必须包含主键", "description": "CREATE TABLE 语句必须包含 PRIMARY KEY 定义,确保每行数据唯一可标识", "rule_type": "DDL", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "建表语句缺少主键定义,请添加 PRIMARY KEY", }, { "name": "建表必须添加表注释", "description": "CREATE TABLE 语句必须包含 COMMENT 表注释,便于维护和理解表用途", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "建表时必须添加表注释 (COMMENT='...')", }, { "name": "建表字段必须有注释", "description": "CREATE TABLE 中的每个字段都应包含 COMMENT 注释", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "字段缺少注释,请为每个字段添加 COMMENT", }, { "name": "禁止 DROP TABLE", "description": "生产环境禁止执行 DROP TABLE 操作,数据将永久丢失", "rule_type": "DDL", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "禁止执行 DROP TABLE 操作,数据将永久丢失", }, { "name": "禁止 TRUNCATE TABLE", "description": "禁止执行 TRUNCATE TABLE 操作,数据不可回滚", "rule_type": "DDL", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "禁止执行 TRUNCATE TABLE 操作,数据不可回滚", }, { "name": "单表索引数量不得超过20个", "description": "单表索引数量过多会影响写入性能", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": "20", "message": "单表索引数量超过20个,会影响写入性能", }, { "name": "单索引字段数不得超过8个", "description": "单索引包含字段数过多会降低索引效率", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": "8", "message": "单索引包含字段数超过8个,请优化索引设计", }, { "name": "禁止使用 FLOAT/DOUBLE 类型", "description": "浮点类型精度丢失,应使用 DECIMAL", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "禁止使用 FLOAT/DOUBLE 类型,建议使用 DECIMAL 避免精度丢失", }, { "name": "CHAR 类型长度不得超过255", "description": "CHAR 类型长度过大会浪费存储空间,超出请使用 VARCHAR", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": "255", "message": "CHAR 类型长度不得超过255,超出请使用 VARCHAR", }, { "name": "表名长度不得超过64字符", "description": "过长的表名不利于维护和使用", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": "64", "message": "表名长度超过64个字符,请缩短表名", }, { "name": "禁止外键约束", "description": "外键约束影响性能,推荐在应用层处理关联关系", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "禁止使用 FOREIGN KEY 约束,建议在应用层维护数据一致性", }, { "name": "必须包含 create_time/update_time 字段", "description": "建表必须包含 create_time 和 update_time 审计字段,便于追踪数据变更", "rule_type": "DDL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "建表缺少 create_time 或 update_time 审计字段", }, ] # ── DML 规则种子 ────────────────────────────────────────────── DML_RULES: List[Dict[str, Any]] = [ { "name": "UPDATE/DELETE 必须包含 WHERE 条件", "description": "UPDATE/DELETE 缺少 WHERE 将影响全表数据,可能导致严重事故", "rule_type": "DML", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "UPDATE/DELETE 语句必须包含 WHERE 条件", }, { "name": "单次 INSERT 最多500行", "description": "单次 INSERT 行数过多会导致事务过大,影响性能和复制延迟", "rule_type": "DML", "severity": "warning", "db_type": "all", "pattern": None, "value": "500", "message": "单次 INSERT 最多插入500行", }, { "name": "UPDATE/DELETE 禁止带 LIMIT", "description": "UPDATE/DELETE 使用 LIMIT 可能导致部分数据变更,数据不一致", "rule_type": "DML", "severity": "warning", "db_type": "mysql", "pattern": None, "value": None, "message": "UPDATE/DELETE 语句禁止使用 LIMIT", }, { "name": "禁止 INSERT 不指定字段列表", "description": "INSERT 必须显式指定字段列表,避免表结构变更导致错误", "rule_type": "DML", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "INSERT 语句必须显式指定字段列表", }, { "name": "禁止批量 UPDATE 不带主键条件", "description": "批量 UPDATE 必须带主键 WHERE 条件,防止误更新大量数据", "rule_type": "DML", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "批量 UPDATE 必须包含主键条件", }, { "name": "DELETE 必须有 LIMIT(除非带主键条件)", "description": "DELETE 必须带 LIMIT 或主键 WHERE 条件,防止误删大量数据", "rule_type": "DML", "severity": "warning", "db_type": "mysql", "pattern": None, "value": None, "message": "DELETE 建议添加 LIMIT 或使用主键条件", }, { "name": "禁止 REPLACE INTO", "description": "REPLACE INTO 会先删除再插入,风险较高", "rule_type": "DML", "severity": "error", "db_type": "mysql", "pattern": None, "value": None, "message": "禁止使用 REPLACE INTO,请使用 INSERT ... ON DUPLICATE KEY UPDATE 替代", }, { "name": "INSERT 必须指定字段列表", "description": "INSERT 语句必须显式指定字段列表,避免表结构变更导致错误", "rule_type": "DML", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "INSERT 语句必须显式指定字段列表", }, ] # ── DQL 规则种子 ────────────────────────────────────────────── DQL_RULES: List[Dict[str, Any]] = [ { "name": "禁止 SELECT *", "description": "SELECT * 返回不必要的列,影响网络传输和内存使用", "rule_type": "DQL", "severity": "info", "db_type": "all", "pattern": None, "value": None, "message": "禁止使用 SELECT *,请明确指定需要的字段", }, { "name": "SELECT 必须有 LIMIT", "description": "SELECT 查询应添加 LIMIT 限制返回行数,防止返回过多数据", "rule_type": "DQL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "SELECT 查询缺少 LIMIT,建议添加 LIMIT 防止返回过多数据", }, { "name": "禁止在 WHERE 中使用函数", "description": "WHERE 中使用函数会导致索引失效,性能下降", "rule_type": "DQL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "WHERE 子句中使用函数会导致索引失效", }, { "name": "禁止隐式类型转换", "description": "不同类型比较会导致索引失效,请确保比较双方类型一致", "rule_type": "DQL", "severity": "warning", "db_type": "all", "pattern": None, "value": None, "message": "WHERE 条件中存在隐式类型转换风险", }, { "name": "ORDER BY 必须使用索引字段", "description": "ORDER BY 非索引字段会导致文件排序(filesort),影响性能", "rule_type": "DQL", "severity": "info", "db_type": "all", "pattern": None, "value": None, "message": "ORDER BY 字段建议添加索引,避免 filesort 性能问题", }, ] # ── 命名规范规则种子 ────────────────────────────────────────── NAMING_RULES: List[Dict[str, Any]] = [ { "name": "表名必须小写下划线命名", "description": "表名应使用小写下划线格式 (snake_case),如 user_info", "rule_type": "naming", "severity": "warning", "db_type": "all", "pattern": r"^[a-z][a-z0-9_]*$", "value": None, "message": "表名不符合小写下划线命名规范", }, { "name": "字段名必须小写下划线命名", "description": "字段名应使用小写下划线格式 (snake_case),如 user_name", "rule_type": "naming", "severity": "warning", "db_type": "all", "pattern": r"^[a-z][a-z0-9_]*$", "value": None, "message": "字段名不符合小写下划线命名规范", }, { "name": "索引名必须以 idx_ 或 uk_ 开头", "description": "索引命名规范:普通索引 idx_,唯一索引 uk_,如 idx_user_name", "rule_type": "naming", "severity": "info", "db_type": "all", "pattern": r"^(idx_|uk_)", "value": None, "message": "索引名应以 idx_ 或 uk_ 开头", }, { "name": "主键名必须为 id 或 pk_ 开头", "description": "主键字段命名规范,建议使用 id 或 pk_ 前缀", "rule_type": "naming", "severity": "info", "db_type": "all", "pattern": r"^(id|pk_)", "value": None, "message": "主键名建议使用 'id' 或 'pk_' 前缀", }, { "name": "布尔字段以 is_/has_ 开头", "description": "布尔类型字段应以 is_ 或 has_ 前缀命名,如 is_active", "rule_type": "naming", "severity": "info", "db_type": "all", "pattern": r"^(is_|has_)", "value": None, "message": "布尔字段建议以 is_ 或 has_ 开头命名", }, ] # ── 性能规则种子 ────────────────────────────────────────────── PERF_RULES: List[Dict[str, Any]] = [ { "name": "禁止大表全表扫描", "description": "大表(>10000行)查询缺少 WHERE 条件会导致全表扫描,严重影响性能", "rule_type": "performance", "severity": "warning", "db_type": "all", "pattern": None, "value": "10000", "message": "SELECT 查询缺少 WHERE 条件,大表下会导致全表扫描", }, { "name": "IN 子句值数量不得超过100个", "description": "IN 子句值过多会导致性能下降,建议使用临时表或 JOIN 替代", "rule_type": "performance", "severity": "warning", "db_type": "all", "pattern": None, "value": "100", "message": "IN 子句包含超过100个值,建议使用临时表或 JOIN 替代", }, { "name": "禁止 LIKE '%xxx%' 前缀模糊匹配", "description": "前缀模糊匹配无法使用索引,建议使用全文检索", "rule_type": "performance", "severity": "warning", "db_type": "all", "pattern": r"LIKE\s+'%[^']+'", "value": None, "message": "LIKE '%xxx%' 无法使用索引,建议改用全文检索或后缀索引", }, { "name": "JOIN 不得超过5个表", "description": "JOIN 表过多会导致查询计划复杂化和性能下降", "rule_type": "performance", "severity": "warning", "db_type": "all", "pattern": None, "value": "5", "message": "JOIN 涉及超过5个表,请优化查询结构", }, { "name": "禁止 ORDER BY RAND()", "description": "ORDER BY RAND() 性能极差,会导致全表扫描", "rule_type": "performance", "severity": "error", "db_type": "mysql", "pattern": r"ORDER\s+BY\s+RAND\s*\(", "value": None, "message": "禁止 ORDER BY RAND(),性能极差,建议在应用层随机采样", }, { "name": "子查询建议改为 JOIN", "description": "相关子查询性能较差,建议改为 JOIN", "rule_type": "performance", "severity": "info", "db_type": "all", "pattern": None, "value": None, "message": "WHERE 中的子查询建议改为 JOIN 以提升性能", }, { "name": "大表禁止 OFFSET 分页", "description": "大表使用 OFFSET 分页性能差,建议使用游标分页", "rule_type": "performance", "severity": "warning", "db_type": "all", "pattern": r"OFFSET\s+\d+", "value": None, "message": "使用 OFFSET 分页在大表上性能较差,建议改用游标分页", }, { "name": "禁止 SELECT DISTINCT(除非必要)", "description": "DISTINCT 需要去重操作,影响性能", "rule_type": "performance", "severity": "info", "db_type": "all", "pattern": r"SELECT\s+DISTINCT\b", "value": None, "message": "SELECT DISTINCT 影响性能,请确认是否必要", }, ] # ── 安全规则种子 ────────────────────────────────────────────── SEC_RULES: List[Dict[str, Any]] = [ { "name": "禁止查询 password/secret/token 字段", "description": "禁止在 SELECT 中查询敏感字段,防止密码等敏感信息泄露", "rule_type": "security", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "禁止在查询中返回 password/secret/token 等敏感字段", }, { "name": "禁止 SQL 中硬编码密码", "description": "SQL 中不应包含硬编码密码,防止密码泄露", "rule_type": "security", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "SQL 中发现可能的硬编码密码,禁止在 SQL 中直接写入密码", }, { "name": "禁止 UNION(SQL 注入风险)", "description": "UNION 语句存在 SQL 注入风险,禁止在工单中使用", "rule_type": "security", "severity": "error", "db_type": "all", "pattern": r"\bUNION\b", "value": None, "message": "禁止 UNION 操作,存在 SQL 注入风险", }, { "name": "禁止注释符", "description": "SQL 中不应包含 -- 或 /* */ 注释,可能存在注入风险", "rule_type": "security", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "SQL 中禁止包含注释符,可能存在注入风险", }, { "name": "禁止多语句(分号分隔)", "description": "禁止单条 SQL 中包含多条语句,防止注入攻击", "rule_type": "security", "severity": "error", "db_type": "all", "pattern": None, "value": None, "message": "禁止提交多条 SQL 语句(分号分隔),请拆分为多个工单", }, ] # ── 合并所有规则 ────────────────────────────────────────────── ALL_V2_RULES: List[Dict[str, Any]] = ( DDL_RULES + DML_RULES + DQL_RULES + NAMING_RULES + PERF_RULES + SEC_RULES ) async def seed_all_rules(db: AsyncSession) -> List[AuditRule]: """将所有 V2 规则种子数据插入数据库(跳过已存在的规则)。 Args: db: 数据库异步会话 Returns: 新创建的 AuditRule 列表 """ created: List[AuditRule] = [] for rule_dict in ALL_V2_RULES: # 检查是否已存在(按 name 去重) exists = await db.execute(select(AuditRule).where(AuditRule.name == rule_dict["name"])) if exists.scalar_one_or_none(): continue # 跳过已存在的规则 rule = AuditRule(**rule_dict) db.add(rule) created.append(rule) await db.commit() for rule in created: await db.refresh(rule) return created