[RHELC-752] Verify kernel boot files after conversion

convert2rhel/checks.py

@@ -1,5 +1,3 @@
-# -*- coding: utf-8 -*-
-#
 # Copyright(C) 2016 Red Hat, Inc.
 #
 # This program is free software: you can redistribute it and/or modify
@@ -75,6 +73,12 @@
     8: "4.18.0",
 }
 
+VMLINUZ_FILEPATH = "/boot/vmlinuz-%s"
+"""The path to the vmlinuz file in a system."""
+
+INITRAMFS_FILEPATH = "/boot/initramfs-%s.img"
+"""The path to the initramfs image in a system."""
+
 # Python 2.6 compatibility.
 # This code is copied from Pthon-3.10's functools module,
 # licensed under the Python Software Foundation License, version 2
@@ -835,3 +839,79 @@ def check_dbus_is_running():
         "Could not find a running DBus Daemon which is needed to register with subscription manager.\n"
         "Please start dbus using `systemctl start dbus` or (on CentOS Linux 6), `service messagebus start`"
     )
+
+
+def _is_initramfs_file_valid(filepath):
+    """Internal function to verify if an initramfs file is corrupted.
+
+    This method will rely on using lsinitrd to do the validation. If the
+    lsinitrd returns other value that is not 0, then it means that the file is
+    probably corrupted or may cause problems during the next reboot.
+
+    :param filepath: The path to the initramfs file.
+    :type filepath: str
+    :return: A boolean to determine if the file is corrupted.
+    :rtype: bool
+    """
+    logger.info("Checking if the '%s' file is valid.", filepath)
+
+    if not os.path.exists(filepath):
+        logger.info("The initramfs file is not present.")
+        return False
+
+    logger.debug("Checking if the '%s' file is not corrupted.", filepath)
+    out, return_code = run_subprocess(
+        cmd=["/usr/bin/lsinitrd", filepath],
+        print_output=False,
+    )
+
+    if return_code != 0:
+        logger.info("Couldn't verify initramfs file. It may be corrupted.")
+        logger.debug("Output of lsinitrd: %s", out)
+        return False
+
+    return True
+
+
+def check_kernel_boot_files():
+    """Check if the required kernel files exist and are valid under the boot partition."""
+    # For Oracle/CentOS Linux 8 the `kernel` is just a meta package, instead,
+    # we check for `kernel-core`. This is not true regarding the 7.* releases.
+    kernel_name = "kernel-core" if system_info.version.major >= 8 else "kernel"
+
+    # Either the package is returned or not. The return_code will be 0 in
+    # either case, so we don't care about checking for that here.
+    output, _ = run_subprocess(["rpm", "-q", "--last", kernel_name], print_output=False)
+
+    # We are parsing the latest kernel installed on the system, which at this
+    # point, should be a RHEL kernel. Since we can't get the kernel version
+    # from `uname -r`, as it requires a reboot in order to take place, we are
+    # detecting the latest kernel by using `rpm` and figuring out which was the
+    # latest kernel installed.
+    latest_installed_kernel = output.split("\n")[0].split(" ")[0]
+    latest_installed_kernel = latest_installed_kernel[len(kernel_name + "-") :]
+    grub2_config_file = grub.get_grub_config_file()
+    initramfs_file = INITRAMFS_FILEPATH % latest_installed_kernel
+    vmlinuz_file = VMLINUZ_FILEPATH % latest_installed_kernel
+
+    logger.info("Checking if the '%s' file exists.", vmlinuz_file)
+    vmlinuz_exists = os.path.exists(vmlinuz_file)
+    if not vmlinuz_exists:
+        logger.info("The vmlinuz file is not present.")
+
+    is_initramfs_valid = _is_initramfs_file_valid(initramfs_file)
+
+    if not is_initramfs_valid or not vmlinuz_exists:
+        logger.warning(
+            "Couldn't verify the kernel boot files in the boot partition. This may cause problems during the next boot "
+            "of your system.\nIn order to fix this problem you may need to free/increase space in your boot partition"
+            " and then run the following commands in your terminal:\n"
+            "1. yum reinstall %s-%s -y\n"
+            "2. grub2-mkconfig -o %s\n"
+            "3. reboot",
+            kernel_name,
+            latest_installed_kernel,
+            grub2_config_file,
+        )
+    else:
+        logger.info("The initramfs and vmlinuz files are valid.")

convert2rhel/grub.py

@@ -591,6 +591,23 @@ def post_ponr_set_efi_configuration():
         _log_critical_error(e.message)
 
 
+def get_grub_config_file():
+    """Get the grub config file path.
+
+    This method will return the grub config file, depending if it is BIOS or
+    UEFI, the method will handle that automatically.
+
+    :return: The path to the grub config file.
+    :rtype: str
+    """
+    grub_config_path = GRUB2_BIOS_CONFIG_FILE
+
+    if is_efi():
+        grub_config_path = os.path.join(RHEL_EFIDIR_CANONICAL_PATH, "grub.cfg")
+
+    return grub_config_path
+
+
 def update_grub_after_conversion():
     """Update GRUB2 images and config after conversion.
 
@@ -611,7 +628,7 @@ def update_grub_after_conversion():
     backup.RestorableFile(GRUB2_BIOS_CONFIG_FILE).backup()
     backup.RestorableFile(GRUB2_BIOS_ENV_FILE).backup()
 
-    grub2_config_file = GRUB2_BIOS_CONFIG_FILE if not is_efi() else os.path.join(RHEL_EFIDIR_CANONICAL_PATH, "grub.cfg")
+    grub2_config_file = get_grub_config_file()
 
     output, ret_code = utils.run_subprocess(["/usr/sbin/grub2-mkconfig", "-o", grub2_config_file], print_output=False)
     logger.debug("Output of the grub2-mkconfig call:\n%s" % output)

convert2rhel/main.py

@@ -132,6 +132,9 @@ def main():
         loggerinst.task("Final: Remove temporary folder %s" % utils.TMP_DIR)
         utils.remove_tmp_dir()
 
+        loggerinst.task("Final: Check kernel boot files")
+        checks.check_kernel_boot_files()
+
         breadcrumbs.breadcrumbs.finish_collection(success=True)
 
         loggerinst.task("Final: Update RHSM custom facts")

convert2rhel/unit_tests/__init__.py

@@ -273,7 +273,6 @@ def run_subprocess_side_effect(*stubs):
     >>>         (("repoquery", "-l"), (REPOQUERY_L_STUB_GOOD, 0)),
     >>>     )
     >>> )
-
     """
 
     def factory(*args, **kwargs):

convert2rhel/unit_tests/checks_test.py

@@ -1612,3 +1612,144 @@ def test_check_dbus_is_running_not_running(caplog, monkeypatch, global_tool_opts
     )
     assert log_msg == caplog.records[-1].message
     assert caplog.records[-1].levelname == "CRITICAL"
+
+
+@pytest.mark.parametrize(
+    ("latest_installed_kernel", "subprocess_output", "expected"),
+    (
+        ("6.1.7-200.fc37.x86_64", ("test", 0), True),
+        ("6.1.7-200.fc37.x86_64", ("error", 1), False),
+    ),
+)
+def test_is_initramfs_file_valid(latest_installed_kernel, subprocess_output, expected, tmpdir, caplog, monkeypatch):
+    initramfs_file = tmpdir.mkdir("/boot").join("initramfs-%s.img")
+    initramfs_file = str(initramfs_file)
+    initramfs_file = initramfs_file % latest_installed_kernel
+    with open(initramfs_file, mode="w") as _:
+        pass
+
+    monkeypatch.setattr(checks, "INITRAMFS_FILEPATH", initramfs_file)
+    monkeypatch.setattr(checks, "run_subprocess", mock.Mock(return_value=subprocess_output))
+    result = checks._is_initramfs_file_valid(initramfs_file)
+    assert result == expected
+
+    if not expected:
+        assert "Couldn't verify initramfs file. It may be corrupted." in caplog.records[-2].message
+        assert "Output of lsinitrd: %s" % subprocess_output[0] in caplog.records[-1].message
+
+
+@centos8
+def test_check_kernel_boot_files(pretend_os, tmpdir, caplog, monkeypatch):
+    rpm_last_kernel_output = ("kernel-core-6.1.8-200.fc37.x86_64 Wed 01 Feb 2023 14:01:01 -03", 0)
+    latest_installed_kernel = "6.1.8-200.fc37.x86_64"
+
+    boot_folder = tmpdir.mkdir("/boot")
+    initramfs_file = boot_folder.join("initramfs-%s.img")
+    vmlinuz_file = boot_folder.join("vmlinuz-%s")
+    initramfs_file = str(initramfs_file)
+    vmlinuz_file = str(vmlinuz_file)
+
+    with open(initramfs_file % latest_installed_kernel, mode="w") as _:
+        pass
+
+    with open(vmlinuz_file % latest_installed_kernel, mode="w") as _:
+        pass
+
+    monkeypatch.setattr(checks, "VMLINUZ_FILEPATH", vmlinuz_file)
+    monkeypatch.setattr(checks, "INITRAMFS_FILEPATH", initramfs_file)
+    monkeypatch.setattr(checks, "run_subprocess", mock.Mock(side_effect=[rpm_last_kernel_output, ("test", 0)]))
+
+    checks.check_kernel_boot_files()
+    assert "The initramfs and vmlinuz files are valid." in caplog.records[-1].message
+
+
+@pytest.mark.parametrize(
+    ("create_initramfs", "create_vmlinuz", "run_piped_subprocess", "rpm_last_kernel_output", "latest_installed_kernel"),
+    (
+        pytest.param(
+            False,
+            False,
+            ("", 0),
+            ("kernel-core-6.1.8-200.fc37.x86_64 Wed 01 Feb 2023 14:01:01 -03", 0),
+            "6.1.8-200.fc37.x86_64",
+            id="both-files-missing",
+        ),
+        pytest.param(
+            True,
+            False,
+            ("test", 0),
+            ("kernel-core-6.1.8-200.fc37.x86_64 Wed 01 Feb 2023 14:01:01 -03", 0),
+            "6.1.8-200.fc37.x86_64",
+            id="vmlinuz-missing",
+        ),
+        pytest.param(
+            False,
+            True,
+            ("test", 0),
+            ("kernel-core-6.1.8-200.fc37.x86_64 Wed 01 Feb 2023 14:01:01 -03", 0),
+            "6.1.8-200.fc37.x86_64",
+            id="initramfs-missing",
+        ),
+        pytest.param(
+            True,
+            True,
+            ("error", 1),
+            ("kernel-core-6.1.8-200.fc37.x86_64 Wed 01 Feb 2023 14:01:01 -03", 0),
+            "6.1.8-200.fc37.x86_64",
+            id="initramfs-corrupted",
+        ),
+    ),
+)
+@centos8
+def test_check_kernel_boot_files_missing(
+    pretend_os,
+    create_initramfs,
+    create_vmlinuz,
+    run_piped_subprocess,
+    rpm_last_kernel_output,
+    latest_installed_kernel,
+    tmpdir,
+    caplog,
+    monkeypatch,
+):
+    """
+    This test will check if we output the warning message correctly if either
+    initramfs or vmlinuz are missing.
+    """
+    # We are mocking both subprocess calls here in order to make it easier for
+    # testing any type of parametrization we may include in the future. Note
+    # that the second iteration may not run sometimes, as this is specific for
+    # when we want to check if a file is corrupted or not.
+    monkeypatch.setattr(
+        checks,
+        "run_subprocess",
+        mock.Mock(
+            side_effect=[
+                rpm_last_kernel_output,
+                run_piped_subprocess,
+            ]
+        ),
+    )
+    boot_folder = tmpdir.mkdir("/boot")
+    if create_initramfs:
+        initramfs_file = boot_folder.join("initramfs-%s.img")
+        initramfs_file = str(initramfs_file)
+        with open(initramfs_file % latest_installed_kernel, mode="w") as _:
+            pass
+
+        monkeypatch.setattr(checks, "INITRAMFS_FILEPATH", initramfs_file)
+    else:
+        monkeypatch.setattr(checks, "INITRAMFS_FILEPATH", "/non-existing-%s.img")
+
+    if create_vmlinuz:
+        vmlinuz_file = boot_folder.join("vmlinuz-%s")
+        vmlinuz_file = str(vmlinuz_file)
+        with open(vmlinuz_file % latest_installed_kernel, mode="w") as _:
+            pass
+
+        monkeypatch.setattr(checks, "VMLINUZ_FILEPATH", vmlinuz_file)
+    else:
+        monkeypatch.setattr(checks, "VMLINUZ_FILEPATH", "/non-existing-%s")
+
+    checks.check_kernel_boot_files()
+    assert "Couldn't verify the kernel boot files in the boot partition." in caplog.records[-1].message

convert2rhel/unit_tests/grub_test.py

@@ -602,3 +602,17 @@ def test_update_grub_after_conversion(
     grub.update_grub_after_conversion()
     if expected is not None:
         assert expected in caplog.records[-1].message
+
+
+@pytest.mark.parametrize(
+    ("is_efi", "config_path"),
+    (
+        (False, "/boot/grub2/grub.cfg"),
+        (True, "/boot/efi/EFI/redhat/grub.cfg"),
+    ),
+)
+def test_get_grub_config_file(is_efi, config_path, monkeypatch):
+    monkeypatch.setattr("convert2rhel.grub.is_efi", mock.Mock(return_value=is_efi))
+    config_file = grub.get_grub_config_file()
+
+    assert config_file == config_path

convert2rhel/unit_tests/main_test.py

@@ -395,6 +395,7 @@ def test_main(monkeypatch):
     remove_tmp_dir_mock = mock.Mock()
     restart_system_mock = mock.Mock()
     finish_collection_mock = mock.Mock()
+    check_kernel_boot_files_mock = mock.Mock()
     update_rhsm_custom_facts_mock = mock.Mock()
 
     monkeypatch.setattr(utils, "require_root", require_root_mock)
@@ -418,6 +419,7 @@ def test_main(monkeypatch):
     monkeypatch.setattr(utils, "remove_tmp_dir", remove_tmp_dir_mock)
     monkeypatch.setattr(utils, "restart_system", restart_system_mock)
     monkeypatch.setattr(breadcrumbs, "finish_collection", finish_collection_mock)
+    monkeypatch.setattr(checks, "check_kernel_boot_files", check_kernel_boot_files_mock)
     monkeypatch.setattr(subscription, "update_rhsm_custom_facts", update_rhsm_custom_facts_mock)
 
     assert main.main() == 0
@@ -440,6 +442,7 @@ def test_main(monkeypatch):
     assert remove_tmp_dir_mock.call_count == 1
     assert restart_system_mock.call_count == 1
     assert finish_collection_mock.call_count == 1
+    assert check_kernel_boot_files_mock.call_count == 1
     assert update_rhsm_custom_facts_mock.call_count == 1