Merge remote-tracking branch 'origin/main' into codex/sre-typed-automation-20260715

# Conflicts:
#	apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py
This commit is contained in:
ogt
2026-07-15 17:45:18 +08:00
6 changed files with 150 additions and 11 deletions

View File

@@ -2,6 +2,7 @@ from __future__ import annotations
# ruff: noqa: E402, I001
import json
import os
from contextlib import asynccontextmanager
from datetime import UTC, datetime
@@ -272,6 +273,78 @@ def test_wazuh_runtime_readback_classifies_failed_check_for_bounded_retry() -> N
assert "ansible_target_unreachable" in payload["active_blockers"]
def test_wazuh_runtime_readback_projects_allowlisted_privilege_blocker() -> None:
run_id = "00000000-0000-0000-0000-000000000312"
recurrence = _current_wazuh_source_recurrence(
"2026-07-15T05:59:58+08:00"
)
row = {
"candidate_op_id": run_id,
"automation_run_id": run_id,
"trace_id": run_id,
"run_id": run_id,
"candidate_project_id": "awoooi",
"work_item_id": "P0-03-WAZUH-ALERT-INGRESS",
"proposal_source": "iwooos_wazuh_alert_ingress_scheduler",
"run_namespace": "awoooi:iwooos:wazuh-alert-ingress",
"source_receipt_ref": "wazuh-alert:2026071500",
"candidate_source_recurrence": recurrence,
"candidate_source_recurrence_verified": True,
"catalog_id": "ansible:wazuh-alertmanager-integration",
"check_op_id": "00000000-0000-0000-0000-000000000313",
"check_status": "failed",
"check_returncode": "2",
"check_timed_out": "false",
"check_automation_run_id": run_id,
"check_trace_id": run_id,
"check_run_id": run_id,
"check_work_item_id": "P0-03-WAZUH-ALERT-INGRESS",
"check_source_recurrence": recurrence,
"check_failure_marker": (
"wazuh_privileged_convergence_capability_missing"
),
"stage_ids": [],
}
payload = build_iwooos_wazuh_controlled_executor_runtime_readback(
row,
executor_enabled=True,
)
assert payload["check_failure_class"] == (
"wazuh_privileged_convergence_capability_missing"
)
assert payload["next_safe_action"] == (
"queue_critical_break_glass_privileged_convergence_capability_"
"then_bounded_retry"
)
assert "wazuh_privileged_convergence_capability_missing" in (
payload["active_blockers"]
)
assert payload["boundaries"]["host_write_performed"] is False
assert payload["boundaries"]["secret_value_collection_allowed"] is False
def test_wazuh_runtime_readback_rejects_untrusted_failure_marker() -> None:
row = {
"candidate_op_id": "00000000-0000-0000-0000-000000000313",
"check_status": "failed",
"check_returncode": "2",
"check_failure_marker": "untrusted_raw_failure_text",
"stage_ids": [],
}
payload = build_iwooos_wazuh_controlled_executor_runtime_readback(
row,
executor_enabled=True,
)
assert payload["check_failure_class"] == "ansible_check_mode_returncode_2"
assert "untrusted_raw_failure_text" not in str(payload)
assert payload["summary"]["runtime_closed_count"] == 0
assert payload["boundaries"]["host_write_performed"] is False
def test_wazuh_posture_catalog_and_playbook_are_bounded_no_write() -> None:
catalog = get_ansible_catalog_item("ansible:wazuh-manager-posture-readback")
assert catalog is not None
@@ -975,6 +1048,10 @@ def test_iwooos_runtime_workspace_defaults_to_live_closure_without_legacy_text_w
).read_text(encoding="utf-8")
assert "useState<IwoooSWorkspaceView>('runtime')" in source
assert "wazuh_privileged_convergence_capability_missing" in source
assert "privilegedCapabilityMissing" in source
assert "ansible:wazuh-alertmanager-integration" in source
assert "labelKey: runtimeSummary?.selected_catalog_id" in source
runtime_panel = source.split(
"<IwoooSWorkspacePanel active={workspaceView === 'runtime'}>", 1
)[1].split("</IwoooSWorkspacePanel>", 1)[0]
@@ -1000,3 +1077,22 @@ def test_iwooos_runtime_workspace_defaults_to_live_closure_without_legacy_text_w
)
assert all(board not in runtime_panel for board in legacy_boards)
assert all(board in ledger_panel for board in legacy_boards)
runtime_failure_keys = None
for locale in ("zh-TW", "en"):
messages = json.loads(
(repo_root / f"apps/web/messages/{locale}.json").read_text(
encoding="utf-8"
)
)
locale_keys = set(
messages["iwooos"]["securityToolClosure"]["runtimeFailure"]
)
assert "privilegedCapabilityMissing" in locale_keys
if runtime_failure_keys is None:
runtime_failure_keys = locale_keys
else:
assert locale_keys == runtime_failure_keys
runway = messages["iwooos"]["securityToolClosure"]["runway"]
assert "ingress" in runway