[UEFI] Handling error code 2 for fwupgmgr refresh

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)

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')