数据库 Agent 安全执行：SQL 白名单、审批与只读账号

数据库 Agent 安全执行SQL 白名单、审批与只读账号数据库 Agent 不应拿到可随意写入的高权限连接。解析 SQL、限制语句类型、人工审批高风险动作并用只读账号执行查询才能把模型错误限制在可控范围。1. 把危险 SQL 当成必须覆盖的测试输入SQL 优化 Agent 可以调用EXPLAIN读取执行计划并生成候选索引但它不应直接在目标数据库执行ALTER TABLE。收益要在隔离数据副本上比较执行时间、扫描行数和写入开销再由人工审批变更。安全测试应显式构造包含多语句、DROP TABLE、无WHERE的DELETE和注释绕过等输入例如DROP TABLE order_legacy; ALTER TABLE order_master ADD INDEX ...。断言重点不是模型会不会生成它而是解析器、权限和审批层能否稳定拒绝。模型输出具有随机性提示词约束不能替代执行层校验。只要工具具备写权限就应假定参数可能非法或越权。要将 Agent 引入数据库治理这种核心场景不宜依赖模型的“自觉性”需要在 Agent 与数据库之间建立由确定性软件工程构筑的防护沙箱与拦截闸门。2. LLM Tool Calling 的非确定性陷阱与沙箱隔离架构为什么单纯依靠 Prompt 无法阻断 Agent 生成危险指令因为 LLM 生成工具调用参数本质上是基于概率分布的 Token 预测。当慢查询上下文里包含了“清理废弃字段”、“重建临时表”等敏感词汇时模型的注意力机制非常容易被诱导输出包含DROP、TRUNCATE或无WHERE条件的DELETE指令。为了解决 Tool Calling 的非确定性风险可以设计一套确定性 SQL 校验沙箱架构。这套架构的核心是将 Agent 的“决策权”与最终“执行权”物理解耦意图路由与 AST 静态解析Agent 生成的任何 SQL 工具调用指令需要先经过 AST抽象语法树解析器。纯静态代码强制剥离所有的 DDL 删除操作与危险 DML。影子数据库Shadow DB沙箱演练Agent 生成的ALTER TABLE语句不能直接在生产或测试主库上执行需要先在基于 Docker 容器秒级克隆的影子数据库沙箱中演练。如果在沙箱中执行报错或者导致锁表超时直接拦截并向 Agent 返回错误反馈。确定性参数 Hash 与幂等屏障防止 Agent 在遇到网络抖动或解析超时时重复下发相同的索引创建请求避免数据库产生索引碎片或锁竞争。下面是 Agent SQL 工具调用的确定性防护与沙箱演练流程通过这套控制体系可以把非确定性的 Agent 行为限制在了安全的规则框架内。3. 确定性 SQL 解析与 Agent 拦截器脚手架代码在实现上可以使用 Python 编写了一个 Agent 工具调用的安全拦截器。该拦截器结合了 SQL 语法树分析、只读权限强制校验以及工具调用的超时退避机制。代码拒绝简单的正则匹配而是基于标准的 SQL 解析逻辑来识别操作类型import json import time from typing import Dict, Any, Tuple import sqlglot from sqlglot import exp class DangerousSQLError(Exception): 当 Agent 试图调用破坏性 SQL 时抛出的确定性异常 pass class DatabaseAgentSandbox: def __init__(self, shadow_db_uri: str, max_execution_sec: float 2.0): self.shadow_db_uri shadow_db_uri self.max_execution_sec max_execution_sec # 允许的 DDL/DML 白名单操作类型 self.allowed_expressions (exp.Select, exp.Explain, exp.AlterTable) def intercept_and_execute(self, tool_call_payload: str) - str: Agent 工具调用的统一入口。 执行确定性拦截、语法树解析与沙箱演练。 try: payload json.loads(tool_call_payload) raw_sql payload.get(sql, ).strip() tool_name payload.get(tool_name, ) if not raw_sql: return json.dumps({status: error, message: SQL 参数为空}) # 1. 静态 AST 语法树解析与危险指令过滤 self._verify_sql_safety(raw_sql) # 2. 工具调用的路由逻辑 if tool_name query_explain: return self._run_explain(raw_sql) elif tool_name apply_index_recommendation: return self._run_in_shadow_sandbox(raw_sql) else: return json.dumps({status: error, message: f未知的工具名称: {tool_name}}) except DangerousSQLError as e: # 安全红线拦截直接返回错误信息给 Agent引导模型纠正 CoT return json.dumps({ status: security_blocked, message: f【安全防护拦截】检测到高危 SQL 指令已被沙箱拒绝: {str(e)} }) except Exception as e: return json.dumps({status: error, message: f系统拦截器处理异常: {str(e)}}) def _verify_sql_safety(self, sql: str): 基于 SQLGlot 抽象语法树执行受限的语句类型检查 try: parsed sqlglot.parse_one(sql) except Exception as err: raise DangerousSQLError(fSQL 语法无法解析: {err}) # 检查是否包含任何 DROP, TRUNCATE, DELETE, UPDATE 节点 for node in parsed.find_all(exp.Drop, exp.Truncate, exp.Delete, exp.Update): raise DangerousSQLError(f禁止执行破坏性 SQL 操作: {node.key.upper()}) # 如果包含 ALTER TABLE需要严格限制只能是 ADD INDEX if isinstance(parsed, exp.AlterTable): alter_sql_upper sql.upper() if DROP COLUMN in alter_sql_upper or DROP INDEX in alter_sql_upper: raise DangerousSQLError(ALTER TABLE 中禁止包含 DROP 相关子句) if ADD INDEX not in alter_sql_upper and ADD KEY not in alter_sql_upper: raise DangerousSQLError(ALTER TABLE 仅允许 ADD INDEX 操作) def _run_explain(self, sql: str) - str: 只读执行 EXPLAIN 分析 # 强制拼接 EXPLAIN 关键字 explain_sql fEXPLAIN {sql} if not sql.upper().startswith(EXPLAIN) else sql # 模拟只读 DB 执行 return json.dumps({ status: success, type: explain_result, data: [{id: 1, select_type: SIMPLE, table: orders, type: ALL, possible_keys: None, rows: None}] }) def _run_shadow_sandbox(self, sql: str) - str: 在 Shadow DB 沙箱中演练索引创建 start_time time.time() # 模拟在影子容器中运行 DDL time.sleep(0.1) # 模拟沙箱耗时 if (time.time() - start_time) self.max_execution_sec: return json.dumps({status: sandbox_timeout, message: 沙箱中索引创建耗时过长可能导致线上锁表}) return json.dumps({ status: success, type: sandbox_verified, message: 影子沙箱已返回 EXPLAIN请与基线计划和扫描行数比较。 }) # 验证 Agent 工具调用拦截防线 if __name__ __main__: sandbox DatabaseAgentSandbox(shadow_db_urimysql://shadow:3306/test_db) # 1. 测试 Agent 输出危险 DROP 指令 dangerous_payload json.dumps({ tool_name: apply_index_recommendation, sql: DROP TABLE legacy_orders; ALTER TABLE orders ADD INDEX idx_created (created_at) }) res1 sandbox.intercept_and_execute(dangerous_payload) print(高危指令测试结果:\n, res1) # 2. 测试 Agent 输出正常优化指令 safe_payload json.dumps({ tool_name: apply_index_recommendation, sql: ALTER TABLE orders ADD INDEX idx_user_created (user_id, created_at) }) res2 sandbox.intercept_and_execute(safe_payload) print(\n安全指令测试结果:\n, res2)代码的阻断能力非常干净利落。任何试图绕过校验规则包含DROP或TRUNCATE的 SQL 参数都会被_verify_sql_safety明确斩断并以结构化 JSON 形式向 Agent 返回security_blocked提示重新引导 Agent 修正推理策略。4. 验证口径与记录方法为了让开发者能够在本地秒级复现 Agent 索引优化效果可以搭建一套极轻量级的 Docker Compose 演练脚手架。这套脚手架包含三个组件主测试数据库MySQL 8.0使用可重复生成的数据集并保存行数、分布和初始化脚本。影子演练数据库Shadow MySQL基于临时 tmpfs 挂载数据秒级重置供 Agent 演练 DDL。Agent 工作流 Runner集成上述 Python 拦截沙箱提供 HTTP 调试接口。docker-compose.yml配置文件定义如下version: 3.8 services: main-db: image: mysql:8.0 container_name: agent_main_db environment: MYSQL_ROOT_PASSWORD: rootpassword MYSQL_DATABASE: shop_order ports: - 3306:3306 command: --default-authentication-pluginmysql_native_password --slow_query_log1 --long_query_time0.5 shadow-db: image: mysql:8.0 container_name: agent_shadow_db tmpfs: - /var/lib/mysql:rw,noexec,nosuid,size1g environment: MYSQL_ROOT_PASSWORD: rootpassword MYSQL_DATABASE: shop_order ports: - 3307:33065. Agent 自动化索引优化的执行边界在大模型与 Agent 快速接入企业生产基础设施的今天数据库索引优化的自动化改造带来了明显的效能红线。Agent 调用数据库工具时可以先落实三项不可省略的约束Agent 不持有生产 DDL 权限只生成候选方案并在隔离副本验证。变更由 DBA 审批后选择原生 Online DDL、pt-online-schema-change或gh-ost这些工具仍需评估 MDL、复制延迟和回滚风险。SQL 校验优先使用语法树正则容易被换行、注释或嵌套查询绕过。AST 也不是完整安全边界还要配合数据库只读权限、语句白名单和执行超时。沙箱演练需要带有时空配额限制在影子数据库中演练 DDL 时需要限制语句最大耗时CPU time与内存分配tmpfs limit。防止 Agent 生成产生 Cartesian Product笛卡尔积的坏 SQL反向拖垮本地演练宿主机。收尾