fix(telegram): sanitize failover alert errors
All checks were successful
Code Review / ai-code-review (push) Successful in 10s
CD Pipeline / tests (push) Successful in 1m5s
CD Pipeline / build-and-deploy (push) Successful in 3m25s
CD Pipeline / post-deploy-checks (push) Successful in 1m16s

This commit is contained in:
Your Name
2026-05-06 16:45:29 +08:00
parent a94435f143
commit 8f715fd3f2
5 changed files with 101 additions and 11 deletions

View File

@@ -12,7 +12,8 @@ from __future__ import annotations
import hashlib
import json
from datetime import datetime, timezone, timedelta
import re
from datetime import datetime, timedelta, timezone
from typing import Any
import structlog
@@ -23,6 +24,7 @@ RECOVERY_DEDUP_TTL_SEC = 3600 # 1h — GCP 健康閃爍時 1 小時內不重複
QUOTA_DEDUP_TTL_SEC = 86400 # 24h每日 quota 告警只發一次)
logger = structlog.get_logger(__name__)
_TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s]+")
class FailoverAlerter:
@@ -157,7 +159,7 @@ class FailoverAlerter:
if impact_lines:
sections.append(f"\n*影響*\n{impact_lines}")
if remediation_lines or next_action_line:
sections.append(f"\n*修復方向*")
sections.append("\n*修復方向*")
if remediation_lines:
sections.append(remediation_lines)
if next_action_line:
@@ -285,8 +287,8 @@ class FailoverAlerter:
2026-04-25 P1.5 by Claude Engineer-D — 告警失敗不能阻斷主流程
"""
try:
from src.services.telegram_gateway import get_telegram_gateway
from src.core.config import get_settings
from src.services.telegram_gateway import get_telegram_gateway
settings = get_settings()
chat_id = getattr(settings, "SRE_GROUP_CHAT_ID", None) or getattr(settings, "OPENCLAW_TG_CHAT_ID", None)
@@ -299,7 +301,13 @@ class FailoverAlerter:
logger.info("telegram_failover_alert_sent", message_len=len(message))
except Exception as e:
# 不 raise — 告警失敗不該阻斷主流程(鐵律)
logger.exception("telegram_failover_send_failed", error=str(e))
# 2026-05-06 Codex: Telegram/httpx exception 字串可能包含 bot token URL
# 禁止用 logger.exception 輸出 chained traceback。
logger.warning(
"telegram_failover_send_failed",
error=_sanitize_telegram_error(str(e)),
error_type=type(e).__name__,
)
# -------------------------------------------------------------------------
@@ -319,6 +327,11 @@ def _escape_md(text: str) -> str:
return text
def _sanitize_telegram_error(text: str) -> str:
"""遮蔽 Telegram Bot URL 中的 token避免例外訊息寫入 log。"""
return _TELEGRAM_BOT_URL_RE.sub(r"\1<redacted>", text)
def _as_dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
@@ -334,7 +347,7 @@ def _lines_from_dict(data: dict[str, Any], max_items: int = 20, compact: bool =
rows.append(f"{_escape_md(str(k))}{_escape_md(str(data.get(k)))}")
idx += 1
if compact and len(rows) >= max_items:
rows.append("...(更多欄位略)")
rows.append(_escape_md("...(更多欄位略)"))
return "\n".join(f" {line}" for line in rows)
@@ -342,7 +355,7 @@ def _lines_from_list(value: Any) -> str:
if not isinstance(value, list):
return ""
return "\n".join(
f" {idx + 1}. {_escape_md(str(item))}"
f" {_escape_md(str(idx + 1))}\\. {_escape_md(str(item))}"
for idx, item in enumerate(value)
)

View File

@@ -25,6 +25,7 @@ SOUL.md 鐵律 (4.1 Telegram 訊息壓縮原則):
import asyncio
import html
import os
import re
from dataclasses import dataclass
from datetime import UTC, datetime
@@ -58,6 +59,12 @@ POLLING_LEADER_RENEW = 20 # seconds - 每 20s 續約
POLLING_LEADER_WATCH = 30 # seconds - 非 Leader Pod 每 30s 嘗試接管
logger = structlog.get_logger(__name__)
_TELEGRAM_BOT_URL_RE = re.compile(r"(api\.telegram\.org/bot)[^/\s]+")
def _sanitize_telegram_error(text: str) -> str:
"""遮蔽 Telegram Bot URL 中的 token避免例外字串污染 log / trace。"""
return _TELEGRAM_BOT_URL_RE.sub(r"\1<redacted>", text)
# 2026-04-27 Claude Sonnet 4.6: B3 — LLM 動態 Telegram 按鈕 Feature Flag
# true → 優先使用 ActionPlan.recommended_actions 動態生成按鈕
@@ -1468,20 +1475,28 @@ class TelegramGateway:
except httpx.HTTPStatusError as e:
span.set_attribute("telegram.http_status", e.response.status_code)
span.set_status(trace.Status(trace.StatusCode.ERROR))
span.record_exception(e)
span.record_exception(
TelegramGatewayError(f"HTTP error: {e.response.status_code}")
)
logger.error("telegram_api_error", method=method, status=e.response.status_code,
response_body=e.response.text[:500])
raise TelegramGatewayError(f"HTTP error: {e.response.status_code}") from e
raise TelegramGatewayError(f"HTTP error: {e.response.status_code}") from None
except TelegramGatewayError:
# 已處理的錯誤,直接拋出
raise
except Exception as e:
safe_error = _sanitize_telegram_error(str(e))
span.set_status(trace.Status(trace.StatusCode.ERROR))
span.record_exception(e)
logger.error("telegram_request_failed", method=method, error=str(e))
raise TelegramGatewayError(str(e)) from e
span.record_exception(TelegramGatewayError(safe_error))
logger.error(
"telegram_request_failed",
method=method,
error=safe_error,
error_type=type(e).__name__,
)
raise TelegramGatewayError(safe_error) from None
async def _build_inline_keyboard(
self,