Supabase MCP插件SQL注入漏洞复盘:Server端权限校验与工具注册安全开发实践
摘要:Supabase MCP插件漏洞复盘:Server端安全开发要点漏洞真实影响:不是“可能”,是已发生的数据泄露Supabase MCP插件的漏洞已在生产环境被利用。攻击者通过构造/mcp/tools路径下的恶意请求,绕过所有权限检查,直接执行任意SQL查询——包括SELECT * FROM auth.users、pg_dump导出语句等。有团队确认其PostgreSQL日志中出现未授权的CO...

Supabase MCP插件漏洞复盘:Server端安全开发要点
漏洞真实影响:不是“可能”,是已发生的数据泄露
Supabase MCP插件的漏洞已在生产环境被利用。攻击者通过构造/mcp/tools路径下的恶意请求,绕过所有权限检查,直接执行任意SQL查询——包括SELECT * FROM auth.users、pg_dump导出语句等。有团队确认其PostgreSQL日志中出现未授权的COPY ... TO PROGRAM调用,说明攻击者已获取数据库文件系统访问权限。
这不是理论风险。它暴露了一个关键事实:MCP Server实现中,工具注册逻辑与权限校验完全解耦。
问题根源:两处硬伤
1. 工具调用零校验
MCP规范要求Server对每个工具调用做三重检查:调用方身份、工具白名单、参数合法性。但Supabase插件只做了第一项(JWT解析),且未验证token中声明的scope是否包含该工具权限。
更严重的是,它把工具函数直接挂载到HTTP路由,例如:
# 错误示范:工具函数直连路由
@app.route('/mcp/tools/query', methods=['POST'])
def query_tool():
# 这里没有检查调用方是否有query权限
return execute_sql(request.json['query']) # ← 直接执行用户输入结果是:任何持有有效JWT(哪怕只是anon角色)的请求,都能触发任意SQL。
2. Agent调用链无边界控制
MCP协议允许Agent间相互调用,但Supabase插件未限制调用深度和目标范围。攻击者发现:
- Agent A(低权限)可调用Agent B(高权限)
- Agent B在执行时未校验调用来源,直接信任传入参数
- 最终形成跳板:
anon → data_cleaner → model_inference → db_admin
日志显示,一次攻击链包含7次跨Agent调用,其中3个Agent运行在相同进程内,共享内存空间——权限隔离彻底失效。
实操方案:现在就能加上的防护
权限校验必须嵌入工具层
不要依赖中间件统一鉴权。每个工具函数启动时,必须显式检查:
- 调用方JWT中的
scope字段是否包含当前工具ID - 请求参数是否在预定义schema内(用Pydantic或JSON Schema)
- 数据库操作是否限定在租户schema下(如
tenant_123.users而非public.users)
from pydantic import BaseModel, Field
from typing import Literal
class QueryToolInput(BaseModel):
query: str = Field(..., max_length=2048)
mode: Literal["read", "explain"] = "read" # 禁止write/delete
@app.route('/mcp/tools/query', methods=['POST'])
@token_required
def query_tool():
try:
input_data = QueryToolInput.model_validate(request.json)
except Exception as e:
return jsonify({"error": "Invalid input"}), 400
# 关键:从token提取租户ID和权限
token_payload = jwt.decode(
request.headers['Authorization'],
key=SECRET_KEY,
algorithms=["HS256"]
)
if "query" not in token_payload.get("scope", []):
return jsonify({"error": "Permission denied"}), 403
# 强制重写SQL前缀
safe_query = f"SET search_path TO tenant_{token_payload['tenant_id']}; {input_data.query}"
# 拦截危险操作
if any(kw in safe_query.lower() for kw in ["drop", "delete", "copy", "create"]):
return jsonify({"error": "Write operations forbidden"}), 403
return jsonify(execute_sql(safe_query))Agent调用必须带签名和超时
禁止裸HTTP调用。所有Agent间通信需满足:
- 调用方用私钥对请求体签名,接收方用公钥验签
- 每次调用附带
caller_id和ttl(建议≤5秒) - 接收方拒绝
ttl过期或caller_id不在白名单的请求
import hmac
import time
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.serialization import load_pem_public_key
def verify_agent_call(payload: dict, signature: str, public_key_pem: str) -> bool:
# 验证签名
pub_key = load_pem_public_key(public_key_pem.encode())
try:
pub_key.verify(
bytes.fromhex(signature),
payload['body'].encode(),
padding.PKCS1v15(),
hashes.SHA256()
)
except Exception:
return False
# 验证时效性
if time.time() - payload['timestamp'] > 5:
return False
# 验证调用方白名单
if payload['caller_id'] not in ALLOWED_AGENTS:
return False
return True
@app.route('/api/call_agent', methods=['POST'])
def call_agent():
data = request.json
if not verify_agent_call(
data,
request.headers.get('X-Signature'),
AGENT_PUBLIC_KEYS[data['target_id']]
):
return jsonify({"error": "Invalid call"}), 403
# 执行调用...安全不是功能,是部署约束
合规Agent的变现能力,取决于它能否通过以下硬性检查:
- 租户隔离:每个请求必须绑定
tenant_id,数据库连接池按租户分隔 - 资源熔断:单次工具调用CPU时间>2s或内存>100MB时强制终止
- 审计留痕:所有工具调用记录
caller_id、tool_id、input_hash、duration_ms到独立审计表
没有这些,所谓“安全增值服务”只是营销话术。用户真正付费的,是能写进SLA的确定性保障——比如“SQL注入防护覆盖率100%”、“跨租户数据泄露零事件”。
立即行动清单
- 检查你的MCP Server:搜索代码中所有
@app.route装饰器,确认每个工具路由是否包含scope校验和参数schema验证 - 禁用危险工具:临时移除
execute_sql、run_shell等高危工具,改用预编译查询模板 - 强制租户上下文:在所有数据库操作前插入
SET search_path TO tenant_xxx,并在连接池初始化时绑定 - 添加调用签名:为Agent间通信生成RSA密钥对,要求所有
/api/call_agent请求携带X-Signature头
别等下一个漏洞。现在就打开编辑器,删掉那行没校验scope的return execute_sql()。