diff --git a/apps/api/scripts/e2e_tool_call_verification.py b/apps/api/scripts/e2e_tool_call_verification.py index f6801c869..b0dd20c7f 100644 --- a/apps/api/scripts/e2e_tool_call_verification.py +++ b/apps/api/scripts/e2e_tool_call_verification.py @@ -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: