From ff4adaeee49392d9b4e43c21bddba7dc7589689d Mon Sep 17 00:00:00 2001 From: Daniel Zatovic Date: Tue, 16 Apr 2024 17:06:18 +0200 Subject: [PATCH] InhibitWhenLuks: modify the inhibitor to use LuksDump Consume LuksDump messages to decide whether the upgrade process should be inhibited. If all devices are LUKS2 with clevis TPM2 binding, don't inhibit. --- .../common/actors/inhibitwhenluks/actor.py | 31 +---- .../libraries/inhibitwhenluks.py | 119 ++++++++++++++++++ .../tests/test_inhibitwhenluks.py | 113 ++++++++++++++--- .../common/actors/luksscanner/actor.py | 6 +- .../luksscanner/libraries/luksscanner.py | 6 +- .../luksscanner/tests/test_luksscaner.py | 9 +- .../system_upgrade/common/models/luksdump.py | 16 ++- 7 files changed, 250 insertions(+), 50 deletions(-) create mode 100644 repos/system_upgrade/common/actors/inhibitwhenluks/libraries/inhibitwhenluks.py diff --git a/repos/system_upgrade/common/actors/inhibitwhenluks/actor.py b/repos/system_upgrade/common/actors/inhibitwhenluks/actor.py index 40b845b01d..fb9e7a0285 100644 --- a/repos/system_upgrade/common/actors/inhibitwhenluks/actor.py +++ b/repos/system_upgrade/common/actors/inhibitwhenluks/actor.py @@ -1,7 +1,7 @@ -from leapp import reporting from leapp.actors import Actor -from leapp.models import CephInfo, StorageInfo -from leapp.reporting import create_report, Report +from leapp.libraries.actor.inhibitwhenluks import check_invalid_luks_devices +from leapp.models import CephInfo, LuksDumps, TargetUserSpaceUpgradeTasks +from leapp.reporting import Report from leapp.tags import ChecksPhaseTag, IPUWorkflowTag @@ -13,28 +13,9 @@ class InhibitWhenLuks(Actor): """ name = 'check_luks_and_inhibit' - consumes = (StorageInfo, CephInfo) - produces = (Report,) + consumes = (LuksDumps, CephInfo) + produces = (Report, TargetUserSpaceUpgradeTasks) tags = (ChecksPhaseTag, IPUWorkflowTag) def process(self): - # If encrypted Ceph volumes present, check if there are more encrypted disk in lsblk than Ceph vol - ceph_vol = [] - try: - ceph_info = next(self.consume(CephInfo)) - if ceph_info: - ceph_vol = ceph_info.encrypted_volumes[:] - except StopIteration: - pass - - for storage_info in self.consume(StorageInfo): - for blk in storage_info.lsblk: - if blk.tp == 'crypt' and blk.name not in ceph_vol: - create_report([ - reporting.Title('LUKS encrypted partition detected'), - reporting.Summary('Upgrading system with encrypted partitions is not supported'), - reporting.Severity(reporting.Severity.HIGH), - reporting.Groups([reporting.Groups.BOOT, reporting.Groups.ENCRYPTION]), - reporting.Groups([reporting.Groups.INHIBITOR]), - ]) - break + check_invalid_luks_devices() diff --git a/repos/system_upgrade/common/actors/inhibitwhenluks/libraries/inhibitwhenluks.py b/repos/system_upgrade/common/actors/inhibitwhenluks/libraries/inhibitwhenluks.py new file mode 100644 index 0000000000..ef567df331 --- /dev/null +++ b/repos/system_upgrade/common/actors/inhibitwhenluks/libraries/inhibitwhenluks.py @@ -0,0 +1,119 @@ +from leapp import reporting +from leapp.libraries.common.config.version import get_target_major_version +from leapp.libraries.stdlib import api +from leapp.models import CephInfo, DracutModule, LuksDumps, TargetUserSpaceUpgradeTasks, UpgradeInitramfsTasks +from leapp.reporting import create_report + +CLEVIS_RHEL8_DOC_URL = 'https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/8/html/security_hardening/configuring-automated-unlocking-of-encrypted-volumes-using-policy-based-decryption_security-hardening#configuring-manual-enrollment-of-volumes-using-tpm2_configuring-automated-unlocking-of-encrypted-volumes-using-policy-based-decryption' # noqa: E501; pylint: disable=line-too-long +CLEVIS_RHEL9_DOC_URL = 'https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/9/html/security_hardening/configuring-automated-unlocking-of-encrypted-volumes-using-policy-based-decryption_security-hardening#configuring-manual-enrollment-of-volumes-using-tpm2_configuring-automated-unlocking-of-encrypted-volumes-using-policy-based-decryption' # noqa: E501; pylint: disable=line-too-long +LUKS2_CONVERT_RHEL8_DOC_URL = 'https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/8/html/security_hardening/encrypting-block-devices-using-luks_security-hardening#luks-versions-in-rhel_encrypting-block-devices-using-luks' # noqa: E501; pylint: disable=line-too-long +LUKS2_CONVERT_RHEL9_DOC_URL = 'https://access.redhat.com/documentation/en-us/red_hat_enterprise_linux/9/html/security_hardening/encrypting-block-devices-using-luks_security-hardening#luks-versions-in-rhel_encrypting-block-devices-using-luks' # noqa: E501; pylint: disable=line-too-long +FMT_LIST_SEPARATOR = '\n - ' + + +def _at_least_one_tpm_token(luks_dump): + return any([token.token_type == "clevis-tpm2" for token in luks_dump.tokens]) + + +def check_invalid_luks_devices(): + target_major_version = get_target_major_version() + if target_major_version == '8': + clevis_doc_url = CLEVIS_RHEL8_DOC_URL + luks2_convert_doc_url = LUKS2_CONVERT_RHEL8_DOC_URL + elif target_major_version == '9': + clevis_doc_url = CLEVIS_RHEL9_DOC_URL + luks2_convert_doc_url = LUKS2_CONVERT_RHEL9_DOC_URL + else: + create_report([ + reporting.Title('LUKS encrypted partition detected'), + reporting.Summary('Upgrading system with encrypted partitions is not supported'), + reporting.Severity(reporting.Severity.HIGH), + reporting.Groups([reporting.Groups.BOOT, reporting.Groups.ENCRYPTION]), + reporting.Groups([reporting.Groups.INHIBITOR]), + ]) + return + ceph_vol = [] + try: + ceph_info = next(api.consume(CephInfo), None) + if ceph_info: + ceph_vol = ceph_info.encrypted_volumes[:] + except StopIteration: + pass + + luks_dumps = next(api.consume(LuksDumps), None) + if luks_dumps is None: + return + + for luks_dump in luks_dumps.dumps: + # if the device is managed by ceph, don't inhibit + if luks_dump.device_name in ceph_vol: + continue + + list_luks1_partitions = [] + list_no_tpm2_partitions = [] + + if luks_dump.version == 1: + list_luks1_partitions.append(luks_dump.device_name) + elif luks_dump.version == 2 and not _at_least_one_tpm_token(luks_dump): + list_no_tpm2_partitions.append(luks_dump.device_name) + + if list_luks1_partitions or list_no_tpm2_partitions: + summary = ( + 'Only systems where all encrypted devices are LUKS2 ' + 'devices with Clevis TPM 2.0 token can be updated.' + ) + report_hints = [] + + if list_luks1_partitions: + luks1_partitions_text = '' + for partition in list_luks1_partitions: + luks1_partitions_text += '{0}{1}'.format(FMT_LIST_SEPARATOR, partition) + + summary += '\nThe following LUKS1 partitions have been discovered on your system: ' + summary += luks1_partitions_text + report_hints.append(reporting.Remediation( + hint=("Convert your LUKS1 encrypted partition to LUKS2 and bind it to TPM2 using clevis.") + )) + report_hints.append(reporting.ExternalLink( + url=luks2_convert_doc_url, + title='LUKS versions in RHEL: Conversion' + )) + + if list_no_tpm2_partitions: + no_tpm2_partitions_text = '' + for partition in list_no_tpm2_partitions: + no_tpm2_partitions_text += '{0}{1}'.format(FMT_LIST_SEPARATOR, partition) + + summary += ('\nThe following LUKS2 devices without clevis TPM2 token ' + 'have been discovered on your system:') + summary += no_tpm2_partitions_text + + report_hints.append(reporting.Remediation(hint="Add Clevis TPM2 binding to the volume.")) + report_hints.append(reporting.ExternalLink( + url=clevis_doc_url, + title='Configuring manual enrollment of LUKS-encrypted volumes by using a TPM 2.0 policy' + ) + ) + + create_report([ + reporting.Title('Invalid LUKS encrypted partition detected'), + reporting.Summary(summary), + reporting.Severity(reporting.Severity.HIGH), + reporting.Groups([reporting.Groups.BOOT, reporting.Groups.ENCRYPTION]), + reporting.Groups([reporting.Groups.INHIBITOR]), + ] + report_hints) + else: + required_crypt_rpms = [ + 'clevis', + 'clevis-dracut', + 'clevis-systemd', + 'clevis-udisks2', + 'clevis-luks', + 'cryptsetup', + 'tpm2-tss', + 'tpm2-tools', + 'tpm2-abrmd' + ] + api.produce(TargetUserSpaceUpgradeTasks(install_rpms=required_crypt_rpms)) + api.produce(UpgradeInitramfsTasks(include_dracut_modules=[ + DracutModule(name='clevis'), DracutModule(name='clevis-pin-tpm2')])) diff --git a/repos/system_upgrade/common/actors/inhibitwhenluks/tests/test_inhibitwhenluks.py b/repos/system_upgrade/common/actors/inhibitwhenluks/tests/test_inhibitwhenluks.py index 405a34295b..031a2d37f0 100644 --- a/repos/system_upgrade/common/actors/inhibitwhenluks/tests/test_inhibitwhenluks.py +++ b/repos/system_upgrade/common/actors/inhibitwhenluks/tests/test_inhibitwhenluks.py @@ -1,34 +1,117 @@ -from leapp.models import CephInfo, LsblkEntry, StorageInfo +from leapp.libraries.common.config import version +from leapp.models import CephInfo, LuksDump, LuksDumps, LuksToken, TargetUserSpaceUpgradeTasks from leapp.reporting import Report from leapp.snactor.fixture import current_actor_context from leapp.utils.report import is_inhibitor -def test_actor_with_luks(current_actor_context): - with_luks = [LsblkEntry(name='luks-132', kname='kname1', maj_min='253:0', rm='0', size='10G', bsize=10*(1 << 39), - ro='0', tp='crypt', mountpoint='', parent_name='', parent_path='')] +def test_actor_with_luks1(monkeypatch, current_actor_context): + monkeypatch.setattr(version, 'get_target_major_version', lambda: '8') + current_actor_context.feed(CephInfo(encrypted_volumes=[])) + luks_dump = LuksDump( + version=1, + uuid="dd09e6d4-b595-4f1c-80b8-fd47540e6464", + device_path="/dev/sda", + device_name="sda") + luks_dumps = LuksDumps(dumps=[luks_dump]) + current_actor_context.feed(luks_dumps) + current_actor_context.run() + assert current_actor_context.consume(Report) + report_fields = current_actor_context.consume(Report)[0].report + assert is_inhibitor(report_fields) + assert not current_actor_context.consume(TargetUserSpaceUpgradeTasks) + + assert report_fields['title'].startswith("Invalid LUKS encrypted partition detected") + assert "LUKS1 partitions have been discovered on your system" in report_fields['summary'] + assert luks_dump.device_name in report_fields['summary'] - current_actor_context.feed(StorageInfo(lsblk=with_luks)) + +def test_actor_with_luks2(monkeypatch, current_actor_context): + monkeypatch.setattr(version, 'get_target_major_version', lambda: '8') + current_actor_context.feed(CephInfo(encrypted_volumes=[])) + luks_dump = LuksDump( + version=2, + uuid="27b57c75-9adf-4744-ab04-9eb99726a301", + device_path="/dev/sda", + device_name="sda") + luks_dumps = LuksDumps(dumps=[luks_dump]) + current_actor_context.feed(luks_dumps) current_actor_context.run() assert current_actor_context.consume(Report) report_fields = current_actor_context.consume(Report)[0].report assert is_inhibitor(report_fields) + assert not current_actor_context.consume(TargetUserSpaceUpgradeTasks) + assert report_fields['title'].startswith("Invalid LUKS encrypted partition detected") + assert "LUKS2 devices without clevis TPM2 token have been discovered on your system" in report_fields['summary'] + assert luks_dump.device_name in report_fields['summary'] -def test_actor_with_luks_ceph_only(current_actor_context): - with_luks = [LsblkEntry(name='luks-132', kname='kname1', maj_min='253:0', rm='0', size='10G', bsize=10*(1 << 39), - ro='0', tp='crypt', mountpoint='', parent_name='', parent_path='')] - ceph_volume = ['luks-132'] - current_actor_context.feed(StorageInfo(lsblk=with_luks)) - current_actor_context.feed(CephInfo(encrypted_volumes=ceph_volume)) + +def test_actor_with_luks2_invalid_token(monkeypatch, current_actor_context): + monkeypatch.setattr(version, 'get_target_major_version', lambda: '8') + current_actor_context.feed(CephInfo(encrypted_volumes=[])) + luks_dump = LuksDump( + version=2, + uuid="dc1dbe37-6644-4094-9839-8fc5dcbec0c6", + device_path="/dev/sda", + device_name="sda", + tokens=[LuksToken(token_id=0, keyslot=1, token_type="clevis")]) + luks_dumps = LuksDumps(dumps=[luks_dump]) + current_actor_context.feed(luks_dumps) + current_actor_context.run() + assert current_actor_context.consume(Report) + report_fields = current_actor_context.consume(Report)[0].report + assert is_inhibitor(report_fields) + + assert report_fields['title'].startswith("Invalid LUKS encrypted partition detected") + assert "LUKS2 devices without clevis TPM2 token have been discovered on your system" in report_fields['summary'] + assert luks_dump.device_name in report_fields['summary'] + assert not current_actor_context.consume(TargetUserSpaceUpgradeTasks) + + +def test_actor_with_luks2_clevis_tpm_token(monkeypatch, current_actor_context): + monkeypatch.setattr(version, 'get_target_major_version', lambda: '8') + current_actor_context.feed(CephInfo(encrypted_volumes=[])) + luks_dump = LuksDump( + version=2, + uuid="83050bd9-61c6-4ff0-846f-bfd3ac9bfc67", + device_path="/dev/sda", + device_name="sda", + tokens=[LuksToken(token_id=0, keyslot=1, token_type="clevis-tpm2")]) + luks_dumps = LuksDumps(dumps=[luks_dump]) + current_actor_context.feed(luks_dumps) current_actor_context.run() assert not current_actor_context.consume(Report) + upgrade_tasks = current_actor_context.consume(TargetUserSpaceUpgradeTasks) + assert len(upgrade_tasks) == 1 + assert set(upgrade_tasks[0].install_rpms) == set([ + 'clevis', + 'clevis-dracut', + 'clevis-systemd', + 'clevis-udisks2', + 'clevis-luks', + 'cryptsetup', + 'tpm2-tss', + 'tpm2-tools', + 'tpm2-abrmd' + ]) -def test_actor_without_luks(current_actor_context): - without_luks = [LsblkEntry(name='sda1', kname='sda1', maj_min='8:0', rm='0', size='10G', bsize=10*(1 << 39), - ro='0', tp='part', mountpoint='/boot', parent_name='', parent_path='')] - current_actor_context.feed(StorageInfo(lsblk=without_luks)) +def test_actor_with_luks2_ceph(monkeypatch, current_actor_context): + monkeypatch.setattr(version, 'get_target_major_version', lambda: '8') + ceph_volume = ['sda'] + current_actor_context.feed(CephInfo(encrypted_volumes=ceph_volume)) + luks_dump = LuksDump( + version=2, + uuid="0edb8c11-1a04-4abd-a12d-93433ee7b8d8", + device_path="/dev/sda", + device_name="sda", + tokens=[LuksToken(token_id=0, keyslot=1, token_type="clevis")]) + luks_dumps = LuksDumps(dumps=[luks_dump]) + current_actor_context.feed(luks_dumps) current_actor_context.run() assert not current_actor_context.consume(Report) + + # make sure we don't needlessly include clevis packages, when there is no clevis token + assert not current_actor_context.consume(TargetUserSpaceUpgradeTasks) diff --git a/repos/system_upgrade/common/actors/luksscanner/actor.py b/repos/system_upgrade/common/actors/luksscanner/actor.py index 382e858b82..a163374be3 100644 --- a/repos/system_upgrade/common/actors/luksscanner/actor.py +++ b/repos/system_upgrade/common/actors/luksscanner/actor.py @@ -1,6 +1,6 @@ from leapp.actors import Actor from leapp.libraries.actor import luksscanner -from leapp.models import LuksDump, StorageInfo +from leapp.models import LuksDumps, StorageInfo from leapp.reporting import Report from leapp.tags import FactsPhaseTag, IPUWorkflowTag @@ -16,8 +16,8 @@ class LuksScanner(Actor): name = 'luks_scanner' consumes = (StorageInfo,) - produces = (Report, LuksDump) + produces = (Report, LuksDumps) tags = (IPUWorkflowTag, FactsPhaseTag) def process(self): - self.produce(*luksscanner.get_luks_dumps()) + self.produce(luksscanner.get_luks_dumps_model()) diff --git a/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py b/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py index 0a80a33223..c042e9fd17 100644 --- a/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py +++ b/repos/system_upgrade/common/actors/luksscanner/libraries/luksscanner.py @@ -4,7 +4,7 @@ from leapp.libraries import stdlib from leapp.libraries.actor.luksdump_parser import LuksDumpParser from leapp.libraries.stdlib import api -from leapp.models import LuksDump, LuksToken, StorageInfo +from leapp.models import LuksDump, LuksDumps, LuksToken, StorageInfo def aslist(f): @@ -118,3 +118,7 @@ def get_luks_dumps(): for blk in storage_info.lsblk: if blk.tp == 'crypt' and blk.parent_path: yield get_luks_dump_by_device(blk.parent_path, blk.parent_name) + + +def get_luks_dumps_model(): + return LuksDumps(dumps=get_luks_dumps()) diff --git a/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py b/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py index 12f4810016..22eb0946f1 100644 --- a/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py +++ b/repos/system_upgrade/common/actors/luksscanner/tests/test_luksscaner.py @@ -3,10 +3,8 @@ import pytest from leapp.libraries.stdlib import api -from leapp.models import CephInfo, LsblkEntry, LuksDump, StorageInfo -from leapp.reporting import Report +from leapp.models import LsblkEntry, LuksDumps, StorageInfo from leapp.snactor.fixture import current_actor_context -from leapp.utils.report import is_inhibitor CUR_DIR = os.path.dirname(os.path.abspath(__file__)) @@ -127,9 +125,10 @@ def test_actor_with_luks(monkeypatch, current_actor_context, variant, luks_versi current_actor_context.feed(StorageInfo(lsblk=with_luks)) current_actor_context.run() - luks_dumps = current_actor_context.consume(LuksDump) + luks_dumps = current_actor_context.consume(LuksDumps) assert len(luks_dumps) == 1 - luks_dump = luks_dumps[0] + assert len(luks_dumps[0].dumps) == 1 + luks_dump = luks_dumps[0].dumps[0] assert luks_dump.version == luks_version assert luks_dump.uuid == uuid diff --git a/repos/system_upgrade/common/models/luksdump.py b/repos/system_upgrade/common/models/luksdump.py index e3a3b8fbd9..511c10cbc7 100644 --- a/repos/system_upgrade/common/models/luksdump.py +++ b/repos/system_upgrade/common/models/luksdump.py @@ -29,7 +29,9 @@ class LuksToken(Model): class LuksDump(Model): """ - Information about LUKS-encrypted device. + Information about a single LUKS-encrypted device. + + Note this model is supposed to be used as a part of LuksDumps msg. """ topic = SystemInfoTopic @@ -57,3 +59,15 @@ class LuksDump(Model): """ List of LUKS2 tokens """ + + +class LuksDumps(Model): + """ + Information about a all LUKS-encrypted devices on the system. + """ + topic = SystemInfoTopic + + dumps = fields.List(fields.Model(LuksDump)) + """ + List of LuksDump representing all the encrypted devices on the system. + """