fix(e2e): 新增 HMAC 簽名支援

E2E 腳本現在會:
- 讀取 WEBHOOK_HMAC_SECRET 環境變數
- 計算 HMAC-SHA256 簽名
- 加入 X-Webhook-Signature header

修復生產環境 401 驗證失敗問題

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-29 20:54:28 +08:00
parent c80a69bd88
commit 8bd51ea7c8

View File

@@ -29,6 +29,10 @@ Version: 2.0 (Phase 18.2 優化)
import argparse
import asyncio
import hashlib
import hmac
import json
import os
import re
import sys
import time
@@ -46,6 +50,17 @@ import httpx
DEFAULT_API_URL = "http://localhost:8000"
TIMEOUT = 60.0
HMAC_SECRET = os.getenv("WEBHOOK_HMAC_SECRET", "")
def compute_hmac_signature(secret: str, payload: bytes) -> str:
"""計算 HMAC-SHA256 簽章"""
signature = hmac.new(
secret.encode(),
payload,
hashlib.sha256,
).hexdigest()
return f"sha256={signature}"
# E2E Signer Pool (用於動態簽署)
SIGNER_POOL = [
@@ -201,16 +216,28 @@ class E2EVerification:
print_info("Severity", self.test_alert["severity"])
try:
# 準備 payload 和 headers
payload_bytes = json.dumps(self.test_alert).encode("utf-8")
headers = {"Content-Type": "application/json"}
# 如果有 HMAC Secret計算簽名
if HMAC_SECRET:
signature = compute_hmac_signature(HMAC_SECRET, payload_bytes)
headers["X-Webhook-Signature"] = signature
print_info("HMAC", f"已簽名 ({signature[:20]}...)")
else:
print_warn("無 HMAC Secret - 僅限測試環境")
async with httpx.AsyncClient(timeout=TIMEOUT) as client:
response = await client.post(
f"{self.api_url}/api/v1/webhooks/alerts",
json=self.test_alert,
headers={"Content-Type": "application/json"},
content=payload_bytes,
headers=headers,
)
if response.status_code == 401:
print_warn("HMAC 驗證啟用中 - 生產環境需要簽名")
print_info("提示", "請在測試環境執行,或配置 HMAC Secret")
print_warn("HMAC 驗證失敗 - 檢查 Secret 是否正確")
print_info("提示", "確認 WEBHOOK_HMAC_SECRET 與 K8s 一致")
return False
if response.status_code != 200: