From cd35a6a0722f740daf5226da452790cf00d35aa0 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 18 Jul 2026 13:00:40 +0800 Subject: [PATCH] fix(telegram): suppress healthy Wazuh posture spam --- .../services/awooop_ansible_audit_service.py | 1 + .../awooop_ansible_check_mode_service.py | 131 +++++++++++++++-- ...wooos_wazuh_controlled_executor_runtime.py | 37 ++++- ...gram_alert_monitoring_coverage_readback.py | 138 +++++++++++++++++- apps/api/src/services/telegram_gateway.py | 105 ++++++++++--- ...est_ansible_incident_ledger_churn_guard.py | 64 ++++++++ ...wooos_wazuh_controlled_executor_runtime.py | 31 ++++ ..._alert_monitoring_coverage_readback_api.py | 14 +- .../tests/test_telegram_message_templates.py | 124 ++++++++++++++++ 9 files changed, 606 insertions(+), 39 deletions(-) diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 4a32a73cb..f5f6fa058 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -1771,6 +1771,7 @@ def build_ansible_decision_audit_payload( "auto_apply_enabled": row["auto_apply_enabled"], "approval_required": row["approval_required"], "risk_level": row["risk_level"], + "no_write_only": row.get("no_write_only") is True, "catalog_revision": row.get("catalog_revision"), "canonical_asset_id": row.get("canonical_asset_id"), "typed_domain": row.get("typed_domain"), diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 762cf8e2a..26edec07d 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -346,6 +346,7 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]: "auto_apply_enabled": bool(catalog_item.get("auto_apply_enabled") is True), "approval_required": bool(catalog_item.get("approval_required") is True), "break_glass_required": bool(catalog_item.get("break_glass_required") is True), + "no_write_only": bool(catalog_item.get("no_write_only") is True), "canonical_asset_id": ( typed_route.get("canonical_asset_id") if typed_schema @@ -589,6 +590,13 @@ def _allowed_controlled_apply_risks() -> set[str]: return {item.strip().lower() for item in raw.split(",") if item.strip()} +def _claim_is_no_write_observation(claim: AnsibleCheckModeClaim) -> bool: + """Classify read-only catalog execution from the current allowlist.""" + + catalog_item = get_ansible_catalog_item(claim.catalog_id) or {} + return catalog_item.get("no_write_only") is True + + def _controlled_apply_allowed(candidate: dict[str, Any]) -> tuple[bool, str | None]: risk = str(candidate.get("risk_level") or "").strip().lower() if settings.ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY is not True: @@ -764,6 +772,8 @@ def build_ansible_check_mode_claim_input( "check_mode_playbook_path": safe["check_mode_playbook_path"], "inventory_hosts": list(safe["inventory_hosts"]), "risk_level": safe["risk_level"], + "no_write_only": safe["no_write_only"], + "runtime_write_expected": not safe["no_write_only"], "canonical_asset_id": safe["canonical_asset_id"], "typed_domain": safe["typed_domain"], "target_scope_validated": safe["target_scope_validated"], @@ -2767,6 +2777,7 @@ async def _read_verified_apply_closure_prerequisites( automation_run_id = _automation_run_id_for_claim(claim) approval_id = _approval_id_for_claim(claim) canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id) + telegram_shadow_allowed = _claim_is_no_write_observation(claim) async with get_db_context(project_id) as db: result = await db.execute( text(""" @@ -2835,8 +2846,26 @@ async def _read_verified_apply_closure_prerequisites( FROM awooop_outbound_message outbound WHERE outbound.project_id = :project_id AND outbound.channel_type = 'telegram' - AND outbound.send_status = 'sent' - AND outbound.provider_message_id IS NOT NULL + AND ( + ( + outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + ) + OR ( + :telegram_shadow_allowed + AND outbound.send_status = 'shadow' + AND outbound.provider_message_id IS NULL + AND outbound.source_envelope #>> + '{notification_policy,disposition}' + = 'suppressed' + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,execution_kind}', + outbound.source_envelope ->> + 'execution_kind' + ) = 'no_write_posture_readback' + ) + ) AND outbound.source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' @@ -2888,8 +2917,17 @@ async def _read_verified_apply_closure_prerequisites( SELECT 1 FROM alert_operation_log lifecycle WHERE lifecycle.incident_id = :incident_id - AND lifecycle.event_type::text - = 'TELEGRAM_RESULT_SENT' + AND ( + lifecycle.event_type::text + = 'TELEGRAM_RESULT_SENT' + OR ( + :telegram_shadow_allowed + AND lifecycle.event_type::text + = 'NOTIFICATION_CLASSIFIED' + AND lifecycle.action_detail + = 'no_write_posture_result_receipt_suppressed' + ) + ) AND lifecycle.success IS TRUE AND lifecycle.context ->> 'apply_op_id' = :apply_op_id @@ -2916,6 +2954,7 @@ async def _read_verified_apply_closure_prerequisites( "approval_id": approval_id, "canonical_playbook_id": canonical_playbook_id, "km_path_type": f"ansible_apply_receipt:{apply_op_id[:8]}", + "telegram_shadow_allowed": telegram_shadow_allowed, }, ) row = result.mappings().one_or_none() @@ -3216,6 +3255,7 @@ def _incident_terminal_disposition_payload( retry_op_id: str | None, telegram_receipt_acknowledged: bool, ) -> dict[str, Any]: + no_write_observation = _claim_is_no_write_observation(claim) return { "schema_version": "ansible_incident_terminal_disposition_v1", "automation_run_id": _automation_run_id_for_claim(claim), @@ -3225,8 +3265,13 @@ def _incident_terminal_disposition_payload( "incident_status": "RESOLVED" if success_terminal else "MITIGATING", "incident_resolved": success_terminal, "no_write_terminal": not success_terminal, + "observation_only": no_write_observation, + "repair_performed": success_terminal and not no_write_observation, + "runtime_write_performed": success_terminal and not no_write_observation, "safe_next_action": ( - "keep_verified_repair_and_monitor_recurrence" + "retain_posture_receipt_and_notify_only_on_state_transition" + if success_terminal and no_write_observation + else "keep_verified_repair_and_monitor_recurrence" if success_terminal else "queue_ai_playbook_or_transport_repair_candidate" ), @@ -3249,7 +3294,11 @@ async def _commit_verified_incident_closure( **_incident_terminal_disposition_payload( claim, apply_op_id=apply_op_id, - terminal_type="controlled_apply_verified_closed", + terminal_type=( + "no_write_posture_verified_closed" + if _claim_is_no_write_observation(claim) + else "controlled_apply_verified_closed" + ), success_terminal=True, retry_op_id=None, telegram_receipt_acknowledged=True, @@ -3831,6 +3880,7 @@ async def _retry_telegram_receipt_acknowledged( project_id: str, ) -> bool: automation_run_id = _automation_run_id_for_claim(claim) + telegram_shadow_allowed = _claim_is_no_write_observation(claim) try: async with get_db_context(project_id) as db: result = await db.execute( @@ -3840,8 +3890,26 @@ async def _retry_telegram_receipt_acknowledged( FROM awooop_outbound_message outbound WHERE outbound.project_id = :project_id AND outbound.channel_type = 'telegram' - AND outbound.send_status = 'sent' - AND outbound.provider_message_id IS NOT NULL + AND ( + ( + outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + ) + OR ( + :telegram_shadow_allowed + AND outbound.send_status = 'shadow' + AND outbound.provider_message_id IS NULL + AND outbound.source_envelope #>> + '{notification_policy,disposition}' + = 'suppressed' + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,execution_kind}', + outbound.source_envelope ->> + 'execution_kind' + ) = 'no_write_posture_readback' + ) + ) AND outbound.source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' @@ -3868,6 +3936,7 @@ async def _retry_telegram_receipt_acknowledged( "automation_run_id": automation_run_id, "incident_id": claim.incident_id, "apply_op_id": apply_op_id, + "telegram_shadow_allowed": telegram_shadow_allowed, }, ) return result.scalar() is True @@ -6202,10 +6271,22 @@ async def _send_controlled_apply_telegram_receipt( try: from src.services.telegram_gateway import get_telegram_gateway + no_write_observation = _claim_is_no_write_observation(claim) + verified_success = bool( + result.returncode == 0 + and str(writeback.get("verification_result") or "") == "success" + ) + effective_execution_kind = ( + "no_write_posture_readback" + if no_write_observation and execution_kind == "controlled_apply" + else execution_kind + ) effective_provider_delivery = str( claim.input_payload.get("telegram_provider_delivery") or provider_delivery ) + if no_write_observation and verified_success: + effective_provider_delivery = "shadow_only" response = await get_telegram_gateway().send_controlled_apply_result_receipt( automation_run_id=str( claim.input_payload.get("automation_run_id") @@ -6222,7 +6303,7 @@ async def _send_controlled_apply_telegram_receipt( duration_ms=result.duration_ms, verifier_written=bool(writeback.get("verification")), learning_written=bool(writeback.get("learning")), - execution_kind=execution_kind, + execution_kind=effective_execution_kind, provider_delivery=effective_provider_delivery, project_id=project_id, ) @@ -6279,12 +6360,21 @@ async def _reconcile_verified_apply_closure_projections( ) telegram_lifecycle = False if telegram_receipt: + notification_classified = _claim_is_no_write_observation(claim) telegram_lifecycle = await _append_alert_lifecycle_receipt( claim, - "TELEGRAM_RESULT_SENT", + ( + "NOTIFICATION_CLASSIFIED" + if notification_classified + else "TELEGRAM_RESULT_SENT" + ), apply_op_id=apply_op_id, success=False, - action_detail="controlled_apply_result_receipt_sent", + action_detail=( + "no_write_posture_failure_digest_recorded" + if notification_classified + else "controlled_apply_result_receipt_sent" + ), project_id=project_id, post_verifier_passed=result.post_verifier_passed, ) @@ -6307,7 +6397,11 @@ async def _reconcile_verified_apply_closure_projections( "EXECUTION_COMPLETED", apply_op_id=apply_op_id, success=True, - action_detail=f"controlled_apply_completed:{claim.catalog_id}", + action_detail=( + f"no_write_posture_readback_completed:{claim.catalog_id}" + if _claim_is_no_write_observation(claim) + else f"controlled_apply_completed:{claim.catalog_id}" + ), project_id=project_id, post_verifier_passed=True, ) @@ -6339,12 +6433,21 @@ async def _reconcile_verified_apply_closure_projections( telegram_lifecycle = False if telegram_receipt: + notification_suppressed = _claim_is_no_write_observation(claim) telegram_lifecycle = await _append_alert_lifecycle_receipt( claim, - "TELEGRAM_RESULT_SENT", + ( + "NOTIFICATION_CLASSIFIED" + if notification_suppressed + else "TELEGRAM_RESULT_SENT" + ), apply_op_id=apply_op_id, success=True, - action_detail="controlled_apply_result_receipt_sent", + action_detail=( + "no_write_posture_result_receipt_suppressed" + if notification_suppressed + else "controlled_apply_result_receipt_sent" + ), project_id=project_id, post_verifier_passed=True, ) diff --git a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py index d09bb0c4c..0ac1858f1 100644 --- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py @@ -302,6 +302,18 @@ _LIVE_RUNTIME_SQL = """ SELECT outbound.message_id::text AS telegram_receipt_id, outbound.send_status AS telegram_send_status, + outbound.provider_message_id AS telegram_provider_message_id, + coalesce( + outbound.source_envelope #>> + '{callback_reply,execution_kind}', + outbound.source_envelope ->> 'execution_kind' + ) AS telegram_execution_kind, + outbound.source_envelope #>> + '{notification_policy,disposition}' + AS telegram_notification_disposition, + outbound.source_envelope #>> + '{notification_policy,provider_delivery}' + AS telegram_provider_delivery, outbound.sent_at AS telegram_sent_at FROM awooop_outbound_message outbound JOIN latest_candidate candidate @@ -610,7 +622,20 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( trust_ready = bool(data.get("playbook_trust_acknowledged")) and ( "playbook_trust" in stage_ids ) - telegram_ready = data.get("telegram_send_status") == "sent" + telegram_provider_sent = bool( + data.get("telegram_send_status") == "sent" + and data.get("telegram_provider_message_id") + ) + telegram_notification_suppressed = bool( + not ingress_mode + and data.get("telegram_send_status") == "shadow" + and not data.get("telegram_provider_message_id") + and data.get("telegram_execution_kind") + == "no_write_posture_readback" + and data.get("telegram_notification_disposition") == "suppressed" + and data.get("telegram_provider_delivery") == "shadow_only" + ) + telegram_ready = telegram_provider_sent or telegram_notification_suppressed incident_closed = all( ( apply_passed, @@ -748,6 +773,10 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( ), "timeline_receipt_count": 1 if "timeline_projection" in stage_ids else 0, "telegram_receipt_count": 1 if telegram_ready else 0, + "telegram_provider_send_count": 1 if telegram_provider_sent else 0, + "telegram_notification_suppressed_count": ( + 1 if telegram_notification_suppressed else 0 + ), "same_run_identity_chain_verified_count": ( 1 if identity_chain_verified else 0 ), @@ -833,6 +862,12 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( "wazuh_active_response_performed": False, "live_wazuh_api_query_performed": False, "raw_wazuh_payload_persisted": False, + "healthy_posture_telegram_provider_send_performed": ( + telegram_provider_sent if not ingress_mode else False + ), + "healthy_posture_notification_suppressed": ( + telegram_notification_suppressed + ), "command_output_returned": False, "inventory_identity_returned": False, "secret_value_collection_allowed": False, diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py index 80c9daef7..a03a6c80b 100644 --- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py +++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py @@ -226,8 +226,10 @@ _RUNTIME_LIFECYCLE_SQL = """ OR ( telegram_result.event_type = 'NOTIFICATION_CLASSIFIED' - AND telegram_result.action_detail - = 'stdin_boundary_no_write_result_receipt_suppressed' + AND telegram_result.action_detail IN ( + 'stdin_boundary_no_write_result_receipt_suppressed', + 'no_write_posture_result_receipt_suppressed' + ) ) ) AND telegram_result.context @@ -253,8 +255,10 @@ _RUNTIME_LIFECYCLE_SQL = """ WHERE event_type = 'TELEGRAM_RESULT_SENT' OR ( event_type = 'NOTIFICATION_CLASSIFIED' - AND action_detail - = 'stdin_boundary_no_write_result_receipt_suppressed' + AND action_detail IN ( + 'stdin_boundary_no_write_result_receipt_suppressed', + 'no_write_posture_result_receipt_suppressed' + ) ) ) AS result_acknowledged_count, MAX(created_at) AS latest_at @@ -1000,6 +1004,20 @@ def build_telegram_alert_monitoring_coverage_readback( runtime_summary.get("automation_runtime_closure_percent") or 0.0 ), "runtime_automation_closure_ready": runtime_closure_ready, + "wazuh_posture_provider_sent_count_1h": _int( + runtime_summary.get("wazuh_posture_provider_sent_count_1h") + ), + "wazuh_posture_notification_suppressed_count_1h": _int( + runtime_summary.get( + "wazuh_posture_notification_suppressed_count_1h" + ) + ), + "latest_wazuh_posture_provider_sent_at": runtime_summary.get( + "latest_wazuh_posture_provider_sent_at" + ), + "latest_wazuh_posture_suppressed_at": runtime_summary.get( + "latest_wazuh_posture_suppressed_at" + ), "verified_context_receipt_count": _int( verifier_rollups.get("verified_context_receipt_count") ), @@ -1274,6 +1292,8 @@ async def _load_runtime_log_readback(*, project_id: str) -> dict[str, Any]: "automation_open_run_count_7d": 0, "automation_runtime_closure_percent": 0.0, "automation_runtime_closure_ready": False, + "wazuh_posture_provider_sent_count_1h": 0, + "wazuh_posture_notification_suppressed_count_1h": 0, "error_type": type(exc).__name__, }, "event_type_counts_7d": {}, @@ -1310,6 +1330,52 @@ async def _load_runtime_log_readback_once(*, project_id: str) -> dict[str, Any]: WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb) ? 'source_refs' ) AS outbound_source_refs_total, + COUNT(*) FILTER ( + WHERE queued_at >= NOW() - INTERVAL '1 hour' + AND send_status = 'sent' + AND provider_message_id IS NOT NULL + AND ( + source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + OR content LIKE + '%ansible:wazuh-manager-posture-readback%' + ) + ) AS wazuh_posture_provider_sent_count_1h, + COUNT(*) FILTER ( + WHERE queued_at >= NOW() - INTERVAL '1 hour' + AND send_status = 'shadow' + AND provider_message_id IS NULL + AND source_envelope #>> + '{notification_policy,disposition}' = 'suppressed' + AND ( + source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + OR content LIKE + '%ansible:wazuh-manager-posture-readback%' + ) + ) AS wazuh_posture_notification_suppressed_count_1h, + MAX(COALESCE(sent_at, queued_at)) FILTER ( + WHERE send_status = 'sent' + AND provider_message_id IS NOT NULL + AND ( + source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + OR content LIKE + '%ansible:wazuh-manager-posture-readback%' + ) + ) AS latest_wazuh_posture_provider_sent_at, + MAX(queued_at) FILTER ( + WHERE send_status = 'shadow' + AND provider_message_id IS NULL + AND source_envelope #>> + '{notification_policy,disposition}' = 'suppressed' + AND source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + ) AS latest_wazuh_posture_suppressed_at, MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at FROM awooop_outbound_message WHERE project_id = :project_id @@ -1387,6 +1453,52 @@ async def _load_runtime_log_readback_direct( WHERE COALESCE(source_envelope::jsonb, '{}'::jsonb) ? 'source_refs' ) AS outbound_source_refs_total, + COUNT(*) FILTER ( + WHERE queued_at >= NOW() - INTERVAL '1 hour' + AND send_status = 'sent' + AND provider_message_id IS NOT NULL + AND ( + source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + OR content LIKE + '%ansible:wazuh-manager-posture-readback%' + ) + ) AS wazuh_posture_provider_sent_count_1h, + COUNT(*) FILTER ( + WHERE queued_at >= NOW() - INTERVAL '1 hour' + AND send_status = 'shadow' + AND provider_message_id IS NULL + AND source_envelope #>> + '{notification_policy,disposition}' = 'suppressed' + AND ( + source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + OR content LIKE + '%ansible:wazuh-manager-posture-readback%' + ) + ) AS wazuh_posture_notification_suppressed_count_1h, + MAX(COALESCE(sent_at, queued_at)) FILTER ( + WHERE send_status = 'sent' + AND provider_message_id IS NOT NULL + AND ( + source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + OR content LIKE + '%ansible:wazuh-manager-posture-readback%' + ) + ) AS latest_wazuh_posture_provider_sent_at, + MAX(queued_at) FILTER ( + WHERE send_status = 'shadow' + AND provider_message_id IS NULL + AND source_envelope #>> + '{notification_policy,disposition}' = 'suppressed' + AND source_envelope #>> + '{callback_reply,catalog_id}' = + 'ansible:wazuh-manager-posture-readback' + ) AS latest_wazuh_posture_suppressed_at, MAX(COALESCE(sent_at, queued_at)) AS latest_outbound_at FROM awooop_outbound_message WHERE project_id = $1 @@ -1501,6 +1613,24 @@ def _runtime_log_readback_from_rows( "outbound_source_refs_total_7d": _int( outbound_row.get("outbound_source_refs_total") ), + "wazuh_posture_provider_sent_count_1h": _int( + outbound_row.get("wazuh_posture_provider_sent_count_1h") + ), + "wazuh_posture_notification_suppressed_count_1h": _int( + outbound_row.get( + "wazuh_posture_notification_suppressed_count_1h" + ) + ), + "latest_wazuh_posture_provider_sent_at": ( + str(outbound_row.get("latest_wazuh_posture_provider_sent_at")) + if outbound_row.get("latest_wazuh_posture_provider_sent_at") + else None + ), + "latest_wazuh_posture_suppressed_at": ( + str(outbound_row.get("latest_wazuh_posture_suppressed_at")) + if outbound_row.get("latest_wazuh_posture_suppressed_at") + else None + ), "latest_alert_operation_log_at": latest_event_at, "latest_outbound_telegram_at": str(outbound_row.get("latest_outbound_at")) if outbound_row.get("latest_outbound_at") diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index a92f66dcf..b3cf76e9f 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -12653,29 +12653,32 @@ class TelegramGateway: """Send and mirror the AI Agent controlled-apply result receipt.""" no_write_replay = execution_kind == "no_write_replay" + no_write_posture = execution_kind == "no_write_posture_readback" success = ( not no_write_replay and verification_result == "success" and returncode == 0 ) - next_step = ( - "monitor_for_regression" - if success - else ( - "queue_ai_transport_or_playbook_repair" - if no_write_replay - else "queue_ai_rollback_or_playbook_repair" + if no_write_posture: + next_step = ( + "record_healthy_posture_without_notification" + if success + else "queue_ai_transport_or_playbook_repair" ) - ) - title = ( - "CONTROLLED APPLY RESULT｜AI Agent 受控執行完成" - if success - else ( - "AI 自動修復摘要｜PlayBook 檢查待修復" - if no_write_replay - else "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復" + title = ( + "AI 姿態讀回｜健康狀態已驗證" + if success + else "AI 自動修復摘要｜姿態檢查待修復" ) - ) + elif success: + next_step = "monitor_for_regression" + title = "CONTROLLED APPLY RESULT｜AI Agent 受控執行完成" + elif no_write_replay: + next_step = "queue_ai_transport_or_playbook_repair" + title = "AI 自動修復摘要｜PlayBook 檢查待修復" + else: + next_step = "queue_ai_rollback_or_playbook_repair" + title = "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復" failure_fingerprint = ( _no_write_replay_failure_fingerprint( catalog_id=catalog_id, @@ -12684,7 +12687,7 @@ class TelegramGateway: returncode=returncode, next_step=next_step, ) - if no_write_replay + if no_write_replay or (no_write_posture and not success) else "" ) truth_chain: dict[str, object] | None = None @@ -12732,10 +12735,45 @@ class TelegramGateway: callback_reply["apply_op_id"] = apply_op_id callback_reply["project_id"] = project_id or "awoooi" callback_reply["execution_kind"] = execution_kind + callback_reply["catalog_id"] = catalog_id + callback_reply["runtime_write_performed"] = not ( + no_write_replay or no_write_posture + ) source_refs = source_extra.get("source_refs") if isinstance(source_refs, dict): source_refs["automation_run_ids"] = [automation_run_id] - if no_write_replay: + if no_write_posture: + effective_provider_delivery = ( + "shadow_only" + if success or provider_delivery == "shadow_only" + else "digest" + ) + source_extra["notification_policy"] = { + "policy_version": "telegram_posture_state_digest_v1", + "failure_fingerprint": failure_fingerprint, + "group_scope": ( + "scheduled_posture_healthy" + if success + else "catalog_posture_failure" + ), + "provider_delivery": effective_provider_delivery, + "digest_window_minutes": ( + _NO_WRITE_REPLAY_DIGEST_WINDOW_MINUTES + ), + "disposition": ( + "suppressed" + if effective_provider_delivery == "shadow_only" + else "send_or_suppress" + ), + "provider_send_performed": False if success else None, + "runtime_write_performed": False, + "details_retained_in": [ + "automation_operation_log", + "awooop_runs", + "km_rag_playbook", + ], + } + elif no_write_replay: effective_provider_delivery = ( "shadow_only" if provider_delivery == "shadow_only" @@ -12775,7 +12813,36 @@ class TelegramGateway: "km_rag_playbook", ], } - if no_write_replay: + if no_write_posture: + lines = [ + f"{html.escape(title)}", + "階段: controlled_observe", + f"目標: {html.escape(str(catalog_id or '--'))}", + ( + "狀態: posture " + f"{html.escape(str(verification_result or 'failed'))} " + f"/ rc {html.escape(str(returncode))}" + ), + ( + "AI 處置: healthy_receipt_recorded" + if success + else "AI 處置: transport_or_playbook_repair_queued" + ), + "Runtime write: 0", + ( + "Receipts: verifier " + f"{html.escape(_bool_code(verifier_written))} / KM " + f"{html.escape(_bool_code(learning_written))}" + ), + ( + "證據: " + f"{html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}" + ), + ( + "通知: 健康週期讀回不外送；異常相同原因 30 分鐘只通知一次。" + ), + ] + elif no_write_replay: lines = [ f"{html.escape(title)}", "階段: controlled_check", diff --git a/apps/api/tests/test_ansible_incident_ledger_churn_guard.py b/apps/api/tests/test_ansible_incident_ledger_churn_guard.py index e49d73361..87b0ee865 100644 --- a/apps/api/tests/test_ansible_incident_ledger_churn_guard.py +++ b/apps/api/tests/test_ansible_incident_ledger_churn_guard.py @@ -381,3 +381,67 @@ async def test_backfill_repairs_orphan_incident_and_reuses_verifier( assert result["incident_closure_written"] == 1 incident_ledger.assert_awaited_once() assert writeback.await_args.kwargs["durable_verifier_receipt"] == receipt + + +@pytest.mark.asyncio +async def test_wazuh_posture_success_receipt_is_shadow_only_no_write( + monkeypatch: pytest.MonkeyPatch, +) -> None: + gateway = AsyncMock() + gateway.send_controlled_apply_result_receipt.return_value = { + "ok": True, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "suppressed", + } + monkeypatch.setattr( + "src.services.telegram_gateway.get_telegram_gateway", + lambda: gateway, + ) + + acknowledged = await service._send_controlled_apply_telegram_receipt( + _wazuh_claim(), + service.AnsibleRunResult( + returncode=0, + stdout="", + stderr="", + duration_ms=5, + post_verifier_passed=True, + ), + apply_op_id="00000000-0000-0000-0000-000000000203", + writeback={ + "verification_result": "success", + "verification": True, + "learning": True, + }, + project_id="awoooi", + ) + + assert acknowledged is True + call = gateway.send_controlled_apply_result_receipt.await_args.kwargs + assert call["execution_kind"] == "no_write_posture_readback" + assert call["provider_delivery"] == "shadow_only" + + +def test_wazuh_posture_shadow_receipt_is_a_terminal_notification_receipt() -> None: + closure_source = service._read_verified_apply_closure_prerequisites.__doc__ or "" + assert service._claim_is_no_write_observation(_wazuh_claim()) is True + assert "durable prerequisite" in closure_source + + import inspect + + readback_source = inspect.getsource( + service._read_verified_apply_closure_prerequisites + ) + retry_source = inspect.getsource( + service._retry_telegram_receipt_acknowledged + ) + reconcile_source = inspect.getsource( + service._reconcile_verified_apply_closure_projections + ) + for source in (readback_source, retry_source): + assert "telegram_shadow_allowed" in source + assert "no_write_posture_readback" in source + assert "send_status = 'shadow'" in source + assert "no_write_posture_result_receipt_suppressed" in reconcile_source + assert "no_write_posture_failure_digest_recorded" in reconcile_source + assert '"NOTIFICATION_CLASSIFIED"' in reconcile_source diff --git a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py index 3eb060dc8..58e021b27 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -122,6 +122,7 @@ def _complete_row() -> dict[str, object]: "auto_repair_success": True, "telegram_receipt_id": "telegram-303", "telegram_send_status": "sent", + "telegram_provider_message_id": "303", "stage_ids": [ "mcp_context", "service_log_evidence", @@ -165,6 +166,36 @@ def test_wazuh_runtime_readback_closes_only_with_complete_same_run_receipts() -> assert payload["boundaries"]["command_output_returned"] is False +def test_wazuh_posture_shadow_receipt_closes_without_provider_send() -> None: + row = _complete_row() + row.update( + { + "telegram_send_status": "shadow", + "telegram_provider_message_id": None, + "telegram_execution_kind": "no_write_posture_readback", + "telegram_notification_disposition": "suppressed", + "telegram_provider_delivery": "shadow_only", + } + ) + + payload = build_iwooos_wazuh_controlled_executor_runtime_readback( + row, + executor_enabled=True, + ) + + assert payload["summary"]["runtime_closed_count"] == 1 + assert payload["summary"]["telegram_receipt_count"] == 1 + assert payload["summary"]["telegram_provider_send_count"] == 0 + assert payload["summary"]["telegram_notification_suppressed_count"] == 1 + assert ( + payload["boundaries"][ + "healthy_posture_telegram_provider_send_performed" + ] + is False + ) + assert payload["boundaries"]["healthy_posture_notification_suppressed"] is True + + def test_wazuh_runtime_readback_prefers_ingress_convergence_semantics() -> None: row = _complete_row() row["catalog_id"] = "ansible:wazuh-alertmanager-integration" diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py index 2a78f32db..95eb291ca 100644 --- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py +++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py @@ -788,7 +788,13 @@ def test_telegram_alert_runtime_log_calculates_incident_closure() -> None: "latest_triggered_at": "2026-07-11T02:00:00+00:00", "latest_closed_at": "2026-07-11T01:59:00+00:00", }, - outbound_row={}, + outbound_row={ + "wazuh_posture_provider_sent_count_1h": 0, + "wazuh_posture_notification_suppressed_count_1h": 4, + "latest_wazuh_posture_suppressed_at": ( + "2026-07-18T04:32:00+00:00" + ), + }, readback_source="test", ) @@ -798,6 +804,11 @@ def test_telegram_alert_runtime_log_calculates_incident_closure() -> None: assert summary["automation_open_run_count_7d"] == 1 assert summary["automation_runtime_closure_percent"] == 66.7 assert summary["automation_runtime_closure_ready"] is False + assert summary["wazuh_posture_provider_sent_count_1h"] == 0 + assert summary["wazuh_posture_notification_suppressed_count_1h"] == 4 + assert summary["latest_wazuh_posture_suppressed_at"] == ( + "2026-07-18T04:32:00+00:00" + ) def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None: @@ -835,6 +846,7 @@ def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None: assert "= 'TELEGRAM_RESULT_SENT'" in sql assert "= 'NOTIFICATION_CLASSIFIED'" in sql assert "stdin_boundary_no_write_result_receipt_suppressed" in sql + assert "no_write_posture_result_receipt_suppressed" in sql assert "result_acknowledged_count" in sql assert "completed_success_count" not in sql diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py index 66498f73a..e3715b42c 100644 --- a/apps/api/tests/test_telegram_message_templates.py +++ b/apps/api/tests/test_telegram_message_templates.py @@ -1441,6 +1441,130 @@ async def test_no_write_replay_receipt_never_claims_runtime_apply(monkeypatch) - ] +@pytest.mark.asyncio +async def test_healthy_posture_readback_is_suppressed_and_never_claims_apply( + monkeypatch, +) -> None: + gateway = TelegramGateway() + sent_requests = [] + + async def fake_send_request(method, payload): + sent_requests.append((method, payload)) + return { + "ok": True, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "suppressed", + } + + async def fake_fetch_truth_chain(**_kwargs): + return { + "truth_status": {}, + "automation_quality": {}, + "execution": {}, + } + + async def fake_fetch_km_completion_summary(**_kwargs): + return {} + + monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) + monkeypatch.setattr(gateway, "_send_request", fake_send_request) + monkeypatch.setattr( + "src.services.awooop_truth_chain_service.fetch_truth_chain", + fake_fetch_truth_chain, + ) + monkeypatch.setattr( + "src.services.telegram_gateway._fetch_km_stale_completion_summary_for_incident", + fake_fetch_km_completion_summary, + ) + + result = await gateway.send_controlled_apply_result_receipt( + automation_run_id="00000000-0000-0000-0000-000000000003", + incident_id="IWZ-POSTURE-2026071806", + catalog_id="ansible:wazuh-manager-posture-readback", + apply_op_id="93b7a95c-3652-4c0d-bb4c-729e500acedb", + playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml", + verification_result="success", + returncode=0, + duration_ms=8649, + verifier_written=True, + learning_written=True, + execution_kind="no_write_posture_readback", + project_id="awoooi", + ) + + assert result["ok"] is True + method, payload = sent_requests[0] + assert method == "sendMessage" + assert "AI 姿態讀回｜健康狀態已驗證" in payload["text"] + assert "Runtime write: 0" in payload["text"] + assert "CONTROLLED APPLY RESULT" not in payload["text"] + assert "Runtime apply: 1" not in payload["text"] + source_extra = payload["_awooop_source_envelope_extra"] + callback = source_extra["callback_reply"] + assert callback["execution_kind"] == "no_write_posture_readback" + assert callback["catalog_id"] == "ansible:wazuh-manager-posture-readback" + assert callback["runtime_write_performed"] is False + policy = source_extra["notification_policy"] + assert policy["policy_version"] == "telegram_posture_state_digest_v1" + assert policy["provider_delivery"] == "shadow_only" + assert policy["disposition"] == "suppressed" + assert policy["provider_send_performed"] is False + assert policy["runtime_write_performed"] is False + + +@pytest.mark.asyncio +async def test_failed_posture_readback_is_digest_deduplicated_and_queues_repair( + monkeypatch, +) -> None: + gateway = TelegramGateway() + sent_requests = [] + + async def fake_send_request(method, payload): + sent_requests.append((method, payload)) + return {"ok": True, "result": {"message_id": 12347}} + + async def fake_fetch_truth_chain(**_kwargs): + return {"truth_status": {}, "automation_quality": {}, "execution": {}} + + async def fake_fetch_km_completion_summary(**_kwargs): + return {} + + monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat")) + monkeypatch.setattr(gateway, "_send_request", fake_send_request) + monkeypatch.setattr( + "src.services.awooop_truth_chain_service.fetch_truth_chain", + fake_fetch_truth_chain, + ) + monkeypatch.setattr( + "src.services.telegram_gateway._fetch_km_stale_completion_summary_for_incident", + fake_fetch_km_completion_summary, + ) + + await gateway.send_controlled_apply_result_receipt( + automation_run_id="00000000-0000-0000-0000-000000000004", + incident_id="IWZ-POSTURE-2026071812", + catalog_id="ansible:wazuh-manager-posture-readback", + apply_op_id="a3b7a95c-3652-4c0d-bb4c-729e500acedb", + playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml", + verification_result="failed", + returncode=2, + duration_ms=9000, + verifier_written=True, + learning_written=True, + execution_kind="no_write_posture_readback", + project_id="awoooi", + ) + + payload = sent_requests[0][1] + assert "AI 自動修復摘要｜姿態檢查待修復" in payload["text"] + assert "transport_or_playbook_repair_queued" in payload["text"] + assert "Runtime write: 0" in payload["text"] + policy = payload["_awooop_source_envelope_extra"]["notification_policy"] + assert policy["provider_delivery"] == "digest" + assert policy["digest_window_minutes"] == 30 + assert len(policy["failure_fingerprint"]) == 24 + + def test_callback_reply_awooop_status_chain_snapshot_marks_manual_gate() -> None: """Callback evidence 要保存當下 AwoooP 狀態鏈，不只保存 live query 結果。""" snapshot = telegram_gateway_module._callback_reply_awooop_status_chain_snapshot(