From d3fa6feef4cf466e7a303aa0dfc501ce7b47104e Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 11:46:38 +0800 Subject: [PATCH] fix(agent): auto-authorize noncritical alert control --- apps/api/src/api/v1/webhooks.py | 168 ++++++++----- apps/api/src/core/trust_engine.py | 58 +++-- apps/api/src/models/approval.py | 11 +- apps/api/src/services/approval_db.py | 66 ++++- apps/api/src/services/operator_outcome.py | 20 +- apps/api/src/services/telegram_gateway.py | 70 +++++- .../test_awooop_operator_timeline_labels.py | 4 +- ...t_current_owner_controlled_apply_policy.py | 236 ++++++++++++++++++ apps/api/tests/test_operator_outcome.py | 4 +- apps/api/tests/test_telegram_integration.py | 2 +- 10 files changed, 542 insertions(+), 97 deletions(-) create mode 100644 apps/api/tests/test_current_owner_controlled_apply_policy.py diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index d7e12cbf3..e7605a80c 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -44,7 +44,6 @@ from src.models.approval import ( RiskLevel, ) from src.models.webhook import AlertPayload, AlertResponse -from src.services.action_parser import is_safe_kubectl_action from src.services.alert_analyzer_service import AlertAnalyzer from src.services.alert_approval_guard import guard_alert_approval_action from src.services.alert_grouping_service import get_alert_grouping_service @@ -217,6 +216,18 @@ def _auto_repair_action_label(result, fallback_target: str) -> str: return f"auto_repair_playbook:{playbook_id} {step_text}".strip() +def _controlled_ai_policy_allows(risk_level: RiskLevel | str) -> bool: + """Return whether the current owner policy waives human approval. + + The decision only opens the single-writer check/apply/verifier route. It + does not bypass catalog allowlists, check-mode, rollback, or post-verifier + requirements. Critical actions remain break-glass only. + """ + + value = risk_level.value if isinstance(risk_level, RiskLevel) else str(risk_level) + return value.strip().lower() in {"low", "medium", "high"} + + async def _try_auto_repair_background( incident_id: str, approval_id: str, @@ -1247,6 +1258,13 @@ async def receive_alert( ) if existing_approval: + policy_promoted = False + promote = getattr(service, "apply_current_owner_policy", None) + if callable(promote): + promoted_approval, policy_promoted = await promote(existing_approval.id) + if promoted_approval is not None: + existing_approval = promoted_approval + # ========================================================================== # 戰略 B Step 3: [收斂] 同指紋告警 - 跳過 LLM，只更新計數！ # ========================================================================== @@ -1263,6 +1281,21 @@ async def receive_alert( updated_approval = await service.increment_hit_count(existing_approval.id) if updated_approval: + if ( + policy_promoted + and getattr(updated_approval, "incident_id", None) + and _controlled_ai_policy_allows(updated_approval.risk_level) + and not is_heartbeat_alertname(alert.alert_type) + ): + background_tasks.add_task( + _try_auto_repair_background, + incident_id=str(updated_approval.incident_id), + approval_id=str(updated_approval.id), + alert_type=alert.alert_type, + target_resource=alert.target_resource, + namespace=alert.namespace, + risk_level=updated_approval.risk_level.value, + ) # ================================================================= # 2026-03-27 ogt: 收斂告警不重複發送 Telegram，只更新 hit_count # 避免 Telegram 洗版，用戶可在 UI 查看聚合次數 @@ -1363,6 +1396,7 @@ async def receive_alert( risk_mapping = { "low": RiskLevel.LOW, "medium": RiskLevel.MEDIUM, + "high": RiskLevel.HIGH, "critical": RiskLevel.CRITICAL, } risk_level = risk_mapping.get(analysis_result.risk_level.value, RiskLevel.MEDIUM) @@ -1500,27 +1534,20 @@ async def receive_alert( except Exception as _shadow_err: logger.warning("shadow_auto_approve_failed", error=str(_shadow_err)) - # 2026-04-27 ogt + Claude Sonnet 4.6: CS1 LLM 高信心度自動執行 - # 設計：confidence ≥ 0.85 + 非 CRITICAL + 非破壞性 + 有 kubectl 指令 → 直接執行 - # 安全防線：CRITICAL / destructive patterns / NO_ACTION/INVESTIGATE/OBSERVE / 空 kubectl → 降級 PENDING - if analysis_result: - _cs1_kubectl = _cmd_cs1 - _cs1_can_auto = ( - bool(_cs1_kubectl) - and analysis_result.confidence >= 0.85 - and _cs1_risk_level != RiskLevel.CRITICAL - and _sa_val not in _non_destructive_actions - and is_safe_kubectl_action(_cs1_kubectl) + # Current policy routes every non-critical alert to the single-writer + # executor. Confidence and command shape remain evidence for check-mode; + # they are not human-approval gates. + if _controlled_ai_policy_allows(_cs1_risk_level) and not is_heartbeat_alertname( + str((alert.labels or {}).get("alertname") or alert.alert_type or "") + ): + await _try_auto_repair_background( + incident_id=_cs1_incident_id, + approval_id=str(approval.id), + alert_type=alert.alert_type, + target_resource=alert.target_resource, + namespace=alert.namespace, + risk_level=_cs1_risk_level.value, ) - if _cs1_can_auto: - await _try_auto_repair_background( - incident_id=_cs1_incident_id, - approval_id=str(approval.id), - alert_type=alert.alert_type, - target_resource=alert.target_resource, - namespace=alert.namespace, - risk_level=_cs1_risk_level.value, - ) logger.info( "approval_auto_created_with_fingerprint", @@ -1749,6 +1776,7 @@ async def _process_new_alert_background( risk_mapping = { "low": RiskLevel.LOW, "medium": RiskLevel.MEDIUM, + "high": RiskLevel.HIGH, "critical": RiskLevel.CRITICAL, } rule_risk = risk_mapping.get( @@ -1860,12 +1888,7 @@ async def _process_new_alert_background( # 2026-04-27 ogt + Claude Sonnet 4.6: CS2 規則引擎自動執行 # 設計：is_rule_based=True 確定性高，滿足條件直接執行，不等人工審核 # 安全防線：CRITICAL / destructive patterns / NO_ACTION / 空 kubectl → 全部降級 PENDING - _can_auto = ( - bool(rule_kubectl) - and rule_risk != RiskLevel.CRITICAL - and is_safe_kubectl_action(rule_kubectl) - and "NO_ACTION" not in rule_action - ) + _can_auto = _controlled_ai_policy_allows(rule_risk) if _can_auto: logger.info( "rule_engine_auto_execution_deferred_to_single_writer", @@ -1919,7 +1942,7 @@ async def _process_new_alert_background( ) _is_heartbeat = is_heartbeat_alertname(alertname) - if can_auto_repair and not _is_heartbeat: + if _controlled_ai_policy_allows(rule_risk) and not _is_heartbeat: await _try_auto_repair_background( incident_id=incident_id, approval_id=str(approval.id), @@ -1928,7 +1951,7 @@ async def _process_new_alert_background( namespace=namespace, risk_level=rule_risk.value, ) - elif not can_auto_repair and not _is_heartbeat: + elif not _is_heartbeat: from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository _op_log_rule = get_alert_operation_log_repository() await _op_log_rule.append( @@ -1936,9 +1959,13 @@ async def _process_new_alert_background( incident_id=incident_id, approval_id=str(approval.id), actor="prometheus-rule", - action_detail=f"Prometheus rule 設定 auto_repair=false，轉入緊急介入: {alertname}", + action_detail=f"Critical alert requires break-glass: {alertname}", success=False, - context={"alertname": alertname, "auto_repair_flag": False}, + context={ + "alertname": alertname, + "legacy_auto_repair_flag": bool(can_auto_repair), + "critical_break_glass": True, + }, ) await _escalate_auto_repair_unavailable( incident_id=incident_id, @@ -1946,8 +1973,8 @@ async def _process_new_alert_background( alert_type=alert_type, target_resource=target_resource, namespace=namespace, - failure_reason="Prometheus rule auto_repair=false，未進入自動修復評估", - attempted_actions="rule_engine -> rule_first -> emergency_intervention", + failure_reason="critical risk requires break-glass authorization", + attempted_actions="rule_engine -> critical_break_glass_gate", ) await _push_to_telegram_background( @@ -1983,6 +2010,7 @@ async def _process_new_alert_background( risk_mapping = { "low": RiskLevel.LOW, "medium": RiskLevel.MEDIUM, + "high": RiskLevel.HIGH, "critical": RiskLevel.CRITICAL, } risk_level = risk_mapping.get(analysis_result.risk_level.value, RiskLevel.MEDIUM) @@ -2078,15 +2106,8 @@ async def _process_new_alert_background( except Exception as _shadow_err_cs3: logger.warning("shadow_auto_approve_failed", error=str(_shadow_err_cs3)) - # 2026-04-27 Claude Sonnet 4.6: CS3 LLM 高信心自動執行（修法3擴展） _cs3_kubectl = _cmd_cs3 - _cs3_can_auto = ( - bool(_cs3_kubectl) - and analysis_result.confidence >= 0.85 - and risk_level != RiskLevel.CRITICAL - and "NO_ACTION" not in (analysis_result.action_title or "") - and is_safe_kubectl_action(_cs3_kubectl) - ) + _cs3_can_auto = _controlled_ai_policy_allows(risk_level) if _cs3_can_auto: logger.info( "cs3_llm_auto_execution_deferred_to_single_writer", @@ -2153,7 +2174,7 @@ async def _process_new_alert_background( alertname=alertname, ) - if can_auto_repair and not _is_heartbeat: + if _controlled_ai_policy_allows(risk_level) and not _is_heartbeat: await _try_auto_repair_background( incident_id=incident_id, approval_id=str(approval.id), @@ -2162,7 +2183,7 @@ async def _process_new_alert_background( namespace=namespace, risk_level=risk_level.value, ) - else: + elif not _is_heartbeat: from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository _op_log_rule = get_alert_operation_log_repository() await _op_log_rule.append( @@ -2170,20 +2191,23 @@ async def _process_new_alert_background( incident_id=incident_id, approval_id=str(approval.id), actor="prometheus-rule", - action_detail=f"Prometheus rule 設定 auto_repair=false，強制人工審核: {alertname}", + action_detail=f"Critical alert requires break-glass: {alertname}", success=False, - context={"alertname": alertname, "auto_repair_flag": False}, + context={ + "alertname": alertname, + "legacy_auto_repair_flag": bool(can_auto_repair), + "critical_break_glass": True, + }, + ) + await _escalate_auto_repair_unavailable( + incident_id=incident_id, + approval_id=str(approval.id), + alert_type=alert_type, + target_resource=target_resource, + namespace=namespace, + failure_reason="critical risk requires break-glass authorization", + attempted_actions="llm_analysis -> critical_break_glass_gate", ) - if not _is_heartbeat: - await _escalate_auto_repair_unavailable( - incident_id=incident_id, - approval_id=str(approval.id), - alert_type=alert_type, - target_resource=target_resource, - namespace=namespace, - failure_reason="Prometheus rule auto_repair=false，未進入自動修復評估", - attempted_actions="llm_analysis -> guardrail:auto_repair_false -> emergency_intervention", - ) await _push_to_telegram_background( approval_id=str(approval.id), @@ -2413,6 +2437,15 @@ async def _process_new_alert_background( from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository _op_log_fallback = get_alert_operation_log_repository() if repair_candidate_result.candidate_found: + if _controlled_ai_policy_allows(fallback_create.risk_level): + await _try_auto_repair_background( + incident_id=fallback_incident_id, + approval_id=str(approval.id), + alert_type=alert_type, + target_resource=target_resource, + namespace=namespace, + risk_level=fallback_create.risk_level.value, + ) await _op_log_fallback.append( "PLAYBOOK_DRAFT_CREATED", incident_id=fallback_incident_id, @@ -2620,7 +2653,8 @@ async def alertmanager_webhook( _alert_labels = alert.labels or {} _alert_annotations = alert.annotations or {} _alertname_for_log = _alert_labels.get("alertname", "UnknownAlert") - # Q9: auto_repair flag — Rule=false 強制 HITL（不觸發自動修復背景任務） + # Legacy rule flag is retained as evidence only. Current owner policy routes + # low/medium/high alerts through the controlled single-writer pipeline. _can_auto_repair_by_rule = _alert_labels.get("auto_repair", "true").lower() == "true" try: from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository @@ -2878,6 +2912,13 @@ async def alertmanager_webhook( ) if existing_approval: + policy_promoted = False + promote = getattr(service, "apply_current_owner_policy", None) + if callable(promote): + promoted_approval, policy_promoted = await promote(existing_approval.id) + if promoted_approval is not None: + existing_approval = promoted_approval + # 收斂告警 - 跳過 LLM logger.info( "alertmanager_converged", @@ -2888,6 +2929,21 @@ async def alertmanager_webhook( updated_approval = await service.increment_hit_count(existing_approval.id) if updated_approval: + if ( + policy_promoted + and getattr(updated_approval, "incident_id", None) + and _controlled_ai_policy_allows(updated_approval.risk_level) + and not is_heartbeat_alertname(alertname) + ): + background_tasks.add_task( + _try_auto_repair_background, + incident_id=str(updated_approval.incident_id), + approval_id=str(updated_approval.id), + alert_type=alert_type, + target_resource=target_resource, + namespace=namespace, + risk_level=updated_approval.risk_level.value, + ) # 2026-03-27 ogt: 收斂告警不重複發送 Telegram，只更新 hit_count # 用戶可在 UI 查看聚合次數，避免 Telegram 洗版 logger.info( diff --git a/apps/api/src/core/trust_engine.py b/apps/api/src/core/trust_engine.py index 4f6b038dc..abc4015a9 100644 --- a/apps/api/src/core/trust_engine.py +++ b/apps/api/src/core/trust_engine.py @@ -4,9 +4,8 @@ Trust Engine - 風險判定與 Multi-Sig 簽核邏輯 CISO-101: 信任引擎核心實作 風險等級與簽核需求: -- LOW: 0 人，自動放行 (如 scale up) -- MEDIUM: 需 1 人簽核 (如 delete pod) -- CRITICAL: 需 2 人 Multi-Sig 雙重簽核 (如 DROP TABLE) +- LOW/MEDIUM/HIGH: 0 人，進入受控自動執行 +- CRITICAL: 1 人 break-glass (如 DROP TABLE / host reboot / secret read) Features: - 自動風險分類 @@ -42,6 +41,37 @@ CRITICAL_KEYWORDS = [ "format", "wipe", "purge all", + "destructive migration", + "restore database", + "restore backup", + "remote delete", + "retention delete", + "retention prune", + "host reboot", + "reboot host", + "node drain", + "firewall cutover", + "dns cutover", + "credentialed exploit", + "active scan", + "paid provider", + "cost limit", + "switch production provider", + "replace production model", + "replace agent runtime", + "force push", + "delete repo", + "delete ref", + "change repo visibility", + "raw runtime secret volume", + "kubectl get secret", + "kubectl describe secret", + "read secret", + "dump secret", + "export token", + "read token", + "private key", + "authorization header", ] MEDIUM_KEYWORDS = [ @@ -69,9 +99,10 @@ LOW_KEYWORDS = [ # ============================================================================= SIGNATURE_REQUIREMENTS: dict[RiskLevel, int] = { - RiskLevel.LOW: 0, # 自動放行 - RiskLevel.MEDIUM: 1, # 單人簽核 - RiskLevel.CRITICAL: 1, # 2026-04-02 ogt: 統帥決策 — 只需 1 層審核，Multi-Sig 停用 + RiskLevel.LOW: 0, + RiskLevel.MEDIUM: 0, + RiskLevel.HIGH: 0, + RiskLevel.CRITICAL: 1, } @@ -162,10 +193,6 @@ def classify_risk( Returns: 最終風險等級 """ - # 如果明確指定，直接使用 - if explicit_level is not None: - return explicit_level - # 從動作分類 action_risk = classify_risk_by_action(action) @@ -175,11 +202,12 @@ def classify_risk( blast_risk = classify_risk_by_blast_radius(blast_radius) # 取較高風險等級 - risk_order = [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.CRITICAL] + risk_order = [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH, RiskLevel.CRITICAL] action_idx = risk_order.index(action_risk) blast_idx = risk_order.index(blast_risk) + explicit_idx = risk_order.index(explicit_level) if explicit_level is not None else 0 - return risk_order[max(action_idx, blast_idx)] + return risk_order[max(action_idx, blast_idx, explicit_idx)] # ============================================================================= @@ -220,7 +248,7 @@ class TrustEngine: 建立新的授權請求 自動根據風險等級設定所需簽核數 - LOW 風險自動批准 + LOW/MEDIUM/HIGH 風險由 current owner policy 自動批准 """ # 分類風險 risk_level = classify_risk( @@ -245,8 +273,8 @@ class TrustEngine: required_signatures=required_sigs, ) - # LOW 風險自動批准 - if risk_level == RiskLevel.LOW: + # Non-critical requests enter controlled automation without a human gate. + if required_sigs == 0: approval.status = ApprovalStatus.APPROVED approval.resolved_at = datetime.now(UTC) if self._on_approved: diff --git a/apps/api/src/models/approval.py b/apps/api/src/models/approval.py index 6a6ce1a9e..2761b35b8 100644 --- a/apps/api/src/models/approval.py +++ b/apps/api/src/models/approval.py @@ -5,8 +5,8 @@ CISO-101: 授權請求與簽核資料模型 Features: - 狀態機 (PENDING → APPROVED/REJECTED/EXPIRED) -- 風險等級判定 (LOW/MEDIUM/CRITICAL) -- Multi-Sig 簽核追蹤 +- 風險等級判定 (LOW/MEDIUM/HIGH/CRITICAL) +- Controlled-apply / break-glass 簽核追蹤 - Pydantic 強型別驗證 """ @@ -41,10 +41,9 @@ class RiskLevel(str, Enum): """ 風險等級 - 決定所需簽核人數 - - LOW: 0 人，自動放行 - - MEDIUM: 需 1 人簽核 - - HIGH: 需 1 人簽核 (信任引擎可降級至 MEDIUM) - - CRITICAL: 需 2 人 Multi-Sig 雙重簽核 (永不降級) + - LOW/MEDIUM/HIGH: current owner policy 自動授權，後續仍須 + check-mode、bounded execution、獨立 verifier 與 rollback receipt + - CRITICAL: break-glass 人工授權 (永不自動降級) 變更紀錄: - 2026-03-25: 新增 HIGH (Phase 16 R2 合併自 trust_engine.py) diff --git a/apps/api/src/services/approval_db.py b/apps/api/src/services/approval_db.py index b2e664bf4..5328c0ab9 100644 --- a/apps/api/src/services/approval_db.py +++ b/apps/api/src/services/approval_db.py @@ -159,7 +159,7 @@ def approval_request_to_record_data( record_data = { "action": request.action, "description": request.description, - "status": ApprovalStatus.APPROVED if risk_level == RiskLevel.LOW else ApprovalStatus.PENDING, + "status": ApprovalStatus.APPROVED if required_sigs == 0 else ApprovalStatus.PENDING, "risk_level": risk_level, "required_signatures": required_sigs, "current_signatures": 0, @@ -169,7 +169,7 @@ def approval_request_to_record_data( "requested_by": request.requested_by, "expires_at": request.expires_at, "extra_metadata": metadata or None, - "resolved_at": now if risk_level == RiskLevel.LOW else None, + "resolved_at": now if required_sigs == 0 else None, # 戰略 B: 告警風暴收斂 "fingerprint": fingerprint, "hit_count": 1, @@ -279,7 +279,7 @@ def _approval_decision_mode(record: ApprovalRecord) -> str: risk_level = _record_value(record.risk_level) if _approval_needs_human(record) or current > 0: return "manual" - if risk_level == "low" and required == 0: + if risk_level in {"low", "medium", "high"} and required == 0: return "auto" return "manual" @@ -442,6 +442,66 @@ class ApprovalDBService: return approval_record_to_request(record) + async def apply_current_owner_policy( + self, + approval_id: UUID, + ) -> tuple[ApprovalRequest | None, bool]: + """Promote a legacy pending non-critical gate to controlled automation. + + Existing alert fingerprints can otherwise keep converging onto an old + PENDING row for 24 hours. This bounded reconciliation updates only one + row, re-runs the current critical classifier, and leaves critical rows + untouched for break-glass handling. + """ + + async with get_db_context() as db: + result = await db.execute( + select(ApprovalRecord) + .where(ApprovalRecord.id == str(approval_id)) + .with_for_update() + ) + record = result.scalar_one_or_none() + if record is None: + return None, False + + current = approval_record_to_request(record) + status = _record_value(record.status) + effective_risk = classify_risk( + action=str(record.action or ""), + blast_radius=current.blast_radius, + explicit_level=current.risk_level, + ) + if status != "pending" or effective_risk == RiskLevel.CRITICAL: + return current, False + + metadata = dict(record.extra_metadata or {}) + metadata.update( + { + "owner_policy": "global_product_governance_v2", + "owner_review_gate": "auto_waived_for_low_medium_high", + "controlled_apply_required": True, + "legacy_pending_promoted": True, + } + ) + record.status = ApprovalStatus.APPROVED + record.risk_level = effective_risk + record.required_signatures = 0 + record.current_signatures = 0 + record.signatures = [] + record.resolved_at = datetime.now(UTC) + record.extra_metadata = metadata + add_approval_created_timeline_event(db, record) + await db.flush() + await db.refresh(record) + + logger.info( + "approval_promoted_by_current_owner_policy", + id=record.id, + risk_level=effective_risk.value, + incident_id=record.incident_id, + ) + return approval_record_to_request(record), True + async def find_by_fingerprint( self, fingerprint: str, diff --git a/apps/api/src/services/operator_outcome.py b/apps/api/src/services/operator_outcome.py index 300872b31..ac92fcda4 100644 --- a/apps/api/src/services/operator_outcome.py +++ b/apps/api/src/services/operator_outcome.py @@ -143,7 +143,7 @@ def normalize_operator_outcome(outcome: dict[str, Any] | None) -> dict[str, Any] if isinstance(normalized.get("notification"), dict) else {} ) - notification["mode"] = "action_required" + notification["mode"] = "human_action_required" if break_glass else "automation_progress" notification["channels"] = list(_ACTION_REQUIRED_CHANNELS) notification["reason"] = str(normalized["human_action_reason"]) normalized["notification"] = notification @@ -200,11 +200,10 @@ def _build_notification( "channels": list(channels), "reason": reason, "source_id": source_id, - "telegram": ( - "reply_to_original_or_standalone_action_required" - if mode == "action_required" - else "reply_to_original_or_standalone_result" - ), + "telegram": { + "human_action_required": "reply_to_original_or_standalone_action_required", + "automation_progress": "reply_to_original_or_standalone_automation_progress", + }.get(mode, "reply_to_original_or_standalone_result"), "awooop": "status_chain_panel", } @@ -607,8 +606,13 @@ def build_operator_outcome( "monitor_for_regression", "monitor_or_reopen_if_alert_recurs", } - mode = "action_required" if needs_human or ai_action_required else "result_only" - channels = _ACTION_REQUIRED_CHANNELS if mode == "action_required" else _RESULT_ONLY_CHANNELS + if needs_human: + mode = "human_action_required" + elif ai_action_required: + mode = "automation_progress" + else: + mode = "result_only" + channels = _ACTION_REQUIRED_CHANNELS if mode != "result_only" else _RESULT_ONLY_CHANNELS return { "schema_version": "operator_outcome_v1", "state": state, diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 774de7eb1..5ddad0403 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -3374,6 +3374,7 @@ class TelegramMessage: repair_candidate_promotion_summary: str = "" # 候選升級合約摘要 repair_candidate_work_item_href: str = "" # AwoooP 修復候選草案連結 repair_candidate_work_item_id: str = "" # AwoooP 修復候選草案 ID + controlled_auto_authorized: bool = False # low/medium/high owner policy # ========================================================================== # Phase 22: Nemotron 協作欄位 (ADR-044) @@ -3529,7 +3530,11 @@ class TelegramMessage: nemotron_state = "tool_ready" if self.nemotron_enabled else "standby" hermes_state = self.playbook_name or "rule_catalog" elephant_state = "timeline_km_pending" - flow = "webhook>investigator>router>llm/rule>safe>approval" + flow = ( + "webhook>investigator>router>policy>check>apply>verify>writeback" + if self.controlled_auto_authorized + else "webhook>investigator>router>safe>break-glass" + ) # 2026-05-04 ogt: Token 用量顯示（有資料才顯示） token_line = "" @@ -3608,8 +3613,10 @@ class TelegramMessage: execute_state = f"diagnostic_recorded:{operation_records}" elif is_noop: execute_state = "no_action_or_observe" + elif self.controlled_auto_authorized: + execute_state = "controlled_queue_evaluation" elif "approval" in verdict or self._automation_mode() == "ai_proposal_ready": - execute_state = "awaiting_approval" + execute_state = "awaiting_break_glass" else: execute_state = "not_started" @@ -3628,8 +3635,10 @@ class TelegramMessage: conclusion = "已記錄診斷/觀察，尚未證明修復" elif is_noop: conclusion = "未自動修復，已排入 AI 證據補強 / PlayBook 候選" + elif self.controlled_auto_authorized: + conclusion = "已由 owner policy 自動授權，等待 check/apply/verifier receipt" elif "approval" in verdict: - conclusion = "等待 controlled policy 判定後執行" + conclusion = "critical break-glass 尚未授權" elif "manual" in verdict: conclusion = "舊人工狀態已轉入 AI 受控處理" else: @@ -3811,8 +3820,13 @@ class TelegramMessage: manual_handoff_block = self._format_manual_handoff_package_block() # ADR-075 TYPE-3 格式組裝 + title = ( + "AI CONTROLLED AUTOMATION" + if self.controlled_auto_authorized + else "BREAK-GLASS REQUIRED" + ) message = ( - f"{self.status_emoji} ACTION REQUIRED | {html.escape(risk_zh)}\n" + f"{self.status_emoji} {title} | {html.escape(risk_zh)}\n" f"──────────────────────\n" f"📋 {html.escape(incident_id)}\n" f"🎯 資源：{safe_resource}\n" @@ -3871,6 +3885,9 @@ class TelegramMessage: Returns: str: 格式化的 Telegram 訊息 (max 1000 字元) """ + if self.controlled_auto_authorized: + return self.format() + # 責任映射 resp_map = { "FE": "👨‍💻 FE (前端)", @@ -5358,6 +5375,7 @@ class TelegramGateway: action_plan: object = None, repair_candidate_work_item_href: str = "", repair_candidate_work_item_id: str = "", + controlled_auto_authorized: bool = False, ) -> dict: """ 建立 Inline Keyboard @@ -5395,6 +5413,35 @@ class TelegramGateway: else is_executable_repair_approval_action(suggested_action) ) + if controlled_auto_authorized: + rows: list[list[dict]] = [] + if incident_id: + first_row: list[dict] = [] + work_item_deep_link = _repair_candidate_work_item_url( + incident_id=incident_id, + repair_candidate_work_item_href=repair_candidate_work_item_href, + repair_candidate_work_item_id=repair_candidate_work_item_id, + ) + if work_item_deep_link: + first_row.append({"text": "🧾 Work Item", "url": work_item_deep_link}) + first_row.extend( + [ + {"text": "🧰 處置包", "callback_data": f"detail:{incident_id}"}, + {"text": "📊 歷史", "callback_data": f"history:{incident_id}"}, + ] + ) + rows.append(first_row) + awooop_row = _awooop_truth_chain_button_row(incident_id) + if awooop_row: + rows.append(awooop_row) + logger.info( + "telegram_keyboard_built", + source="controlled_auto_authorized", + approval_id=approval_id, + incident_id=incident_id, + ) + return {"inline_keyboard": rows} + # 可執行修復卡第一排置頂批准/拒絕；純觀察卡不得提供誤導性的執行批准。 first_row: list[dict] = [ {"text": "✅ 批准", "callback_data": approve_nonce}, @@ -5782,6 +5829,12 @@ class TelegramGateway: Returns: dict: Telegram API 回應 """ + controlled_auto_authorized = risk_level.strip().lower() in { + "low", + "medium", + "high", + } + # 取得狀態 Emoji emoji = RISK_EMOJI_MAP.get(risk_level.lower(), "⚠️") @@ -5869,6 +5922,7 @@ class TelegramGateway: repair_candidate_promotion_summary=repair_candidate_promotion_summary, repair_candidate_work_item_href=repair_candidate_work_item_href, repair_candidate_work_item_id=repair_candidate_work_item_id, + controlled_auto_authorized=controlled_auto_authorized, ) # 格式化訊息 — Phase 22: 如果 Nemotron 啟用，使用雙軌格式 @@ -5887,6 +5941,7 @@ class TelegramGateway: notification_type=notification_type, repair_candidate_work_item_href=repair_candidate_work_item_href, repair_candidate_work_item_id=repair_candidate_work_item_id, + controlled_auto_authorized=controlled_auto_authorized, ) # 發送訊息：2026-04-30 統帥指示，告警卡片完整切到 SRE 戰情室群組。 @@ -6030,6 +6085,11 @@ class TelegramGateway: 由 asyncio.create_task 非同步呼叫，失敗不影響主告警流程。 """ try: + controlled_auto_authorized = risk_level.strip().lower() in { + "low", + "medium", + "high", + } emoji = RISK_EMOJI_MAP.get(risk_level.lower(), "⚠️") signoz_metrics = None @@ -6072,6 +6132,7 @@ class TelegramGateway: nemotron_validation=nemotron_validation, nemotron_latency_ms=nemotron_latency_ms, remediation_summary=remediation_summary, + controlled_auto_authorized=controlled_auto_authorized, ) text = message.format_with_nemotron() if nemotron_enabled else message.format() @@ -6083,6 +6144,7 @@ class TelegramGateway: suggested_action=suggested_action, alert_category=alert_category, notification_type=notification_type, + controlled_auto_authorized=controlled_auto_authorized, ) resp = await self.send_to_group(text=text, reply_markup=_group_keyboard) diff --git a/apps/api/tests/test_awooop_operator_timeline_labels.py b/apps/api/tests/test_awooop_operator_timeline_labels.py index 87dab48bf..811b8dfba 100644 --- a/apps/api/tests/test_awooop_operator_timeline_labels.py +++ b/apps/api/tests/test_awooop_operator_timeline_labels.py @@ -2781,7 +2781,7 @@ def test_awooop_status_chain_does_not_treat_audit_ops_as_repair() -> None: assert chain["next_step"] == "auto_generate_repair_candidate_from_diagnostic_evidence" assert chain["needs_human"] is False assert chain["operator_outcome"]["state"] == "diagnostic_only_ai_repair_required" - assert chain["operator_outcome"]["notification"]["mode"] == "action_required" + assert chain["operator_outcome"]["notification"]["mode"] == "automation_progress" assert chain["evidence"]["operation_records"] == 1 assert chain["evidence"]["auto_repair_records"] == 0 @@ -2914,7 +2914,7 @@ def test_awooop_status_chain_surfaces_expired_approval_outcome() -> None: assert chain["repair_state"] == "approval_expired_ai_retry" assert chain["needs_human"] is False assert chain["operator_outcome"]["state"] == "approval_expired_ai_retry" - assert chain["operator_outcome"]["notification"]["mode"] == "action_required" + assert chain["operator_outcome"]["notification"]["mode"] == "automation_progress" def test_legacy_mcp_timeline_summary_surfaces_tool_context() -> None: diff --git a/apps/api/tests/test_current_owner_controlled_apply_policy.py b/apps/api/tests/test_current_owner_controlled_apply_policy.py new file mode 100644 index 000000000..f5bbc3e65 --- /dev/null +++ b/apps/api/tests/test_current_owner_controlled_apply_policy.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +from contextlib import asynccontextmanager +from datetime import UTC, datetime +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +from src.api.v1.webhooks import _controlled_ai_policy_allows +from src.core.trust_engine import classify_risk, get_required_signatures +from src.models.approval import ApprovalStatus, RiskLevel +from src.services import approval_db as approval_db_module +from src.services.approval_db import ApprovalDBService, approval_request_to_record_data +from src.services.telegram_gateway import TelegramGateway, TelegramMessage + + +@pytest.mark.parametrize( + "risk_level", + [RiskLevel.LOW, RiskLevel.MEDIUM, RiskLevel.HIGH], +) +def test_noncritical_risks_are_auto_authorized_for_controlled_apply( + risk_level: RiskLevel, +) -> None: + assert get_required_signatures(risk_level) == 0 + assert _controlled_ai_policy_allows(risk_level) is True + + +def test_critical_risk_remains_break_glass() -> None: + assert get_required_signatures(RiskLevel.CRITICAL) == 1 + assert _controlled_ai_policy_allows(RiskLevel.CRITICAL) is False + + +@pytest.mark.parametrize( + "action", + [ + "host reboot 192.168.0.110", + "kubectl get secret awoooi-secrets", + "git force push main", + "restore database from snapshot", + ], +) +def test_critical_action_overrides_explicit_high_risk(action: str) -> None: + assert classify_risk(action, explicit_level=RiskLevel.HIGH) == RiskLevel.CRITICAL + + +def test_medium_approval_record_is_created_auto_approved() -> None: + request = SimpleNamespace( + action="kubectl rollout restart deployment/api", + description="bounded rollout", + metadata={"source": "alertmanager"}, + blast_radius=None, + dry_run_checks=[], + requested_by="OpenClaw", + expires_at=None, + incident_id="INC-CONTROLLED", + matched_playbook_id=None, + ) + + data = approval_request_to_record_data( + request, + RiskLevel.MEDIUM, + get_required_signatures(RiskLevel.MEDIUM), + ) + + assert data["status"] == ApprovalStatus.APPROVED + assert data["required_signatures"] == 0 + assert data["resolved_at"] is not None + + +def test_medium_telegram_card_is_automation_progress_not_action_required() -> None: + message = TelegramMessage( + status_emoji="⚠️", + risk_level="MEDIUM", + resource_name="awoooi-api", + root_cause="restart required", + suggested_action="kubectl rollout restart deployment/awoooi-api", + estimated_downtime="30s", + approval_id="approval-1", + incident_id="INC-CONTROLLED", + controlled_auto_authorized=True, + ) + + rendered = message.format() + + assert "AI CONTROLLED AUTOMATION" in rendered + assert "ACTION REQUIRED" not in rendered + assert "awaiting_approval" not in rendered + assert "check>apply>verify>writeback" in rendered + + +def test_critical_telegram_card_stays_break_glass() -> None: + message = TelegramMessage( + status_emoji="🚨", + risk_level="CRITICAL", + resource_name="database", + root_cause="destructive restore requested", + suggested_action="restore database from snapshot", + estimated_downtime="unknown", + approval_id="approval-critical", + incident_id="INC-CRITICAL", + ) + + rendered = message.format() + + assert "BREAK-GLASS REQUIRED" in rendered + assert "AI CONTROLLED AUTOMATION" not in rendered + + +@pytest.mark.asyncio +async def test_controlled_automation_keyboard_has_no_approval_buttons() -> None: + gateway = TelegramGateway() + + keyboard = await gateway._build_inline_keyboard( + approval_id="approval-1", + incident_id="INC-CONTROLLED", + suggested_action="kubectl rollout restart deployment/awoooi-api", + controlled_auto_authorized=True, + ) + button_texts = { + button["text"] + for row in keyboard["inline_keyboard"] + for button in row + } + + assert "✅ 批准" not in button_texts + assert "❌ 拒絕" not in button_texts + assert "🧰 處置包" in button_texts + assert "📊 歷史" in button_texts + + +def _approval_record(*, action: str, risk_level: RiskLevel) -> SimpleNamespace: + now = datetime.now(UTC) + return SimpleNamespace( + id=str(uuid4()), + action=action, + description="legacy pending alert", + status=ApprovalStatus.PENDING, + risk_level=risk_level, + required_signatures=1, + current_signatures=0, + signatures=[], + blast_radius={ + "affected_pods": 1, + "estimated_downtime": "30s", + "related_services": ["awoooi-api"], + "data_impact": "write", + }, + dry_run_checks=[], + requested_by="OpenClaw", + created_at=now, + expires_at=None, + resolved_at=None, + rejection_reason=None, + extra_metadata={"source": "alertmanager"}, + fingerprint="fingerprint", + hit_count=1, + last_seen_at=now, + incident_id="INC-LEGACY-PENDING", + matched_playbook_id=None, + telegram_message_id=None, + telegram_chat_id=None, + ) + + +class _FakeResult: + def __init__(self, record: SimpleNamespace) -> None: + self._record = record + + def scalar_one_or_none(self) -> SimpleNamespace: + return self._record + + +class _FakeDB: + def __init__(self, record: SimpleNamespace) -> None: + self.record = record + self.events: list[object] = [] + + async def execute(self, _statement: object) -> _FakeResult: + return _FakeResult(self.record) + + def add(self, event: object) -> None: + self.events.append(event) + + async def flush(self) -> None: + return None + + async def refresh(self, _record: object) -> None: + return None + + +@pytest.mark.asyncio +async def test_legacy_medium_pending_gate_is_durably_promoted(monkeypatch) -> None: + record = _approval_record( + action="kubectl rollout restart deployment/awoooi-api", + risk_level=RiskLevel.MEDIUM, + ) + db = _FakeDB(record) + + @asynccontextmanager + async def fake_db_context(): + yield db + + monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context) + + approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id) + + assert promoted is True + assert approval is not None + assert approval.status == ApprovalStatus.APPROVED + assert approval.required_signatures == 0 + assert approval.metadata["legacy_pending_promoted"] is True + assert len(db.events) == 1 + + +@pytest.mark.asyncio +async def test_legacy_critical_pending_gate_is_not_promoted(monkeypatch) -> None: + record = _approval_record( + action="host reboot 192.168.0.110", + risk_level=RiskLevel.HIGH, + ) + db = _FakeDB(record) + + @asynccontextmanager + async def fake_db_context(): + yield db + + monkeypatch.setattr(approval_db_module, "get_db_context", fake_db_context) + + approval, promoted = await ApprovalDBService().apply_current_owner_policy(record.id) + + assert promoted is False + assert approval is not None + assert approval.status == ApprovalStatus.PENDING + assert approval.required_signatures == 1 + assert db.events == [] diff --git a/apps/api/tests/test_operator_outcome.py b/apps/api/tests/test_operator_outcome.py index 14b7db8a6..4b3a48bac 100644 --- a/apps/api/tests/test_operator_outcome.py +++ b/apps/api/tests/test_operator_outcome.py @@ -66,7 +66,7 @@ def test_operator_outcome_marks_diagnostic_only_as_ai_repair_required() -> None: assert outcome["schema_version"] == "operator_outcome_v1" assert outcome["state"] == "diagnostic_only_ai_repair_required" assert outcome["needs_human"] is False - assert outcome["notification"]["mode"] == "action_required" + assert outcome["notification"]["mode"] == "automation_progress" assert "telegram_sre_war_room" in outcome["notification"]["channels"] assert outcome["next_action"] == "auto_generate_repair_candidate_from_diagnostic_evidence" assert outcome["execution_result"]["completion_status"] == ( @@ -199,7 +199,7 @@ def test_operator_outcome_marks_expired_approval_as_manual_review() -> None: assert outcome["state"] == "approval_expired_ai_retry" assert outcome["needs_human"] is False - assert outcome["notification"]["mode"] == "action_required" + assert outcome["notification"]["mode"] == "automation_progress" assert outcome["next_action"] == "ai_retry_or_rebuild_controlled_packet" assert outcome["execution_result"]["completion_status"] == "expired_route_requeued" diff --git a/apps/api/tests/test_telegram_integration.py b/apps/api/tests/test_telegram_integration.py index 8850d3b17..ad7ba800e 100644 --- a/apps/api/tests/test_telegram_integration.py +++ b/apps/api/tests/test_telegram_integration.py @@ -131,7 +131,7 @@ class TestTelegramMessageStructure: result = msg.format() # 驗證關鍵區塊存在（ADR-075 TYPE-3 新格式） - assert "ACTION REQUIRED" in result + assert "BREAK-GLASS REQUIRED" in result assert "嚴重" in result # risk_level 改為中文 assert "api-service" in result assert "INC-20260321-0001" in result