MCP（Model Context Protocol）让大模型能够调用数据库工具，但“能调用”不等于“应该拥有数据库管理员权限”。KaiwuDB 开源的 KWDB MCP Server 中定义了一个很清晰的能力边界：read-query 用于执行 SELECT、SHOW、EXPLAIN 等只读查询；write-query 同时支持 DML 和 DDL，包括 INSERT、UPDATE、DELETE、CREATE、ALTER、DROP。

由此可见，模型是否“理解了自然语言”不是重点，而一次自然语言请求是否被转换成了一个拥有过大权限、缺少审批和缺少回滚边界的 SQL，才是我们更值得关心的。本文以 KWDB MCP Server 的官方能力为基线，给出一套可验证的防护方案：最小权限、工具分层、SQL AST 检查、影响行数阈值、事务与超时、双人审批和审计。

一、先厘清 KWDB MCP Server 的真实能力

KWDB MCP Server 是一个基于 MCP（Model Context Protocol，模型上下文协议）协议的服务器实现，它通过 MCP 协议提供一套工具和资源，用于与 KaiwuDB 数据库交互和提供商业智能功能。KWDB MCP Server 支持读取、写入、查询、修改数据以及执行 DDL 操作。以下内容来自 KaiwuDB 文档：

官方文档展示的能力	安全含义
`read-query` 支持 `SELECT`、`SHOW`、`EXPLAIN` 等只读查询	适合作为低风险工具，但仍需限制结果量、执行时间和可访问对象
没有 `LIMIT` 的 `SELECT` 会自动添加 `LIMIT 20`	这是结果集保护，不是权限控制；`UPDATE`、`DELETE` 和 DDL 不受此规则保护
`write-query` 支持 DML 与 DDL，包括 `DROP TABLE`	一个写工具同时拥有改数据和改结构的能力，不能直接绑定高权限账号
MCP Resources 暴露产品信息、数据库元数据和表结构	元数据应按不可信输入处理，不能因为“来自数据库”就绕过提示注入和敏感字段检查
支持 stdio 和 HTTP；HTTP 示例使用 `http://localhost:8080/mcp`	HTTP 部署必须增加认证、TLS、网络隔离和请求审计
连接字符串使用 PostgreSQL 风格 DSN	可以复用 PostgreSQL 生态的连接池、事务和驱动，但具体 SQL/权限特性仍需在目标 KaiwuDB 版本验证
验证示例返回 `abnormal_vehicle_count = 5`，并标记 `auto_limited: true`、`affected_rows: 0`	返回的元数据可以作为自动化审计字段和回归测试断言

这里有一个容易被忽略的细节：示例配置中的 "autoApprove": [] 只描述客户端的自动批准配置；它不等价于数据库 GRANT，也不能替代服务端授权。客户端是否显示 Auto-approve 入口，同样不应被当作数据库安全控制。

二、威胁模型：自然语言是入口，SQL 才是实际动作

把请求拆成四层，可以避免把责任错误地归给模型：

用户自然语言
    ↓
LLM 选择 MCP 工具并生成参数
    ↓
MCP Server 执行 SQL
    ↓
KaiwuDB 角色、事务、锁和数据发生变化

常见危险转换包括：

用户原话	可能生成的 SQL	主要风险
“清理三个月前的订单”	`DELETE FROM orders WHERE created_at < ...`	条件过宽，影响行数不可预期
“把异常车辆都改成已处理”	`UPDATE vehicle SET status = 'done' ...`	大范围更新，业务语义可能被误解
“删除不用的临时表”	`DROP TABLE tmp_x`	DDL 破坏结构，通常无法靠业务事务恢复
“把用户邮箱导出来”	`SELECT id, email FROM users`	只读但可能泄露个人信息

因此，安全目标不是让模型永远不犯错，而是让错误请求最多停在“被拒绝、被审批或被回滚”的阶段。

三、第一条边界：读写工具和数据库角色必须分离

3.1 只读角色

下面是 PostgreSQL 风格的最小权限示例。KaiwuDB 使用 PostgreSQL 风格连接串，但角色和默认权限语法请在目标版本的兼容性文档中验证后再上线：

-- 管理员执行：为 MCP 只读服务创建独立角色
CREATE ROLE mcp_ro LOGIN PASSWORD 'rotate-this-secret';

GRANT CONNECT ON DATABASE appdb TO mcp_ro;
GRANT USAGE ON SCHEMA public TO mcp_ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_ro;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO mcp_ro;

-- 防止未来新表默认变成不可控的权限扩散
ALTER DEFAULT PRIVILEGES IN SCHEMA public
  GRANT SELECT ON TABLES TO mcp_ro;

-- 明确拒绝结构变更权限（即使未来有人误授了更宽的 schema 权限，也要复核）
REVOKE CREATE ON SCHEMA public FROM mcp_ro;

3.2 受限写角色

把“改业务数据”和“改数据库结构”拆开。生产环境通常应该让 MCP 写角色只能访问明确的表和列，并且不授予 CREATE、ALTER、DROP：

CREATE ROLE mcp_rw LOGIN PASSWORD 'rotate-this-secret-too';

GRANT CONNECT ON DATABASE appdb TO mcp_rw;
GRANT USAGE ON SCHEMA app TO mcp_rw;
GRANT SELECT, INSERT, UPDATE, DELETE
  ON app.orders, app.vehicle
  TO mcp_rw;

REVOKE CREATE ON SCHEMA app FROM mcp_rw;

DDL 应进入独立的迁移流水线，而不是由通用 write-query 工具直接执行。若确实需要保留 DDL 能力，至少使用单独的 MCP Server、单独的数据库账号、单独的网络入口和强制人工审批。

四、第二条边界：在 MCP Server 前增加 SQL 策略网关

提示词可以要求“不要删表”，但提示词不是强制策略。下面的 Python 示例使用 sqlglot 做 AST 级检查，避免只靠字符串查找 DROP 或 WHERE。示例面向 PostgreSQL 风格 SQL；部署前应使用目标 KaiwuDB 版本的真实语句补充回归测试。

安装依赖：

python -m pip install "psycopg[binary]" sqlglot

策略代码：

from dataclasses import dataclass
from typing import Literal

import sqlglot
from sqlglot import exp


class PolicyError(ValueError):
    pass


@dataclass(frozen=True)
class Decision:
    sql: str
    kind: str
    requires_approval: bool


MAX_READ_ROWS = 20
MAX_WRITE_ROWS = 100


def guard_sql(sql: str, approved: bool = False) -> Decision:
    """在真正提交给数据库前检查 SQL；多语句和 DDL 默认拒绝。"""
    trees = sqlglot.parse(sql, read="postgres")
    if len(trees) != 1:
        raise PolicyError("只允许一条 SQL 语句")

    tree = trees[0]
    kind = type(tree).__name__

    # 结构变更不走通用写查询通道。
    if kind in {"Create", "Alter", "Drop", "Truncate"}:
        raise PolicyError(f"禁止通过通用 MCP 工具执行 DDL: {kind}")

    if kind in {"Update", "Delete"}:
        if tree.args.get("where") is None:
            raise PolicyError(f"{kind} 必须包含 WHERE 条件")
        # 影响范围较大的写操作必须由人工审批令牌放行。
        return Decision(sql=tree.sql(dialect="postgres"),
                        kind=kind,
                        requires_approval=not approved)

    if kind == "Insert":
        return Decision(sql=tree.sql(dialect="postgres"),
                        kind=kind,
                        requires_approval=not approved)

    if kind == "Select":
        # KaiwuDB 的 read-query 会自动补 LIMIT 20；网关再做一次防御，
        # 使策略不会依赖某个客户端是否真的经过该工具。
        if tree.args.get("limit") is None:
            tree = tree.limit(MAX_READ_ROWS)
        return Decision(sql=tree.sql(dialect="postgres"),
                        kind=kind,
                        requires_approval=False)

    if kind in {"Show", "Describe", "Explain"}:
        return Decision(sql=tree.sql(dialect="postgres"),
                        kind=kind,
                        requires_approval=False)

    raise PolicyError(f"不在允许列表中的语句类型: {kind}")

这段代码的关键点有三个：

拒绝多语句：UPDATE ...; DROP TABLE ... 不能借助一个参数同时完成。
拒绝通用 DDL：即使模型把“清理”理解为删除表，也会在数据库执行前失败。
只读查询显式限行：官方示例中的 auto_limited: true 是一个有用的事实，但网关仍应自己设置上限，避免绕过 read-query 的调用路径。

AST 检查不是万能的。它不能替代数据库权限，也不能自动判断业务条件是否正确；它的作用是把一部分高风险语句稳定地挡在执行层之前。

五、第三条边界：事务、超时和影响范围要可回滚

对于 UPDATE、DELETE，最小执行单元应包含超时、事务和影响行数检查。下面的代码展示了一个防御性执行器：

import psycopg


def execute_write(conn: psycopg.Connection, sql: str, approved: bool = False) -> int:
    decision = guard_sql(sql, approved=approved)
    if decision.kind not in {"Insert", "Update", "Delete"}:
        raise PolicyError("写执行器只接受 DML")
    if decision.requires_approval:
        raise PolicyError("该操作需要人工审批后重新提交")

    with conn.transaction():
        with conn.cursor() as cur:
            # SET LOCAL 只在当前事务生效，避免污染连接池中的后续请求。
            cur.execute("SET LOCAL statement_timeout = '5s'")
            cur.execute("SET LOCAL lock_timeout = '2s'")
            cur.execute(decision.sql)

            affected = cur.rowcount
            if affected is not None and affected >= 0 and affected > MAX_WRITE_ROWS:
                raise PolicyError(
                    f"影响行数 {affected} 超过阈值 {MAX_WRITE_ROWS}，事务将回滚"
                )
            return affected

需要特别说明两点：

rowcount 是执行后的防线，不应被用来替代 WHERE 校验；不同驱动或语句类型可能返回 -1，此时应在执行前用同一事务内的 SELECT COUNT(*) 做预估，或要求人工审批。
DROP TABLE 这类 DDL 即使处于事务中，也可能带来锁等待、依赖对象变化和运维恢复成本，因此“能回滚”不等于“风险低”。DDL 应独立审批、备份并记录变更单号。

六、第四条边界：把“审批”设计成数据结构，而不是聊天确认

“你确定吗？”是弱确认。更可靠的审批令牌应绑定 SQL 摘要、操作者、目标库、过期时间和最大影响行数，例如：

{
  "request_id": "mcp-20261020-00017",
  "database": "appdb",
  "sql_sha256": "<sha256 of normalized SQL>",
  "max_affected_rows": 100,
  "approved_by": ["dba@example.com", "owner@example.com"],
  "expires_at": "2026-10-20T12:30:00+08:00"
}

执行器应重新规范化 SQL 并计算摘要，只有摘要、数据库和阈值全部匹配时才放行。这样可以防止“审批了安全的 UPDATE ... WHERE id = 1，执行时却被替换成另一条 SQL”。

七、第五条边界：审计要记录“模型为什么这样做”

至少记录以下字段：

{
  "request_id": "mcp-20261020-00017",
  "actor": "mcp-service",
  "tool": "write-query",
  "natural_language": "把异常车辆标记为已处理",
  "normalized_sql": "UPDATE app.vehicle SET status = $1 WHERE id = $2",
  "database": "appdb",
  "decision": "approved",
  "affected_rows": 1,
  "duration_ms": 42,
  "transaction": "committed",
  "created_at": "2026-10-20T12:20:10+08:00"
}

日志中不要直接写入数据库密码、完整身份证号、邮箱等敏感值。自然语言原文可以保留用于事故回放，但应按敏感信息等级脱敏和限权访问。

八、可重复的安全回归测试

下面的测试集不依赖真实数据，先验证策略层是否拒绝明显危险语句：

CASES = [
    ("SELECT * FROM app.vehicle", "Select"),
    ("SELECT * FROM app.vehicle LIMIT 5", "Select"),
    ("UPDATE app.vehicle SET status = 'done' WHERE id = 1", "Update"),
    ("DELETE FROM app.vehicle WHERE id = 1", "Delete"),
]

DENIED = [
    "UPDATE app.vehicle SET status = 'done'",
    "DELETE FROM app.vehicle",
    "DROP TABLE app.vehicle",
    "UPDATE app.vehicle SET status = 'done'; DROP TABLE app.vehicle",
]

for sql, expected_kind in CASES:
    decision = guard_sql(sql, approved=True)
    assert decision.kind == expected_kind

for sql in DENIED:
    try:
        guard_sql(sql, approved=True)
    except PolicyError:
        pass
    else:
        raise AssertionError(f"危险 SQL 未被拒绝: {sql}")

print("policy regression: PASS")

上线前应再增加三类真实回归：

语法兼容性：把 KaiwuDB 目标版本支持的 SHOW、EXPLAIN、分页、分区表和函数加入测试。
权限有效性：使用 mcp_ro 真实连接确认 INSERT/UPDATE/DELETE/DDL 均被数据库拒绝。
影响范围：在脱敏数据集上测量每类操作的预计行数、锁等待、超时和回滚结果。

九、如何解读官方验证数据

题图中的验证场景询问“当前有多少个异常车辆”，最终返回 abnormal_vehicle_count = 5。响应元数据还显示：affected_rows = 0、row_count = 1、auto_limited = true。这组数据能够证明：

read-only 查询可以通过 MCP 工具路由到 KaiwuDB；
结果被包装成结构化 JSON，便于上层 Agent 消费；
自动限行状态可以进入审计和回归断言。

但它不能证明：写操作安全、模型不会生成错误 SQL、权限已经最小化，或者 HTTP 端点已经完成认证。安全结论必须由权限测试、拒绝测试和故障演练共同支撑。

十、推荐的生产部署形态

LLM Agent
   │
   ├── read-query ──> 只读 MCP Server ──> mcp_ro ──> KaiwuDB
   │                    │
   │                    └── AST、限行、超时、审计
   │
   └── write-query ─> 写入 MCP Server ──> mcp_rw ──> KaiwuDB
                        │
                        └── 审批令牌、影响行数阈值、事务回滚、审计

DDL / DROP TABLE ──> 迁移流水线 + DBA 审批（不走通用 MCP）

如果只能做三件事，优先级是：

生产默认只暴露 read-query，并使用只读账号。
写入请求必须经过 AST 检查、影响范围阈值和人工审批。
DDL 从通用 write-query 中剥离，交给版本化迁移流程。

结语

MCP 的价值是把数据库能力标准化地提供给 Agent；它并不会自动替你完成权限治理。KaiwuDB 官方 MCP 同时提供只读查询和写查询，尤其是 write-query 覆盖 DML 与 DDL，这正是必须建立安全边界的原因。

最可靠的做法是把自然语言当作“不可信意图”，把 SQL 当作“待审批命令”，把数据库角色和事务当作最终执行边界。模型可以负责理解和解释，是否允许改变数据和结构，必须由可测试、可审计、可回滚的工程策略决定。

物联网行业解决方案

工业物联网

能源电力

车联网

冶金

智慧矿山

智慧烟草

赋能千行百业数字化转型