fix(reboot): surface windows99 ssh candidate readback

scripts/reboot-recovery/tests/test_windows99_management_channel_probe.py

@@ -69,3 +69,63 @@ def test_management_probe_surfaces_no_secret_console_channels(monkeypatch):
     assert payload["can_collect_vmware_verify_without_secret"] is False
     assert "windows99_remote_execution_channel_unavailable" in payload["blockers"]
     assert "read_windows_password" in payload["forbidden_actions"]
+
+
+def test_management_probe_surfaces_multiple_ssh_candidates(monkeypatch):
+    module = _load_module()
+
+    def fake_tcp_status(_host: str, port: int, _timeout: float) -> str:
+        return "open" if port in {22, 3389} else "timeout"
+
+    def fake_ssh_batch_status(
+        _host: str,
+        user: str,
+        _timeout: int,
+        _port_open: bool,
+    ) -> dict[str, object]:
+        return {
+            "checked": True,
+            "ready": user == "wooo",
+            "status": "ready" if user == "wooo" else "permission_denied",
+        }
+
+    monkeypatch.setattr(module, "tcp_status", fake_tcp_status)
+    monkeypatch.setattr(
+        module,
+        "ping_status",
+        lambda _host: {"checked": True, "ok": True, "status": "ok"},
+    )
+    monkeypatch.setattr(module, "ssh_batch_status", fake_ssh_batch_status)
+
+    payload = module.build_payload(
+        argparse.Namespace(
+            host="192.168.0.99",
+            ssh_users=["administrator", "wooo", "ogt"],
+            tcp_timeout=0.01,
+            ssh_timeout=1,
+            ports=None,
+            skip_ssh=False,
+            generated_at="2026-07-02T16:00:00+08:00",
+            output=None,
+        )
+    )
+
+    assert payload["ssh_users"] == ["administrator", "wooo", "ogt"]
+    assert payload["ssh_user"] == "wooo"
+    assert payload["ssh_batch"]["status"] == "ready"
+    assert payload["remote_execution_channel_ready"] is True
+    assert payload["can_collect_vmware_verify_without_secret"] is True
+    assert payload["ssh_batch_candidates"] == [
+        {
+            "user": "administrator",
+            "checked": True,
+            "ready": False,
+            "status": "permission_denied",
+        },
+        {
+            "user": "wooo",
+            "checked": True,
+            "ready": True,
+            "status": "ready",
+        },
+    ]

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

@@ -10,6 +10,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import os
 import shutil
 import socket
 import subprocess
@@ -28,7 +29,12 @@ def parse_args() -> argparse.Namespace:
         description="Probe no-secret management channels for Windows host 99.",
     )
     parser.add_argument("--host", default="192.168.0.99")
-    parser.add_argument("--ssh-user", default="administrator")
+    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(
@@ -41,7 +47,12 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--skip-ssh", action="store_true")
     parser.add_argument("--generated-at", help="Override generated_at.")
     parser.add_argument("--output", type=Path, help="Write JSON to this path.")
-    return parser.parse_args()
+    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:
@@ -148,6 +159,12 @@ def ssh_batch_status(host: str, user: str, timeout: int, port_open: bool) -> dic
 
 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)
@@ -165,16 +182,27 @@ def build_payload(args: argparse.Namespace) -> dict[str, Any]:
     if hyperv_vmconnect_open:
         console_collection_channels.append("hyperv_vmconnect")
     local_console_channel_reachable = bool(console_collection_channels)
-    ssh_probe = (
-        {"checked": False, "ready": False, "status": "skipped"}
-        if args.skip_ssh
-        else ssh_batch_status(
-            args.host,
-            args.ssh_user,
-            args.ssh_timeout,
-            tcp_ports.get("22") == "open",
+    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
     blockers: list[str] = []
     if not host_reachable:
@@ -194,8 +222,21 @@ def build_payload(args: argparse.Namespace) -> dict[str, Any]:
         "ping": ping,
         "host_reachable": host_reachable,
         "tcp_ports": tcp_ports,
-        "ssh_user": args.ssh_user,
+        "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,