awoooi/scripts/reboot-recovery/windows99-management-channel-probe.py

#!/usr/bin/env python3
"""No-secret Windows 99 management-channel readback.

This probe only checks whether a command channel is available for collecting the
Windows 99 VMware verifier. It does not read credentials, start VMs, or change
Windows/VMware state.
"""

from __future__ import annotations

import argparse
import json
import os
import shutil
import socket
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


SCHEMA_VERSION = "windows99_management_channel_readback_v1"
DEFAULT_PORTS = (22, 135, 445, 2179, 3389, 5985, 5986)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Probe no-secret management channels for Windows host 99.",
    )
    parser.add_argument("--host", default="192.168.0.99")
    parser.add_argument(
        "--ssh-user",
        action="append",
        dest="ssh_users",
        help="SSH user to probe with BatchMode public-key auth. May be repeated.",
    )
    parser.add_argument("--tcp-timeout", type=float, default=2.0)
    parser.add_argument("--ssh-timeout", type=int, default=8)
    parser.add_argument(
        "--port",
        action="append",
        type=int,
        dest="ports",
        help="TCP port to probe. May be passed more than once.",
    )
    parser.add_argument("--skip-ssh", action="store_true")
    parser.add_argument(
        "--console-artifact-status",
        default="not_attempted",
        choices=[
            "not_attempted",
            "blocked_clipboard_unreliable",
            "blocked_focus_unreliable",
            "blocked_truncated_output",
            "collected_stdout",
            "validated_artifact",
        ],
        help=(
            "Optional no-secret console artifact collection status. Use a blocked "
            "value only after an attempted console stdout capture path fails."
        ),
    )
    parser.add_argument(
        "--console-artifact-blocker",
        action="append",
        dest="console_artifact_blockers",
        help="Optional machine-readable console artifact blocker. May be repeated.",
    )
    parser.add_argument("--generated-at", help="Override generated_at.")
    parser.add_argument("--output", type=Path, help="Write JSON to this path.")
    args = parser.parse_args()
    if not args.ssh_users:
        env_users = os.environ.get("WINDOWS99_SSH_USERS", "").split()
        args.ssh_users = env_users or ["administrator"]
    args.ssh_user = args.ssh_users[0]
    return args


def tcp_status(host: str, port: int, timeout: float) -> str:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return "open"
    except TimeoutError:
        return "timeout"
    except ConnectionRefusedError:
        return "refused"
    except OSError as exc:
        name = exc.__class__.__name__
        if getattr(exc, "errno", None) is not None:
            return f"{name}_{exc.errno}"
        return name


def ping_status(host: str) -> dict[str, Any]:
    if not shutil.which("ping"):
        return {"checked": False, "ok": False, "status": "ping_missing"}
    timeout_arg = "1000" if sys.platform == "darwin" else "1"
    command = ["ping", "-c", "2", "-W", timeout_arg, host]
    try:
        result = subprocess.run(
            command,
            text=True,
            capture_output=True,
            timeout=5,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return {"checked": True, "ok": False, "status": "timeout"}
    return {
        "checked": True,
        "ok": result.returncode == 0,
        "status": "ok" if result.returncode == 0 else "failed",
    }


def classify_ssh_failure(stderr: str, returncode: int) -> str:
    lowered = stderr.lower()
    if "permission denied" in lowered:
        return "permission_denied"
    if "connection timed out" in lowered or "operation timed out" in lowered:
        return "timeout"
    if "connection refused" in lowered:
        return "refused"
    if "no route to host" in lowered:
        return "no_route"
    if returncode == 124:
        return "timeout"
    return "failed"


def ssh_batch_status(host: str, user: str, timeout: int, port_open: bool) -> dict[str, Any]:
    if not port_open:
        return {"checked": False, "ready": False, "status": "port_not_open"}
    ssh = shutil.which("ssh")
    if not ssh:
        return {"checked": False, "ready": False, "status": "ssh_missing"}
    command = [
        ssh,
        "-o",
        "BatchMode=yes",
        "-o",
        f"ConnectTimeout={timeout}",
        "-o",
        "PreferredAuthentications=publickey",
        "-o",
        "StrictHostKeyChecking=no",
        "-o",
        "UserKnownHostsFile=/dev/null",
        "-o",
        "GlobalKnownHostsFile=/dev/null",
        f"{user}@{host}",
        "cmd",
        "/c",
        "echo",
        "AWOOOI_WINDOWS99_SSH_OK",
    ]
    try:
        result = subprocess.run(
            command,
            text=True,
            capture_output=True,
            timeout=timeout + 3,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return {"checked": True, "ready": False, "status": "timeout"}

    ready = (
        result.returncode == 0
        and "AWOOOI_WINDOWS99_SSH_OK" in (result.stdout or "")
    )
    return {
        "checked": True,
        "ready": ready,
        "status": "ready"
        if ready
        else classify_ssh_failure(result.stderr or "", result.returncode),
    }


def build_payload(args: argparse.Namespace) -> dict[str, Any]:
    ports = tuple(args.ports or DEFAULT_PORTS)
    ssh_users = list(
        getattr(args, "ssh_users", None)
        or [getattr(args, "ssh_user", "administrator")]
    )
    if not ssh_users:
        ssh_users = ["administrator"]
    generated_at = args.generated_at or datetime.now().astimezone().isoformat(timespec="seconds")
    tcp_ports = {
        str(port): tcp_status(args.host, port, args.tcp_timeout)
        for port in ports
    }
    ping = ping_status(args.host)
    host_reachable = ping["ok"] or any(status == "open" for status in tcp_ports.values())
    winrm_http_open = tcp_ports.get("5985") == "open"
    winrm_https_open = tcp_ports.get("5986") == "open"
    hyperv_vmconnect_open = tcp_ports.get("2179") == "open"
    rdp_console_reachable = tcp_ports.get("3389") == "open"
    console_collection_channels = []
    if rdp_console_reachable:
        console_collection_channels.append("rdp_console")
    if hyperv_vmconnect_open:
        console_collection_channels.append("hyperv_vmconnect")
    local_console_channel_reachable = bool(console_collection_channels)
    ssh_batch_candidates: list[dict[str, Any]] = []
    if args.skip_ssh:
        ssh_probe = {"checked": False, "ready": False, "status": "skipped"}
    else:
        for user in ssh_users:
            candidate = ssh_batch_status(
                args.host,
                user,
                args.ssh_timeout,
                tcp_ports.get("22") == "open",
            )
            candidate = {"user": user, **candidate}
            ssh_batch_candidates.append(candidate)
            if candidate["ready"] is True:
                break
        ready_candidate = next(
            (candidate for candidate in ssh_batch_candidates if candidate["ready"] is True),
            None,
        )
        ssh_probe = dict(ready_candidate or ssh_batch_candidates[0])
        ssh_probe.pop("user", None)
    remote_execution_ready = ssh_probe["ready"] is True
    console_artifact_status = str(
        getattr(args, "console_artifact_status", "not_attempted")
        or "not_attempted"
    )
    console_artifact_blockers = list(
        getattr(args, "console_artifact_blockers", None) or []
    )
    if console_artifact_status == "blocked_clipboard_unreliable":
        console_artifact_blockers.append("windows99_console_clipboard_unreliable")
    elif console_artifact_status == "blocked_focus_unreliable":
        console_artifact_blockers.append("windows99_console_focus_unreliable")
    elif console_artifact_status == "blocked_truncated_output":
        console_artifact_blockers.append("windows99_console_verify_output_truncated")
    console_artifact_blockers = list(dict.fromkeys(console_artifact_blockers))
    console_artifact_reliable = console_artifact_status in {
        "collected_stdout",
        "validated_artifact",
    }
    blockers: list[str] = []
    if not host_reachable:
        blockers.append("windows99_host_unreachable_from_management_probe")
    if not remote_execution_ready:
        blockers.append("windows99_remote_execution_channel_unavailable")
    if not (winrm_http_open or winrm_https_open):
        blockers.append("windows99_winrm_unavailable")
    if tcp_ports.get("22") == "open" and ssh_probe["status"] == "permission_denied":
        blockers.append("windows99_ssh_batch_denied")
    blockers.extend(console_artifact_blockers)

    return {
        "schema_version": SCHEMA_VERSION,
        "readback_present": True,
        "generated_at": generated_at,
        "host": args.host,
        "ping": ping,
        "host_reachable": host_reachable,
        "tcp_ports": tcp_ports,
        "ssh_user": (
            next(
                (
                    candidate["user"]
                    for candidate in ssh_batch_candidates
                    if candidate.get("ready") is True
                ),
                ssh_users[0],
            )
            if ssh_users
            else ""
        ),
        "ssh_users": ssh_users,
        "ssh_batch": ssh_probe,
        "ssh_batch_candidates": ssh_batch_candidates,
        "winrm_http_open": winrm_http_open,
        "winrm_https_open": winrm_https_open,
        "hyperv_vmconnect_open": hyperv_vmconnect_open,
        "rdp_console_reachable": rdp_console_reachable,
        "local_console_channel_reachable": local_console_channel_reachable,
        "console_collection_channels": console_collection_channels,
        "console_artifact_status": console_artifact_status,
        "console_artifact_reliable": console_artifact_reliable,
        "console_artifact_blockers": console_artifact_blockers,
        "console_artifact_safe_next_step": (
            "validate_collected_console_stdout_then_rerun_reboot_scorecard"
            if console_artifact_reliable
            else (
                "use_authorized_no_secret_management_channel_or_manual_console_stdout_capture"
                if console_artifact_blockers
                else "attempt_console_stdout_capture_only_if_focus_and_clipboard_are_reliable"
            )
        ),
        "remote_execution_channel_ready": remote_execution_ready,
        "can_collect_vmware_verify_without_secret": remote_execution_ready,
        "blockers": blockers,
        "forbidden_actions": [
            "read_windows_password",
            "read_secret_value",
            "start_vm",
            "reboot_host",
            "restart_service",
            "write_windows_policy",
        ],
    }


def main() -> int:
    args = parse_args()
    payload = build_payload(args)
    output = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    if args.output:
        args.output.write_text(output, encoding="utf-8")
    else:
        sys.stdout.write(output)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())