1c21c7acc8
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m31s
CD Pipeline / build-and-deploy (push) Successful in 16m46s
CD Pipeline / post-deploy-checks (push) Successful in 2m10s
419 lines
14 KiB
Python
419 lines
14 KiB
Python
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"
|
|
)
|
|
WORKER_DEPLOYMENT = ROOT / "k8s" / "awoooi-prod" / "08-deployment-worker.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"
|
|
)
|
|
|
|
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),
|
|
)
|
|
)
|
|
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
|
|
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),
|
|
)
|
|
|
|
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()
|
|
|
|
|
|
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")
|