Feature: [Ubuntu, no-CVM] Adding reboot, hibernate checks and aligning all pre-reqs together

src/core/src/core_logic/PatchInstaller.py

@@ -808,24 +808,139 @@ def try_update_certificates_for_default_patching(self):
             self.composite_logger.log_debug("Not updating certificates since this is not a default patching operation.")
             return
 
+        if self.__are_prerequisites_for_updating_certs_met():
+            try:
+                certs_updated = self.package_manager.update_certs()
+                if not certs_updated:
+                    error_msg = "UEFI certificate update did not complete successfully. Certificates may not have been updated."
+                    self.composite_logger.log_error(error_msg)
+                    self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+            except Exception as e:
+                error_msg = "An error was encountered while attempting to update certificates. Continuing with patch installation... [Error: {0}]".format(str(e))
+                self.composite_logger.log_error(error_msg)
+                self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+
+    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 None and
+                self.execution_config.health_store_id != "" and
+                self.execution_config.operation.lower() == Constants.INSTALLATION.lower())
+
+    def __are_prerequisites_for_updating_certs_met(self):
+        # type: () -> bool
+        """ Returns true if all the pre-requisites for updating certs are met. These pre-reqs are:
+        1. Should not be a CVM
+        2. Hibernation should be turned off
+        3. Should not already contain latest certificates
+        4. System uptime should not be more than 7 days, if it is, we issue a reboot if permitted"""
+
+        if self.__is_cvm():
+            return False
+
+        if not self.can_continue_cert_update_after_hibernation_check():
+            return False
+
+        if not self.can_continue_cert_update_after_latest_cert_check():
+            return False
+
+        if not self.can_continue_cert_update_after_reboot_check():
+            return False
+
+        return True
+
+    def __is_cvm(self):
+        # type: () -> bool
+        """ Returns true if this is a Confidential VM (CVM) """
         try:
             is_confidential_vm, detection_details = self.env_layer.detect_confidential_vm()
         except Exception as e:
             self.composite_logger.log_warning("Unable to determine whether the VM is a Confidential VM before attempting the UEFI certificate update. Continuing with patch installation... [Error: {0}]".format(str(e)))
-            return
+            return True
 
         if is_confidential_vm:
             self.composite_logger.log("Skipping UEFI certificate update because this VM was detected as a Confidential VM. [Detection={0}]".format(detection_details))
-            return
 
+        return is_confidential_vm
+
+    def can_continue_cert_update_after_reboot_check(self):
+        # type: () -> bool
+        """Attempts pre-cert reboot when required.
+
+        Returns True only when cert update can continue in the current cycle.
+        Returns False when reboot is required-but-blocked/failed, or when reboot was initiated
+        and cert update must wait for the post-reboot cycle.
+        """
+
+        should_reboot_before_cert_update = self.package_manager.should_reboot_before_cert_update()
+        if should_reboot_before_cert_update:
+            if self.reboot_manager.is_setting(Constants.REBOOT_NEVER):
+                error_msg = "UEFI certificate update requires a reboot first (VM uptime exceeded 7 days), but reboot setting is 'Never'. Skipping certificate update."
+                self.composite_logger.log_error(error_msg)
+                self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+                return False
+
+            reboot_started = False
+            original_force_reboot = self.package_manager.force_reboot
+            self.package_manager.force_reboot = True
+            try:
+                remaining_time = self.maintenance_window.get_remaining_time_in_minutes(None, False)
+                reboot_started = self.reboot_manager.start_reboot_if_required_and_time_available(remaining_time)
+            except Exception as e:
+                error_msg = "Unable to reboot before UEFI certificate update. Skipping certificate update. [Error: {0}]".format(str(e))
+                self.composite_logger.log_warning(error_msg)
+                self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+                return False
+            finally:
+                self.package_manager.force_reboot = original_force_reboot
+
+            if not reboot_started:
+                error_msg = "UEFI certificate update requires a reboot first (VM uptime exceeded 7 days), but reboot could not be initiated. Skipping certificate update."
+                self.composite_logger.log_warning(error_msg)
+                self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+                return False
+
+            # If reboot was started, this process is expected to exit; avoid running cert update in the same cycle.
+            return False
+
+        return True
+
+    def can_continue_cert_update_after_hibernation_check(self):
+        # type: () -> bool
+        """Attempts hibernation pre-cert check.
+
+        Returns True when cert update can continue in the current cycle.
+        Returns False when hibernation is enabled and cert update must be skipped or hibernate state cannot be determined.
+        """
         try:
-            self.package_manager.update_certs()
+            is_hibernation_enabled = self.package_manager.is_hibernation_enabled_for_cert_update()
         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)))
+            error_msg = "Unable to determine hibernation state before UEFI certificate update. Not continuing with certificate update. [Error: {0}]".format(str(e))
+            self.composite_logger.log_warning(error_msg)
+            self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+            return False
 
-    def __is_default_patching(self):
+        if is_hibernation_enabled:
+            error_msg = "UEFI certificate update requires hibernation to be turned off for the duration of the update. Turn off hibernation and re-run patching. Skipping certificate update."
+            self.composite_logger.log_warning(error_msg)
+            self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+            return False
+
+        return True
+
+    def can_continue_cert_update_after_latest_cert_check(self):
         # type: () -> bool
-        """ Returns true if the patching run is a default patching run"""
-        return (self.execution_config.health_store_id is not None and
-                self.execution_config.health_store_id != "" and
-                self.execution_config.operation.lower() == Constants.INSTALLATION.lower())
+        """Returns False when latest certs are already installed and cert update can be skipped."""
+        try:
+            latest_certs_present = self.package_manager.are_latest_certs_present()
+        except Exception as e:
+            self.composite_logger.log_warning("Unable to determine whether latest certs are already installed before UEFI certificate update. Not continuing with certificate update. [Error: {0}]".format(str(e)))
+            return False
+
+        if latest_certs_present:
+            self.composite_logger.log("Latest UEFI certificates are already present. Skipping certificate update.")
+            return False
+
+        return True
+
+

src/core/src/package_managers/AptitudePackageManager.py

@@ -102,6 +102,9 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         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_update_cmd = "sudo fwupdmgr update -y"
+        self.get_uptime_seconds_cmd = "cat /proc/uptime"
+        self.get_hibernation_state_cmd = "cat /sys/power/disk"
+        self.pre_cert_reboot_uptime_threshold_seconds = 7 * 24 * 60 * 60
 
     # region Sources Management
     def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_classification=str()):
@@ -957,6 +960,57 @@ def get_package_install_expected_avg_time_in_seconds(self):
         return self.package_install_expected_avg_time_in_seconds
 
     # region Update certificates in factory defaults
+    def should_reboot_before_cert_update(self):
+        # type: () -> bool
+        """Return True when certificate update should be preceded by a reboot."""
+        uptime_seconds = self.__get_vm_uptime_seconds()
+        if uptime_seconds is None:
+            # Best-effort check: do not block cert flow if uptime cannot be determined.
+            return False
+
+        requires_reboot = uptime_seconds > self.pre_cert_reboot_uptime_threshold_seconds
+        self.composite_logger.log_debug("[APM][Certs] Pre-cert reboot check. [UptimeInSeconds={0}][ThresholdInSeconds={1}][RequiresReboot={2}]"
+                                        .format(str(uptime_seconds), str(self.pre_cert_reboot_uptime_threshold_seconds), str(requires_reboot)))
+        return requires_reboot
+
+    def is_hibernation_enabled_for_cert_update(self):
+        # type: () -> bool
+        """Returns True when hibernation appears enabled based on /sys/power/disk state."""
+        cmd = self.get_hibernation_state_cmd
+        self.composite_logger.log_verbose('[APM][Certs] Checking hibernation state before cert update. [Command={0}]'.format(cmd))
+        code, out = self.env_layer.run_command_output(cmd, False, False)
+        out = str(out) if out is not None else str()
+
+        if code != self.apt_exitcode_ok:
+            self.composite_logger.log_debug('[APM][Certs] Unable to determine hibernation state for cert update. Assuming disabled. [Code={0}][Output={1}]'.format(str(code), out))
+            return False
+
+        # Linux shows current mode in brackets; [disabled] means hibernation is turned off.
+        is_hibernation_enabled = '[disabled]' not in out
+        self.composite_logger.log_debug('[APM][Certs] Hibernation state for cert update. [Enabled={0}][RawState={1}]'.format(str(is_hibernation_enabled), out.strip()))
+        return is_hibernation_enabled
+
+    def __get_vm_uptime_seconds(self):
+        # type: () -> any
+        """Return VM uptime in seconds, or None when unavailable."""
+        cmd = self.get_uptime_seconds_cmd
+        code, out = self.env_layer.run_command_output(cmd, False, False)
+        self.composite_logger.log_debug("[APM][Certs] VM uptime command executed. [Command={0}][Code={1}][Output={2}]".format(str(cmd), str(code), str(out)))
+        if code != self.apt_exitcode_ok:
+            self.composite_logger.log_warning("[APM][Certs] Unable to determine VM uptime. [Command={0}][Code={1}][Output={2}]".format(str(cmd), str(code), str(out)))
+            return None
+
+        output = str(out).strip()
+        if output == str() or len(output.split()) == 0:
+            self.composite_logger.log_warning("[APM][Certs] VM uptime command returned an empty result.")
+            return None
+
+        try:
+            return int(float(output.split()[0]))
+        except Exception as error:
+            self.composite_logger.log_warning("[APM][Certs] Failed to parse VM uptime output. [Output={0}][Error={1}]".format(str(output), repr(error)))
+            return None
+
     def try_install_mokutil(self):
         # type: () -> bool
         """ Attempts to install mokutil """
@@ -969,9 +1023,6 @@ def try_install_mokutil(self):
     def try_update_certs(self):
         """ Attempts to update certificate status """
         self.composite_logger.log("[APM][Certs] Starting cert update flow.")
-        if self.are_latest_certs_present():
-            self.composite_logger.log("[APM][Certs] Latest certs already present. Skipping.")
-            return True
 
         success = False
         try:

src/core/src/package_managers/Dnf5PackageManager.py

@@ -718,4 +718,12 @@ def try_install_mokutil(self):
     def try_update_certs(self):
         """ Attempts to update certificate status """
         pass
+
+    def is_hibernation_enabled_for_cert_update(self):
+        """Checks whether hibernation is enabled"""
+        return False
+
+    def should_reboot_before_cert_update(self):
+        """ Checks if a reboot is required before updating certificates """
+        return False
     # endregion

src/core/src/package_managers/PackageManager.py

@@ -15,7 +15,6 @@
 # Requires Python 2.7+
 
 """The is base package manager, which defines the package management relevant operations"""
-import json
 import os
 from abc import ABCMeta, abstractmethod
 from core.src.bootstrap.Constants import Constants
@@ -498,31 +497,27 @@ def get_package_install_expected_avg_time_in_seconds(self):
         pass
 
     # region Update certificates in factory defaults
-    # TODO (Before removing in-VM feature flag): This update should not be done in CVMs in current form
     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
-        # 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
-
+        # type: () -> bool
+        """ Updates the certificates if the required ones do not exist. Returns True if certs were updated successfully, False otherwise. """
         self.composite_logger.log_verbose("[PM] Updating current certificates if needed...")
-        is_mokutil_installed = self.is_mokutil_installed()
-        try_install_mokutil_status = False
-        if not is_mokutil_installed:
-            try_install_mokutil_status = self.try_install_mokutil()
+        return self.try_update_certs()
 
-        if is_mokutil_installed or try_install_mokutil_status:
-            self.try_update_certs()
-        else:
-            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)
-            raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
+    def ensure_mokutil_available_for_cert_checks(self):
+        # type: (bool) -> bool
+        """Ensures mokutil is available before certificate status checks."""
+        if self.is_mokutil_installed():
+            return True
+
+        self.composite_logger.log_verbose("[PM][Certs] Mokutil is not installed. Attempting installation before cert check.")
+        try_install_mokutil_status = self.try_install_mokutil()
+        if try_install_mokutil_status:
+            return True
+
+        error_msg = "[PM][UpdateCerts] Mokutil is not installed and could not be installed. Cannot fetch current certs."
+        self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.CERTIFICATE_UPDATE)
+        self.composite_logger.log_error(error_msg)
+        raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
 
     def is_mokutil_installed(self):
         # type: () -> bool
@@ -534,6 +529,9 @@ def is_mokutil_installed(self):
         return code == 0
 
     def are_latest_certs_present(self):
+        if not self.ensure_mokutil_available_for_cert_checks():
+            return False
+
         kek_code, kek_out = self.fetch_current_certs(Constants.Certificates.KEK, self.get_kek_cert_status_cmd, raise_on_exception=False)
         db_code, db_out = self.fetch_current_certs(Constants.Certificates.DB, self.get_db_cert_status_cmd, raise_on_exception=False)
         return self.is_latest_cert_installed(kek_code, kek_out) and self.is_latest_cert_installed(db_code, db_out)
@@ -577,5 +575,16 @@ def try_install_mokutil(self):
     def try_update_certs(self):
         """ Attempts to update certificate status """
         pass
+
+    @abstractmethod
+    def is_hibernation_enabled_for_cert_update(self):
+        # type: () -> bool
+        """Checks whether hibernation is enabled."""
+        pass
+
+    @abstractmethod
+    def should_reboot_before_cert_update(self):
+        """ Checks if a reboot is required before updating certificates """
+        pass
     # endregion
 

src/core/src/package_managers/TdnfPackageManager.py

@@ -773,5 +773,13 @@ def try_install_mokutil(self):
     def try_update_certs(self):
         """ Attempts to update certificate status """
         pass
+
+    def is_hibernation_enabled_for_cert_update(self):
+        """Checks whether hibernation is enabled."""
+        return False
+
+    def should_reboot_before_cert_update(self):
+        """ Checks if a reboot is required before updating certificates """
+        return False
     # endregion
 

src/core/src/package_managers/YumPackageManager.py

@@ -1130,5 +1130,13 @@ def try_install_mokutil(self):
     def try_update_certs(self):
         """ Attempts to update certificate status """
         pass
+
+    def is_hibernation_enabled_for_cert_update(self):
+        """Checks whether hibernation is enabled."""
+        return False
+
+    def should_reboot_before_cert_update(self):
+        """ Checks if a reboot is required before updating certificates """
+        return False
     # endregion
 

src/core/src/package_managers/ZypperPackageManager.py

@@ -882,5 +882,13 @@ def try_install_mokutil(self):
     def try_update_certs(self):
         """ Attempts to update certificate status """
         pass
+
+    def is_hibernation_enabled_for_cert_update(self):
+        """Checks whether hibernation is enabled."""
+        return False
+
+    def should_reboot_before_cert_update(self):
+        """ Checks if a reboot is required before updating certificates """
+        return False
     # endregion