diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index 6fcf4b44f..c1d64946b 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -13,6 +13,7 @@ import json import os import random from collections.abc import Awaitable, Callable, Mapping +from datetime import UTC, datetime from types import SimpleNamespace from typing import Any from uuid import NAMESPACE_URL, uuid4, uuid5 @@ -64,6 +65,68 @@ _WAZUH_POSTURE_WORK_ITEM_ID = "P0-03-WAZUH-MANAGER-POSTURE" _WAZUH_POSTURE_PUBLIC_ASSET_ALIAS = "security_manager_primary" _WAZUH_INGRESS_CATALOG_ID = "ansible:wazuh-alertmanager-integration" _WAZUH_INGRESS_WORK_ITEM_ID = "P0-03-WAZUH-ALERT-INGRESS" +_WAZUH_SOURCE_RECURRENCE_DRIFT_WORK_ITEM_ID = ( + "DRIFT-WAZUH-SOURCE-RECURRENCE" +) +_WAZUH_ALLOWED_RECURRENCE_SOURCES = frozenset( + {"current_alert_webhook", "awooop_conversation_event"} +) + + +def _validated_wazuh_source_recurrence( + receipt: Mapping[str, Any] | None, + *, + observed_now: datetime, +) -> dict[str, Any] | None: + """Accept only a current external source receipt, never scheduler evidence.""" + + if not isinstance(receipt, Mapping): + return None + if ( + receipt.get("schema_version") != "alert_source_recurrence_receipt_v1" + or receipt.get("verified") is not True + ): + return None + fingerprint = str(receipt.get("fingerprint") or "").strip() + occurrence_id = str(receipt.get("occurrence_id") or "").strip() + identity_anchor = str(receipt.get("identity_anchor") or "").strip() + canonical_asset_id = str( + receipt.get("canonical_asset_id") or "" + ).strip() + source = str(receipt.get("source") or "").strip() + observed_at_raw = str(receipt.get("observed_at") or "").strip() + if ( + not fingerprint + or not occurrence_id + or not identity_anchor + or canonical_asset_id != "service:wazuh-manager" + or source not in _WAZUH_ALLOWED_RECURRENCE_SOURCES + or not observed_at_raw + ): + return None + try: + observed_at = datetime.fromisoformat( + observed_at_raw.replace("Z", "+00:00") + ) + except ValueError: + return None + if observed_at.tzinfo is None or observed_now.tzinfo is None: + return None + age_seconds = ( + observed_now.astimezone(UTC) - observed_at.astimezone(UTC) + ).total_seconds() + if age_seconds < -60 or age_seconds > _CURRENT_RECURRENCE_MAX_AGE_SECONDS: + return None + return { + "schema_version": "alert_source_recurrence_receipt_v1", + "verified": True, + "fingerprint": fingerprint, + "occurrence_id": occurrence_id, + "observed_at": observed_at.isoformat(), + "identity_anchor": identity_anchor, + "canonical_asset_id": canonical_asset_id, + "source": source, + } async def _fetch_missing_candidate_incidents( @@ -789,6 +852,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "check-mode and independent verifier chain" ), _incident_binder: IncidentBinder | None = None, + source_recurrence: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Enqueue one fresh, no-write Wazuh manager posture executor run.""" @@ -801,6 +865,23 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "active_blockers": ["wazuh_manager_posture_executor_disabled"], } + observed_now = now_taipei() + validated_recurrence = _validated_wazuh_source_recurrence( + source_recurrence, + observed_now=observed_now, + ) + if validated_recurrence is None: + return { + "status": "blocked_current_source_recurrence_not_verified", + "queued": 0, + "existing": 0, + "evidence_ready": 0, + "work_item_id": _work_item_id, + "drift_work_item_id": _WAZUH_SOURCE_RECURRENCE_DRIFT_WORK_ITEM_ID, + "controlled_apply_allowed": False, + "active_blockers": ["current_source_recurrence_not_verified"], + } + freshness_hours = max( 1, int(settings.IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS), @@ -866,6 +947,12 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( ) learning ON TRUE WHERE candidate.operation_type = 'ansible_candidate_matched' AND candidate.input -> 'executor_candidates' @> CAST(:catalog_match AS jsonb) + AND candidate.input ->> 'project_id' = :project_id + AND candidate.input ->> 'work_item_id' = :work_item_id + AND candidate.input ->> 'proposal_source' = :proposal_source + AND candidate.input ->> 'run_namespace' = :run_namespace + AND candidate.input #>> '{source_recurrence,fingerprint}' = :source_fingerprint + AND candidate.input #>> '{source_recurrence,occurrence_id}' = :source_occurrence_id AND candidate.created_at >= NOW() - (:freshness_hours * INTERVAL '1 hour') ORDER BY candidate.created_at DESC LIMIT 1 @@ -875,6 +962,12 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( [{"catalog_id": _catalog_id}] ), "freshness_hours": freshness_hours, + "project_id": project_id, + "work_item_id": _work_item_id, + "proposal_source": _scheduler_source, + "run_namespace": _run_namespace, + "source_fingerprint": validated_recurrence["fingerprint"], + "source_occurrence_id": validated_recurrence["occurrence_id"], }, ) existing_row = existing.mappings().first() @@ -979,7 +1072,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( ], } - now = now_taipei() + now = observed_now bucket_hour = (now.hour // freshness_hours) * freshness_hours bucket = now.replace(hour=bucket_hour, minute=0, second=0, microsecond=0) automation_run_id = str( @@ -994,6 +1087,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( bucket_ref=bucket.strftime("%Y%m%d%H"), attempt_ref=retry_attempt_ref, ) + incident["run_namespace"] = _run_namespace if _incident_binder is not None: try: incident_bound = await _incident_binder( @@ -1064,6 +1158,17 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "risk_level": _risk_level, "execution_priority": 0, "action": _proposal_action, + "source_fingerprint": validated_recurrence["fingerprint"], + "source_occurrence_id": validated_recurrence["occurrence_id"], + "source_received_at": validated_recurrence["observed_at"], + "source_recurrence_verified": True, + "source_recurrence_source": validated_recurrence["source"], + "source_recurrence_anchor": validated_recurrence[ + "identity_anchor" + ], + "source_canonical_asset_id": validated_recurrence[ + "canonical_asset_id" + ], } if build_ansible_decision_audit_payload( incident=incident, @@ -1145,6 +1250,7 @@ async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( evidence_collector: EvidenceCollector = collect_ansible_candidate_pre_decision_evidence, evidence_verifier: EvidenceVerifier = verify_ansible_candidate_pre_decision_evidence, incident_binder: IncidentBinder = ensure_iwooos_wazuh_scheduled_incident, + source_recurrence: Mapping[str, Any] | None = None, ) -> dict[str, Any]: """Enqueue idempotent Wazuh event-ingress convergence before posture work.""" @@ -1165,6 +1271,7 @@ async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( "check-mode, bounded apply, rollback, and independent verifier chain" ), _incident_binder=incident_binder, + source_recurrence=source_recurrence, ) diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index d08a0a872..356d9ceea 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -26,6 +26,14 @@ from src.services.controlled_alert_target_router import ( logger = structlog.get_logger(__name__) _TYPED_ROUTE_GENERATION_PREFIX = "typed-route-v2" +_RECURRENCE_REQUIRED_PROPOSAL_SOURCES = frozenset( + { + "alert_webhook_controlled_router", + "truth_chain_candidate_backfill", + "iwooos_wazuh_manager_posture_scheduler", + "iwooos_wazuh_alert_ingress_scheduler", + } +) def _source_fingerprint(incident: dict[str, Any], proposal: dict[str, Any]) -> str: @@ -75,6 +83,10 @@ def _source_recurrence_receipt( proposal.get("source_recurrence_anchor") or "" ) or None, + "canonical_asset_id": str( + proposal.get("source_canonical_asset_id") or "" + ) + or None, "source": ( "current_alert_webhook" if proposal_source == "alert_webhook_controlled_router" @@ -1462,6 +1474,7 @@ def build_ansible_decision_audit_payload( "trace_id": str(incident_payload.get("trace_id") or ""), "run_id": str(incident_payload.get("run_id") or ""), "work_item_id": str(incident_payload.get("work_item_id") or ""), + "run_namespace": str(incident_payload.get("run_namespace") or ""), "source_receipt_ref": str(incident_payload.get("source_receipt_ref") or ""), "asset_scope_aliases": [ str(alias) @@ -1579,11 +1592,7 @@ async def preflight_ansible_candidate_generation( source_recurrence = payload["input"].get("source_recurrence") or {} proposal_source = str(payload["input"].get("proposal_source") or "") if ( - proposal_source - in { - "alert_webhook_controlled_router", - "truth_chain_candidate_backfill", - } + proposal_source in _RECURRENCE_REQUIRED_PROPOSAL_SOURCES and source_recurrence.get("verified") is not True ): return { @@ -1732,11 +1741,7 @@ async def record_ansible_decision_audit( source_recurrence = payload["input"].get("source_recurrence") or {} proposal_source = str(payload["input"].get("proposal_source") or "") if ( - proposal_source - in { - "alert_webhook_controlled_router", - "truth_chain_candidate_backfill", - } + proposal_source in _RECURRENCE_REQUIRED_PROPOSAL_SOURCES and source_recurrence.get("verified") is not True ): logger.warning( 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 d0ed335c8..01395ee46 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -100,6 +100,20 @@ _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS = frozenset( } ) _NO_WRITE_REPLAY_REQUIRED_STAGE_IDS = _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS +_WAZUH_SCHEDULED_IDENTITY = { + "iwooos_wazuh_manager_posture_scheduler": ( + "P0-03-WAZUH-MANAGER-POSTURE", + "awoooi:iwooos:wazuh-manager-posture", + ), + "iwooos_wazuh_alert_ingress_scheduler": ( + "P0-03-WAZUH-ALERT-INGRESS", + "awoooi:iwooos:wazuh-alert-ingress", + ), +} +_WAZUH_ALLOWED_RECURRENCE_SOURCES = frozenset( + {"current_alert_webhook", "awooop_conversation_event"} +) +_WAZUH_RECURRENCE_MAX_AGE_SECONDS = 15 * 60 FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap" REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519") REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts") @@ -563,6 +577,7 @@ def build_ansible_check_mode_claim_input( *, source_candidate_op_id: str, candidate_input: dict[str, Any], + observed_now: datetime | None = None, ) -> dict[str, Any]: safe = _safe_candidate(candidate_input) incident_id = _incident_id_from_payload(candidate_input) @@ -573,9 +588,86 @@ def build_ansible_check_mode_claim_input( if proposal_risk_level == "critical": controlled_apply_allowed = False controlled_apply_blocker = "critical_break_glass_required" + proposal_source = str(candidate_input.get("proposal_source") or "").strip() + source_recurrence = candidate_input.get("source_recurrence") + if not isinstance(source_recurrence, Mapping): + source_recurrence = {} + scheduled_identity = _WAZUH_SCHEDULED_IDENTITY.get(proposal_source) + if scheduled_identity is not None: + expected_work_item_id, expected_run_namespace = scheduled_identity + project_id = str(candidate_input.get("project_id") or "").strip() + trace_id = str(candidate_input.get("trace_id") or "").strip() + run_id = str(candidate_input.get("run_id") or "").strip() + work_item_id = str(candidate_input.get("work_item_id") or "").strip() + run_namespace = str(candidate_input.get("run_namespace") or "").strip() + automation_run_id = str( + candidate_input.get("automation_run_id") or "" + ).strip() + recurrence_fields_ready = all( + str(source_recurrence.get(key) or "").strip() + for key in ( + "fingerprint", + "occurrence_id", + "observed_at", + "identity_anchor", + "canonical_asset_id", + "source", + ) + ) + recurrence_observed_at: datetime | None = None + try: + recurrence_observed_at = datetime.fromisoformat( + str(source_recurrence.get("observed_at") or "").replace( + "Z", "+00:00" + ) + ) + except ValueError: + recurrence_fields_ready = False + current_time = observed_now or datetime.now(UTC) + recurrence_age_seconds: float | None = None + if ( + recurrence_observed_at is not None + and recurrence_observed_at.tzinfo is not None + and current_time.tzinfo is not None + ): + recurrence_age_seconds = ( + current_time.astimezone(UTC) + - recurrence_observed_at.astimezone(UTC) + ).total_seconds() + recurrence_current = bool( + recurrence_age_seconds is not None + and -60 <= recurrence_age_seconds <= _WAZUH_RECURRENCE_MAX_AGE_SECONDS + and str(source_recurrence.get("source") or "") + in _WAZUH_ALLOWED_RECURRENCE_SOURCES + and source_recurrence.get("canonical_asset_id") + == "service:wazuh-manager" + ) + if not ( + project_id + and trace_id == run_id == automation_run_id == str(source_candidate_op_id) + and work_item_id == expected_work_item_id + and run_namespace == expected_run_namespace + and str(candidate_input.get("source_receipt_ref") or "").strip() + and source_recurrence.get("schema_version") + == "alert_source_recurrence_receipt_v1" + and source_recurrence.get("verified") is True + and recurrence_fields_ready + and recurrence_current + ): + raise ValueError("scheduled_candidate_identity_chain_not_verified") return { "automation_run_id": str(source_candidate_op_id), "incident_id": incident_id, + "project_id": str(candidate_input.get("project_id") or ""), + "trace_id": str(candidate_input.get("trace_id") or ""), + "run_id": str(candidate_input.get("run_id") or ""), + "work_item_id": str(candidate_input.get("work_item_id") or ""), + "run_namespace": str(candidate_input.get("run_namespace") or ""), + "source_receipt_ref": str( + candidate_input.get("source_receipt_ref") or "" + ), + "source_recurrence": dict(source_recurrence), + "proposal_source": proposal_source, "approval_id": str(candidate_input.get("approval_id") or ""), "executor": "ansible", "execution_backend": "ansible", @@ -1211,6 +1303,21 @@ def _claim_from_stale_check_mode_row( source_candidate_op_id=source_candidate_op_id, candidate_input={ "incident_id": incident_id, + **{ + key: input_payload.get(key) + for key in ( + "project_id", + "trace_id", + "run_id", + "work_item_id", + "run_namespace", + "source_receipt_ref", + "source_recurrence", + "proposal_source", + "automation_run_id", + ) + if key in input_payload + }, "executor": "ansible", "executor_candidates": [ { @@ -1305,6 +1412,21 @@ def _claim_from_failed_apply_catalog_repair_row( source_candidate_op_id=source_candidate_op_id, candidate_input={ "incident_id": incident_id, + **{ + key: input_payload.get(key) + for key in ( + "project_id", + "trace_id", + "run_id", + "work_item_id", + "run_namespace", + "source_receipt_ref", + "source_recurrence", + "proposal_source", + "automation_run_id", + ) + if key in input_payload + }, "approval_id": input_payload.get("approval_id"), "proposal_risk_level": input_payload.get( "proposal_risk_level" @@ -1887,6 +2009,11 @@ def _runtime_stage_receipt( "stage_id": stage_id, "automation_run_id": _automation_run_id_for_claim(claim), "incident_id": claim.incident_id, + "project_id": str(claim.input_payload.get("project_id") or ""), + "trace_id": str(claim.input_payload.get("trace_id") or ""), + "run_id": str(claim.input_payload.get("run_id") or ""), + "work_item_id": str(claim.input_payload.get("work_item_id") or ""), + "source_recurrence": claim.input_payload.get("source_recurrence") or {}, "evidence_ref": evidence_ref, "detail": detail, "durable_receipt": True, @@ -5473,6 +5600,12 @@ async def _record_post_apply_verifier_and_learning( ) post_state = { **verifier_receipt, + "automation_run_id": automation_run_id, + "project_id": str(claim.input_payload.get("project_id") or ""), + "trace_id": str(claim.input_payload.get("trace_id") or ""), + "run_id": str(claim.input_payload.get("run_id") or ""), + "work_item_id": str(claim.input_payload.get("work_item_id") or ""), + "source_recurrence": claim.input_payload.get("source_recurrence") or {}, "playbook_path": claim.apply_playbook_path, "timed_out": result.timed_out, "action_taken": action_label, @@ -8399,6 +8532,20 @@ async def _insert_skipped_candidate( input_payload = { "automation_run_id": str(source_candidate_op_id), "incident_id": _incident_id_from_payload(candidate_input), + **{ + key: candidate_input.get(key) + for key in ( + "project_id", + "trace_id", + "run_id", + "work_item_id", + "run_namespace", + "source_receipt_ref", + "source_recurrence", + "proposal_source", + ) + if key in candidate_input + }, "executor": "ansible", "execution_backend": "ansible", "execution_mode": "check_mode", 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 d206b8041..75d4ccc46 100644 --- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py @@ -26,6 +26,11 @@ SCHEMA_VERSION = "iwooos_wazuh_controlled_executor_runtime_readback_v1" CATALOG_ID = "ansible:wazuh-manager-posture-readback" INGRESS_CATALOG_ID = "ansible:wazuh-alertmanager-integration" WORK_ITEM_ID = "P0-03-WAZUH-MANAGER-POSTURE" +INGRESS_WORK_ITEM_ID = "P0-03-WAZUH-ALERT-INGRESS" +POSTURE_PROPOSAL_SOURCE = "iwooos_wazuh_manager_posture_scheduler" +INGRESS_PROPOSAL_SOURCE = "iwooos_wazuh_alert_ingress_scheduler" +POSTURE_RUN_NAMESPACE = "awoooi:iwooos:wazuh-manager-posture" +INGRESS_RUN_NAMESPACE = "awoooi:iwooos:wazuh-alert-ingress" _LIVE_RUNTIME_SQL = """ WITH latest_candidate AS ( @@ -35,8 +40,16 @@ _LIVE_RUNTIME_SQL = """ input ->> 'automation_run_id' AS automation_run_id, input ->> 'trace_id' AS trace_id, input ->> 'run_id' AS run_id, + input ->> 'project_id' AS candidate_project_id, input ->> 'work_item_id' AS work_item_id, + input ->> 'proposal_source' AS proposal_source, + input ->> 'run_namespace' AS run_namespace, input ->> 'source_receipt_ref' AS source_receipt_ref, + input -> 'source_recurrence' AS candidate_source_recurrence, + coalesce( + (input -> 'source_recurrence' ->> 'verified')::boolean, + false + ) AS candidate_source_recurrence_verified, input -> 'executor_candidates' -> 0 ->> 'catalog_id' AS catalog_id, jsonb_array_length( coalesce(input -> 'asset_scope_aliases', '[]'::jsonb) @@ -46,12 +59,21 @@ _LIVE_RUNTIME_SQL = """ FROM automation_operation_log WHERE operation_type = 'ansible_candidate_matched' AND input -> 'executor_candidates' @> CAST(:catalog_match AS jsonb) + AND input ->> 'project_id' = :project_id + AND input ->> 'work_item_id' = :work_item_id + AND input ->> 'proposal_source' = :proposal_source + AND input ->> 'run_namespace' = :run_namespace ORDER BY created_at DESC LIMIT 1 ), latest_check AS ( SELECT check_mode.op_id::text AS check_op_id, check_mode.status AS check_status, + check_mode.input ->> 'automation_run_id' AS check_automation_run_id, + check_mode.input ->> 'trace_id' AS check_trace_id, + check_mode.input ->> 'run_id' AS check_run_id, + check_mode.input ->> 'work_item_id' AS check_work_item_id, + check_mode.input -> 'source_recurrence' AS check_source_recurrence, coalesce( check_mode.output ->> 'returncode', check_mode.dry_run_result ->> 'returncode' @@ -78,6 +100,10 @@ _LIVE_RUNTIME_SQL = """ ) AS apply_returncode, apply.duration_ms AS apply_duration_ms, apply.input ->> 'automation_run_id' AS apply_automation_run_id, + apply.input ->> 'trace_id' AS apply_trace_id, + apply.input ->> 'run_id' AS apply_run_id, + apply.input ->> 'work_item_id' AS apply_work_item_id, + apply.input -> 'source_recurrence' AS apply_source_recurrence, apply.input -> 'runtime_stage_receipts' AS runtime_stage_receipts, apply.created_at AS apply_created_at FROM automation_operation_log apply @@ -87,14 +113,29 @@ _LIVE_RUNTIME_SQL = """ ORDER BY apply.created_at DESC LIMIT 1 ), runtime_stages AS ( - SELECT coalesce( - array_agg(DISTINCT receipt.value ->> 'stage_id') FILTER ( + SELECT + coalesce( + array_agg(DISTINCT receipt.value ->> 'stage_id') FILTER ( + WHERE receipt.value ->> 'stage_id' IS NOT NULL + AND receipt.value ->> 'schema_version' = 'ai_automation_stage_receipt_v1' + AND coalesce((receipt.value ->> 'durable_receipt')::boolean, false) + ), + ARRAY[]::text[] + ) AS stage_ids, + coalesce(bool_and( + receipt.value ->> 'automation_run_id' + = apply.apply_automation_run_id + AND receipt.value ->> 'trace_id' = apply.apply_trace_id + AND receipt.value ->> 'run_id' = apply.apply_run_id + AND receipt.value ->> 'work_item_id' = apply.apply_work_item_id + AND receipt.value -> 'source_recurrence' + = apply.apply_source_recurrence + ) FILTER ( WHERE receipt.value ->> 'stage_id' IS NOT NULL AND receipt.value ->> 'schema_version' = 'ai_automation_stage_receipt_v1' AND coalesce((receipt.value ->> 'durable_receipt')::boolean, false) ), - ARRAY[]::text[] - ) AS stage_ids + false) AS runtime_stage_identity_verified FROM latest_apply apply LEFT JOIN LATERAL jsonb_array_elements( coalesce(apply.runtime_stage_receipts, '[]'::jsonb) @@ -104,6 +145,11 @@ _LIVE_RUNTIME_SQL = """ evidence.id AS verifier_id, coalesce(evidence.verification_result, 'missing') AS verification_result, evidence.post_execution_state ->> 'automation_run_id' AS verifier_run_id, + evidence.post_execution_state ->> 'trace_id' AS verifier_trace_id, + evidence.post_execution_state ->> 'run_id' AS verifier_identity_run_id, + evidence.post_execution_state ->> 'work_item_id' AS verifier_work_item_id, + evidence.post_execution_state -> 'source_recurrence' + AS verifier_source_recurrence, evidence.collected_at AS verifier_created_at FROM incident_evidence evidence JOIN latest_apply apply @@ -177,6 +223,7 @@ _LIVE_RUNTIME_SQL = """ check_mode.*, apply.*, runtime_stages.stage_ids, + runtime_stages.runtime_stage_identity_verified, verifier.*, km.*, learning.*, @@ -203,6 +250,7 @@ async def load_latest_iwooos_wazuh_controlled_executor_runtime( async with get_db_context(project_id) as db: row = None for catalog_id in (INGRESS_CATALOG_ID, CATALOG_ID): + ingress_mode = catalog_id == INGRESS_CATALOG_ID result = await db.execute( text(_LIVE_RUNTIME_SQL), { @@ -210,6 +258,19 @@ async def load_latest_iwooos_wazuh_controlled_executor_runtime( [{"catalog_id": catalog_id}] ), "project_id": project_id, + "work_item_id": ( + INGRESS_WORK_ITEM_ID if ingress_mode else WORK_ITEM_ID + ), + "proposal_source": ( + INGRESS_PROPOSAL_SOURCE + if ingress_mode + else POSTURE_PROPOSAL_SOURCE + ), + "run_namespace": ( + INGRESS_RUN_NAMESPACE + if ingress_mode + else POSTURE_RUN_NAMESPACE + ), }, ) row = result.mappings().first() @@ -239,8 +300,98 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( ) selected_catalog_id = str(data.get("catalog_id") or CATALOG_ID) ingress_mode = selected_catalog_id == INGRESS_CATALOG_ID + expected_work_item_id = INGRESS_WORK_ITEM_ID if ingress_mode else WORK_ITEM_ID + expected_proposal_source = ( + INGRESS_PROPOSAL_SOURCE if ingress_mode else POSTURE_PROPOSAL_SOURCE + ) + expected_run_namespace = ( + INGRESS_RUN_NAMESPACE if ingress_mode else POSTURE_RUN_NAMESPACE + ) catalog_ready = get_ansible_catalog_item(selected_catalog_id) is not None candidate_present = bool(data.get("candidate_op_id")) + candidate_op_id = str(data.get("candidate_op_id") or "") + candidate_source_recurrence = data.get("candidate_source_recurrence") + if not isinstance(candidate_source_recurrence, Mapping): + candidate_source_recurrence = {} + source_recurrence_verified = bool( + data.get("candidate_source_recurrence_verified") is True + and candidate_source_recurrence.get("verified") is True + and all( + str(candidate_source_recurrence.get(key) or "").strip() + for key in ( + "fingerprint", + "occurrence_id", + "observed_at", + "identity_anchor", + "canonical_asset_id", + "source", + ) + ) + and candidate_source_recurrence.get("canonical_asset_id") + == "service:wazuh-manager" + ) + candidate_identity_verified = bool( + candidate_present + and str(data.get("automation_run_id") or "") == candidate_op_id + and str(data.get("trace_id") or "") == candidate_op_id + and str(data.get("run_id") or "") == candidate_op_id + and str(data.get("candidate_project_id") or "").strip() + and str(data.get("work_item_id") or "") == expected_work_item_id + and str(data.get("proposal_source") or "") == expected_proposal_source + and str(data.get("run_namespace") or "") == expected_run_namespace + and str(data.get("source_receipt_ref") or "").strip() + and source_recurrence_verified + ) + check_identity_verified = bool( + data.get("check_op_id") + and str(data.get("check_automation_run_id") or "") == candidate_op_id + and str(data.get("check_trace_id") or "") == candidate_op_id + and str(data.get("check_run_id") or "") == candidate_op_id + and str(data.get("check_work_item_id") or "") == expected_work_item_id + and data.get("check_source_recurrence") == candidate_source_recurrence + ) + apply_identity_verified = bool( + data.get("apply_op_id") + and str(data.get("apply_automation_run_id") or "") == candidate_op_id + and str(data.get("apply_trace_id") or "") == candidate_op_id + and str(data.get("apply_run_id") or "") == candidate_op_id + and str(data.get("apply_work_item_id") or "") == expected_work_item_id + and data.get("apply_source_recurrence") == candidate_source_recurrence + ) + verifier_identity_verified = bool( + data.get("verifier_id") + and str(data.get("verifier_run_id") or "") == candidate_op_id + and str(data.get("verifier_trace_id") or "") == candidate_op_id + and str(data.get("verifier_identity_run_id") or "") == candidate_op_id + and str(data.get("verifier_work_item_id") or "") == expected_work_item_id + and data.get("verifier_source_recurrence") == candidate_source_recurrence + ) + runtime_stage_identity_verified = bool( + data.get("runtime_stage_identity_verified") is True + ) + identity_chain_verified = all( + ( + candidate_identity_verified, + check_identity_verified, + apply_identity_verified, + verifier_identity_verified, + runtime_stage_identity_verified, + ) + ) + identity_chain_mismatch = bool( + candidate_present + and ( + not candidate_identity_verified + or (data.get("check_op_id") and not check_identity_verified) + or (data.get("apply_op_id") and not apply_identity_verified) + or (data.get("verifier_id") and not verifier_identity_verified) + or ( + data.get("apply_op_id") + and data.get("stage_ids") + and not runtime_stage_identity_verified + ) + ) + ) check_passed = ( data.get("check_status") == "success" and str(data.get("check_returncode") or "") == "0" @@ -252,12 +403,19 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( data.get("apply_returncode") or "" ) == "0" apply_failed = apply_terminal and not apply_passed - verifier_passed = data.get("verification_result") == "success" and str( - data.get("verifier_run_id") or "" - ) == str(data.get("candidate_op_id") or "") + verifier_passed = bool( + data.get("verification_result") == "success" + and candidate_identity_verified + and check_identity_verified + and apply_identity_verified + and verifier_identity_verified + ) auto_repair_ready = data.get("auto_repair_success") is True km_ready = bool(data.get("km_entry_id")) - stage_ids = {str(stage_id) for stage_id in data.get("stage_ids") or [] if stage_id} + raw_stage_ids = { + str(stage_id) for stage_id in data.get("stage_ids") or [] if stage_id + } + stage_ids = raw_stage_ids if runtime_stage_identity_verified else set() rag_ready = "rag_writeback" in stage_ids trust_ready = bool(data.get("playbook_trust_acknowledged")) and ( "playbook_trust" in stage_ids @@ -270,14 +428,15 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( auto_repair_ready, km_ready, telegram_ready, + identity_chain_verified, ) ) present_stages = set(stage_ids) for stage_id, present in ( - ("candidate", candidate_present), - ("check_mode", check_passed), - ("controlled_apply", apply_passed), + ("candidate", candidate_identity_verified), + ("check_mode", check_passed and check_identity_verified), + ("controlled_apply", apply_passed and apply_identity_verified), ("auto_repair_execution_receipt", auto_repair_ready), ("post_apply_verifier", verifier_passed), ("km_playbook_writeback", km_ready), @@ -294,7 +453,9 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( missing_stage_ids = [ stage_id for stage_id in required_stages if stage_id not in present_stages ] - runtime_closed = bool(apply_passed and not missing_stage_ids) + runtime_closed = bool( + apply_passed and identity_chain_verified and not missing_stage_ids + ) required_count = len(required_stages) present_count = len(eligible_present_stages) completion_percent = round((present_count / required_count) * 100) @@ -308,6 +469,10 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( active_blockers.append("wazuh_controlled_executor_disabled") if not catalog_ready: active_blockers.append("wazuh_controlled_catalog_missing") + if candidate_present and not source_recurrence_verified: + active_blockers.append("current_source_recurrence_not_verified") + if identity_chain_mismatch: + active_blockers.append("same_run_identity_chain_not_verified") if candidate_present and data.get("check_status") == "failed": active_blockers.append("wazuh_controlled_check_mode_failed") if check_failure_class: @@ -344,6 +509,8 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( data.get("automation_run_id") or data.get("candidate_op_id") or "" ), "same_run_correlation_required": True, + "identity_chain_verified": identity_chain_verified, + "source_recurrence_verified": source_recurrence_verified, }, "summary": { "selected_catalog_id": selected_catalog_id, @@ -369,6 +536,12 @@ 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, + "same_run_identity_chain_verified_count": ( + 1 if identity_chain_verified else 0 + ), + "source_recurrence_verified_count": ( + 1 if source_recurrence_verified else 0 + ), "same_run_required_stage_count": required_count, "same_run_present_stage_count": present_count, "same_run_missing_stage_count": len(missing_stage_ids), @@ -398,6 +571,7 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( apply_terminal=apply_terminal, apply_passed=apply_passed, verifier_passed=verifier_passed, + identity_chain_mismatch=identity_chain_mismatch, runtime_closed=runtime_closed, missing_stage_ids=missing_stage_ids, ), @@ -469,6 +643,7 @@ def _next_action( apply_terminal: bool, apply_passed: bool, verifier_passed: bool, + identity_chain_mismatch: bool, runtime_closed: bool, missing_stage_ids: list[str], ) -> str: @@ -478,6 +653,8 @@ def _next_action( if ingress_mode else "keep_scheduled_wazuh_manager_posture_runtime_receipts_fresh" ) + if identity_chain_mismatch: + return "repair_same_run_identity_chain_before_closure" if not candidate_present: return ( "internal_worker_enqueue_wazuh_alert_ingress_candidate" 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 ef49b528a..d5b2ad40a 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -24,25 +24,61 @@ from src.services.iwooos_wazuh_controlled_executor_runtime import ( ) +def _current_wazuh_source_recurrence(observed_at: str) -> dict[str, object]: + return { + "schema_version": "alert_source_recurrence_receipt_v1", + "verified": True, + "fingerprint": "wazuh-posture-source-fingerprint", + "occurrence_id": "wazuh-posture-source-occurrence", + "observed_at": observed_at, + "identity_anchor": "awooop_conversation_event:wazuh-posture", + "canonical_asset_id": "service:wazuh-manager", + "source": "awooop_conversation_event", + } + + def _complete_row() -> dict[str, object]: run_id = "00000000-0000-0000-0000-000000000301" + recurrence = _current_wazuh_source_recurrence( + "2026-07-15T05:59:58+08:00" + ) return { "candidate_op_id": run_id, "candidate_status": "dry_run", "automation_run_id": run_id, "trace_id": run_id, "run_id": run_id, + "candidate_project_id": "awoooi", "work_item_id": "P0-03-WAZUH-MANAGER-POSTURE", + "proposal_source": "iwooos_wazuh_manager_posture_scheduler", + "run_namespace": "awoooi:iwooos:wazuh-manager-posture", + "source_receipt_ref": "scheduled-wazuh-manager-posture:2026071500", + "candidate_source_recurrence": recurrence, + "candidate_source_recurrence_verified": True, "asset_scope_alias_count": 1, "check_op_id": "00000000-0000-0000-0000-000000000302", "check_status": "success", "check_returncode": "0", + "check_automation_run_id": run_id, + "check_trace_id": run_id, + "check_run_id": run_id, + "check_work_item_id": "P0-03-WAZUH-MANAGER-POSTURE", + "check_source_recurrence": recurrence, "apply_op_id": "00000000-0000-0000-0000-000000000303", "apply_status": "success", "apply_returncode": "0", + "apply_automation_run_id": run_id, + "apply_trace_id": run_id, + "apply_run_id": run_id, + "apply_work_item_id": "P0-03-WAZUH-MANAGER-POSTURE", + "apply_source_recurrence": recurrence, "verifier_id": "verifier-303", "verification_result": "success", "verifier_run_id": run_id, + "verifier_trace_id": run_id, + "verifier_identity_run_id": run_id, + "verifier_work_item_id": "P0-03-WAZUH-MANAGER-POSTURE", + "verifier_source_recurrence": recurrence, "km_entry_id": "km-303", "km_status": "review", "playbook_trust_acknowledged": True, @@ -62,6 +98,7 @@ def _complete_row() -> dict[str, object]: "playbook_trust", "timeline_projection", ], + "runtime_stage_identity_verified": True, } @@ -96,6 +133,11 @@ def test_wazuh_runtime_readback_prefers_ingress_convergence_semantics() -> None: row = _complete_row() row["catalog_id"] = "ansible:wazuh-alertmanager-integration" row["work_item_id"] = "P0-03-WAZUH-ALERT-INGRESS" + row["proposal_source"] = "iwooos_wazuh_alert_ingress_scheduler" + row["run_namespace"] = "awoooi:iwooos:wazuh-alert-ingress" + row["check_work_item_id"] = "P0-03-WAZUH-ALERT-INGRESS" + row["apply_work_item_id"] = "P0-03-WAZUH-ALERT-INGRESS" + row["verifier_work_item_id"] = "P0-03-WAZUH-ALERT-INGRESS" payload = build_iwooos_wazuh_controlled_executor_runtime_readback( row, @@ -140,6 +182,23 @@ def test_wazuh_runtime_readback_keeps_partial_execution_open() -> None: assert "same_run_mcp_context_missing" in payload["active_blockers"] +def test_wazuh_runtime_readback_rejects_mismatched_identity_chain() -> None: + row = _complete_row() + row["verifier_work_item_id"] = "other-lane" + + payload = build_iwooos_wazuh_controlled_executor_runtime_readback( + row, + executor_enabled=True, + ) + + assert payload["summary"]["runtime_closed_count"] == 0 + assert payload["trace"]["identity_chain_verified"] is False + assert "same_run_identity_chain_not_verified" in payload["active_blockers"] + assert payload["next_safe_action"] == ( + "repair_same_run_identity_chain_before_closure" + ) + + def test_wazuh_runtime_readback_separates_probe_success_from_writeback_failure() -> None: row = _complete_row() row["apply_status"] = "failed" @@ -172,10 +231,26 @@ def test_wazuh_runtime_readback_classifies_failed_check_for_bounded_retry() -> N "automation_run_id": "00000000-0000-0000-0000-000000000311", "trace_id": "00000000-0000-0000-0000-000000000311", "run_id": "00000000-0000-0000-0000-000000000311", + "candidate_project_id": "awoooi", "work_item_id": "P0-03-WAZUH-MANAGER-POSTURE", + "proposal_source": "iwooos_wazuh_manager_posture_scheduler", + "run_namespace": "awoooi:iwooos:wazuh-manager-posture", + "source_receipt_ref": "scheduled-wazuh-manager-posture:2026071500", + "candidate_source_recurrence": _current_wazuh_source_recurrence( + "2026-07-15T05:59:58+08:00" + ), + "candidate_source_recurrence_verified": True, + "check_op_id": "00000000-0000-0000-0000-000000000312", "check_status": "failed", "check_returncode": "4", "check_timed_out": "false", + "check_automation_run_id": "00000000-0000-0000-0000-000000000311", + "check_trace_id": "00000000-0000-0000-0000-000000000311", + "check_run_id": "00000000-0000-0000-0000-000000000311", + "check_work_item_id": "P0-03-WAZUH-MANAGER-POSTURE", + "check_source_recurrence": _current_wazuh_source_recurrence( + "2026-07-15T05:59:58+08:00" + ), "stage_ids": [], } @@ -327,6 +402,9 @@ async def test_wazuh_posture_scheduler_enqueues_after_mcp_and_log_evidence( recorder=recorder, evidence_collector=collector, evidence_verifier=verifier, + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), ) assert result["status"] == "queued_for_controlled_executor" @@ -405,6 +483,9 @@ async def test_wazuh_posture_scheduler_retries_failed_check_once_per_deploy( recorder=recorder, evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-2")), evidence_verifier=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), ) assert result["status"] == "queued_for_controlled_executor_retry" @@ -481,6 +562,9 @@ async def test_wazuh_posture_scheduler_queues_one_context_retry_per_deploy( recorder=recorder, evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-3")), evidence_verifier=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), ) assert result["status"] == "queued_for_controlled_executor_context_retry" @@ -502,6 +586,9 @@ async def test_wazuh_posture_scheduler_queues_one_context_retry_per_deploy( duplicate = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( recorder=recorder, evidence_collector=duplicate_collector, + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), ) assert duplicate["status"] == "context_retry_already_attempted_for_deploy" @@ -576,6 +663,9 @@ async def test_wazuh_posture_scheduler_retries_incomplete_learning_once_per_depl recorder=recorder, evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-4")), evidence_verifier=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), ) assert result["status"] == "queued_for_controlled_executor_retry" @@ -598,6 +688,9 @@ async def test_wazuh_posture_scheduler_retries_incomplete_learning_once_per_depl duplicate = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( recorder=recorder, evidence_collector=duplicate_collector, + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-11T05:59:58+08:00" + ), ) assert duplicate["status"] == ( @@ -656,6 +749,9 @@ async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate( result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( recorder=recorder, + source_recurrence=_current_wazuh_source_recurrence( + datetime.now(UTC).isoformat() + ), ) assert result["status"] == "fresh_candidate_already_present" @@ -708,6 +804,9 @@ async def test_wazuh_posture_scheduler_does_not_retry_pending_candidate_without_ result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( recorder=recorder, evidence_collector=collector, + source_recurrence=_current_wazuh_source_recurrence( + datetime.now(UTC).isoformat() + ), ) assert result["status"] == ( @@ -774,6 +873,9 @@ async def test_wazuh_posture_scheduler_limits_failed_retry_to_once_per_deploy( result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once( recorder=recorder, evidence_collector=collector, + source_recurrence=_current_wazuh_source_recurrence( + datetime.now(UTC).isoformat() + ), ) assert result["status"] == ( @@ -827,6 +929,9 @@ async def test_wazuh_posture_scheduler_fails_closed_with_degraded_context( recorder=recorder, evidence_collector=AsyncMock(return_value=None), evidence_verifier=AsyncMock(return_value=False), + source_recurrence=_current_wazuh_source_recurrence( + datetime.now(UTC).isoformat() + ), ) assert result["status"] == "blocked_predecision_evidence_not_verified" diff --git a/apps/api/tests/test_wazuh_alertmanager_integration.py b/apps/api/tests/test_wazuh_alertmanager_integration.py index 07b7cdf00..373d61276 100644 --- a/apps/api/tests/test_wazuh_alertmanager_integration.py +++ b/apps/api/tests/test_wazuh_alertmanager_integration.py @@ -17,11 +17,14 @@ os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/ from src.services.awooop_ansible_audit_service import ( _catalog_hints, + build_ansible_decision_audit_payload, get_ansible_catalog_item, + preflight_ansible_candidate_generation, ) from src.services.awooop_ansible_post_verifier import postconditions_for_catalog from src.services.awooop_ansible_check_mode_service import ( build_ansible_apply_command, + build_ansible_check_mode_claim_input, build_ansible_check_mode_command, ) from src.core.config import settings @@ -52,6 +55,21 @@ WORKER_DEPLOYMENT = ROOT / "k8s" / "awoooi-prod" / "08-deployment-worker.yaml" CATALOG_ID = "ansible:wazuh-alertmanager-integration" +def _current_wazuh_source_recurrence( + observed_at: str | None = None, +) -> dict[str, object]: + return { + "schema_version": "alert_source_recurrence_receipt_v1", + "verified": True, + "fingerprint": "wazuh-source-fingerprint-5710", + "occurrence_id": "wazuh-source-occurrence-5710", + "observed_at": observed_at or datetime.now(UTC).isoformat(), + "identity_anchor": "awooop_conversation_event:wazuh-5710", + "canonical_asset_id": "service:wazuh-manager", + "source": "awooop_conversation_event", + } + + def _load_bridge_module(): spec = importlib.util.spec_from_file_location("wazuh_awoooi_bridge", SCRIPT) assert spec is not None and spec.loader is not None @@ -284,10 +302,37 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat return _Mappings() class _Db: - async def execute(self, _statement, params): + async def execute(self, statement, params): assert json.loads(params["catalog_match"]) == [ {"catalog_id": CATALOG_ID} ] + assert params["project_id"] == "awoooi" + assert params["work_item_id"] == "P0-03-WAZUH-ALERT-INGRESS" + assert params["proposal_source"] == ( + "iwooos_wazuh_alert_ingress_scheduler" + ) + assert params["run_namespace"] == ( + "awoooi:iwooos:wazuh-alert-ingress" + ) + assert params["source_fingerprint"] == ( + "wazuh-source-fingerprint-5710" + ) + assert params["source_occurrence_id"] == ( + "wazuh-source-occurrence-5710" + ) + query = str(statement) + assert "candidate.input ->> 'project_id' = :project_id" in query + assert "candidate.input ->> 'work_item_id' = :work_item_id" in query + assert "candidate.input ->> 'proposal_source' = :proposal_source" in query + assert "candidate.input ->> 'run_namespace' = :run_namespace" in query + assert ( + "candidate.input #>> '{source_recurrence,fingerprint}' = :source_fingerprint" + in query + ) + assert ( + "candidate.input #>> '{source_recurrence,occurrence_id}' = :source_occurrence_id" + in query + ) return _Result() @asynccontextmanager @@ -318,6 +363,9 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-ingress")), evidence_verifier=AsyncMock(return_value=True), incident_binder=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-15T05:59:58+08:00" + ), ) assert result["status"] == "queued_for_controlled_executor" @@ -333,6 +381,49 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat assert recorded["incident"]["source_receipt_ref"] == ( "scheduled-wazuh-alert-ingress:2026071500" ) + candidate = build_ansible_decision_audit_payload( + incident=recorded["incident"], + proposal_data=recorded["proposal_data"], + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + ) + assert candidate is not None + candidate["input"]["automation_run_id"] = result["automation_run_id"] + claim = build_ansible_check_mode_claim_input( + source_candidate_op_id=result["automation_run_id"], + candidate_input=candidate["input"], + observed_now=datetime.fromisoformat("2026-07-15T05:59:59+08:00"), + ) + assert claim["project_id"] == "awoooi" + assert claim["trace_id"] == claim["run_id"] == result["automation_run_id"] + assert claim["work_item_id"] == "P0-03-WAZUH-ALERT-INGRESS" + assert claim["run_namespace"] == "awoooi:iwooos:wazuh-alert-ingress" + assert claim["source_recurrence"]["verified"] is True + mismatched_candidate = {**candidate["input"], "work_item_id": "other-lane"} + with pytest.raises( + ValueError, + match="scheduled_candidate_identity_chain_not_verified", + ): + build_ansible_check_mode_claim_input( + source_candidate_op_id=result["automation_run_id"], + candidate_input=mismatched_candidate, + observed_now=datetime.fromisoformat("2026-07-15T05:59:59+08:00"), + ) + stale_candidate = { + **candidate["input"], + "source_recurrence": _current_wazuh_source_recurrence( + "2026-07-15T05:30:00+08:00" + ), + } + with pytest.raises( + ValueError, + match="scheduled_candidate_identity_chain_not_verified", + ): + build_ansible_check_mode_claim_input( + source_candidate_op_id=result["automation_run_id"], + candidate_input=stale_candidate, + observed_now=datetime.fromisoformat("2026-07-15T05:59:59+08:00"), + ) write_not_acknowledged = ( await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( @@ -342,6 +433,9 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat ), evidence_verifier=AsyncMock(return_value=True), incident_binder=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence( + "2026-07-15T05:59:58+08:00" + ), ) ) assert write_not_acknowledged["status"] == "candidate_write_not_acknowledged" @@ -351,6 +445,96 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat ] +@pytest.mark.asyncio +async def test_wazuh_ingress_scheduler_fails_closed_without_current_recurrence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + recorder = AsyncMock(return_value=True) + collector = AsyncMock(return_value=MagicMock(snapshot_id="must-not-run")) + verifier = AsyncMock(return_value=True) + binder = AsyncMock(return_value=True) + monkeypatch.setattr( + job.settings, + "ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR", + True, + ) + + result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( + recorder=recorder, + evidence_collector=collector, + evidence_verifier=verifier, + incident_binder=binder, + ) + + assert result == { + "status": "blocked_current_source_recurrence_not_verified", + "queued": 0, + "existing": 0, + "evidence_ready": 0, + "work_item_id": "P0-03-WAZUH-ALERT-INGRESS", + "drift_work_item_id": "DRIFT-WAZUH-SOURCE-RECURRENCE", + "controlled_apply_allowed": False, + "active_blockers": ["current_source_recurrence_not_verified"], + } + recorder.assert_not_awaited() + collector.assert_not_awaited() + verifier.assert_not_awaited() + binder.assert_not_awaited() + + unrelated_recurrence = { + **_current_wazuh_source_recurrence(), + "canonical_asset_id": "service:sentry", + } + unrelated = ( + await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( + recorder=recorder, + evidence_collector=collector, + evidence_verifier=verifier, + incident_binder=binder, + source_recurrence=unrelated_recurrence, + ) + ) + assert unrelated["status"] == ( + "blocked_current_source_recurrence_not_verified" + ) + assert unrelated["controlled_apply_allowed"] is False + + +@pytest.mark.asyncio +async def test_wazuh_scheduler_audit_preflight_requires_source_recurrence() -> None: + run_id = "00000000-0000-0000-0000-000000000571" + incident = { + "incident_id": "IWZ-INGRESS-2026071500", + "project_id": "awoooi", + "severity": "high", + "alertname": "WazuhAlertmanagerIntegrationConvergence", + "alert_category": "security", + "affected_services": ["wazuh-manager", "siem"], + "trace_id": run_id, + "run_id": run_id, + "work_item_id": "P0-03-WAZUH-ALERT-INGRESS", + "run_namespace": "awoooi:iwooos:wazuh-alert-ingress", + "signals": [{"labels": {"service": "wazuh-manager"}}], + } + + result = await preflight_ansible_candidate_generation( + incident=incident, + proposal_data={ + "source": "iwooos_wazuh_alert_ingress_scheduler", + "risk_level": "high", + "action": "converge_wazuh_alertmanager_integration", + }, + decision_path="repair_candidate_controlled_queue", + not_used_reason="test", + project_id="awoooi", + ) + + assert result["status"] == "current_source_recurrence_not_verified" + assert result["should_collect_evidence"] is False + + @pytest.mark.asyncio async def test_wazuh_ingress_scheduler_reports_unresolved_payload_before_recording( monkeypatch: pytest.MonkeyPatch, @@ -391,6 +575,7 @@ async def test_wazuh_ingress_scheduler_reports_unresolved_payload_before_recordi evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-ingress")), evidence_verifier=AsyncMock(return_value=True), incident_binder=AsyncMock(return_value=True), + source_recurrence=_current_wazuh_source_recurrence(), ) assert result["status"] == "blocked_canonical_asset_or_ansible_catalog_unresolved" @@ -437,6 +622,7 @@ async def test_wazuh_ingress_scheduler_blocks_orphan_candidate( evidence_collector=collector, evidence_verifier=AsyncMock(return_value=True), incident_binder=AsyncMock(return_value=False), + source_recurrence=_current_wazuh_source_recurrence(), ) assert result["status"] == "blocked_scheduled_incident_write_not_acknowledged"