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

Merge remote-tracking branch 'gitea/main' into codex/delivery-workbench-release-20260626-ffsync

Browse Source
This commit is contained in:
ogt
2026-06-26 23:43:46 +08:00
parent 2d8d12e750 88b63f4d1d
commit e359d65feb
15 changed files with 684 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
apps/api/src/api/v1/iwooos.py
View File

@@ -9,12 +9,8 @@ from __future__ import annotations
import asyncio
import json
import os
from base64 import b64encode
from typing import Any
from urllib.parse import urljoin, urlparse
import httpx
from fastapi import APIRouter, HTTPException, status
from fastapi.responses import JSONResponse
@@ -24,180 +20,18 @@ from src.services.iwooos_runtime_security_readback import (
from src.services.iwooos_security_control_coverage import (
load_latest_iwooos_security_control_coverage,
)
from src.services.iwooos_wazuh_readonly_status import (
load_iwooos_wazuh_readonly_status,
)
from src.services.public_redaction import redact_public_lan_topology
router = APIRouter(tags=["IwoooS Security"])
REQUEST_TIMEOUT_SECONDS = 5.0
def _wazuh_env() -> dict[str, str]:
return {
"enabled": os.getenv("IWOOOS_WAZUH_READONLY_ENABLED", "").strip().lower(),
"base_url": os.getenv("WAZUH_API_BASE_URL", "").strip(),
"username": os.getenv("WAZUH_API_USERNAME", "").strip(),
"password": os.getenv("WAZUH_API_PASSWORD", "").strip(),
"expected_min_agent_count": os.getenv("IWOOOS_WAZUH_EXPECTED_MIN_AGENT_COUNT", "").strip(),
}
def _expected_min_agent_count(value: str) -> int:
try:
return max(0, int(value))
except ValueError:
return 0
def _https_url(value: str) -> str | None:
parsed = urlparse(value)
if parsed.scheme != "https" or not parsed.netloc:
return None
return value.rstrip("/") + "/"
def _boundary_response(status_text: str, http_status: int = 200) -> JSONResponse:
return JSONResponse(
status_code=http_status,
content={
"schema_version": "iwooos_wazuh_readonly_status_v1",
"status": status_text,
"mode": "metadata_only_no_active_response_no_raw_payload",
"configured": False,
"summary": {
"wazuh_platform_reported_count": 1,
"readonly_api_enabled_count": 0,
"wazuh_manager_query_accepted_count": 0,
"wazuh_event_accepted_count": 0,
"host_forensics_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
"expected_min_agent_count": _expected_min_agent_count(_wazuh_env()["expected_min_agent_count"]),
"agent_registry_empty_count": 0,
"agent_below_expected_minimum_count": 0,
"agent_visibility_no_false_green_count": 1,
},
"boundaries": _boundaries(),
},
)
def _boundaries() -> dict[str, bool]:
return {
"active_response_authorized": False,
"host_write_authorized": False,
"secret_value_collection_allowed": False,
"raw_wazuh_payload_storage_allowed": False,
"agent_identity_public_display_allowed": False,
"internal_ip_public_display_allowed": False,
"not_authorization": True,
}
def _redacted_agent(agent: dict[str, Any], index: int) -> dict[str, Any]:
os_info = agent.get("os") if isinstance(agent.get("os"), dict) else {}
return {
"alias": f"agent-{index + 1:02d}",
"status": agent.get("status", "unknown"),
"os": os_info.get("platform") or os_info.get("name") or "unknown",
"last_seen_present": bool(agent.get("lastKeepAlive")),
}
def _int_or_default(value: Any, default: int) -> int:
return value if isinstance(value, int) else default
def _agent_visibility_status(agent_total: int, expected_min_agent_count: int) -> str:
if agent_total <= 0:
return "wazuh_agent_registry_empty"
if expected_min_agent_count > 0 and agent_total < expected_min_agent_count:
return "wazuh_agent_registry_below_expected"
return "readonly_metadata_available"
async def _fetch_json(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> dict[str, Any]:
response = await client.get(url, headers=headers)
response.raise_for_status()
payload = response.json()
return payload if isinstance(payload, dict) else {}
async def _wazuh_readonly_status() -> JSONResponse:
env = _wazuh_env()
if env["enabled"] != "true":
return _boundary_response("disabled_waiting_iwooos_wazuh_owner_gate")
base_url = _https_url(env["base_url"])
if not base_url or not env["username"] or not env["password"]:
return _boundary_response("misconfigured_missing_server_side_wazuh_env", 503)
try:
auth_header = b64encode(f"{env['username']}:{env['password']}".encode("utf-8")).decode("ascii")
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS) as client:
auth = await _fetch_json(
client,
urljoin(base_url, "security/user/authenticate"),
{"Authorization": f"Basic {auth_header}"},
)
token = (auth.get("data") or {}).get("token")
if not token:
return _boundary_response("wazuh_auth_token_missing", 502)
bearer_headers = {"Authorization": f"Bearer {token}"}
status_payload = await _fetch_json(
client,
urljoin(base_url, "agents/summary/status"),
bearer_headers,
)
agents_payload = await _fetch_json(
client,
urljoin(base_url, "agents?limit=100&select=id,status,os.name,os.platform,lastKeepAlive"),
bearer_headers,
)
except (httpx.HTTPError, ValueError):
return _boundary_response("wazuh_readonly_metadata_unavailable", 502)
connection = ((status_payload.get("data") or {}).get("connection") or {})
affected_items = ((agents_payload.get("data") or {}).get("affected_items") or [])
if not isinstance(affected_items, list):
affected_items = []
expected_min_agent_count = _expected_min_agent_count(env["expected_min_agent_count"])
agent_total = _int_or_default(connection.get("total"), len(affected_items))
agent_active = _int_or_default(connection.get("active"), 0)
agent_disconnected = _int_or_default(connection.get("disconnected"), 0)
agent_pending = _int_or_default(connection.get("pending"), 0)
agent_registry_empty = agent_total <= 0
agent_below_expected = expected_min_agent_count > 0 and agent_total < expected_min_agent_count
return JSONResponse(
content={
"schema_version": "iwooos_wazuh_readonly_status_v1",
"status": _agent_visibility_status(agent_total, expected_min_agent_count),
"mode": "metadata_only_no_active_response_no_raw_payload",
"configured": True,
"summary": {
"wazuh_platform_reported_count": 1,
"readonly_api_enabled_count": 1,
"agent_total": agent_total,
"agent_active": agent_active,
"agent_disconnected": agent_disconnected,
"agent_pending": agent_pending,
"expected_min_agent_count": expected_min_agent_count,
"agent_registry_empty_count": 1 if agent_registry_empty else 0,
"agent_below_expected_minimum_count": 1 if agent_below_expected else 0,
"agent_visibility_no_false_green_count": 1,
"wazuh_manager_query_accepted_count": 0,
"wazuh_event_accepted_count": 0,
"host_forensics_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
},
"agents": [_redacted_agent(agent, index) for index, agent in enumerate(affected_items[:20])],
"boundaries": _boundaries(),
},
)
result = await load_iwooos_wazuh_readonly_status()
return JSONResponse(status_code=result.http_status, content=result.payload)
@router.get("/api/iwooos/wazuh")
@@ -216,14 +50,20 @@ async def get_iwooos_wazuh_readonly_status_v1() -> JSONResponse:
summary="取得 IwoooS runtime security readback",
description=(
"讀取最新已提交的 IwoooS 資安只讀快照，彙總 Wazuh、Kali、SOC/SIEM、"
"告警可讀性、owner dispatch 與外部入侵防護 Gate。此端點不呼叫 Wazuh / Kali / "
"主機 / Docker / Nginx / firewall / Telegram，不收集 secret，不授權 runtime 寫入。"
"告警可讀性、owner dispatch 與外部入侵防護 Gate，並附上 Wazuh 只讀路由的"
"公開安全 aggregate 讀回。此端點不呼叫 Kali / 主機 / Docker / Nginx / firewall / "
"Telegram，不保存 raw Wazuh payload，不收集 secret，不授權 runtime 寫入。"
),
)
async def get_iwooos_runtime_security_readback() -> dict[str, Any]:
"""回傳 IwoooS 資安 runtime readback 只讀總板。"""
try:
payload = await asyncio.to_thread(load_latest_iwooos_runtime_security_readback)
wazuh_result = await load_iwooos_wazuh_readonly_status()
payload = await asyncio.to_thread(
load_latest_iwooos_runtime_security_readback,
wazuh_live_status=wazuh_result.payload,
wazuh_live_http_status=wazuh_result.http_status,
)
return redact_public_lan_topology(payload)
except FileNotFoundError as exc:
raise HTTPException(

62
apps/api/src/services/iwooos_runtime_security_readback.py
View File

@@ -60,6 +60,8 @@ _FALSE_BOUNDARY_KEYS = {
def load_latest_iwooos_runtime_security_readback(
security_dir: Path | None = None,
wazuh_live_status: dict[str, Any] | None = None,
wazuh_live_http_status: int = 0,
) -> dict[str, Any]:
"""Load and normalize the current IwoooS runtime security readback."""
directory = security_dir or _DEFAULT_SECURITY_DIR
@@ -72,6 +74,7 @@ def load_latest_iwooos_runtime_security_readback(
alert_summary = _summary(snapshots["alert_readability"])
dispatch_summary = _summary(snapshots["owner_dispatch"])
intrusion_summary = _summary(snapshots["intrusion_prevention"])
live_wazuh = _wazuh_live_summary(wazuh_live_status, wazuh_live_http_status)
source_refs = [f"docs/security/{filename}" for filename in _SNAPSHOT_FILES.values()]
runtime_gate_count = _max_summary_count(
@@ -85,11 +88,11 @@ def load_latest_iwooos_runtime_security_readback(
return {
"schema_version": "iwooos_runtime_security_readback_v1",
"status": "blocked_waiting_owner_evidence_and_runtime_gates",
"mode": "committed_snapshot_readback_only_no_runtime_query",
"mode": "committed_snapshot_readback_with_public_safe_wazuh_route_metadata",
"source_refs": source_refs,
"summary": {
"source_snapshot_count": len(source_refs),
"p0_lane_count": 6,
"p0_lane_count": 7,
"control_plane_visibility_percent": _average_percent(
soc_summary.get("coverage_percent_after_soc_integration_control"),
intrusion_summary.get("coverage_percent_after_prevention_control"),
@@ -106,6 +109,15 @@ def load_latest_iwooos_runtime_security_readback(
"wazuh_dashboard_api_degraded_observed_count": _int(
wazuh_summary.get("dashboard_api_degraded_observed_count")
),
"wazuh_live_route_http_status": live_wazuh["http_status"],
"wazuh_live_route_degraded_count": live_wazuh["degraded_count"],
"wazuh_live_readonly_api_enabled_count": live_wazuh["readonly_api_enabled_count"],
"wazuh_live_agent_total": live_wazuh["agent_total"],
"wazuh_live_agent_active": live_wazuh["agent_active"],
"wazuh_live_registry_empty_count": live_wazuh["agent_registry_empty_count"],
"wazuh_live_below_expected_count": live_wazuh["agent_below_expected_minimum_count"],
"wazuh_live_metadata_available_count": live_wazuh["metadata_available_count"],
"wazuh_live_status": live_wazuh["status"],
"kali_active_scan_authorized_count": _int(soc_summary.get("kali_active_scan_authorized_count")),
"kali_execute_authorized_count": _int(soc_summary.get("kali_execute_authorized_count")),
"kali_finding_envelope_accepted_count": _int(soc_summary.get("kali_finding_envelope_accepted_count")),
@@ -128,6 +140,21 @@ def load_latest_iwooos_runtime_security_readback(
},
["docs/security/wazuh-managed-host-coverage-gate.snapshot.json"],
),
_lane(
"wazuh_live_route",
live_wazuh["status"],
0 if live_wazuh["degraded_count"] else 30,
"steady" if live_wazuh["metadata_available_count"] else "warn",
"enable_readonly_metadata_owner_gate_and_manager_registry_cross_check",
{
"http_status": live_wazuh["http_status"],
"readonly_enabled": live_wazuh["readonly_api_enabled_count"],
"agent_total": live_wazuh["agent_total"],
"metadata_available": live_wazuh["metadata_available_count"],
"route_degraded": live_wazuh["degraded_count"],
},
["GET /api/iwooos/wazuh"],
),
_lane(
"wazuh_dashboard_api",
"degraded_api_connection_not_green",
@@ -223,6 +250,7 @@ def load_latest_iwooos_runtime_security_readback(
"owner_request_draft_is_not_owner_acceptance",
"kali_health_is_not_active_scan_authorization",
"alert_format_contract_is_not_telegram_send_receipt",
"wazuh_live_route_disabled_or_degraded_is_p0_not_green",
],
}
@@ -250,6 +278,36 @@ def _int(value: Any) -> int:
return value if isinstance(value, int) else 0
def _wazuh_live_summary(payload: dict[str, Any] | None, http_status: int) -> dict[str, Any]:
if not isinstance(payload, dict):
return {
"status": "not_checked_by_snapshot_loader",
"http_status": 0,
"degraded_count": 1,
"readonly_api_enabled_count": 0,
"agent_total": 0,
"agent_active": 0,
"agent_registry_empty_count": 0,
"agent_below_expected_minimum_count": 0,
"metadata_available_count": 0,
}
summary = payload.get("summary")
summary = summary if isinstance(summary, dict) else {}
status_text = str(payload.get("status") or "unknown")
metadata_available = status_text == "readonly_metadata_available"
return {
"status": status_text,
"http_status": http_status if isinstance(http_status, int) else 0,
"degraded_count": 0 if metadata_available else 1,
"readonly_api_enabled_count": _int(summary.get("readonly_api_enabled_count")),
"agent_total": _int(summary.get("agent_total")),
"agent_active": _int(summary.get("agent_active")),
"agent_registry_empty_count": _int(summary.get("agent_registry_empty_count")),
"agent_below_expected_minimum_count": _int(summary.get("agent_below_expected_minimum_count")),
"metadata_available_count": 1 if metadata_available else 0,
}
def _average_percent(*values: Any) -> int:
percents = [_int(value) for value in values if isinstance(value, int)]
return int(round(sum(percents) / len(percents))) if percents else 0

194
apps/api/src/services/iwooos_wazuh_readonly_status.py Normal file
View File

@@ -0,0 +1,194 @@
"""
IwoooS Wazuh read-only metadata status.
This module returns public-safe aggregate metadata only. It never stores raw
Wazuh payloads, never exposes agent identity or LAN topology, and never
authorizes active response or host writes.
"""
from __future__ import annotations
import os
from base64 import b64encode
from dataclasses import dataclass
from typing import Any
from urllib.parse import urljoin, urlparse
import httpx
REQUEST_TIMEOUT_SECONDS = 5.0
@dataclass(frozen=True)
class WazuhReadonlyStatus:
payload: dict[str, Any]
http_status: int = 200
def wazuh_env() -> dict[str, str]:
return {
"enabled": os.getenv("IWOOOS_WAZUH_READONLY_ENABLED", "").strip().lower(),
"base_url": os.getenv("WAZUH_API_BASE_URL", "").strip(),
"username": os.getenv("WAZUH_API_USERNAME", "").strip(),
"password": os.getenv("WAZUH_API_PASSWORD", "").strip(),
"expected_min_agent_count": os.getenv("IWOOOS_WAZUH_EXPECTED_MIN_AGENT_COUNT", "").strip(),
}
def expected_min_agent_count(value: str) -> int:
try:
return max(0, int(value))
except ValueError:
return 0
def https_url(value: str) -> str | None:
parsed = urlparse(value)
if parsed.scheme != "https" or not parsed.netloc:
return None
return value.rstrip("/") + "/"
def boundaries() -> dict[str, bool]:
return {
"active_response_authorized": False,
"host_write_authorized": False,
"secret_value_collection_allowed": False,
"raw_wazuh_payload_storage_allowed": False,
"agent_identity_public_display_allowed": False,
"internal_ip_public_display_allowed": False,
"not_authorization": True,
}
def boundary_status(status_text: str, http_status: int = 200) -> WazuhReadonlyStatus:
return WazuhReadonlyStatus(
http_status=http_status,
payload={
"schema_version": "iwooos_wazuh_readonly_status_v1",
"status": status_text,
"mode": "metadata_only_no_active_response_no_raw_payload",
"configured": False,
"summary": {
"wazuh_platform_reported_count": 1,
"readonly_api_enabled_count": 0,
"wazuh_manager_query_accepted_count": 0,
"wazuh_event_accepted_count": 0,
"host_forensics_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
"expected_min_agent_count": expected_min_agent_count(wazuh_env()["expected_min_agent_count"]),
"agent_registry_empty_count": 0,
"agent_below_expected_minimum_count": 0,
"agent_visibility_no_false_green_count": 1,
},
"boundaries": boundaries(),
},
)
def redacted_agent(agent: dict[str, Any], index: int) -> dict[str, Any]:
os_info = agent.get("os") if isinstance(agent.get("os"), dict) else {}
return {
"alias": f"agent-{index + 1:02d}",
"status": agent.get("status", "unknown"),
"os": os_info.get("platform") or os_info.get("name") or "unknown",
"last_seen_present": bool(agent.get("lastKeepAlive")),
}
def int_or_default(value: Any, default: int) -> int:
return value if isinstance(value, int) else default
def agent_visibility_status(agent_total: int, expected_minimum: int) -> str:
if agent_total <= 0:
return "wazuh_agent_registry_empty"
if expected_minimum > 0 and agent_total < expected_minimum:
return "wazuh_agent_registry_below_expected"
return "readonly_metadata_available"
async def fetch_json(client: httpx.AsyncClient, url: str, headers: dict[str, str]) -> dict[str, Any]:
response = await client.get(url, headers=headers)
response.raise_for_status()
payload = response.json()
return payload if isinstance(payload, dict) else {}
async def load_iwooos_wazuh_readonly_status() -> WazuhReadonlyStatus:
env = wazuh_env()
if env["enabled"] != "true":
return boundary_status("disabled_waiting_iwooos_wazuh_owner_gate")
base_url = https_url(env["base_url"])
if not base_url or not env["username"] or not env["password"]:
return boundary_status("misconfigured_missing_server_side_wazuh_env", 503)
try:
auth_header = b64encode(f"{env['username']}:{env['password']}".encode("utf-8")).decode("ascii")
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS) as client:
auth = await fetch_json(
client,
urljoin(base_url, "security/user/authenticate"),
{"Authorization": f"Basic {auth_header}"},
)
token = (auth.get("data") or {}).get("token")
if not token:
return boundary_status("wazuh_auth_token_missing", 502)
bearer_headers = {"Authorization": f"Bearer {token}"}
status_payload = await fetch_json(
client,
urljoin(base_url, "agents/summary/status"),
bearer_headers,
)
agents_payload = await fetch_json(
client,
urljoin(base_url, "agents?limit=100&select=id,status,os.name,os.platform,lastKeepAlive"),
bearer_headers,
)
except (httpx.HTTPError, ValueError):
return boundary_status("wazuh_readonly_metadata_unavailable", 502)
connection = ((status_payload.get("data") or {}).get("connection") or {})
affected_items = ((agents_payload.get("data") or {}).get("affected_items") or [])
if not isinstance(affected_items, list):
affected_items = []
expected_minimum = expected_min_agent_count(env["expected_min_agent_count"])
agent_total = int_or_default(connection.get("total"), len(affected_items))
agent_active = int_or_default(connection.get("active"), 0)
agent_disconnected = int_or_default(connection.get("disconnected"), 0)
agent_pending = int_or_default(connection.get("pending"), 0)
agent_registry_empty = agent_total <= 0
agent_below_expected = expected_minimum > 0 and agent_total < expected_minimum
return WazuhReadonlyStatus(
payload={
"schema_version": "iwooos_wazuh_readonly_status_v1",
"status": agent_visibility_status(agent_total, expected_minimum),
"mode": "metadata_only_no_active_response_no_raw_payload",
"configured": True,
"summary": {
"wazuh_platform_reported_count": 1,
"readonly_api_enabled_count": 1,
"agent_total": agent_total,
"agent_active": agent_active,
"agent_disconnected": agent_disconnected,
"agent_pending": agent_pending,
"expected_min_agent_count": expected_minimum,
"agent_registry_empty_count": 1 if agent_registry_empty else 0,
"agent_below_expected_minimum_count": 1 if agent_below_expected else 0,
"agent_visibility_no_false_green_count": 1,
"wazuh_manager_query_accepted_count": 0,
"wazuh_event_accepted_count": 0,
"host_forensics_accepted_count": 0,
"active_response_authorized_count": 0,
"host_write_authorized_count": 0,
"runtime_gate_count": 0,
},
"agents": [redacted_agent(agent, index) for index, agent in enumerate(affected_items[:20])],
"boundaries": boundaries(),
},
)

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

@@ -21,11 +21,15 @@ def test_iwooos_runtime_security_readback_preserves_zero_runtime_gates() -> None
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"] == 6
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
@@ -41,6 +45,7 @@ def test_iwooos_runtime_security_readback_lanes_are_candidate_only() -> None:
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",
@@ -52,16 +57,69 @@ def test_iwooos_runtime_security_readback_lanes_are_candidate_only() -> None:
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() -> None:
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