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

fix(agent): fail closed on incomplete Wazuh context

Browse Source
This commit is contained in:
ogt
2026-07-11 11:57:38 +08:00
parent f4e85fac38
commit 0fe6bd3ba0
2 changed files with 270 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -359,8 +359,17 @@ def _wazuh_posture_incident(
receipt_ref = (
f"{bucket_ref}-{attempt_ref.upper()}" if attempt_ref else bucket_ref
)
incident_id = f"IWZ-POSTURE-{bucket_ref}"
if attempt_ref:
attempt_token = uuid5(
NAMESPACE_URL,
f"awoooi:iwooos:wazuh-manager-posture-attempt:{attempt_ref}",
).hex[:12].upper()
incident_id = f"IWZ-P-{bucket_ref}-{attempt_token}"
return {
"incident_id": f"IWZ-POSTURE-{receipt_ref}",
# IncidentEvidence.incident_id is VARCHAR(30). Keep retry identities
# deterministic and bounded while preserving the full ref separately.
"incident_id": incident_id,
"project_id": "awoooi",
"status": "INVESTIGATING",
"severity": "low",
@@ -422,7 +431,29 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
candidate.status AS candidate_status,
candidate.input ->> 'automation_run_id' AS automation_run_id,
candidate.input ->> 'source_receipt_ref' AS source_receipt_ref,
check_mode.status AS latest_check_status
check_mode.status AS latest_check_status,
EXISTS (
SELECT 1
FROM incident_evidence evidence
WHERE evidence.incident_id = candidate.input ->> 'incident_id'
AND evidence.collected_at <= candidate.created_at
AND (
evidence.post_execution_state IS NULL
OR CAST(evidence.post_execution_state AS jsonb) = 'null'::jsonb
)
AND evidence.sensors_succeeded > 0
AND NULLIF(btrim(evidence.recent_logs), '') IS NOT NULL
AND EXISTS (
SELECT 1
FROM jsonb_each(
coalesce(
CAST(evidence.mcp_health AS jsonb),
'{}'::jsonb
)
) health
WHERE health.value = 'true'::jsonb
)
) AS predecision_evidence_ready
FROM automation_operation_log candidate
LEFT JOIN LATERAL (
SELECT check_receipt.status
@@ -446,8 +477,20 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
},
)
existing_row = existing.mappings().first()
existing_evidence_ready = bool(
existing_row
and existing_row.get("predecision_evidence_ready") is True
)
latest_check_status = str(
existing_row.get("latest_check_status") or ""
) if existing_row else ""
retry_attempt_ref: str | None = None
if existing_row and existing_row.get("latest_check_status") != "failed":
retry_reason: str | None = None
if (
existing_row
and latest_check_status != "failed"
and existing_evidence_ready
):
return {
"status": "fresh_candidate_already_present",
"queued": 0,
@@ -463,30 +506,63 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
}
if existing_row:
source_sha = os.getenv("AWOOOI_BUILD_COMMIT_SHA", "").strip().lower()
retry_attempt_ref = (
deploy_ref = (
source_sha[:12]
if len(source_sha) >= 12
and all(character in "0123456789abcdef" for character in source_sha)
else "runtime-retry"
)
existing_source_ref = str(
existing_row.get("source_receipt_ref") or ""
).lower()
if existing_source_ref.endswith(f"-{retry_attempt_ref.lower()}"):
if latest_check_status == "failed":
retry_attempt_ref = deploy_ref
retry_reason = "failed_check_mode"
elif latest_check_status == "success":
retry_attempt_ref = f"{deploy_ref}-context"
retry_reason = "predecision_context"
else:
return {
"status": "failed_check_retry_already_attempted_for_deploy",
"status": "fresh_candidate_predecision_evidence_not_verified",
"queued": 0,
"existing": 1,
"evidence_ready": 1,
"evidence_ready": 0,
"automation_run_id": str(
existing_row.get("automation_run_id")
or existing_row.get("op_id")
or ""
),
"work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
"retry_of_failed_check_mode": True,
"active_blockers": [
"predecision_mcp_or_log_evidence_missing"
],
}
existing_source_ref = str(
existing_row.get("source_receipt_ref") or ""
).lower()
if existing_source_ref.endswith(f"-{retry_attempt_ref.lower()}"):
return {
"status": (
"failed_check_retry_already_attempted_for_deploy"
if retry_reason == "failed_check_mode"
else "context_retry_already_attempted_for_deploy"
),
"queued": 0,
"existing": 1,
"evidence_ready": 1 if existing_evidence_ready else 0,
"automation_run_id": str(
existing_row.get("automation_run_id")
or existing_row.get("op_id")
or ""
),
"work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
"retry_of_failed_check_mode": (
retry_reason == "failed_check_mode"
),
"retry_of_incomplete_predecision_context": (
retry_reason == "predecision_context"
),
"active_blockers": [
"bounded_retry_already_attempted_for_deploy"
if retry_reason == "failed_check_mode"
else "bounded_context_retry_already_attempted_for_deploy"
],
}
@@ -522,6 +598,23 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
error_type=type(exc).__name__,
)
if not evidence_ready:
return {
"status": "blocked_predecision_evidence_not_verified",
"queued": 0,
"existing": 1 if existing_row else 0,
"evidence_ready": 0,
"automation_run_id": automation_run_id,
"work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
"retry_of_failed_check_mode": retry_reason == "failed_check_mode",
"retry_of_incomplete_predecision_context": (
retry_reason == "predecision_context"
),
"active_blockers": [
"predecision_mcp_or_log_evidence_missing"
],
}
queued = await recorder(
incident=incident,
proposal_data={
@@ -540,26 +633,23 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
return {
"status": (
"queued_for_controlled_executor_retry"
if queued and evidence_ready and retry_attempt_ref
if queued and retry_reason == "failed_check_mode"
else "queued_for_controlled_executor_context_retry"
if queued and retry_reason == "predecision_context"
else "queued_for_controlled_executor"
if queued and evidence_ready
else (
"queued_for_controlled_executor_retry_with_degraded_context"
if queued and retry_attempt_ref
else "queued_for_controlled_executor_with_degraded_context"
if queued
else "candidate_race_deduplicated"
)
if queued
else "candidate_race_deduplicated"
),
"queued": 1 if queued else 0,
"existing": 0 if queued else 1,
"evidence_ready": 1 if evidence_ready else 0,
"automation_run_id": automation_run_id,
"work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID,
"retry_of_failed_check_mode": bool(retry_attempt_ref),
"active_blockers": (
[] if evidence_ready else ["predecision_mcp_or_log_evidence_missing"]
"retry_of_failed_check_mode": retry_reason == "failed_check_mode",
"retry_of_incomplete_predecision_context": (
retry_reason == "predecision_context"
),
"active_blockers": [] if queued else ["candidate_write_not_acknowledged"],
}
except Exception as exc:
logger.warning(

163
apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py
View File

@@ -338,15 +338,106 @@ async def test_wazuh_posture_scheduler_retries_failed_check_once_per_deploy(
assert result["queued"] == 1
assert result["retry_of_failed_check_mode"] is True
recorded = recorder.await_args.kwargs
assert recorded["incident"]["incident_id"] == (
"IWZ-POSTURE-2026071100-ABCDEF123456"
assert recorded["incident"]["incident_id"].startswith(
"IWZ-P-2026071100-"
)
assert len(recorded["incident"]["incident_id"]) == 29
assert recorded["incident"]["source_receipt_ref"] == (
"scheduled-wazuh-manager-posture:2026071100-ABCDEF123456"
)
assert recorded["automation_run_id"] == result["automation_run_id"]
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_queues_one_context_retry_per_deploy(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
existing_row = {
"op_id": "00000000-0000-0000-0000-000000000322",
"candidate_status": "success",
"automation_run_id": "00000000-0000-0000-0000-000000000322",
"source_receipt_ref": "scheduled-wazuh-manager-posture:2026071100",
"latest_check_status": "success",
"predecision_evidence_ready": False,
}
class _Mappings:
def first(self):
return existing_row
class _Result:
def mappings(self):
return _Mappings()
class _Db:
async def execute(self, _statement, _params):
return _Result()
@asynccontextmanager
async def fake_db_context(_project_id: str):
yield _Db()
recorder = AsyncMock(return_value=True)
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(
job.settings,
"ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
True,
)
monkeypatch.setattr(
job.settings,
"IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS",
6,
)
monkeypatch.setattr(
job,
"now_taipei",
MagicMock(return_value=datetime.fromisoformat("2026-07-11T05:59:59+08:00")),
)
monkeypatch.setenv(
"AWOOOI_BUILD_COMMIT_SHA",
"abcdef1234567890abcdef1234567890abcdef12",
)
result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once(
recorder=recorder,
evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-3")),
evidence_verifier=AsyncMock(return_value=True),
)
assert result["status"] == "queued_for_controlled_executor_context_retry"
assert result["queued"] == 1
assert result["retry_of_failed_check_mode"] is False
assert result["retry_of_incomplete_predecision_context"] is True
recorded = recorder.await_args.kwargs
assert recorded["incident"]["source_receipt_ref"] == (
"scheduled-wazuh-manager-posture:2026071100-ABCDEF123456-CONTEXT"
)
assert recorded["incident"]["incident_id"].startswith(
"IWZ-P-2026071100-"
)
assert len(recorded["incident"]["incident_id"]) == 29
existing_row["automation_run_id"] = result["automation_run_id"]
existing_row["source_receipt_ref"] = recorded["incident"]["source_receipt_ref"]
duplicate_collector = AsyncMock()
duplicate = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once(
recorder=recorder,
evidence_collector=duplicate_collector,
)
assert duplicate["status"] == "context_retry_already_attempted_for_deploy"
assert duplicate["queued"] == 0
assert duplicate["retry_of_incomplete_predecision_context"] is True
assert duplicate["active_blockers"] == [
"bounded_context_retry_already_attempted_for_deploy"
]
recorder.assert_awaited_once()
duplicate_collector.assert_not_awaited()
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate(
monkeypatch: pytest.MonkeyPatch,
@@ -360,6 +451,7 @@ async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate(
"candidate_status": "dry_run",
"automation_run_id": "00000000-0000-0000-0000-000000000331",
"latest_check_status": "pending",
"predecision_evidence_ready": True,
}
class _Result:
@@ -396,6 +488,65 @@ async def test_wazuh_posture_scheduler_does_not_duplicate_active_candidate(
recorder.assert_not_awaited()
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_does_not_retry_pending_candidate_without_context(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
class _Mappings:
def first(self):
return {
"op_id": "00000000-0000-0000-0000-000000000332",
"candidate_status": "dry_run",
"automation_run_id": "00000000-0000-0000-0000-000000000332",
"latest_check_status": "pending",
"predecision_evidence_ready": False,
}
class _Result:
def mappings(self):
return _Mappings()
class _Db:
async def execute(self, _statement, _params):
return _Result()
@asynccontextmanager
async def fake_db_context(_project_id: str):
yield _Db()
recorder = AsyncMock(return_value=True)
collector = AsyncMock()
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(
job.settings,
"ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR",
True,
)
monkeypatch.setattr(
job.settings,
"IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS",
6,
)
result = await job.enqueue_iwooos_wazuh_manager_posture_candidate_once(
recorder=recorder,
evidence_collector=collector,
)
assert result["status"] == (
"fresh_candidate_predecision_evidence_not_verified"
)
assert result["queued"] == 0
assert result["evidence_ready"] == 0
assert result["active_blockers"] == [
"predecision_mcp_or_log_evidence_missing"
]
recorder.assert_not_awaited()
collector.assert_not_awaited()
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_limits_failed_retry_to_once_per_deploy(
monkeypatch: pytest.MonkeyPatch,
@@ -463,7 +614,7 @@ async def test_wazuh_posture_scheduler_limits_failed_retry_to_once_per_deploy(
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_queues_bounded_probe_with_degraded_context(
async def test_wazuh_posture_scheduler_fails_closed_with_degraded_context(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from src.jobs import awooop_ansible_candidate_backfill_job as job
@@ -503,11 +654,11 @@ async def test_wazuh_posture_scheduler_queues_bounded_probe_with_degraded_contex
evidence_verifier=AsyncMock(return_value=False),
)
assert result["status"] == ("queued_for_controlled_executor_with_degraded_context")
assert result["queued"] == 1
assert result["status"] == "blocked_predecision_evidence_not_verified"
assert result["queued"] == 0
assert result["evidence_ready"] == 0
assert result["active_blockers"] == ["predecision_mcp_or_log_evidence_missing"]
recorder.assert_awaited_once()
recorder.assert_not_awaited()
def test_wazuh_runtime_api_is_public_safe(monkeypatch: pytest.MonkeyPatch) -> None: