wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(iwooos): add wazuh allowlisted dry-run gate

Browse Source
This commit is contained in:
Your Name
2026-06-28 19:10:37 +08:00
parent 19bc2d059b
commit 3c5ca49cbf
10 changed files with 1436 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

223
apps/api/tests/test_iwooos_runtime_security_readback.py
View File

@@ -7,6 +7,9 @@ from src.api.v1.iwooos import router
from src.services.iwooos_runtime_security_readback import (
load_latest_iwooos_runtime_security_readback,
)
from src.services.iwooos_wazuh_allowlisted_check_mode_dry_run import (
load_latest_iwooos_wazuh_allowlisted_check_mode_dry_run,
)
from src.services.iwooos_wazuh_runtime_controlled_apply_preflight import (
load_latest_iwooos_wazuh_runtime_controlled_apply_preflight,
)
@@ -98,13 +101,41 @@ def _valid_runtime_gate_owner_review_packet() -> dict[str, object]:
}
def _valid_allowlisted_check_mode_dry_run_packet() -> dict[str, object]:
return {
"dry_run_intent": "stage_allowlisted_check_mode_dry_run_readback_only",
"target_selector_aliases": [
"managed_core_node_a",
"managed_core_node_b",
"managed_core_node_c",
"managed_edge_node_a",
"managed_edge_node_b",
"managed_lab_node_a",
],
"check_mode_plan_ref": "playbooks/wazuh-allowlisted-check-mode#redacted-plan",
"dry_run_evidence_ref": "evidence/iwooos/wazuh-allowlisted-dry-run-redacted-v1",
"dry_run_result_ref": "evidence/iwooos/wazuh-allowlisted-dry-run-result-redacted-v1",
"dry_run_result_state": "staged_passed_no_runtime_action",
"dry_run_redaction_attestation": "redacted_refs_only_no_raw_output",
"post_dry_run_verifier_ref": "verifiers/iwooos-wazuh-post-dry-run-readback#public-safe",
"rollback_revalidation_ref": "playbooks/wazuh-controlled-apply-rollback#revalidated-redacted",
"km_playbook_writeback_ref": "km/playbook-trust/wazuh-allowlisted-check-mode-dry-run-v1",
"followup_owner": "iwooos-security-reviewer",
"audit_receipt_ref": "audit/iwooos-wazuh-allowlisted-check-mode-dry-run-redacted-v1",
"runtime_boundary_ack": "runtime_gate_remains_closed",
"live_wazuh_query_boundary_ack": "no_live_wazuh_query_performed",
"host_write_boundary_ack": "no_host_write_performed",
"secret_boundary_ack": "no_secret_value_collected",
}
def test_iwooos_runtime_security_readback_preserves_zero_runtime_gates() -> None:
payload = load_latest_iwooos_runtime_security_readback()
assert payload["schema_version"] == "iwooos_runtime_security_readback_v1"
assert payload["status"] == "blocked_waiting_owner_evidence_and_runtime_gates"
assert payload["summary"]["source_snapshot_count"] == 12
assert payload["summary"]["p0_lane_count"] == 11
assert payload["summary"]["source_snapshot_count"] == 13
assert payload["summary"]["p0_lane_count"] == 12
assert payload["summary"]["runtime_gate_count"] == 0
assert payload["summary"]["owner_response_received_count"] == 0
assert payload["summary"]["owner_response_accepted_count"] == 0
@@ -170,6 +201,40 @@ def test_iwooos_runtime_security_readback_preserves_zero_runtime_gates() -> None
)
assert payload["summary"]["wazuh_runtime_owner_review_packet_accepted_count"] == 1
assert payload["summary"]["wazuh_runtime_owner_review_runtime_gate_count"] == 0
assert (
payload["summary"][
"wazuh_allowlisted_check_mode_dry_run_target_selector_count"
]
== 6
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_check_mode_plan_count"]
== 1
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_evidence_ref_count"]
== 1
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_result_ref_count"]
== 1
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_packet_received_count"]
== 1
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_packet_accepted_count"]
== 1
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_post_verifier_count"]
== 1
)
assert (
payload["summary"]["wazuh_allowlisted_check_mode_dry_run_runtime_gate_count"]
== 0
)
assert payload["summary"]["kali_active_scan_authorized_count"] == 0
assert payload["summary"]["kali_execute_authorized_count"] == 0
assert payload["summary"]["alert_receipt_runtime_send_count"] == 0
@@ -190,6 +255,7 @@ def test_iwooos_runtime_security_readback_lanes_are_candidate_only() -> None:
"wazuh_owner_evidence_preflight",
"wazuh_runtime_controlled_apply_preflight",
"wazuh_runtime_gate_owner_review",
"wazuh_allowlisted_check_mode_dry_run",
"wazuh_dashboard_api",
"kali_intake",
"alert_readability",
@@ -218,6 +284,15 @@ def test_iwooos_runtime_security_readback_lanes_are_candidate_only() -> None:
)
for lane in payload["lanes"]
)
assert all(
lane["lane_id"] != "wazuh_allowlisted_check_mode_dry_run"
or (
lane["completion_percent"] == 65
and lane["metrics"]["packet_accepted"] == 1
and lane["metrics"]["runtime_gate"] == 0
)
for lane in payload["lanes"]
)
assert all(
lane["lane_id"] != "wazuh_owner_evidence_preflight"
or lane["metrics"]["owner_accepted"] == 0
@@ -802,3 +877,147 @@ def test_iwooos_wazuh_runtime_gate_owner_review_validator_rejects_runtime_action
assert data["runtime_action_rejected"] is True
assert data["summary"]["owner_review_runtime_action_rejected_count"] == 1
assert data["summary"]["runtime_gate_count"] == 0
def test_iwooos_wazuh_allowlisted_check_mode_dry_run_api_is_public_safe(
monkeypatch,
) -> None:
monkeypatch.delenv("IWOOOS_WAZUH_READONLY_ENABLED", raising=False)
monkeypatch.delenv("WAZUH_API_BASE_URL", raising=False)
monkeypatch.delenv("WAZUH_API_USERNAME", raising=False)
monkeypatch.delenv("WAZUH_API_PASSWORD", raising=False)
payload = load_latest_iwooos_wazuh_allowlisted_check_mode_dry_run()
assert (
payload["schema_version"]
== "iwooos_wazuh_allowlisted_check_mode_dry_run_readback_v1"
)
assert payload["status"] == "allowlisted_check_mode_dry_run_staged_no_runtime_action"
assert payload["summary"]["dry_run_packet_accepted_count"] == 1
assert payload["summary"]["allowlisted_target_selector_count"] == 6
assert payload["summary"]["runtime_gate_count"] == 0
assert payload["boundaries"]["runtime_execution_authorized"] is False
response = _client().get("/api/v1/iwooos/wazuh-allowlisted-check-mode-dry-run")
assert response.status_code == 200
data = response.json()
assert (
data["schema_version"]
== "iwooos_wazuh_allowlisted_check_mode_dry_run_readback_v1"
)
assert data["summary"]["dry_run_packet_received_count"] == 1
assert data["summary"]["dry_run_packet_review_ready_count"] == 1
assert data["summary"]["dry_run_packet_accepted_count"] == 1
assert data["summary"]["runtime_gate_count"] == 0
assert data["summary"]["wazuh_api_live_query_authorized_count"] == 0
assert data["summary"]["wazuh_active_response_authorized_count"] == 0
assert data["summary"]["host_write_authorized_count"] == 0
assert data["summary"]["secret_value_collection_allowed_count"] == 0
assert data["boundaries"]["payload_persisted"] is False
assert data["boundaries"]["wazuh_api_live_query_authorized"] is False
assert data["boundaries"]["wazuh_active_response_authorized"] is False
assert data["boundaries"]["host_write_authorized"] is False
assert data["boundaries"]["runtime_gate_open"] is False
assert data["boundaries"]["not_authorization"] is True
assert len(data["target_selectors"]) == 6
assert len(data["dry_run_items"]) == 6
assert any(
marker == "wazuh_allowlisted_check_mode_dry_run_validation_api_available=true"
for marker in data["boundary_markers"]
)
assert "192.168.0." not in response.text
assert "工作視窗" not in response.text
assert "批准！繼續" not in response.text
assert "WAZUH_API_PASSWORD" not in response.text
def test_iwooos_wazuh_allowlisted_check_mode_dry_run_validator_accepts_redacted_packet(
monkeypatch,
) -> None:
monkeypatch.delenv("IWOOOS_WAZUH_READONLY_ENABLED", raising=False)
monkeypatch.delenv("WAZUH_API_BASE_URL", raising=False)
monkeypatch.delenv("WAZUH_API_USERNAME", raising=False)
monkeypatch.delenv("WAZUH_API_PASSWORD", raising=False)
client = _client()
before = client.get("/api/v1/iwooos/wazuh-allowlisted-check-mode-dry-run").json()
response = client.post(
"/api/v1/iwooos/wazuh-allowlisted-check-mode-dry-run/validate-dry-run-packet",
json=_valid_allowlisted_check_mode_dry_run_packet(),
)
after = client.get("/api/v1/iwooos/wazuh-allowlisted-check-mode-dry-run").json()
assert response.status_code == 200
data = response.json()
assert (
data["schema_version"]
== "iwooos_wazuh_allowlisted_check_mode_dry_run_validation_result_v1"
)
assert data["status"] == "accepted_for_allowlisted_check_mode_dry_run_readback_only"
assert data["accepted_for_allowlisted_check_mode_dry_run_readback_only"] is True
assert data["summary"]["dry_run_packet_received_count"] == 1
assert data["summary"]["dry_run_packet_review_ready_count"] == 1
assert data["summary"]["dry_run_packet_accepted_count"] == 1
assert data["summary"]["runtime_gate_count"] == 0
assert data["summary"]["wazuh_api_live_query_authorized_count"] == 0
assert data["summary"]["wazuh_active_response_authorized_count"] == 0
assert data["summary"]["host_write_authorized_count"] == 0
assert data["summary"]["secret_value_collection_allowed_count"] == 0
assert data["boundaries"]["payload_persisted"] is False
assert data["boundaries"]["runtime_execution_authorized"] is False
assert data["boundaries"]["runtime_gate_open"] is False
assert before["summary"] == after["summary"]
assert "192.168.0." not in response.text
assert "工作視窗" not in response.text
assert "批准！繼續" not in response.text
def test_iwooos_wazuh_allowlisted_check_mode_dry_run_validator_quarantines_sensitive_payload(
monkeypatch,
) -> None:
monkeypatch.delenv("IWOOOS_WAZUH_READONLY_ENABLED", raising=False)
monkeypatch.delenv("WAZUH_API_BASE_URL", raising=False)
monkeypatch.delenv("WAZUH_API_USERNAME", raising=False)
monkeypatch.delenv("WAZUH_API_PASSWORD", raising=False)
packet = _valid_allowlisted_check_mode_dry_run_packet()
packet[
"redacted_evidence_ref"
] = "dry-run raw output includes 10.1.2.3 and Authorization: Bearer abcdefghijklmnop"
response = _client().post(
"/api/v1/iwooos/wazuh-allowlisted-check-mode-dry-run/validate-dry-run-packet",
json=packet,
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "quarantine_sensitive_payload"
assert data["quarantined"] is True
assert data["summary"]["dry_run_packet_quarantined_count"] == 1
assert data["summary"]["runtime_gate_count"] == 0
assert "10.1.2.3" not in response.text
assert "Bearer abcdefghijklmnop" not in response.text
def test_iwooos_wazuh_allowlisted_check_mode_dry_run_validator_rejects_runtime_action(
monkeypatch,
) -> None:
monkeypatch.delenv("IWOOOS_WAZUH_READONLY_ENABLED", raising=False)
monkeypatch.delenv("WAZUH_API_BASE_URL", raising=False)
monkeypatch.delenv("WAZUH_API_USERNAME", raising=False)
monkeypatch.delenv("WAZUH_API_PASSWORD", raising=False)
packet = _valid_allowlisted_check_mode_dry_run_packet()
packet["wazuh_active_response"] = True
response = _client().post(
"/api/v1/iwooos/wazuh-allowlisted-check-mode-dry-run/validate-dry-run-packet",
json=packet,
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "reject_runtime_action_request"
assert data["runtime_action_rejected"] is True
assert data["summary"]["dry_run_runtime_action_rejected_count"] == 1
assert data["summary"]["runtime_gate_count"] == 0