Data Agent 的安全风险

把自然语言接到数据库，等于把部分数据访问能力暴露给终端用户。典型风险：

风险 攻击示例 后果
SQL 注入 '; DROP TABLE bi_metric; -- 数据破坏
越权读表 拼接未授权表名 敏感数据泄露
全表扫描 无 LIMIT 的大结果集 DB 负载、OOM
写操作 UPDATE / DELETE 混入 数据篡改
口径误导 绕过 is_valid 过滤 错误决策

安全默认：假设每一层都可能被绕过，需要纵深防御。

纵深防御架构

1
2
3
4
5
6
7
8
9
flowchart TB
    Q[用户问题] --> P[元数据模板拼 SQL]
    P --> V1[层1: 语句类型]
    V1 --> V2[层2: 关键字黑名单]
    V2 --> V3[层3: 表白名单]
    V3 --> V4[层4: 语法解析]
    V4 --> E[只读账号执行]
    E --> R[层5: 结果行数兜底]
    R --> O[输出 + 审计日志]

生产路径不让 LLM 直接产出可执行 SQL，已消除最大注入面；但仍需校验模板拼接与配置错误。

层 1：仅允许 SELECT

1
2
3
4
5
def check_select_only(sql: str) -> tuple[bool, str]:
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return False, "仅允许 SELECT 语句"
    return True, ""

拒绝：WITH ... DELETE、SELECT INTO 等变体需在层 4 一并拦截。

层 2：DML/DDL 关键字黑名单

1
2
3
4
5
6
7
8
9
10
11
12
13
FORBIDDEN_KEYWORDS = [
    "INSERT", "UPDATE", "DELETE", "DROP", "TRUNCATE",
    "ALTER", "CREATE", "REPLACE", "GRANT", "REVOKE",
    "EXEC", "EXECUTE", "CALL", "LOAD", "OUTFILE",
    "INTO DUMPFILE", "INFORMATION_SCHEMA",
]

def check_forbidden_keywords(sql: str) -> tuple[bool, str]:
    upper = sql.upper()
    for kw in FORBIDDEN_KEYWORDS:
        if kw in upper:
            return False, f"禁止关键字: {kw}"
    return True, ""

注意：INFORMATION_SCHEMA 拦截防止通过元数据探库。

层 3：表白名单

元数据表 bi_table（虚构）：

1
2
3
SELECT table_code, table_name_cn, is_whitelisted
FROM bi_table
WHERE is_whitelisted = 1;
table_code table_name_cn is_whitelisted
dws_sales_metric_wide 销售指标宽表 1
dim_region 区域维表 1
ods_order_raw 订单原始表 0

校验逻辑：

1
2
3
4
5
6
def check_table_whitelist(sql: str, allowed: set[str]) -> tuple[bool, str]:
    tables = extract_tables(sql)  # sqlparse
    for t in tables:
        if t not in allowed:
            return False, f"表 {t} 不在白名单"
    return True, ""

规则：SQL 中的每个物理表必须 ∈ bi_table.is_whitelisted = 1。

层 4：sqlparse 语法校验

1
2
3
4
5
6
7
8
9
10
import sqlparse

def check_syntax(sql: str) -> tuple[bool, str]:
    statements = sqlparse.parse(sql)
    if len(statements) != 1:
        return False, "不允许多语句"
    stmt = statements[0]
    if stmt.get_type() != "SELECT":
        return False, "非 SELECT 语句"
    return True, ""

拦截：

  • 多语句：SELECT 1; SELECT 2
  • 注释注入：SELECT/**/1
  • 子查询中的 UNION 接危险语句（需递归检查）

validate_sql 统一入口

1
2
3
4
5
6
7
8
9
10
11
12
def validate_sql(sql: str) -> dict:
    checks = [
        ("select_only", check_select_only),
        ("forbidden_kw", check_forbidden_keywords),
        ("whitelist", lambda s: check_table_whitelist(s, load_whitelist())),
        ("syntax", check_syntax),
    ]
    for name, fn in checks:
        ok, msg = fn(sql)
        if not ok:
            return {"valid": False, "failed_check": name, "message": msg}
    return {"valid": True}

失败时不执行，返回可读错误给前端，并记安全日志。

攻击示例与防护结果

输入场景 生成的 SQL（若未防护） 拦截层
用户问句含 '; DROP 模板通常不拼接原文 层 1（无 SELECT）
配置篡改表名 FROM ods_order_raw 层 3 白名单
手工改 API 传 SQL DELETE FROM bi_metric 层 1 + 层 2
多语句注入 SELECT 1; UPDATE... 层 4
子查询越权 FROM (SELECT * FROM secret) 层 3

默认过滤条件

取数逻辑中的 default_filter 必须并入 WHERE，不可被用户覆盖：

1
2
3
4
5
6
7
-- 元数据配置
default_filter = "is_valid = '1'"

-- 最终 SQL 片段
WHERE index_name = '销售收入'
  AND region_name = '华东区'
  AND is_valid = '1'

虚构宽表 dws_sales_metric_wide 中，is_valid = '0' 为作废记录，跳过会导致指标偏大。

LIMIT 与结果兜底

1
2
3
4
5
6
7
MAX_ROWS = 1000

def enforce_limit(sql: str) -> str:
    if "LIMIT" not in sql.upper():
        return sql.rstrip(";") + f" LIMIT {MAX_ROWS}"
    # 解析已有 LIMIT，若超过上限则改写
    return cap_limit(sql, MAX_ROWS)
兜底策略 说明
强制 LIMIT 默认 1000，可配置
超时 执行超过 30s 杀查询
空结果 返回友好提示 + 检查清单
超大数值 科学计数或单位换算展示

数据库账号权限

1
2
3
4
5
-- 虚构：只读账号
CREATE USER 'data_agent_ro'@'%' IDENTIFIED BY '***';
GRANT SELECT ON analytics.dws_sales_metric_wide TO 'data_agent_ro'@'%';
GRANT SELECT ON analytics.dim_region TO 'data_agent_ro'@'%';
-- 禁止 WRITE、禁止其他库

应用层白名单 + DB 层权限双保险。

审计日志

1
2
3
4
5
6
7
8
9
10
11
CREATE TABLE sql_audit_log (
  id            BIGINT AUTO_INCREMENT PRIMARY KEY,
  user_id       VARCHAR(64),
  session_id    VARCHAR(64),
  question      TEXT,
  sql_text      TEXT,
  validation    JSON,
  row_count     INT,
  duration_ms   INT,
  created_at    DATETIME DEFAULT CURRENT_TIMESTAMP
);

安全事件单独告警：failed_check 非空、同一用户 1 分钟内 10 次校验失败。

敏感字段脱敏（可选）

若宽表含手机号、身份证等（本专题宽表为指标汇总，通常不需要），在 explain_result 层做列级脱敏规则。

安全校验流程图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
sequenceDiagram
    participant API as /ask
    participant V as validate_sql
    participant DB as MySQL RO
    participant L as audit_log

    API->>V: SQL from build_sql
    V->>V: 四层检查
    alt 校验失败
        V-->>API: valid=false
        API->>L: 记录安全事件
        API-->>API: 拒绝执行
    else 校验通过
        V-->>API: valid=true
        API->>DB: execute + LIMIT
        DB-->>API: rows
        API->>L: 正常审计
    end

本章小结

  • 安全靠纵深：模板化 SQL + 四层校验 + 只读账号 + LIMIT 兜底
  • 表白名单以 bi_table 为唯一来源，禁止运行时拼未知表
  • default_filter（如 is_valid = '1'）是数据质量防线，不可省略
  • 所有 SQL 执行应审计，校验失败应告警

思考题

  1. 若业务强烈要求支持 WITH 公用表表达式，白名单校验应如何扩展？
  2. LIMIT 1000 会不会导致用户误以为「总数就是 1000」？产品如何提示？
  3. 只读账号仍可能读到敏感汇总，还需哪些业务层权限控制？

系列导航

篇 主题
01 架构全景
02 Text-to-SQL
03 RAG 指标库
04 多轮澄清
05 本篇
06 Vue3 工作台
07 演示验证
08 反馈闭环

← 返回专题首页