Supabase MCP插件SQL注入漏洞復(fù)盤:Server端權(quán)限校驗與工具注冊安全開發(fā)實踐

Supabase MCP插件漏洞復(fù)盤:Server端安全開發(fā)要點
漏洞真實影響:不是“可能”,是已發(fā)生的數(shù)據(jù)泄露
Supabase MCP插件的漏洞已在生產(chǎn)環(huán)境被利用。攻擊者通過構(gòu)造/mcp/tools路徑下的惡意請求,繞過所有權(quán)限檢查,直接執(zhí)行任意SQL查詢——包括SELECT * FROM auth.users、pg_dump導(dǎo)出語句等。有團隊確認其PostgreSQL日志中出現(xiàn)未授權(quán)的COPY ... TO PROGRAM調(diào)用,說明攻擊者已獲取數(shù)據(jù)庫文件系統(tǒng)訪問權(quán)限。
這不是理論風(fēng)險。它暴露了一個關(guān)鍵事實:MCP Server實現(xiàn)中,工具注冊邏輯與權(quán)限校驗完全解耦。
問題根源:兩處硬傷
1. 工具調(diào)用零校驗
MCP規(guī)范要求Server對每個工具調(diào)用做三重檢查:調(diào)用方身份、工具白名單、參數(shù)合法性。但Supabase插件只做了第一項(JWT解析),且未驗證token中聲明的scope是否包含該工具權(quán)限。
更嚴重的是,它把工具函數(shù)直接掛載到HTTP路由,例如:
# 錯誤示范:工具函數(shù)直連路由
@app.route('/mcp/tools/query', methods=['POST'])
def query_tool():
# 這里沒有檢查調(diào)用方是否有query權(quán)限
return execute_sql(request.json['query']) # ← 直接執(zhí)行用戶輸入結(jié)果是:任何持有有效JWT(哪怕只是anon角色)的請求,都能觸發(fā)任意SQL。
2. Agent調(diào)用鏈無邊界控制
MCP協(xié)議允許Agent間相互調(diào)用,但Supabase插件未限制調(diào)用深度和目標(biāo)范圍。攻擊者發(fā)現(xiàn):
- Agent A(低權(quán)限)可調(diào)用Agent B(高權(quán)限)
- Agent B在執(zhí)行時未校驗調(diào)用來源,直接信任傳入?yún)?shù)
- 最終形成跳板:
anon → data_cleaner → model_inference → db_admin
日志顯示,一次攻擊鏈包含7次跨Agent調(diào)用,其中3個Agent運行在相同進程內(nèi),共享內(nèi)存空間——權(quán)限隔離徹底失效。
實操方案:現(xiàn)在就能加上的防護
權(quán)限校驗必須嵌入工具層
不要依賴中間件統(tǒng)一鑒權(quán)。每個工具函數(shù)啟動時,必須顯式檢查:
- 調(diào)用方JWT中的
scope字段是否包含當(dāng)前工具ID - 請求參數(shù)是否在預(yù)定義schema內(nèi)(用Pydantic或JSON Schema)
- 數(shù)據(jù)庫操作是否限定在租戶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
# 關(guān)鍵:從token提取租戶ID和權(quán)限
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調(diào)用必須帶簽名和超時
禁止裸HTTP調(diào)用。所有Agent間通信需滿足:
- 調(diào)用方用私鑰對請求體簽名,接收方用公鑰驗簽
- 每次調(diào)用附帶
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
# 驗證調(diào)用方白名單
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
# 執(zhí)行調(diào)用...安全不是功能,是部署約束
合規(guī)Agent的變現(xiàn)能力,取決于它能否通過以下硬性檢查:
- 租戶隔離:每個請求必須綁定
tenant_id,數(shù)據(jù)庫連接池按租戶分隔 - 資源熔斷:單次工具調(diào)用CPU時間>2s或內(nèi)存>100MB時強制終止
- 審計留痕:所有工具調(diào)用記錄
caller_id、tool_id、input_hash、duration_ms到獨立審計表
沒有這些,所謂“安全增值服務(wù)”只是營銷話術(shù)。用戶真正付費的,是能寫進SLA的確定性保障——比如“SQL注入防護覆蓋率100%”、“跨租戶數(shù)據(jù)泄露零事件”。
立即行動清單
- 檢查你的MCP Server:搜索代碼中所有
@app.route裝飾器,確認每個工具路由是否包含scope校驗和參數(shù)schema驗證 - 禁用危險工具:臨時移除
execute_sql、run_shell等高危工具,改用預(yù)編譯查詢模板 - 強制租戶上下文:在所有數(shù)據(jù)庫操作前插入
SET search_path TO tenant_xxx,并在連接池初始化時綁定 - 添加調(diào)用簽名:為Agent間通信生成RSA密鑰對,要求所有
/api/call_agent請求攜帶X-Signature頭
別等下一個漏洞。現(xiàn)在就打開編輯器,刪掉那行沒校驗scope的return execute_sql()。