refactor(api): Phase A P1 快速勝利 (3 項)
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 <noreply@anthropic.com>
This commit is contained in:
@@ -30,6 +30,7 @@ from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request,
|
|||||||
from pydantic import BaseModel, Field
|
from pydantic import BaseModel, Field
|
||||||
|
|
||||||
from src.core.config import settings
|
from src.core.config import settings
|
||||||
|
from src.core.constants import is_cicd_alertname
|
||||||
from src.core.logging import get_logger
|
from src.core.logging import get_logger
|
||||||
|
|
||||||
# Phase 15.2: Trace Context (moved to SignalProducerService)
|
# Phase 15.2: Trace Context (moved to SignalProducerService)
|
||||||
@@ -1159,19 +1160,10 @@ async def alertmanager_webhook(
|
|||||||
alertname = alert.labels.get("alertname", "UnknownAlert")
|
alertname = alert.labels.get("alertname", "UnknownAlert")
|
||||||
|
|
||||||
# ==========================================================================
|
# ==========================================================================
|
||||||
# 2026-03-30 ogt: CI/CD 告警偵測 - 跳過 AI 仲裁,使用簡潔格式
|
# 2026-03-30 P1: CI/CD 告警偵測 - 配置化 (constants.py)
|
||||||
# CI/CD 告警 alertname 模式: CD_*, CI_*, E2E_*, SmokeTest, *_Build, *_Test
|
# 跳過 AI 仲裁,使用簡潔格式
|
||||||
# ==========================================================================
|
# ==========================================================================
|
||||||
cicd_prefixes = ("CD_", "CI_", "E2E_", "SmokeTest", "Build_", "Test_", "Deploy_")
|
if is_cicd_alertname(alertname):
|
||||||
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:
|
|
||||||
# CI/CD 告警 - 使用簡潔格式,不走 AI 仲裁
|
# CI/CD 告警 - 使用簡潔格式,不走 AI 仲裁
|
||||||
logger.info(
|
logger.info(
|
||||||
"alertmanager_cicd_detected",
|
"alertmanager_cicd_detected",
|
||||||
|
|||||||
@@ -43,3 +43,107 @@ APPROVAL_TO_INCIDENT_STATUS = {
|
|||||||
|
|
||||||
# Incident 狀態 → 是否活躍
|
# Incident 狀態 → 是否活躍
|
||||||
INCIDENT_ACTIVE_STATUSES = frozenset({"investigating", "mitigating"})
|
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
|
||||||
|
|||||||
@@ -20,6 +20,12 @@ import uuid
|
|||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import Protocol, runtime_checkable
|
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.logging import get_logger
|
||||||
from src.core.sse import EventType, SSEEvent, get_publisher
|
from src.core.sse import EventType, SSEEvent, get_publisher
|
||||||
from src.models.terminal import (
|
from src.models.terminal import (
|
||||||
@@ -511,7 +517,7 @@ class TerminalService:
|
|||||||
publisher, topic, "Approval-Query", "completed",
|
publisher, topic, "Approval-Query", "completed",
|
||||||
result={"pending_count": len(pending_approvals)}
|
result={"pending_count": len(pending_approvals)}
|
||||||
)
|
)
|
||||||
await asyncio.sleep(0.3)
|
await asyncio.sleep(SSE_DELAY_SECONDS)
|
||||||
|
|
||||||
if not pending_approvals:
|
if not pending_approvals:
|
||||||
await self._publish_thought(
|
await self._publish_thought(
|
||||||
@@ -522,14 +528,15 @@ class TerminalService:
|
|||||||
|
|
||||||
# 顯示待簽核清單摘要
|
# 顯示待簽核清單摘要
|
||||||
summary_lines = [f"發現 {len(pending_approvals)} 個待簽核項目:"]
|
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"
|
risk = approval.risk_level.value.upper() if approval.risk_level else "UNKNOWN"
|
||||||
summary_lines.append(
|
summary_lines.append(
|
||||||
f" {i}. [{risk}] {approval.action[:50]}"
|
f" {i}. [{risk}] {approval.action[:50]}"
|
||||||
)
|
)
|
||||||
|
|
||||||
if len(pending_approvals) > 5:
|
if len(pending_approvals) > MAX_APPROVAL_DISPLAY:
|
||||||
summary_lines.append(f" ... 還有 {len(pending_approvals) - 5} 個")
|
remaining = len(pending_approvals) - MAX_APPROVAL_DISPLAY
|
||||||
|
summary_lines.append(f" ... 還有 {remaining} 個")
|
||||||
|
|
||||||
await self._publish_thought(
|
await self._publish_thought(
|
||||||
publisher, topic, "Executor",
|
publisher, topic, "Executor",
|
||||||
@@ -561,11 +568,11 @@ class TerminalService:
|
|||||||
logger.error("approval_query_failed", error=str(e))
|
logger.error("approval_query_failed", error=str(e))
|
||||||
await self._publish_tool_call(
|
await self._publish_tool_call(
|
||||||
publisher, topic, "Approval-Query", "failed",
|
publisher, topic, "Approval-Query", "failed",
|
||||||
result={"error": str(e)[:100]}
|
result={"error": sanitize_error_message(str(e))}
|
||||||
)
|
)
|
||||||
await self._publish_thought(
|
await self._publish_thought(
|
||||||
publisher, topic, "System",
|
publisher, topic, "System",
|
||||||
f"查詢簽核項目失敗: {str(e)[:50]}"
|
f"查詢簽核項目失敗: {sanitize_error_message(str(e), ERROR_MESSAGE_DISPLAY_LENGTH)}"
|
||||||
)
|
)
|
||||||
|
|
||||||
async def _handle_metrics_query(
|
async def _handle_metrics_query(
|
||||||
|
|||||||
Reference in New Issue
Block a user