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.

repos/system_upgrade/common/actors/inhibitwhenluks/actor.py

@@ -1,9 +1,15 @@
 from leapp import reporting
 from leapp.actors import Actor
-from leapp.models import CephInfo, StorageInfo
+from leapp.libraries.common.config.version import get_target_major_version
+from leapp.models import CephInfo, LuksDump, TargetUserSpaceUpgradeTasks
 from leapp.reporting import create_report, Report
 from leapp.tags import ChecksPhaseTag, IPUWorkflowTag
 
+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
+
 
 class InhibitWhenLuks(Actor):
     """
@@ -13,12 +19,18 @@ class InhibitWhenLuks(Actor):
     """
 
     name = 'check_luks_and_inhibit'
-    consumes = (StorageInfo, CephInfo)
-    produces = (Report,)
+    consumes = (LuksDump, 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
+        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
         ceph_vol = []
         try:
             ceph_info = next(self.consume(CephInfo))
@@ -27,14 +39,51 @@ def process(self):
         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
+        required_crypt_rpms = [
+            'clevis',
+            'clevis-dracut',
+            'clevis-systemd',
+            'clevis-udisks2',
+            'clevis-luks',
+            'cryptsetup',
+            'tpm2-tss',
+            'tpm2-tools',
+            'tpm2-abrmd'
+        ]
+        self.produce(TargetUserSpaceUpgradeTasks(install_rpms=required_crypt_rpms))
+
+        for luks_dump in self.consume(LuksDump):
+            # if the device is managed by ceph, don't inhibit
+            if luks_dump.device_name in ceph_vol:
+                continue
+
+            if luks_dump.version == 1:
+                create_report([
+                    reporting.Title('Detected LUKS1 encrypted partition ({})'.format(luks_dump.device_name)),
+                    reporting.Summary('Upgrading system with LUKS1 disks is not supported.'),
+                    reporting.Severity(reporting.Severity.HIGH),
+                    reporting.Groups([reporting.Groups.BOOT, reporting.Groups.ENCRYPTION]),
+                    reporting.Groups([reporting.Groups.INHIBITOR]),
+                    reporting.Remediation(
+                        hint=("Convert your LUKS1 encrypted partition to LUKS2 and bind it to TPM2 using clevis.")
+                    ),
+                    reporting.ExternalLink(
+                        url=luks2_convert_doc_url,
+                        title='LUKS versions in RHEL: Conversion'
+                    )
+                ])
+            elif luks_dump.version == 2 and \
+                    not any(map(lambda x: x.token_type == "clevis-tpm2", luks_dump.tokens)):
+                create_report([
+                    reporting.Title('LUKS2 partition ({}) without clevis TPM2 binding detected'
+                                    .format(luks_dump.device_name)),
+                    reporting.Summary('Upgrading system with LUKS1 disks is not supported.'),
+                    reporting.Severity(reporting.Severity.HIGH),
+                    reporting.Groups([reporting.Groups.BOOT, reporting.Groups.ENCRYPTION]),
+                    reporting.Groups([reporting.Groups.INHIBITOR]),
+                    reporting.Remediation(hint="Add Clevis TPM2 binding to the volume."),
+                    reporting.ExternalLink(
+                        url=clevis_doc_url,
+                        title='Configuring manual enrollment of LUKS-encrypted volumes by using a TPM 2.0 policy'
+                    ),
+                ])

repos/system_upgrade/common/actors/inhibitwhenluks/tests/test_inhibitwhenluks.py

@@ -1,34 +1,89 @@
-from leapp.models import CephInfo, LsblkEntry, StorageInfo
+from leapp.libraries.common.config import version
+from leapp.models import CephInfo, LuksDump, LuksToken
 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")
+    current_actor_context.feed(luks_dump)
+    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("Detected LUKS1 encrypted partition")
+    assert luks_dump.device_name in report_fields['title']
 
-    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")
+    current_actor_context.feed(luks_dump)
     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 luks_dump.device_name in report_fields['title']
+    assert "without clevis TPM2 binding" in report_fields['title']
 
-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")])
+    current_actor_context.feed(luks_dump)
     current_actor_context.run()
-    assert not current_actor_context.consume(Report)
+    assert current_actor_context.consume(Report)
+    report_fields = current_actor_context.consume(Report)[0].report
+    assert is_inhibitor(report_fields)
+
+    assert luks_dump.device_name in report_fields['title']
+    assert "without clevis TPM2 binding" in report_fields['title']
 
 
-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='')]
+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")])
+    current_actor_context.feed(luks_dump)
+    current_actor_context.run()
+    assert not current_actor_context.consume(Report)
 
-    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")])
+    current_actor_context.feed(luks_dump)
     current_actor_context.run()
     assert not current_actor_context.consume(Report)