fix(agent99): stop Host112 verifier churn
This commit is contained in:
ogt
@@ -205,6 +205,7 @@ def test_agent99_99_config_has_bounded_transport_and_recovery_defaults() -> None
|
||||
assert host111["enabled"] is True
|
||||
assert host111["verifyTcpPorts"] == [22]
|
||||
assert host111["verifyTimeoutSeconds"] <= 300
|
||||
assert host111["failureBackoffMinutes"] == 30
|
||||
host112 = next(item for item in config["vms"] if item["host"] == "192.168.0.112")
|
||||
assert host112["vmx"] == r"S:\VMs\host112\kali-linux-2025.4-vmware-amd64.vmx"
|
||||
|
||||
@@ -220,6 +221,13 @@ def test_agent99_host111_recovery_is_allowlisted_and_independently_verified() ->
|
||||
assert "Invoke-AgentExternalHostRecovery" in function
|
||||
assert "Test-AgentExternalHostReadiness" in function
|
||||
assert 'terminal = if ($after.ready) { "verified_ready" }' in function
|
||||
assert 'terminal = "failure_backoff_active"' in function
|
||||
assert '"recovered_during_backoff_verified"' in function
|
||||
assert "recoveryObserved" in function
|
||||
assert "if ($result.changed -or $result.recoveryObserved)" in function
|
||||
assert "external-host-recovery-$safeName.failure.marker" in function
|
||||
assert "failureBackoffMinutes" in function
|
||||
assert "failureBackoffActive" in function
|
||||
assert 'verifier = "icmp_and_required_tcp_ports"' in function
|
||||
for prohibited in (
|
||||
"host_reboot",
|
||||
|
||||
317
apps/api/tests/test_ansible_incident_ledger_churn_guard.py
Normal file
317
apps/api/tests/test_ansible_incident_ledger_churn_guard.py
Normal file
@@ -0,0 +1,317 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault(
|
||||
"DATABASE_URL",
|
||||
"postgresql+asyncpg://test:test@localhost/test",
|
||||
)
|
||||
|
||||
from src.jobs.awooop_ansible_candidate_backfill_job import ( # noqa: E402
|
||||
_wazuh_posture_incident,
|
||||
)
|
||||
from src.services import awooop_ansible_audit_service as audit # noqa: E402
|
||||
from src.services import awooop_ansible_check_mode_service as service # noqa: E402
|
||||
|
||||
|
||||
class _MappingResult:
|
||||
def __init__(self, row: dict | None = None) -> None:
|
||||
self._row = row
|
||||
|
||||
def mappings(self) -> _MappingResult:
|
||||
return self
|
||||
|
||||
def one_or_none(self) -> dict | None:
|
||||
return self._row
|
||||
|
||||
|
||||
class _RowsResult:
|
||||
def __init__(self, rows: list[dict]) -> None:
|
||||
self._rows = rows
|
||||
|
||||
def mappings(self) -> _RowsResult:
|
||||
return self
|
||||
|
||||
def all(self) -> list[dict]:
|
||||
return self._rows
|
||||
|
||||
|
||||
class _SequenceDB:
|
||||
def __init__(self, *results) -> None:
|
||||
self._results = list(results)
|
||||
self.statements: list[str] = []
|
||||
self.parameters: list[dict] = []
|
||||
|
||||
async def execute(self, statement, parameters=None):
|
||||
self.statements.append(str(statement))
|
||||
self.parameters.append(dict(parameters or {}))
|
||||
return self._results.pop(0)
|
||||
|
||||
|
||||
def _wazuh_claim() -> service.AnsibleCheckModeClaim:
|
||||
run_id = "00000000-0000-0000-0000-000000000202"
|
||||
return service.AnsibleCheckModeClaim(
|
||||
op_id="00000000-0000-0000-0000-000000000201",
|
||||
source_candidate_op_id="00000000-0000-0000-0000-000000000200",
|
||||
incident_id="IWZ-POSTURE-2026071506",
|
||||
catalog_id="ansible:wazuh-manager-posture-readback",
|
||||
playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml",
|
||||
apply_playbook_path=(
|
||||
"infra/ansible/playbooks/wazuh-manager-posture-readback.yml"
|
||||
),
|
||||
inventory_hosts=("host_112",),
|
||||
risk_level="low",
|
||||
input_payload={
|
||||
"automation_run_id": run_id,
|
||||
"trace_id": run_id,
|
||||
"run_id": run_id,
|
||||
"work_item_id": "P0-03-WAZUH-MANAGER-POSTURE",
|
||||
"source_receipt_ref": "scheduled-wazuh-manager-posture:2026071506",
|
||||
"asset_scope_aliases": ["security_manager_primary"],
|
||||
"canonical_asset_id": "service:wazuh-manager",
|
||||
"typed_domain": "windows_vmware_control_plane",
|
||||
"target_selector": {
|
||||
"affected_services": ["wazuh-manager", "siem"],
|
||||
"canonical_asset_id": "service:wazuh-manager",
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _durable_verifier_receipt(
|
||||
claim: service.AnsibleCheckModeClaim,
|
||||
*,
|
||||
apply_op_id: str,
|
||||
) -> dict:
|
||||
return {
|
||||
"schema_version": service.ANSIBLE_POST_VERIFIER_SCHEMA_VERSION,
|
||||
"verifier": "asset_specific_read_only_host_postconditions",
|
||||
"independent_source": "broker_ssh_host_runtime_readback",
|
||||
"automation_run_id": claim.input_payload["automation_run_id"],
|
||||
"apply_op_id": apply_op_id,
|
||||
"catalog_id": claim.catalog_id,
|
||||
"verification_result": "success",
|
||||
"all_postconditions_passed": True,
|
||||
"required_postcondition_count": 2,
|
||||
"passed_postcondition_count": 2,
|
||||
"postconditions": [
|
||||
{"condition_id": "unit_loaded", "passed": True},
|
||||
{"condition_id": "runtime", "passed": True},
|
||||
],
|
||||
"active_blockers": [],
|
||||
"executor_returncode": 0,
|
||||
"executor_returncode_trusted": False,
|
||||
"raw_output_stored": False,
|
||||
"writes_on_verify": False,
|
||||
}
|
||||
|
||||
|
||||
def test_wazuh_candidate_builds_bounded_canonical_incident_ledger() -> None:
|
||||
run_id = "00000000-0000-0000-0000-000000000202"
|
||||
incident = _wazuh_posture_incident(
|
||||
automation_run_id=run_id,
|
||||
bucket_ref="2026071506",
|
||||
)
|
||||
|
||||
params = audit._canonical_incident_ledger_params(
|
||||
incident,
|
||||
automation_run_id=run_id,
|
||||
)
|
||||
|
||||
assert params is not None
|
||||
assert params["incident_id"] == "IWZ-POSTURE-2026071506"
|
||||
assert params["status"] == "INVESTIGATING"
|
||||
assert params["severity"] == "P3"
|
||||
assert params["notification_type"] is None
|
||||
decision_chain = json.loads(params["decision_chain"])
|
||||
assert decision_chain["trace_id"] == run_id
|
||||
assert decision_chain["work_item_id"] == "P0-03-WAZUH-MANAGER-POSTURE"
|
||||
assert decision_chain["normalized_asset_identity"] is True
|
||||
assert decision_chain["candidate_runtime_apply_executed"] is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_incident_ledger_insert_is_idempotent_and_read_back() -> None:
|
||||
db = _SequenceDB(_MappingResult({"inserted": True, "ready": True}))
|
||||
run_id = "00000000-0000-0000-0000-000000000202"
|
||||
incident = _wazuh_posture_incident(
|
||||
automation_run_id=run_id,
|
||||
bucket_ref="2026071506",
|
||||
)
|
||||
|
||||
result = await audit._ensure_ansible_incident_ledger_with_db(
|
||||
db,
|
||||
incident,
|
||||
automation_run_id=run_id,
|
||||
)
|
||||
|
||||
assert result == {"ready": True, "inserted": True}
|
||||
assert "ON CONFLICT (incident_id) DO NOTHING" in db.statements[0]
|
||||
assert "UPDATE incidents" not in db.statements[0]
|
||||
assert "EXISTS" in db.statements[0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_durable_verifier_receipt_skips_a_second_ssh_probe(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
claim = _wazuh_claim()
|
||||
apply_op_id = "00000000-0000-0000-0000-000000000203"
|
||||
receipt = _durable_verifier_receipt(claim, apply_op_id=apply_op_id)
|
||||
ssh_probe = AsyncMock(side_effect=AssertionError("SSH verifier repeated"))
|
||||
monkeypatch.setattr(service, "run_ansible_asset_post_verifier", ssh_probe)
|
||||
|
||||
resolved, reused = await service._resolve_ansible_post_verifier_receipt(
|
||||
claim,
|
||||
service.AnsibleRunResult(
|
||||
returncode=0,
|
||||
stdout="",
|
||||
stderr="",
|
||||
duration_ms=5,
|
||||
),
|
||||
apply_op_id=apply_op_id,
|
||||
durable_verifier_receipt=receipt,
|
||||
)
|
||||
|
||||
assert reused is True
|
||||
assert resolved == receipt
|
||||
ssh_probe.assert_not_awaited()
|
||||
|
||||
|
||||
def test_backfill_accepts_only_same_run_successful_durable_verifier() -> None:
|
||||
claim = _wazuh_claim()
|
||||
apply_op_id = "00000000-0000-0000-0000-000000000203"
|
||||
receipt = _durable_verifier_receipt(claim, apply_op_id=apply_op_id)
|
||||
|
||||
assert (
|
||||
service._durable_post_verifier_receipt_for_backfill(
|
||||
{"durable_verifier_receipt": receipt},
|
||||
claim,
|
||||
apply_op_id=apply_op_id,
|
||||
)
|
||||
== receipt
|
||||
)
|
||||
|
||||
wrong_run = {**receipt, "automation_run_id": "wrong-run"}
|
||||
assert (
|
||||
service._durable_post_verifier_receipt_for_backfill(
|
||||
{"durable_verifier_receipt": wrong_run},
|
||||
claim,
|
||||
apply_op_id=apply_op_id,
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_backfill_repairs_orphan_incident_and_reuses_verifier(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
claim = _wazuh_claim()
|
||||
apply_op_id = "00000000-0000-0000-0000-000000000203"
|
||||
receipt = _durable_verifier_receipt(claim, apply_op_id=apply_op_id)
|
||||
row = {
|
||||
"op_id": apply_op_id,
|
||||
"parent_op_id": claim.op_id,
|
||||
"incident_id": claim.incident_id,
|
||||
"input": {
|
||||
**claim.input_payload,
|
||||
"catalog_id": claim.catalog_id,
|
||||
"source_candidate_op_id": claim.source_candidate_op_id,
|
||||
"check_mode_op_id": claim.op_id,
|
||||
"playbook_path": claim.playbook_path,
|
||||
"apply_playbook_path": claim.apply_playbook_path,
|
||||
"inventory_hosts": list(claim.inventory_hosts),
|
||||
"risk_level": claim.risk_level,
|
||||
},
|
||||
"output": {"returncode": 0},
|
||||
"dry_run_result": {},
|
||||
"error": None,
|
||||
"duration_ms": 5,
|
||||
"status": "success",
|
||||
"incident_ledger_ready": False,
|
||||
"verifier_ready": True,
|
||||
"durable_verifier_receipt": receipt,
|
||||
}
|
||||
db = _SequenceDB(_MappingResult(), _RowsResult([row]))
|
||||
|
||||
@asynccontextmanager
|
||||
async def fake_db_context(_project_id: str):
|
||||
yield db
|
||||
|
||||
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
||||
monkeypatch.setattr(
|
||||
service,
|
||||
"backfill_missing_retry_terminal_projections_once",
|
||||
AsyncMock(
|
||||
return_value={
|
||||
"scanned": 0,
|
||||
"written": 0,
|
||||
"retry_receipt_written": 0,
|
||||
"telegram_receipt_acknowledged": 0,
|
||||
"incident_receipt_written": 0,
|
||||
"lifecycle_written": 0,
|
||||
"verified": 0,
|
||||
"runtime_apply_executed": False,
|
||||
"error": None,
|
||||
}
|
||||
),
|
||||
)
|
||||
incident_ledger = AsyncMock(return_value={"ready": True, "inserted": True})
|
||||
monkeypatch.setattr(service, "ensure_ansible_incident_ledger", incident_ledger)
|
||||
writeback = AsyncMock(
|
||||
return_value={
|
||||
"verification": True,
|
||||
"verification_passed": True,
|
||||
"verification_result": "success",
|
||||
"learning": True,
|
||||
"trust_learning": True,
|
||||
"rag_writeback": True,
|
||||
"runtime_stage_receipts": [],
|
||||
"durable_verifier_receipt_reused": True,
|
||||
}
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service,
|
||||
"_record_post_apply_verifier_and_learning",
|
||||
writeback,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service,
|
||||
"_record_auto_repair_execution_receipt",
|
||||
AsyncMock(return_value=True),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service,
|
||||
"_record_timeline_projection_receipt",
|
||||
AsyncMock(return_value=None),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service,
|
||||
"_record_runtime_stage_receipts",
|
||||
AsyncMock(return_value=True),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
service,
|
||||
"_reconcile_verified_apply_closure_projections",
|
||||
AsyncMock(
|
||||
return_value={
|
||||
"closed": True,
|
||||
"telegram_receipt_acknowledged": True,
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
result = await service.backfill_missing_auto_repair_execution_receipts_once(limit=1)
|
||||
|
||||
assert result["incident_ledger_reconciled"] == 1
|
||||
assert result["incident_ledger_reconcile_failed"] == 0
|
||||
assert result["durable_verifier_receipt_reused"] == 1
|
||||
assert result["incident_closure_written"] == 1
|
||||
incident_ledger.assert_awaited_once()
|
||||
assert writeback.await_args.kwargs["durable_verifier_receipt"] == receipt
|
||||
@@ -127,6 +127,9 @@ class _GenerationResult:
|
||||
def first(self):
|
||||
return self._row
|
||||
|
||||
def one_or_none(self):
|
||||
return self._row
|
||||
|
||||
def scalar(self):
|
||||
return self._scalar
|
||||
|
||||
@@ -143,6 +146,10 @@ class _GenerationDb:
|
||||
self.parameters.append(parameters or {})
|
||||
if "pg_advisory_xact_lock" in sql:
|
||||
return _GenerationResult()
|
||||
if "WITH inserted AS" in sql and "INSERT INTO incidents" in sql:
|
||||
return _GenerationResult(
|
||||
row={"inserted": False, "ready": True}
|
||||
)
|
||||
if "SELECT" in sql and "candidate.op_id::text AS op_id" in sql:
|
||||
return _GenerationResult(row=self.latest)
|
||||
if "typed_route_reconciler" in sql:
|
||||
|
||||
Reference in New Issue
Block a user