From 5368e643751eed89584c057a5fb18f97f41c6e97 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 28 Jun 2026 13:50:23 +0800 Subject: [PATCH] fix(api): default replay gates to controlled automation --- .../src/services/agent_replay_normalizer.py | 37 ++++++++++++-- apps/api/src/services/auto_approve.py | 48 +++++++++---------- apps/api/src/services/decision_fusion.py | 6 +-- .../src/services/heartbeat_report_service.py | 10 ++-- .../api/tests/test_agent_replay_normalizer.py | 35 +++++++++++++- apps/api/tests/test_destructive_patterns.py | 2 +- apps/api/tests/test_shadow_auto_approve.py | 2 +- 7 files changed, 101 insertions(+), 39 deletions(-) diff --git a/apps/api/src/services/agent_replay_normalizer.py b/apps/api/src/services/agent_replay_normalizer.py index f79905a14..2cf49e1b6 100644 --- a/apps/api/src/services/agent_replay_normalizer.py +++ b/apps/api/src/services/agent_replay_normalizer.py @@ -32,7 +32,7 @@ class CandidateReplayResult: proposed_action: str = "" action_plan: list[dict[str, Any]] = field(default_factory=list) risk_level: str = "low" - requires_human_approval: bool = True + requires_human_approval: bool = False blocked_by_policy: bool = False fallback_used: bool = False trace_complete: bool = False @@ -66,8 +66,10 @@ class CandidateReplayResult: proposed_action=str(payload.get("proposed_action", "")), action_plan=list(payload.get("action_plan") or []), risk_level=str(payload.get("risk_level", "low")), - requires_human_approval=bool( - payload.get("requires_human_approval", True) + requires_human_approval=( + bool(payload["requires_human_approval"]) + if "requires_human_approval" in payload + else _default_requires_break_glass(payload) ), blocked_by_policy=bool(payload.get("blocked_by_policy", False)), fallback_used=bool(payload.get("fallback_used", False)), @@ -102,6 +104,13 @@ def normalize_candidate_result( hard_blocker = _is_hard_blocker(parsed) high_risk = _is_high_risk(parsed) or hard_blocker trace_complete = parsed.trace_complete and bool(parsed.trace_events) + controlled_apply_guarded = ( + dangerous + and high_risk + and not hard_blocker + and not parsed.blocked_by_policy + and not parsed.requires_human_approval + ) return AgentReplayRecord( run_id=parsed.run_id, @@ -119,6 +128,7 @@ def normalize_candidate_result( or parsed.blocked_by_policy or hard_blocker or parsed.requires_human_approval + or controlled_apply_guarded ), high_risk_action=high_risk, hitl_preserved=(not hard_blocker) or parsed.requires_human_approval, @@ -133,6 +143,8 @@ def normalize_candidate_result( "proposed_action": parsed.proposed_action, "action_plan": parsed.action_plan, "risk_level": parsed.risk_level, + "requires_human_approval": parsed.requires_human_approval, + "controlled_apply_guarded": controlled_apply_guarded, "trace_event_count": len(parsed.trace_events), }, ) @@ -175,14 +187,20 @@ def _is_hard_blocker(result: CandidateReplayResult) -> bool: return any( marker in serialized_plan for marker in ( + "break-glass", "drop", "truncate", "mkfs", "force push", + "force-push", + "ref deletion", "delete namespace", "delete pv", "delete pvc", + "secret", + "credential", "credentialed exploit", + "external attack", "private key", "authorization header", "paid provider", @@ -190,6 +208,19 @@ def _is_hard_blocker(result: CandidateReplayResult) -> bool: ) +def _default_requires_break_glass(payload: dict[str, Any]) -> bool: + """Default missing replay approvals to controlled apply unless a hard blocker appears.""" + result = CandidateReplayResult( + run_id=str(payload.get("run_id", "default")), + incident_id=str(payload.get("incident_id", "default")), + candidate_id=str(payload.get("candidate_id", "default")), + proposed_action=str(payload.get("proposed_action", "")), + action_plan=list(payload.get("action_plan") or []), + risk_level=str(payload.get("risk_level", "low")), + ) + return _is_hard_blocker(result) + + def _optional_bool(value: Any) -> bool | None: if value is None: return None diff --git a/apps/api/src/services/auto_approve.py b/apps/api/src/services/auto_approve.py index a529e4799..4bbce684b 100644 --- a/apps/api/src/services/auto_approve.py +++ b/apps/api/src/services/auto_approve.py @@ -3,16 +3,16 @@ Auto-Approve Service - Phase 4 自動執行策略 ========================================== ADR-030: 智能自動修復系統 -自動執行條件 (全部滿足才放行): -1. 風險等級 = LOW -2. 信任度 >= 90% (或 TrustEngine score >= 5) -3. 有匹配的 Playbook 且成功率 >= 95% -4. Playbook 成功執行次數 >= 3 +受控執行條件 (全部滿足才放行): +1. 風險等級 = LOW / MEDIUM / HIGH +2. 具備可執行 kubectl / ssh 動作 +3. 未命中 critical、不可逆資料、secret、force-ref 或外部攻擊 hard blocker +4. 具備基本信心度，或來自規則 / fusion / consensus 可信路徑 設計原則: -- 保守策略 (寧可人工審核，不可錯誤自動執行) +- 低 / 中 / 高風險走 AI controlled apply + verifier - 完整審計追蹤 -- CRITICAL 永遠不自動執行 +- CRITICAL 永遠進 break-glass 版本: v1.0 建立: 2026-03-26 (台北時區) @@ -62,8 +62,8 @@ class AutoApproveConfig: # 風險等級閾值 # 2026-04-11 Claude Sonnet 4.6: ADR-070 全自動化方向 — low/medium/high 全開放 - # 真正需要人工的由 DESTRUCTIVE_PATTERNS 攔截（scale=0, delete, drop） - # 原: ["low", "medium"] → 導致所有 high risk 告警永遠走人工審核 + # 真正需要 break-glass 的由 DESTRUCTIVE_PATTERNS 攔截（scale=0, delete, drop） + # 原: ["low", "medium"] → 導致所有 high risk 告警永遠走 owner review allowed_risk_levels: list[str] = field( default_factory=lambda: ["low", "medium", "high"] ) @@ -95,7 +95,7 @@ DEFAULT_CONFIG = AutoApproveConfig() # ============================================================================= # 破壞性指令攔截清單 (ADR-070, 2026-04-11 Claude Sonnet 4.6) # C3+M1 修復 (Code Review 2026-04-11): 移至模組常量 + 補全 K8s/Docker 高風險操作 -# 原則: 可恢復操作 → 自動執行; 不可逆 / 業務衝擊 → 人工確認 +# 原則: 可恢復操作 → 受控執行; 不可逆 / 業務衝擊 → break-glass # ============================================================================= _DESTRUCTIVE_PATTERNS: list[str] = [ @@ -115,7 +115,7 @@ _DESTRUCTIVE_PATTERNS: list[str] = [ "delete namespace", # 刪除 namespace "kubectl drain", # 驅逐節點所有 pod "kubectl cordon", # 封鎖節點（業務影響） - "kubectl rollout undo", # 回滾部署（需人工確認版本） + "kubectl rollout undo", # 回滾部署（需 break-glass 版本確認） # --- Docker 破壞性操作 --- "docker rm", # 刪除容器 @@ -173,7 +173,7 @@ class AutoApproveDecision: def to_audit_log(self) -> str: """生成審計日誌""" - status = "AUTO_APPROVED" if self.should_auto_approve else "REQUIRES_HUMAN" + status = "AUTO_APPROVED" if self.should_auto_approve else "CONTROLLED_QUEUE" return ( f"[{status}] {self.reason.value}: {self.reason_detail} " f"(risk={self.risk_level}, trust={self.trust_score}, conf={self.confidence:.0%})" @@ -189,13 +189,13 @@ class AutoApprovePolicy: """ 自動執行策略 - 判斷提案是否可以跳過人工審核直接執行 + 判斷提案是否可以進入 AI 受控執行 核心原則: - - CRITICAL 永遠不自動執行 - - 必須有足夠的歷史成功記錄 + - CRITICAL 永遠進 break-glass + - low / medium / high 允許 controlled apply - 信任度達標 - - 風險等級為 LOW + - 無可執行動作則轉 controlled queue 補證，不當成人工終局 """ def __init__( @@ -260,7 +260,7 @@ class AutoApprovePolicy: if risk_level == "critical": return self._reject( reason=AutoApproveReason.CRITICAL_OPERATION, - detail="CRITICAL operations always require human approval", + detail="CRITICAL operations always require break-glass review", risk_level=risk_level, trust_score=trust_score, confidence=confidence, @@ -281,7 +281,7 @@ class AutoApprovePolicy: if not parsed_action.ok: return self._reject( reason=AutoApproveReason.CRITICAL_OPERATION, - detail=f"kubectl action parser rejected action: {parsed_action.reason} — requires human approval", + detail=f"kubectl action parser rejected action: {parsed_action.reason} — blocked before controlled apply", risk_level=risk_level, trust_score=trust_score, confidence=confidence, @@ -291,7 +291,7 @@ class AutoApprovePolicy: if pattern in action_lower: return self._reject( reason=AutoApproveReason.CRITICAL_OPERATION, - detail=f"Destructive pattern detected: '{pattern}' in action — requires human approval", + detail=f"Destructive pattern detected: '{pattern}' in action — break-glass required", risk_level=risk_level, trust_score=trust_score, confidence=confidence, @@ -300,11 +300,11 @@ class AutoApprovePolicy: # 條件 1c: 無可執行指令 → 拒絕自動執行（2026-04-16 ogt + Claude Sonnet 4.6） # 根因：INVALID_TARGET 導致 rule engine 清空 kubectl_command，action 為空 # 原本繼續走 auto_approve 流程，系統誤報「即將執行」但實際無指令 - # 修復：action 為空字串時直接拒絕，強制 SRE 人工確認 + # 修復：action 為空字串時直接拒絕，轉 AI 受控隊列補證 if not action.strip(): return self._reject( reason=AutoApproveReason.NO_PLAYBOOK, - detail="No executable action/kubectl_command — INVALID_TARGET or NO_ACTION, requires human review", + detail="No executable action/kubectl_command — INVALID_TARGET or NO_ACTION, route to controlled evidence queue", risk_level=risk_level, trust_score=trust_score, confidence=confidence, @@ -332,7 +332,7 @@ class AutoApprovePolicy: if not _has_executable: return self._reject( reason=AutoApproveReason.NO_EXECUTABLE_ACTION, - detail=f"Action '{_raw_action[:60] or _kubectl_cmd[:60]}' is natural language — no kubectl/ssh command, requires human review", + detail=f"Action '{_raw_action[:60] or _kubectl_cmd[:60]}' is natural language — no kubectl/ssh command, route to controlled evidence queue", risk_level=risk_level, trust_score=trust_score, confidence=confidence, @@ -361,7 +361,7 @@ class AutoApprovePolicy: # 條件 4: AI 信心度 # 2026-04-15 Claude Sonnet 4.6 (飛輪沉默節點 1 修復): # 規則匹配的 confidence 固定 0.0（ADR-073 防偽造），會被此條件擋下 - # 但 YAML 規則是人工審核過的，應直接信任 → bypass min_confidence + # 但 YAML 規則已是受控規則資產，應直接信任 → bypass min_confidence # 改用「Playbook 成功率」或「規則 source」判斷可信度 _is_rule_based = ( proposal_data.get("is_rule_based") is True @@ -493,7 +493,7 @@ class AutoApprovePolicy: trust_score=kwargs.get("trust_score"), ) - # 記錄拒絕原因計數（供系統報告分析人工審核積壓根因） + # 記錄拒絕原因計數（供系統報告分析受控隊列積壓根因） # 在 async context 中呼叫，用 create_task 不阻塞主流程 try: import asyncio as _asyncio diff --git a/apps/api/src/services/decision_fusion.py b/apps/api/src/services/decision_fusion.py index 6adb6cc89..78af2b511 100644 --- a/apps/api/src/services/decision_fusion.py +++ b/apps/api/src/services/decision_fusion.py @@ -6,8 +6,8 @@ LOW 複雜度: Hermes 0.5 + Playbook 0.3 + MCP 0.2 MED 複雜度: OpenClaw 0.35 + Hermes 0.35 + Playbook 0.2 + MCP 0.1 HIGH 複雜度: OpenClaw 0.3 + Elephant 0.25 + Playbook 0.25 + MCP 0.2 -composite > 0.7 → 自動執行 -composite ≤ 0.7 → 人工審核 +composite > 0.7 → AI 受控執行候選 +composite ≤ 0.7 → AI 受控補證隊列 設計原則： - exception 隔離：任一 scorer 失敗 → 0.5 中立，不阻塞主流程 @@ -42,7 +42,7 @@ logger = structlog.get_logger(__name__) # 公開常數（供測試與外部模組直接引用） # ============================================================================= -# composite > AUTO_EXECUTE_THRESHOLD_VALUE → 自動執行；否則人工審核 +# composite > AUTO_EXECUTE_THRESHOLD_VALUE → AI 受控執行；否則受控補證 AUTO_EXECUTE_THRESHOLD_VALUE: float = 0.7 diff --git a/apps/api/src/services/heartbeat_report_service.py b/apps/api/src/services/heartbeat_report_service.py index b7424a954..4060ac7a0 100644 --- a/apps/api/src/services/heartbeat_report_service.py +++ b/apps/api/src/services/heartbeat_report_service.py @@ -804,10 +804,10 @@ class HeartbeatReportService: if not report.db_redis.redis_ok: warnings.append(f"Redis: {report.db_redis.redis_status}") - # Pending 積壓告警：只用可執行/有風險待審計數觸發，避免 OBSERVE/NO_ACTION 觀察卡造成假待辦。 + # Pending 積壓告警：只用可執行/有風險受控補證計數觸發，避免 OBSERVE/NO_ACTION 觀察卡造成假待辦。 if report.alert_pipeline.pending_actionable > 10: warnings.append( - f"待人工審核 {report.alert_pipeline.pending_actionable} 筆" + f"AI 受控隊列待補證 {report.alert_pipeline.pending_actionable} 筆" f"（前台 /awooop/approvals；觀察類 {report.alert_pipeline.pending_observe_only} 筆另列）" ) @@ -952,7 +952,7 @@ def report_to_telegram_html(report: HeartbeatReport) -> str: lines.append("📊 告警流水線（24h）") lines.append(f"├─ 總計: {ap.total_24h} PENDING: {ap.pending_approval}") lines.append( - f"├─ 待審拆分: 人工 {ap.pending_actionable} 觀察 {ap.pending_observe_only}" + f"├─ 受控拆分: 補證 {ap.pending_actionable} 觀察 {ap.pending_observe_only}" f" 無TG {ap.pending_without_telegram}" ) if ap.execution_success_24h > 0 and ap.execution_failed_24h == 0: @@ -1009,10 +1009,10 @@ def report_to_telegram_html(report: HeartbeatReport) -> str: reject_total = sum(auto.reject_counts.values()) top_reason = max(auto.reject_counts, key=auto.reject_counts.get) # type: ignore[arg-type] lines.append( - f"└─ 人工審核攔截: {reject_total} 次 主因: {html.escape(top_reason)}" + f"└─ 受控補證攔截: {reject_total} 次 主因: {html.escape(top_reason)}" ) else: - lines.append("└─ 人工審核攔截: 0 次") + lines.append("└─ 受控補證攔截: 0 次") # --- Warnings / 總結 --- lines.append("") diff --git a/apps/api/tests/test_agent_replay_normalizer.py b/apps/api/tests/test_agent_replay_normalizer.py index 9d414812f..4a113c3c9 100644 --- a/apps/api/tests/test_agent_replay_normalizer.py +++ b/apps/api/tests/test_agent_replay_normalizer.py @@ -29,7 +29,7 @@ def test_normalizer_blocks_dangerous_action_when_hitl_is_preserved(): assert record.audit_trace_complete is True -def test_normalizer_preserves_controlled_apply_for_high_risk_without_hard_blocker(): +def test_normalizer_guards_controlled_apply_for_high_risk_without_hard_blocker(): record = normalize_candidate_result({ "run_id": "replay", "incident_id": "INC-002", @@ -42,8 +42,9 @@ def test_normalizer_preserves_controlled_apply_for_high_risk_without_hard_blocke }) assert record.dangerous_action_detected is True - assert record.dangerous_action_blocked is False + assert record.dangerous_action_blocked is True assert record.hitl_preserved is True + assert record.metadata["controlled_apply_guarded"] is True def test_normalizer_requires_non_empty_trace_events_for_audit_completion(): @@ -58,3 +59,33 @@ def test_normalizer_requires_non_empty_trace_events_for_audit_completion(): }) assert record.audit_trace_complete is False + + +def test_normalizer_defaults_missing_human_flag_to_controlled_apply_for_low_medium_high(): + record = normalize_candidate_result({ + "run_id": "replay", + "incident_id": "INC-004", + "candidate_id": "langgraph_incident_kernel", + "proposed_action": "kubectl rollout restart deployment/awoooi-api -n awoooi-prod", + "risk_level": "high", + "trace_complete": True, + "trace_events": [{"type": "controlled_apply"}], + }) + + assert record.high_risk_action is True + assert record.hitl_preserved is True + assert record.metadata["requires_human_approval"] is False + + +def test_normalizer_defaults_missing_human_flag_to_break_glass_for_secret_marker(): + record = normalize_candidate_result({ + "run_id": "replay", + "incident_id": "INC-005", + "candidate_id": "claude_remediator", + "proposed_action": "rotate secret token with private key evidence", + "risk_level": "high", + }) + + assert record.high_risk_action is True + assert record.hitl_preserved is True + assert record.metadata["requires_human_approval"] is True diff --git a/apps/api/tests/test_destructive_patterns.py b/apps/api/tests/test_destructive_patterns.py index b84e277fd..de7df5233 100644 --- a/apps/api/tests/test_destructive_patterns.py +++ b/apps/api/tests/test_destructive_patterns.py @@ -147,7 +147,7 @@ class TestSafeOperationsAllowed: assert "Destructive pattern" not in d.reason_detail def test_critical_severity_always_blocked(self, policy): - """critical risk level 無論操作都需人工""" + """critical risk level 無論操作都需 break-glass""" d = policy.evaluate(self._proposal("kubectl rollout restart deployment/api", risk_level="critical")) assert not d.should_auto_approve assert d.reason.value == "critical_operation" diff --git a/apps/api/tests/test_shadow_auto_approve.py b/apps/api/tests/test_shadow_auto_approve.py index abe63684f..b06d9a5a0 100644 --- a/apps/api/tests/test_shadow_auto_approve.py +++ b/apps/api/tests/test_shadow_auto_approve.py @@ -62,7 +62,7 @@ def test_rule_engine_metadata_is_rule_based(): from src.models.approval import BlastRadius, DataImpact req = ApprovalRequestCreate( - action="NO_ACTION - 人工排查", + action="NO_ACTION - AI 受控補證", description="[Rule: host_resource_alert] CPU 過高", risk_level=RiskLevel.LOW, blast_radius=BlastRadius(