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 的價值在于標準化，但標準化的前提是每個實現都守住底線。