feat(aiops): normalize multi-signal alert cards

apps/api/src/services/telegram_gateway.py

@@ -108,6 +108,160 @@ _HOST_PROCESS_LINE_RE = re.compile(
     r"(?:\S+\s+){6}"
     r"(?P<command>.+?)\s*$"
 )
+_PRIVATE_IP_RE = re.compile(
+    r"\b(?:10|192\.168|172\.(?:1[6-9]|2\d|3[0-1]))(?:\.\d{1,3}){2}\b"
+)
+_URL_RE = re.compile(r"https?://\S+")
+_ABSOLUTE_PATH_RE = re.compile(r"/(?:[\w@.+-]+/){2,}[\w@.+-]*")
+_SECRET_ASSIGNMENT_RE = re.compile(
+    r"(?i)\b(token|secret|password|authorization|api[_-]?key|bearer)\b\s*[:=]\s*\S+"
+)
+_BEARER_RE = re.compile(r"(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+")
+_AI_SIGNAL_TARGET_LABEL_RE = re.compile(
+    r"\b(?:host|target|service|domain|agent\.name|node|route)\s*=\s*\"?(?P<target>[A-Za-z0-9_.:-]+)\"?"
+)
+
+
+@dataclass(frozen=True)
+class AiOpsSignalSpec:
+    event_type: str
+    title: str
+    lane: str
+    source_tokens: tuple[str, ...]
+    gate_tokens: tuple[str, ...]
+    impact: str
+    next_step: str
+    recommendation_lines: tuple[str, ...]
+    forbidden_summary: str
+    default_severity: str = "P2"
+
+
+_AI_OPS_SIGNAL_SPECS: tuple[AiOpsSignalSpec, ...] = (
+    AiOpsSignalSpec(
+        event_type="wazuh_intrusion_signal",
+        title="Wazuh 入侵訊號",
+        lane="security_intrusion_triage",
+        source_tokens=(
+            "wazuh",
+            "rule.level",
+            "syscheck",
+            "authentication_failed",
+            "authentication failure",
+            "wazuh_alert",
+        ),
+        gate_tokens=("alert", "rule.level", "intrusion", "critical", "auth", "fired"),
+        impact="可能涉及入侵、權限濫用或檔案完整性變更；先轉 SIEM evidence，不直接封鎖或改 firewall",
+        next_step="建立 Wazuh security triage candidate，要求 agent、rule、severity、evidence ref 與 reviewer 欄位",
+        recommendation_lines=(
+            "├ 彙整 Wazuh rule / agent / evidence ref，建立 security incident candidate",
+            "├ 要求 owner 補影響範圍、誤報判斷、封鎖條件與驗證計畫",
+            "└ 回寫 Timeline / KM / PlayBook，Verifier 需讀回 SIEM 或 no-write evidence",
+        ),
+        forbidden_summary="不封鎖 IP、不改 firewall、不停服務、不貼 raw log；除非 owner / window / verifier 齊全。",
+        default_severity="P1",
+    ),
+    AiOpsSignalSpec(
+        event_type="kali_assessment_signal",
+        title="Kali 評估訊號",
+        lane="security_assessment_review",
+        source_tokens=("kali_assessment_signal", "kali", "nmap", "nikto", "sqlmap", "metasploit"),
+        gate_tokens=("assessment", "finding", "scan", "vulnerability", "critical", "high"),
+        impact="安全評估結果需要轉成 owner review；不得把 Kali active scan 當成已授權 remediation",
+        next_step="建立 Kali finding review candidate，分離 passive evidence、active scan request 與 remediation gate",
+        recommendation_lines=(
+            "├ 將 finding 轉成脫敏 evidence ref，不把完整掃描輸出貼進 Telegram",
+            "├ 補 owner scope、允許的測試窗口、目標清單與 false positive 判斷",
+            "└ 若需 active scan 或修復，必須另開批准與維護窗口",
+        ),
+        forbidden_summary="不觸發 active scan、不打 exploit、不改服務、不改 firewall；目前僅建立 review candidate。",
+    ),
+    AiOpsSignalSpec(
+        event_type="nginx_config_drift",
+        title="Nginx 配置漂移",
+        lane="public_gateway_config_drift",
+        source_tokens=("nginx_config_drift", "nginx drift", "nginx", "vhost", "nginx -t"),
+        gate_tokens=("drift", "config", "reload", "failed", "nginx -t", "route"),
+        impact="公開入口配置可能與 repo / rendered diff 不一致；先要求脫敏 live conf 與 route health readback",
+        next_step="建立 Public Gateway rendered diff candidate，等待 redacted export、nginx -t evidence 與 rollback owner",
+        recommendation_lines=(
+            "├ 比對 repo source、rendered diff、redacted live export 與 route ownership",
+            "├ 補 nginx -t、reload/no-reload、route smoke、TLS/ACME 與 rollback evidence",
+            "└ 將差異回寫 Public Gateway Gate，不直接 reload",
+        ),
+        forbidden_summary="不讀 raw live conf、不執行 nginx -t、不 reload、不改 DNS/TLS；只建立漂移候選。",
+    ),
+    AiOpsSignalSpec(
+        event_type="backup_restore_escrow_signal",
+        title="備份 / 還原 / 金庫訊號",
+        lane="backup_restore_escrow_triage",
+        source_tokens=(
+            "backupaggregaterunfailed",
+            "backupconfigcapturepartial",
+            "coldstartrecoveryblocked",
+            "escrow_missing",
+            "restic",
+            "rclone",
+            "offsite",
+            "backup",
+        ),
+        gate_tokens=("failed", "missing", "blocked", "stale", "escrow", "restore"),
+        impact="資料保護或 DR 信任度下降；先讀 freshness / escrow / restore evidence，不假性標綠",
+        next_step="建立 backup/restore readback candidate，要求 freshness SLO、restore target、escrow metadata 與 no-false-green 判讀",
+        recommendation_lines=(
+            "├ 讀回 backup-status、offsite verify、escrow missing count 與 restore drill evidence",
+            "├ 缺 escrow 或 restore evidence 時標 blocked，不偽造 marker",
+            "└ 下一步沉澱到 DR scorecard、KM / PlayBook / Verifier",
+        ),
+        forbidden_summary="不跑 backup/restore、不 remote delete、不寫 escrow marker、不改 retention；只建立 evidence candidate。",
+        default_severity="P1",
+    ),
+    AiOpsSignalSpec(
+        event_type="provider_freshness_signal",
+        title="Provider freshness 訊號",
+        lane="source_provider_freshness_triage",
+        source_tokens=(
+            "sourceprovideringestionstale",
+            "provider_freshness",
+            "ingestion stale",
+            "sentry",
+            "signoz",
+            "freshness",
+        ),
+        gate_tokens=("stale", "freshness", "window", "provider", "ingestion", "timeout"),
+        impact="觀測來源可能過期或 ingestion 中斷；不能把 route 200 或 dashboard up 當健康",
+        next_step="建立 provider freshness candidate，要求 provider window、last seen、direct ref 與 verifier readback",
+        recommendation_lines=(
+            "├ 對齊 Sentry / SigNoz / provider freshness window 與 direct reference",
+            "├ 標出缺資料、過期、timeout 與 no-false-green 風險",
+            "└ 回寫報表資料信任度與 AgentOps readiness，不直接切 provider",
+        ),
+        forbidden_summary="不切 provider、不呼叫付費模型、不改 env、不 reload；先補 freshness evidence。",
+    ),
+    AiOpsSignalSpec(
+        event_type="supply_chain_drift",
+        title="供應鏈漂移訊號",
+        lane="supply_chain_drift_review",
+        source_tokens=(
+            "dependency_supply_chain_drift",
+            "supply-chain",
+            "supply chain",
+            "npm audit",
+            "pip-audit",
+            "cve-",
+            "dependency drift",
+            "package drift",
+        ),
+        gate_tokens=("drift", "audit", "vulnerability", "cve-", "package", "dependency"),
+        impact="依賴、套件或 runner build 來源有風險；需接 package owner 與 CI evidence，不直接升級",
+        next_step="建立 supply-chain review candidate，要求 package scope、CVE/evidence ref、CI run 與 rollback plan",
+        recommendation_lines=(
+            "├ 彙整 package、lockfile、CVE/evidence ref、CI run 與 affected surface",
+            "├ 判斷是否需要 owner review、maintenance window 與 rollback plan",
+            "└ 產出 upgrade/downgrade/dismiss candidate，等待 reviewer acceptance",
+        ),
+        forbidden_summary="不直接升級套件、不改 lockfile、不重跑 deploy、不修改 secrets；先建立 review candidate。",
+    ),
+)
 
 
 def _is_host_resource_alert_text(text: str) -> bool:
@@ -122,6 +276,120 @@ def _is_host_resource_alert_text(text: str) -> bool:
     )
 
 
+def _contains_any_token(text: str, tokens: tuple[str, ...]) -> bool:
+    lowered = text.lower()
+    return any(token.lower() in lowered for token in tokens)
+
+
+def _classify_aiops_signal_alert(text: str) -> AiOpsSignalSpec | None:
+    for spec in _AI_OPS_SIGNAL_SPECS:
+        if _contains_any_token(text, spec.source_tokens) and _contains_any_token(
+            text,
+            spec.gate_tokens,
+        ):
+            return spec
+    return None
+
+
+def _is_aiops_signal_alert_text(text: str) -> bool:
+    return _classify_aiops_signal_alert(text) is not None
+
+
+def _sanitize_aiops_signal_value(value: str) -> str:
+    sanitized = _BEARER_RE.sub("Bearer [redacted-secret]", value)
+    sanitized = _SECRET_ASSIGNMENT_RE.sub(lambda m: f"{m.group(1)}=[redacted-secret]", sanitized)
+    sanitized = _URL_RE.sub("[redacted-url]", sanitized)
+    sanitized = _PRIVATE_IP_RE.sub("[redacted-private-ip]", sanitized)
+    sanitized = _ABSOLUTE_PATH_RE.sub("[redacted-path]", sanitized)
+    sanitized = re.sub(r"\s+", " ", sanitized).strip()
+    return sanitized[:140]
+
+
+def _aiops_signal_target(text: str, spec: AiOpsSignalSpec) -> str:
+    match = _AI_SIGNAL_TARGET_LABEL_RE.search(text)
+    if match:
+        target = _sanitize_aiops_signal_value(match.group("target"))
+        return target or spec.event_type
+    if spec.event_type == "backup_restore_escrow_signal":
+        return "backup_restore"
+    if spec.event_type == "nginx_config_drift":
+        return "public_gateway"
+    if spec.event_type == "supply_chain_drift":
+        return "source_control"
+    return spec.event_type
+
+
+def _aiops_signal_severity(text: str, spec: AiOpsSignalSpec) -> str:
+    lowered = text.lower()
+    level_match = re.search(r"\brule\.level\s*=\s*\"?(?P<level>\d+)\"?", text)
+    if level_match and int(level_match.group("level")) >= 10:
+        return "P1"
+    if any(token in lowered for token in ("p0", "critical", "crit", "failed", "blocked")):
+        return "P1"
+    if any(token in lowered for token in ("p1", "high", "warning", "warn", "stale")):
+        return "P2"
+    return spec.default_severity
+
+
+def _aiops_signal_evidence_lines(text: str) -> list[str]:
+    candidate_lines: list[str] = []
+    for raw_line in text.splitlines():
+        stripped = raw_line.strip()
+        if not stripped:
+            continue
+        if len(candidate_lines) >= 3:
+            break
+        sanitized = _sanitize_aiops_signal_value(stripped)
+        if not sanitized:
+            continue
+        candidate_lines.append(sanitized)
+
+    if not candidate_lines:
+        return ["└ 尚未收到可解析的脫敏 evidence，請補 owner evidence ref。"]
+
+    lines = [f"├ <code>{html.escape(line)}</code>" for line in candidate_lines]
+    lines[-1] = "└" + lines[-1][1:]
+    return lines
+
+
+def format_aiops_signal_alert_card(text: str) -> str:
+    """把資安 / 維運 raw signal 轉成 AI 自動化事件包。"""
+    spec = _classify_aiops_signal_alert(text)
+    if spec is None:
+        return text
+
+    target = _aiops_signal_target(text, spec)
+    severity = _aiops_signal_severity(text, spec)
+    evidence_lines = _aiops_signal_evidence_lines(text)
+
+    return "\n".join(
+        [
+            f"<b>{severity} AIOps 訊號｜{html.escape(spec.title)}</b>",
+            "<code>ai_automation_alert_card_v1</code>",
+            "",
+            "<b>一眼摘要</b>",
+            f"├ 事件類型：<code>{html.escape(spec.event_type)}</code>",
+            f"├ Target：<code>{html.escape(target)}</code>",
+            f"├ 影響：<b>{html.escape(spec.impact)}</b>",
+            "└ 產品判讀：<code>raw_signal_normalized</code> / <code>no_runtime_write</code>",
+            "",
+            "<b>AI 自動化判讀</b>",
+            f"├ Lane：<code>{html.escape(spec.lane)}</code>",
+            "├ Gate：<code>candidate_only</code> / <code>runtime_write_gate=0</code>",
+            f"└ 下一步：{html.escape(spec.next_step)}",
+            "",
+            "<b>Top evidence</b>",
+            *evidence_lines,
+            "",
+            "<b>建議下一步</b>",
+            *spec.recommendation_lines,
+            "",
+            "<b>禁止事項</b>",
+            f"└ {html.escape(spec.forbidden_summary)}",
+        ]
+    )
+
+
 def _resource_load_bar(load_text: str) -> str:
     try:
         load = max(float(load_text), 0.0)
@@ -387,9 +655,11 @@ def format_host_resource_alert_card(text: str) -> str:
 
 
 def normalize_alert_notification_payload(text: str, parse_mode: str) -> tuple[str, str]:
-    """最後出口統一脫敏監控告警；host 類 raw dump 一律改成 HTML 事件卡。"""
+    """最後出口統一脫敏監控告警；raw signal 一律改成 HTML 事件卡。"""
     if _is_host_resource_alert_text(text):
         return format_host_resource_alert_card(text), "HTML"
+    if _is_aiops_signal_alert_text(text):
+        return format_aiops_signal_alert_card(text), "HTML"
     return text, parse_mode