diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index 165605c25..e1f01a88c 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -839,7 +839,8 @@ async def preview_iwooos_wazuh_controlled_executor_handoff( response_model=dict[str, Any], summary="取得 Wazuh controlled executor live runtime receipt 讀回", description=( - "讀取內部 worker 排程的 Wazuh manager bounded no-write posture executor " + "優先讀取內部 worker 排程的 Wazuh sanitized alert-ingress convergence，" + "尚無 ingress candidate 時才回退到 manager bounded no-write posture executor；" "candidate、check-mode、execution、independent verifier、KM / RAG / PlayBook、" "MCP context 與 Telegram durable receipts。此端點不執行命令、不回傳 command output、" "inventory identity、主機位址、raw Wazuh payload 或機密值。" diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index 053cc8646..ab75c5dae 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -747,8 +747,8 @@ class Settings(BaseSettings): ENABLE_IWOOOS_WAZUH_MANAGER_POSTURE_EXECUTOR: bool = Field( default=False, description=( - "True=periodically enqueue the allowlisted no-write Wazuh manager " - "posture playbook into the existing Ansible controlled executor." + "True=periodically converge the allowlisted Wazuh sanitized alert " + "ingress before enqueueing the no-write manager posture playbook." ), ) IWOOOS_WAZUH_MANAGER_POSTURE_FRESHNESS_HOURS: int = Field( @@ -756,7 +756,7 @@ class Settings(BaseSettings): ge=1, le=24, description=( - "Minimum freshness window between Wazuh manager posture executor runs." + "Minimum freshness window between Wazuh ingress and posture runs." ), ) AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS: str = Field( @@ -812,6 +812,13 @@ class Settings(BaseSettings): default="/etc/ssh-mcp/known_hosts", description="known_hosts path for Ansible check-mode SSH transport.", ) + AWOOOP_ANSIBLE_BECOME_PASSWORD_FILE: str = Field( + default="/run/secrets/host112-become/password", + description=( + "Optional Ansible become password file. The value is consumed by " + "Ansible through its password-file contract and never placed in argv." + ), + ) AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS: int = Field( default=24, ge=1, 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 e89e2d5a2..0745852e2 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -58,6 +58,8 @@ def _error_backoff_seconds() -> float: _WAZUH_POSTURE_CATALOG_ID = "ansible:wazuh-manager-posture-readback" _WAZUH_POSTURE_WORK_ITEM_ID = "P0-03-WAZUH-MANAGER-POSTURE" _WAZUH_POSTURE_PUBLIC_ASSET_ALIAS = "security_manager_primary" +_WAZUH_INGRESS_CATALOG_ID = "ansible:wazuh-alertmanager-integration" +_WAZUH_INGRESS_WORK_ITEM_ID = "P0-03-WAZUH-ALERT-INGRESS" async def _fetch_missing_candidate_incidents( @@ -527,12 +529,75 @@ def _wazuh_posture_incident( } +def _wazuh_ingress_incident( + *, + automation_run_id: str, + bucket_ref: str, + attempt_ref: str | None = None, +) -> dict[str, Any]: + receipt_ref = ( + f"{bucket_ref}-{attempt_ref.upper()}" if attempt_ref else bucket_ref + ) + incident_id = f"IWZ-INGRESS-{bucket_ref}" + if attempt_ref: + attempt_token = uuid5( + NAMESPACE_URL, + f"awoooi:iwooos:wazuh-alert-ingress-attempt:{attempt_ref}", + ).hex[:12].upper() + incident_id = f"IWZ-I-{bucket_ref}-{attempt_token}" + return { + "incident_id": incident_id, + "project_id": "awoooi", + "status": "INVESTIGATING", + "severity": "high", + "alertname": "WazuhAlertmanagerIntegrationConvergence", + "alert_category": "security", + "notification_type": "controlled_security_configuration", + "trace_id": automation_run_id, + "run_id": automation_run_id, + "work_item_id": _WAZUH_INGRESS_WORK_ITEM_ID, + "source_receipt_ref": f"scheduled-wazuh-alert-ingress:{receipt_ref}", + "asset_scope_aliases": [_WAZUH_POSTURE_PUBLIC_ASSET_ALIAS], + "affected_services": ["wazuh-manager", "siem"], + "signals": [ + { + "alert_name": "WazuhAlertmanagerIntegrationConvergence", + "source": "iwooos_scheduler", + "labels": { + "alertname": "WazuhAlertmanagerIntegrationConvergence", + "service": "wazuh-manager", + "component": "wazuh-manager", + "tool": "wazuh", + "asset_alias": _WAZUH_POSTURE_PUBLIC_ASSET_ALIAS, + }, + "annotations": { + "summary": ( + "converge sanitized Wazuh event ingress with bounded " + "rollback and independent verification" + ) + }, + } + ], + } + + async def enqueue_iwooos_wazuh_manager_posture_candidate_once( *, project_id: str = "awoooi", recorder: Recorder = record_ansible_decision_audit, evidence_collector: EvidenceCollector = collect_ansible_candidate_pre_decision_evidence, evidence_verifier: EvidenceVerifier = verify_ansible_candidate_pre_decision_evidence, + _catalog_id: str = _WAZUH_POSTURE_CATALOG_ID, + _work_item_id: str = _WAZUH_POSTURE_WORK_ITEM_ID, + _incident_factory: Callable[..., dict[str, Any]] = _wazuh_posture_incident, + _run_namespace: str = "awoooi:iwooos:wazuh-manager-posture", + _scheduler_source: str = "iwooos_wazuh_manager_posture_scheduler", + _proposal_action: str = "run_bounded_no_write_wazuh_manager_posture_readback", + _risk_level: str = "low", + _reason: str = ( + "scheduled Wazuh manager posture readback entered the allowlisted " + "check-mode and independent verifier chain" + ), ) -> dict[str, Any]: """Enqueue one fresh, no-write Wazuh manager posture executor run.""" @@ -616,7 +681,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( """), { "catalog_match": json.dumps( - [{"catalog_id": _WAZUH_POSTURE_CATALOG_ID}] + [{"catalog_id": _catalog_id}] ), "freshness_hours": freshness_hours, }, @@ -647,7 +712,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( or existing_row.get("op_id") or "" ), - "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "work_item_id": _work_item_id, "active_blockers": [], } if existing_row: @@ -678,7 +743,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( or existing_row.get("op_id") or "" ), - "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "work_item_id": _work_item_id, "retry_reason": retry_reason, "active_blockers": [ "predecision_mcp_or_log_evidence_missing" @@ -704,7 +769,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( or existing_row.get("op_id") or "" ), - "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "work_item_id": _work_item_id, "retry_reason": retry_reason, "retry_of_failed_check_mode": ( retry_reason == "check_mode_failed" @@ -729,11 +794,11 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( automation_run_id = str( uuid5( NAMESPACE_URL, - "awoooi:iwooos:wazuh-manager-posture:" + f"{_run_namespace}:" f"{bucket.isoformat()}:{retry_attempt_ref or 'initial'}", ) ) - incident = _wazuh_posture_incident( + incident = _incident_factory( automation_run_id=automation_run_id, bucket_ref=bucket.strftime("%Y%m%d%H"), attempt_ref=retry_attempt_ref, @@ -762,7 +827,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "existing": 1 if existing_row else 0, "evidence_ready": 0, "automation_run_id": automation_run_id, - "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "work_item_id": _work_item_id, "retry_reason": retry_reason, "retry_of_failed_check_mode": retry_reason == "check_mode_failed", "retry_of_incomplete_predecision_context": ( @@ -780,16 +845,13 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( queued = await recorder( incident=incident, proposal_data={ - "source": "iwooos_wazuh_manager_posture_scheduler", - "risk_level": "low", + "source": _scheduler_source, + "risk_level": _risk_level, "execution_priority": 0, - "action": "run_bounded_no_write_wazuh_manager_posture_readback", + "action": _proposal_action, }, decision_path=_BACKFILL_DECISION_PATH, - not_used_reason=( - "scheduled Wazuh manager posture readback entered the allowlisted " - "check-mode and independent verifier chain" - ), + not_used_reason=_reason, automation_run_id=automation_run_id, ) return { @@ -806,7 +868,7 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "existing": 0 if queued else 1, "evidence_ready": 1 if evidence_ready else 0, "automation_run_id": automation_run_id, - "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "work_item_id": _work_item_id, "retry_reason": retry_reason, "retry_of_failed_check_mode": retry_reason == "check_mode_failed", "retry_of_incomplete_predecision_context": ( @@ -828,11 +890,39 @@ async def enqueue_iwooos_wazuh_manager_posture_candidate_once( "queued": 0, "existing": 0, "evidence_ready": 0, - "work_item_id": _WAZUH_POSTURE_WORK_ITEM_ID, + "work_item_id": _work_item_id, "active_blockers": [f"candidate_error:{type(exc).__name__}"], } +async def enqueue_iwooos_wazuh_alertmanager_integration_candidate_once( + *, + project_id: str = "awoooi", + recorder: Recorder = record_ansible_decision_audit, + evidence_collector: EvidenceCollector = collect_ansible_candidate_pre_decision_evidence, + evidence_verifier: EvidenceVerifier = verify_ansible_candidate_pre_decision_evidence, +) -> dict[str, Any]: + """Enqueue idempotent Wazuh event-ingress convergence before posture work.""" + + return await enqueue_iwooos_wazuh_manager_posture_candidate_once( + project_id=project_id, + recorder=recorder, + evidence_collector=evidence_collector, + evidence_verifier=evidence_verifier, + _catalog_id=_WAZUH_INGRESS_CATALOG_ID, + _work_item_id=_WAZUH_INGRESS_WORK_ITEM_ID, + _incident_factory=_wazuh_ingress_incident, + _run_namespace="awoooi:iwooos:wazuh-alert-ingress", + _scheduler_source="iwooos_wazuh_alert_ingress_scheduler", + _proposal_action="converge_wazuh_alertmanager_integration", + _risk_level="high", + _reason=( + "scheduled Wazuh alert ingress convergence entered the allowlisted " + "check-mode, bounded apply, rollback, and independent verifier chain" + ), + ) + + def _wazuh_posture_retry_reason( existing_row: Mapping[str, Any] | None, ) -> str | None: @@ -1075,6 +1165,9 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None: settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS ) try: + ingress_result = ( + await enqueue_iwooos_wazuh_alertmanager_integration_candidate_once() + ) posture_result = await enqueue_iwooos_wazuh_manager_posture_candidate_once() result = await enqueue_missing_ansible_candidates_once( limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT, @@ -1093,6 +1186,11 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None: "iwooos_wazuh_manager_posture_candidate_tick", **posture_result, ) + if ingress_result.get("queued") or ingress_result.get("active_blockers"): + logger.info( + "iwooos_wazuh_alert_ingress_candidate_tick", + **ingress_result, + ) except Exception as exc: logger.warning( "awooop_ansible_candidate_backfill_worker_failed", error=str(exc) diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 42dc7d31b..868a537e6 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -507,6 +507,36 @@ _CATALOG: tuple[dict[str, Any], ...] = ( "risk_level": "low", "no_write_only": True, }, + { + "catalog_id": "ansible:wazuh-alertmanager-integration", + "playbook_path": ( + "infra/ansible/playbooks/wazuh-alertmanager-integration.yml" + ), + "check_mode_playbook_path": ( + "infra/ansible/playbooks/wazuh-alertmanager-integration.yml" + ), + "inventory_hosts": ["host_112"], + "domains": [ + "wazuh", + "siem", + "security_event_ingress", + "alertmanager", + ], + "keywords": [ + "wazuhalertmanagerintegrationconvergence", + "wazuh alertmanager integration convergence", + "custom-awoooi-alertmanager", + "security event ingress", + ], + "activation_keywords": [ + "wazuhalertmanagerintegrationconvergence", + ], + "supports_check_mode": True, + "auto_apply_enabled": True, + "approval_required": False, + "risk_level": "high", + "catalog_revision": "2026-07-15-wazuh-alert-ingress-v1", + }, { "catalog_id": "ansible:110-host-pressure-readonly", "playbook_path": "infra/ansible/playbooks/110-host-pressure-readonly.yml", diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 271f598d0..d0ed335c8 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -634,17 +634,33 @@ def _resolve_playbook_path(playbook_root: Path, playbook_path: str) -> Path: return resolved -def _ansible_command_env(playbook_root: Path) -> dict[str, str]: +def _ansible_command_env( + playbook_root: Path, + *, + allow_become_password_file: bool = False, +) -> dict[str, str]: roles_path = str((playbook_root / "roles").resolve()) existing_roles_path = os.environ.get("ANSIBLE_ROLES_PATH") if existing_roles_path: roles_path = os.pathsep.join([roles_path, existing_roles_path]) - return { + env = { **os.environ, "ANSIBLE_HOST_KEY_CHECKING": "true", "ANSIBLE_RETRY_FILES_ENABLED": "false", "ANSIBLE_ROLES_PATH": roles_path, } + become_password_file = Path( + settings.AWOOOP_ANSIBLE_BECOME_PASSWORD_FILE + ) + if ( + allow_become_password_file + and become_password_file.is_file() + and os.access(become_password_file, os.R_OK) + ): + env["ANSIBLE_BECOME_PASSWORD_FILE"] = str(become_password_file) + else: + env.pop("ANSIBLE_BECOME_PASSWORD_FILE", None) + return env def build_ansible_check_mode_command( @@ -687,7 +703,14 @@ def build_ansible_check_mode_command( "--extra-vars", json.dumps(extra_vars, ensure_ascii=False, separators=(",", ":")), ] - env = _ansible_command_env(root) + env = _ansible_command_env( + root, + allow_become_password_file=( + playbook_path + == "infra/ansible/playbooks/wazuh-alertmanager-integration.yml" + and inventory_hosts == ("host_112",) + ), + ) return AnsibleCommandSpec( command=command, cwd=root, @@ -738,7 +761,14 @@ def build_ansible_apply_command( "--extra-vars", json.dumps(extra_vars, ensure_ascii=False, separators=(",", ":")), ] - env = _ansible_command_env(root) + env = _ansible_command_env( + root, + allow_become_password_file=( + playbook_path + == "infra/ansible/playbooks/wazuh-alertmanager-integration.yml" + and inventory_hosts == ("host_112",) + ), + ) return AnsibleCommandSpec( command=command, cwd=root, diff --git a/apps/api/src/services/awooop_ansible_post_verifier.py b/apps/api/src/services/awooop_ansible_post_verifier.py index c3a21d668..db212a768 100644 --- a/apps/api/src/services/awooop_ansible_post_verifier.py +++ b/apps/api/src/services/awooop_ansible_post_verifier.py @@ -72,6 +72,47 @@ _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = { "systemctl is-active --quiet wazuh-manager.service", ), ), + "ansible:wazuh-alertmanager-integration": ( + AssetPostcondition( + "host_112_wazuh_alertmanager_script", + "security", + "host_112", + "test -r /var/lib/awoooi-wazuh-alert-ingress/receipt.env && " + "grep -Fqx 'integration_name=custom-awoooi-alertmanager' " + "/var/lib/awoooi-wazuh-alert-ingress/receipt.env", + ), + AssetPostcondition( + "host_112_wazuh_alertmanager_config", + "security", + "host_112", + "grep -Fqx 'config_present=1' " + "/var/lib/awoooi-wazuh-alert-ingress/receipt.env && " + "grep -Fqx 'raw_payload_persisted=0' " + "/var/lib/awoooi-wazuh-alert-ingress/receipt.env && " + "grep -Fqx 'active_response_enabled=0' " + "/var/lib/awoooi-wazuh-alert-ingress/receipt.env", + ), + AssetPostcondition( + "host_112_wazuh_manager_runtime_after_ingress_convergence", + "service", + "host_112", + "systemctl is-active --quiet wazuh-manager.service", + ), + AssetPostcondition( + "host_112_wazuh_integrator_runtime", + "service", + "host_112", + "pgrep -x wazuh-integratord >/dev/null && " + "grep -Fqx 'integrator_running=1' " + "/var/lib/awoooi-wazuh-alert-ingress/receipt.env", + ), + AssetPostcondition( + "host_112_awoooi_alertmanager_ingress_reachable", + "network", + "host_112", + "curl -fsS --max-time 10 https://awoooi.wooo.work/api/v1/webhooks/health >/dev/null", + ), + ), "ansible:110-host-pressure-readonly": ( AssetPostcondition( "host_110_pressure_controller_asset", diff --git a/apps/api/src/services/controlled_alert_target_router.py b/apps/api/src/services/controlled_alert_target_router.py index 7ae678a4d..3cd976a0e 100644 --- a/apps/api/src/services/controlled_alert_target_router.py +++ b/apps/api/src/services/controlled_alert_target_router.py @@ -328,6 +328,32 @@ def resolve_typed_alert_target( execution_role="single_writer_windows_vmware_executor", ) + if compact_alert == "wazuhalertmanagerintegrationconvergence": + identity = registry.resolve_identity(target_resource) + inventory_host = _host_scope(str(identity.get("host") or "")) + if ( + identity.get("resolution_status") == "resolved" + and identity.get("canonical_id") == "service:wazuh-manager" + and inventory_host is not None + and inventory_host[0] == "host_112" + ): + return _typed_route( + resolution_status="resolved", + target_kind="host_systemd", + target_resource=target_resource, + namespace=namespace, + canonical_asset_id="service:wazuh-manager", + host=inventory_host[1], + executor="host_ansible_executor", + verifier="wazuh_alert_ingress_independent_verifier", + risk_class="high", + allowed_catalog_ids=[ + "ansible:wazuh-alertmanager-integration" + ], + allowed_inventory_hosts=["host_112"], + stateful_level=str(identity.get("stateful_level") or ""), + ) + host_scope = _host_scope( str(labels.get("host") or ""), str(labels.get("instance") or ""), diff --git a/apps/api/src/services/incident_service.py b/apps/api/src/services/incident_service.py index f1f27b64a..c9845dbf1 100644 --- a/apps/api/src/services/incident_service.py +++ b/apps/api/src/services/incident_service.py @@ -194,7 +194,7 @@ def classify_alert_early( # ADR-075 TYPE-5S (2026-04-12 ogt) if any(alertname.startswith(p) for p in ( "UnauthorizedSSH", "KubeAudit", "CVECritical", "WAFAttack", - "PodAbnormal", "SecurityBreach", + "PodAbnormal", "SecurityBreach", "WazuhSecurityEvent", )): return "secops", "TYPE-5S" diff --git a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py index 7f886bc91..d206b8041 100644 --- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py @@ -24,6 +24,7 @@ from src.services.awooop_ansible_audit_service import get_ansible_catalog_item SCHEMA_VERSION = "iwooos_wazuh_controlled_executor_runtime_readback_v1" CATALOG_ID = "ansible:wazuh-manager-posture-readback" +INGRESS_CATALOG_ID = "ansible:wazuh-alertmanager-integration" WORK_ITEM_ID = "P0-03-WAZUH-MANAGER-POSTURE" _LIVE_RUNTIME_SQL = """ @@ -36,6 +37,7 @@ _LIVE_RUNTIME_SQL = """ input ->> 'run_id' AS run_id, input ->> 'work_item_id' AS work_item_id, input ->> 'source_receipt_ref' AS source_receipt_ref, + input -> 'executor_candidates' -> 0 ->> 'catalog_id' AS catalog_id, jsonb_array_length( coalesce(input -> 'asset_scope_aliases', '[]'::jsonb) ) AS asset_scope_alias_count, @@ -199,14 +201,20 @@ async def load_latest_iwooos_wazuh_controlled_executor_runtime( try: async with get_db_context(project_id) as db: - result = await db.execute( - text(_LIVE_RUNTIME_SQL), - { - "catalog_match": json.dumps([{"catalog_id": CATALOG_ID}]), - "project_id": project_id, - }, - ) - row = result.mappings().first() + row = None + for catalog_id in (INGRESS_CATALOG_ID, CATALOG_ID): + result = await db.execute( + text(_LIVE_RUNTIME_SQL), + { + "catalog_match": json.dumps( + [{"catalog_id": catalog_id}] + ), + "project_id": project_id, + }, + ) + row = result.mappings().first() + if row: + break except Exception as exc: return build_iwooos_wazuh_controlled_executor_runtime_readback( None, @@ -229,7 +237,9 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( if executor_enabled is None else executor_enabled ) - catalog_ready = get_ansible_catalog_item(CATALOG_ID) is not None + selected_catalog_id = str(data.get("catalog_id") or CATALOG_ID) + ingress_mode = selected_catalog_id == INGRESS_CATALOG_ID + catalog_ready = get_ansible_catalog_item(selected_catalog_id) is not None candidate_present = bool(data.get("candidate_op_id")) check_passed = ( data.get("check_status") == "success" @@ -295,15 +305,15 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( if read_error_type: active_blockers.append(f"runtime_receipt_read_unavailable:{read_error_type}") if not enabled: - active_blockers.append("wazuh_manager_posture_executor_disabled") + active_blockers.append("wazuh_controlled_executor_disabled") if not catalog_ready: - active_blockers.append("wazuh_manager_posture_catalog_missing") + active_blockers.append("wazuh_controlled_catalog_missing") if candidate_present and data.get("check_status") == "failed": - active_blockers.append("wazuh_manager_posture_check_mode_failed") + active_blockers.append("wazuh_controlled_check_mode_failed") if check_failure_class: active_blockers.append(check_failure_class) if apply_failed: - active_blockers.append("wazuh_manager_posture_controlled_probe_failed") + active_blockers.append("wazuh_controlled_apply_failed") if apply_passed and not verifier_passed: active_blockers.append("independent_post_verifier_missing_or_failed") if verifier_passed and not (km_ready and rag_ready and trust_ready): @@ -316,6 +326,7 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( return { "schema_version": SCHEMA_VERSION, "status": _status( + ingress_mode=ingress_mode, read_error_type=read_error_type, candidate_present=candidate_present, check_passed=check_passed, @@ -335,12 +346,18 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( "same_run_correlation_required": True, }, "summary": { + "selected_catalog_id": selected_catalog_id, "executor_policy_enabled_count": 1 if enabled and catalog_ready else 0, "dispatch_candidate_count": 1 if candidate_present else 0, "check_mode_passed_count": 1 if check_passed else 0, "check_mode_failed_count": 1 if check_failed else 0, "runtime_execution_performed_count": 1 if apply_terminal else 0, - "bounded_posture_execution_passed_count": 1 if apply_passed else 0, + "bounded_posture_execution_passed_count": ( + 1 if apply_passed and not ingress_mode else 0 + ), + "bounded_ingress_convergence_passed_count": ( + 1 if apply_passed and ingress_mode else 0 + ), "independent_post_verifier_passed_count": 1 if verifier_passed else 0, "auto_repair_execution_receipt_count": 1 if auto_repair_ready else 0, "km_writeback_count": 1 if km_ready else 0, @@ -357,7 +374,9 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( "same_run_missing_stage_count": len(missing_stage_ids), "same_run_completion_percent": completion_percent, "runtime_closed_count": 1 if runtime_closed else 0, - "host_write_performed_count": 0, + "host_write_performed_count": ( + 1 if apply_terminal and ingress_mode else 0 + ), "wazuh_active_response_performed_count": 0, "secret_value_collection_count": 0, }, @@ -371,6 +390,7 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( "missing_stage_ids": missing_stage_ids, "check_failure_class": check_failure_class, "next_safe_action": _next_action( + ingress_mode=ingress_mode, candidate_present=candidate_present, check_passed=check_passed, check_failed=check_failed, @@ -385,8 +405,12 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( "latest_receipt_at": _latest_receipt_at(data), "boundaries": { "runtime_execution_performed": apply_terminal, - "bounded_no_write_posture_probe": True, - "host_write_performed": False, + "bounded_no_write_posture_probe": not ingress_mode, + "bounded_alert_ingress_convergence": ingress_mode, + "sanitized_alert_ingress_configured": ( + ingress_mode and verifier_passed + ), + "host_write_performed": apply_terminal and ingress_mode, "wazuh_active_response_performed": False, "live_wazuh_api_query_performed": False, "raw_wazuh_payload_persisted": False, @@ -397,6 +421,8 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( }, "source_refs": [ "infra/ansible/playbooks/wazuh-manager-posture-readback.yml", + "infra/ansible/playbooks/wazuh-alertmanager-integration.yml", + "infra/ansible/files/wazuh/custom-awoooi-alertmanager.py", "apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py", "apps/api/src/services/awooop_ansible_check_mode_service.py", "apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py", @@ -406,6 +432,7 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( def _status( *, + ingress_mode: bool, read_error_type: str | None, candidate_present: bool, check_passed: bool, @@ -414,25 +441,27 @@ def _status( apply_passed: bool, runtime_closed: bool, ) -> str: + prefix = "wazuh_alert_ingress" if ingress_mode else "wazuh_manager_posture" if read_error_type: return "degraded_runtime_receipt_read_unavailable" if runtime_closed: - return "wazuh_manager_posture_same_run_runtime_closed" + return f"{prefix}_same_run_runtime_closed" if apply_terminal and not apply_passed: - return "wazuh_manager_posture_execution_failed_waiting_retry_or_repair" + return f"{prefix}_execution_failed_waiting_retry_or_repair" if apply_passed: - return "wazuh_manager_posture_executed_runtime_writeback_open" + return f"{prefix}_executed_runtime_writeback_open" if check_passed: - return "wazuh_manager_posture_check_passed_waiting_bounded_execution" + return f"{prefix}_check_passed_waiting_bounded_execution" if check_failed: - return "wazuh_manager_posture_check_failed_waiting_bounded_retry" + return f"{prefix}_check_failed_waiting_bounded_retry" if candidate_present: - return "wazuh_manager_posture_dispatched_waiting_check_mode" - return "waiting_for_scheduled_wazuh_manager_posture_dispatch" + return f"{prefix}_dispatched_waiting_check_mode" + return f"waiting_for_scheduled_{prefix}_dispatch" def _next_action( *, + ingress_mode: bool, candidate_present: bool, check_passed: bool, check_failed: bool, @@ -444,9 +473,17 @@ def _next_action( missing_stage_ids: list[str], ) -> str: if runtime_closed: - return "keep_scheduled_wazuh_manager_posture_runtime_receipts_fresh" + return ( + "keep_scheduled_wazuh_alert_ingress_runtime_receipts_fresh" + if ingress_mode + else "keep_scheduled_wazuh_manager_posture_runtime_receipts_fresh" + ) if not candidate_present: - return "internal_worker_enqueue_wazuh_manager_posture_candidate" + return ( + "internal_worker_enqueue_wazuh_alert_ingress_candidate" + if ingress_mode + else "internal_worker_enqueue_wazuh_manager_posture_candidate" + ) if check_failed: return ( "repair_and_enqueue_bounded_retry:" @@ -455,7 +492,11 @@ def _next_action( if not check_passed: return "worker_claim_and_run_allowlisted_ansible_check_mode" if not apply_terminal: - return "run_bounded_no_write_wazuh_manager_posture_execution" + return ( + "run_bounded_wazuh_alert_ingress_convergence" + if ingress_mode + else "run_bounded_no_write_wazuh_manager_posture_execution" + ) if not apply_passed: return "classify_transport_or_manager_failure_then_bounded_retry" if not verifier_passed: diff --git a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py index 19027bc93..ef49b528a 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -1,5 +1,7 @@ from __future__ import annotations +# ruff: noqa: E402, I001 + import os from contextlib import asynccontextmanager from datetime import UTC, datetime @@ -90,6 +92,30 @@ def test_wazuh_runtime_readback_closes_only_with_complete_same_run_receipts() -> assert payload["boundaries"]["command_output_returned"] is False +def test_wazuh_runtime_readback_prefers_ingress_convergence_semantics() -> None: + row = _complete_row() + row["catalog_id"] = "ansible:wazuh-alertmanager-integration" + row["work_item_id"] = "P0-03-WAZUH-ALERT-INGRESS" + + payload = build_iwooos_wazuh_controlled_executor_runtime_readback( + row, + executor_enabled=True, + ) + + assert payload["status"] == "wazuh_alert_ingress_same_run_runtime_closed" + assert payload["work_item_id"] == "P0-03-WAZUH-ALERT-INGRESS" + assert payload["summary"]["selected_catalog_id"] == ( + "ansible:wazuh-alertmanager-integration" + ) + assert payload["summary"]["bounded_posture_execution_passed_count"] == 0 + assert payload["summary"]["bounded_ingress_convergence_passed_count"] == 1 + assert payload["summary"]["host_write_performed_count"] == 1 + assert payload["boundaries"]["bounded_no_write_posture_probe"] is False + assert payload["boundaries"]["bounded_alert_ingress_convergence"] is True + assert payload["boundaries"]["sanitized_alert_ingress_configured"] is True + assert payload["boundaries"]["raw_wazuh_payload_persisted"] is False + + def test_wazuh_runtime_readback_keeps_partial_execution_open() -> None: row = _complete_row() row["stage_ids"] = [ diff --git a/apps/api/tests/test_wazuh_alertmanager_integration.py b/apps/api/tests/test_wazuh_alertmanager_integration.py new file mode 100644 index 000000000..f8c235b45 --- /dev/null +++ b/apps/api/tests/test_wazuh_alertmanager_integration.py @@ -0,0 +1,334 @@ +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, + get_ansible_catalog_item, +) +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_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" +) +CATALOG_ID = "ansible:wazuh-alertmanager-integration" + + +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 playbook[0]["hosts"] == "host_112" + assert "backup: true" in text + assert "remote_src: true" in text + assert "wazuh_privileged_convergence_capability_missing" 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_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} + ] + 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), + ) + + 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" + ) + + +def test_wazuh_security_event_uses_secops_lifecycle() -> None: + assert classify_alert_early( + "WazuhSecurityEvent_5710", + "critical", + {"tool": "wazuh"}, + ) == ("secops", "TYPE-5S") diff --git a/infra/ansible/files/wazuh/custom-awoooi-alertmanager.py b/infra/ansible/files/wazuh/custom-awoooi-alertmanager.py new file mode 100644 index 000000000..dc4848cca --- /dev/null +++ b/infra/ansible/files/wazuh/custom-awoooi-alertmanager.py @@ -0,0 +1,238 @@ +#!/usr/bin/env python3 +"""Forward a public-safe Wazuh alert envelope to AWOOOI Alertmanager ingress.""" + +from __future__ import annotations + +import hashlib +import json +import re +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from urllib.error import HTTPError +from urllib.parse import urlsplit +from urllib.request import HTTPRedirectHandler, Request, build_opener + +_ALLOWED_HOST = "awoooi.wooo.work" +_ALLOWED_PATH = "/api/v1/webhooks/alertmanager" +_MAX_ALERT_BYTES = 1024 * 1024 +_SAFE_TOKEN_RE = re.compile(r"[^A-Za-z0-9_.:-]+") +_MITRE_RE = re.compile(r"^T[0-9]{4}(?:\.[0-9]{3})?$") + + +class _NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + return None + + +def _safe_token(value: Any, *, fallback: str, limit: int = 80) -> str: + token = _SAFE_TOKEN_RE.sub("-", str(value or "").strip()).strip("-._:") + return (token or fallback)[:limit] + + +def _rule_groups(rule: dict[str, Any]) -> list[str]: + raw_groups = rule.get("groups") or [] + if isinstance(raw_groups, str): + raw_groups = raw_groups.split(",") + if not isinstance(raw_groups, list): + return [] + groups = { + _safe_token(value, fallback="unknown", limit=48) + for value in raw_groups + if str(value or "").strip() + } + groups.discard("unknown") + return sorted(groups)[:8] + + +def _mitre_techniques(rule: dict[str, Any]) -> list[str]: + mitre = rule.get("mitre") or {} + raw_ids = mitre.get("id") if isinstance(mitre, dict) else [] + if isinstance(raw_ids, str): + raw_ids = [raw_ids] + if not isinstance(raw_ids, list): + return [] + return sorted({str(value).strip() for value in raw_ids if _MITRE_RE.fullmatch(str(value).strip())})[:8] + + +def _agent_alias(alert: dict[str, Any]) -> str: + manager = alert.get("manager") or {} + agent = alert.get("agent") or {} + manager_identity = str(manager.get("name") or "primary-manager") + agent_identity = str(agent.get("id") or agent.get("name") or "manager-local") + digest = hashlib.sha256( + f"awoooi-wazuh-agent-v1\0{manager_identity}\0{agent_identity}".encode() + ).hexdigest()[:16] + return f"wazuh-agent-{digest}" + + +def _event_timestamp(alert: dict[str, Any], *, now: datetime | None = None) -> str: + value = str(alert.get("timestamp") or "").strip() + if value and len(value) <= 64 and re.match(r"^\d{4}-\d{2}-\d{2}T", value): + return value + return (now or datetime.now(timezone.utc)).isoformat().replace("+00:00", "Z") + + +def build_alertmanager_payload( + alert: dict[str, Any], *, now: datetime | None = None +) -> dict[str, Any]: + """Build a bounded envelope without raw logs or endpoint identities.""" + + rule = alert.get("rule") or {} + if not isinstance(rule, dict): + raise ValueError("invalid_rule") + try: + rule_level = int(rule.get("level")) + except (TypeError, ValueError) as exc: + raise ValueError("invalid_rule_level") from exc + if not 0 <= rule_level <= 16: + raise ValueError("invalid_rule_level") + + rule_id = _safe_token(rule.get("id"), fallback="unknown", limit=24) + agent_alias = _agent_alias(alert) + groups = _rule_groups(rule) + techniques = _mitre_techniques(rule) + severity = "critical" if rule_level >= 12 else "warning" + recurrence_source = "|".join( + ("wazuh-event-v1", rule_id, agent_alias, ",".join(groups)) + ) + fingerprint = hashlib.sha256(recurrence_source.encode()).hexdigest()[:32] + alertname = f"WazuhSecurityEvent_{rule_id}" + starts_at = _event_timestamp(alert, now=now) + + labels = { + "alertname": alertname, + "severity": severity, + "namespace": "awoooi", + "component": "wazuh-manager", + "service": "wazuh-manager", + "tool": "wazuh", + "alert_category": "security", + "deployment": agent_alias, + "asset_alias": agent_alias, + "wazuh_rule_id": rule_id, + "wazuh_rule_level": str(rule_level), + "wazuh_rule_groups": ",".join(groups), + "mitre_techniques": ",".join(techniques), + "raw_payload_persisted": "false", + "active_response_requested": "false", + } + annotations = { + "summary": f"Wazuh rule {rule_id} level {rule_level} requires AI security triage", + "description": "Sanitized Wazuh security signal; full evidence remains in the source system.", + } + return { + "version": "4", + "groupKey": fingerprint, + "status": "firing", + "receiver": "awoooi-wazuh", + "groupLabels": { + "alertname": alertname, + "deployment": agent_alias, + }, + "commonLabels": labels, + "commonAnnotations": annotations, + "alerts": [ + { + "status": "firing", + "labels": labels, + "annotations": annotations, + "startsAt": starts_at, + "fingerprint": fingerprint, + } + ], + } + + +def _validated_hook_url(raw_url: str) -> str: + parsed = urlsplit(raw_url.strip()) + if ( + parsed.scheme != "https" + or parsed.hostname != _ALLOWED_HOST + or parsed.port not in {None, 443} + or parsed.path != _ALLOWED_PATH + or parsed.query + or parsed.fragment + or parsed.username + or parsed.password + ): + raise ValueError("hook_url_not_allowlisted") + return f"https://{_ALLOWED_HOST}{_ALLOWED_PATH}" + + +def _load_alert(path: str) -> dict[str, Any]: + alert_path = Path(path) + with alert_path.open("rb") as handle: + raw = handle.read(_MAX_ALERT_BYTES + 1) + if len(raw) > _MAX_ALERT_BYTES: + raise ValueError("alert_file_too_large") + payload = json.loads(raw.decode("utf-8")) + if not isinstance(payload, dict): + raise ValueError("alert_payload_not_object") + return payload + + +def _post_payload(hook_url: str, payload: dict[str, Any]) -> int: + body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() + request = Request( + hook_url, + data=body, + method="POST", + headers={ + "Content-Type": "application/json", + "User-Agent": "wazuh-custom-awoooi/1", + }, + ) + try: + with build_opener(_NoRedirect()).open(request, timeout=10) as response: + status = int(response.getcode()) + except HTTPError as exc: + status = int(exc.code) + if not 200 <= status < 300: + raise RuntimeError(f"webhook_http_status_{status}") + return status + + +def _self_test() -> None: + fixture = { + "timestamp": "2026-07-15T00:00:00Z", + "manager": {"name": "manager"}, + "agent": {"id": "000", "name": "synthetic"}, + "rule": { + "id": "100001", + "level": 10, + "groups": ["awoooi_canary"], + "mitre": {"id": ["T1110"]}, + }, + "full_log": "must-not-leave-source", + } + encoded = json.dumps(build_alertmanager_payload(fixture), sort_keys=True) + if "must-not-leave-source" in encoded or "synthetic" in encoded: + raise RuntimeError("raw_field_exposure") + _validated_hook_url(f"https://{_ALLOWED_HOST}{_ALLOWED_PATH}") + + +def main(argv: list[str]) -> int: + try: + if argv[1:] == ["--self-test"]: + _self_test() + return 0 + if len(argv) < 4: + raise ValueError("expected_alert_api_key_hook_url_arguments") + # argv[2] is the Wazuh Integrator api_key slot. This integration does + # not configure, read, log, or forward it. + alert = _load_alert(argv[1]) + hook_url = _validated_hook_url(argv[3]) + _post_payload(hook_url, build_alertmanager_payload(alert)) + return 0 + except Exception as exc: # Wazuh retries on a non-zero integration result. + print( + f"awoooi_wazuh_bridge_failed:{type(exc).__name__}", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/infra/ansible/playbooks/wazuh-alertmanager-integration.yml b/infra/ansible/playbooks/wazuh-alertmanager-integration.yml new file mode 100644 index 000000000..c5b31ee76 --- /dev/null +++ b/infra/ansible/playbooks/wazuh-alertmanager-integration.yml @@ -0,0 +1,281 @@ +--- +- name: Converge the Wazuh to AWOOOI sanitized alert ingress + hosts: host_112 + gather_facts: false + become: true + serial: 1 + any_errors_fatal: true + vars: + wazuh_integration_src: >- + {{ playbook_dir }}/../files/wazuh/custom-awoooi-alertmanager.py + wazuh_integration_dest: /var/ossec/integrations/custom-awoooi-alertmanager + wazuh_ossec_conf: /var/ossec/etc/ossec.conf + wazuh_hook_url: https://awoooi.wooo.work/api/v1/webhooks/alertmanager + wazuh_receipt_dir: /var/lib/awoooi-wazuh-alert-ingress + wazuh_receipt_path: /var/lib/awoooi-wazuh-alert-ingress/receipt.env + + tasks: + - name: Verify the repo-owned integration asset before host mutation + ansible.builtin.command: + argv: + - /usr/bin/python3 + - "{{ wazuh_integration_src }}" + - --self-test + delegate_to: localhost + become: false + check_mode: false + changed_when: false + + - name: Compute the repo-owned integration source fingerprint + ansible.builtin.command: + argv: + - /usr/bin/sha256sum + - "{{ wazuh_integration_src }}" + delegate_to: localhost + become: false + register: wazuh_source_sha256 + check_mode: false + changed_when: false + + - name: Probe the exact privileged convergence capability + ansible.builtin.command: + argv: + - /usr/bin/true + become: true + register: wazuh_privilege_preflight + check_mode: false + changed_when: false + failed_when: false + ignore_errors: true + no_log: true + + - name: Fail closed before mutation when privilege is unavailable + ansible.builtin.assert: + that: + - not (wazuh_privilege_preflight.failed | default(true)) + - (wazuh_privilege_preflight.rc | default(1)) == 0 + fail_msg: wazuh_privileged_convergence_capability_missing + become: false + + - name: Verify the Wazuh manager is active before convergence + ansible.builtin.command: + argv: + - /usr/bin/systemctl + - is-active + - wazuh-manager.service + register: wazuh_manager_before + check_mode: false + changed_when: false + failed_when: wazuh_manager_before.rc != 0 + no_log: true + + - name: Inspect existing integration script state without reading content + ansible.builtin.stat: + path: "{{ wazuh_integration_dest }}" + register: wazuh_integration_before + + - name: Inspect existing public-safe convergence receipt + ansible.builtin.stat: + path: "{{ wazuh_receipt_path }}" + register: wazuh_receipt_before + + - name: Converge integration with remote-only rollback artifacts + block: + - name: Install the public-safe Wazuh integration script + ansible.builtin.copy: + src: "{{ wazuh_integration_src }}" + dest: "{{ wazuh_integration_dest }}" + owner: root + group: wazuh + mode: "0750" + backup: true + register: wazuh_integration_copy + + - name: Install the bounded Wazuh Integrator configuration + ansible.builtin.blockinfile: + path: "{{ wazuh_ossec_conf }}" + insertbefore: "" + marker: "" + block: |2 + + custom-awoooi-alertmanager + {{ wazuh_hook_url }} + 10 + json + 10 + 3 + + backup: true + register: wazuh_config_update + no_log: true + + - name: Validate current Wazuh XML without returning configuration + ansible.builtin.command: + argv: + - /usr/bin/python3 + - -c + - >- + import xml.etree.ElementTree as ET; + ET.parse('{{ wazuh_ossec_conf }}') + check_mode: false + changed_when: false + no_log: true + + - name: Validate the installed integration script + ansible.builtin.command: + argv: + - "{{ wazuh_integration_dest }}" + - --self-test + when: not ansible_check_mode + changed_when: false + + - name: Restart Wazuh only when managed assets changed + ansible.builtin.systemd: + name: wazuh-manager.service + state: restarted + when: + - not ansible_check_mode + - wazuh_integration_copy.changed or wazuh_config_update.changed + + - name: Verify Wazuh manager after controlled convergence + ansible.builtin.command: + argv: + - /usr/bin/systemctl + - is-active + - wazuh-manager.service + register: wazuh_manager_after + retries: 6 + delay: 5 + until: wazuh_manager_after.rc == 0 + check_mode: false + changed_when: false + no_log: true + + - name: Verify Wazuh Integrator process after apply + ansible.builtin.command: + argv: + - /var/ossec/bin/wazuh-control + - status + register: wazuh_process_status + when: not ansible_check_mode + changed_when: false + failed_when: >- + 'wazuh-integratord is running' not in wazuh_process_status.stdout + no_log: true + + - name: Verify AWOOOI ingress reachability from the Wazuh manager + ansible.builtin.uri: + url: https://awoooi.wooo.work/api/v1/webhooks/health + method: GET + status_code: 200 + return_content: false + validate_certs: true + timeout: 10 + changed_when: false + + - name: Verify installed script fingerprint matches repo source + ansible.builtin.stat: + path: "{{ wazuh_integration_dest }}" + checksum_algorithm: sha256 + get_checksum: true + register: wazuh_installed_script + when: not ansible_check_mode + no_log: true + + - name: Assert installed script fingerprint parity + ansible.builtin.assert: + that: + - wazuh_installed_script.stat.checksum == (wazuh_source_sha256.stdout.split() | first) + when: not ansible_check_mode + no_log: true + + - name: Create the public-safe convergence receipt directory + ansible.builtin.file: + path: "{{ wazuh_receipt_dir }}" + state: directory + owner: root + group: root + mode: "0755" + + - name: Write the public-safe convergence receipt + ansible.builtin.copy: + dest: "{{ wazuh_receipt_path }}" + owner: root + group: root + mode: "0644" + backup: true + content: | + schema_version=awoooi_wazuh_alert_ingress_receipt_v1 + integration_name=custom-awoooi-alertmanager + script_sha256={{ wazuh_source_sha256.stdout.split() | first }} + config_present=1 + manager_active=1 + integrator_running=1 + ingress_reachable=1 + raw_payload_persisted=0 + secret_value_exposed=0 + active_response_enabled=0 + register: wazuh_receipt_update + + rescue: + - name: Restore the prior Wazuh configuration from remote backup + ansible.builtin.copy: + remote_src: true + src: "{{ wazuh_config_update.backup_file }}" + dest: "{{ wazuh_ossec_conf }}" + owner: root + group: wazuh + mode: "0660" + when: + - wazuh_config_update is defined + - wazuh_config_update.backup_file | default('') | length > 0 + no_log: true + + - name: Restore the prior integration script from remote backup + ansible.builtin.copy: + remote_src: true + src: "{{ wazuh_integration_copy.backup_file }}" + dest: "{{ wazuh_integration_dest }}" + owner: root + group: wazuh + mode: "0750" + when: + - wazuh_integration_before.stat.exists + - wazuh_integration_copy is defined + - wazuh_integration_copy.backup_file | default('') | length > 0 + no_log: true + + - name: Remove a newly-created integration script during rollback + ansible.builtin.file: + path: "{{ wazuh_integration_dest }}" + state: absent + when: not wazuh_integration_before.stat.exists + + - name: Restore the prior public-safe receipt from remote backup + ansible.builtin.copy: + remote_src: true + src: "{{ wazuh_receipt_update.backup_file }}" + dest: "{{ wazuh_receipt_path }}" + owner: root + group: root + mode: "0644" + when: + - wazuh_receipt_before.stat.exists + - wazuh_receipt_update is defined + - wazuh_receipt_update.backup_file | default('') | length > 0 + no_log: true + + - name: Remove a newly-created public-safe receipt during rollback + ansible.builtin.file: + path: "{{ wazuh_receipt_path }}" + state: absent + when: not wazuh_receipt_before.stat.exists + + - name: Restart Wazuh after rollback + ansible.builtin.systemd: + name: wazuh-manager.service + state: restarted + + - name: Stop with a bounded rollback terminal + ansible.builtin.fail: + msg: wazuh_alertmanager_integration_rolled_back diff --git a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml index 6eb7bfd10..0d45839b6 100644 --- a/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml +++ b/k8s/awoooi-prod/08-deployment-ansible-executor-broker.yaml @@ -89,6 +89,8 @@ spec: value: "/run/secrets/ssh_mcp_key" - name: AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH value: "/etc/ssh-mcp/known_hosts" + - name: AWOOOP_ANSIBLE_BECOME_PASSWORD_FILE + value: "/run/secrets/host112-become/password" volumeMounts: - name: ssh-mcp-key mountPath: /run/secrets/ssh_mcp_key @@ -98,6 +100,9 @@ spec: mountPath: /etc/ssh-mcp/known_hosts subPath: known_hosts readOnly: true + - name: host112-become-password + mountPath: /run/secrets/host112-become + readOnly: true resources: requests: cpu: "100m" @@ -136,3 +141,11 @@ spec: secret: secretName: ssh-mcp-key defaultMode: 0400 + - name: host112-become-password + secret: + secretName: awoooi-secrets + optional: true + defaultMode: 0400 + items: + - key: HOST112_ANSIBLE_BECOME_PASSWORD + path: password