防禦方式辦法

1. n8n 工作流層：

   • Queue/Worker 併發控制：限制同時執行的工作流數量，防止後端系統過載
   • 節流節點：在迴圈前加入 Wait 節點，控制每秒處理筆數
   • 重試風暴控制：使用指數退避算法，避免失敗重試放大攻擊流量

2. MCP 工具層：

   • Per-Tool 差異化限制：危險工具如 port_scan、LLM 調用設定更嚴格配額（1-2次/分鐘）
   • Per-API-Key + 角色分級：依身份給予不同配額（user: 5次/分鐘，admin: 10次/分鐘）
   • Per-Input 大小限制：控制 payload 長度、檔案大小，防止單次超量請求

防禦實作

這部分為了方便演示，我只在 MCP 針對用戶（Admin/User）做流量限制，因為前面的防禦已經實作不同身份別會使用不同工作流允續 Agent 使用不同工具，因此我在身份別上也有做一個流量的管控差別。

範例程式碼：

 ====== 總量限流 (Per-Role) ======
_role_timestamps: dict = {}

ROLE_LIMITS = {
    "user": ,    
    "admin": 10,  
}

def _now_epoch() -> int:
    return int(time.time())

def _clean_and_count(bucket: list, window_seconds: int) -> int:
    cutoff = _now_epoch() - window_seconds
    # 將過期的 timestamp 移除（bucket 為遞增排序）
    while bucket and bucket[0] <= cutoff:
        bucket.pop(0)
    return len(bucket)

def _check_rate_limit(api_key: str | None, user_role: str | None):
    """
    依照角色做每分鐘總量限制
    """
    role = user_role if user_role in ROLE_LIMITS else "user"
    limit = ROLE_LIMITS.get(role)
    window = 60  

    if role not in _role_timestamps:
        _role_timestamps[role] = []

    # 清理過期紀錄並取得當前數量
    count = _clean_and_count(_role_timestamps[role], window)

    if count >= limit:
        raise Exception(f"RATE_LIMITED: 角色 {role} 超出配額每分鐘上限 {limit}，請稍後再試")

    # 記錄本次請求
    _role_timestamps[role].append(_now_epoch())

核心防禦策略

1. 角色分級限流

   • user 角色：每分鐘限制次數（程式碼中數值被省略）
   • admin 角色：每分鐘 10 次
   • 不同角色有不同的資源配額，避免所有使用者擠在同一個限制池

2. 滑動視窗演算法

   • _clean_and_count() 函式負責清理過期的時間戳記
   • 使用 60 秒滑動視窗，而非固定時間段重置
   • 這比傳統的「每分鐘重置」更精確，避免突發流量在重置時間點造成衝擊

3. 共享配額池設計

   if role not in _role_timestamps:
       _role_timestamps[role] = []
   

所有 user 角色的請求共享同一個配額池、admin 角色的請求共享另一個配額池
這種設計適合防範分散式攻擊，即使攻擊者使用多個不同的 API Key，仍會受到角色層級的總量限制。

k6 測試

![]
測試概況：
總共執行 50 次請求，全部完成
3 個虛擬使用者並發執行
執行時間 47.1 秒
平均響應時間 2697.29ms（約 2.7 秒）

限流效果：
Rate Limited: 30 次以上 - 表示有 60% 的請求被成功攔截
從日誌可以看到大量 "抱歉，目前因為配額限制，無法立即取得當前時間" 的回應

總結

剛好近期有一些 Key 外洩的事件發生，若是沒有做好相關防禦就有可能變成「被 Google 詐騙」的冤大頭哈哈，看看笑一笑也別忘了好好檢查自己各項專案的安全性！