wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(wazuh): require current same-run recurrence

Browse Source
This commit is contained in:
ogt
2026-07-15 16:40:46 +08:00
parent 6dc148ad2f
commit 9c5b85f9bb
6 changed files with 751 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
View File

@@ -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,
)

25
apps/api/src/services/awooop_ansible_audit_service.py
View File

@@ -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(

147
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -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",

201
apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
View File

@@ -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"