#!/usr/bin/env python3
"""Validate retained legacy Windows evidence bundles before first rollout."""

from __future__ import annotations

import argparse
import hashlib
import json
import re
import sys
import zipfile
from pathlib import Path
from typing import Any


TOOL_NAME = "legacy-windows-first-iteration-readiness"
EVIDENCE_BUNDLE_TOOL = "legacy-windows-first-iteration-evidence-bundle"

REQUIRED_ARTIFACT_ROLES = {
    "lab_smoke",
    "collect_manifest",
    "build_manifest",
    "collect_file:operator-context.json",
    "collect_file:watchdog.log",
    "collect_file:legacy-windows-lab-smoke.json",
    "collect_file:task-SK-Agent.txt",
    "collect_file:task-SK-Watchdog.txt",
    "collect_file:task-SK-Tray.txt",
    "collect_file:service-SKRecoverDaemon.txt",
    "collect_file:firewall-SiteKings-Recover-Daemon.txt",
    "collect_file:netstat-ano.txt",
    "collect_file:tasklist.txt",
}

WIN7_BRIDGE_ROLES = {
    "win10_upgrade_status",
    "win10_upgrade_log",
}

WIN7_PROBE_ROLE = "win10_probe"
WIN10_BRIDGE_START_PREFIX = "Starting Windows 10 bridge upgrade"
WIN10_MEDIA_URL_QUERY_RE = re.compile(r"\?[^\s]*(?:[A-Za-z0-9_.~-]+)=")
REQUIRED_WIN10_PROBE_FIELDS = (
    "bios_version",
    "manufacturer",
    "model",
    "processor_name",
    "total_ram_gb",
)

OS_IDENTITY_FIELDS = (
    "os_caption",
    "os_version",
    "os_service_pack",
    "service_pack",
)

REQUIRED_SMOKE_CHECKS = {
    "task:SK Agent",
    "task:SK Watchdog",
    "task:SK Tray",
    "service:SKRecoverDaemon",
    "recover_listen_47778",
    "firewall:SKRecoverDaemon 47778",
    "process:sk-agent-legacy",
    "process:sk-tray",
    "peer_recover",
    "recover_health",
    "self_rescue",
    "watchdog.log",
}


def _read_json(zf: zipfile.ZipFile, name: str) -> tuple[dict[str, Any] | None, str | None]:
    try:
        data = json.loads(zf.read(name).decode("utf-8"))
    except Exception as exc:
        return None, f"{name}: cannot read JSON: {exc}"
    if not isinstance(data, dict):
        return None, f"{name}: root JSON value must be an object"
    return data, None


def _sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _is_sha256_hex(value: str) -> bool:
    if len(value) != 64:
        return False
    try:
        int(value, 16)
    except ValueError:
        return False
    return True


def _validate_manifest_entry(
    zf: zipfile.ZipFile,
    names: set[str],
    hostname: str,
    role: str,
    item: dict[str, Any],
) -> list[str]:
    errors: list[str] = []
    archive_path = str(item.get("archive_path") or "")
    if archive_path not in names:
        return [f"{hostname}: missing archived evidence file for {role}: {archive_path}"]
    try:
        data = zf.read(archive_path)
    except Exception as exc:
        return [f"{hostname}: cannot read archived evidence file for {role}: {exc}"]
    raw_size = item.get("size_bytes")
    raw_sha = item.get("sha256")
    expected_size = raw_size if isinstance(raw_size, int) and not isinstance(raw_size, bool) else 0
    expected_sha = raw_sha.strip() if isinstance(raw_sha, str) else ""
    if expected_size <= 0 or not _is_sha256_hex(expected_sha):
        errors.append(f"{hostname}: invalid manifest metadata for {role}")
    if expected_size != len(data):
        errors.append(f"{hostname}: artifact size mismatch for {role}")
    if _is_sha256_hex(expected_sha) and expected_sha != _sha256_bytes(data):
        errors.append(f"{hostname}: artifact hash mismatch for {role}")
    return errors


def _classify_legacy_os(smoke: dict[str, Any]) -> str:
    hay = " ".join(
        value.lower()
        for key in OS_IDENTITY_FIELDS
        if isinstance((value := smoke.get(key)), str)
    )
    is_sp1 = "service pack 1" in hay or "sp1" in hay or "6.1.7601" in hay
    if "server 2008 r2" in hay and is_sp1:
        return "server_2008_r2"
    if ("windows 7" in hay or "windows nt 6.1" in hay) and is_sp1:
        return "windows_7_sp1"
    return "unknown"


def _smoke_check_map(smoke: dict[str, Any]) -> dict[str, bool]:
    checks: dict[str, bool] = {}
    raw_checks = smoke.get("checks")
    if not isinstance(raw_checks, list):
        return checks
    for item in raw_checks:
        if not isinstance(item, dict):
            continue
        name = str(item.get("name") or "").strip()
        if name:
            checks[name] = item.get("ok") is True
    return checks


def _smoke_check_detail_map(smoke: dict[str, Any]) -> dict[str, str]:
    details: dict[str, str] = {}
    raw_checks = smoke.get("checks")
    if not isinstance(raw_checks, list):
        return details
    for item in raw_checks:
        if not isinstance(item, dict):
            continue
        raw_name = item.get("name")
        raw_detail = item.get("detail")
        if isinstance(raw_name, str) and isinstance(raw_detail, str):
            name = raw_name.strip()
            if name:
                details[name] = raw_detail
    return details


def _required_smoke_summary(smoke_checks: dict[str, bool]) -> dict[str, bool]:
    return {
        check: smoke_checks.get(check) is True
        for check in sorted(REQUIRED_SMOKE_CHECKS)
    }


def _validate_operator_context(hostname: str, payload: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    if payload.get("schema") != "legacy-windows-lab-operator-context":
        errors.append(f"{hostname}: operator-context.json: schema is not legacy-windows-lab-operator-context")
    if payload.get("is_elevated") is not True:
        errors.append(f"{hostname}: operator-context.json: is_elevated must be true")
    for key in ("username", "computer_name", "script_root"):
        value = payload.get(key)
        if not isinstance(value, str):
            errors.append(f"{hostname}: operator-context.json: {key} must be a string")
        elif not value.strip():
            errors.append(f"{hostname}: operator-context.json: {key} is required")
    raw_computer_name = payload.get("computer_name")
    computer_name = raw_computer_name.strip() if isinstance(raw_computer_name, str) else ""
    if computer_name and computer_name.lower() != hostname.lower():
        errors.append(
            f"{hostname}: operator-context.json: computer_name {computer_name} does not match bundle hostname {hostname}"
        )
    return errors


def _operator_context_summary(payload: dict[str, Any]) -> dict[str, Any]:
    computer_name = payload.get("computer_name")
    script_root = payload.get("script_root")
    username = payload.get("username")
    return {
        "computer_name": computer_name.strip() if isinstance(computer_name, str) else "",
        "is_elevated": payload.get("is_elevated") is True,
        "script_root": script_root.strip() if isinstance(script_root, str) else "",
        "username": username.strip() if isinstance(username, str) else "",
    }


def _win10_bridge_summary(zf: zipfile.ZipFile, roles: dict[str, dict[str, Any]]) -> dict[str, Any]:
    status_role = roles.get("win10_upgrade_status") or {}
    log_role = roles.get("win10_upgrade_log") or {}
    status_path = str(status_role.get("archive_path") or "")
    log_path = str(log_role.get("archive_path") or "")
    if not status_path or not log_path:
        return {}

    status, status_error = _read_json(zf, status_path)
    if status_error or status is None:
        return {}
    try:
        log_text = zf.read(log_path).decode("utf-8", errors="replace")
    except Exception:
        log_text = ""
    stage = status.get("stage")
    stage_text = stage.strip() if isinstance(stage, str) else ""
    probe_verdict = status.get("probe_verdict")
    probe_verdict_text = probe_verdict.strip() if isinstance(probe_verdict, str) else ""
    percent = status.get("percent")
    return {
        "acknowledgements_recorded": "Operator acknowledgements recorded" in log_text,
        "evidence_notes_recorded": "Operator evidence notes recorded: customer-scoped enablement" in log_text,
        "media_url_query_redacted": _win10_media_url_query_redacted(log_text),
        "percent": percent,
        "percent_valid": _win10_bridge_percent_valid(stage_text, percent),
        "probe_verdict": probe_verdict_text,
        "setup_launched": "Launching setup.exe" in log_text,
        "stage": stage_text,
    }


def _win10_media_url_query_redacted(log_text: str) -> bool:
    for line in log_text.splitlines():
        if WIN10_BRIDGE_START_PREFIX not in line or "URL=" not in line:
            continue
        media_url = line.split("URL=", 1)[1].strip()
        redacted_url = "<redacted>" in media_url or "%3credacted%3e" in media_url.lower()
        if WIN10_MEDIA_URL_QUERY_RE.search(media_url) and not redacted_url:
            return False
    return True


def _win10_bridge_percent_valid(stage: str, percent: Any) -> bool:
    if isinstance(percent, bool) or not isinstance(percent, int) or percent < 0 or percent > 100:
        return False
    if stage == "setup_running":
        return percent >= 60
    if stage == "completed":
        return percent == 100
    return False


def _win10_probe_summary(zf: zipfile.ZipFile, roles: dict[str, dict[str, Any]]) -> dict[str, Any]:
    role = roles.get(WIN7_PROBE_ROLE) or {}
    archive_path = str(role.get("archive_path") or "")
    if not archive_path:
        return {}
    payload, error = _read_json(zf, archive_path)
    if error or payload is None:
        return {}
    return {
        "ac_power_required": payload.get("ac_power_required") is True,
        "architecture": _probe_context_value(payload, "architecture"),
        "bios_version": _probe_context_value(payload, "bios_version"),
        "manufacturer": _probe_context_value(payload, "manufacturer"),
        "model": _probe_context_value(payload, "model"),
        "processor_name": _probe_context_value(payload, "processor_name"),
        "service_pack": _probe_context_value(payload, "service_pack"),
        "total_ram_gb": _probe_context_value(payload, "total_ram_gb"),
        "verdict": _probe_context_value(payload, "verdict"),
    }


def _probe_context_value(payload: dict[str, Any], field: str) -> str:
    value = payload.get(field)
    if isinstance(value, str):
        return value.strip()
    if field == "total_ram_gb" and isinstance(value, (int, float)) and not isinstance(value, bool):
        return str(value)
    return ""


def _positive_probe_ram_gb(summary: dict[str, Any]) -> bool:
    value = summary.get("total_ram_gb")
    if isinstance(value, bool):
        return False
    if isinstance(value, (int, float)):
        return value > 0
    if isinstance(value, str):
        try:
            return float(value.strip()) > 0
        except ValueError:
            return False
    return False


def _missing_win10_probe_context(summary: dict[str, Any]) -> list[str]:
    return [
        field for field in REQUIRED_WIN10_PROBE_FIELDS
        if not str(summary.get(field) or "").strip()
    ]


def _collector_summary(zf: zipfile.ZipFile, roles: dict[str, dict[str, Any]]) -> dict[str, Any]:
    manifest_role = roles.get("collect_manifest") or {}
    manifest_path = str(manifest_role.get("archive_path") or "")
    if not manifest_path:
        return {}
    payload, error = _read_json(zf, manifest_path)
    if error or payload is None:
        return {}
    files = payload.get("files")
    if not isinstance(files, list):
        files = []
    return {
        "build_manifest_sha256": payload.get("build_manifest_sha256"),
        "collector_exit": payload.get("collector_exit"),
        "files_count": len(files),
        "smoke_exit": payload.get("smoke_exit"),
    }


def _validate_collector_exit_status(hostname: str, summary: dict[str, Any]) -> list[str]:
    errors: list[str] = []
    for key in ("collector_exit", "smoke_exit"):
        value = summary.get(key)
        if isinstance(value, bool) or not isinstance(value, int) or value != 0:
            errors.append(f"{hostname}: collector manifest {key} is required and must be 0")
    return errors


def _validate_collector_build_manifest(
    zf: zipfile.ZipFile,
    hostname: str,
    roles: dict[str, dict[str, Any]],
    collector_summary: dict[str, Any],
) -> list[str]:
    errors: list[str] = []
    expected_sha = collector_summary.get("build_manifest_sha256")
    if not isinstance(expected_sha, str) or not _is_sha256_hex(expected_sha.strip()):
        return [f"{hostname}: collector manifest build_manifest_sha256 must be a 64-character hex string"]
    build_manifest_role = roles.get("build_manifest") or {}
    archive_path = str(build_manifest_role.get("archive_path") or "")
    if not archive_path:
        return [f"{hostname}: missing archived build manifest for collector provenance check"]
    try:
        actual_sha = _sha256_bytes(zf.read(archive_path))
    except Exception as exc:
        return [f"{hostname}: cannot read archived build manifest for collector provenance check: {exc}"]
    if expected_sha.strip().lower() != actual_sha.lower():
        errors.append(f"{hostname}: collector manifest build_manifest_sha256 must match archived build manifest")
    return errors


def _bundle_summary(path: Path) -> tuple[dict[str, Any] | None, list[str]]:
    errors: list[str] = []
    try:
        with zipfile.ZipFile(path) as zf:
            name_list = zf.namelist()
            names = set(name_list)
            duplicates = sorted({name for name in name_list if name_list.count(name) > 1})
            manifest, error = _read_json(zf, "manifest.json")
            if error:
                return None, [f"{path.name}: {error}"]
            assert manifest is not None
            if duplicates:
                errors.append(f"{path.name}: duplicate archive entries are not allowed: {', '.join(duplicates)}")
            if manifest.get("schema") != "legacy-windows-first-iteration-evidence-bundle":
                errors.append(f"{path.name}: manifest schema mismatch")
            raw_hostname = manifest.get("hostname")
            hostname = raw_hostname.strip() if isinstance(raw_hostname, str) else ""
            if not hostname:
                errors.append(f"{path.name}: manifest hostname is required")
                hostname = path.name
            artifacts = manifest.get("artifacts")
            if not isinstance(artifacts, list):
                artifacts = []
                errors.append(f"{path.name}: manifest artifacts must be a list")
            roles: dict[str, dict[str, Any]] = {}
            seen_artifact_roles: set[str] = set()
            seen_artifact_paths: set[str] = set()
            for item in artifacts:
                if not isinstance(item, dict):
                    errors.append(f"{path.name}: manifest artifact entries must be objects")
                    continue
                role = str(item.get("role") or "")
                archive_path = str(item.get("archive_path") or "")
                if role in seen_artifact_roles:
                    errors.append(f"{path.name}: duplicate manifest artifact role: {role}")
                seen_artifact_roles.add(role)
                if archive_path in seen_artifact_paths:
                    errors.append(f"{path.name}: duplicate manifest artifact archive_path: {archive_path}")
                seen_artifact_paths.add(archive_path)
                roles[role] = item
            for role in sorted(REQUIRED_ARTIFACT_ROLES):
                item = roles.get(role)
                if not item:
                    errors.append(f"{hostname}: missing required evidence role: {role}")
                    continue
                errors.extend(_validate_manifest_entry(zf, names, hostname, role, item))
            for role in sorted(WIN7_BRIDGE_ROLES | {WIN7_PROBE_ROLE}):
                item = roles.get(role)
                if item:
                    errors.extend(_validate_manifest_entry(zf, names, hostname, role, item))

            docs = manifest.get("docs")
            if not isinstance(docs, list):
                docs = []
                errors.append(f"{path.name}: manifest docs must be a list")
            for item in docs:
                if not isinstance(item, dict):
                    continue
                role = str(item.get("role") or "release_doc")
                errors.extend(_validate_manifest_entry(zf, names, hostname, role, item))

            smoke_role = roles.get("lab_smoke") or {}
            smoke_path = str(smoke_role.get("archive_path") or "artifacts/legacy-windows-lab-smoke.json")
            smoke, smoke_error = _read_json(zf, smoke_path)
            if smoke_error:
                errors.append(f"{hostname}: {smoke_error}")
                smoke = {}
            assert smoke is not None
            os_kind = _classify_legacy_os(smoke)
            if os_kind == "unknown":
                errors.append(f"{hostname}: evidence bundle is not Windows 7 SP1 or Windows Server 2008 R2 SP1")
            manifest_os_kind = str(manifest.get("os_kind") or "").strip()
            if not manifest_os_kind:
                errors.append(f"{hostname}: manifest os_kind is required")
            elif manifest_os_kind != os_kind:
                errors.append(f"{hostname}: manifest os_kind {manifest_os_kind} does not match smoke-derived {os_kind}")
            smoke_checks = _smoke_check_map(smoke)
            smoke_details = _smoke_check_detail_map(smoke)
            for check in sorted(REQUIRED_SMOKE_CHECKS):
                if smoke_checks.get(check) is not True:
                    errors.append(f"{hostname}: required smoke check failed or missing: {check}")
            if smoke.get("exercise_peer_recover") is not True:
                errors.append(f"{hostname}: signed peer_recover was not exercised")
            if (
                smoke_checks.get("peer_recover") is True
                and "fresh restart-agent.flag observed" not in smoke_details.get("peer_recover", "")
            ):
                errors.append(f"{hostname}: peer_recover detail does not prove fresh restart-agent.flag")
            if smoke.get("exercise_self_rescue") is not True:
                errors.append(f"{hostname}: signed self_rescue was not exercised")
            if (
                smoke_checks.get("self_rescue") is True
                and "fresh restart-agent.flag and update-flag.txt observed" not in smoke_details.get("self_rescue", "")
            ):
                errors.append(f"{hostname}: self_rescue detail does not prove fresh restart-agent.flag/update-flag.txt")

            operator_context_role = roles.get("collect_file:operator-context.json") or {}
            operator_context_path = str(operator_context_role.get("archive_path") or "")
            operator_context_summary: dict[str, Any] = {}
            if operator_context_path:
                operator_context, operator_context_error = _read_json(zf, operator_context_path)
                if operator_context_error:
                    errors.append(f"{hostname}: {operator_context_error}")
                    operator_context = {}
                assert operator_context is not None
                errors.extend(_validate_operator_context(hostname, operator_context))
                operator_context_summary = _operator_context_summary(operator_context)
            collector_summary = _collector_summary(zf, roles)
            errors.extend(_validate_collector_exit_status(hostname, collector_summary))
            errors.extend(_validate_collector_build_manifest(zf, hostname, roles, collector_summary))

            if os_kind == "windows_7_sp1":
                missing_bridge = [role for role in sorted(WIN7_BRIDGE_ROLES) if role not in roles]
                if missing_bridge:
                    errors.append(f"{hostname}: missing Windows 10 bridge launch artifacts")
                if WIN7_PROBE_ROLE not in roles:
                    errors.append(f"{hostname}: missing Windows 10 probe evidence")
                win10_bridge = _win10_bridge_summary(zf, roles)
                win10_probe = _win10_probe_summary(zf, roles)
                if WIN7_PROBE_ROLE in roles and win10_probe.get("verdict") != "candidate":
                    errors.append(f"{hostname}: Windows 10 probe verdict is not candidate")
                missing_probe_context = _missing_win10_probe_context(win10_probe)
                if WIN7_PROBE_ROLE in roles and missing_probe_context:
                    errors.append(
                        f"{hostname}: Windows 10 probe missing hardware context: {', '.join(missing_probe_context)}"
                    )
                if (
                    WIN7_PROBE_ROLE in roles
                    and "total_ram_gb" not in missing_probe_context
                    and not _positive_probe_ram_gb(win10_probe)
                ):
                    errors.append(f"{hostname}: Windows 10 probe total_ram_gb must be greater than 0")
                if win10_bridge.get("acknowledgements_recorded") is not True:
                    errors.append(f"{hostname}: Windows 10 bridge acknowledgement proof is missing")
                if win10_bridge.get("evidence_notes_recorded") is not True:
                    errors.append(f"{hostname}: Windows 10 bridge evidence note proof is missing")
                if win10_bridge.get("media_url_query_redacted") is not True:
                    errors.append(f"{hostname}: Windows 10 bridge media URL redaction proof is invalid")
                if win10_bridge.get("setup_launched") is not True:
                    errors.append(f"{hostname}: Windows 10 bridge setup launch proof is missing")
                if win10_bridge.get("stage") not in {"setup_running", "completed"}:
                    errors.append(f"{hostname}: Windows 10 bridge stage proof is invalid")
                if win10_bridge.get("percent_valid") is not True:
                    errors.append(f"{hostname}: Windows 10 bridge percent proof is invalid")
                if win10_bridge.get("probe_verdict") != "candidate":
                    errors.append(f"{hostname}: Windows 10 bridge status probe verdict proof is invalid")
            else:
                win10_bridge = {}
                win10_probe = {}
                present_bridge = [role for role in sorted(WIN7_BRIDGE_ROLES) if role in roles]
                if os_kind == "server_2008_r2" and present_bridge:
                    errors.append(
                        f"{hostname}: Windows Server 2008 R2 SP1 bundle must not include Windows 10 bridge artifacts"
                    )
                present_win10_probe = WIN7_PROBE_ROLE in roles
                if os_kind == "server_2008_r2" and present_win10_probe:
                    errors.append(
                        f"{hostname}: Windows Server 2008 R2 SP1 bundle must not include Windows 10 probe evidence"
                    )

            return {
                "path": str(path),
                "collector": collector_summary,
                "hostname": hostname,
                "operator_context": operator_context_summary,
                "os_kind": os_kind,
                "roles": sorted(roles),
                "smoke_checks": _required_smoke_summary(smoke_checks),
                "win10_bridge": win10_bridge,
                "win10_probe": win10_probe,
            }, errors
    except zipfile.BadZipFile as exc:
        return None, [f"{path.name}: cannot open evidence ZIP: {exc}"]


def validate(paths: list[Path]) -> tuple[list[dict[str, Any]], list[str]]:
    summaries: list[dict[str, Any]] = []
    errors: list[str] = []
    for path in paths:
        summary, bundle_errors = _bundle_summary(path)
        errors.extend(bundle_errors)
        if summary:
            summaries.append(summary)

    if not any(item.get("os_kind") == "windows_7_sp1" for item in summaries):
        errors.append("missing Windows 7 SP1 evidence bundle")
    if not any(item.get("os_kind") == "server_2008_r2" for item in summaries):
        errors.append("missing Windows Server 2008 R2 SP1 evidence bundle")
    return summaries, errors


def main(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("bundle", nargs="+", type=Path, help="Retained legacy Windows evidence bundle ZIP(s)")
    args = parser.parse_args(argv)

    summaries, errors = validate(args.bundle)
    if errors:
        print("legacy Windows first-iteration readiness failed", file=sys.stderr)
        for error in errors:
            print(f"- {error}", file=sys.stderr)
        return 2

    win7 = next(item for item in summaries if item.get("os_kind") == "windows_7_sp1")
    server = next(item for item in summaries if item.get("os_kind") == "server_2008_r2")
    print(f"Windows 7 SP1: {win7['hostname']}")
    print(f"Windows Server 2008 R2 SP1: {server['hostname']}")
    print("legacy Windows first-iteration readiness passed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))
