Move virtio installation helper to coriolis core

The virtio installation helper is currently duplicated across all
the providers. We'll move it to Coriolis core instead.

coriolis/osmorphing/windows.py

@@ -10,6 +10,7 @@ import re
 import uuid
 
 from oslo_log import log as logging
+from packaging import version
 
 from coriolis import constants
 from coriolis import exception
@@ -705,6 +706,114 @@ class BaseWindowsMorphingTools(base.BaseOSMorphingTools):
         finally:
             self._unload_registry_hive("HKLM\\%s" % key_name)
 
+    def _add_virtio_drivers(self):
+        arch = "amd64"
+
+        CLIENT = 1
+        SERVER = 2
+
+        # Ordered by version number
+        virtio_dirs = [
+            ("xp", version.Version("5.1"), CLIENT),
+            ("2k3", version.Version("5.2"), SERVER),
+            ("2k8", version.Version("6.0"), SERVER | CLIENT),
+            ("w7", version.Version("6.1"), CLIENT),
+            ("2k8R2", version.Version("6.1"), SERVER),
+            ("w8", version.Version("6.2"), CLIENT),
+            ("2k12", version.Version("6.2"), SERVER),
+            ("w8.1", version.Version("6.3"), CLIENT),
+            ("2k12R2", version.Version("6.3"), SERVER),
+            # NOTE: although modern VirtIO releases feature a dedicated '2k19'
+            # directory for Server 2019, some just use 'w10' for both:
+            ("w10", version.Version("10.0"), SERVER | CLIENT),
+            # NOTE: this list is traversed in reverse, so the Server-specific
+            # one should come after the shared 'w10' ones:
+            ("2k16", version.Version("10.0.14393"), SERVER),
+            ("2k19", version.Version("10.0.17763"), SERVER),
+            ("2k22", version.Version("10.0.20348"), SERVER),
+            ("w11", version.Version("10.0.22000"), CLIENT),
+            ("2k25", version.Version("10.0.26100"), SERVER),
+        ]
+
+        # The list of all possible editions is huge, this is a simplification
+        if "Server" in self._edition_id:
+            edition_type = SERVER
+        else:
+            edition_type = CLIENT
+
+        drivers = [
+            "Balloon",
+            "NetKVM",
+            "qxl",
+            "qxldod",
+            "pvpanic",
+            "viorng",
+            "vioscsi",
+            "vioserial",
+            "viostor",
+            "viogpudo",
+        ]
+
+        self._event_manager.progress_update("Downloading virtio-win drivers")
+
+        virtio_iso_path = "c:\\virtio-win.iso"
+        virtio_iso_url = self._osmorphing_parameters["windows_virtio_iso_url"]
+        utils.retry_on_error(sleep_seconds=5)(self._conn.download_file)(
+            virtio_iso_url, virtio_iso_path
+        )
+
+        self._event_manager.progress_update("Adding virtio-win drivers")
+
+        virtio_drive = self._mount_disk_image(virtio_iso_path)
+        try:
+            virtio_dir = None
+
+            for main_dir, dir_version, dir_edition_type in reversed(
+                    virtio_dirs):
+                if version.Version(
+                    str(self._version_number)) >= dir_version and (
+                        edition_type & dir_edition_type
+                ):
+                    path = "%s:\\Balloon\\%s\\%s" % (
+                        virtio_drive, main_dir, arch)
+                    if self._conn.test_path(path):
+                        virtio_dir = main_dir
+                        break
+
+            if not virtio_dir:
+                raise exception.CoriolisException(
+                    "Could not determine virtio driver directory "
+                    "for OS '%s' (edition '%s'). "
+                    "Check virtio-win ISO contents and mount status."
+                    % (self._version_number, self._edition_id)
+                )
+
+            driver_paths = [
+                "%s:\\%s\\%s\\%s" % (
+                    virtio_drive, d, virtio_dir, arch) for d in drivers
+            ]
+
+            sid = self._get_sid()
+            # Fails on Nano Server without explicitly granting permissions
+            file_repo_path = (
+                "%sWindows\\System32\\DriverStore\\FileRepository" %
+                self._os_root_dir
+            )
+            self._grant_permissions(file_repo_path, "*%s" % sid)
+            try:
+                for driver_path in driver_paths:
+                    if self._conn.test_path(driver_path):
+                        self._add_dism_driver(driver_path)
+                    else:
+                        LOG.warn(
+                            "Could not locate driver dir '%s', skipping.",
+                            driver_path
+                        )
+            finally:
+                self._revoke_permissions(file_repo_path, "*%s" % sid)
+        finally:
+            self._dismount_disk_image(virtio_iso_path)
+
     def get_packages(self):
         return [], []
 

coriolis/tests/osmorphing/test_windows.py

@@ -6,6 +6,9 @@ import json
 import logging
 from unittest import mock
 
+import ddt
+from distutils import version
+
 from coriolis import exception
 from coriolis.osmorphing import windows
 from coriolis.tests import test_base
@@ -25,9 +28,12 @@ class CoriolisTestException(Exception):
     pass
 
 
+@ddt.ddt
 class BaseWindowsMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
     """Test suite for the BaseWindowsMorphingTools class."""
 
+    _FAKE_VIRTIO_ISO_URL = "fake-virtio-url"
+
     def setUp(self):
         super(BaseWindowsMorphingToolsTestCase, self).setUp()
         self.detected_os_info = {
@@ -43,11 +49,14 @@ class BaseWindowsMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
         self.conn = mock.MagicMock()
         self.event_manager = mock.MagicMock()
         self.os_root_dir = 'C:\\'
+        self.osmorphing_parameters = {
+            "windows_virtio_iso_url": self._FAKE_VIRTIO_ISO_URL
+        }
         self.morphing_tools = windows.BaseWindowsMorphingTools(
             self.conn, self.os_root_dir,
             mock.sentinel.os_root_dev, mock.sentinel.hypervisor,
             self.event_manager, self.detected_os_info,
-            mock.sentinel.osmorphing_parameters,
+            self.osmorphing_parameters,
             mock.sentinel.operation_timeout)
 
     def test_get_required_detected_os_info_fields(self):
@@ -989,3 +998,207 @@ class BaseWindowsMorphingToolsTestCase(test_base.CoriolisBaseTestCase):
             self.morphing_tools._conn,
             f"C:\\Cloudbase-Init\\LocalScripts/{script_filename}",
             mock_script)
+
+    @ddt.data(
+        {
+            "version_number": version.LooseVersion("10.0.26500"),
+            "edition_id": "ServerDatacenterEval",
+            "exp_virtio_dir": "2k25",
+        },
+        {
+            "version_number": version.LooseVersion("10.0.23000"),
+            "edition_id": "Enterprise",
+            "exp_virtio_dir": "w11",
+        }
+    )
+    @ddt.unpack
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_mount_disk_image")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_get_sid")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_grant_permissions")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_add_dism_driver")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_revoke_permissions")
+    @mock.patch.object(
+        windows.BaseWindowsMorphingTools, "_dismount_disk_image")
+    def test_add_virtio_drivers(
+        self,
+        mock_dismount,
+        mock_revoke_permissions,
+        mock_add_dism_driver,
+        mock_grant_permissions,
+        mock_get_sid,
+        mock_mount_disk_image,
+        version_number=None,
+        edition_id=None,
+        exp_virtio_dir=None,
+    ):
+        null_sid = "S-1-0-0"
+        mock_get_sid.return_value = null_sid
+
+        self.morphing_tools._version_number = version_number
+        self.morphing_tools._edition_id = edition_id
+
+        fake_image_mountpoint = "e"
+        mock_mount_disk_image.return_value = fake_image_mountpoint
+
+        self.morphing_tools._add_virtio_drivers()
+
+        exp_iso_path = "c:\\virtio-win.iso"
+        self.morphing_tools._conn.download_file.assert_called_once_with(
+            self._FAKE_VIRTIO_ISO_URL, exp_iso_path)
+        mock_mount_disk_image.assert_called_once_with(exp_iso_path)
+        mock_dismount.assert_called_once_with(exp_iso_path)
+
+        exp_repo_path = "C:\\Windows\\System32\\DriverStore\\FileRepository"
+        mock_grant_permissions.assert_called_once_with(
+            exp_repo_path, f"*{null_sid}")
+        mock_revoke_permissions.assert_called_once_with(
+            exp_repo_path, f"*{null_sid}")
+
+        drivers = [
+            "Balloon",
+            "NetKVM",
+            "qxl",
+            "qxldod",
+            "pvpanic",
+            "viorng",
+            "vioscsi",
+            "vioserial",
+            "viostor",
+            "viogpudo",
+        ]
+        exp_driver_paths = [
+            f"{fake_image_mountpoint}:\\{driver}\\{exp_virtio_dir}\\amd64"
+            for driver in drivers]
+        mock_add_dism_driver.assert_has_calls(
+            [mock.call(path) for path in exp_driver_paths])
+
+    @ddt.data(
+        {
+            "version_number": version.LooseVersion("10.0.26500"),
+            "edition_id": "ServerDatacenterEval",
+        },
+    )
+    @ddt.unpack
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_mount_disk_image")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_get_sid")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_grant_permissions")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_add_dism_driver")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_revoke_permissions")
+    @mock.patch.object(
+        windows.BaseWindowsMorphingTools, "_dismount_disk_image")
+    def test_add_virtio_drivers_failure(
+        self,
+        mock_dismount,
+        mock_revoke_permissions,
+        mock_add_dism_driver,
+        mock_grant_permissions,
+        mock_get_sid,
+        mock_mount_disk_image,
+        version_number=None,
+        edition_id=None,
+    ):
+        # Ensure proper cleanup in case of driver installation failure.
+        null_sid = "S-1-0-0"
+        mock_get_sid.return_value = null_sid
+
+        self.morphing_tools._version_number = version_number
+        self.morphing_tools._edition_id = edition_id
+
+        fake_image_mountpoint = "e"
+        mock_mount_disk_image.return_value = fake_image_mountpoint
+        mock_add_dism_driver.side_effect = IOError
+
+        self.assertRaises(
+            IOError,
+            self.morphing_tools._add_virtio_drivers)
+
+        exp_iso_path = "c:\\virtio-win.iso"
+        self.morphing_tools._conn.download_file.assert_called_once_with(
+            self._FAKE_VIRTIO_ISO_URL, exp_iso_path)
+        mock_mount_disk_image.assert_called_once_with(exp_iso_path)
+        mock_dismount.assert_called_once_with(exp_iso_path)
+
+        exp_repo_path = "C:\\Windows\\System32\\DriverStore\\FileRepository"
+        mock_grant_permissions.assert_called_once_with(
+            exp_repo_path, f"*{null_sid}")
+        mock_revoke_permissions.assert_called_once_with(
+            exp_repo_path, f"*{null_sid}")
+
+    @ddt.data(
+        {
+            "version_number": version.LooseVersion("10.0.26500"),
+            "edition_id": "ServerDatacenterEval",
+        },
+    )
+    @ddt.unpack
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_mount_disk_image")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_get_sid")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_grant_permissions")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_add_dism_driver")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_revoke_permissions")
+    @mock.patch.object(
+        windows.BaseWindowsMorphingTools, "_dismount_disk_image")
+    def test_add_virtio_drivers_missing_virtio_dir(
+        self,
+        mock_dismount,
+        mock_revoke_permissions,
+        mock_add_dism_driver,
+        mock_grant_permissions,
+        mock_get_sid,
+        mock_mount_disk_image,
+        version_number=None,
+        edition_id=None,
+    ):
+        self.morphing_tools._version_number = version_number
+        self.morphing_tools._edition_id = edition_id
+        self.conn.test_path.return_value = False
+
+        self.assertRaises(
+            exception.CoriolisException,
+            self.morphing_tools._add_virtio_drivers)
+
+        exp_iso_path = "c:\\virtio-win.iso"
+        self.morphing_tools._conn.download_file.assert_called_once_with(
+            self._FAKE_VIRTIO_ISO_URL, exp_iso_path)
+        mock_mount_disk_image.assert_called_once_with(exp_iso_path)
+        mock_dismount.assert_called_once_with(exp_iso_path)
+        mock_add_dism_driver.assert_not_called()
+
+    @ddt.data(
+        {
+            "version_number": version.LooseVersion("3.1"),
+            "edition_id": "Sever",
+        },
+    )
+    @ddt.unpack
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_mount_disk_image")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_get_sid")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_grant_permissions")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_add_dism_driver")
+    @mock.patch.object(windows.BaseWindowsMorphingTools, "_revoke_permissions")
+    @mock.patch.object(
+        windows.BaseWindowsMorphingTools, "_dismount_disk_image")
+    def test_add_virtio_drivers_unsupported_version(
+        self,
+        mock_dismount,
+        mock_revoke_permissions,
+        mock_add_dism_driver,
+        mock_grant_permissions,
+        mock_get_sid,
+        mock_mount_disk_image,
+        version_number=None,
+        edition_id=None,
+    ):
+        self.morphing_tools._version_number = version_number
+        self.morphing_tools._edition_id = edition_id
+
+        self.assertRaises(
+            exception.CoriolisException,
+            self.morphing_tools._add_virtio_drivers)
+
+        exp_iso_path = "c:\\virtio-win.iso"
+        self.morphing_tools._conn.download_file.assert_called_once_with(
+            self._FAKE_VIRTIO_ISO_URL, exp_iso_path)
+        mock_mount_disk_image.assert_called_once_with(exp_iso_path)
+        mock_dismount.assert_called_once_with(exp_iso_path)
+        mock_add_dism_driver.assert_not_called()