diff --git a/src/core/src/package_managers/AptitudePackageManager.py b/src/core/src/package_managers/AptitudePackageManager.py index 35af2147..cb591daa 100644 --- a/src/core/src/package_managers/AptitudePackageManager.py +++ b/src/core/src/package_managers/AptitudePackageManager.py @@ -97,10 +97,10 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ self.install_mokutil_cmd = "sudo apt-get install -y -qq mokutil" self.apt_update_cmd = "sudo apt-get -q update" self.min_fwupd_version = "2.0.8" # Refer public docs: https://github.com/fwupd/fwupd/releases/tag/2.0.8 and https://discourse.ubuntu.com/t/microsoft-uefi-ca-rotation-what-it-means-for-ubuntu-users-and-vendors/82652 - self.get_installed_fwupd_version_cmd = "fwupdmgr --version" self.remove_fwupd_cmd = "sudo apt-get purge -y fwupd" + self.get_installed_fwupd_version_cmd = "fwupdmgr --version" self.install_fwupd_cmd = "sudo apt-get install -y fwupd" - self.fwupd_refresh_cmd = "sudo fwupdmgr refresh" # NOTE: This could be made generic in package manager, depending on what solution type is adopted for other distros + self.fwupd_force_refresh_cmd = "sudo fwupdmgr refresh --force" # NOTE: This could be made generic in package manager, depending on what solution type is adopted for other distros self.fwupd_update_cmd = "sudo fwupdmgr update -y" self.get_uptime_seconds_cmd = "cat /proc/uptime" self.get_hibernation_state_cmd = "cat /sys/power/disk" @@ -1032,7 +1032,7 @@ def try_update_certs(self): self.__ensure_fwupd_installation() # shell fwupd commands to update certificates - self.__run_cert_shell_command(self.fwupd_refresh_cmd, step_name="FwupdRefresh", raise_on_error=True) + self.__run_cert_shell_command(self.fwupd_force_refresh_cmd, step_name="FwupdRefresh", raise_on_error=False) self.__run_cert_shell_command(self.fwupd_update_cmd, step_name="FwupdUpdate", raise_on_error=True) """ NOTE: They tooling used to update here is fwupd (firmware update manager). In this method of updating certs, the exact version of current certs is never pinned @@ -1139,7 +1139,11 @@ def __run_cert_apt_command(self, command, step_name, raise_on_error=False): def __run_cert_shell_command(self, command, step_name, raise_on_error=False): """Run non-apt utility commands directly.""" code, out = self.env_layer.run_command_output(command, False, False) - if code != 0: + if command == self.fwupd_force_refresh_cmd: + # As per the suggestion from trusted launch VM partners, we need to ignore all errors returned by force refresh command and continue to the next step. + self.composite_logger.log_debug("[APM][UpdateCerts] Force refresh shell step executed. [Step={0}][Command={1}][Code={2}][Output={3}]".format(step_name, str(command), str(code), str(out))) + return True, out + elif code != 0: msg = "[APM][UpdateCerts] Shell step failed. [Step={0}][Command={1}][Code={2}][Output={3}]".format(step_name, str(command), str(code), str(out)) self.composite_logger.log_error(msg) self.status_handler.add_error_to_status(msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE) diff --git a/src/core/tests/Test_AptitudePackageManager.py b/src/core/tests/Test_AptitudePackageManager.py index 71b414f3..a0956461 100644 --- a/src/core/tests/Test_AptitudePackageManager.py +++ b/src/core/tests/Test_AptitudePackageManager.py @@ -135,9 +135,23 @@ def mock_run_command_output_fwupd_version_not_found_in_first_attempt_and_found_l return 0, "" def mock_run_command_output_fwupd_refresh_fails(self, cmd, no_output=False, chk_err=True): - self.assertGreater(cmd.find(self.runtime.package_manager.fwupd_refresh_cmd), -1) + self.assertGreater(cmd.find(self.runtime.package_manager.fwupd_force_refresh_cmd), -1) return 1, "Error" + def mock_run_command_output_with_fwupd_exit_code_overrides(self, cmd, no_output=False, chk_err=True): + """Reuse the legacy env layer command mocks and override only the fwupd refresh/update exit codes driven by test state.""" + code, output = self.legacy_run_command_output(cmd, no_output, chk_err) + + if cmd.find(self.runtime.package_manager.fwupd_force_refresh_cmd) > -1: + if self.fwupd_refresh_code == 2: + return self.fwupd_refresh_code, "Nothing to do (no updates available)" + return self.fwupd_refresh_code, output if self.fwupd_refresh_code == 0 else "Error" + + if cmd.find(self.runtime.package_manager.fwupd_update_cmd) > -1: + return self.fwupd_update_code, output if self.fwupd_update_code == 0 else "Error" + + return code, output + def mock_run_command_output_apt_update_cmd_fails(self, cmd, no_output=False, chk_err=True): if cmd.find(self.runtime.package_manager.apt_update_cmd) > -1: self.latest_apt_update_cmd_attempt += 1 @@ -1564,6 +1578,80 @@ def test_try_update_certs_command_failure_paths__with_various_use_cases(self): finally: package_manager.env_layer.run_command_output = backup_run_command_output + def test_try_update_certs_fwupd_exit_code_handling__with_various_use_cases(self): + """fwupd refresh command ignores all error codes and succeeds regardless. Other fwupd commands still fail on non-zero exit codes.""" + package_manager = self.container.get('package_manager') + + backup_run_command_output = package_manager.env_layer.run_command_output + backup_add_error_to_status = package_manager.status_handler.add_error_to_status + backup_are_latest_certs_present_with_mokutil_check = package_manager.are_latest_certs_present_with_mokutil_check + backup_is_reboot_pending = package_manager.is_reboot_pending + + use_cases = [ + { + "name": "fwupd_refresh_returns_code_2_is_treated_as_success", + "fwupd_refresh_code": 2, + "fwupd_update_code": 0, + "expected_result": True, + "expected_error_count": 0 + }, + { + "name": "fwupd_refresh_returns_code_1_is_ignored_and_treated_as_success", + "fwupd_refresh_code": 1, + "fwupd_update_code": 0, + "expected_result": True, + "expected_error_count": 0 + }, + { + "name": "fwupd_refresh_returns_code_3_is_ignored_and_treated_as_success", + "fwupd_refresh_code": 3, + "fwupd_update_code": 0, + "expected_result": True, + "expected_error_count": 0 + }, + { + "name": "fwupd_update_returns_code_2_is_treated_as_failure", + "fwupd_refresh_code": 0, + "fwupd_update_code": 2, + "expected_result": False, + "expected_error_count": 1 + }, + { + "name": "all_fwupd_commands_succeed", + "fwupd_refresh_code": 0, + "fwupd_update_code": 0, + "expected_result": True, + "expected_error_count": 0 + } + ] + + try: + # reuse the legacy env layer command mocks (fwupdmgr --version, refresh, update, apt-get -q update) and only override exit codes under test + self.legacy_run_command_output = backup_run_command_output + package_manager.env_layer.run_command_output = self.mock_run_command_output_with_fwupd_exit_code_overrides + package_manager.status_handler.add_error_to_status = self.mock_add_error_to_status_to_capture_errors + package_manager.are_latest_certs_present_with_mokutil_check = self.mock_are_latest_certs_present_return_true + package_manager.is_reboot_pending = self.mock_is_reboot_pending_returns_bool_False + + for use_case in use_cases: + self.captured_errors_tracking = [] + self.fwupd_refresh_code = use_case["fwupd_refresh_code"] + self.fwupd_update_code = use_case["fwupd_update_code"] + + self.assertEqual(package_manager.try_update_certs(), use_case["expected_result"], "Failed use case: {0}".format(use_case["name"])) + self.assertEqual(len(self.captured_errors_tracking), use_case["expected_error_count"], "Failed use case: {0}".format(use_case["name"])) + + if use_case["expected_error_count"] > 0: + self.assertIn("Shell step failed", self.captured_errors_tracking[0][0], "Failed use case: {0}".format(use_case["name"])) + self.assertEqual(self.captured_errors_tracking[0][1], Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE, "Failed use case: {0}".format(use_case["name"])) + else: + self.assertEqual(package_manager.status_handler.is_reboot_pending, False, "Failed use case: {0}".format(use_case["name"])) + finally: + package_manager.env_layer.run_command_output = backup_run_command_output + package_manager.status_handler.add_error_to_status = backup_add_error_to_status + package_manager.are_latest_certs_present_with_mokutil_check = backup_are_latest_certs_present_with_mokutil_check + package_manager.is_reboot_pending = backup_is_reboot_pending + def test_is_hibernation_enabled_for_cert_update__with_various_use_cases(self): """Test is_hibernation_enabled_for_cert_update with various hibernation states""" package_manager = self.container.get('package_manager')