awoooi/apps/api/tests/test_iwooos_runtime_security_readback.py

from __future__ import annotations

from fastapi import FastAPI
from fastapi.testclient import TestClient

from src.api.v1.iwooos import router
from src.services.iwooos_runtime_security_readback import (
    load_latest_iwooos_runtime_security_readback,
)


def _client() -> TestClient:
    app = FastAPI()
    app.include_router(router)
    return TestClient(app)


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"] == 8
    assert payload["summary"]["p0_lane_count"] == 7
    assert payload["summary"]["runtime_gate_count"] == 0
    assert payload["summary"]["owner_response_received_count"] == 0
    assert payload["summary"]["owner_response_accepted_count"] == 0
    assert payload["summary"]["wazuh_manager_registry_accepted_count"] == 0
    assert payload["summary"]["wazuh_live_status"] == "not_checked_by_snapshot_loader"
    assert payload["summary"]["wazuh_live_route_degraded_count"] == 1
    assert payload["summary"]["wazuh_live_readonly_api_enabled_count"] == 0
    assert payload["summary"]["wazuh_live_metadata_available_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
    assert payload["boundaries"]["runtime_execution_authorized"] is False
    assert payload["boundaries"]["active_scan_authorized"] is False
    assert payload["boundaries"]["wazuh_active_response_authorized"] is False
    assert payload["boundaries"]["telegram_send_authorized"] is False


def test_iwooos_runtime_security_readback_lanes_are_candidate_only() -> None:
    payload = load_latest_iwooos_runtime_security_readback()

    lane_ids = {lane["lane_id"] for lane in payload["lanes"]}
    assert lane_ids == {
        "wazuh_registry",
        "wazuh_live_route",
        "wazuh_dashboard_api",
        "kali_intake",
        "alert_readability",
        "owner_dispatch",
        "intrusion_prevention",
    }
    assert all(lane["metrics"] for lane in payload["lanes"])
    assert all(lane["next_gate"] for lane in payload["lanes"])
    assert all(lane["source_refs"] for lane in payload["lanes"])
    assert any(lane["completion_percent"] > 0 for lane in payload["lanes"])
    assert all(lane["lane_id"] != "wazuh_registry" or lane["completion_percent"] == 0 for lane in payload["lanes"])
    assert all(lane["lane_id"] != "wazuh_live_route" or lane["metrics"]["route_degraded"] == 1 for lane in payload["lanes"])


def test_iwooos_runtime_security_readback_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)

    response = _client().get("/api/v1/iwooos/runtime-security-readback")

    assert response.status_code == 200
    data = response.json()
    assert data["schema_version"] == "iwooos_runtime_security_readback_v1"
    assert data["summary"]["runtime_gate_count"] == 0
    assert data["summary"]["wazuh_live_status"] == "disabled_waiting_iwooos_wazuh_owner_gate"
    assert data["summary"]["wazuh_live_route_http_status"] == 200
    assert data["summary"]["wazuh_live_route_degraded_count"] == 1
    assert data["summary"]["wazuh_live_metadata_available_count"] == 0
    assert data["boundaries"]["secret_value_collection_allowed"] is False
    assert "192.168.0." not in response.text
    assert "工作視窗" not in response.text
    assert "批准！繼續" not in response.text


def test_iwooos_runtime_security_readback_api_includes_live_wazuh_empty_registry(monkeypatch) -> None:
    import httpx

    monkeypatch.setenv("IWOOOS_WAZUH_READONLY_ENABLED", "true")
    monkeypatch.setenv("WAZUH_API_BASE_URL", "https://wazuh.example.test:55000")
    monkeypatch.setenv("WAZUH_API_USERNAME", "readonly")
    monkeypatch.setenv("WAZUH_API_PASSWORD", "placeholder")
    monkeypatch.setenv("IWOOOS_WAZUH_EXPECTED_MIN_AGENT_COUNT", "6")

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/security/user/authenticate":
            return httpx.Response(200, json={"data": {"token": "token-value"}})
        if request.url.path == "/agents/summary/status":
            return httpx.Response(
                200,
                json={"data": {"connection": {"total": 0, "active": 0, "disconnected": 0, "pending": 0}}},
            )
        if request.url.path == "/agents":
            return httpx.Response(200, json={"data": {"affected_items": []}})
        return httpx.Response(404)

    transport = httpx.MockTransport(handler)
    original_async_client = httpx.AsyncClient

    def client_factory(*args, **kwargs):
        kwargs["transport"] = transport
        return original_async_client(*args, **kwargs)

    monkeypatch.setattr(httpx, "AsyncClient", client_factory)

    response = _client().get("/api/v1/iwooos/runtime-security-readback")

    assert response.status_code == 200
    data = response.json()
    assert data["summary"]["wazuh_live_status"] == "wazuh_agent_registry_empty"
    assert data["summary"]["wazuh_live_readonly_api_enabled_count"] == 1
    assert data["summary"]["wazuh_live_agent_total"] == 0
    assert data["summary"]["wazuh_live_registry_empty_count"] == 1
    assert data["summary"]["wazuh_live_route_degraded_count"] == 1
    assert data["summary"]["runtime_gate_count"] == 0
    assert "token-value" not in response.text