Files
awoooi/apps/api/tests/test_iwooos_security_tool_closure_readback.py

182 lines
7.5 KiB
Python

from __future__ import annotations
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.v1.iwooos import router
from src.services.iwooos_security_tool_closure_readback import (
load_latest_iwooos_security_tool_closure_readback,
)
def _client() -> TestClient:
app = FastAPI()
app.include_router(router)
return TestClient(app)
def test_iwooos_security_tool_closure_readback_rolls_up_tool_tracks() -> None:
payload = load_latest_iwooos_security_tool_closure_readback()
assert payload["schema_version"] == "iwooos_security_tool_closure_readback_v2"
assert payload["status"] == "tool_source_readiness_available_runtime_closure_open"
assert payload["summary"]["source_readback_count"] == 8
assert payload["summary"]["tool_track_count"] == 8
assert payload["summary"]["p0_tool_track_count"] == 3
assert payload["summary"]["wazuh_expected_host_scope_count"] == 6
assert payload["summary"]["wazuh_manager_registry_accepted_count"] == 6
assert payload["summary"]["wazuh_registry_complete_count"] == 1
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
assert payload["summary"]["closed_tool_track_count"] == 0
assert payload["summary"]["tool_runtime_closure_percent"] == 0
assert payload["summary"]["tool_closure_percent"] == 0
assert payload["summary"]["tool_source_readiness_percent"] > 0
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_count"]
== 0
)
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_signal_count"]
== 5
)
assert payload["summary"]["supply_chain_scan_closed_count"] == 0
assert payload["summary"]["runtime_detection_closed_count"] == 0
assert payload["summary"]["web_api_dast_closed_count"] == 0
assert payload["summary"]["controlled_apply_policy_active_count"] == 1
assert payload["summary"]["runtime_execution_performed_count"] == 0
track_ids = {track["track_id"] for track in payload["tool_tracks"]}
assert track_ids == {
"wazuh_detection_response",
"supply_chain_sbom_scan",
"source_control_scorecard",
"k8s_policy_attestation",
"runtime_threat_detection",
"web_api_dast",
"normalized_detection_schema",
"ai_agent_controlled_apply",
}
def test_iwooos_security_tool_closure_readback_uses_controlled_apply_policy() -> None:
payload = load_latest_iwooos_security_tool_closure_readback()
assert payload["boundaries"]["runtime_execution_performed"] is False
assert payload["boundaries"]["controlled_apply_policy_active"] is True
assert payload["boundaries"]["low_risk_controlled_apply_allowed"] is True
assert payload["boundaries"]["medium_risk_controlled_apply_allowed"] is True
assert payload["boundaries"]["high_risk_controlled_apply_allowed"] is True
assert payload["boundaries"]["critical_break_glass_required"] is True
assert (
payload["boundaries"]["owner_review_required_for_low_medium_high"]
is False
)
assert payload["boundaries"]["secret_value_collection_allowed"] is False
assert payload["boundaries"]["readback_endpoint_is_executor"] is False
assert all(
track["controlled_apply_policy_allowed"] is True
for track in payload["tool_tracks"]
)
assert all(
track["runtime_execution_performed"] is False
for track in payload["tool_tracks"]
)
assert all(track["owner_review_required"] is False for track in payload["tool_tracks"])
assert all(
track["secret_value_collection_allowed"] is False
for track in payload["tool_tracks"]
)
def test_iwooos_security_tool_closure_readback_marks_partial_vs_missing_tracks() -> None:
payload = load_latest_iwooos_security_tool_closure_readback()
tracks = {track["track_id"]: track for track in payload["tool_tracks"]}
assert tracks["wazuh_detection_response"]["closure_state"] == (
"partial_source_readiness_runtime_open"
)
assert tracks["wazuh_detection_response"]["ready_signal_count"] == 5
assert tracks["wazuh_detection_response"]["missing_signal_count"] == 1
assert tracks["supply_chain_sbom_scan"]["closure_state"] == (
"source_missing_runtime_open"
)
assert tracks["runtime_threat_detection"]["tool_ids"] == ["falco"]
assert tracks["web_api_dast"]["tool_ids"] == ["owasp_zap_or_equivalent_dast"]
assert tracks["ai_agent_controlled_apply"]["ready_signal_count"] > 0
assert tracks["ai_agent_controlled_apply"]["missing_signal_count"] >= 0
def test_iwooos_security_tool_closure_readback_accepts_alert_receipt_signal() -> None:
payload = load_latest_iwooos_security_tool_closure_readback(
alert_receipt_readback={
"events": [{"provider_event_id": "alertmanager:received:tool-closure"}],
"total": 1,
"limit": 20,
"source_status": "ready",
}
)
tracks = {track["track_id"]: track for track in payload["tool_tracks"]}
assert payload["summary"]["alert_receipt_accepted_count"] == 1
assert payload["summary"]["alert_receipt_observed_count"] == 1
assert payload["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_verifier_ready_count"]
== 1
)
assert (
payload["summary"][
"wazuh_fim_vulnerability_alert_source_verifier_ready_count"
]
== 1
)
assert (
payload["summary"]["wazuh_fim_vulnerability_alert_runtime_closed_count"]
== 0
)
assert tracks["wazuh_detection_response"]["ready_signal_count"] == 6
assert tracks["wazuh_detection_response"]["missing_signal_count"] == 0
assert tracks["wazuh_detection_response"]["closure_state"] == (
"source_ready_runtime_open"
)
assert tracks["wazuh_detection_response"]["source_readiness_percent"] == 100
assert tracks["wazuh_detection_response"]["runtime_closed"] is False
assert tracks["wazuh_detection_response"]["execution_candidate_ready"] is True
assert tracks["normalized_detection_schema"]["ready_signal_count"] == 1
assert payload["summary"]["controlled_apply_policy_active_count"] == 1
def test_iwooos_security_tool_closure_readback_api_is_public_safe(monkeypatch) -> None:
async def fake_recent_events(**_: object) -> dict[str, object]:
return {
"events": [{"provider_event_id": "alertmanager:received:tool-closure"}],
"total": 1,
"limit": 20,
"source_status": "ready",
}
monkeypatch.setattr(
"src.api.v1.iwooos.list_recent_channel_events",
fake_recent_events,
)
response = _client().get("/api/v1/iwooos/security-tool-closure-readback")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "iwooos_security_tool_closure_readback_v2"
assert data["summary"]["tool_track_count"] == 8
assert data["summary"]["alert_receipt_accepted_count"] == 1
assert data["summary"]["wazuh_fim_vulnerability_alert_closed_count"] == 0
assert data["summary"]["tool_runtime_closure_percent"] == 0
assert data["summary"]["controlled_apply_policy_active_count"] == 1
assert data["boundaries"]["controlled_apply_policy_active"] is True
assert any(
item["track_id"] == "supply_chain_sbom_scan"
for item in data["tool_tracks"]
)
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