4ed96a83a5
Some checks failed
Code Review / ai-code-review (push) Successful in 17s
CD Pipeline / tests (push) Successful in 1m39s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
249 lines
12 KiB
Python
249 lines
12 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": "代理服務與傳輸連線已觀察，仍待管理器清單驗收",
|
|
"no_agent_transport_observed": "未觀察到代理傳輸，需確認安裝、服務與負責人決策",
|
|
"ssh_readback_blocked": "合法只讀讀回仍受阻，需 owner export 或維護窗口",
|
|
}
|
|
|
|
_NEXT_GATE_LABELS = {
|
|
"manager_registry_cross_check": "補管理器清單交叉驗收",
|
|
"agent_install_or_service_owner_decision": "補代理安裝狀態或服務負責人決策",
|
|
"read_only_access_or_owner_export": "補只讀 access 或脫敏 owner export",
|
|
}
|
|
|
|
|
|
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_count 維持 0 時不得宣稱所有主機已納管",
|
|
"重新註冊、重啟、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}")
|
|
if raw_item.get("manager_registry_accepted") is not False:
|
|
raise ValueError(f"Wazuh 受管主機覆蓋 {node_id} manager_registry_accepted 必須維持 false")
|
|
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": False,
|
|
}
|
|
)
|
|
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 (
|
|
"manager_registry_accepted_count",
|
|
"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")
|
|
|
|
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")
|