77d55ddedc
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m25s
CD Pipeline / build-and-deploy (push) Successful in 15m19s
CD Pipeline / post-deploy-checks (push) Successful in 2m12s
384 lines
12 KiB
Python
384 lines
12 KiB
Python
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 _ScalarResult:
|
|
def __init__(self, value) -> None:
|
|
self._value = value
|
|
|
|
def scalar(self):
|
|
return self._value
|
|
|
|
|
|
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_post_apply_verifier_persists_canonical_same_run_identity(
|
|
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)
|
|
verifier_db = _SequenceDB(_ScalarResult("evidence-id"))
|
|
context_calls = 0
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(_project_id: str):
|
|
nonlocal context_calls
|
|
context_calls += 1
|
|
if context_calls == 1:
|
|
yield verifier_db
|
|
return
|
|
raise RuntimeError("KM persistence is outside this focused receipt test")
|
|
|
|
monkeypatch.setattr(service, "get_db_context", fake_db_context)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_resolve_ansible_post_verifier_receipt",
|
|
AsyncMock(return_value=(receipt, True)),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_apply_post_verifier_terminal",
|
|
AsyncMock(return_value=True),
|
|
)
|
|
monkeypatch.setattr(
|
|
service,
|
|
"_record_learning_writeback_receipt",
|
|
AsyncMock(return_value=None),
|
|
)
|
|
|
|
result = await service._record_post_apply_verifier_and_learning(
|
|
claim,
|
|
service.AnsibleRunResult(
|
|
returncode=0,
|
|
stdout="",
|
|
stderr="",
|
|
duration_ms=5,
|
|
),
|
|
apply_op_id=apply_op_id,
|
|
project_id="default",
|
|
durable_verifier_receipt=receipt,
|
|
)
|
|
|
|
post_state = json.loads(verifier_db.parameters[0]["post_execution_state"])
|
|
assert post_state["automation_run_id"] == claim.input_payload["automation_run_id"]
|
|
assert post_state["trace_id"] == claim.input_payload["trace_id"]
|
|
assert post_state["run_id"] == claim.input_payload["run_id"]
|
|
assert result["verification"] is True
|
|
assert result["verification_passed"] is True
|
|
assert result["durable_verifier_receipt_reused"] is True
|
|
|
|
|
|
@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
|