Update GRUB core on legacy (BIOS) systems.

On legacy (BIOS) systems, GRUB core (located in the gap between the MBR and the
first partition), does not get automatically updated when GRUB is upgraded.

This actor also helps to mitigate an issue with randomly booting into
old RHEL7 kernel.

repos/system_upgrade/el7toel8/actors/checkgrubcore/actor.py

@@ -0,0 +1,53 @@
+from leapp.actors import Actor
+from leapp.libraries.common.config import architecture
+from leapp.models import FirmwareFacts, GrubDevice, UpdateGrub
+from leapp.reporting import Report, create_report
+from leapp import reporting
+from leapp.tags import ChecksPhaseTag, IPUWorkflowTag
+
+
+GRUB_SUMMARY = ('On legacy (BIOS) systems, GRUB core (located in the gap between the MBR and the '
+                'first partition) does not get automatically updated when GRUB is upgraded.')
+
+
+class CheckGrubCore(Actor):
+    """
+    Check whether we are on legacy (BIOS) system and instruct Leapp to upgrade GRUB core
+    """
+
+    name = 'check_grub_core'
+    consumes = (FirmwareFacts, GrubDevice)
+    produces = (Report, UpdateGrub)
+    tags = (ChecksPhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        if architecture.matches_architecture(architecture.ARCH_S390X):
+            # s390x archs use ZIPL instead of GRUB
+            return
+
+        ff = next(self.consume(FirmwareFacts), None)
+        if ff and ff.firmware == 'bios':
+            dev = next(self.consume(GrubDevice), None)
+            if dev:
+                self.produce(UpdateGrub(grub_device=dev.grub_device))
+                create_report([
+                    reporting.Title(
+                        'GRUB core on {} will be updated during upgrade'.format(dev.grub_device)
+                    ),
+                    reporting.Summary(GRUB_SUMMARY),
+                    reporting.Severity(reporting.Severity.HIGH),
+                    reporting.Tags([reporting.Tags.BOOT]),
+                ])
+            else:
+                create_report([
+                    reporting.Title('Leapp could not identify where GRUB core is located'),
+                    reporting.Summary(
+                        'We assume GRUB core is located on the same device as /boot. Leapp needs to '
+                        'update GRUB core as it is not done automatically on legacy (BIOS) systems. '
+                    ),
+                    reporting.Severity(reporting.Severity.HIGH),
+                    reporting.Tags([reporting.Tags.BOOT]),
+                    reporting.Remediation(
+                        hint='Please use "LEAPP_GRUB_DEVICE" environment variable to point Leapp to '
+                             'device where GRUB core is located'),
+                ])

repos/system_upgrade/el7toel8/actors/checkgrubcore/tests/test_checkgrubcore.py

@@ -0,0 +1,34 @@
+from leapp.libraries.common.config import mock_configs
+from leapp.models import GrubDevice, UpdateGrub, FirmwareFacts
+from leapp.reporting import Report
+
+
+def test_actor_update_grub(current_actor_context):
+    current_actor_context.feed(FirmwareFacts(firmware='bios'))
+    current_actor_context.feed(GrubDevice(grub_device='/dev/vda'))
+    current_actor_context.run(config_model=mock_configs.CONFIG)
+    assert current_actor_context.consume(Report)
+    assert current_actor_context.consume(UpdateGrub)
+    assert current_actor_context.consume(UpdateGrub)[0].grub_device == '/dev/vda'
+
+
+def test_actor_no_grub_device(current_actor_context):
+    current_actor_context.feed(FirmwareFacts(firmware='bios'))
+    current_actor_context.run(config_model=mock_configs.CONFIG)
+    assert current_actor_context.consume(Report)
+    assert not current_actor_context.consume(UpdateGrub)
+
+
+def test_actor_with_efi(current_actor_context):
+    current_actor_context.feed(FirmwareFacts(firmware='efi'))
+    current_actor_context.run(config_model=mock_configs.CONFIG)
+    assert not current_actor_context.consume(Report)
+    assert not current_actor_context.consume(UpdateGrub)
+
+
+def test_s390x(current_actor_context):
+    current_actor_context.feed(FirmwareFacts(firmware='bios'))
+    current_actor_context.feed(GrubDevice(grub_device='/dev/vda'))
+    current_actor_context.run(config_model=mock_configs.CONFIG_S390X)
+    assert not current_actor_context.consume(Report)
+    assert not current_actor_context.consume(UpdateGrub)

repos/system_upgrade/el7toel8/actors/grubdevname/actor.py

@@ -0,0 +1,18 @@
+from leapp.actors import Actor
+from leapp.libraries.actor.library import get_grub_device
+from leapp.models import GrubDevice
+from leapp.tags import FactsPhaseTag, IPUWorkflowTag
+
+
+class Grubdevname(Actor):
+    """
+    Get name of block device where GRUB is located
+    """
+
+    name = 'grubdevname'
+    consumes = ()
+    produces = (GrubDevice,)
+    tags = (FactsPhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        get_grub_device()

repos/system_upgrade/el7toel8/actors/grubdevname/libraries/library.py

@@ -0,0 +1,69 @@
+import os
+
+from leapp.libraries.stdlib import run, api, CalledProcessError
+from leapp.exceptions import StopActorExecution
+from leapp.models import GrubDevice
+
+
+def has_grub(blk_dev):
+    """
+    Check whether GRUB is present on block device
+    """
+    try:
+        result = run(['dd', 'status=none', 'if={}'.format(blk_dev), 'bs=512', 'count=1'], encoding=None)
+    except CalledProcessError:
+        api.current_logger().warning(
+            'Could not read first sector of {} in order to identify the bootloader'.format(blk_dev)
+        )
+        raise StopActorExecution()
+    return b'GRUB' in result['stdout']
+
+
+def blk_dev_from_partition(partition):
+    """
+    Find parent device of /boot partition
+    """
+    try:
+        result = run(['lsblk', '-spnlo', 'name', partition])
+    except CalledProcessError:
+        api.current_logger().warning(
+            'Could not get parent device of {} partition'.format(partition)
+        )
+        raise StopActorExecution()
+    # lsblk "-s" option prints dependencies in inverse order, so the parent device will always
+    # be the last or the only device.
+    # Command result example:
+    # 'result', {'signal': 0, 'pid': 3872, 'exit_code': 0, 'stderr': u'', 'stdout': u'/dev/vda1\n/dev/vda\n'}
+    return result['stdout'].strip().split()[-1]
+
+
+def get_boot_partition():
+    """
+    Get /boot partition
+    """
+    try:
+        # call grub2-probe to identify /boot partition
+        result = run(['grub2-probe', '--target=device', '/boot'])
+    except CalledProcessError:
+        api.current_logger().warning(
+            'Could not get name of underlying /boot partition'
+        )
+        raise StopActorExecution()
+    return result['stdout'].strip()
+
+
+def get_grub_device():
+    """
+    Get block device where GRUB is located. We assume GRUB is on the same device
+    as /boot partition is.
+
+    """
+    grub_dev = os.getenv('LEAPP_GRUB_DEVICE', None)
+    if grub_dev:
+        api.produce(GrubDevice(grub_device=grub_dev))
+        return
+    boot_partition = get_boot_partition()
+    grub_dev = blk_dev_from_partition(boot_partition)
+    if grub_dev:
+        if has_grub(grub_dev):
+            api.produce(GrubDevice(grub_device=grub_dev))

repos/system_upgrade/el7toel8/actors/grubdevname/tests/test_grubdevname.py

@@ -0,0 +1,89 @@
+import pytest
+
+from leapp.exceptions import StopActorExecution
+from leapp.libraries.stdlib import api, CalledProcessError
+from leapp.libraries.common import testutils
+from leapp.libraries.actor import library
+
+
+BOOT_PARTITION = '/dev/vda1'
+
+BOOT_DEVICE = '/dev/vda'
+BOOT_DEVICE_ENV = '/dev/sda'
+
+VALID_DD = b'GRUB GeomHard DiskRead Error'
+INVALID_DD = b'Nothing here'
+
+
+def raise_call_error(args=None):
+    raise CalledProcessError(
+        message='A Leapp Command Error occured.',
+        command=args,
+        result={'signal': None, 'exit_code': 1, 'pid': 0, 'stdout': 'fake', 'stderr': 'fake'}
+    )
+
+
+class RunMocked(object):
+
+    def __init__(self, no_grub=False, raise_err=False):
+        self.called = 0
+        self.args = None
+        self.no_grub = no_grub
+        self.raise_err = raise_err
+
+    def __call__(self, args, encoding=None):
+        self.called += 1
+        self.args = args
+        if self.raise_err:
+            raise_call_error(args)
+
+        if self.args == ['grub2-probe', '--target=device', '/boot']:
+            stdout = BOOT_PARTITION
+
+        elif self.args == ['lsblk', '-spnlo', 'name', BOOT_PARTITION]:
+            stdout = BOOT_DEVICE
+
+        elif self.args == [
+            'dd', 'status=none', 'if={}'.format(BOOT_DEVICE), 'bs=512', 'count=1'
+        ]:
+            stdout = VALID_DD if not self.no_grub else INVALID_DD
+
+        return {'stdout': stdout}
+
+
+def test_get_grub_device(monkeypatch):
+    run_mocked = RunMocked()
+    monkeypatch.setattr(library, 'run', run_mocked)
+    monkeypatch.setattr(api, 'produce', testutils.produce_mocked())
+    library.get_grub_device()
+    assert library.run.called == 3
+    assert BOOT_DEVICE == api.produce.model_instances[0].grub_device
+
+
+def test_get_grub_device_fail(monkeypatch):
+    run_mocked = RunMocked(raise_err=True)
+    monkeypatch.setattr(library, 'run', run_mocked)
+    monkeypatch.setattr(api, 'produce', testutils.produce_mocked())
+    with pytest.raises(StopActorExecution):
+        library.get_grub_device()
+    assert library.run.called == 1
+    assert not api.produce.model_instances
+
+
+def test_grub_device_env_var(monkeypatch):
+    run_mocked = RunMocked()
+    monkeypatch.setenv('LEAPP_GRUB_DEVICE', BOOT_DEVICE_ENV)
+    monkeypatch.setattr(library, 'run', run_mocked)
+    monkeypatch.setattr(api, 'produce', testutils.produce_mocked())
+    library.get_grub_device()
+    assert library.run.called == 0
+    assert BOOT_DEVICE_ENV == api.produce.model_instances[0].grub_device
+
+
+def test_device_no_grub(monkeypatch):
+    run_mocked = RunMocked(no_grub=True)
+    monkeypatch.setattr(library, 'run', run_mocked)
+    monkeypatch.setattr(api, 'produce', testutils.produce_mocked())
+    library.get_grub_device()
+    assert library.run.called == 3
+    assert not api.produce.model_instances

repos/system_upgrade/el7toel8/actors/updategrubcore/actor.py

@@ -0,0 +1,22 @@
+from leapp.actors import Actor
+from leapp.libraries.actor.library import update_grub_core
+from leapp.models import TransactionCompleted, UpdateGrub
+from leapp.reporting import Report
+from leapp.tags import RPMUpgradePhaseTag, IPUWorkflowTag
+
+
+class UpdateGrubCore(Actor):
+    """
+    On legacy (BIOS) systems, GRUB core (located in the gap between the MBR and the
+    first partition), does not get automatically updated when GRUB is upgraded.
+    """
+
+    name = 'update_grub_core'
+    consumes = (TransactionCompleted, UpdateGrub)
+    produces = (Report,)
+    tags = (RPMUpgradePhaseTag, IPUWorkflowTag)
+
+    def process(self):
+        dev = next(self.consume(UpdateGrub), None)
+        if dev:
+            update_grub_core(dev.grub_device)

repos/system_upgrade/el7toel8/actors/updategrubcore/libraries/library.py

@@ -0,0 +1,32 @@
+from leapp.libraries.stdlib import api, run, CalledProcessError
+from leapp.exceptions import StopActorExecution
+from leapp import reporting
+
+
+def update_grub_core(grub_dev):
+    """
+    Update GRUB core after upgrade from RHEL7 to RHEL8
+
+    On legacy systems, GRUB core does not get automatically updated when GRUB packages
+    are updated.
+    """
+    try:
+        run(['grub2-install', grub_dev])
+    except CalledProcessError as err:
+        reporting.create_report([
+            reporting.Title('GRUB core update failed'),
+            reporting.Summary(str(err)),
+            reporting.Tags([reporting.Tags.BOOT]),
+            reporting.Severity(reporting.Severity.HIGH),
+            reporting.Remediation(
+                hint='Please run "grub2-install <GRUB_DEVICE>" manually after upgrade'
+            )
+        ])
+        api.current_logger().warning('GRUB core update on {} failed'.format(grub_dev))
+        raise StopActorExecution()
+    reporting.create_report([
+        reporting.Title('GRUB core successfully updated'),
+        reporting.Summary('GRUB core on {} was successfully updated'.format(grub_dev)),
+        reporting.Tags([reporting.Tags.BOOT]),
+        reporting.Severity(reporting.Severity.INFO)
+    ])

repos/system_upgrade/el7toel8/actors/updategrubcore/tests/test_updategrubcore.py

@@ -0,0 +1,59 @@
+import pytest
+
+from leapp.exceptions import StopActorExecution
+from leapp.snactor.fixture import current_actor_context
+from leapp.models import UpdateGrub
+from leapp.reporting import Report
+from leapp import reporting
+from leapp.libraries.common import testutils
+from leapp.libraries.actor import library
+from leapp.libraries.stdlib import CalledProcessError, api
+
+
+UPDATE_OK_TITLE = 'GRUB core successfully updated'
+UPDATE_FAILED_TITLE = 'GRUB core update failed'
+
+
+def raise_call_error(args=None):
+    raise CalledProcessError(
+        message='A Leapp Command Error occured.',
+        command=args,
+        result={'signal': None, 'exit_code': 1, 'pid': 0, 'stdout': 'fake', 'stderr': 'fake'}
+    )
+
+
+class run_mocked(object):
+    def __init__(self, raise_err=False):
+        self.called = 0
+        self.args = []
+        self.raise_err = raise_err
+
+    def __call__(self, *args):
+        self.called += 1
+        self.args.append(args)
+        if self.raise_err:
+            raise_call_error(args)
+
+
+def test_update_grub(monkeypatch):
+    monkeypatch.setattr(api, 'consume', lambda x: iter([UpdateGrub(grub_device='/dev/vda')]))
+    monkeypatch.setattr(reporting, "create_report", testutils.create_report_mocked())
+    monkeypatch.setattr(library, 'run', run_mocked())
+    library.update_grub_core('/dev/vda')
+    assert reporting.create_report.called
+    assert UPDATE_OK_TITLE == reporting.create_report.report_fields['title']
+
+
+def test_update_grub_failed(monkeypatch):
+    monkeypatch.setattr(api, 'consume', lambda x: iter([UpdateGrub(grub_device='/dev/vda')]))
+    monkeypatch.setattr(reporting, "create_report", testutils.create_report_mocked())
+    monkeypatch.setattr(library, 'run', run_mocked(raise_err=True))
+    with pytest.raises(StopActorExecution):
+        library.update_grub_core('/dev/vda')
+    assert reporting.create_report.called
+    assert UPDATE_FAILED_TITLE == reporting.create_report.report_fields['title']
+
+
+def test_update_grub_negative(current_actor_context):
+    current_actor_context.run()
+    assert not current_actor_context.consume(Report)