"""WinRM connectivity test for Windows hosts using domain admin credentials.""" from __future__ import annotations import socket from dataclasses import dataclass from app.models import Host class WinAdminNotConfiguredError(Exception): pass class HostNotWindowsError(Exception): pass class HostTargetMissingError(Exception): pass @dataclass(frozen=True) class WinRmTestResult: ok: bool message: str target: str hostname: str | None = None def resolve_windows_host_target(host: Host) -> str: if not is_windows_host(host): raise HostNotWindowsError("Host is not Windows") target = (host.ipv4 or "").strip() or (host.hostname or "").strip() if not target: raise HostTargetMissingError("Host has no IPv4 or hostname for WinRM test") return target def is_windows_host(host: Host) -> bool: if (host.os_family or "").strip().lower() == "windows": return True return (host.product or "").strip() == "rdp-login-monitor" def test_winrm_connection( *, target: str, user: str, password: str, timeout_sec: int = 15, ) -> WinRmTestResult: target = target.strip() user = user.strip() if not target or not user or not password: raise WinAdminNotConfiguredError("Windows admin credentials or target missing") try: import winrm except ImportError as exc: raise RuntimeError("pywinrm is not installed on SAC server") from exc endpoint = f"http://{target}:5985/wsman" try: session = winrm.Session( endpoint, auth=(user, password), transport="ntlm", read_timeout_sec=timeout_sec, operation_timeout_sec=timeout_sec, ) result = session.run_cmd("hostname") except winrm.exceptions.WinRMTransportError as exc: return WinRmTestResult( ok=False, message=f"WinRM transport error: {exc}", target=target, ) except (socket.timeout, TimeoutError): return WinRmTestResult( ok=False, message=f"WinRM timeout ({timeout_sec}s) to {target}:5985", target=target, ) except Exception as exc: return WinRmTestResult( ok=False, message=f"WinRM error: {exc}", target=target, ) stdout = (result.std_out or b"").decode("utf-8", errors="replace").strip() stderr = (result.std_err or b"").decode("utf-8", errors="replace").strip() if result.status_code == 0 and stdout: return WinRmTestResult( ok=True, message=f"WinRM OK, hostname={stdout}", target=target, hostname=stdout, ) detail = stderr or stdout or f"exit code {result.status_code}" return WinRmTestResult( ok=False, message=f"WinRM command failed: {detail}", target=target, )