265 lines
13 KiB
Python
265 lines
13 KiB
Python
"""
|
|
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")
|