""" IwoooS Wazuh managed host coverage readback. This service exposes the committed Wazuh managed-host coverage snapshot as a public-safe, alias-only payload. It never queries Wazuh, reads secrets, reenrolls agents, restarts services, writes hosts, or enables active response. """ from __future__ import annotations import json from pathlib import Path from typing import Any from src.services.snapshot_paths import default_security_dir _DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__)) _SNAPSHOT_FILE = "wazuh-managed-host-coverage-gate.snapshot.json" _EXPECTED_SCHEMA = "wazuh_managed_host_coverage_gate_v1" _REQUIRED_FALSE_BOUNDARIES = { "host_write_authorized", "kali_active_scan_authorized", "raw_wazuh_payload_storage_allowed", "runtime_execution_authorized", "secret_value_collection_allowed", "wazuh_active_response_authorized", "wazuh_agent_reenroll_authorized", "wazuh_agent_restart_authorized", "wazuh_api_live_query_authorized", "wazuh_manager_restart_authorized", } _STATUS_LABELS = { "agent_active_transport_observed": "代理服務與傳輸連線已觀察，管理器清單 evidence 已驗收", "no_agent_transport_observed": "管理器清單 evidence 已驗收，仍需 runtime/服務狀態獨立確認", "ssh_readback_blocked": "管理器清單 evidence 已驗收，合法只讀讀回與 runtime postcheck 仍需獨立 gate", } _NEXT_GATE_LABELS = { "manager_registry_cross_check": "補管理器清單交叉驗收", "agent_install_or_service_owner_decision": "補代理安裝狀態或服務負責人決策", "read_only_access_or_owner_export": "補只讀 access 或脫敏 owner export", "runtime_gate_owner_review": "進 runtime gate owner review 與 postcheck", } def load_latest_iwooos_wazuh_managed_host_coverage( security_dir: Path | None = None, ) -> dict[str, Any]: """Load Wazuh managed-host coverage as an alias-only public readback.""" directory = security_dir or _DEFAULT_SECURITY_DIR snapshot = _load_snapshot(directory) _require_boundaries(snapshot) summary = _summary(snapshot) matrix = _host_scope_matrix(snapshot) evidence = _required_evidence(snapshot) merged_summary = { "expected_host_scope_count": _int(summary.get("expected_host_scope_count")), "manager_service_active_observed_count": _int(summary.get("manager_service_active_observed_count")), "manager_api_unauthenticated_response_count": _int(summary.get("manager_api_unauthenticated_response_count")), "manager_transport_established_connection_count": _int( summary.get("manager_transport_established_connection_count") ), "direct_agent_active_observed_count": _int(summary.get("direct_agent_active_observed_count")), "direct_agent_transport_observed_count": _int(summary.get("direct_agent_transport_observed_count")), "direct_agent_missing_or_no_transport_count": _int( summary.get("direct_agent_missing_or_no_transport_count") ), "ssh_readback_blocked_count": _int(summary.get("ssh_readback_blocked_count")), "manager_registry_accepted_count": _int(summary.get("manager_registry_accepted_count")), "manager_registry_gap_count": max( _int(summary.get("expected_host_scope_count")) - _int(summary.get("manager_registry_accepted_count")), 0, ), "dashboard_api_degraded_observed_count": _int(summary.get("dashboard_api_degraded_observed_count")), "live_metadata_env_enabled_count": _int(summary.get("live_metadata_env_enabled_count")), "active_response_authorized_count": _int(summary.get("active_response_authorized_count")), "host_write_authorized_count": _int(summary.get("host_write_authorized_count")), "agent_reenroll_authorized_count": _int(summary.get("agent_reenroll_authorized_count")), "agent_restart_authorized_count": _int(summary.get("agent_restart_authorized_count")), "runtime_gate_count": _int(summary.get("runtime_gate_count")), "host_scope_matrix_count": len(matrix), "required_evidence_before_green_count": len(evidence), "required_evidence_accepted_count": sum(1 for item in evidence if item.get("accepted") is True), } return { "schema_version": "iwooos_wazuh_managed_host_coverage_readback_v1", "status": snapshot.get("status", "blocked_waiting_full_host_registry_readback"), "mode": "committed_snapshot_readback_alias_only_no_wazuh_live_query", "source_refs": [ f"docs/security/{_SNAPSHOT_FILE}", "scripts/security/wazuh-managed-host-coverage-gate.py", ], "summary": merged_summary, "host_scope_matrix": matrix, "required_evidence_before_green": evidence, "operator_interpretation": _list_of_strings(snapshot.get("operator_interpretation")), "forbidden_completion_claims": _list_of_strings(snapshot.get("forbidden_completion_claims")), "forbidden_actions": _list_of_strings(snapshot.get("forbidden_actions")), "boundary_markers": _boundary_markers(merged_summary), "boundaries": { "wazuh_api_live_query_authorized": False, "wazuh_active_response_authorized": False, "wazuh_agent_reenroll_authorized": False, "wazuh_agent_restart_authorized": False, "wazuh_manager_restart_authorized": False, "host_write_authorized": False, "kali_active_scan_authorized": False, "secret_value_collection_allowed": False, "raw_wazuh_payload_storage_allowed": False, "runtime_execution_authorized": False, "not_authorization": True, }, "no_false_green_rules": [ "Wazuh Dashboard 可見不等於 manager registry 已恢復", "transport 連線、agent service active 或 HTTP 200 不可替代逐主機 registry matrix", "manager registry accepted 只代表 committed redacted evidence 覆蓋公開別名，不代表 runtime 授權", "重新註冊、重啟、active response、主機寫入、Nginx、firewall、Kali 掃描與機密調整都不是此讀回授權", ], } def _load_snapshot(directory: Path) -> dict[str, Any]: path = directory / _SNAPSHOT_FILE if not path.is_file(): raise FileNotFoundError(f"{path}: Wazuh 受管主機覆蓋快照不存在") with path.open(encoding="utf-8") as handle: payload = json.load(handle) if not isinstance(payload, dict): raise ValueError(f"{path}: expected JSON object") if payload.get("schema_version") != _EXPECTED_SCHEMA: raise ValueError(f"{path}: expected schema_version={_EXPECTED_SCHEMA}") return payload def _summary(payload: dict[str, Any]) -> dict[str, Any]: summary = payload.get("summary") return summary if isinstance(summary, dict) else {} def _int(value: Any) -> int: return value if isinstance(value, int) else 0 def _list_of_strings(value: Any) -> list[str]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, str)] def _host_scope_matrix(payload: dict[str, Any]) -> list[dict[str, Any]]: raw_items = payload.get("host_scope_matrix") if not isinstance(raw_items, list): raise ValueError("Wazuh 受管主機覆蓋 host_scope_matrix 缺失") matrix: list[dict[str, Any]] = [] for raw_item in raw_items: if not isinstance(raw_item, dict): raise ValueError("Wazuh 受管主機覆蓋 host_scope_matrix 必須是 object list") node_id = str(raw_item.get("node_id", "")) role = str(raw_item.get("role", "")) readback_status = str(raw_item.get("readback_status", "")) next_gate = str(raw_item.get("next_gate", "")) if not node_id.startswith("managed_"): raise ValueError(f"Wazuh 受管主機覆蓋 node_id 必須維持公開別名：{node_id}") manager_registry_accepted = raw_item.get("manager_registry_accepted") is True matrix.append( { "node_id": node_id, "role": role, "readback_status": readback_status, "readback_status_label": _STATUS_LABELS.get(readback_status, "待補只讀讀回"), "next_gate": next_gate, "next_gate_label": _NEXT_GATE_LABELS.get(next_gate, "待補負責人驗收"), "manager_registry_accepted": manager_registry_accepted, } ) return matrix def _required_evidence(payload: dict[str, Any]) -> list[dict[str, Any]]: raw_items = payload.get("required_evidence_before_green") if not isinstance(raw_items, list): raise ValueError("Wazuh 受管主機覆蓋 required_evidence_before_green 缺失") evidence: list[dict[str, Any]] = [] for raw_item in raw_items: if not isinstance(raw_item, dict): raise ValueError("Wazuh 受管主機覆蓋 required_evidence_before_green 必須是 object list") evidence_id = str(raw_item.get("evidence_id", "")) accepted = raw_item.get("accepted") is True evidence.append({"evidence_id": evidence_id, "accepted": accepted}) return evidence def _boundary_markers(summary: dict[str, int]) -> list[str]: return [ "wazuh_managed_host_coverage_gate_visible=true", f"wazuh_managed_host_coverage_expected_host_scope_count={summary['expected_host_scope_count']}", f"wazuh_managed_host_coverage_host_scope_matrix_count={summary['host_scope_matrix_count']}", f"wazuh_managed_host_coverage_direct_agent_active_observed_count={summary['direct_agent_active_observed_count']}", f"wazuh_managed_host_coverage_direct_agent_missing_or_no_transport_count={summary['direct_agent_missing_or_no_transport_count']}", f"wazuh_managed_host_coverage_ssh_readback_blocked_count={summary['ssh_readback_blocked_count']}", f"wazuh_managed_host_coverage_manager_registry_accepted_count={summary['manager_registry_accepted_count']}", f"wazuh_managed_host_coverage_manager_registry_gap_count={summary['manager_registry_gap_count']}", f"wazuh_managed_host_coverage_required_evidence_before_green_count={summary['required_evidence_before_green_count']}", f"wazuh_managed_host_coverage_required_evidence_accepted_count={summary['required_evidence_accepted_count']}", f"wazuh_managed_host_coverage_dashboard_api_degraded_observed_count={summary['dashboard_api_degraded_observed_count']}", f"wazuh_managed_host_coverage_live_metadata_env_enabled_count={summary['live_metadata_env_enabled_count']}", f"wazuh_managed_host_coverage_runtime_gate_count={summary['runtime_gate_count']}", f"wazuh_agent_reenroll_authorized_count={summary['agent_reenroll_authorized_count']}", f"wazuh_agent_restart_authorized_count={summary['agent_restart_authorized_count']}", "wazuh_agent_reenroll_authorized=false", "wazuh_agent_restart_authorized=false", "wazuh_manager_restart_authorized=false", "wazuh_active_response_authorized=false", "host_write_authorized=false", "secret_value_collection_allowed=false", "raw_wazuh_payload_storage_allowed=false", "runtime_execution_authorized=false", "not_authorization=true", ] def _require_boundaries(payload: dict[str, Any]) -> None: summary = _summary(payload) for key in ( "live_metadata_env_enabled_count", "active_response_authorized_count", "host_write_authorized_count", "agent_reenroll_authorized_count", "agent_restart_authorized_count", "runtime_gate_count", ): if _int(summary.get(key)) != 0: raise ValueError(f"Wazuh 受管主機覆蓋 summary.{key} 必須維持 0") expected_hosts = _int(summary.get("expected_host_scope_count")) manager_accepted = _int(summary.get("manager_registry_accepted_count")) if manager_accepted < 0: raise ValueError("Wazuh 受管主機覆蓋 summary.manager_registry_accepted_count 不得為負數") if manager_accepted > expected_hosts: raise ValueError("Wazuh 受管主機覆蓋 summary.manager_registry_accepted_count 不得大於 expected_host_scope_count") raw_matrix = payload.get("host_scope_matrix") if isinstance(raw_matrix, list): matrix_accepted = sum( 1 for item in raw_matrix if isinstance(item, dict) and item.get("manager_registry_accepted") is True ) if matrix_accepted != manager_accepted: raise ValueError("Wazuh 受管主機覆蓋 manager registry accepted matrix/count 不一致") boundaries = payload.get("execution_boundaries") if not isinstance(boundaries, dict): raise ValueError("Wazuh 受管主機覆蓋 execution_boundaries 缺失") for key in _REQUIRED_FALSE_BOUNDARIES: if boundaries.get(key) is not False: raise ValueError(f"Wazuh 受管主機覆蓋 execution_boundaries.{key} 必須維持 false") if boundaries.get("not_authorization") is not True: raise ValueError("Wazuh 受管主機覆蓋 not_authorization 必須維持 true")