from __future__ import annotations # ruff: noqa: E402, I001 import importlib.util import json import os from contextlib import asynccontextmanager from datetime import UTC, datetime from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest import yaml os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") 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 from src.services.controlled_alert_target_router import resolve_typed_alert_target from src.services.incident_service import classify_alert_early ROOT = Path(__file__).resolve().parents[3] SCRIPT = ( ROOT / "infra" / "ansible" / "files" / "wazuh" / "custom-awoooi-alertmanager.py" ) PLAYBOOK = ( ROOT / "infra" / "ansible" / "playbooks" / "wazuh-alertmanager-integration.yml" ) BROKER_DEPLOYMENT = ( ROOT / "k8s" / "awoooi-prod" / "08-deployment-ansible-executor-broker.yaml" ) WORKER_DEPLOYMENT = ROOT / "k8s" / "awoooi-prod" / "08-deployment-worker.yaml" CD_WORKFLOW = ROOT / ".gitea" / "workflows" / "cd.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 module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def _alert(*, agent_id: str = "017") -> dict: return { "timestamp": "2026-07-15T05:30:00Z", "manager": {"name": "wazuh-manager-internal"}, "agent": { "id": agent_id, "name": "private-endpoint-name", "ip": "192.168.0.55", }, "rule": { "id": "5710", "level": 12, "description": "private security evidence", "groups": ["authentication_failures", "sshd"], "mitre": {"id": ["T1110", "not-a-technique"]}, }, "full_log": "raw source event must remain in Wazuh", "data": {"srcip": "203.0.113.7"}, } def test_wazuh_bridge_emits_only_public_safe_stable_metadata() -> None: bridge = _load_bridge_module() payload = bridge.build_alertmanager_payload( _alert(), now=datetime(2026, 7, 15, 5, 30, tzinfo=UTC), ) encoded = json.dumps(payload, sort_keys=True) labels = payload["alerts"][0]["labels"] assert labels["alertname"] == "WazuhSecurityEvent_5710" assert labels["component"] == "wazuh-manager" assert labels["severity"] == "critical" assert labels["wazuh_rule_groups"] == "authentication_failures,sshd" assert labels["mitre_techniques"] == "T1110" assert labels["deployment"].startswith("wazuh-agent-") assert labels["raw_payload_persisted"] == "false" assert labels["active_response_requested"] == "false" for forbidden in ( "private-endpoint-name", "192.168.0.55", "203.0.113.7", "private security evidence", "raw source event must remain in Wazuh", "wazuh-manager-internal", ): assert forbidden not in encoded def test_wazuh_bridge_fingerprint_clusters_same_rule_and_agent() -> None: bridge = _load_bridge_module() first = bridge.build_alertmanager_payload(_alert(agent_id="017")) second = bridge.build_alertmanager_payload(_alert(agent_id="017")) other_agent = bridge.build_alertmanager_payload(_alert(agent_id="018")) assert first["groupKey"] == second["groupKey"] assert first["groupKey"] != other_agent["groupKey"] assert ( first["alerts"][0]["labels"]["deployment"] != other_agent["alerts"][0]["labels"]["deployment"] ) @pytest.mark.parametrize( "url", [ "http://awoooi.wooo.work/api/v1/webhooks/alertmanager", "https://example.com/api/v1/webhooks/alertmanager", "https://awoooi.wooo.work/api/v1/webhooks/alertmanager?raw=1", "https://user:pass@awoooi.wooo.work/api/v1/webhooks/alertmanager", ], ) def test_wazuh_bridge_rejects_non_allowlisted_hook_urls(url: str) -> None: bridge = _load_bridge_module() with pytest.raises(ValueError, match="hook_url_not_allowlisted"): bridge._validated_hook_url(url) def test_wazuh_integration_catalog_and_rollback_contract_are_bounded() -> None: catalog = get_ansible_catalog_item(CATALOG_ID) playbook = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8")) text = PLAYBOOK.read_text(encoding="utf-8") conditions = postconditions_for_catalog(CATALOG_ID) assert catalog is not None assert catalog["inventory_hosts"] == ["host_112"] assert catalog["risk_level"] == "high" assert catalog["auto_apply_enabled"] is True assert catalog["approval_required"] is False assert catalog["supports_check_mode"] is True assert catalog["catalog_revision"] == "2026-07-15-wazuh-alert-ingress-v2" assert playbook[0]["hosts"] == "host_112" self_test_task = playbook[0]["tasks"][0] assert self_test_task["delegate_to"] == "localhost" assert self_test_task["become"] is False assert self_test_task["ansible.builtin.command"]["argv"][0] == ( "{{ ansible_playbook_python }}" ) secret_reference_assert = next( task for task in playbook[0]["tasks"] if task.get("ansible.builtin.assert", {}).get("fail_msg") == "wazuh_privileged_convergence_secret_reference_missing" ) assert secret_reference_assert.get("no_log") is not True assert "backup: true" in text assert "remote_src: true" in text assert "wazuh_privileged_convergence_capability_missing" in text assert "wazuh_privileged_convergence_secret_reference_missing" in text assert "ANSIBLE_BECOME_PASSWORD_FILE" in text assert "get_checksum: false" in text assert "/var/lib/awoooi-wazuh-alert-ingress/receipt.env" in text assert "raw_payload_persisted=0" in text assert "active_response_enabled=0" in text assert "wazuh_alertmanager_integration_rolled_back" in text assert "slurp:" not in text assert "api_key>" not in text assert "ansible_become_pass" not in text assert "raw.githubusercontent.com" not in text assert {condition.condition_id for condition in conditions} == { "host_112_wazuh_alertmanager_script", "host_112_wazuh_alertmanager_config", "host_112_wazuh_manager_runtime_after_ingress_convergence", "host_112_wazuh_integrator_runtime", "host_112_awoooi_alertmanager_ingress_reachable", } def test_wazuh_become_password_projection_is_path_only_and_optional( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: password_file = tmp_path / "become-password" password_file.write_text("test-only-secret-value\n", encoding="utf-8") monkeypatch.setattr( settings, "AWOOOP_ANSIBLE_BECOME_PASSWORD_FILE", str(password_file), ) spec = build_ansible_check_mode_command( playbook_path="infra/ansible/playbooks/wazuh-alertmanager-integration.yml", inventory_hosts=("host_112",), playbook_root=ROOT / "infra" / "ansible", check_mode_ssh_key_path=tmp_path / "ssh-key", check_mode_known_hosts_path=tmp_path / "known-hosts", ) manifest = yaml.safe_load(BROKER_DEPLOYMENT.read_text(encoding="utf-8")) volume = next( item for item in manifest["spec"]["template"]["spec"]["volumes"] if item["name"] == "host112-become-password" ) assert spec.env["ANSIBLE_BECOME_PASSWORD_FILE"] == str(password_file) assert "test-only-secret-value" not in " ".join(spec.command) assert volume["secret"]["optional"] is True assert volume["secret"]["items"] == [ {"key": "HOST112_ANSIBLE_BECOME_PASSWORD", "path": "password"} ] assert "test-only-secret-value" not in BROKER_DEPLOYMENT.read_text( encoding="utf-8" ) apply_spec = build_ansible_apply_command( playbook_path="infra/ansible/playbooks/wazuh-alertmanager-integration.yml", inventory_hosts=("host_112",), playbook_root=ROOT / "infra" / "ansible", check_mode_ssh_key_path=tmp_path / "ssh-key", check_mode_known_hosts_path=tmp_path / "known-hosts", ) assert apply_spec.env["ANSIBLE_BECOME_PASSWORD_FILE"] == str(password_file) assert "test-only-secret-value" not in " ".join(apply_spec.command) unrelated_spec = build_ansible_check_mode_command( playbook_path="infra/ansible/playbooks/wazuh-manager-posture-readback.yml", inventory_hosts=("host_112",), playbook_root=ROOT / "infra" / "ansible", check_mode_ssh_key_path=tmp_path / "ssh-key", check_mode_known_hosts_path=tmp_path / "known-hosts", ) assert "ANSIBLE_BECOME_PASSWORD_FILE" not in unrelated_spec.env def test_cd_projects_wazuh_privilege_reference_without_readback() -> None: workflow = CD_WORKFLOW.read_text(encoding="utf-8") assert ( "HOST112_ANSIBLE_BECOME_PASSWORD: " "${{ secrets.HOST112_ANSIBLE_BECOME_PASSWORD }}" ) in workflow assert ( 'HOST112_ANSIBLE_BECOME_PASSWORD_B64="$(secret_b64_env ' 'HOST112_ANSIBLE_BECOME_PASSWORD)"' ) in workflow assert '"path":"/data/HOST112_ANSIBLE_BECOME_PASSWORD"' in workflow assert "host112_become_secret_reference_injected=1" in workflow assert "host112_become_secret_projection_ready=1" in workflow assert ( "test -s /run/secrets/host112-become/password && " "test -r /run/secrets/host112-become/password" ) in workflow assert "kubectl get secret awoooi-secrets -o json" not in workflow def test_wazuh_ingress_convergence_uses_only_the_mutating_catalog() -> None: route = resolve_typed_alert_target( alertname="WazuhAlertmanagerIntegrationConvergence", target_resource="wazuh-manager", namespace="awoooi", labels={"service": "wazuh-manager", "tool": "wazuh"}, ) hints = _catalog_hints( { "incident_id": "IWZ-INGRESS-2026071500", "alertname": "WazuhAlertmanagerIntegrationConvergence", "alert_category": "security", "affected_services": ["wazuh-manager", "siem"], "signals": [ { "alert_name": "WazuhAlertmanagerIntegrationConvergence", "labels": { "service": "wazuh-manager", "component": "wazuh-manager", "tool": "wazuh", }, } ], }, None, ) assert route["canonical_asset_id"] == "service:wazuh-manager" assert route["allowed_catalog_ids"] == [CATALOG_ID] assert [item["catalog_id"] for item in hints["candidates"]] == [CATALOG_ID] @pytest.mark.asyncio async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidate( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job class _Mappings: def first(self): return None class _Result: def mappings(self): return _Mappings() class _Db: 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" ) assert params["scheduled_recurrence"] is False 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 async def fake_db_context(project_id: str): assert project_id == "awoooi" 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-15T05:59:59+08:00")), ) result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( recorder=recorder, 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" assert result["work_item_id"] == "P0-03-WAZUH-ALERT-INGRESS" recorded = recorder.await_args.kwargs assert recorded["incident"]["alertname"] == ( "WazuhAlertmanagerIntegrationConvergence" ) assert recorded["incident"]["incident_id"] == "IWZ-INGRESS-2026071500" assert len(recorded["incident"]["incident_id"]) <= 30 assert recorded["proposal_data"]["risk_level"] == "high" assert recorded["proposal_data"]["execution_priority"] == 0 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"), ) durable_scheduled_recurrence = { "schema_version": "alert_source_recurrence_receipt_v1", "verified": True, "fingerprint": "a" * 64, "occurrence_id": "scheduled-wazuh-source:202607150550", "observed_at": "2026-07-15T05:50:00+08:00", "identity_anchor": ( "automation_operation_log:" "00000000-0000-0000-0000-000000000803" ), "canonical_asset_id": "service:wazuh-manager", "source": "iwooos_asset_sensor_receipt", "durable_readback_verified": True, } durable_scheduled_proposal = { **recorded["proposal_data"], "source_fingerprint": durable_scheduled_recurrence["fingerprint"], "source_occurrence_id": durable_scheduled_recurrence["occurrence_id"], "source_received_at": durable_scheduled_recurrence["observed_at"], "source_recurrence_source": durable_scheduled_recurrence["source"], "source_recurrence_anchor": durable_scheduled_recurrence[ "identity_anchor" ], "source_canonical_asset_id": "service:wazuh-manager", "source_recurrence_durable_readback_verified": True, } durable_candidate = build_ansible_decision_audit_payload( incident=recorded["incident"], proposal_data=durable_scheduled_proposal, decision_path="repair_candidate_controlled_queue", not_used_reason="test", ) assert durable_candidate is not None assert durable_candidate["input"]["source_recurrence"] == ( durable_scheduled_recurrence ) scheduled_candidate = { **candidate["input"], "source_recurrence": durable_scheduled_recurrence, } scheduled_claim = build_ansible_check_mode_claim_input( source_candidate_op_id=result["automation_run_id"], candidate_input=scheduled_candidate, observed_now=datetime.fromisoformat("2026-07-15T05:59:59+08:00"), ) assert scheduled_claim["source_recurrence"] == ( durable_scheduled_recurrence ) unacknowledged_scheduled_candidate = { **scheduled_candidate, "source_recurrence": { **durable_scheduled_recurrence, "durable_readback_verified": False, }, } 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=unacknowledged_scheduled_candidate, observed_now=datetime.fromisoformat("2026-07-15T05:59:59+08:00"), ) write_not_acknowledged = ( await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( recorder=AsyncMock(return_value=False), 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 write_not_acknowledged["status"] == "candidate_write_not_acknowledged" assert write_not_acknowledged["existing"] == 0 assert write_not_acknowledged["active_blockers"] == [ "candidate_write_not_acknowledged" ] @pytest.mark.asyncio @pytest.mark.parametrize( ( "failure_class", "expected_status", "expected_controlled_apply_allowed", "expected_blocker", ), [ ( "wazuh_privileged_convergence_capability_missing", "controlled_privilege_capability_repair_already_queued", True, "wazuh_controlled_privilege_capability_repair_pending", ), ( "wazuh_privileged_convergence_secret_reference_missing", "critical_secret_reference_break_glass_already_queued", False, "wazuh_critical_secret_reference_provisioning_pending", ), ], ) async def test_wazuh_ingress_scheduler_does_not_repeat_failed_check_while_capability_repair_is_open( monkeypatch: pytest.MonkeyPatch, failure_class: str, expected_status: str, expected_controlled_apply_allowed: bool, expected_blocker: str, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job run_id = "00000000-0000-0000-0000-000000000572" class _Mappings: def first(self): return { "op_id": run_id, "automation_run_id": run_id, "source_receipt_ref": "wazuh-alert:2026071500", "latest_check_status": "failed", "latest_apply_status": None, "latest_learning_status": None, "latest_controlled_capability_packet_status": "pending", "latest_controlled_capability_queue_status": "queued", "latest_controlled_capability_packet_expires_at": ( "2026-07-15T06:29:59+08:00" ), "latest_controlled_capability_failure_class": failure_class, "predecision_evidence_ready": True, } class _Result: def mappings(self): return _Mappings() class _Db: async def execute(self, statement, params): query = str(statement) assert ":capability_packet_operation_type" in query assert params["capability_packet_operation_type"] == ( "ansible_controlled_capability_repair_queued" ) return _Result() @asynccontextmanager async def fake_db_context(_project_id: str): yield _Db() 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, "get_db_context", fake_db_context) monkeypatch.setattr( job.settings, "ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR", True, ) monkeypatch.setattr( job, "now_taipei", MagicMock(return_value=datetime.fromisoformat("2026-07-15T05:59:59+08:00")), ) result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( recorder=recorder, evidence_collector=collector, evidence_verifier=verifier, incident_binder=binder, source_recurrence=_current_wazuh_source_recurrence( "2026-07-15T05:59:58+08:00" ), ) assert result["status"] == expected_status assert result["queued"] == 0 assert result["retry_of_failed_check_mode"] is True assert result["controlled_apply_allowed"] is ( expected_controlled_apply_allowed ) assert result["active_blockers"] == [expected_blocker] recorder.assert_not_awaited() collector.assert_not_awaited() verifier.assert_not_awaited() binder.assert_not_awaited() def test_wazuh_controlled_capability_packet_cooldown_expires() -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job observed_now = datetime.fromisoformat("2026-07-15T05:59:59+08:00") row = { "latest_controlled_capability_packet_status": "pending", "latest_controlled_capability_queue_status": "queued", "latest_controlled_capability_packet_expires_at": ( "2026-07-15T06:29:59+08:00" ), } assert job._controlled_capability_packet_is_active( row, observed_now=observed_now, ) is True assert job._controlled_capability_packet_is_active( { **row, "latest_controlled_capability_packet_expires_at": ( "2026-07-15T05:59:58+08:00" ), }, observed_now=observed_now, ) is False @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, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job class _Mappings: def first(self): return None 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=False) monkeypatch.setattr(job, "get_db_context", fake_db_context) monkeypatch.setattr( job.settings, "ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR", True, ) monkeypatch.setattr( job, "build_ansible_decision_audit_payload", MagicMock(return_value=None), ) result = await job.enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( recorder=recorder, 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" assert result["existing"] == 0 assert result["active_blockers"] == [ "canonical_asset_or_ansible_catalog_unresolved" ] recorder.assert_not_awaited() @pytest.mark.asyncio async def test_wazuh_ingress_scheduler_blocks_orphan_candidate( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job class _Mappings: def first(self): return None 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(return_value=MagicMock(snapshot_id="must-not-run")) monkeypatch.setattr(job, "get_db_context", fake_db_context) 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=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" assert result["active_blockers"] == [ "scheduled_incident_write_or_readback_not_acknowledged" ] collector.assert_not_awaited() recorder.assert_not_awaited() @pytest.mark.asyncio async def test_wazuh_scheduled_incident_bind_is_idempotent_and_public_safe( monkeypatch: pytest.MonkeyPatch, ) -> None: from src.jobs import awooop_ansible_candidate_backfill_job as job calls: list[tuple[str, dict]] = [] class _Mappings: def first(self): return { "incident_id": "IWZ-INGRESS-2026071500", "project_id": "awoooi", "status": "INVESTIGATING", "alertname": "WazuhAlertmanagerIntegrationConvergence", "notification_type": "TYPE-5S", "alert_category": "security", "ttl_days": 7, "vectorized": False, "signal_count": 1, "affected_service_count": 2, } class _Result: def mappings(self): return _Mappings() class _Db: async def execute(self, statement, params): calls.append((str(statement), dict(params))) return _Result() @asynccontextmanager async def fake_db_context(project_id: str): assert project_id == "awoooi" yield _Db() monkeypatch.setattr(job, "get_db_context", fake_db_context) incident = job._wazuh_ingress_incident( automation_run_id="00000000-0000-0000-0000-000000000112", bucket_ref="2026071500", ) assert await job.ensure_iwooos_wazuh_scheduled_incident( incident=incident, project_id="awoooi", ) is True assert "ON CONFLICT (incident_id) DO NOTHING" in calls[0][0] inserted = calls[0][1] assert inserted["notification_type"] == "TYPE-5S" assert inserted["severity"] == "P1" assert inserted["ttl_days"] == 7 assert inserted["vectorized"] is False assert "ttl_days, vectorized" in calls[0][0] encoded = json.dumps(inserted, default=str, sort_keys=True) assert "sensor-agent" in encoded assert "raw" not in encoded.lower() assert "secret" not in encoded.lower() def test_wazuh_ingress_scheduler_worker_mounts_canonical_service_registry() -> None: deployment = yaml.safe_load(WORKER_DEPLOYMENT.read_text(encoding="utf-8")) pod_spec = deployment["spec"]["template"]["spec"] container = pod_spec["containers"][0] mount = next( item for item in container["volumeMounts"] if item["name"] == "service-registry" ) volume = next(item for item in pod_spec["volumes"] if item["name"] == "service-registry") assert mount == { "name": "service-registry", "mountPath": "/app/ops/config/service-registry.yaml", "subPath": "service-registry.yaml", "readOnly": True, } assert volume["configMap"]["name"] == "service-registry" def test_wazuh_security_event_uses_secops_lifecycle() -> None: assert classify_alert_early( "WazuhSecurityEvent_5710", "critical", {"tool": "wazuh"}, ) == ("secops", "TYPE-5S")