Merge pull request #1740 from sarthurdev/tpm_luks

config: T4919: Add support for encrypted config with TPM

4 files changed, 424 insertions, 19 deletions

diff --git a/src/helpers/vyos-config-encrypt.py b/src/helpers/vyos-config-encrypt.py
new file mode 100755
index 000000000..8f7359767
--- /dev/null
+++ b/src/helpers/vyos-config-encrypt.py
@@ -0,0 +1,276 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2024 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import re
+import shutil
+import sys
+
+from argparse import ArgumentParser
+from cryptography.fernet import Fernet
+from tempfile import NamedTemporaryFile
+from tempfile import TemporaryDirectory
+
+from vyos.tpm import clear_tpm_key
+from vyos.tpm import init_tpm
+from vyos.tpm import read_tpm_key
+from vyos.tpm import write_tpm_key
+from vyos.util import ask_input
+from vyos.util import ask_yes_no
+from vyos.util import cmd
+
+persistpath_cmd = '/opt/vyatta/sbin/vyos-persistpath'
+mount_paths = ['/config', '/opt/vyatta/etc/config']
+dm_device = '/dev/mapper/vyos_config'
+
+def is_opened():
+    return os.path.exists(dm_device)
+
+def get_current_image():
+    with open('/proc/cmdline', 'r') as f:
+        args = f.read().split(" ")
+        for arg in args:
+            if 'vyos-union' in arg:
+                k, v = arg.split("=")
+                path_split = v.split("/")
+                return path_split[-1]
+    return None
+
+def load_config(key):
+    if not key:
+        return
+
+    persist_path = cmd(persistpath_cmd).strip()
+    image_name = get_current_image()
+    image_path = os.path.join(persist_path, 'luks', image_name)
+
+    if not os.path.exists(image_path):
+        raise Exception("Encrypted config volume doesn't exist")
+
+    if is_opened():
+        print('Encrypted config volume is already mounted')
+        return
+
+    with NamedTemporaryFile(dir='/dev/shm', delete=False) as f:
+        f.write(key)
+        key_file = f.name
+
+    cmd(f'cryptsetup -q open {image_path} vyos_config --key-file={key_file}')
+
+    for path in mount_paths:
+        cmd(f'mount /dev/mapper/vyos_config {path}')
+        cmd(f'chgrp -R vyattacfg {path}')
+
+    os.unlink(key_file)
+
+    return True
+
+def encrypt_config(key, recovery_key):
+    if is_opened():
+        raise Exception('An encrypted config volume is already mapped')
+
+    # Clear and write key to TPM
+    try:
+        clear_tpm_key()
+    except:
+        pass
+    write_tpm_key(key)
+
+    persist_path = cmd(persistpath_cmd).strip()
+    size = ask_input('Enter size of encrypted config partition (MB): ', numeric_only=True, default=512)
+
+    luks_folder = os.path.join(persist_path, 'luks')
+
+    if not os.path.isdir(luks_folder):
+        os.mkdir(luks_folder)
+
+    image_name = get_current_image()
+    image_path = os.path.join(luks_folder, image_name)
+
+    # Create file for encrypted config
+    cmd(f'fallocate -l {size}M {image_path}')
+
+    # Write TPM key for slot #1
+    with NamedTemporaryFile(dir='/dev/shm', delete=False) as f:
+        f.write(key)
+        key_file = f.name
+
+    # Format and add main key to volume
+    cmd(f'cryptsetup -q luksFormat {image_path} {key_file}')
+
+    if recovery_key:
+        # Write recovery key for slot 2
+        with NamedTemporaryFile(dir='/dev/shm', delete=False) as f:
+            f.write(recovery_key)
+            recovery_key_file = f.name
+
+        cmd(f'cryptsetup -q luksAddKey {image_path} {recovery_key_file} --key-file={key_file}')
+
+    # Open encrypted volume and format with ext4
+    cmd(f'cryptsetup -q open {image_path} vyos_config --key-file={key_file}')
+    cmd('mkfs.ext4 /dev/mapper/vyos_config')
+
+    with TemporaryDirectory() as d:
+        cmd(f'mount /dev/mapper/vyos_config {d}')
+
+        # Move /config to encrypted volume
+        shutil.copytree('/config', d, copy_function=shutil.move, dirs_exist_ok=True)
+
+        cmd(f'umount {d}')
+
+    os.unlink(key_file)
+
+    if recovery_key:
+        os.unlink(recovery_key_file)
+
+    for path in mount_paths:
+        cmd(f'mount /dev/mapper/vyos_config {path}')
+        cmd(f'chgrp vyattacfg {path}')
+
+    return True
+
+def decrypt_config(key):
+    if not key:
+        return
+
+    persist_path = cmd(persistpath_cmd).strip()
+    image_name = get_current_image()
+    image_path = os.path.join(persist_path, 'luks', image_name)
+
+    if not os.path.exists(image_path):
+        raise Exception("Encrypted config volume doesn't exist")
+
+    key_file = None
+
+    if not is_opened():
+        with NamedTemporaryFile(dir='/dev/shm', delete=False) as f:
+            f.write(key)
+            key_file = f.name
+
+        cmd(f'cryptsetup -q open {image_path} vyos_config --key-file={key_file}')
+
+    # unmount encrypted volume mount points
+    for path in mount_paths:
+        if os.path.ismount(path):
+            cmd(f'umount {path}')
+
+    # If /config is populated, move to /config.old
+    if len(os.listdir('/config')) > 0:
+        print('Moving existing /config folder to /config.old')
+        shutil.move('/config', '/config.old')
+
+    # Temporarily mount encrypted volume and migrate files to /config on rootfs
+    with TemporaryDirectory() as d:
+        cmd(f'mount /dev/mapper/vyos_config {d}')
+
+        # Move encrypted volume to /config
+        shutil.copytree(d, '/config', copy_function=shutil.move, dirs_exist_ok=True)
+        cmd(f'chgrp -R vyattacfg /config')
+
+        cmd(f'umount {d}')
+
+    # Close encrypted volume
+    cmd('cryptsetup -q close vyos_config')
+
+    # Remove encrypted volume image file and key
+    if key_file:
+        os.unlink(key_file)
+    os.unlink(image_path)
+
+    try:
+        clear_tpm_key()
+    except:
+        pass
+
+    return True
+
+if __name__ == '__main__':
+    if len(sys.argv) < 2:
+        print("Must specify action.")
+        sys.exit(1)
+
+    parser = ArgumentParser(description='Config encryption')
+    parser.add_argument('--disable', help='Disable encryption', action="store_true")
+    parser.add_argument('--enable', help='Enable encryption', action="store_true")
+    parser.add_argument('--load', help='Load encrypted config volume', action="store_true")
+    args = parser.parse_args()
+
+    tpm_exists = os.path.exists('/sys/class/tpm/tpm0')
+
+    key = None
+    recovery_key = None
+    need_recovery = False
+
+    question_key_str = 'recovery key' if tpm_exists else 'key'
+
+    if tpm_exists:
+        if args.enable:
+            key = Fernet.generate_key()
+        elif args.disable or args.load:
+            try:
+                key = read_tpm_key()
+                need_recovery = False
+            except:
+                print('Failed to read key from TPM, recovery key required')
+                need_recovery = True
+    else:
+        need_recovery = True
+
+    if args.enable and not tpm_exists:
+        print('WARNING: VyOS will boot into a default config when encrypted without a TPM')
+        print('You will need to manually login with default credentials and use "encryption load"')
+        print('to mount the encrypted volume and use "load /config/config.boot"')
+
+        if not ask_yes_no('Are you sure you want to proceed?'):
+            sys.exit(0)
+
+    if need_recovery or (args.enable and not ask_yes_no(f'Automatically generate a {question_key_str}?', default=True)):
+        while True:
+            recovery_key = ask_input(f'Enter {question_key_str}:', default=None).encode()
+
+            if len(recovery_key) >= 32:
+                break
+
+            print('Invalid key - must be at least 32 characters, try again.')
+    else:
+        recovery_key = Fernet.generate_key()
+
+    try:
+        if args.disable:
+            decrypt_config(key or recovery_key)
+
+            print('Encrypted config volume has been disabled')
+            print('Contents have been migrated to /config on rootfs')
+        elif args.load:
+            load_config(key or recovery_key)
+
+            print('Encrypted config volume has been mounted')
+            print('Use "load /config/config.boot" to load configuration')
+        elif args.enable and tpm_exists:
+            encrypt_config(key, recovery_key)
+
+            print('Encrypted config volume has been enabled with TPM')
+            print('Backup the recovery key in a safe place!')
+            print('Recovery key: ' + recovery_key.decode())
+        elif args.enable:
+            encrypt_config(recovery_key)
+
+            print('Encrypted config volume has been enabled without TPM')
+            print('Backup the key in a safe place!')
+            print('Key: ' + recovery_key.decode())
+    except Exception as e:
+        word = 'decrypt' if args.disable or args.load else 'encrypt'
+        print(f'Failed to {word} config: {e}')
diff --git a/src/init/vyos-router b/src/init/vyos-router
index 912a9ef3b..adf892371 100755
--- a/src/init/vyos-router
+++ b/src/init/vyos-router
@@ -64,6 +64,69 @@ disabled () {
     grep -q -w no-vyos-$1 /proc/cmdline
 }
 
+# Load encrypted config volume
+mount_encrypted_config() {
+    persist_path=$(/opt/vyatta/sbin/vyos-persistpath)
+    if [ $? == 0 ]; then
+        if [ -e $persist_path/boot ]; then
+            image_name=$(cat /proc/cmdline | sed -e s+^.*vyos-union=/boot/++ | sed -e 's/ .*$//')
+
+            if [ -z "$image_name" ]; then
+                return
+            fi
+
+            if [ ! -f $persist_path/luks/$image_name ]; then
+                return
+            fi
+
+            vyos_tpm_key=$(python3 -c 'from vyos.tpm import read_tpm_key; print(read_tpm_key().decode())' 2>/dev/null)
+
+            if [ $? -ne 0 ]; then
+                echo "ERROR: Failed to fetch encryption key from TPM. Encrypted config volume has not been mounted"
+                echo "Use 'encryption load' to load volume with recovery key"
+                echo "or 'encryption disable' to decrypt volume with recovery key"
+                return
+            fi
+
+            echo $vyos_tpm_key | tr -d '\r\n' | cryptsetup open $persist_path/luks/$image_name vyos_config --key-file=-
+
+            if [ $? -ne 0 ]; then
+                echo "ERROR: Failed to decrypt config volume. Encrypted config volume has not been mounted"
+                echo "Use 'encryption load' to load volume with recovery key"
+                echo "or 'encryption disable' to decrypt volume with recovery key"
+                return
+            fi
+
+            mount /dev/mapper/vyos_config /config
+            mount /dev/mapper/vyos_config $vyatta_sysconfdir/config
+
+            echo "Mounted encrypted config volume"
+        fi
+    fi
+}
+
+unmount_encrypted_config() {
+    persist_path=$(/opt/vyatta/sbin/vyos-persistpath)
+    if [ $? == 0 ]; then
+        if [ -e $persist_path/boot ]; then
+            image_name=$(cat /proc/cmdline | sed -e s+^.*vyos-union=/boot/++ | sed -e 's/ .*$//')
+
+            if [ -z "$image_name" ]; then
+                return
+            fi
+
+            if [ ! -f $persist_path/luks/$image_name ]; then
+                return
+            fi
+
+            umount /config
+            umount $vyatta_sysconfdir/config
+
+            cryptsetup close vyos_config
+        fi
+    fi
+}
+
 # if necessary, provide initial config
 init_bootfile () {
     if [ ! -r $BOOTFILE ] ; then
@@ -402,6 +465,8 @@ start ()
       && chgrp ${GROUP} ${vyatta_configdir}
     log_action_end_msg $?
 
+    mount_encrypted_config
+
     # T5239: early read of system hostname as this value is read-only once during
     # FRR initialisation
     tmp=$(${vyos_libexec_dir}/read-saved-value.py --path "system host-name")
@@ -470,6 +535,8 @@ stop()
     log_action_end_msg $?
 
     systemctl stop frr.service
+
+    unmount_encrypted_config
 }
 
 case "$action" in
diff --git a/src/op_mode/image_installer.py b/src/op_mode/image_installer.py
index d677c2cf8..85ebd19ba 100755
--- a/src/op_mode/image_installer.py
+++ b/src/op_mode/image_installer.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 #
-# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2023-2024 VyOS maintainers and contributors <maintainers@vyos.io>
 #
 # This file is part of VyOS.
 #
@@ -65,6 +65,8 @@ MSG_INPUT_ROOT_SIZE_SET: str = 'Please specify the size (in GB) of the root part
 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,14 +214,17 @@ 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):
@@ -227,32 +232,61 @@ def search_previous_installation(disks: list[str]) -> None:
                     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)
diff --git a/src/op_mode/image_manager.py b/src/op_mode/image_manager.py
index e64a85b95..1510a667c 100755
--- a/src/op_mode/image_manager.py
+++ b/src/op_mode/image_manager.py
@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 #
-# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+# Copyright 2023-2024 VyOS maintainers and contributors <maintainers@vyos.io>
 #
 # This file is part of VyOS.
 #
@@ -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"""