luks: Adds tpm2_pcrs config option for LUKS-encrypted devices

Adds the mentioned config option. The default value for it is 7, which
is the default value used by systemd-cryptenroll.

coriolis/cmd/worker.py

@@ -9,6 +9,7 @@ from oslo_reports import guru_meditation_report as gmr
 from oslo_reports import opts as gmr_opts
 
 from coriolis import constants
+from coriolis.osmorphing.osmount import luks_mixin
 from coriolis import service
 from coriolis import utils
 from coriolis.worker.rpc import server as rpc_server
@@ -25,6 +26,7 @@ CONF.register_opts(worker_opts, 'worker')
 def main():
     worker_count, args = service.get_worker_count_from_args(sys.argv)
     CONF(args[1:], project='coriolis', version="1.0.0")
+    luks_mixin.validate_config()
     if not worker_count:
         worker_count = CONF.worker.worker_count
     utils.setup_logging()

coriolis/osmorphing/osmount/luks_mixin.py

@@ -6,16 +6,33 @@ import json
 import os
 import re
 
+from oslo_config import cfg
+from oslo_config import types
 from oslo_log import log as logging
 
 from coriolis import constants
 from coriolis import exception
 from coriolis import utils
 
+luks_opts = [
+    cfg.ListOpt(
+        'tpm2_pcrs',
+        item_type=types.Integer(min=0, max=15),
+        default=['7'],
+        help='List of TPM2 PCR indexes the LUKS firstboot script binds its '
+             'TPM2 enrollment to. Defaults to PCR 7, matching the default '
+             'used by systemd-cryptenroll. Cannot be empty. Valid PCR '
+             'indexes are 0-15.'),
+]
+
+CONF = cfg.CONF
+CONF.register_opts(luks_opts, 'luks')
+
 LOG = logging.getLogger(__name__)
 
 _LUKS_KEYFILE_DIR = "/etc/luks"
 _DRACUT_LUKS_CONF_PATH = "/etc/dracut.conf.d/99-coriolis-luks.conf"
+_TPM2_PCRS_PLACEHOLDER = "__CORIOLIS_TPM2_PCRS__"
 
 # cryptsetup loads TPM2 token plugins via dlopen, so dracut's ldd analysis
 # misses them. List candidate paths in order of preference; the first one
@@ -41,6 +58,36 @@ _LUKS_FIRSTBOOT_SCRIPTS = {
 }
 
 
+def validate_config():
+    """Validate LUKS-related config options.
+
+    Called at worker service startup, so a misconfigured [luks] config section
+    fails fast, rather than surfacing into an actual LUKS migration.
+    """
+    if not CONF.luks.tpm2_pcrs:
+        raise exception.CoriolisException(
+            "The luks.tpm2_pcrs config option must not be empty; at least one "
+            "PCR index is required to enroll a TPM2 device.")
+
+
+def _render_tpm2_pcrs(script_content, initramfs_tool):
+    """Substitute the configured TPM2 PCR list into the given script_content.
+
+    The list is formatted based on the given initramfs_tool's enrollment
+    command syntax.
+    """
+    pcrs = CONF.luks.tpm2_pcrs
+    if initramfs_tool == "dracut":
+        # systemd-cryptenroll --tpm2-pcrs= syntax: "+"-separated entries,
+        # each optionally suffixed with a hash bank.
+        pcrs_value = "+".join("%s:sha256" % pcr for pcr in pcrs)
+    else:
+        # clevis tpm2 pin pcr_ids syntax: a plain comma-separated list.
+        pcrs_value = ",".join(str(pcr) for pcr in pcrs)
+
+    return script_content.replace(_TPM2_PCRS_PLACEHOLDER, pcrs_value)
+
+
 class LinuxLUKSMixin:
     """Mixin providing LUKS-related methods for BaseLinuxOSMountTools.
 
@@ -532,6 +579,7 @@ class LinuxLUKSMixin:
             raise exception.CoriolisException(
                 "No initramfs tool found in OS at '%s'; cannot install "
                 "LUKS firstboot cleanup script." % os_root_dir)
+        script_content = _render_tpm2_pcrs(script_content, initramfs_tool)
 
         os_morphing_tools.register_firstboot_script(
             script_content, user_provided=False,

coriolis/osmorphing/osmount/resources/luks_firstboot_dracut.sh

@@ -76,16 +76,23 @@ enroll_systemd_cryptenroll() {
     for dev in "${!dev_to_keyfile[@]}"; do
         local keyfile="${dev_to_keyfile[$dev]}"
 
-        if ! systemd-cryptenroll --tpm2-device=auto --tpm2-pcrs= \
+        # The PCRs list is substituted in by Coriolis.
+        if ! systemd-cryptenroll --tpm2-device=auto --tpm2-pcrs=__CORIOLIS_TPM2_PCRS__ \
                 --unlock-key-file="$keyfile" "$dev" 2>/dev/null; then
             echo "ERROR: systemd-cryptenroll failed for $dev; aborting to avoid lockout." >&2
             return 1
         fi
 
-        if ! cryptsetup luksDump "$dev" 2>/dev/null | grep -q 'systemd-tpm2'; then
+        local dump
+        dump=$(cryptsetup luksDump "$dev" 2>/dev/null)
+        if ! echo "$dump" | grep -q 'systemd-tpm2'; then
             echo "ERROR: systemd-tpm2 token not found in LUKS header for $dev; aborting to avoid lockout." >&2
             return 1
         fi
+        if ! echo "$dump" | grep 'tpm2-hash-pcrs:' | grep -q '[0-9]'; then
+            echo "ERROR: systemd-tpm2 token for $dev has no PCRs bound; aborting." >&2
+            return 1
+        fi
 
         echo "systemd-cryptenroll TPM2 enrollment verified for $dev."
     done

coriolis/osmorphing/osmount/resources/luks_firstboot_initramfs_tools.sh

@@ -58,7 +58,9 @@ enroll_clevis() {
     for dev in "${!dev_to_keyfile[@]}"; do
         local keyfile="${dev_to_keyfile[$dev]}"
 
-	if ! clevis luks bind -k "$keyfile" -d "$dev" tpm2 '{"pcr_ids":""}'; then
+        # The PCR IDs list is substituted in by Coriolis.
+        if ! clevis luks bind -k "$keyfile" -d "$dev" tpm2 \
+                '{"pcr_ids":"__CORIOLIS_TPM2_PCRS__","pcr_bank":"sha256"}'; then
             echo "ERROR: clevis luks bind failed for $dev; aborting to avoid lockout." >&2
             return 1
         fi
@@ -68,6 +70,11 @@ enroll_clevis() {
             return 1
         fi
 
+        if ! clevis luks list -d "$dev" 2>/dev/null | grep -q '"pcr_ids":"[0-9]'; then
+            echo "ERROR: clevis TPM2 pin for $dev has no PCRs bound; aborting." >&2
+            return 1
+        fi
+
 	echo "clevis TPM2 enrollment verified for $dev."
     done
 }

coriolis/tests/cmd/test_worker.py

@@ -6,6 +6,7 @@ from unittest import mock
 
 from coriolis.cmd import worker
 from coriolis import constants
+from coriolis.osmorphing.osmount import luks_mixin
 from coriolis import service
 from coriolis.tests import test_base
 from coriolis import utils
@@ -15,6 +16,7 @@ from coriolis.worker.rpc import server as rpc_server
 class WorkerTestCase(test_base.CoriolisBaseTestCase):
     """Test suite for the Coriolis worker CMD"""
 
+    @mock.patch.object(luks_mixin, 'validate_config')
     @mock.patch.object(service, 'service')
     @mock.patch.object(rpc_server, 'WorkerServerEndpoint')
     @mock.patch.object(service, 'MessagingService')
@@ -33,7 +35,8 @@ class WorkerTestCase(test_base.CoriolisBaseTestCase):
         mock_setup_logging,
         mock_MessagingService,
         mock_WorkerServerEndpoint,
-        mock_service
+        mock_service,
+        mock_validate_luks_config,
     ):
         worker_count = mock.sentinel.worker_count
         args = ['mock_arg_1', 'mock_arg_2']
@@ -44,6 +47,7 @@ class WorkerTestCase(test_base.CoriolisBaseTestCase):
         mock_get_worker_count_from_args.assert_called_once_with(mock_argv)
         mock_conf.assert_called_once_with(
             ['mock_arg_2'], project='coriolis', version="1.0.0")
+        mock_validate_luks_config.assert_called_once_with()
         mock_setup_logging.assert_called_once()
         mock_MessagingService.assert_called_once_with(
             constants.WORKER_MAIN_MESSAGING_TOPIC,

coriolis/tests/osmorphing/osmount/test_luks_mixin.py

@@ -51,6 +51,44 @@ class LinuxLUKSMixinTestCase(test_base.CoriolisBaseTestCase):
         )
         self.mixin._ssh = mock.MagicMock()
 
+    def test_validate_config_pcrs(self):
+        luks_mixin.CONF.set_override('tpm2_pcrs', [], group='luks')
+        self.addCleanup(
+            luks_mixin.CONF.clear_override, 'tpm2_pcrs', group='luks')
+
+        self.assertRaises(
+            exception.CoriolisException,
+            luks_mixin.validate_config)
+
+        luks_mixin.CONF.set_override('tpm2_pcrs', [4, 7], group='luks')
+
+        # Should not raise an exception anymore.
+        luks_mixin.validate_config()
+
+        # PCR indexes outside of the valid 0-15 range are rejected by the
+        # option's item_type as soon as they're set.
+        self.assertRaises(
+            ValueError,
+            luks_mixin.CONF.set_override, 'tpm2_pcrs', ['16'], group='luks')
+
+        self.assertRaises(
+            ValueError,
+            luks_mixin.CONF.set_override, 'tpm2_pcrs', ['foo'], group='luks')
+
+    def test_render_tpm2_pcrs(self):
+        luks_mixin.CONF.set_override('tpm2_pcrs', ['7', '11'], group='luks')
+        self.addCleanup(
+            luks_mixin.CONF.clear_override, 'tpm2_pcrs', group='luks')
+
+        dracut_out = luks_mixin._render_tpm2_pcrs(
+            luks_mixin._LUKS_FIRSTBOOT_SCRIPTS["dracut"], "dracut")
+        self.assertIn("--tpm2-pcrs=7:sha256+11:sha256", dracut_out)
+
+        clevis_out = luks_mixin._render_tpm2_pcrs(
+            luks_mixin._LUKS_FIRSTBOOT_SCRIPTS["update-initramfs"],
+            "update-initramfs")
+        self.assertIn('"pcr_ids":"7,11"', clevis_out)
+
     @mock.patch.object(luks_mixin.LinuxLUKSMixin, "_unlock_luks_device")
     def test__unlock_luks_devices(self, mock_unlock):
         mock_unlock.return_value = "/dev/mapper/coriolis_sda"
@@ -668,12 +706,14 @@ class LinuxLUKSMixinTestCase(test_base.CoriolisBaseTestCase):
             "--include /etc/crypttab /etc/crypttab" % _OS_ROOT_DIR
         )
 
+    @mock.patch.object(luks_mixin, '_render_tpm2_pcrs')
     @mock.patch.object(luks_mixin.LinuxLUKSMixin, '_detect_initramfs_tool')
     @mock.patch.object(luks_mixin.LinuxLUKSMixin, '_rebuild_initramfs')
     @mock.patch.object(luks_mixin.LinuxLUKSMixin, '_fix_grub_luks_root')
     @mock.patch.object(luks_mixin.LinuxLUKSMixin, '_write_migration_keyfiles')
     def test_install_encryption_firstboot_setup(
-        self, mock_write_keyfiles, mock_grub, mock_rebuild, mock_detect_tool
+        self, mock_write_keyfiles, mock_grub, mock_rebuild, mock_detect_tool,
+        mock_render_tpm2_pcrs,
     ):
         mock_morphing_tools = mock.MagicMock()
 
@@ -694,8 +734,11 @@ class LinuxLUKSMixinTestCase(test_base.CoriolisBaseTestCase):
         mock_write_keyfiles.assert_called_once_with(_OS_ROOT_DIR)
         mock_grub.assert_called_once_with(_OS_ROOT_DIR)
         mock_rebuild.assert_called_once_with(_OS_ROOT_DIR)
+
+        mock_render_tpm2_pcrs.assert_called_once_with(
+            luks_mixin._LUKS_FIRSTBOOT_SCRIPTS["dracut"], "dracut")
         mock_morphing_tools.register_firstboot_script.assert_called_once_with(
-            luks_mixin._LUKS_FIRSTBOOT_SCRIPTS["dracut"],
+            mock_render_tpm2_pcrs.return_value,
             user_provided=False,
             script_filename="luks-firstboot.sh",
         )