""" Auto-Approve Service - Phase 4 自動執行策略 ========================================== ADR-030: 智能自動修復系統 自動執行條件 (全部滿足才放行): 1. 風險等級 = LOW 2. 信任度 >= 90% (或 TrustEngine score >= 5) 3. 有匹配的 Playbook 且成功率 >= 95% 4. Playbook 成功執行次數 >= 3 設計原則: - 保守策略 (寧可人工審核，不可錯誤自動執行) - 完整審計追蹤 - CRITICAL 永遠不自動執行 版本: v1.0 建立: 2026-03-26 (台北時區) """ from dataclasses import dataclass, field from datetime import UTC, datetime from enum import Enum from typing import Any import structlog from src.models.playbook import Playbook from src.services.playbook_rag import PlaybookMatch from src.services.trust_engine import TrustScoreManager, get_trust_manager logger = structlog.get_logger(__name__) # ============================================================================= # Configuration # ============================================================================= class AutoApproveReason(str, Enum): """自動執行/拒絕原因""" # 自動執行 PLAYBOOK_MATCH = "playbook_match" # Playbook 匹配成功 TRUST_SCORE = "trust_score" # 信任分數達標 LOW_RISK = "low_risk" # 低風險操作 # 拒絕自動執行 HIGH_RISK = "high_risk" # 風險過高 CRITICAL_OPERATION = "critical_operation" # 關鍵操作 LOW_TRUST = "low_trust" # 信任不足 NO_PLAYBOOK = "no_playbook" # 無匹配 Playbook LOW_SUCCESS_RATE = "low_success_rate" # Playbook 成功率不足 INSUFFICIENT_HISTORY = "insufficient_history" # 執行歷史不足 @dataclass class AutoApproveConfig: """自動執行配置""" # 風險等級閾值 # 2026-04-11 Claude Sonnet 4.6: ADR-070 全自動化方向 — low/medium/high 全開放 # 真正需要人工的由 DESTRUCTIVE_PATTERNS 攔截（scale=0, delete, drop） # 原: ["low", "medium"] → 導致所有 high risk 告警永遠走人工審核 allowed_risk_levels: list[str] = field( default_factory=lambda: ["low", "medium", "high"] ) # 信任度閾值 # 2026-04-10 Claude Sonnet 4.6: trust_score 是 in-memory，Pod 重啟歸零 # → 改為 0，讓 medium risk + confidence >= 0.65 的操作直接自動執行 # 歷史原因: min_trust_score=1 導致所有告警永遠走審批，從未自動修復 min_trust_score: int = 0 # 不要求執行歷史 (原: 1) # 2026-04-11 Claude Sonnet 4.6: ADR-070 全自動化 — 0.5 即可執行 # 真正風險由 DESTRUCTIVE_PATTERNS + risk_level=critical 把關 min_confidence: float = 0.50 # AI 有基本把握即可 (原: 0.90, 後: 0.65) # Playbook 閾值 # 2026-04-01 ogt: 降低啟動門檻，1次成功記錄即可 min_playbook_success_rate: float = 0.80 # 成功率 >= 80% (原: 95%) min_playbook_success_count: int = 1 # 成功次數 >= 1 (原: 3) # 功能開關 enabled: bool = True # 總開關 require_playbook: bool = False # 不強制要求 Playbook (原: True)，Gemini 分析直接可執行 audit_all: bool = True # 是否記錄所有判斷 # 預設配置 (實用策略 - 2026-04-01 ogt Phase 22 P1) DEFAULT_CONFIG = AutoApproveConfig() # ============================================================================= # 破壞性指令攔截清單 (ADR-070, 2026-04-11 Claude Sonnet 4.6) # C3+M1 修復 (Code Review 2026-04-11): 移至模組常量 + 補全 K8s/Docker 高風險操作 # 原則: 可恢復操作 → 自動執行; 不可逆 / 業務衝擊 → 人工確認 # ============================================================================= _DESTRUCTIVE_PATTERNS: list[str] = [ # --- Scale to zero (停機) --- "--replicas=0", # kubectl scale --replicas=0 '"replicas": 0', # kubectl patch JSON patch "'replicas': 0", # kubectl patch 單引號變體 "replicas=0", # 任何形式的 replicas=0 # --- K8s 刪除操作 --- "delete pod", # 強制刪除 pod (kubectl delete pod / pods) "delete pods", # 複數形式 "delete deployment", # 刪除 deployment "delete pvc", # 刪除 PVC (資料丟失) "delete namespace", # 刪除 namespace "kubectl drain", # 驅逐節點所有 pod "kubectl cordon", # 封鎖節點（業務影響） "kubectl rollout undo", # 回滾部署（需人工確認版本） # --- Docker 破壞性操作 --- "docker rm", # 刪除容器 "docker stop", # 停止容器（不同於 restart） "docker kill", # 強制殺死容器 # --- DB DDL (不可逆) --- "drop table", "drop database", "truncate table", # --- SSH 危險指令 --- "rm -rf", # 遞迴刪除 "rm -f /", # 刪除根目錄 ] # ============================================================================= # Data Models # ============================================================================= @dataclass class AutoApproveDecision: """自動執行決策結果""" should_auto_approve: bool reason: AutoApproveReason reason_detail: str # 判斷依據 risk_level: str trust_score: int confidence: float playbook_match: PlaybookMatch | None = None playbook_success_rate: float | None = None playbook_success_count: int | None = None # 時間戳 decided_at: datetime = field(default_factory=lambda: datetime.now(UTC)) def to_dict(self) -> dict[str, Any]: return { "should_auto_approve": self.should_auto_approve, "reason": self.reason.value, "reason_detail": self.reason_detail, "risk_level": self.risk_level, "trust_score": self.trust_score, "confidence": self.confidence, "playbook_match": self.playbook_match.to_dict() if self.playbook_match else None, "playbook_success_rate": self.playbook_success_rate, "playbook_success_count": self.playbook_success_count, "decided_at": self.decided_at.isoformat(), } def to_audit_log(self) -> str: """生成審計日誌""" status = "AUTO_APPROVED" if self.should_auto_approve else "REQUIRES_HUMAN" return ( f"[{status}] {self.reason.value}: {self.reason_detail} " f"(risk={self.risk_level}, trust={self.trust_score}, conf={self.confidence:.0%})" ) # ============================================================================= # Auto-Approve Policy # ============================================================================= class AutoApprovePolicy: """ 自動執行策略 判斷提案是否可以跳過人工審核直接執行 核心原則: - CRITICAL 永遠不自動執行 - 必須有足夠的歷史成功記錄 - 信任度達標 - 風險等級為 LOW """ def __init__( self, config: AutoApproveConfig | None = None, trust_manager: TrustScoreManager | None = None, ): self.config = config or DEFAULT_CONFIG self._trust_manager = trust_manager @property def trust_manager(self) -> TrustScoreManager: """Lazy load trust manager""" if self._trust_manager is None: self._trust_manager = get_trust_manager() return self._trust_manager def evaluate( self, proposal_data: dict[str, Any], playbook: Playbook | None = None, playbook_match: PlaybookMatch | None = None, ) -> AutoApproveDecision: """ 評估提案是否可自動執行 Args: proposal_data: 提案資料 (含 risk_level, confidence, action 等) playbook: 匹配的 Playbook (可選) playbook_match: RAG 匹配結果 (可選) Returns: AutoApproveDecision 包含決策結果和原因 """ # 基本資訊 risk_level = proposal_data.get("risk_level", "medium").lower() confidence = proposal_data.get("confidence", 0.0) # 🔴 無信心度=規則匹配 action = proposal_data.get("action", "") or proposal_data.get("kubectl_command", "") action_pattern = self._extract_action_pattern(action) # 取得信任分數 trust_record = self.trust_manager.get_trust_record(action_pattern) trust_score = trust_record.score if trust_record else 0 # Playbook 資訊 playbook_success_rate = playbook.success_rate if playbook else None playbook_success_count = playbook.success_count if playbook else None # ========== 檢查條件 ========== # 條件 0: 功能是否啟用 if not self.config.enabled: return self._reject( reason=AutoApproveReason.LOW_TRUST, detail="Auto-approve is disabled", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 1: CRITICAL 永遠不自動執行 if risk_level == "critical": return self._reject( reason=AutoApproveReason.CRITICAL_OPERATION, detail="CRITICAL operations always require human approval", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 1b: 破壞性指令攔截 (ADR-070: 2026-04-11 Claude Sonnet 4.6) # 即使是 low/medium risk，以下操作仍需人工確認 # 原則: 可恢復操作 → 自動執行; 不可逆 / 業務衝擊 → 人工 # M1+C3 修復 2026-04-11 (Code Review): 移至模組常量 + 補全 K8s/Docker 高風險操作 action_lower = action.lower() for pattern in _DESTRUCTIVE_PATTERNS: if pattern in action_lower: return self._reject( reason=AutoApproveReason.CRITICAL_OPERATION, detail=f"Destructive pattern detected: '{pattern}' in action — requires human approval", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 2: 風險等級必須在允許列表中 if risk_level not in self.config.allowed_risk_levels: return self._reject( reason=AutoApproveReason.HIGH_RISK, detail=f"Risk level '{risk_level}' not in allowed list {self.config.allowed_risk_levels}", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 3: 信任分數 if trust_score < self.config.min_trust_score: return self._reject( reason=AutoApproveReason.LOW_TRUST, detail=f"Trust score {trust_score} < {self.config.min_trust_score}", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 4: AI 信心度 # 2026-04-15 Claude Sonnet 4.6 (飛輪沉默節點 1 修復): # 規則匹配的 confidence 固定 0.0（ADR-073 防偽造），會被此條件擋下 # 但 YAML 規則是人工審核過的，應直接信任 → bypass min_confidence # 改用「Playbook 成功率」或「規則 source」判斷可信度 _is_rule_based = ( proposal_data.get("is_rule_based") is True or proposal_data.get("source") == "expert_system" or (proposal_data.get("rule_id") or "") != "" or (proposal_data.get("matched_rule") or "") != "" ) if not _is_rule_based and confidence < self.config.min_confidence: return self._reject( reason=AutoApproveReason.LOW_TRUST, detail=f"Confidence {confidence:.0%} < {self.config.min_confidence:.0%}", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 5: Playbook 匹配 (如果要求) if self.config.require_playbook: if playbook is None: return self._reject( reason=AutoApproveReason.NO_PLAYBOOK, detail="No matching Playbook found", risk_level=risk_level, trust_score=trust_score, confidence=confidence, ) # 條件 6: Playbook 成功率 if playbook_success_rate is not None: if playbook_success_rate < self.config.min_playbook_success_rate: return self._reject( reason=AutoApproveReason.LOW_SUCCESS_RATE, detail=f"Playbook success rate {playbook_success_rate:.0%} < {self.config.min_playbook_success_rate:.0%}", risk_level=risk_level, trust_score=trust_score, confidence=confidence, playbook_match=playbook_match, playbook_success_rate=playbook_success_rate, playbook_success_count=playbook_success_count, ) # 條件 7: Playbook 成功次數 if playbook_success_count is not None: if playbook_success_count < self.config.min_playbook_success_count: return self._reject( reason=AutoApproveReason.INSUFFICIENT_HISTORY, detail=f"Playbook success count {playbook_success_count} < {self.config.min_playbook_success_count}", risk_level=risk_level, trust_score=trust_score, confidence=confidence, playbook_match=playbook_match, playbook_success_rate=playbook_success_rate, playbook_success_count=playbook_success_count, ) # ========== 所有條件通過 ========== return self._approve( reason=AutoApproveReason.PLAYBOOK_MATCH if playbook else AutoApproveReason.TRUST_SCORE, detail=f"All conditions met: risk={risk_level}, trust={trust_score}, confidence={confidence:.0%}", risk_level=risk_level, trust_score=trust_score, confidence=confidence, playbook_match=playbook_match, playbook_success_rate=playbook_success_rate, playbook_success_count=playbook_success_count, ) def _approve( self, reason: AutoApproveReason, detail: str, **kwargs, ) -> AutoApproveDecision: """建立自動執行決策""" decision = AutoApproveDecision( should_auto_approve=True, reason=reason, reason_detail=detail, **kwargs, ) if self.config.audit_all: logger.info( "auto_approve_decision", approved=True, reason=reason.value, detail=detail, trust_score=kwargs.get("trust_score"), ) return decision def _reject( self, reason: AutoApproveReason, detail: str, **kwargs, ) -> AutoApproveDecision: """建立拒絕自動執行決策""" decision = AutoApproveDecision( should_auto_approve=False, reason=reason, reason_detail=detail, **kwargs, ) if self.config.audit_all: logger.debug( "auto_approve_decision", approved=False, reason=reason.value, detail=detail, trust_score=kwargs.get("trust_score"), ) return decision def _extract_action_pattern(self, action: str) -> str: """ 從 action 字串提取 pattern 例如: - "kubectl rollout restart deployment/awoooi-api" → "rollout_restart:awoooi-api" - "kubectl scale deployment/nginx --replicas=3" → "scale:nginx" """ if not action: return "unknown" parts = action.split() if len(parts) < 3: return "unknown" # kubectl / verb = parts[1] if len(parts) > 1 else "unknown" resource_part = parts[2] if len(parts) > 2 else "" if "/" in resource_part: resource_name = resource_part.split("/")[-1] else: resource_name = resource_part # 移除可能的選項 resource_name = resource_name.split()[0] if " " in resource_name else resource_name return f"{verb}:{resource_name}" # ============================================================================= # Singleton # ============================================================================= _auto_approve_policy: AutoApprovePolicy | None = None def get_auto_approve_policy() -> AutoApprovePolicy: """取得自動執行策略 singleton""" global _auto_approve_policy if _auto_approve_policy is None: _auto_approve_policy = AutoApprovePolicy() return _auto_approve_policy