From af6e5848a8ba7c7e7e0a4f901e82cf30c151e31e Mon Sep 17 00:00:00 2001 From: ogt Date: Wed, 15 Jul 2026 16:03:48 +0800 Subject: [PATCH] fix(security): bind scheduled Wazuh runs to incidents --- .../awooop_ansible_candidate_backfill_job.py | 191 +++++++++++++++++- .../test_wazuh_alertmanager_integration.py | 104 ++++++++++ 2 files changed, 294 insertions(+), 1 deletion(-) diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index dc835a6d1..29f877c8a 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/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, diff --git a/apps/api/tests/test_wazuh_alertmanager_integration.py b/apps/api/tests/test_wazuh_alertmanager_integration.py index 53778c169..07b7cdf00 100644 --- a/apps/api/tests/test_wazuh_alertmanager_integration.py +++ b/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"]