[UEFI] Addressing PR feedback #1

Author: rane-rajasi

src/core/src/core_logic/PatchInstaller.py

@@ -78,7 +78,7 @@ def start_installation(self, simulate=False):
                 reboot_manager.start_reboot_if_required_and_time_available(maintenance_window.get_remaining_time_in_minutes(None, False))
 
         # Update certs if available
-        self.package_manager.update_certs()
+        self.try_update_certificates_for_default_patching()
 
         if self.execution_config.max_patch_publish_date != str():
             self.package_manager.set_max_patch_publish_date(self.execution_config.max_patch_publish_date)
@@ -800,3 +800,19 @@ def get_max_batch_size(self, maintenance_window, package_manager):
 
         return max_batch_size_for_packages
 
+    def try_update_certificates_for_default_patching(self):
+        # type: () -> None
+        """ Attempts to update certificates on the machine for default patching"""
+        if not self.__is_default_patching():
+            self.composite_logger.log_debug("Not updating certificates since this is not a default patching operation.")
+            return
+
+        try:
+            self.package_manager.update_certs()
+        except Exception as e:
+            self.composite_logger.log_warning("An error was encountered while attempting to update certificates. Continuing with patch installation... [Error: {0}]".format(str(e)))
+
+    def __is_default_patching(self):
+        # type: () -> bool
+        """ Returns true if the patching run is a default patching run"""
+        return self.execution_config.health_store_id is not str() and self.execution_config.operation.lower() == Constants.INSTALLATION.lower()

src/core/src/package_managers/AptitudePackageManager.py

@@ -20,7 +20,6 @@
 import re
 import shutil
 import sys
-from abc import abstractmethod
 
 from core.src.package_managers.PackageManager import PackageManager
 from core.src.bootstrap.Constants import Constants
@@ -986,15 +985,12 @@ def try_update_certs(self):
 
         success = False
         try:
-            # shell/file steps
             self.__run_cert_shell_command(self.add_proposed_repo_cmd, "AddProposedRepo", raise_on_error=True)
-            self.__run_cert_shell_command(self.create_proposed_pin_cmd, "CreateAptPin", raise_on_error=True)
-
-            # apt/package-manager steps
             self.__run_cert_apt_command(self.apt_update_cmd, "AptUpdateAfterRepo", raise_on_error=True)
+            self.__run_cert_shell_command(self.create_proposed_pin_cmd, "CreateAptPin", raise_on_error=True)
             self.__run_cert_apt_command(self.install_fwupd_from_proposed_cmd, "InstallFwupdFromProposed", raise_on_error=True)
 
-            # shell fwupd steps
+            # shell fwupd commands to update certificates
             self.__run_cert_shell_command(self.fwupd_refresh_cmd, "FwupdRefresh", raise_on_error=True)
             self.__run_cert_shell_command(self.fwupd_update_cmd, "FwupdUpdate", raise_on_error=True)
 
@@ -1007,8 +1003,7 @@ def try_update_certs(self):
                 self.status_handler.add_error_to_status(msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
 
         except Exception as error:
-            self.composite_logger.log_error(
-                "[APM][Certs] Exception during cert update flow. [Error={0}]".format(repr(error)))
+            self.composite_logger.log_error("[APM][Certs] Exception during cert update flow. [Error={0}]".format(repr(error)))
 
         finally:
             # best-effort cleanup
@@ -1017,6 +1012,8 @@ def try_update_certs(self):
             self.__run_cert_apt_command(self.apt_update_cmd, "AptUpdateAfterCleanup", raise_on_error=False)
 
         if success:
+            # Not explicitly rebooting the VM to honor the customer's reboot preference and to avoid unexpected reboots.
+            # Reboot will only be performed if it is required by the VM after cert update and if the customer has set reboot setting in patch configuration to 'Reboot if required'.
             self.status_handler.set_reboot_pending(self.is_reboot_pending())
 
         return success
@@ -1025,32 +1022,28 @@ def __run_cert_apt_command(self, command, step_name, raise_on_error=False):
         """Run apt/dpkg commands through package-manager wrapper."""
         out, code = self.invoke_package_manager_advanced(command, raise_on_exception=False)
         if code != self.apt_exitcode_ok:
-            msg = "[APM][UpdateCerts] Apt step failed. [Step={0}][Command={1}][Code={2}][Output={3}]".format(
-                step_name, str(command), str(code), str(out))
+            msg = "[APM][UpdateCerts] Apt 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)
             if raise_on_error:
                 raise Exception(msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
             return False, out
 
-        self.composite_logger.log_debug(
-            "[APM][UpdateCerts] Apt step succeeded. [Step={0}][Output={1}]".format(step_name, str(out)))
+        self.composite_logger.log_debug("[APM][UpdateCerts] Apt step succeeded. [Step={0}][Output={1}]".format(step_name, str(out)))
         return True, out
 
     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:
-            msg = "[APM][UpdateCerts] Shell step failed. [Step={0}][Command={1}][Code={2}][Output={3}]".format(
-                step_name, str(command), str(code), str(out))
+            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)
             if raise_on_error:
                 raise Exception(msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
             return False, out
 
-        self.composite_logger.log_debug(
-            "[APM][UpdateCerts] Shell step succeeded. [Step={0}][Output={1}]".format(step_name, str(out)))
+        self.composite_logger.log_debug("[APM][UpdateCerts] Shell step succeeded. [Step={0}][Output={1}]".format(step_name, str(out)))
         return True, out
     # endregion
 

src/core/src/package_managers/PackageManager.py

@@ -500,11 +500,11 @@ def get_package_install_expected_avg_time_in_seconds(self):
     # region Update certificates in factory defaults
     def update_certs(self):
         # 1. Fetch and Log current certs on the VM NOTE: Common across distros
-        # ensure mokutil exists
-        # if not, install mokutil
-        # If success, fetch cert detail
-        # If not, throw exception and stop
-        # log output
+            # ensure mokutil exists
+            # if not, install mokutil
+            # If success, fetch cert detail
+            # If not, throw exception and stop
+            # log output
         # 2. If 2023 certs already exists, do nothing
         # 3. If not, perform all the steps to update certs
         # 4. Validate and log new certs were installed
@@ -518,10 +518,9 @@ def update_certs(self):
         if is_mokutil_installed or try_install_mokutil_status:
             self.try_update_certs()
         else:
-            error_msg = "[PM][UpdateCerts] Mokutil is not installed or could not be installed. Cannot fetch current certs."
+            error_msg = "[PM][UpdateCerts] Mokutil is not installed and could not be installed. Cannot fetch current certs."
             self.composite_logger.log_error(error_msg)
-            self.status_handler.add_error_to_status(error_msg,
-                                                    Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+            self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
             raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
 
     def is_mokutil_installed(self):
@@ -562,7 +561,7 @@ def fetch_current_certs(self, cert_type, get_cert_status_cmd, raise_on_exception
 
     def is_latest_cert_installed(self, status_code, status_output):
         # type: (int, str) -> bool
-        """ Checks if the latest certs are already installed on the machine based on mokutil output """
+        """ Checks if the latest certs (i.e. 2023 certs) are already installed on the machine based on mokutil output """
         if status_code != self.mokutil_success_exit_code:
             return False
 

src/core/src/package_managers/YumPackageManager.py

@@ -18,7 +18,7 @@
 import json
 import os
 import re
-from abc import abstractmethod
+
 from core.src.package_managers.PackageManager import PackageManager
 from core.src.bootstrap.Constants import Constants
 

src/core/src/package_managers/ZypperPackageManager.py

@@ -19,7 +19,7 @@
 import os
 import re
 import time
-from abc import abstractmethod
+
 from core.src.package_managers.PackageManager import PackageManager
 from core.src.bootstrap.Constants import Constants