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

AI 生成 SQL 差点扫全表：业务接 AI 前，必须先做执行前审计

news 2026/5/23 1:18:41

我见过一个很典型的事故苗头。

一个团队做了一个内部数据助手，产品经理可以直接用自然语言查数据。

比如输入：

帮我查一下最近 7 天支付失败的订单数量。

AI 会生成 SQL，然后后端执行查询，把结果返回给页面。

Demo 的时候效果很好。

产品经理问“昨天退款订单有多少”，AI 能答。
运营问“本周注册用户有多少”，AI 能答。
老板问“某个渠道的转化率怎么样”，AI 也能答。

大家都觉得这个东西很有用。

直到有一天，有人问了一句：

帮我看看所有用户最近的订单情况。

AI 生成了一条 SQL：

```sql
SELECT * FROM orders;
```

如果这条 SQL 真跑到生产库，轻则慢查询拖垮数据库，重则把不该展示的数据直接暴露出来。

幸好当时还在测试环境。

但这件事给团队提了一个醒：AI 生成 SQL 最大的问题，不是它会不会写 SQL，而是它写出来之后你敢不敢执行。

现在很多团队接 AI，都停在第一层：

自然语言转 SQL。
自然语言查报表。
自然语言生成接口参数。
自然语言驱动 Agent 调工具。

这些能力确实越来越强。OpenAI 的 Codex cloud 已经可以在云环境里后台处理代码任务，包括读、改、运行代码；Google 也在 Search 中引入更强模型能力，让用户可以通过提问使用 Agent 能力。AI 正在从“回答问题”走向“执行动作”。

但越能执行动作，越不能直接执行。

尤其是 SQL。

AI 生成 SQL 至少有四类风险。

第一，扫全表。

```sql
SELECT * FROM orders;
```

这类 SQL 在小数据测试库里没事，在生产库里可能直接把数据库打满。

第二，越权查敏感字段。

```sql
SELECT user_id, phone, email, address FROM users;
```

业务问题可能只是“统计用户数量”，但 AI 为了“完整”，把手机号、邮箱、地址都查出来了。

第三，生成危险操作。

```sql
DELETE FROM orders WHERE status = 'failed';
```

哪怕你只想查数据，模型也可能因为理解偏差生成写操作。

第四，缺少 LIMIT。

```sql
SELECT * FROM logs ORDER BY created_at DESC;
```

日志表、订单表、行为表，一旦没有 LIMIT，很容易出问题。

所以业务系统接 AI 生成 SQL，第一步不是优化 Prompt。

而是做一层 SQL 执行前审计器。

一、不要直接执行 AI 生成的 SQL

很多人最初会这么写：

```python
def run_ai_sql(sql: str, db):
return db.execute(sql).fetchall()
```

这段代码看起来简单，风险也很清楚：

AI 生成什么，就执行什么。

这不叫智能助手。

这叫让模型直接摸生产库。

至少应该改成：

```python
def run_ai_sql_safely(sql: str, db):
audit_result = audit_sql(sql)

if not audit_result["ok"]:
return {
"ok": False,
"stage": "sql_audit",
"message": audit_result["message"],
"risks": audit_result["risks"]
}

return {
"ok": True,
"data": db.execute(sql).fetchall()
}
```

关键不在 execute，而在 audit_sql。

二、先限制 SQL 类型：只允许 SELECT

内部数据助手的第一条规则应该是：只读。

不允许：

```sql
INSERT
UPDATE
DELETE
DROP
ALTER
TRUNCATE
CREATE
```

先写一个最基础的检查：

```python
import re

FORBIDDEN_KEYWORDS = [
"INSERT",
"UPDATE",
"DELETE",
"DROP",
"ALTER",
"TRUNCATE",
"CREATE",
"REPLACE",
"MERGE"
]

def normalize_sql(sql: str) -> str:
return re.sub(r"\s+", " ", sql.strip()).upper()

def check_readonly(sql: str) -> list[str]:
normalized = normalize_sql(sql)
risks = []

if not normalized.startswith("SELECT"):
risks.append("只允许执行 SELECT 查询")

for keyword in FORBIDDEN_KEYWORDS:
if re.search(rf"\b{keyword}\b", normalized):
risks.append(f"检测到禁止关键字：{keyword}")

return risks
```

这只是第一层。

它不能证明 SQL 安全，但能挡掉最危险的一批操作。

三、必须限制可访问表

即使只允许 SELECT，也不能让 AI 随便查所有表。

比如这些表就不应该开放给 AI：

```text
users
user_passwords
payment_cards
admin_logs
access_tokens
internal_audit_logs
```

更好的方式是维护一个白名单。

```python
ALLOWED_TABLES = {
"orders",
"order_items",
"products",
"refunds",
"channels"
}

BLOCKED_TABLES = {
"users",
"user_passwords",
"payment_cards",
"access_tokens",
"admin_logs"
}
```

然后解析 SQL 里出现的表名。

简单版本可以先用正则：

```python
def extract_tables(sql: str) -> set[str]:
pattern = r"\bFROM\s+([a-zA-Z_][a-zA-Z0-9_]*)|\bJOIN\s+([a-zA-Z_][a-zA-Z0-9_]*)"
matches = re.findall(pattern, sql, flags=re.IGNORECASE)

tables = set()
for from_table, join_table in matches:
if from_table:
tables.add(from_table.lower())
if join_table:
tables.add(join_table.lower())

return tables
```

做表级校验：

```python
def check_tables(sql: str) -> list[str]:
risks = []
tables = extract_tables(sql)

for table in tables:
if table in BLOCKED_TABLES:
risks.append(f"禁止访问敏感表：{table}")

if table not in ALLOWED_TABLES:
risks.append(f"表不在白名单中：{table}")

return risks
```

真实项目里建议用 SQL parser，而不是只靠正则。

比如 Python 可以考虑 sqlglot 这类解析库，把 SQL 解析成 AST 再检查表、字段、条件和 LIMIT。

但就算先用正则，也比直接执行强很多。

四、限制敏感字段

表能查，不代表所有字段都能查。

比如 orders 表可以开放，但这些字段可能不该直接暴露：

```text
buyer_phone
buyer_address
payment_account
invoice_title
internal_note
```

可以做字段黑名单：

```python
BLOCKED_COLUMNS = {
"phone",
"email",
"address",
"buyer_phone",
"buyer_address",
"payment_account",
"id_card",
"token",
"password"
}
```

检查 SQL 是否选择了敏感字段：

```python
def check_sensitive_columns(sql: str) -> list[str]:
risks = []
normalized = normalize_sql(sql)

for column in BLOCKED_COLUMNS:
if re.search(rf"\b{column.upper()}\b", normalized):
risks.append(f"检测到敏感字段：{column}")

return risks
```

有人可能会说：如果 AI 写 SELECT * 呢？

这就更应该拦。

```python
def check_select_star(sql: str) -> list[str]:
normalized = normalize_sql(sql)

if re.search(r"SELECT\s+\*", normalized):
return ["禁止使用 SELECT *，必须显式声明字段"]

return []
```

在生产业务里，SELECT * 是 AI SQL 的高危信号。

五、强制 LIMIT

任何 AI 生成的查询，都应该有结果数量限制。

```python
def check_limit(sql: str) -> list[str]:
normalized = normalize_sql(sql)

if " LIMIT " not in normalized:
return ["查询必须包含 LIMIT"]

return []
```

但只检查有没有 LIMIT 还不够。

如果 AI 写：

```sql
LIMIT 1000000
```

仍然危险。

可以限制最大值：

```python
def extract_limit(sql: str) -> int | None:
match = re.search(r"\bLIMIT\s+(\d+)\b", sql, flags=re.IGNORECASE)
if not match:
return None
return int(match.group(1))

def check_limit_size(sql: str, max_limit: int = 1000) -> list[str]:
limit = extract_limit(sql)

if limit is None:
return ["查询必须包含 LIMIT"]

if limit > max_limit:
return [f"LIMIT 过大：{limit}，最大允许 {max_limit}"]

return []
```

对于内部运营查询，通常 100、500、1000 就够了。

真要导出大数据，不应该走 AI 即席查询，而应该走异步导出流程。

六、必须有时间范围

很多事故都来自没有时间条件。

比如订单表：

```sql
SELECT order_id, amount FROM orders LIMIT 1000;
```

这条 SQL 有 LIMIT，也只读，但仍然不够安全。

因为它没有时间范围。

更合理的是：

```sql
SELECT order_id, amount
FROM orders
WHERE created_at >= NOW() - INTERVAL '7 days'
LIMIT 1000;
```

可以对大表强制要求时间条件。

```python
TIME_REQUIRED_TABLES = {
"orders",
"refunds",
"logs",
"user_events"
}

TIME_COLUMNS = {
"created_at",
"updated_at",
"event_time",
"paid_at"
}

def check_time_condition(sql: str) -> list[str]:
risks = []
tables = extract_tables(sql)
normalized = normalize_sql(sql)

need_time = any(table in TIME_REQUIRED_TABLES for table in tables)

if not need_time:
return risks

has_time_condition = any(column.upper() in normalized for column in TIME_COLUMNS)

if not has_time_condition:
risks.append("大表查询必须包含时间范围条件")

return risks
```

七、组合成 SQL 审计器

把前面的规则合起来：

```python
def audit_sql(sql: str) -> dict:
risks = []

risks.extend(check_readonly(sql))
risks.extend(check_tables(sql))
risks.extend(check_sensitive_columns(sql))
risks.extend(check_select_star(sql))
risks.extend(check_limit_size(sql, max_limit=1000))
risks.extend(check_time_condition(sql))

return {
"ok": len(risks) == 0,
"risks": risks,
"message": "SQL 审计通过" if not risks else "SQL 审计未通过"
}
```

测试一下：

```python
sql = "SELECT * FROM orders;"

print(audit_sql(sql))
```

可能输出：

```python
{
"ok": False,
"risks": [
"禁止使用 SELECT *，必须显式声明字段",
"查询必须包含 LIMIT",
"大表查询必须包含时间范围条件"
],
"message": "SQL 审计未通过"
}
```

八、审计失败后，不要直接报错给用户

如果 SQL 被拦截，不要把内部规则生硬地丢给用户。

可以返回可理解的提示：

```python
def build_user_friendly_message(risks: list[str]) -> str:
if any("SELECT *" in risk for risk in risks):
return "查询范围过宽，请指定要查看的字段。"

if any("LIMIT" in risk for risk in risks):
return "查询结果可能过多，请缩小范围或增加数量限制。"

if any("时间范围" in risk for risk in risks):
return "请补充时间范围，例如最近 7 天、上个月或某个日期区间。"

if any("敏感" in risk or "禁止访问" in risk for risk in risks):
return "该查询涉及敏感数据，无法直接执行。请改为查看汇总数据。"

return "当前查询不满足安全规则，请调整问题后重试。"
```

完整返回：

```python
def handle_ai_sql(sql: str, db):
audit_result = audit_sql(sql)

if not audit_result["ok"]:
return {
"ok": False,
"message": build_user_friendly_message(audit_result["risks"]),
"audit_risks": audit_result["risks"]
}

rows = db.execute(sql).fetchall()

return {
"ok": True,
"data": rows
}
```

九、执行高风险 SQL 前必须人工确认

有些 SQL 不一定要完全禁止，但需要人工确认。

比如：

查询数据量接近上限。
涉及多个大表 JOIN。
包含财务金额字段。
包含用户维度聚合。
执行计划成本过高。

可以给审计结果增加等级：

```python
def audit_sql_with_level(sql: str) -> dict:
risks = audit_sql(sql)["risks"]

if not risks:
return {
"ok": True,
"level": "safe",
"risks": []
}

high_risk_keywords = ["敏感", "禁止访问", "DELETE", "UPDATE", "DROP"]
level = "high" if any(
any(keyword in risk for keyword in high_risk_keywords)
for risk in risks
) else "medium"

return {
"ok": False,
"level": level,
"risks": risks
}
```

执行策略：

```python
def decide_next_action(audit_result: dict) -> str:
if audit_result["ok"]:
return "execute"

if audit_result["level"] == "medium":
return "ask_user_to_refine"

if audit_result["level"] == "high":
return "human_review"

return "reject"
```

十、生产环境还要加数据库层防线

应用层审计不是唯一防线。

数据库层也要配合。

至少做到：

AI 查询使用只读账号。
只读账号只能访问白名单视图。
敏感字段不在视图里。
查询超时自动取消。
单次返回行数限制。
慢查询记录日志。
高风险查询写审计记录。

比如不要让 AI 直接访问 orders 表，可以建一个视图：