config: T4919: Support copying encrypted volumes during install

Re-implements vyos/vyatta-cfg-system#194

src/op_mode/image_installer.py

@@ -65,6 +65,8 @@
 MSG_INPUT_CONSOLE_TYPE: str = 'What console should be used by default? (K: KVM, S: Serial, U: USB-Serial)?'
 MSG_INPUT_COPY_DATA: str = 'Would you like to copy data to the new image?'
 MSG_INPUT_CHOOSE_COPY_DATA: str = 'From which image would you like to save config information?'
+MSG_INPUT_COPY_ENC_DATA: str = 'Would you like to copy the encrypted config to the new image?'
+MSG_INPUT_CHOOSE_COPY_ENC_DATA: str = 'From which image would you like to copy the encrypted config?'
 MSG_WARN_ISO_SIGN_INVALID: str = 'Signature is not valid. Do you want to continue with installation?'
 MSG_WARN_ISO_SIGN_UNAVAL: str = 'Signature is not available. Do you want to continue with installation?'
 MSG_WARN_ROOT_SIZE_TOOBIG: str = 'The size is too big. Try again.'
@@ -212,47 +214,79 @@ def search_previous_installation(disks: list[str]) -> None:
         disks (list[str]): a list of available disks
     """
     mnt_config = '/mnt/config'
+    mnt_encrypted_config = '/mnt/encrypted_config'
     mnt_ssh = '/mnt/ssh'
     mnt_tmp = '/mnt/tmp'
     rmtree(Path(mnt_config), ignore_errors=True)
     rmtree(Path(mnt_ssh), ignore_errors=True)
     Path(mnt_tmp).mkdir(exist_ok=True)
+    Path(mnt_encrypted_config).unlink(missing_ok=True)
 
     print('Searching for data from previous installations')
     image_data = []
+    encrypted_configs = []
     for disk_name in disks:
         for partition in disk.partition_list(disk_name):
             if disk.partition_mount(partition, mnt_tmp):
                 if Path(mnt_tmp + '/boot').exists():
                     for path in Path(mnt_tmp + '/boot').iterdir():
                         if path.joinpath('rw/config/.vyatta_config').exists():
                             image_data.append((path.name, partition))
+                if Path(mnt_tmp + '/luks').exists():
+                    for path in Path(mnt_tmp + '/luks').iterdir():
+                        encrypted_configs.append((path.name, partition))
 
                 disk.partition_umount(partition)
 
-    if len(image_data) == 1:
-        image_name, image_drive = image_data[0]
-        print('Found data from previous installation:')
-        print(f'\t{image_name} on {image_drive}')
-        if not ask_yes_no(MSG_INPUT_COPY_DATA, default=True):
-            return
-
-    elif len(image_data) > 1:
-        print('Found data from previous installations')
-        if not ask_yes_no(MSG_INPUT_COPY_DATA, default=True):
-            return
-
-        image_name, image_drive = select_entry(image_data,
-                                               'Available versions:',
-                                               MSG_INPUT_CHOOSE_COPY_DATA,
-                                               search_format_selection)
+    image_name = None
+    image_drive = None
+    encrypted = False
+
+    if len(image_data) > 0:
+        if len(image_data) == 1:
+            print('Found data from previous installation:')
+            print(f'\t{" on ".join(image_data[0])}')
+            if ask_yes_no(MSG_INPUT_COPY_DATA, default=True):
+                image_name, image_drive = image_data[0]
+
+        elif len(image_data) > 1:
+            print('Found data from previous installations')
+            if ask_yes_no(MSG_INPUT_COPY_DATA, default=True):
+                image_name, image_drive = select_entry(image_data,
+                                                       'Available versions:',
+                                                       MSG_INPUT_CHOOSE_COPY_DATA,
+                                                       search_format_selection)
+    elif len(encrypted_configs) > 0:
+        if len(encrypted_configs) == 1:
+            print('Found encrypted config from previous installation:')
+            print(f'\t{" on ".join(encrypted_configs[0])}')
+            if ask_yes_no(MSG_INPUT_COPY_ENC_DATA, default=True):
+                image_name, image_drive = encrypted_configs[0]
+                encrypted = True
+
+        elif len(encrypted_configs) > 1:
+            print('Found encrypted configs from previous installations')
+            if ask_yes_no(MSG_INPUT_COPY_ENC_DATA, default=True):
+                image_name, image_drive = select_entry(encrypted_configs,
+                                          'Available versions:',
+                                          MSG_INPUT_CHOOSE_COPY_ENC_DATA,
+                                          search_format_selection)
+                encrypted = True
+
     else:
         print('No previous installation found')
         return
 
+    if not image_name:
+        return
+
     disk.partition_mount(image_drive, mnt_tmp)
 
-    copytree(f'{mnt_tmp}/boot/{image_name}/rw/config', mnt_config)
+    if not encrypted:
+        copytree(f'{mnt_tmp}/boot/{image_name}/rw/config', mnt_config)
+    else:
+        copy(f'{mnt_tmp}/luks/{image_name}', mnt_encrypted_config)
+
     Path(mnt_ssh).mkdir()
     host_keys: list[str] = glob(f'{mnt_tmp}/boot/{image_name}/rw/etc/ssh/ssh_host*')
     for host_key in host_keys:
@@ -279,6 +313,12 @@ def copy_previous_installation_data(target_dir: str) -> None:
                  dirs_exist_ok=True)
 
 
+def copy_previous_encrypted_config(target_dir: str, image_name: str) -> None:
+    if Path('/mnt/encrypted_config').exists():
+        Path(target_dir).mkdir(exist_ok=True)
+        copy('/mnt/encrypted_config', Path(target_dir).joinpath(image_name))
+
+
 def ask_single_disk(disks_available: dict[str, int]) -> str:
     """Ask user to select a disk for installation
 
@@ -712,6 +752,9 @@ def install_image() -> None:
         # owner restored on copy of config data by chmod_2775, above
         copy_previous_installation_data(f'{DIR_DST_ROOT}/boot/{image_name}/rw')
 
+        # copy saved encrypted config volume
+        copy_previous_encrypted_config(f'{DIR_DST_ROOT}/luks', image_name)
+
         if is_raid_install(install_target):
             write_dir: str = f'{DIR_DST_ROOT}/boot/{image_name}/rw'
             raid.update_default(write_dir)

src/op_mode/image_manager.py

@@ -95,6 +95,15 @@ def delete_image(image_name: Optional[str] = None,
     except Exception as err:
         exit(f'Unable to remove the image "{image_name}": {err}')
 
+    # remove LUKS volume if it exists
+    luks_path: Path = Path(f'{persistence_storage}/luks/{image_name}')
+    if luks_path.is_file():
+        try:
+            luks_path.unlink()
+            print(f'The encrypted config for "{image_name}" was successfully deleted')
+        except Exception as err:
+            exit(f'Unable to remove the encrypted config for "{image_name}": {err}')
+
 
 @compat.grub_cfg_update
 def set_image(image_name: Optional[str] = None,
@@ -174,6 +183,16 @@ def rename_image(name_old: str, name_new: str) -> None:
     except Exception as err:
         exit(f'Unable to rename image "{name_old}" to "{name_new}": {err}')
 
+    # rename LUKS volume if it exists
+    old_luks_path: Path = Path(f'{persistence_storage}/luks/{name_old}')
+    if old_luks_path.is_file():
+        try:
+            new_luks_path: Path = Path(f'{persistence_storage}/luks/{name_new}')
+            old_luks_path.rename(new_luks_path)
+            print(f'The encrypted config for "{name_old}" was successfully renamed to "{name_new}"')
+        except Exception as err:
+            exit(f'Unable to rename the encrypted config for "{name_old}" to "{name_new}": {err}')
+
 
 def list_images() -> None:
     """Print list of available images for CLI hints"""