把 Text2SQL 接上真实数据库：只读账号加 SQL 白名单加结果校验的安全闭环

不要用业务主库账号连 Agent。建一个只授 SELECT 的只读角色，连一个副本或只读实例。即使账号泄露，攻击者也只能读、不能改，把爆炸半径压到最小。

-- PostgreSQL 示例：只读角色
CREATE ROLE agent_ro LOGIN PASSWORD '强密码';
GRANT CONNECT ON DATABASE shop TO agent_ro;
GRANT USAGE ON SCHEMA public TO agent_ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_ro;
-- 敏感表（用户、支付）不授权，Agent 根本看不到

系统提示明确只允许查询指定表，禁止任何写操作与系统表。这一步是「软约束」：靠模型听话，不是硬保险，所以必须配合第三步的代码层白名单校验，两层一起才稳妥。提示里把允许的范围写得越具体（具体到表名、列名），模型越不容易跑偏。

SYSTEM = '''你是一个只读查询助手。
只允许生成 SELECT 语句，查询范围仅限表：orders、products、users（仅 id 与 city 两列）。
禁止 INSERT/UPDATE/DELETE/DROP，禁止查询 information_schema、payment、password 相关列。
只输出一条 SQL，不要解释。'''

def ask_sql(question):
    resp = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}])
    return resp.choices[0].message.content.strip()

再严格的提示也可能被绕过（比如用注释、Union 偷渡）。执行前用正则做硬校验：必须 SELECT 开头、只允许白名单表、禁止多语句与危险关键字。

import re
ALLOWED_TABLES = {"orders", "products", "users"}
FORBID = re.compile(r"\b(insert|update|delete|drop|alter|union|information_schema|password)\b", re.I)

def safe_sql(sql):
    s = sql.lower().replace(";", " ").strip()
    if not s.startswith("select"):
        raise ValueError("只允许 SELECT")
    if FORBID.search(s):
        raise ValueError("命中禁词或禁表")
    tables = set(re.findall(r"from\s+([a-z_]+)", s)) | set(re.findall(r"join\s+([a-z_]+)", s))
    if not tables <= ALLOWED_TABLES:
        raise ValueError(f"超出白名单表：{tables - ALLOWED_TABLES}")
    return sql

用户问「所有订单」，模型很可能漏写 LIMIT，或者写出没有上限的聚合查询。执行前统一补一个上限，结果行数再做二次封顶。这既保护内存，也避免一次返回几十万行把下游渲染或模型上下文直接撑爆。封顶值视表大小而定，宁可少返回让模型追问，也不要贪多。

def with_limit(sql, cap=200):
    if "limit" not in sql.lower():
        sql = sql.rstrip(";") + f" LIMIT {cap}"
    return sql

def run_query(sql):
    sql = safe_sql(sql)
    sql = with_limit(sql)
    cur = conn.execute(sql)
    rows = cur.fetchall()
    if len(rows) > 200:
        rows = rows[:200]        # 二次封顶，保护内存
    return rows

users 表即便只暴露 id 与 city，也可能和别处的公开信息组合出可识别个体。对手机号、邮箱、身份证这类字段，统一打码后再返回给模型或用户。脱敏要放在「返回给模型之前」：模型做聚合统计并不需要真实号码，打码后的数据足够它回答「哪个城市下单最多」这类问题。

def mask(value):
    s = str(value)
    if "@" in s:                       # 邮箱
        return s[0] + "***@" + s.split("@")[-1]
    if s.isdigit() and len(s) >= 7:    # 手机号
        return s[:3] + "****" + s[-2:]
    return s

def mask_row(row):
    return tuple(mask(v) for v in row)

把「生成 SQL 到 校验 到 加 LIMIT 到 执行 到 脱敏 到 封顶」收成一个工具，Agent 就能用自然语言查库。

def db_tool(question):
    try:
        sql = ask_sql(question)
        sql = safe_sql(sql)
        sql = with_limit(sql)
        rows = run_query(sql)
        return {"sql": sql, "rows": [mask_row(r) for r in rows], "count": len(rows)}
    except Exception as e:
        return {"error": str(e), "rows": []}

print(db_tool("上个月销售额 top 5 的商品"))

上线核对清单：
1. 账号是否只读、是否只连授权表（账号层）
2. SQL 是否过白名单校验、是否强制 LIMIT（代码层）
3. 结果是否脱敏、是否记审计日志（数据层）