fix(security): bind scheduled Wazuh runs to incidents

apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py

@@ -40,6 +40,7 @@ logger = structlog.get_logger(__name__)
 Recorder = Callable[..., Awaitable[bool]]
 EvidenceCollector = Callable[..., Awaitable[EvidenceSnapshot | None]]
 EvidenceVerifier = Callable[..., Awaitable[bool]]
+IncidentBinder = Callable[..., Awaitable[bool]]
 
 _BACKFILL_DECISION_PATH = "repair_candidate_controlled_queue"
 _BACKFILL_REASON = (
@@ -581,6 +582,163 @@ def _wazuh_ingress_incident(
     }
 
 
+def _public_safe_wazuh_scheduled_incident_record(
+    incident: Mapping[str, Any],
+    *,
+    project_id: str,
+) -> dict[str, Any]:
+    """Build the bounded durable Incident row shared by the Wazuh scheduler."""
+
+    incident_id = str(incident.get("incident_id") or "").strip()
+    if not incident_id.startswith("IWZ-") or len(incident_id) > 30:
+        raise ValueError("invalid_wazuh_scheduler_incident_id")
+    if str(incident.get("project_id") or project_id) != project_id:
+        raise ValueError("wazuh_scheduler_project_mismatch")
+
+    alertname = str(incident.get("alertname") or "")[:100]
+    ingress_lane = alertname == "WazuhAlertmanagerIntegrationConvergence"
+    severity = {
+        "critical": "P0",
+        "high": "P1",
+        "medium": "P2",
+        "warning": "P2",
+        "low": "P3",
+    }.get(str(incident.get("severity") or "").lower(), "P2")
+    signal_input = next(
+        (
+            value
+            for value in incident.get("signals") or []
+            if isinstance(value, Mapping)
+        ),
+        {},
+    )
+    raw_labels = signal_input.get("labels")
+    raw_annotations = signal_input.get("annotations")
+    label_keys = {
+        "alertname",
+        "service",
+        "component",
+        "tool",
+        "asset_alias",
+    }
+    labels = {
+        str(key): str(value)[:120]
+        for key, value in (
+            raw_labels.items() if isinstance(raw_labels, Mapping) else []
+        )
+        if str(key) in label_keys and str(value).strip()
+    }
+    annotations = (
+        {"summary": str(raw_annotations.get("summary") or "")[:240]}
+        if isinstance(raw_annotations, Mapping)
+        else {}
+    )
+    observed_at = now_taipei()
+    signal = {
+        "signal_id": uuid5(
+            NAMESPACE_URL,
+            f"awoooi:iwooos:wazuh-scheduled-signal:{incident_id}",
+        ).hex[:8],
+        "alert_name": alertname,
+        "severity": severity,
+        "source": "sensor-agent",
+        "fired_at": observed_at.isoformat(),
+        "resolved_at": None,
+        "labels": labels,
+        "annotations": annotations,
+        "fingerprint": uuid5(
+            NAMESPACE_URL,
+            f"awoooi:iwooos:wazuh-scheduled-fingerprint:{incident_id}",
+        ).hex,
+    }
+    return {
+        "incident_id": incident_id,
+        "project_id": project_id,
+        "status": "INVESTIGATING",
+        "severity": severity,
+        "signals": json.dumps([signal], ensure_ascii=False),
+        "affected_services": json.dumps(
+            [
+                str(value)[:120]
+                for value in incident.get("affected_services") or []
+                if str(value).strip()
+            ][:20],
+            ensure_ascii=False,
+        ),
+        "proposal_ids": "[]",
+        "alertname": alertname,
+        "notification_type": "TYPE-5S" if ingress_lane else "TYPE-2",
+        "alert_category": "security",
+        "created_at": observed_at,
+        "updated_at": observed_at,
+    }
+
+
+async def ensure_iwooos_wazuh_scheduled_incident(
+    *,
+    incident: Mapping[str, Any],
+    project_id: str,
+) -> bool:
+    """Idempotently bind a scheduled Wazuh run to a durable Incident row."""
+
+    record = _public_safe_wazuh_scheduled_incident_record(
+        incident,
+        project_id=project_id,
+    )
+    async with get_db_context(project_id) as db:
+        await db.execute(
+            text("""
+                INSERT INTO incidents (
+                    incident_id, project_id, status, severity,
+                    signals, affected_services, proposal_ids,
+                    alertname, notification_type, alert_category,
+                    created_at, updated_at
+                ) VALUES (
+                    :incident_id, :project_id, :status, :severity,
+                    CAST(:signals AS json),
+                    CAST(:affected_services AS json),
+                    CAST(:proposal_ids AS json),
+                    :alertname, :notification_type, :alert_category,
+                    :created_at, :updated_at
+                )
+                ON CONFLICT (incident_id) DO NOTHING
+            """),
+            record,
+        )
+        readback = await db.execute(
+            text("""
+                SELECT
+                    incident_id,
+                    project_id,
+                    status::text AS status,
+                    alertname,
+                    notification_type,
+                    alert_category,
+                    json_array_length(signals) AS signal_count,
+                    json_array_length(affected_services) AS affected_service_count
+                FROM incidents
+                WHERE incident_id = :incident_id
+                  AND project_id = :project_id
+                LIMIT 1
+            """),
+            {
+                "incident_id": record["incident_id"],
+                "project_id": record["project_id"],
+            },
+        )
+        row = readback.mappings().first()
+    return bool(
+        row
+        and str(row.get("status") or "").upper()
+        in {"INVESTIGATING", "MITIGATING"}
+        and row.get("alertname") == record["alertname"]
+        and row.get("notification_type") == record["notification_type"]
+        and row.get("alert_category") == "security"
+        and int(row.get("signal_count") or 0) > 0
+        and int(row.get("affected_service_count") or 0) > 0
+    )
+
+
 async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
     *,
     project_id: str = "awoooi",
@@ -598,6 +756,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
         "scheduled Wazuh manager posture readback entered the allowlisted "
         "check-mode and independent verifier chain"
     ),
+    _incident_binder: IncidentBinder | None = None,
 ) -> dict[str, Any]:
     """Enqueue one fresh, no-write Wazuh manager posture executor run."""
 
@@ -803,6 +962,32 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once(
             bucket_ref=bucket.strftime("%Y%m%d%H"),
             attempt_ref=retry_attempt_ref,
         )
+        if _incident_binder is not None:
+            try:
+                incident_bound = await _incident_binder(
+                    incident=incident,
+                    project_id=project_id,
+                )
+            except Exception as exc:
+                incident_bound = False
+                logger.warning(
+                    "iwooos_wazuh_scheduled_incident_bind_failed",
+                    incident_id=incident["incident_id"],
+                    error_type=type(exc).__name__,
+                )
+            if not incident_bound:
+                return {
+                    "status": "blocked_scheduled_incident_write_not_acknowledged",
+                    "queued": 0,
+                    "existing": 1 if existing_row else 0,
+                    "evidence_ready": 0,
+                    "automation_run_id": automation_run_id,
+                    "work_item_id": _work_item_id,
+                    "retry_reason": retry_reason,
+                    "active_blockers": [
+                        "scheduled_incident_write_or_readback_not_acknowledged"
+                    ],
+                }
         try:
             snapshot = await evidence_collector(
                 incident=incident,
@@ -927,6 +1112,7 @@ async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
     recorder: Recorder = record_ansible_decision_audit,
     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,
 ) -> dict[str, Any]:
     """Enqueue idempotent Wazuh event-ingress convergence before posture work."""
 
@@ -946,6 +1132,7 @@ async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once(
             "scheduled Wazuh alert ingress convergence entered the allowlisted "
             "check-mode, bounded apply, rollback, and independent verifier chain"
         ),
+        _incident_binder=incident_binder,
     )
 
 
@@ -1194,7 +1381,9 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None:
             ingress_result = (
                 await enqueue_iwooos_wazuh_alertmanager_integration_candidate_once()
             )
-            posture_result = await enqueue_iwooos_wazuh_manager_posture_candidate_once()
+            posture_result = await enqueue_iwooos_wazuh_manager_posture_candidate_once(
+                _incident_binder=ensure_iwooos_wazuh_scheduled_incident,
+            )
             result = await enqueue_missing_ansible_candidates_once(
                 limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT,
                 window_hours=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS,

apps/api/tests/test_wazuh_alertmanager_integration.py

@@ -317,6 +317,7 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat
         recorder=recorder,
         evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-ingress")),
         evidence_verifier=AsyncMock(return_value=True),
+        incident_binder=AsyncMock(return_value=True),
     )
 
     assert result["status"] == "queued_for_controlled_executor"
@@ -340,6 +341,7 @@ async def test_wazuh_ingress_scheduler_records_one_high_risk_controlled_candidat
                 return_value=MagicMock(snapshot_id="wazuh-ingress")
             ),
             evidence_verifier=AsyncMock(return_value=True),
+            incident_binder=AsyncMock(return_value=True),
         )
     )
     assert write_not_acknowledged["status"] == "candidate_write_not_acknowledged"
@@ -388,6 +390,7 @@ async def test_wazuh_ingress_scheduler_reports_unresolved_payload_before_recordi
         recorder=recorder,
         evidence_collector=AsyncMock(return_value=MagicMock(snapshot_id="wazuh-ingress")),
         evidence_verifier=AsyncMock(return_value=True),
+        incident_binder=AsyncMock(return_value=True),
     )
 
     assert result["status"] == "blocked_canonical_asset_or_ansible_catalog_unresolved"
@@ -398,6 +401,107 @@ async def test_wazuh_ingress_scheduler_reports_unresolved_payload_before_recordi
     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),
+    )
+
+    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",
+                "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"
+    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"]