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

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 certificates if feature flag to update certs is set
-        if self.execution_config.enable_uefi_cert_update:
+        if self.execution_config.enable_uefi_cert_update and self.package_manager.is_certificate_update_enabled_for_package_manager:
             self.try_update_certificates_for_default_patching()
 
         if self.execution_config.max_patch_publish_date != str():
@@ -808,24 +808,138 @@ 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:
+                self.composite_logger.log_verbose("Updating current certificates if needed...")
+                certs_updated = self.package_manager.try_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"""
+
+        self.composite_logger.log_verbose("Verifying all pre-requisites for updating certificates are met...")
+        if not self.__is_definitely_not_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.ensure_pre_cert_update_reboot_completed():
+            return False
+
+        return True
+
+    def __is_definitely_not_cvm(self):
+        # type: () -> bool
+        """ Returns true if this is not a Confidential VM (CVM), false otherwise (i.e. CVM or undeterministic) """
         try:
+            self.composite_logger.log_verbose("Verifying if this is a Confidential VM...")
             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 False
 
         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 not is_confidential_vm
+
+    def ensure_pre_cert_update_reboot_completed(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.
+        """
+        self.composite_logger.log_verbose("Ensuring pre certificate update reboot is completed...")
+        is_reboot_required_before_cert_update = self.package_manager.is_reboot_required_before_cert_update()
+        if is_reboot_required_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."""
+        self.composite_logger.log_verbose("Checking if latest UEFI certificates are already present...")
+        latest_certs_present = self.package_manager.are_latest_certs_present_with_mokutil_check()
+        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

@@ -93,7 +93,8 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
 
         self.package_install_expected_avg_time_in_seconds = 90  # As per telemetry data, the average time to install package is around 81 seconds for apt.
 
-        # Update certificates in factory defaults using default Ubuntu repos.
+        # Update certificates in factory defaults.
+        self.is_certificate_update_enabled_for_package_manager = True
         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
@@ -102,6 +103,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 # NOTE: This needs to be reviewed again later as these aren't specific to this class
 
     # region Sources Management
     def __get_custom_sources_to_spec(self, max_patch_published_date=str(), base_classification=str()):
@@ -957,6 +961,59 @@ 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 is_reboot_required_before_cert_update(self):
+        # type: () -> bool
+        """ Long-running VMs may not have the minimum firmware required for certificate updates.
+        We will reboot VMs with uptime > threshold to load the required firmware version. """
+        """ 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 or out.strip() == str():
+            self.composite_logger.log_debug('[APM][Certs] Unable to determine hibernation state for cert update. Assuming enabled. [Code={0}][Output={1}]'.format(str(code), out))
+            return True
+
+        # 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 +1026,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:
@@ -985,7 +1039,7 @@ def try_update_certs(self):
             """ 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
             or set/referred while installing. fwupd fetches and installs latest available certs. This is beneficial because our code doesn't become dated in the future
             (if it was pinned to an old version) but adds the need to verify if the certs we intend to update were applied. Hence the validation on latest certs. """
-            if self.are_latest_certs_present():
+            if self.are_latest_certs_present_with_mokutil_check():
                 self.composite_logger.log("[APM][Certs] Cert update completed and verified.")
                 success = True
             else:

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 is_reboot_required_before_cert_update(self):
+        """ Checks if a reboot is required before updating certificates """
+        return False
     # endregion