#!/usr/bin/env python3
"""
Wazuh 主機納管覆蓋 Gate。

本工具只驗證 repo 內的脫敏 snapshot；不連線 Wazuh、不讀 secret、
不重新註冊 agent、不重啟服務、不修改主機，也不啟用 active response。
"""

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path
from typing import Any


SNAPSHOT_PATH = Path("docs/security/wazuh-managed-host-coverage-gate.snapshot.json")
SCHEMA_VERSION = "wazuh_managed_host_coverage_gate_v1"

HOST_SCOPE_MATRIX = [
    {
        "node_id": "managed_core_node_a",
        "role": "核心服務節點",
        "readback_status": "agent_active_transport_observed",
        "manager_registry_accepted": True,
        "next_gate": "runtime_gate_owner_review",
    },
    {
        "node_id": "managed_core_node_b",
        "role": "資料服務節點",
        "readback_status": "agent_active_transport_observed",
        "manager_registry_accepted": True,
        "next_gate": "runtime_gate_owner_review",
    },
    {
        "node_id": "managed_dev_node_a",
        "role": "開發工作節點",
        "readback_status": "no_agent_transport_observed",
        "manager_registry_accepted": True,
        "next_gate": "runtime_gate_owner_review",
    },
    {
        "node_id": "managed_dev_node_b",
        "role": "開發工作節點",
        "readback_status": "ssh_readback_blocked",
        "manager_registry_accepted": True,
        "next_gate": "runtime_gate_owner_review",
    },
    {
        "node_id": "managed_control_node_a",
        "role": "控制平面節點",
        "readback_status": "ssh_readback_blocked",
        "manager_registry_accepted": True,
        "next_gate": "runtime_gate_owner_review",
    },
    {
        "node_id": "managed_control_node_b",
        "role": "控制平面節點",
        "readback_status": "ssh_readback_blocked",
        "manager_registry_accepted": True,
        "next_gate": "runtime_gate_owner_review",
    },
]

REQUIRED_EVIDENCE_BEFORE_GREEN = [
    "manager_registry_agent_counts",
    "per_host_agent_scope_matrix",
    "dashboard_api_rbac_tls_repair_readback",
    "readonly_credential_metadata_without_secret",
    "owner_response_and_rollback_owner",
    "post_enable_iwooos_readback",
]

FORBIDDEN_COMPLETION_CLAIMS = [
    "所有 Wazuh 用戶端已恢復",
    "所有主機已納入 Wazuh",
    "Wazuh agent registry 已驗收等於 runtime 已授權",
    "Dashboard 可見等於 registry 已恢復",
    "transport 連線等於全數納管",
]

FORBIDDEN_ACTIONS = [
    "wazuh_agent_reenroll",
    "wazuh_agent_restart",
    "wazuh_manager_restart",
    "wazuh_dashboard_secret_patch",
    "active_response_enable",
    "host_write",
    "firewall_change",
    "nginx_reload",
    "kali_active_scan",
]

FORBIDDEN_TEXT_PATTERNS = [
    re.compile(r"\b(?:10|127|172\.(?:1[6-9]|2\d|3[01])|192\.168)\.\d{1,3}\.\d{1,3}\b"),
    re.compile(r"Authorization\s*:", re.IGNORECASE),
    re.compile(r"Bearer\s+[A-Za-z0-9._-]{10,}", re.IGNORECASE),
    re.compile(r"Basic\s+[A-Za-z0-9+/=]{10,}", re.IGNORECASE),
    re.compile(r"password\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
    re.compile(r"token\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
    re.compile(r"cookie\s*[:=]\s*['\"][^'\"]+['\"]", re.IGNORECASE),
    re.compile(r"client\.keys", re.IGNORECASE),
    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
]


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def assert_equal(label: str, actual: Any, expected: Any) -> None:
    if actual != expected:
        raise SystemExit(f"BLOCKED {label}: expected {expected!r}, got {actual!r}")


def assert_false(label: str, actual: Any) -> None:
    assert_equal(label, actual, False)


def assert_zero(label: str, actual: Any) -> None:
    assert_equal(label, actual, 0)


def collect_string_values(value: Any) -> list[str]:
    if isinstance(value, str):
        return [value]
    if isinstance(value, list):
        values: list[str] = []
        for item in value:
            values.extend(collect_string_values(item))
        return values
    if isinstance(value, dict):
        values = []
        for item in value.values():
            values.extend(collect_string_values(item))
        return values
    return []


def validate_no_forbidden_text(snapshot: dict[str, Any]) -> None:
    for text in collect_string_values(snapshot):
        for pattern in FORBIDDEN_TEXT_PATTERNS:
            if pattern.search(text):
                raise SystemExit("BLOCKED wazuh_managed_host_coverage_gate: snapshot contains forbidden sensitive text")


def build_snapshot(generated_at: str) -> dict[str, Any]:
    return {
        "schema_version": SCHEMA_VERSION,
        "generated_at": generated_at,
        "status": "manager_registry_readback_accepted_runtime_gate_closed",
        "mode": "committed_manager_registry_readback_no_runtime_no_secret_collection",
        "scope": "wazuh_managed_host_coverage",
        "summary": {
            "expected_host_scope_count": len(HOST_SCOPE_MATRIX),
            "manager_service_active_observed_count": 1,
            "manager_api_unauthenticated_response_count": 1,
            "manager_transport_established_connection_count": 6,
            "direct_agent_active_observed_count": 2,
            "direct_agent_transport_observed_count": 2,
            "direct_agent_missing_or_no_transport_count": 1,
            "ssh_readback_blocked_count": 3,
            "manager_registry_accepted_count": len(HOST_SCOPE_MATRIX),
            "dashboard_api_degraded_observed_count": 1,
            "live_metadata_env_enabled_count": 0,
            "active_response_authorized_count": 0,
            "host_write_authorized_count": 0,
            "agent_reenroll_authorized_count": 0,
            "agent_restart_authorized_count": 0,
            "runtime_gate_count": 0,
        },
        "host_scope_matrix": HOST_SCOPE_MATRIX,
        "required_evidence_before_green": [
            {"evidence_id": evidence_id, "accepted": True}
            for evidence_id in REQUIRED_EVIDENCE_BEFORE_GREEN
        ],
        "forbidden_completion_claims": FORBIDDEN_COMPLETION_CLAIMS,
        "forbidden_actions": FORBIDDEN_ACTIONS,
        "operator_interpretation": [
            "manager registry accepted readback 已用 6 個公開節點別名提交；此讀回只代表脫敏 evidence 覆蓋，不代表 runtime 授權。",
            "Dashboard API、RBAC、rate-limit 或 TLS 退化會讓 UI 代理清單看起來消失，但不能用 UI 畫面單獨判定 agent 全部恢復。",
            "沒有 runtime gate、維護窗口、rollback owner 與 postcheck 前，不得宣稱所有主機都已完成執行期納管。",
            "重新註冊 agent、重啟 Wazuh、修改主機或改機密都必須走獨立維護窗口與 rollback owner。",
        ],
        "execution_boundaries": {
            "wazuh_api_live_query_authorized": False,
            "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,
            "kali_active_scan_authorized": False,
            "runtime_execution_authorized": False,
            "not_authorization": True,
        },
    }


def validate(root: Path) -> None:
    snapshot = load_json(root / SNAPSHOT_PATH)
    assert_equal("schema_version", snapshot.get("schema_version"), SCHEMA_VERSION)
    assert_equal("status", snapshot.get("status"), "manager_registry_readback_accepted_runtime_gate_closed")
    assert_equal("mode", snapshot.get("mode"), "committed_manager_registry_readback_no_runtime_no_secret_collection")
    assert_equal("scope", snapshot.get("scope"), "wazuh_managed_host_coverage")

    summary = snapshot.get("summary", {})
    assert_equal("summary.expected_host_scope_count", summary.get("expected_host_scope_count"), len(HOST_SCOPE_MATRIX))
    assert_equal("summary.manager_service_active_observed_count", summary.get("manager_service_active_observed_count"), 1)
    assert_equal("summary.manager_api_unauthenticated_response_count", summary.get("manager_api_unauthenticated_response_count"), 1)
    assert_equal("summary.manager_transport_established_connection_count", summary.get("manager_transport_established_connection_count"), 6)
    assert_equal("summary.direct_agent_active_observed_count", summary.get("direct_agent_active_observed_count"), 2)
    assert_equal("summary.direct_agent_transport_observed_count", summary.get("direct_agent_transport_observed_count"), 2)
    assert_equal("summary.direct_agent_missing_or_no_transport_count", summary.get("direct_agent_missing_or_no_transport_count"), 1)
    assert_equal("summary.ssh_readback_blocked_count", summary.get("ssh_readback_blocked_count"), 3)
    assert_equal(
        "summary.manager_registry_accepted_count",
        summary.get("manager_registry_accepted_count"),
        len(HOST_SCOPE_MATRIX),
    )
    assert_equal("summary.dashboard_api_degraded_observed_count", summary.get("dashboard_api_degraded_observed_count"), 1)
    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",
    ]:
        assert_zero(f"summary.{key}", summary.get(key))

    assert_equal("host_scope_matrix", snapshot.get("host_scope_matrix"), HOST_SCOPE_MATRIX)
    for item in snapshot.get("host_scope_matrix", []):
        assert_equal(f"host_scope_matrix.{item.get('node_id')}.manager_registry_accepted", item.get("manager_registry_accepted"), True)

    required = snapshot.get("required_evidence_before_green", [])
    assert_equal("required_evidence_before_green.count", len(required), len(REQUIRED_EVIDENCE_BEFORE_GREEN))
    assert_equal(
        "required_evidence_before_green.ids",
        [item.get("evidence_id") for item in required],
        REQUIRED_EVIDENCE_BEFORE_GREEN,
    )
    for item in required:
        assert_equal(f"required_evidence_before_green.{item.get('evidence_id')}.accepted", item.get("accepted"), True)

    assert_equal("forbidden_completion_claims", snapshot.get("forbidden_completion_claims"), FORBIDDEN_COMPLETION_CLAIMS)
    assert_equal("forbidden_actions", snapshot.get("forbidden_actions"), FORBIDDEN_ACTIONS)
    boundaries = snapshot.get("execution_boundaries", {})
    for key, value in boundaries.items():
        if key == "not_authorization":
            assert_equal(f"execution_boundaries.{key}", value, True)
        else:
            assert_false(f"execution_boundaries.{key}", value)
    validate_no_forbidden_text(snapshot)


def main() -> None:
    parser = argparse.ArgumentParser(description="Wazuh 主機納管覆蓋 Gate")
    parser.add_argument("--root", type=Path, default=Path.cwd())
    parser.add_argument("--output", type=Path)
    parser.add_argument("--generated-at", default="2026-06-25T11:45:31+08:00")
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    root = args.root.resolve()
    if args.output:
        snapshot = build_snapshot(args.generated_at)
        output = args.output if args.output.is_absolute() else root / args.output
        output.parent.mkdir(parents=True, exist_ok=True)
        output.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")

    validate(root)
    snapshot = load_json(root / SNAPSHOT_PATH)
    if args.json:
        print(json.dumps(snapshot, ensure_ascii=False, sort_keys=True))
        return
    summary = snapshot["summary"]
    print(
        "WAZUH_MANAGED_HOST_COVERAGE_GATE_OK "
        f"scope={summary['expected_host_scope_count']} "
        f"direct_active={summary['direct_agent_active_observed_count']} "
        f"no_transport={summary['direct_agent_missing_or_no_transport_count']} "
        f"ssh_blocked={summary['ssh_readback_blocked_count']} "
        f"registry={summary['manager_registry_accepted_count']} "
        f"runtime_gate={summary['runtime_gate_count']}"
    )


if __name__ == "__main__":
    main()
