当前位置：首页 > news >正文

智能SQL审核系统设计：用LLM检测危险DDL并生成安全替代方案

news 2026/7/9 0:51:14

智能SQL审核系统设计：用LLM检测危险DDL并生成安全替代方案

一、一个DROP TABLE引发的凌晨紧急回滚

凌晨 2:15，运维群里弹出一条消息："有人在生产库执行了ALTER TABLE orders DROP COLUMN total_amount，线上的订单金额全部变 NULL 了。"回滚 Binlog、恢复备份、补数据、修复下游任务——整个团队忙到天亮。

事后复盘发现，这条 DDL 来自一个"看起来无害"的 JIRA 工单上的 SQL 脚本，由一位初级工程师编写。问题不是工程师的水平，而是缺少一个自动化的 DDL 审核门禁——如果这条 SQL 在到达数据库之前就被拦截并提示风险，整个事故就不会发生。

传统的 SQL 审核依赖规则引擎（如 Yearning、Inception），通过硬编码的规则表检测危险关键字（DROP、TRUNCATE）和索引缺失。但规则引擎有三个天生短板：无法理解业务语义（不知道total_amount是核心字段）、无法评估影响范围（不知道这个删除会影响多少下游）、无法生成替代方案（只能拒绝，不能建议）。

LLM 的引入将 SQL 审核从"模式匹配"升级到"语义理解"——不仅能检测危险关键字，还能理解表结构、评估影响、并生成安全的替代方案。

二、多层审核架构：从语法检测到语义评估

flowchart TB A[DDL/DML SQL] --> B{第一层：语法解析} B -->|语法错误| R1[直接拒绝] B -->|通过| C{第二层：规则引擎} C -->|命中危险规则| R2[高风险拦截] C -->|通过| D{第三层：LLM语义审核} subgraph LLM["LLM 审核层"] D --> E[Schema 上下文注入] E --> F[风险评估] F --> G[影响范围分析] G --> H{风险等级} H -->|高风险| I[生成安全替代方案] H -->|中风险| J[人工审批 + 建议] H -->|低风险| K[自动通过] end subgraph Feedback["反馈闭环"] L[审批结果记录] --> M[规则学习] M --> C end

三层防御策略：

第一层（语法解析）：用 MySQL 解析器验证 SQL 语法正确性——这是零成本的校验
第二层（规则引擎）：基于启发式规则快速拦截已知危险模式（DROP、TRUNCATE、ALTER ... CHANGE）
第三层（LLM 语义审核）：这是本文的核心——注入完整的 Schema 上下文，让 LLM 评估 SQL 的业务风险

三层中，规则引擎仍然是第一道防线（因为它在 0ms 内完成），LLM 审核作为深度语义检查的补充（耗时 1-3 秒）。

三、完整系统实现

3.1 Schema 上下文构建

import pymysql from typing import Dict, List, Optional from dataclasses import dataclass import json @dataclass class ColumnInfo: name: str data_type: str nullable: bool default_value: Optional[str] comment: str in_indexes: List[str] @dataclass class TableContext: """包含完整的表结构信息用于 LLM 审核""" table_name: str engine: str row_count_estimate: int columns: List[ColumnInfo] indexes: List[Dict] foreign_keys: List[Dict] dependent_views: List[str] recent_query_count: int # 近24小时查询次数 class SchemaContextFetcher: """提取目标表的完整上下文""" def __init__(self, db_config: dict): self.conn = pymysql.connect(**db_config) def fetch(self, table_name: str) -> TableContext: """提取表的所有元信息""" # 1. 基本表信息 table_info = self._get_table_info(table_name) # 2. 列信息 + 索引关联 columns = self._get_columns_with_index(table_name) # 3. 索引信息 indexes = self._get_indexes(table_name) # 4. 外键与依赖关系 foreign_keys = self._get_foreign_keys(table_name) dependent_views = self._get_dependent_views(table_name) # 5. 近24小时查询频率（来自 Performance Schema） recent_queries = self._get_recent_query_count(table_name) return TableContext( table_name=table_name, engine=table_info.get('ENGINE', ''), row_count_estimate=table_info.get('TABLE_ROWS', 0), columns=columns, indexes=indexes, foreign_keys=foreign_keys, dependent_views=dependent_views, recent_query_count=recent_queries ) def _get_columns_with_index(self, table_name: str) -> List[ColumnInfo]: """获取列信息并关联索引""" cursor = self.conn.cursor(pymysql.cursors.DictCursor) # 列信息 cursor.execute(f"SHOW FULL COLUMNS FROM `{table_name}`") columns = cursor.fetchall() # 索引信息（哪些列在索引中） cursor.execute(f"SHOW INDEX FROM `{table_name}`") index_rows = cursor.fetchall() # 构建列名 → 索引列表的映射 col_indexes = {} for row in index_rows: col_name = row['Column_name'] if col_name not in col_indexes: col_indexes[col_name] = [] col_indexes[col_name].append(row['Key_name']) result = [] for col in columns: result.append(ColumnInfo( name=col['Field'], data_type=col['Type'], nullable=col['Null'] == 'YES', default_value=col.get('Default'), comment=col.get('Comment', ''), in_indexes=col_indexes.get(col['Field'], []) )) return result def _get_foreign_keys(self, table_name: str) -> List[Dict]: """获取外键依赖关系""" cursor = self.conn.cursor(pymysql.cursors.DictCursor) cursor.execute(""" SELECT CONSTRAINT_NAME, COLUMN_NAME, REFERENCED_TABLE_NAME, REFERENCED_COLUMN_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND REFERENCED_TABLE_NAME IS NOT NULL """, (table_name,)) return cursor.fetchall()

3.2 LLM SQL 审核引擎

from typing import Tuple, Optional class SQLAuditor: """基于 LLM 的 SQL 语义审核器""" AUDIT_PROMPT = """你是一个 MySQL DBA 专家。请审核以下 SQL 语句，评估其安全风险。 ## 目标表上下文 {table_context} ## SQL 语句 {sql_statement} ## 审核要求 请从以下维度评估风险，并以 JSON 格式返回结果： 1. **risk_level**: 风险等级 (HIGH / MEDIUM / LOW) 2. **primary_risk**: 主要风险描述（如果安全则填"无"） 3. **impact_analysis**: - 影响行数预估 - 影响的下游依赖（视图、外键表） - 是否会影响业务核心流程 - 回滚难度评估 4. **safer_alternative**: 如果有风险，提供更安全的替代方案 SQL（如果安全则填空字符串） 5. **pre_execution_steps**: 建议执行前的准备步骤（如备份、从库验证） 6. **rollback_plan**: 如果出现问题，回滚方案 返回格式（严格 JSON）： {{ "risk_level": "HIGH|MEDIUM|LOW", "primary_risk": "描述", "impact_analysis": {{ "estimated_rows": 0, "dependent_objects": [], "business_critical": true, "rollback_difficulty": "EASY|MEDIUM|HARD|IMPOSSIBLE" }}, "safer_alternative": "SQL 或空字符串", "pre_execution_steps": ["步骤1", "步骤2"], "rollback_plan": "回滚方案" }}""" def __init__(self, schema_fetcher: SchemaContextFetcher, llm_client): self.schema_fetcher = schema_fetcher self.llm = llm_client def audit(self, sql: str) -> Dict: """审核 SQL 语句""" # 1. 从 SQL 中提取目标表名 tables = self._extract_table_names(sql) # 2. 获取所有目标表的 Schema 上下文 context_parts = [] for table in tables: ctx = self.schema_fetcher.fetch(table) context_parts.append(self._format_context(ctx)) full_context = "\n\n".join(context_parts) # 3. 调用 LLM 审核 prompt = self.AUDIT_PROMPT.format( table_context=full_context, sql_statement=sql ) response = self.llm.chat( system="你是一个专业的 MySQL DBA 审核专家，请严格按照 JSON 格式返回审核结果。", user_message=prompt, temperature=0.1 ) # 4. 解析结果 result = self._parse_response(response) return result def _format_context(self, ctx: TableContext) -> str: """将表上下文格式化为 LLM 友好的文本""" lines = [ f"表名: {ctx.table_name}", f"引擎: {ctx.engine}", f"预估行数: {ctx.row_count_estimate:,}", f"近24小时查询次数: {ctx.recent_query_count}", "", "列信息:", ] for col in ctx.columns: index_note = f" [索引: {', '.join(col.in_indexes)}]" if col.in_indexes else "" nullable = "NULL" if col.nullable else "NOT NULL" lines.append(f" - {col.name} ({col.data_type}, {nullable}){index_note} -- {col.comment}") if ctx.foreign_keys: lines.append("\n外键依赖:") for fk in ctx.foreign_keys: lines.append( f" - {fk['COLUMN_NAME']} → {fk['REFERENCED_TABLE_NAME']}.{fk['REFERENCED_COLUMN_NAME']}" ) if ctx.dependent_views: lines.append(f"\n依赖视图: {', '.join(ctx.dependent_views)}") return "\n".join(lines) def _extract_table_names(self, sql: str) -> List[str]: """从 SQL 中提取表名（简化实现）""" import re patterns = [ r'(?:ALTER|DROP|TRUNCATE|RENAME)\s+TABLE\s+(?:IF\s+EXISTS\s+)?`?(\w+)`?', r'(?:FROM|JOIN|UPDATE|INTO)\s+`?(\w+)`?', ] tables = [] for pattern in patterns: matches = re.findall(pattern, sql, re.IGNORECASE) tables.extend(matches) return list(set(tables)) def _parse_response(self, response: str) -> Dict: """解析 LLM 返回的 JSON""" import re # 提取 JSON 块 json_match = re.search(r'```json\s*(.*?)```', response, re.DOTALL) if json_match: response = json_match.group(1) try: return json.loads(response) except json.JSONDecodeError: # 尝试修复常见错误 # 移除可能的注释和尾随逗号 cleaned = re.sub(r'//.*', '', response) cleaned = re.sub(r',\s*}', '}', cleaned) cleaned = re.sub(r',\s*]', ']', cleaned) return json.loads(cleaned)