Supabase MCP插件安全隐患：开源不等于安全，协议层鉴权缺失才是关键

问题本质：MCP endpoint 没有协议级防护

Supabase 的 MCP 插件暴露了一个 /mcp endpoint，它直接将客户端传入的 SQL 查询转发给数据库执行。这个设计跳过了所有应用层访问控制——没有用户身份校验、没有权限分级、没有操作白名单。只要能发 HTTP 请求到这个地址，就能读取任意表、导出全库、甚至删库。

这不是配置疏忽，是协议实现层面的缺失：MCP 规范本身要求服务端对每个操作做鉴权和沙箱约束，但当前插件把这部分完全交给了上层应用，而多数开发者根本没意识到这里需要补漏。

协议层该有的两道防线

鉴权不是可选项

MCP 协议明确要求服务端验证调用方身份，并基于角色或策略限制其可执行的操作类型（如只读、只查特定表）。Supabase 插件目前把 Authorization 头直接忽略，所有请求都以数据库超级用户身份执行。

后果很直接：

• 攻击者用 curl -X POST https://your-project.supabase.co/mcp -d '{"query":"SELECT * FROM users"}' 就能拿到全部用户数据
• 如果数据库连接配置了高权限账号，DROP TABLE、COPY TO PROGRAM 等危险操作也能成功

沙箱不是性能负担，是必要隔离

沙箱机制要解决两个问题：

• 语法隔离：禁止 DELETE/UPDATE/INSERT/TRUNCATE 等写操作，只允许 SELECT
• 语义隔离：限制 SELECT 范围，比如禁止跨 schema 查询、禁止 information_schema 探针、限制返回行数

当前插件不做任何解析，原样执行传入的 SQL 字符串。一个 UNION SELECT pg_read_file('/etc/passwd') 就可能触发数据库侧信道泄露。

实际风险场景

全库导出无需提权

攻击者构造如下请求即可导出整个数据库结构和内容：

curl -X POST https://your-project.supabase.co/mcp \
  -H "Content-Type: application/json" \
  -d '{
        "query": "SELECT table_name FROM information_schema.tables WHERE table_schema = \'public\'"
      }'

拿到表名后，循环执行 SELECT * FROM <table>，几条 shell 脚本就能完成全量 dump。Supabase 默认开启 Row Level Security（RLS），但 MCP 插件绕过了 RLS——它直连数据库，不走 PostgREST 层。

敏感数据在 AI 流水线中裸奔

很多 AI Agent 项目用 MCP 插件做“数据库工具调用”，把用户会话 ID、支付记录、原始日志等直接喂给 LLM。一旦 MCP endpoint 泄露，这些数据就变成明文 CSV 流向外部。

更危险的是：LLM 可能被诱导生成恶意查询。比如用户输入“帮我删掉测试数据”，Agent 解析成 DELETE FROM logs WHERE is_test = true 并提交——插件照单执行，且无审计日志。

写操作导致服务雪崩

UPDATE accounts SET balance = balance - 100 WHERE user_id = 'attacker' 这类语句不会报错，但会污染业务数据。更糟的是 VACUUM FULL 或 CREATE INDEX CONCURRENTLY 等维护命令可能锁表，让整个应用不可用。

怎么修：在 endpoint 层加两道硬闸

必须加鉴权：JWT 校验 + 权限映射

不要依赖网络层防火墙或 IP 白名单。每个请求必须携带有效 JWT，且 token payload 中需声明 allowed_tables 和 allowed_operations。

from flask import Flask, request, jsonify
import jwt
from datetime import datetime

app = Flask(__name__)

def require_mcp_auth(f):
    def decorated(*args, **kwargs):
        auth = request.headers.get('Authorization')
        if not auth or not auth.startswith('Bearer '):
            return jsonify({'error': 'Missing or invalid Authorization header'}), 401
        
        token = auth[7:]
        try:
            payload = jwt.decode(token, 'your-supabase-jwt-secret', algorithms=['HS256'])
            # 强制检查权限字段
            if 'allowed_tables' not in payload or 'allowed_operations' not in payload:
                return jsonify({'error': 'Invalid token: missing permissions'}), 403
            request.mcp_permissions = payload
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token expired'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': 'Invalid token'}), 401
        return f(*args, **kwargs)
    return decorated

@app.route('/mcp', methods=['POST'])
@require_mcp_auth
def mcp_handler():
    data = request.get_json()
    query = data.get('query', '').strip()
    # 后续校验 query 是否符合 permissions...

必须加沙箱：SQL 解析 + 白名单执行

别用字符串拼接或正则匹配禁用关键词——%00DELETE、注释绕过、Unicode 变体都能绕过。用真正的 SQL 解析器（如 sqlparse）提取 AST，只允许 SELECT 类型节点，且 FROM 子句中的表名必须在 allowed_tables 列表中。

import sqlparse
from sqlparse.sql import IdentifierList, Identifier, Where, Comparison
from sqlparse.tokens import Keyword, DML

def validate_query(query: str, allowed_tables: list) -> bool:
    parsed = sqlparse.parse(query)[0]
    
    # 检查是否为 SELECT
    if not parsed.token_first().ttype is DML and parsed.token_first().value.upper() != 'SELECT':
        return False
    
    # 提取所有表名
    tables = set()
    for token in parsed.flatten():
        if isinstance(token, Identifier) and token.has_alias():
            tables.add(token.get_real_name())
        elif isinstance(token, IdentifierList):
            for identifier in token.get_identifiers():
                if hasattr(identifier, 'get_real_name'):
                    tables.add(identifier.get_real_name())
    
    # 检查表名是否都在白名单中
    return all(table in allowed_tables for table in tables)

@app.route('/mcp', methods=['POST'])
@require_mcp_auth
def mcp_handler():
    data = request.get_json()
    query = data.get('query', '').strip()
    
    if not validate_query(query, request.mcp_permissions['allowed_tables']):
        return jsonify({'error': 'Query violates table access policy'}), 403
    
    # 执行查询（使用参数化，避免二次注入）
    with db_engine.connect() as conn:
        result = conn.execute(text(query))
        return jsonify([dict(row) for row in result]), 200

真实项目怎么落地

不要自己重写 MCP Server

Supabase 官方插件的问题在于它把协议实现简化成了“HTTP → SQL”管道。更稳妥的做法是：

• 用 MCP SDK 启动标准 server
• 在 tool_call handler 中注入鉴权逻辑
• 所有数据库操作走 Supabase Client（自动带 RLS）而非直连

这样既能复用社区规范，又把权限控制收回到应用层。

关键检查清单

部署前确认以下五点：

• ✅ MCP endpoint 的数据库连接账号权限最小化（只读 + 仅限必要 schema）
• ✅ JWT secret 与 Supabase 项目密钥分离，不硬编码在代码里
• ✅ allowed_tables 按业务场景动态生成（如客服 Agent 只能查 tickets 和 users）
• ✅ 所有 SELECT 查询自动加上 LIMIT 1000，防止大结果集拖垮内存
• ✅ 日志记录每次 MCP 调用的 token subject、表名、查询哈希（不记原始 SQL）

商业项目必须签 DPA

如果你的 Agent 服务把客户数据传给第三方 MCP provider，必须签数据处理协议（DPA），明确：

• provider 不得存储原始数据
• 查询日志保留不超过 7 天
• 审计权开放给客户（提供 API 查看调用记录）
• 违约赔偿条款（例如每条泄露记录罚 $500）

GDPR 和国内《个人信息保护法》都把这种“通过协议接口传输数据”的行为认定为委托处理，责任主体仍是你的产品方。

下一步动作

• 立即禁用生产环境的 Supabase MCP 插件，改用 PostgREST + 自定义函数封装数据库操作
• 把 supabase-mcp 仓库 fork 出来，在 src/server.ts 的 handleMcpRequest 函数里插入鉴权和 SQL 白名单逻辑
• 用 sqlglot 替代正则做 SQL 解析——它支持 Postgres 语法树，能准确识别 CTE、子查询、函数调用等复杂结构

安全不是加个中间件，是把权限决策点放在协议入口处。MCP 的价值在于标准化，但标准化的前提是每个实现都守住底线。