Add windows ssh

coriolis/osmorphing/osmount/windows.py

@@ -6,7 +6,7 @@
 
 from oslo_log import log as logging
 
-from coriolis import constants, exception, utils, wsman
+from coriolis import constants, exception, utils, windows_conn
 from coriolis.osmorphing.osmount import base
 
 LOG = logging.getLogger(__name__)
@@ -22,14 +22,19 @@ def __init__(self, *args, **kwargs):
 
     def _connect(self):
         connection_info = self._connection_info
-
         host = connection_info["ip"]
-        port = connection_info.get("port", 5986)
-        self._event_manager.progress_update(
-            "Connecting to WinRM host: %(host)s:%(port)s" % {"host": host, "port": port}
-        )
+        if windows_conn.uses_winrm(connection_info):
+            port = connection_info.get("port", windows_conn.WINRM_HTTPS_PORT)
+            self._event_manager.progress_update(
+                "Connecting to WinRM host: %(host)s:%(port)s"
+                % {"host": host, "port": port}
+            )
+        else:
+            self._event_manager.progress_update(
+                "Connecting through SSH to OSMorphing host on: %s" % host
+            )
 
-        self._conn = wsman.WSManConnection.from_connection_info(
+        self._conn = windows_conn.from_connection_info(
             connection_info, self._osmount_operation_timeout
         )
 

coriolis/providers/provider_utils.py

@@ -6,7 +6,7 @@
 import requests
 from oslo_log import log as logging
 
-from coriolis import constants, exception, utils, wsman
+from coriolis import constants, exception, utils, windows_conn, wsman
 
 LOG = logging.getLogger(__name__)
 
@@ -170,15 +170,19 @@ def _poll_instance_until_reachable_ssh(
     timeout: int = 600,
     poll_interval: int = 10,
 ):
+    if not connection_info.get("password") and not connection_info.get("pkey"):
+        raise exception.InvalidInput(
+            "SSH connection info must include password or pkey."
+        )
     start = time.time()
     while (time.time() - start) < timeout:
         try:
             ssh = utils.connect_ssh(
                 hostname=connection_info["ip"],
                 port=connection_info["port"],
                 username=connection_info["username"],
-                password=connection_info["password"],
-                pkey=connection_info["pkey"],
+                password=connection_info.get("password"),
+                pkey=connection_info.get("pkey"),
             )
             try:
                 # "exit 0" should work across platforms.
@@ -225,9 +229,17 @@ def _poll_instance_until_reachable_winrm(
     )
 
 
+def _protocol_from_connection_info(connection_info, protocol):
+    if protocol:
+        return protocol
+    if windows_conn.uses_winrm(connection_info):
+        return constants.PROTOCOL_WINRM
+    return constants.PROTOCOL_SSH
+
+
 def poll_instance_until_reachable(
     connection_info: dict,
-    protocol: str = constants.PROTOCOL_SSH,
+    protocol: str = None,
     timeout: int = 600,
     poll_interval: int = 10,
 ) -> paramiko.SSHClient:
@@ -239,22 +251,27 @@ def poll_instance_until_reachable(
         * username
         * password
         * pkey - Paramiko keypair
-    :param protocol: connection protocol, "ssh" or "winrm"
+    :param protocol: connection protocol, "ssh" or "winrm". If omitted,
+        port 5986 selects WinRM. Any other port selects SSH.
     :param timeout: the maximum amount of time to wait
     :param poll_interval: the amount of time to wait between retries
     """
-    # TODO(lpetrut): consider including the connection protocol in the
-    # connection info. We'd have to modify a few schemas used during os
-    # morphing. We currently pick the protocol based on the OS type but
-    # we may want to use SSH on Windows as well.
-    if protocol == constants.PROTOCOL_SSH:
-        helper = _poll_instance_until_reachable_ssh
-    elif protocol == constants.PROTOCOL_WINRM:
-        helper = _poll_instance_until_reachable_winrm
-    else:
-        raise exception.InvalidInput(
-            f"Unsupported instance connection protocol: {protocol}"
+    resolved = _protocol_from_connection_info(connection_info, protocol)
+    if resolved == constants.PROTOCOL_SSH:
+        ssh_connection_info = dict(connection_info)
+        if ssh_connection_info.get("port") is None:
+            ssh_connection_info["port"] = 22
+        return _poll_instance_until_reachable_ssh(
+            connection_info=ssh_connection_info,
+            timeout=timeout,
+            poll_interval=poll_interval,
+        )
+    if resolved == constants.PROTOCOL_WINRM:
+        return _poll_instance_until_reachable_winrm(
+            connection_info=connection_info,
+            timeout=timeout,
+            poll_interval=poll_interval,
         )
-    return helper(
-        connection_info=connection_info, timeout=timeout, poll_interval=poll_interval
+    raise exception.InvalidInput(
+        f"Unsupported instance connection protocol: {resolved}"
     )

coriolis/tests/osmorphing/osmount/test_windows.py

@@ -24,8 +24,8 @@ class CoriolisTestException(Exception):
 class WindowsMountToolsTestCase(test_base.CoriolisBaseTestCase):
     """Test suite for the WindowsMountTools class."""
 
-    @mock.patch.object(windows.wsman, 'WSManConnection')
-    def setUp(self, mock_wsman_connection):
+    @mock.patch.object(windows.windows_conn, 'from_connection_info')
+    def setUp(self, mock_from_connection_info):
         super(WindowsMountToolsTestCase, self).setUp()
         self.event_manager = mock.MagicMock()
         self.ssh = mock.MagicMock()
@@ -43,9 +43,9 @@ def setUp(self, mock_wsman_connection):
             mock.sentinel.ignore_devices,
             mock.sentinel.operation_timeout,
         )
-        self.tools._conn = mock_wsman_connection
+        self.tools._conn = mock.MagicMock()
 
-    @mock.patch.object(windows.wsman.WSManConnection, 'from_connection_info')
+    @mock.patch.object(windows.windows_conn, 'from_connection_info')
     def test__connect(self, mock_from_connection_info):
         result = self.tools._connect()
         self.assertIsNone(result)

coriolis/tests/providers/test_provider_utils.py

@@ -351,6 +351,21 @@ def test_poll_instance_ssh_timeout(
             poll_interval=poll_interval,
         )
 
+    def test_poll_instance_missing_auth(self):
+        connection_info = {
+            "ip": "1.2.3.4",
+            "port": 22,
+            "username": "Administrator",
+        }
+        self.assertRaises(
+            exception.InvalidInput,
+            provider_utils.poll_instance_until_reachable,
+            connection_info=connection_info,
+            protocol=constants.PROTOCOL_SSH,
+            timeout=600,
+            poll_interval=5,
+        )
+
     @mock.patch("coriolis.wsman.WSManConnection", new_callable=mock.Mock)
     @mock.patch("time.sleep")
     def test_poll_instance_winrm(
@@ -400,3 +415,12 @@ def test_poll_instance_winrm_timeout(
             timeout=30,
             poll_interval=poll_interval,
         )
+
+    def test_poll_instance_unsupported_protocol(self):
+        connection_info = self._get_mock_conn_info()
+        self.assertRaises(
+            exception.InvalidInput,
+            provider_utils.poll_instance_until_reachable,
+            connection_info=connection_info,
+            protocol="ftp",
+        )

coriolis/tests/test_windows_conn.py

@@ -0,0 +1,30 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from unittest import mock
+
+from coriolis import windows_conn, windows_ssh, wsman
+from coriolis.tests import test_base
+
+
+class WindowsConnTestCase(test_base.CoriolisBaseTestCase):
+    def test_uses_winrm_default_port(self):
+        self.assertTrue(windows_conn.uses_winrm({"ip": "10.0.0.1"}))
+        self.assertTrue(windows_conn.uses_winrm({"ip": "10.0.0.1", "port": 5986}))
+        self.assertFalse(windows_conn.uses_winrm({"ip": "10.0.0.1", "port": 22}))
+
+    @mock.patch.object(wsman.WSManConnection, "from_connection_info")
+    @mock.patch.object(windows_ssh.WindowsSSHConnection, "from_connection_info")
+    def test_from_connection_info_winrm(self, mock_ssh, mock_winrm):
+        conn_info = {"ip": "10.0.0.1", "port": 5986, "username": "u", "password": "p"}
+        windows_conn.from_connection_info(conn_info, timeout=30)
+        mock_winrm.assert_called_once_with(conn_info, 30)
+        mock_ssh.assert_not_called()
+
+    @mock.patch.object(wsman.WSManConnection, "from_connection_info")
+    @mock.patch.object(windows_ssh.WindowsSSHConnection, "from_connection_info")
+    def test_from_connection_info_ssh(self, mock_ssh, mock_winrm):
+        conn_info = {"ip": "10.0.0.1", "port": 22, "username": "u", "password": "p"}
+        windows_conn.from_connection_info(conn_info, timeout=30)
+        mock_ssh.assert_called_once_with(conn_info, 30)
+        mock_winrm.assert_not_called()