awoooi/scripts/reboot-recovery/tests/test_windows99_vmx_source_locator.py

from __future__ import annotations

import os
import subprocess
import textwrap
from pathlib import Path


ROOT = Path(__file__).resolve().parents[3]
BASH_SCRIPT = ROOT / "scripts" / "reboot-recovery" / "locate-windows99-vmx-source.sh"
PS_SCRIPT = ROOT / "scripts" / "reboot-recovery" / "windows99-vmx-source-locator.ps1"


def _write_executable(path: Path, text: str) -> None:
    path.write_text(textwrap.dedent(text).lstrip())
    path.chmod(0o755)


def _run_locator(fake_bin: Path, *args: str) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env["PATH"] = f"{fake_bin}:{env['PATH']}"
    env["WINDOWS99_KNOWN_HOSTS_FILE"] = str(fake_bin / "known_hosts")
    return subprocess.run(
        ["bash", str(BASH_SCRIPT), *args],
        cwd=ROOT,
        env=env,
        capture_output=True,
        text=True,
        check=False,
    )


def _key_values(stdout: str) -> dict[str, str]:
    values: dict[str, str] = {}
    for line in stdout.splitlines():
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key] = value
    return values


def test_locator_contract_forbids_secret_runtime_actions_and_vmx_content_reads() -> None:
    bash_text = BASH_SCRIPT.read_text()
    ps_text = PS_SCRIPT.read_text()

    assert "BatchMode=yes" in bash_text
    assert "PreferredAuthentications=publickey" in bash_text
    assert "PubkeyAuthentication=yes" in bash_text
    assert "PasswordAuthentication=no" in bash_text
    assert "KbdInteractiveAuthentication=no" in bash_text
    assert "NumberOfPasswordPrompts=0" in bash_text
    assert "--via-host" in bash_text
    assert "via_host=" in bash_text
    assert "in_memory_stdin_scriptblock" in bash_text
    assert "[Console]::In.ReadToEnd()" in bash_text
    assert "Get-ChildItem" in ps_text
    assert "Test-Path" in ps_text
    for text in [bash_text, ps_text]:
        for forbidden in [
            "sshpass",
            "PasswordAuthentication=yes",
            "KbdInteractiveAuthentication=yes",
            "net use",
            "shutdown",
            "Restart-Computer",
            "Start-VM",
            "vmrun start",
            "Set-ItemProperty",
            "Register-ScheduledTask",
            "Set-Content",
            "Copy-Item",
            "Move-Item",
            "Remove-Item",
            "Get-Content",
            "Select-String",
        ]:
            assert forbidden not in text


def test_check_mode_reports_auth_probe_without_remote_locator(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    _write_executable(
        fake_bin / "nc",
        """
        #!/usr/bin/env bash
        port="${!#}"
        if [[ "$port" == "22" || "$port" == "3389" ]]; then
          exit 0
        fi
        exit 1
        """,
    )
    _write_executable(
        fake_bin / "ssh",
        """
        #!/usr/bin/env bash
        printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
        exit 0
        """,
    )

    result = _run_locator(fake_bin, "--check", "--max-auth-users", "1")

    assert result.returncode == 0
    values = _key_values(result.stdout)
    assert values["schema_version"] == "windows99_vmx_source_locator_collector_v1"
    assert values["dry_run"] == "true"
    assert values["target_host_alias"] == "99"
    assert values["target_vm_alias"] == "111"
    assert values["ssh_batchmode_auth_ready"] == "1"
    assert values["remote_locator_attempted"] == "0"
    assert values["locator_collection_status"] == "ready_ssh_batchmode_auth_probe_only"
    assert values["secret_value_read"] == "false"
    assert values["password_prompt_allowed"] == "false"
    assert values["remote_write_performed"] == "false"
    assert values["host_reboot_performed"] == "false"
    assert values["vm_power_change_performed"] == "false"
    assert values["vmx_file_content_read"] == "false"
    assert values["raw_path_output"] == "false"


def test_collect_mode_blocks_without_publickey_auth(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    _write_executable(
        fake_bin / "nc",
        """
        #!/usr/bin/env bash
        [[ "${!#}" == "22" ]]
        """,
    )
    _write_executable(
        fake_bin / "ssh",
        """
        #!/usr/bin/env bash
        exit 255
        """,
    )

    result = _run_locator(fake_bin, "--collect", "--max-auth-users", "1")

    assert result.returncode == 75
    values = _key_values(result.stdout)
    assert values["ssh_batchmode_auth_ready"] == "0"
    assert values["remote_locator_attempted"] == "0"
    assert values["locator_collection_status"] == "blocked_ssh_publickey_auth_missing"
    assert values["secret_value_read"] == "false"
    assert values["remote_write_performed"] == "false"


def test_collect_mode_accepts_valid_remote_locator_stdout(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    _write_executable(
        fake_bin / "nc",
        """
        #!/usr/bin/env bash
        [[ "${!#}" == "22" ]]
        """,
    )
    _write_executable(
        fake_bin / "ssh",
        """
        #!/usr/bin/env bash
        joined="$*"
        if [[ "$joined" == *"AWOOOI_WINDOWS99_SSH_READY"* ]]; then
          printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
          exit 0
        fi
        cat >/dev/null
        printf '%s\n' 'EXPECTED_VMX alias=111 index=1 present=0 fingerprint=abc'
        printf '%s\n' 'AWOOOI_WINDOWS99_VMX_SOURCE_LOCATOR=1'
        printf '%s\n' 'schema_version=windows99_vmx_source_locator_v1'
        printf '%s\n' 'MODE=Check'
        printf '%s\n' 'target_host_alias=99'
        printf '%s\n' 'target_vm_alias=111'
        printf '%s\n' 'expected_vmx_path_count=1'
        printf '%s\n' 'expected_vmx_path_present_count=0'
        printf '%s\n' 'candidate_source_count=0'
        printf '%s\n' 'source_locator_status=blocked_missing_expected_vmx_source_no_candidate_found'
        printf '%s\n' 'file_content_read=false'
        printf '%s\n' 'remote_write_performed=false'
        exit 0
        """,
    )

    result = _run_locator(fake_bin, "--collect", "--max-auth-users", "1")

    assert result.returncode == 0
    values = _key_values(result.stdout)
    assert values["dry_run"] == "false"
    assert values["remote_locator_attempted"] == "1"
    assert values["remote_locator_exit_status"] == "0"
    assert values["locator_collection_status"] == (
        "collected_windows99_vmx_source_locator_stdout"
    )
    assert "remote_locator_output_begin" in result.stdout
    assert "source_locator_status=blocked_missing_expected_vmx_source_no_candidate_found" in result.stdout
    assert "remote_locator_output_end" in result.stdout


def test_collect_mode_supports_no_secret_via_host_path(tmp_path: Path) -> None:
    fake_bin = tmp_path / "bin"
    fake_bin.mkdir()
    _write_executable(
        fake_bin / "nc",
        """
        #!/usr/bin/env bash
        [[ "${!#}" == "22" ]]
        """,
    )
    _write_executable(
        fake_bin / "ssh",
        """
        #!/usr/bin/env bash
        joined="$*"
        if [[ "$joined" == *"AWOOOI_WINDOWS99_SSH_READY"* ]]; then
          printf '%s\n' 'AWOOOI_WINDOWS99_SSH_READY'
          exit 0
        fi
        if [[ "$joined" == *"powershell"* ]]; then
          cat >/dev/null
          printf '%s\n' 'EXPECTED_VMX alias=111 index=1 present=0 fingerprint=abc'
          printf '%s\n' 'AWOOOI_WINDOWS99_VMX_SOURCE_LOCATOR=1'
          printf '%s\n' 'schema_version=windows99_vmx_source_locator_v1'
          printf '%s\n' 'MODE=Check'
          printf '%s\n' 'target_host_alias=99'
          printf '%s\n' 'target_vm_alias=111'
          printf '%s\n' 'expected_vmx_path_count=1'
          printf '%s\n' 'expected_vmx_path_present_count=0'
          printf '%s\n' 'candidate_source_count=0'
          printf '%s\n' 'source_locator_status=blocked_missing_expected_vmx_source_no_candidate_found'
          printf '%s\n' 'file_content_read=false'
          printf '%s\n' 'remote_write_performed=false'
          exit 0
        fi
        exit 255
        """,
    )

    result = _run_locator(
        fake_bin,
        "--collect",
        "--via-host",
        "wooo@192.168.0.110",
        "--users",
        "wooo",
        "--max-auth-users",
        "1",
    )

    assert result.returncode == 0
    values = _key_values(result.stdout)
    assert values["via_host"] == "wooo@192.168.0.110"
    assert values["remote_locator_attempted"] == "1"
    assert values["locator_collection_status"] == (
        "collected_windows99_vmx_source_locator_stdout"
    )
    assert values["secret_value_read"] == "false"
    assert values["remote_write_performed"] == "false"
    assert values["vm_power_change_performed"] == "false"