Unlock Bitlocker encrypted volumes

coriolis/minion_manager/rpc/server.py

@@ -7,6 +7,7 @@
 
 from oslo_config import cfg
 from oslo_log import log as logging
+from oslo_utils import strutils
 from oslo_utils import timeutils
 from taskflow import deciders as taskflow_deciders
 from taskflow.patterns import graph_flow
@@ -508,7 +509,9 @@ def _check_pool_minion_count(
                         pool_id, instances, action['id']))
         LOG.debug(
             "Successfully validated minion pool selections for action '%s' "
-            "with properties: %s", action['id'], action)
+            "with properties: %s",
+            action['id'],
+            strutils.mask_dict_password(action))
 
     def allocate_minion_machines_for_transfer(
             self, ctxt, transfer):

coriolis/osmorphing/manager.py

@@ -5,6 +5,7 @@
 
 from oslo_config import cfg
 from oslo_log import log as logging
+from oslo_utils import strutils
 
 from coriolis import constants
 from coriolis import events
@@ -84,7 +85,8 @@ def get_osmorphing_tools_class_for_provider(
     LOG.debug(
         "OSMorphing tools classes returned by provider '%s' for os_type '%s' "
         "and 'osmorphing_info' %s: %s",
-        type(provider), os_type, osmorphing_info, available_tools_cls)
+        type(provider), os_type,
+        strutils.mask_dict_password(osmorphing_info), available_tools_cls)
 
     osmorphing_base_class = base_osmorphing.BaseOSMorphingTools
     for toolscls in available_tools_cls:

coriolis/osmorphing/osmount/windows.py

@@ -6,6 +6,7 @@
 
 from oslo_log import log as logging
 
+from coriolis import constants
 from coriolis import exception
 from coriolis.osmorphing.osmount import base
 from coriolis import utils
@@ -15,6 +16,13 @@
 
 
 class WindowsMountTools(base.BaseOSMountTools):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        # A list of BitLocker encrypted volumes that were unlocked
+        # by us. We'll use a first-boot script to resume BitLocker.
+        self._unlocked_volumes: list[str] = []
+
     def _connect(self):
         connection_info = self._connection_info
 
@@ -223,9 +231,120 @@ def _set_volumes_drive_letter(self):
                         f"Error was: {utils.get_exception_details()}")
             self._rebring_disks_online(disk_nums=disk_nums)
 
+    def _get_encrypted_volume_ids(self):
+        out = self._conn.exec_ps_command(
+            'gwmi -ns "Root\\CIMV2\\Security\\MicrosoftVolumeEncryption" '
+            '-class Win32_EncryptableVolume | % {$_.DeviceID}')
+        return [x for x in out.replace("\r\n", "\n").split("\n") if x]
+
+    def _unlock_encrypted_volume(self, volume_id: str, recovery_password: str):
+        self._conn.exec_ps_command(
+            f'manage-bde -unlock "{volume_id}" '
+            f'-RecoveryPassword "{recovery_password}"')
+
+    def _unlock_encrypted_volumes(self):
+        recovery_password = self._osmorphing_info.get(
+            constants.ENCRYPTED_DISKS_PASS)
+        if not recovery_password:
+            LOG.info("No encrypted disk password specified, "
+                     "skipping BitLocker unlock.")
+            return
+
+        encrypted_volume_ids = self._get_encrypted_volume_ids()
+        if not encrypted_volume_ids:
+            LOG.warning("Received encrypted disk password but no "
+                        "BitLocker encrypted volumes found.")
+            return
+
+        unlocked = False
+        for encrypted_volume_id in encrypted_volume_ids:
+            try:
+                self._unlock_encrypted_volume(
+                    encrypted_volume_id, recovery_password)
+                LOG.info(
+                    "Successfully unlocked BitLocker encrypted volume: %s",
+                    encrypted_volume_id)
+                unlocked = True
+            except Exception:
+                LOG.info(
+                    "Could not unlock volume %s using the specified "
+                    "recovery password.",
+                    encrypted_volume_id)
+                continue
+            self._unlocked_volumes.append(encrypted_volume_id)
+
+        if not unlocked:
+            raise exception.CoriolisException(
+                "Could not unlock any volume using the specified "
+                "BitLocker recovery password.")
+
+    def _suspend_bitlocker(self, volume_id: str):
+        """Suspend BitLocker until the next reboot for a given volume.
+
+        It doesn't decrypt the device, it just adds a publicly accessible
+        BitLocker protector that automatically unlocks the volume.
+
+        When the replica instance boots, the TPM protector will be reconfigured
+        automatically. Unfortunately the '-RebootCount' parameter isn't
+        honored, perhaps due to the fact that the disks are attached to a
+        separate VM. For this reason, we'll use a first-boot script to resume
+        BitLocker explicitly.
+        """
+        self._conn.exec_ps_command(f'Suspend-BitLocker "{volume_id}"')
+
+    def _resume_bitlocker(self, volume_id: str):
+        self._conn.exec_ps_command(f'Resume-BitLocker "{volume_id}"')
+
+    def install_encryption_firstboot_setup(
+        self,
+        os_root_dir,
+        os_morphing_tools,
+    ):
+        if not self._unlocked_volumes:
+            LOG.info(
+                "No unlocked BitLocker volumes, skipping first-boot setup.")
+            return
+
+        # We'll inject a first-boot script to resume BitLocker explicitly.
+        # Unfortunately the "-RebootCount" parameter of "Suspend-BitLocker"
+        # isn't honored, perhaps due to the fact that the disks are attached
+        # to a different VM.
+        script_content = ""
+
+        try:
+            for encrypted_volume_id in self._unlocked_volumes:
+                LOG.info(
+                    "Suspending BitLocker for volume %s, scheduling it to be "
+                    "resumed after first-boot.",
+                    encrypted_volume_id)
+                # Suspend BitLocker until the replica boots.
+                self._suspend_bitlocker(encrypted_volume_id)
+                # Add a resume command to the first-boot script.
+                script_content += (
+                    f'Resume-BitLocker "{encrypted_volume_id}"\r\n')
+
+            # Resume BitLocker after bringing the disks online,
+            # which has a script priority of 10.
+            os_morphing_tools.register_firstboot_script(
+                script_content,
+                user_provided=False,
+                script_filename="11-bitlocker-firstboot.ps1")
+        except Exception:
+            LOG.exception("First-boot preparation failed, attempting to "
+                          "resume BitLocker.")
+            for encrypted_volume_id in self._unlocked_volumes:
+                try:
+                    self._resume_bitlocker(encrypted_volume_id)
+                except Exception:
+                    LOG.exception(
+                        "Unable to resume BitLocker for volume: %s" %
+                        encrypted_volume_id)
+            raise
+
     def mount_os(self):
         self._set_basic_disks_rw_mode()
         self._bring_disks_online()
+        self._unlock_encrypted_volumes()
         self._set_volumes_drive_letter()
         fs_roots = utils.retry_on_error(sleep_seconds=5)(self._get_fs_roots)(
             fail_if_empty=True)

coriolis/tests/osmorphing/osmount/test_windows.py

@@ -4,6 +4,9 @@
 import logging
 from unittest import mock
 
+from oslo_utils import strutils
+
+from coriolis import constants
 from coriolis import exception
 from coriolis.osmorphing.osmount import windows
 from coriolis.tests import test_base
@@ -336,3 +339,203 @@ def test_dismount_os(self):
 
         self.tools._conn.exec_ps_command.assert_called_once_with(
             '(Get-Disk | Where-Object { $_.IsBoot -eq $False }).Number')
+
+    def test_get_encrypted_volume_ids(self):
+        # Powershell wouldn't mix line endings, we're just ensuring that
+        # we can properly handle both line ending types.
+        self.tools._conn.exec_ps_command.return_value = (
+            "\\\\?\\Volume{2750d574-b333-4e7b-a0a2-d739279d39e9}\\\r\n"
+            "\\\\?\\Volume{7723f315-c13c-450c-8be6-f58e06f4ad45}\\\r\n"
+            "\\\\?\\Volume{cb7399af-8f6a-4a7b-a55c-e885ec3ff5fd}\\\n"
+        )
+
+        exp_ret = [
+            "\\\\?\\Volume{2750d574-b333-4e7b-a0a2-d739279d39e9}\\",
+            "\\\\?\\Volume{7723f315-c13c-450c-8be6-f58e06f4ad45}\\",
+            "\\\\?\\Volume{cb7399af-8f6a-4a7b-a55c-e885ec3ff5fd}\\",
+        ]
+        ret = self.tools._get_encrypted_volume_ids()
+
+        self.assertEqual(exp_ret, ret)
+        self.tools._conn.exec_ps_command.assert_called_once_with(
+            'gwmi -ns "Root\\CIMV2\\Security\\MicrosoftVolumeEncryption" '
+            '-class Win32_EncryptableVolume | % {$_.DeviceID}')
+
+    def test_unlock_encrypted_volume(self):
+        vol = "\\\\?\\Volume{2750d574-b333-4e7b-a0a2-d739279d39e9}\\"
+        password = "6010ba47-28e4-4105-8b0a-69eed0a54283"
+
+        self.tools._unlock_encrypted_volume(vol, password)
+
+        exp_cmd = 'manage-bde -unlock "%s" -RecoveryPassword "%s"' % (
+            vol, password)
+        self.tools._conn.exec_ps_command.assert_called_once_with(
+            exp_cmd)
+
+    def test_sanitize_recovery_password(self):
+        vol = "\\\\?\\Volume{2750d574-b333-4e7b-a0a2-d739279d39e9}\\"
+        password = "6010ba47-28e4-4105-8b0a-69eed0a54283"
+
+        cmd = 'manage-bde -unlock "%s" -RecoveryPassword "%s"' % (
+            vol, password)
+        exp_cmd = 'manage-bde -unlock "%s" -RecoveryPassword "%s"' % (
+            vol, '***')
+
+        self.assertEqual(exp_cmd, strutils.mask_password(cmd))
+
+    def test_suspend_bitlocker(self):
+        vol = "\\\\?\\Volume{2750d574-b333-4e7b-a0a2-d739279d39e9}\\"
+
+        self.tools._suspend_bitlocker(vol)
+
+        exp_cmd = 'Suspend-BitLocker "%s"' % (vol)
+        self.tools._conn.exec_ps_command.assert_called_once_with(
+            exp_cmd)
+
+    def test_resume_bitlocker(self):
+        vol = "\\\\?\\Volume{2750d574-b333-4e7b-a0a2-d739279d39e9}\\"
+
+        self.tools._resume_bitlocker(vol)
+
+        exp_cmd = 'Resume-BitLocker "%s"' % (vol)
+        self.tools._conn.exec_ps_command.assert_called_once_with(
+            exp_cmd)
+
+    @mock.patch.object(windows.WindowsMountTools, "_get_encrypted_volume_ids")
+    def test_unlock_encrypted_volumes_no_password(
+        self,
+        mock_get_encrypted_volume_ids,
+    ):
+        self.tools._unlock_encrypted_volumes()
+        mock_get_encrypted_volume_ids.assert_not_called()
+
+    @mock.patch.object(windows.WindowsMountTools, "_get_encrypted_volume_ids")
+    @mock.patch.object(windows.WindowsMountTools, "_unlock_encrypted_volume")
+    def test_unlock_encrypted_volumes_not_encrypted(
+        self,
+        mock_unlock_encrypted_volume,
+        mock_get_encrypted_volume_ids,
+    ):
+        fake_pass = "fake-recovery-password"
+        self.tools._osmorphing_info[constants.ENCRYPTED_DISKS_PASS] = fake_pass
+
+        mock_get_encrypted_volume_ids.return_value = []
+
+        self.tools._unlock_encrypted_volumes()
+        mock_unlock_encrypted_volume.assert_not_called()
+
+    @mock.patch.object(windows.WindowsMountTools, "_get_encrypted_volume_ids")
+    @mock.patch.object(windows.WindowsMountTools, "_unlock_encrypted_volume")
+    def test_unlock_encrypted_volumes_all_failed(
+        self,
+        mock_unlock_encrypted_volume,
+        mock_get_encrypted_volume_ids,
+    ):
+        fake_pass = "fake-recovery-password"
+        self.tools._osmorphing_info[constants.ENCRYPTED_DISKS_PASS] = fake_pass
+
+        mock_get_encrypted_volume_ids.return_value = [
+            mock.sentinel.volume0,
+            mock.sentinel.volume1,
+        ]
+        mock_unlock_encrypted_volume.side_effect = ValueError
+
+        self.assertRaises(
+            exception.CoriolisException,
+            self.tools._unlock_encrypted_volumes,
+        )
+
+    @mock.patch.object(windows.WindowsMountTools, "_get_encrypted_volume_ids")
+    @mock.patch.object(windows.WindowsMountTools, "_unlock_encrypted_volume")
+    def test_unlock_encrypted_volumes_one_failed(
+        self,
+        mock_unlock_encrypted_volume,
+        mock_get_encrypted_volume_ids,
+    ):
+        fake_pass = "fake-recovery-password"
+        self.tools._osmorphing_info[constants.ENCRYPTED_DISKS_PASS] = fake_pass
+
+        encrypted_volume_ids = [
+            mock.sentinel.volume0,
+            mock.sentinel.volume1,
+        ]
+        mock_get_encrypted_volume_ids.return_value = encrypted_volume_ids
+        mock_unlock_encrypted_volume.side_effect = [ValueError, None]
+
+        self.tools._unlock_encrypted_volumes()
+
+        mock_unlock_encrypted_volume.assert_has_calls(
+            [mock.call(vol_id, fake_pass) for vol_id in encrypted_volume_ids])
+
+        self.assertEqual(
+            self.tools._unlocked_volumes, [mock.sentinel.volume1])
+
+    def test_install_encryption_firstboot_setup_noop(self):
+        # No unlocked volumes, nothing to do.
+        mock_morphing_tools = mock.Mock()
+        self.tools.install_encryption_firstboot_setup(
+            mock.sentinel.os_root_dir,
+            mock_morphing_tools)
+        mock_morphing_tools.register_firstboot_script.assert_not_called()
+
+    @mock.patch.object(windows.WindowsMountTools, "_suspend_bitlocker")
+    def test_install_encryption_firstboot_setup(self, mock_suspend_bitlocker):
+        self.tools._unlocked_volumes = ["vol1", "vol2"]
+        mock_morphing_tools = mock.Mock()
+        self.tools.install_encryption_firstboot_setup(
+            mock.sentinel.os_root_dir,
+            mock_morphing_tools)
+
+        expected_script = (
+            'Resume-BitLocker "vol1"\r\nResume-BitLocker "vol2"\r\n')
+        mock_morphing_tools.register_firstboot_script.assert_called_once_with(
+            expected_script,
+            user_provided=False,
+            script_filename="11-bitlocker-firstboot.ps1")
+        mock_suspend_bitlocker.assert_has_calls(
+            [mock.call(volume) for volume in self.tools._unlocked_volumes])
+
+    @mock.patch.object(windows.WindowsMountTools, "_suspend_bitlocker")
+    @mock.patch.object(windows.WindowsMountTools, "_resume_bitlocker")
+    def test_install_encryption_firstboot_setup_register_failure(
+        self,
+        mock_resume_bitlocker,
+        mock_suspend_bitlocker
+    ):
+        self.tools._unlocked_volumes = ["vol1", "vol2"]
+        mock_morphing_tools = mock.Mock()
+        mock_morphing_tools.register_firstboot_script.side_effect = IOError
+        self.assertRaises(
+            IOError,
+            self.tools.install_encryption_firstboot_setup,
+            mock.sentinel.os_root_dir,
+            mock_morphing_tools)
+
+        mock_suspend_bitlocker.assert_has_calls(
+            [mock.call(volume) for volume in self.tools._unlocked_volumes])
+        mock_resume_bitlocker.assert_has_calls(
+            [mock.call(volume) for volume in self.tools._unlocked_volumes])
+
+    @mock.patch.object(windows.WindowsMountTools, "_suspend_bitlocker")
+    @mock.patch.object(windows.WindowsMountTools, "_resume_bitlocker")
+    def test_install_encryption_firstboot_setup_suspend_failure(
+        self,
+        mock_resume_bitlocker,
+        mock_suspend_bitlocker
+    ):
+        self.tools._unlocked_volumes = ["vol1", "vol2"]
+        mock_morphing_tools = mock.Mock()
+        mock_suspend_bitlocker.side_effect = [None, IOError]
+        # Let resume-bitlocker fail to ensure that we're still trying to
+        # cover all volumes.
+        mock_resume_bitlocker.side_effect = ValueError
+        self.assertRaises(
+            IOError,
+            self.tools.install_encryption_firstboot_setup,
+            mock.sentinel.os_root_dir,
+            mock_morphing_tools)
+
+        mock_suspend_bitlocker.assert_has_calls(
+            [mock.call(volume) for volume in self.tools._unlocked_volumes])
+        mock_resume_bitlocker.assert_has_calls(
+            [mock.call(volume) for volume in self.tools._unlocked_volumes])