From 187a30c0504c8853ff01b4143cf001736a67af4a Mon Sep 17 00:00:00 2001 From: Mihaela Balutoiu Date: Thu, 18 Jun 2026 10:24:20 +0300 Subject: [PATCH 1/3] Validate windows_virtio_iso_url OSMorphing parameter Signed-off-by: Mihaela Balutoiu --- coriolis/osmorphing/windows.py | 12 ++++++++++-- coriolis/tests/osmorphing/test_windows.py | 15 +++++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/coriolis/osmorphing/windows.py b/coriolis/osmorphing/windows.py index fcfe7a69..e5e768ed 100644 --- a/coriolis/osmorphing/windows.py +++ b/coriolis/osmorphing/windows.py @@ -71,6 +71,8 @@ REQUIRED_DETECTED_WINDOWS_OS_FIELDS = [ "version_number", "edition_id", "installation_type", "product_name"] +VIRTIO_WIN_ISO_PATH = "c:\\virtio-win.iso" + INTERFACES_PATH_FORMAT = ( "HKLM:\\%s\\ControlSet001\\Services\\Tcpip\\Parameters\\Interfaces") @@ -754,10 +756,16 @@ def _add_virtio_drivers(self): "viogpudo", ] + virtio_iso_url = self._osmorphing_parameters.get( + "windows_virtio_iso_url") + if not virtio_iso_url: + raise exception.CoriolisException( + "VirtIO drivers ISO URL ('windows_virtio_iso_url') not set " + "in the OSMorphing parameters") + 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"] + virtio_iso_path = VIRTIO_WIN_ISO_PATH utils.retry_on_error(sleep_seconds=5)(self._conn.download_file)( virtio_iso_url, virtio_iso_path ) diff --git a/coriolis/tests/osmorphing/test_windows.py b/coriolis/tests/osmorphing/test_windows.py index f20e5b62..8a27323d 100644 --- a/coriolis/tests/osmorphing/test_windows.py +++ b/coriolis/tests/osmorphing/test_windows.py @@ -1202,3 +1202,18 @@ def test_add_virtio_drivers_unsupported_version( 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() + + @mock.patch.object(windows.BaseWindowsMorphingTools, "_mount_disk_image") + def test_add_virtio_drivers_missing_iso_url( + self, mock_mount_disk_image): + self.morphing_tools._version_number = version.LooseVersion( + "10.0.26500") + self.morphing_tools._edition_id = "ServerDatacenterEval" + self.morphing_tools._osmorphing_parameters = {} + + self.assertRaises( + exception.CoriolisException, + self.morphing_tools._add_virtio_drivers) + + self.morphing_tools._conn.download_file.assert_not_called() + mock_mount_disk_image.assert_not_called() From f243272caf03cc81d19e49a743c25b78a4e8ffcd Mon Sep 17 00:00:00 2001 From: Mihaela Balutoiu Date: Thu, 18 Jun 2026 10:27:16 +0300 Subject: [PATCH 2/3] Extract _install_dism_drivers helper Signed-off-by: Mihaela Balutoiu --- coriolis/osmorphing/windows.py | 46 +++++++++++++--------- coriolis/tests/osmorphing/test_windows.py | 48 +++++++++++++++++++++++ 2 files changed, 76 insertions(+), 18 deletions(-) diff --git a/coriolis/osmorphing/windows.py b/coriolis/osmorphing/windows.py index e5e768ed..68e8772b 100644 --- a/coriolis/osmorphing/windows.py +++ b/coriolis/osmorphing/windows.py @@ -801,27 +801,37 @@ def _add_virtio_drivers(self): 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) + self._install_dism_drivers(driver_paths) finally: self._dismount_disk_image(virtio_iso_path) + def _install_dism_drivers(self, driver_paths): + """Installs the drivers located at the given paths into the offline + Windows image via DISM. + + Temporarily grants permissions on the DriverStore FileRepository + (required on Nano Server) for the duration of the installation. + Paths which do not exist are skipped with a warning. + """ + 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.warning( + "Could not locate driver dir '%s', skipping.", + driver_path + ) + finally: + self._revoke_permissions(file_repo_path, "*%s" % sid) + def get_packages(self): return [], [] diff --git a/coriolis/tests/osmorphing/test_windows.py b/coriolis/tests/osmorphing/test_windows.py index 8a27323d..11d4c202 100644 --- a/coriolis/tests/osmorphing/test_windows.py +++ b/coriolis/tests/osmorphing/test_windows.py @@ -1217,3 +1217,51 @@ def test_add_virtio_drivers_missing_iso_url( self.morphing_tools._conn.download_file.assert_not_called() mock_mount_disk_image.assert_not_called() + + @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") + def test_install_dism_drivers( + self, mock_revoke_permissions, mock_add_dism_driver, + mock_grant_permissions, mock_get_sid): + null_sid = "S-1-0-0" + mock_get_sid.return_value = null_sid + + existing_driver_path = "e:\\NetKVM\\w11\\amd64" + missing_driver_path = "e:\\Missing\\w11\\amd64" + driver_paths = [existing_driver_path, missing_driver_path] + self.conn.test_path.side_effect = lambda path: ( + path == existing_driver_path) + + self.morphing_tools._install_dism_drivers(driver_paths) + + 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}") + # Only the existing driver path should be installed: + mock_add_dism_driver.assert_called_once_with(existing_driver_path) + + @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") + def test_install_dism_drivers_revokes_on_failure( + self, mock_revoke_permissions, mock_add_dism_driver, + mock_grant_permissions, mock_get_sid): + null_sid = "S-1-0-0" + mock_get_sid.return_value = null_sid + mock_add_dism_driver.side_effect = IOError + + self.assertRaises( + IOError, + self.morphing_tools._install_dism_drivers, ["e:\\NetKVM"]) + + exp_repo_path = "C:\\Windows\\System32\\DriverStore\\FileRepository" + mock_grant_permissions.assert_called_once_with( + exp_repo_path, f"*{null_sid}") + # Permissions must be revoked even if driver installation fails: + mock_revoke_permissions.assert_called_once_with( + exp_repo_path, f"*{null_sid}") From 8ef01ae5c0ce148ca3ffddb524cf71d95f258f5a Mon Sep 17 00:00:00 2001 From: Mihaela Balutoiu Date: Thu, 18 Jun 2026 10:31:24 +0300 Subject: [PATCH 3/3] Add QEMU guest agent installer support Signed-off-by: Mihaela Balutoiu --- coriolis/osmorphing/windows.py | 72 ++++++++++++++++++ coriolis/tests/osmorphing/test_windows.py | 93 +++++++++++++++++++++++ 2 files changed, 165 insertions(+) diff --git a/coriolis/osmorphing/windows.py b/coriolis/osmorphing/windows.py index 68e8772b..ee02ac8b 100644 --- a/coriolis/osmorphing/windows.py +++ b/coriolis/osmorphing/windows.py @@ -71,6 +71,18 @@ REQUIRED_DETECTED_WINDOWS_OS_FIELDS = [ "version_number", "edition_id", "installation_type", "product_name"] +QEMU_GUEST_AGENT_INSTALL_SCRIPT_FORMAT = ( + "msiexec.exe /i '%(agent_msi_path)s' /qn /passive " + "/L*V '%(agent_msi_log_dir)s/qemu-guest-agent-msiexec.log'") + +QEMU_GUEST_AGENT_INSTALL_FROM_DIR_SCRIPT_FORMAT = ( + "$dst = 'C:\\Program Files\\Qemu-ga'\n" + "New-Item -ItemType Directory -Force -Path $dst | Out-Null\n" + "Copy-Item '%(qemu_ga_dir)s\\x64\\*' -Destination $dst -Force\n" + "Copy-Item '%(qemu_ga_dir)s\\mingw64\\*' -Destination $dst -Force\n" + '& "$dst\\qemu-ga.exe" --service install\n' + "Start-Service QEMU-GA") + VIRTIO_WIN_ISO_PATH = "c:\\virtio-win.iso" INTERFACES_PATH_FORMAT = ( @@ -832,6 +844,66 @@ def _install_dism_drivers(self, driver_paths): finally: self._revoke_permissions(file_repo_path, "*%s" % sid) + def _setup_qemu_agent_installation_local_script( + self, msi_source_path=None, msi_url=None, + msi_name="qemu-ga-x86_64.msi", qemu_ga_source_dir=None): + """Stages the QEMU guest agent in the Cloudbase-Init directory and + registers a first-boot script which installs it. + + Two source modes are supported: + * MSI: pass 'msi_source_path' or 'msi_url'. The MSI is staged + and installed via msiexec on first boot. + * Directory: pass 'qemu_ga_source_dir' pointing at a 'qemu-ga' + directory (e.g. on the SUSE VMDP ISO) containing 'x64' and + 'mingw64' subdirs. + """ + + if not any([msi_source_path, msi_url, qemu_ga_source_dir]): + raise exception.CoriolisException( + "No QEMU guest agent source provided: one of " + "'msi_source_path', 'msi_url' or 'qemu_ga_source_dir' " + "must be set.") + + cloudbaseinit_base_dir = self._get_cbslinit_base_dir() + guest_cloudbase_init_base_dir = "C%s" % cloudbaseinit_base_dir[1:] + + self._event_manager.progress_update( + "Setting up guest local script for installing the " + "QEMU guest agent on first boot") + + if qemu_ga_source_dir: + qemu_ga_exe_path = "%s\\x64\\qemu-ga.exe" % qemu_ga_source_dir + if not self._conn.test_path(qemu_ga_exe_path): + LOG.warning( + "qemu-ga is not packaged in the configured source. " + "Skipping qemu-ga installation.") + return + # Copy qemu-ga/ to the Cloudbase-Init directory on the migrated + # disk: + self._conn.exec_ps_command( + "Copy-Item -Recurse -Force '%s' '%s'" % ( + qemu_ga_source_dir, cloudbaseinit_base_dir), + ignore_stdout=True) + local_script = QEMU_GUEST_AGENT_INSTALL_FROM_DIR_SCRIPT_FORMAT % { + "qemu_ga_dir": "%s\\qemu-ga" % guest_cloudbase_init_base_dir} + else: + msi_dest_path = "%s\\%s" % (cloudbaseinit_base_dir, msi_name) + if msi_url: + utils.retry_on_error(sleep_seconds=5)( + self._conn.download_file)(msi_url, msi_dest_path) + else: + self._conn.exec_ps_command( + "Copy-Item '%s' -Destination '%s'" % ( + msi_source_path, msi_dest_path)) + local_script = QEMU_GUEST_AGENT_INSTALL_SCRIPT_FORMAT % { + "agent_msi_path": "%s\\%s" % ( + guest_cloudbase_init_base_dir, msi_name), + "agent_msi_log_dir": guest_cloudbase_init_base_dir} + + self.register_firstboot_script( + local_script, user_provided=False, + script_filename="coriolis_qemu_agent_install.ps1") + def get_packages(self): return [], [] diff --git a/coriolis/tests/osmorphing/test_windows.py b/coriolis/tests/osmorphing/test_windows.py index 11d4c202..820a9695 100644 --- a/coriolis/tests/osmorphing/test_windows.py +++ b/coriolis/tests/osmorphing/test_windows.py @@ -1265,3 +1265,96 @@ def test_install_dism_drivers_revokes_on_failure( # Permissions must be revoked even if driver installation fails: mock_revoke_permissions.assert_called_once_with( exp_repo_path, f"*{null_sid}") + + @mock.patch.object( + windows.BaseWindowsMorphingTools, "register_firstboot_script") + def test_setup_qemu_agent_installation_local_script_from_url( + self, mock_register_firstboot_script): + fake_msi_url = "fake-qemu-ga-msi-url" + + self.morphing_tools._setup_qemu_agent_installation_local_script( + msi_url=fake_msi_url) + + exp_msi_dest_path = "C:\\Cloudbase-Init\\qemu-ga-x86_64.msi" + self.morphing_tools._conn.download_file.assert_called_once_with( + fake_msi_url, exp_msi_dest_path) + self.morphing_tools._conn.exec_ps_command.assert_not_called() + + exp_script = windows.QEMU_GUEST_AGENT_INSTALL_SCRIPT_FORMAT % { + "agent_msi_path": exp_msi_dest_path, + "agent_msi_log_dir": "C:\\Cloudbase-Init"} + mock_register_firstboot_script.assert_called_once_with( + exp_script, user_provided=False, + script_filename="coriolis_qemu_agent_install.ps1") + + @mock.patch.object( + windows.BaseWindowsMorphingTools, "register_firstboot_script") + def test_setup_qemu_agent_installation_local_script_from_path( + self, mock_register_firstboot_script): + fake_msi_source_path = "e:\\guest-agent\\qemu-ga-x86_64.msi" + + self.morphing_tools._setup_qemu_agent_installation_local_script( + msi_source_path=fake_msi_source_path) + + exp_msi_dest_path = "C:\\Cloudbase-Init\\qemu-ga-x86_64.msi" + self.morphing_tools._conn.download_file.assert_not_called() + self.morphing_tools._conn.exec_ps_command.assert_called_once_with( + "Copy-Item '%s' -Destination '%s'" % ( + fake_msi_source_path, exp_msi_dest_path)) + + exp_script = windows.QEMU_GUEST_AGENT_INSTALL_SCRIPT_FORMAT % { + "agent_msi_path": exp_msi_dest_path, + "agent_msi_log_dir": "C:\\Cloudbase-Init"} + mock_register_firstboot_script.assert_called_once_with( + exp_script, user_provided=False, + script_filename="coriolis_qemu_agent_install.ps1") + + @mock.patch.object( + windows.BaseWindowsMorphingTools, "register_firstboot_script") + def test_setup_qemu_agent_installation_local_script_no_source( + self, mock_register_firstboot_script): + self.assertRaises( + exception.CoriolisException, + self.morphing_tools._setup_qemu_agent_installation_local_script) + + self.morphing_tools._conn.download_file.assert_not_called() + self.morphing_tools._conn.exec_ps_command.assert_not_called() + mock_register_firstboot_script.assert_not_called() + + @mock.patch.object( + windows.BaseWindowsMorphingTools, "register_firstboot_script") + def test_setup_qemu_agent_installation_local_script_from_qemu_ga_dir( + self, mock_register_firstboot_script): + fake_qemu_ga_source_dir = "f:\\qemu-ga" + self.morphing_tools._conn.test_path.return_value = True + + self.morphing_tools._setup_qemu_agent_installation_local_script( + qemu_ga_source_dir=fake_qemu_ga_source_dir) + + self.morphing_tools._conn.test_path.assert_called_once_with( + "%s\\x64\\qemu-ga.exe" % fake_qemu_ga_source_dir) + self.morphing_tools._conn.download_file.assert_not_called() + self.morphing_tools._conn.exec_ps_command.assert_called_once_with( + "Copy-Item -Recurse -Force '%s' '%s'" % ( + fake_qemu_ga_source_dir, "C:\\Cloudbase-Init"), + ignore_stdout=True) + + exp_script = ( + windows.QEMU_GUEST_AGENT_INSTALL_FROM_DIR_SCRIPT_FORMAT % { + "qemu_ga_dir": "C:\\Cloudbase-Init\\qemu-ga"}) + mock_register_firstboot_script.assert_called_once_with( + exp_script, user_provided=False, + script_filename="coriolis_qemu_agent_install.ps1") + + @mock.patch.object( + windows.BaseWindowsMorphingTools, "register_firstboot_script") + def test_setup_qemu_agent_installation_local_script_qemu_ga_dir_missing( + self, mock_register_firstboot_script): + self.morphing_tools._conn.test_path.return_value = False + + self.morphing_tools._setup_qemu_agent_installation_local_script( + qemu_ga_source_dir="f:\\qemu-ga") + + self.morphing_tools._conn.download_file.assert_not_called() + self.morphing_tools._conn.exec_ps_command.assert_not_called() + mock_register_firstboot_script.assert_not_called()