QueryShield

Name: QueryShield
Author: bch1212

AI 智能体与企业数据库之间的安全 SQL 代理。

智能体通过纯英语（或结构化 SQL）调用单一端点。QueryShield 执行以下操作：

通过带有提示词缓存的 Claude 将自然语言转换为 SQL。
在 AST 层面验证每条查询 — 仅允许 SELECT，禁止堆叠语句、禁止使用受限函数，且必须包含 LIMIT。
应用针对智能体的行级安全性：模式/表白名单以及 WHERE 子句注入。
在客户数据库上执行并返回行数据。
将每次尝试记录到只读审计表中 — 仅记录元数据，绝不记录行内容。

智能体永远无法看到连接字符串。

快速入门

pip install -r requirements.txt
cp .env.example .env
# Set ANTHROPIC_API_KEY, DATABASE_URL, VAULT_KEY (see below)

python -m queryshield.start

为 VAULT_KEY 生成一次 Fernet 密钥，并妥善保存：

python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

端到端流程 (curl)

# 1) Boot a tenant. Returns the admin API key — copy it.
curl -X POST localhost:8000/v1/tenants?name=Acme

# 2) Register the customer DB. Connection string is encrypted at rest.
curl -X POST localhost:8000/v1/databases \
  -H 'X-Admin-Key: qs_...' \
  -H 'Content-Type: application/json' \
  -d '{
    "alias": "prod",
    "db_type": "postgresql",
    "connection_string": "postgresql://reader:secret@db.acme.internal:5432/app",
    "allowed_tables": ["users", "orders"]
  }'

# 3) Provision a scoped agent (different from admin) for your AI app.
curl -X POST localhost:8000/v1/agents \
  -H 'X-Admin-Key: qs_...' \
  -H 'Content-Type: application/json' \
  -d '{ "name": "reporting", "tenant_id": "<tenant>" }'

# 4) Set the agent's RLS policy.
curl -X POST localhost:8000/v1/policies \
  -H 'X-Admin-Key: qs_...' \
  -H 'Content-Type: application/json' \
  -d '{
    "agent_id": "<agent>",
    "database_alias": "prod",
    "allowed_tables": ["users", "orders"],
    "row_filters": { "users": "tenant_id = 42" }
  }'

# 5) The agent queries.
curl -X POST localhost:8000/v1/query \
  -H 'X-API-Key: qs_...' \
  -H 'Content-Type: application/json' \
  -d '{
    "database_alias": "prod",
    "query": "how many active users do we have?",
    "mode": "nl",
    "max_rows": 10
  }'

MCP 集成

已在 官方 MCP 注册表 中列出，名称为 io.github.bch1212/queryshield。

安装客户端：

pip install queryshield-mcp

然后将其放入你的 Claude Desktop / Cursor / 智能体配置中：

{
  "queryshield": {
    "command": "queryshield-mcp",
    "env": { "QUERYSHIELD_API_KEY": "qs_..." }
  }
}

独立 PyPI 包的源码位于 packages/queryshield-mcp/。

MCP 集成（旧版）

将其放入任何支持 MCP 的客户端（Claude Desktop、Cursor、自定义智能体）：

{
  "queryshield": {
    "command": "python",
    "args": ["-m", "queryshield.mcp_server"],
    "env": {
      "QUERYSHIELD_API_KEY": "qs_...",
      "QUERYSHIELD_BASE_URL": "https://api.queryshield.io"
    }
  }
}

暴露的工具：

query_database(database_alias, question, max_rows) — 自然语言查询
query_database_sql(database_alias, sql, max_rows) — 预构建的 SELECT 查询
get_audit_log(limit) — 获取调用智能体的近期尝试记录

安全模型

威胁	防御
智能体构造 `DROP TABLE`	sqlglot AST 拒绝非 SELECT 语句
智能体插入 `;` 和第二条语句	解析器拒绝 `len(statements) > 1`
智能体使用 `pg_sleep`, `xp_cmdshell` 等	在 AST 节点层面进行函数黑名单过滤
智能体读取其范围之外的表	RLS 模式 + 表白名单
智能体读取其他租户的行	通过 AST `.where()` 注入 `row_filters`
连接字符串通过堆栈跟踪泄露	Fernet 加密，绝不在任何 API 中返回
审计日志成为数据泄露向量	仅存储元数据 — 绝不存储行数据
`VAULT_KEY` 轮换	使用新密钥重新加密行（脚本驱动）

safety.py 是最重要的模块。任何添加到此处的额外检查都应附带 tests/test_safety.py 中的测试用例。

定价

层级	月费	数据库数量	每月查询量	备注
入门版	$500	3	1,000,000
专业版	$1,500	10	10,000,000	审计导出
企业版	$3,500	不限	不限	SSO, SIEM webhook

目标为 $32.5K MRR @ 15 个客户（10 个专业版 + 5 个企业版）。

部署

该仓库已适配 Railway。python -m queryshield.start 是入口点（通过 os.getenv 读取 PORT，因为 Railway 在没有 shell 的情况下执行启动命令）。从 Railway 市场配置 Postgres +（可选）Redis，其余配置通过环境变量完成。

railway up

/health 是存活检查。如果控制平面数据库不可达，/ready 将返回 503。

测试

pip install pytest
python -m pytest tests/

42 个测试涵盖：

AST 安全性（24 个案例 — 直接 DDL、注释、编码关键字、多条语句、禁止函数、缺少 LIMIT）
RLS 引擎（6 个案例 — 白名单强制执行、WHERE 注入、与现有谓词的连接）
针对 SQLite “客户数据库”的代理端到端测试（5 个案例 — 正常路径、拦截 DML、RLS 行过滤、表白名单、缓存命中）
通过 FastAPI TestClient 进行的 HTTP 集成测试（7 个案例 — 完整配置 → 查询流程、带 RLS 的受限智能体、认证失败）

QueryShield

QueryShield

快速入门

端到端流程 (curl)

MCP 集成

MCP 集成（旧版）

安全模型

定价

部署

测试

Resources

Latest Blog Posts

MCP directory API