fix(iwooos): restore wazuh manager posture access
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 2m34s
CD Pipeline / build-and-deploy (push) Successful in 6m32s
CD Pipeline / post-deploy-checks (push) Successful in 1m49s

This commit is contained in:
ogt
2026-07-11 11:13:19 +08:00
parent 39f064ec3c
commit ffe1218d8c
5 changed files with 50 additions and 19 deletions

View File

@@ -46,10 +46,10 @@ ProbeRunner = Callable[..., Awaitable[ReadOnlyProbeResult]]
_POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = {
"ansible:wazuh-manager-posture-readback": (
AssetPostcondition(
"host_112_wazuh_manager_control_binary",
"host_112_wazuh_manager_unit_loaded",
"security",
"host_112",
"test -f /var/ossec/bin/wazuh-control",
"systemctl show --property=LoadState --value wazuh-manager.service | grep -Fxq loaded",
),
AssetPostcondition(
"host_112_wazuh_manager_runtime",

View File

@@ -7,6 +7,7 @@ from pathlib import Path
from unittest.mock import AsyncMock, MagicMock
import pytest
import yaml
from fastapi import FastAPI
from fastapi.testclient import TestClient
@@ -158,14 +159,14 @@ def test_wazuh_posture_catalog_and_playbook_are_bounded_no_write() -> None:
"ansible:wazuh-manager-posture-readback"
)
assert {condition.condition_id for condition in postconditions} == {
"host_112_wazuh_manager_control_binary",
"host_112_wazuh_manager_unit_loaded",
"host_112_wazuh_manager_runtime",
}
assert {condition.inventory_host for condition in postconditions} == {
"host_112"
}
assert {condition.probe for condition in postconditions} == {
"test -f /var/ossec/bin/wazuh-control",
"systemctl show --property=LoadState --value wazuh-manager.service | grep -Fxq loaded",
"systemctl is-active --quiet wazuh-manager.service",
}
@@ -187,6 +188,25 @@ def test_wazuh_posture_catalog_and_playbook_are_bounded_no_write() -> None:
):
assert forbidden not in source
inventory = yaml.safe_load(
(repo_root / "infra/ansible/inventory/hosts.yml").read_text()
)
host_112 = inventory["all"]["children"]["security"]["hosts"]["host_112"]
assert host_112["ansible_user"] == "kali"
assert "ansible_skip_tags" not in host_112
service_registry = yaml.safe_load(
(repo_root / "ops/monitoring/service-registry.yaml").read_text()
)
wazuh_manager = next(
node
for node in service_registry["nodes"]
if node.get("name") == "wazuh-manager"
)
assert wazuh_manager["ip"] == "192.168.0.112"
assert wazuh_manager["port"] == 55000
assert wazuh_manager["criticality"] == "P0"
@pytest.mark.asyncio
async def test_wazuh_posture_scheduler_enqueues_after_mcp_and_log_evidence(