From bb85d898744d51b9cffd9a1a5dd69c80406c5ed5 Mon Sep 17 00:00:00 2001 From: OG T Date: Mon, 30 Mar 2026 01:44:42 +0800 Subject: [PATCH] =?UTF-8?q?refactor(api):=20Phase=20A=20P1=20=E5=BF=AB?= =?UTF-8?q?=E9=80=9F=E5=8B=9D=E5=88=A9=20(3=20=E9=A0=85)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 常數提取: SSE_DELAY_SECONDS, MAX_APPROVAL_DISPLAY 2. 錯誤訊息安全化: sanitize_error_message() 移除敏感資訊 3. CI/CD alertname 配置化: is_cicd_alertname() 函數 首席架構師審查 P1 改進 (非阻塞) Co-Authored-By: Claude Opus 4.5 --- apps/api/src/api/v1/webhooks.py | 16 +--- apps/api/src/core/constants.py | 104 ++++++++++++++++++++++ apps/api/src/services/terminal_service.py | 19 ++-- 3 files changed, 121 insertions(+), 18 deletions(-) diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index 356fef92..58c8f768 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -30,6 +30,7 @@ from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, from pydantic import BaseModel, Field from src.core.config import settings +from src.core.constants import is_cicd_alertname from src.core.logging import get_logger # Phase 15.2: Trace Context (moved to SignalProducerService) @@ -1159,19 +1160,10 @@ async def alertmanager_webhook( alertname = alert.labels.get("alertname", "UnknownAlert") # ========================================================================== - # 2026-03-30 ogt: CI/CD 告警偵測 - 跳過 AI 仲裁，使用簡潔格式 - # CI/CD 告警 alertname 模式: CD_*, CI_*, E2E_*, SmokeTest, *_Build, *_Test + # 2026-03-30 P1: CI/CD 告警偵測 - 配置化 (constants.py) + # 跳過 AI 仲裁，使用簡潔格式 # ========================================================================== - cicd_prefixes = ("CD_", "CI_", "E2E_", "SmokeTest", "Build_", "Test_", "Deploy_") - cicd_suffixes = ("_Build", "_Test", "_Deploy", "_E2E") - is_cicd_alert = ( - alertname.startswith(cicd_prefixes) or - alertname.endswith(cicd_suffixes) or - "CI/CD" in alertname or - "cicd" in alertname.lower() - ) - - if is_cicd_alert: + if is_cicd_alertname(alertname): # CI/CD 告警 - 使用簡潔格式，不走 AI 仲裁 logger.info( "alertmanager_cicd_detected", diff --git a/apps/api/src/core/constants.py b/apps/api/src/core/constants.py index 715985f7..e77d5f5d 100644 --- a/apps/api/src/core/constants.py +++ b/apps/api/src/core/constants.py @@ -43,3 +43,107 @@ APPROVAL_TO_INCIDENT_STATUS = { # Incident 狀態 → 是否活躍 INCIDENT_ACTIVE_STATUSES = frozenset({"investigating", "mitigating"}) + +# ============================================================================= +# Terminal Service Settings (Phase 19.4 P1) +# 2026-03-30 Claude Code: 首席架構師審查 - 常數提取 +# ============================================================================= + +# SSE 事件間隔 (秒) - 避免前端過載 +SSE_DELAY_SECONDS = 0.3 + +# Approval 清單顯示上限 +MAX_APPROVAL_DISPLAY = 5 + +# 錯誤訊息截斷長度 (安全性) +ERROR_MESSAGE_MAX_LENGTH = 100 +ERROR_MESSAGE_DISPLAY_LENGTH = 50 + +# ============================================================================= +# CI/CD Alert Detection (2026-03-30 P1 配置化) +# ============================================================================= + +# CI/CD 告警 alertname 前綴 +CICD_ALERT_PREFIXES = ( + "CD_", + "CI_", + "E2E_", + "SmokeTest", + "Build_", + "Test_", + "Deploy_", +) + +# CI/CD 告警 alertname 後綴 +CICD_ALERT_SUFFIXES = ( + "_Build", + "_Test", + "_Deploy", + "_E2E", +) + +# CI/CD 告警關鍵字 (不區分大小寫) +CICD_ALERT_KEYWORDS = ("CI/CD", "cicd") + + +def is_cicd_alertname(alertname: str) -> bool: + """ + 判斷 alertname 是否為 CI/CD 告警 + + Args: + alertname: Alertmanager alertname 標籤值 + + Returns: + True 如果是 CI/CD 告警 + """ + return ( + alertname.startswith(CICD_ALERT_PREFIXES) + or alertname.endswith(CICD_ALERT_SUFFIXES) + or any(kw in alertname for kw in CICD_ALERT_KEYWORDS) + or "cicd" in alertname.lower() + ) + + +def sanitize_error_message(error: str, max_length: int = ERROR_MESSAGE_MAX_LENGTH) -> str: + """ + 安全化錯誤訊息 - 截斷並移除敏感資訊 + + 2026-03-30 Claude Code: 首席架構師審查 - 錯誤訊息安全化 + + Args: + error: 原始錯誤訊息 + max_length: 最大長度 (預設 100) + + Returns: + 安全化後的錯誤訊息 + """ + if not error: + return "Unknown error" + + # 移除可能的敏感資訊模式 + sensitive_patterns = [ + "password", + "token", + "secret", + "api_key", + "apikey", + "credential", + "bearer", + ] + + sanitized = error + for pattern in sensitive_patterns: + # 不區分大小寫替換 + import re + sanitized = re.sub( + rf"({pattern})\s*[:=]\s*\S+", + f"{pattern}=[REDACTED]", + sanitized, + flags=re.IGNORECASE, + ) + + # 截斷 + if len(sanitized) > max_length: + return sanitized[:max_length - 3] + "..." + + return sanitized diff --git a/apps/api/src/services/terminal_service.py b/apps/api/src/services/terminal_service.py index 08bf8345..faa88f56 100644 --- a/apps/api/src/services/terminal_service.py +++ b/apps/api/src/services/terminal_service.py @@ -20,6 +20,12 @@ import uuid from enum import Enum from typing import Protocol, runtime_checkable +from src.core.constants import ( + ERROR_MESSAGE_DISPLAY_LENGTH, + MAX_APPROVAL_DISPLAY, + SSE_DELAY_SECONDS, + sanitize_error_message, +) from src.core.logging import get_logger from src.core.sse import EventType, SSEEvent, get_publisher from src.models.terminal import ( @@ -511,7 +517,7 @@ class TerminalService: publisher, topic, "Approval-Query", "completed", result={"pending_count": len(pending_approvals)} ) - await asyncio.sleep(0.3) + await asyncio.sleep(SSE_DELAY_SECONDS) if not pending_approvals: await self._publish_thought( @@ -522,14 +528,15 @@ class TerminalService: # 顯示待簽核清單摘要 summary_lines = [f"發現 {len(pending_approvals)} 個待簽核項目:"] - for i, approval in enumerate(pending_approvals[:5], 1): # 最多顯示 5 個 + for i, approval in enumerate(pending_approvals[:MAX_APPROVAL_DISPLAY], 1): risk = approval.risk_level.value.upper() if approval.risk_level else "UNKNOWN" summary_lines.append( f" {i}. [{risk}] {approval.action[:50]}" ) - if len(pending_approvals) > 5: - summary_lines.append(f" ... 還有 {len(pending_approvals) - 5} 個") + if len(pending_approvals) > MAX_APPROVAL_DISPLAY: + remaining = len(pending_approvals) - MAX_APPROVAL_DISPLAY + summary_lines.append(f" ... 還有 {remaining} 個") await self._publish_thought( publisher, topic, "Executor", @@ -561,11 +568,11 @@ class TerminalService: logger.error("approval_query_failed", error=str(e)) await self._publish_tool_call( publisher, topic, "Approval-Query", "failed", - result={"error": str(e)[:100]} + result={"error": sanitize_error_message(str(e))} ) await self._publish_thought( publisher, topic, "System", - f"查詢簽核項目失敗: {str(e)[:50]}" + f"查詢簽核項目失敗: {sanitize_error_message(str(e), ERROR_MESSAGE_DISPLAY_LENGTH)}" ) async def _handle_metrics_query(