fix(reboot): accept windows99 console verify artifacts

apps/api/src/services/reboot_auto_recovery_slo_scorecard.py

@@ -1853,6 +1853,11 @@ def _build_windows99_verify_collection_packet(
         "no_secret_collector_collect_command": (
             "bash scripts/reboot-recovery/collect-windows99-vmware-verify.sh --collect"
         ),
+        "no_secret_console_artifact_validate_command": (
+            "python3.11 scripts/reboot-recovery/"
+            "validate-windows99-console-verify-artifact.py --input "
+            "<console-stdout.txt> --output <windows99-vmware-verify.txt>"
+        ),
         "post_verifier": (
             "rerun_reboot_auto_recovery_slo_scorecard_with_"
             "windows99_vmware_file_no_secret_no_reboot"
@@ -1860,6 +1865,7 @@ def _build_windows99_verify_collection_packet(
         "collection_blockers": collection_blockers,
         "safe_collection_channels": [
             "authorized_windows99_console_verify_stdout_only",
+            "validated_windows99_console_verify_artifact_file",
             "existing_management_channel_verify_mode_only",
             "no_secret_ssh_batchmode_verify_collector",
             "committed_no_secret_artifact_file_then_scorecard_rerun",

apps/api/tests/test_reboot_auto_recovery_slo_scorecard_api.py

@@ -1027,6 +1027,13 @@ def _assert_reboot_slo_payload(payload: dict):
     assert collection["no_secret_collector_collect_command"] == (
         "bash scripts/reboot-recovery/collect-windows99-vmware-verify.sh --collect"
     )
+    console_validate_command = collection["no_secret_console_artifact_validate_command"]
+    assert "validate-windows99-console-verify-artifact.py" in console_validate_command
+    assert "--input <console-stdout.txt>" in console_validate_command
+    assert "--output <windows99-vmware-verify.txt>" in console_validate_command
+    assert "validated_windows99_console_verify_artifact_file" in collection[
+        "safe_collection_channels"
+    ]
     assert (
         "no_secret_ssh_batchmode_verify_collector"
         in collection["safe_collection_channels"]

scripts/reboot-recovery/reboot-auto-recovery-slo-scorecard.py

@@ -967,6 +967,11 @@ def build_windows99_verify_collection_packet(
         "no_secret_collector_collect_command": (
             "bash scripts/reboot-recovery/collect-windows99-vmware-verify.sh --collect"
         ),
+        "no_secret_console_artifact_validate_command": (
+            "python3.11 scripts/reboot-recovery/"
+            "validate-windows99-console-verify-artifact.py --input "
+            "<console-stdout.txt> --output <windows99-vmware-verify.txt>"
+        ),
         "post_verifier": (
             "rerun_reboot_auto_recovery_slo_scorecard_with_"
             "windows99_vmware_file_no_secret_no_reboot"
@@ -974,6 +979,7 @@ def build_windows99_verify_collection_packet(
         "collection_blockers": collection_blockers,
         "safe_collection_channels": [
             "authorized_windows99_console_verify_stdout_only",
+            "validated_windows99_console_verify_artifact_file",
             "existing_management_channel_verify_mode_only",
             "no_secret_ssh_batchmode_verify_collector",
             "committed_no_secret_artifact_file_then_scorecard_rerun",

scripts/reboot-recovery/tests/test_reboot_auto_recovery_slo_scorecard.py

@@ -466,6 +466,15 @@ def test_missing_windows99_vmware_readback_fails_closed(tmp_path: Path) -> None:
     assert payload["windows99_verify_collection"][
         "no_secret_collector_collect_command"
     ] == "bash scripts/reboot-recovery/collect-windows99-vmware-verify.sh --collect"
+    console_validate_command = payload["windows99_verify_collection"][
+        "no_secret_console_artifact_validate_command"
+    ]
+    assert "validate-windows99-console-verify-artifact.py" in console_validate_command
+    assert "--input <console-stdout.txt>" in console_validate_command
+    assert "--output <windows99-vmware-verify.txt>" in console_validate_command
+    assert "validated_windows99_console_verify_artifact_file" in payload[
+        "windows99_verify_collection"
+    ]["safe_collection_channels"]
     assert "no_secret_ssh_batchmode_verify_collector" in payload[
         "windows99_verify_collection"
     ]["safe_collection_channels"]

scripts/reboot-recovery/tests/test_windows99_console_verify_artifact_validator.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import subprocess
+import textwrap
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = (
+    ROOT
+    / "scripts"
+    / "reboot-recovery"
+    / "validate-windows99-console-verify-artifact.py"
+)
+
+
+VALID_RED_VERIFY = """\
+AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1
+MODE=Verify
+VMRUN_PRESENT=1
+HOST111_VMWARE_TARGET=explicit_or_required
+VMX alias=111 path= present=0
+VMX alias=188 path=D:\\Documents\\Virtual Machines\\188\\188.vmx present=1
+VMX alias=120 path=D:\\Documents\\Virtual Machines\\120\\120.vmx present=1
+VMX alias=121 path=D:\\Documents\\Virtual Machines\\121\\121.vmx present=1
+VMX alias=112 path=D:\\Downloads\\kali\\kali.vmx present=1
+VMWARE_SERVICE name=VMAuthdService present=1 status=Running startup_type=Auto ok=1
+VMWARE_SERVICE name=VMnetDHCP present=1 status=Running startup_type=Auto ok=1
+VMWARE_AUTOSTART_TASK name=AWOOOI-Start-VMware-VMs present=1 enabled=1 state=Ready trigger_count=1 action_count=1 ok=1
+WINDOWS_UPDATE_POLICY name=NoAutoRebootWithLoggedOnUsers value=1 expected=1 ok=1
+VM_POWER alias=111 vmx_present=0 running=0
+VM_POWER alias=188 vmx_present=1 running=0
+VM_POWER alias=120 vmx_present=1 running=0
+VM_POWER alias=121 vmx_present=1 running=0
+VM_POWER alias=112 vmx_present=1 running=0
+VMWARE_AUTOSTART_CONFIG_READY=0
+VMWARE_AUTOSTART_POWER_READY=0
+WINDOWS_UPDATE_NO_AUTO_REBOOT_READY=1
+VMWARE_AUTOSTART_VERIFY_READY=0
+MISSING_VMX_ALIASES=111
+"""
+
+
+def _run(input_path: Path, output_path: Path | None = None) -> subprocess.CompletedProcess[str]:
+    cmd = ["python3.11", str(SCRIPT), "--input", str(input_path)]
+    if output_path is not None:
+        cmd.extend(["--output", str(output_path)])
+    return subprocess.run(
+        cmd,
+        cwd=ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def _kv(stdout: str) -> dict[str, str]:
+    values: dict[str, str] = {}
+    for line in stdout.splitlines():
+        if "=" in line:
+            key, value = line.split("=", 1)
+            values[key] = value
+    return values
+
+
+def test_valid_console_verify_red_artifact_is_accepted_and_normalized(
+    tmp_path: Path,
+) -> None:
+    raw = tmp_path / "console.txt"
+    normalized = tmp_path / "windows99-vmware-verify.txt"
+    raw.write_text(VALID_RED_VERIFY.replace("\n", "\r\n"), encoding="utf-8")
+
+    result = _run(raw, normalized)
+
+    assert result.returncode == 0
+    values = _kv(result.stdout)
+    assert values["schema_version"] == "windows99_console_verify_artifact_validator_v1"
+    assert values["status"] == "valid_console_verify_artifact_vmware_not_ready"
+    assert values["missing_required_field_count"] == "0"
+    assert values["missing_vmx_aliases"] == "111"
+    assert values["powered_off_aliases"] == "111,188,120,121,112"
+    assert values["secret_value_read"] == "false"
+    assert normalized.read_text(encoding="utf-8").startswith(
+        "AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1\n"
+    )
+
+
+def test_collector_envelope_is_extracted(tmp_path: Path) -> None:
+    raw = tmp_path / "collector.txt"
+    normalized = tmp_path / "normalized.txt"
+    raw.write_text(
+        textwrap.dedent(
+            f"""
+            schema_version=windows99_vmware_verify_collector_v1
+            remote_verify_output_begin
+            {VALID_RED_VERIFY}
+            remote_verify_output_end
+            """
+        ).lstrip(),
+        encoding="utf-8",
+    )
+
+    result = _run(raw, normalized)
+
+    assert result.returncode == 0
+    values = _kv(result.stdout)
+    assert values["status"] == "valid_console_verify_artifact_vmware_not_ready"
+    assert "remote_verify_output_begin" not in normalized.read_text(encoding="utf-8")
+
+
+def test_secret_like_console_output_is_rejected(tmp_path: Path) -> None:
+    raw = tmp_path / "console.txt"
+    raw.write_text(VALID_RED_VERIFY + "password=not-allowed\n", encoding="utf-8")
+
+    result = _run(raw)
+
+    assert result.returncode == 76
+    values = _kv(result.stdout)
+    assert values["status"] == "blocked_secret_like_console_output"
+    assert values["secret_like_line_count"] == "1"
+
+
+def test_missing_required_fields_fail_closed(tmp_path: Path) -> None:
+    raw = tmp_path / "console.txt"
+    raw.write_text("AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1\nMODE=Verify\n", encoding="utf-8")
+
+    result = _run(raw)
+
+    assert result.returncode == 75
+    values = _kv(result.stdout)
+    assert values["status"] == "blocked_missing_required_console_verify_fields"
+    assert int(values["missing_required_field_count"]) > 0

scripts/reboot-recovery/validate-windows99-console-verify-artifact.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+"""Validate no-secret Windows99 VMware Verify stdout captured from console."""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from pathlib import Path
+
+
+SCHEMA_VERSION = "windows99_console_verify_artifact_validator_v1"
+DEFAULT_REQUIRED_ALIASES = ("111", "188", "120", "121", "112")
+SENSITIVE_RE = re.compile(
+    r"(?i)(password|passwd|secret|token|authorization|cookie|private[_ -]?key)\s*[=:]"
+)
+
+
+def _normalize(text: str) -> list[str]:
+    text = text.replace("\r\n", "\n").replace("\r", "\n")
+    lines = [line.strip() for line in text.splitlines()]
+    if "remote_verify_output_begin" in lines and "remote_verify_output_end" in lines:
+        begin = lines.index("remote_verify_output_begin") + 1
+        end = lines.index("remote_verify_output_end")
+        lines = lines[begin:end]
+    return [line for line in lines if line]
+
+
+def _value_for(lines: list[str], key: str) -> str:
+    prefix = f"{key}="
+    for line in lines:
+        if line.startswith(prefix):
+            return line.split("=", 1)[1]
+    return ""
+
+
+def _line_matches(lines: list[str], pattern: str) -> bool:
+    compiled = re.compile(pattern)
+    return any(compiled.search(line) for line in lines)
+
+
+def _csv(items: list[str]) -> str:
+    return ",".join(items)
+
+
+def validate(lines: list[str], required_aliases: tuple[str, ...]) -> tuple[int, list[str]]:
+    missing_fields: list[str] = []
+    secret_like_lines = [line for line in lines if SENSITIVE_RE.search(line)]
+
+    required_exact_keys = [
+        "AWOOOI_WINDOWS99_VMWARE_AUTOSTART",
+        "MODE",
+        "VMRUN_PRESENT",
+        "VMWARE_AUTOSTART_CONFIG_READY",
+        "VMWARE_AUTOSTART_POWER_READY",
+        "WINDOWS_UPDATE_NO_AUTO_REBOOT_READY",
+        "VMWARE_AUTOSTART_VERIFY_READY",
+    ]
+    for key in required_exact_keys:
+        if not _value_for(lines, key):
+            missing_fields.append(key)
+
+    if _value_for(lines, "AWOOOI_WINDOWS99_VMWARE_AUTOSTART") != "1":
+        missing_fields.append("AWOOOI_WINDOWS99_VMWARE_AUTOSTART=1")
+    if _value_for(lines, "MODE") != "Verify":
+        missing_fields.append("MODE=Verify")
+
+    if not any(line.startswith("VMWARE_SERVICE ") for line in lines):
+        missing_fields.append("VMWARE_SERVICE")
+    if not any(line.startswith("VMWARE_AUTOSTART_TASK ") for line in lines):
+        missing_fields.append("VMWARE_AUTOSTART_TASK")
+    if not any(line.startswith("WINDOWS_UPDATE_POLICY ") for line in lines):
+        missing_fields.append("WINDOWS_UPDATE_POLICY")
+
+    missing_vmx_aliases: list[str] = []
+    powered_off_aliases: list[str] = []
+    for alias in required_aliases:
+        vmx_pattern = rf"^VMX alias={re.escape(alias)}\s+.*\spresent=([01])$"
+        power_pattern = rf"^VM_POWER alias={re.escape(alias)}\s+.*\srunning=([01])$"
+        vmx_line = next((line for line in lines if re.search(vmx_pattern, line)), "")
+        power_line = next((line for line in lines if re.search(power_pattern, line)), "")
+        if not vmx_line:
+            missing_fields.append(f"VMX alias={alias}")
+        elif not _line_matches([vmx_line], rf"\spresent=1$"):
+            missing_vmx_aliases.append(alias)
+        if not power_line:
+            missing_fields.append(f"VM_POWER alias={alias}")
+        elif not _line_matches([power_line], rf"\srunning=1$"):
+            powered_off_aliases.append(alias)
+
+    status = "ready_console_verify_artifact"
+    exit_code = 0
+    if secret_like_lines:
+        status = "blocked_secret_like_console_output"
+        exit_code = 76
+    elif missing_fields:
+        status = "blocked_missing_required_console_verify_fields"
+        exit_code = 75
+    elif _value_for(lines, "VMWARE_AUTOSTART_VERIFY_READY") != "1":
+        status = "valid_console_verify_artifact_vmware_not_ready"
+
+    summary = [
+        f"schema_version={SCHEMA_VERSION}",
+        f"status={status}",
+        f"required_aliases={_csv(list(required_aliases))}",
+        f"missing_required_field_count={len(missing_fields)}",
+        f"missing_required_fields={_csv(missing_fields)}",
+        f"secret_like_line_count={len(secret_like_lines)}",
+        f"missing_vmx_aliases={_csv(missing_vmx_aliases)}",
+        f"powered_off_aliases={_csv(powered_off_aliases)}",
+        f"vmware_autostart_verify_ready={_value_for(lines, 'VMWARE_AUTOSTART_VERIFY_READY') or 'unknown'}",
+        "secret_value_read=false",
+        "password_prompt_allowed=false",
+        "remote_write_performed=false",
+        "host_reboot_performed=false",
+        "vm_power_change_performed=false",
+        "windows_update_policy_apply_performed=false",
+    ]
+    return exit_code, summary
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Validate no-secret stdout from windows99-vmware-autostart.ps1 -Mode Verify "
+            "captured through RDP / Hyper-V / console."
+        )
+    )
+    parser.add_argument("--input", required=True, help="Raw console stdout file.")
+    parser.add_argument(
+        "--output",
+        help="Optional normalized artifact for --windows99-vmware-file.",
+    )
+    parser.add_argument(
+        "--required-aliases",
+        default=",".join(DEFAULT_REQUIRED_ALIASES),
+        help="Comma or space separated required VM aliases.",
+    )
+    args = parser.parse_args()
+
+    required_aliases = tuple(
+        item
+        for item in re.split(r"[\s,]+", args.required_aliases.strip())
+        if item
+    ) or DEFAULT_REQUIRED_ALIASES
+    input_path = Path(args.input)
+    raw_text = input_path.read_text(encoding="utf-8", errors="replace")
+    lines = _normalize(raw_text)
+    exit_code, summary = validate(lines, required_aliases)
+
+    if exit_code == 0 and args.output:
+        output_path = Path(args.output)
+        output_path.write_text("\n".join(lines + [""]) , encoding="utf-8")
+
+    for line in summary:
+        print(line)
+    if args.output:
+        print(f"normalized_output={args.output if exit_code == 0 else ''}")
+    return exit_code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())