[Livepatching][MVP] Extension changes to support livepatching in Canonical

src/core/src/bootstrap/Constants.py

@@ -58,6 +58,7 @@ class SystemPaths(EnumBackport):
 
     class AzGPSPaths(EnumBackport):
         EULA_SETTINGS = "/var/lib/azure/linuxpatchextension/patch.eula.settings"
+        LIVEPATCH_CUSTOMER_SETTINGS = "/var/lib/azure/linuxpatchextension/livepatching.settings"
 
     class EnvSettings(EnumBackport):
         LOG_FOLDER = "logFolder"
@@ -87,6 +88,12 @@ class EulaSettings(EnumBackport):
         ACCEPTED_BY = 'AcceptedBy'
         LAST_MODIFIED = 'LastModified'
 
+    class LivePatchSettings(EnumBackport):
+        ENABLE_LIVEPATCH = 'EnableLivePatch' # boolean config that if set to true indicates customer's ask on AzGPS to apply livepatches on their VM
+        LIVEPATCH_ONLY = 'LivePatchOnly' # boolean config that indicates if the customer is requesting only livepatches i.e. no cold patch or regular patching with reboots
+        ENABLED_BY = 'EnabledBy'
+        LAST_MODIFIED = 'LastModified'
+
     TEMP_FOLDER_DIR_NAME = "tmp"
     TEMP_FOLDER_CLEANUP_ARTIFACT_LIST = ["*.list", "azgps*"]
 
@@ -312,6 +319,7 @@ class PatchOperationErrorCodes(EnumBackport):
         NEWER_OPERATION_SUPERSEDED = "NEWER_OPERATION_SUPERSEDED"
         UA_ESM_REQUIRED = "UA_ESM_REQUIRED"
         TRUNCATION = "PACKAGE_LIST_TRUNCATED"
+        LIVEPATCH_ERROR = "LIVEPATCH_ERROR"
 
     ERROR_ADDED_TO_STATUS = "Error_added_to_status"
 

src/core/src/bootstrap/EnvLayer.py

@@ -421,4 +421,10 @@ def datetime_string_to_posix_time(datetime_string, format_string):
             epoch = datetime.datetime(1970, 1, 1)
             return int((dt - epoch).total_seconds())
 
+        @staticmethod
+        def datetime_iso_basic_string_to_extended_string(datetime_string):
+            # type: (str) -> str
+            """Converts ISO 8601 basic date format (20240401T000000Z) to extended format (2024-04-01T00:00:00Z)."""
+            return datetime.datetime.strptime(datetime_string, "%Y%m%dT%H%M%SZ").strftime("%Y-%m-%dT%H:%M:%SZ")
+
 # endregion - DateTime emulator and extensions

src/core/src/core_logic/ExecutionConfig.py

@@ -92,6 +92,11 @@ def __init__(self, env_layer, composite_logger, execution_parameters):
         # EULA config
         self.accept_package_eula = self.__is_eula_accepted_for_all_patches()
 
+        # LivePatch config
+        self.livepatch_customer_config_settings = self.__get_livepatch_customer_config_in_json()
+        self.is_livepatch_requested = self.__is_livepatch_requested(self.livepatch_customer_config_settings)
+        self.is_livepatch_only_requested = self.__is_livepatch_only_requested(self.livepatch_customer_config_settings)
+
     def __transform_execution_config_for_auto_assessment(self):
         self.activity_id = str(uuid.uuid4())
         self.included_classifications_list = self.included_package_name_mask_list = self.excluded_package_name_mask_list = []
@@ -246,10 +251,10 @@ def __is_eula_accepted_for_all_patches(self):
         try:
             if os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS):
                 eula_settings = json.loads(self.env_layer.file_system.read_with_retry(Constants.AzGPSPaths.EULA_SETTINGS) or 'null')
-                accept_eula_for_all_patches = self.__fetch_specific_eula_setting(eula_settings, Constants.EulaSettings.ACCEPT_EULA_FOR_ALL_PATCHES)
-                accepted_by = self.__fetch_specific_eula_setting(eula_settings, Constants.EulaSettings.ACCEPTED_BY)
-                last_modified = self.__fetch_specific_eula_setting(eula_settings, Constants.EulaSettings.LAST_MODIFIED)
-                if accept_eula_for_all_patches is not None and accept_eula_for_all_patches in [True, 'True', 'true', '1', 1]:
+                accept_eula_for_all_patches = self.__fetch_specific_setting(eula_settings, Constants.EulaSettings.ACCEPT_EULA_FOR_ALL_PATCHES)
+                accepted_by = self.__fetch_specific_setting(eula_settings, Constants.EulaSettings.ACCEPTED_BY)
+                last_modified = self.__fetch_specific_setting(eula_settings, Constants.EulaSettings.LAST_MODIFIED)
+                if self.__is_truthy(accept_eula_for_all_patches):
                     is_eula_accepted = True
                 self.composite_logger.log_debug("EULA config values from disk: [AcceptEULAForAllPatches={0}] [AcceptedBy={1}] [LastModified={2}]. Computed value of [IsEULAAccepted={3}]"
                                                 .format(str(accept_eula_for_all_patches), str(accepted_by), str(last_modified), str(is_eula_accepted)))
@@ -260,10 +265,72 @@ def __is_eula_accepted_for_all_patches(self):
 
         return is_eula_accepted
 
+    def __get_livepatch_customer_config_in_json(self):
+        # type: () -> dict
+        """ Reads customer provided config on livepatch from disk and returns a dict with the config values.
+            NOTE: This is a temporary solution and will be deprecated soon """
+        livepatch_customer_config = dict()
+        try:
+            if os.path.exists(Constants.AzGPSPaths.LIVEPATCH_CUSTOMER_SETTINGS):
+                livepatch_customer_config = json.loads(self.env_layer.file_system.read_with_retry(Constants.AzGPSPaths.LIVEPATCH_CUSTOMER_SETTINGS) or 'null') or dict()
+                self.composite_logger.log_debug("Livepatch customer config values from disk: [Config={0}]".format(str(livepatch_customer_config)))
+            else:
+                self.composite_logger.log_debug("No livepatch customer config found on the VM. Returning empty config.")
+        except Exception as error:
+            self.composite_logger.log_debug("Error occurred while reading and parsing livepatch customer config. Returning empty config. Error=[{0}]".format(repr(error)))
+
+        return livepatch_customer_config
+
+    def __is_livepatch_requested(self, livepatch_settings):
+        # type: (dict) -> bool
+        """ Determines if livepatch is requested in config settings. Returns a boolean."""
+        livepatch_requested = False
+
+        enable_livepatch = self.__fetch_specific_setting(livepatch_settings, Constants.LivePatchSettings.ENABLE_LIVEPATCH)
+        enabled_by = self.__fetch_specific_setting(livepatch_settings, Constants.LivePatchSettings.ENABLED_BY)
+        last_modified = self.__fetch_specific_setting(livepatch_settings, Constants.LivePatchSettings.LAST_MODIFIED)
+        if self.__is_truthy(enable_livepatch):
+            livepatch_requested = True
+            self.composite_logger.log_debug("Livepatch config values read from disk: [EnableLivePatch={0}] [EnabledBy={1}] [LastModified={2}]. Computed value of [LivePatchRequested={3}]"
+                                            .format(str(enable_livepatch), str(enabled_by), str(last_modified), str(livepatch_requested)))
+        else:
+            self.composite_logger.log_debug("LivePatch is not requested for the VM. [EnableLivePatchValueFromConfig={0}]. Computed value of [LivePatchRequested={1}]"
+                                            .format(str(enable_livepatch),str(livepatch_requested)))
+
+        return livepatch_requested
+
+    def __is_livepatch_only_requested(self, livepatch_settings):
+        # type: (dict) -> bool
+        """ Determines if livepatch only, i.e. no cold patch, is requested in config settings. Returns a boolean."""
+        """ NOTE: This is not in use currently but can be added in MVP if needed"""
+        livepatch_only_config = self.__fetch_specific_setting(livepatch_settings, Constants.LivePatchSettings.LIVEPATCH_ONLY)
+        livepatch_only_requested = self.__is_truthy(livepatch_only_config)
+        self.composite_logger.log_debug("Livepatch only config values from disk: [LivePatchOnly={0}]. Computed value of [LivePatchOnlyRequested={1}]"
+                                        .format(str(livepatch_only_config), str(livepatch_only_requested)))
+        return livepatch_only_requested
+
     @staticmethod
-    def __fetch_specific_eula_setting(settings_source, setting_to_fetch):
-        """ Returns the specific setting value from eula_settings_source or None if not found """
+    def __fetch_specific_setting(settings_source, setting_to_fetch):
+        # type: (dict, str) -> str or None
+        """ Returns the specific setting value from the given settings_source or None if not found """
         if settings_source is not None and setting_to_fetch is not None and setting_to_fetch in settings_source:
             return settings_source[setting_to_fetch]
         return None
 
+    @staticmethod
+    def __is_truthy(value):
+        # type: (any) -> bool
+        """Case-insensitive truthy evaluator for config values."""
+        if isinstance(value, bool):
+            return value
+        if isinstance(value, int):
+            return value == 1
+
+        # Cross-version text types:
+        # py2 -> (str, unicode)
+        # py3 -> (str, str)
+        text_types = (str, type(u""))
+        if isinstance(value, text_types):
+            return value.strip().lower() in ("true", "1")
+        return False
+

src/core/src/core_logic/PatchInstaller.py

@@ -58,13 +58,16 @@ def start_installation(self, simulate=False):
         self.raise_if_telemetry_unsupported()
         self.raise_if_min_python_version_not_met()
 
+        maintenance_window = self.maintenance_window
+        package_manager = self.package_manager
+        reboot_manager = self.reboot_manager
+
         self.composite_logger.log("\nStarting patch installation... [MachineId: " + self.env_layer.platform.vm_name() + "][ActivityId: " + self.execution_config.activity_id + "][StartTime: " + self.execution_config.start_time + "][MaintenanceWindowDuration: " + self.execution_config.duration + "]")
 
         self.stopwatch.start()
 
-        maintenance_window = self.maintenance_window
-        package_manager = self.package_manager
-        reboot_manager = self.reboot_manager
+        if self.execution_config.is_livepatch_requested:
+            package_manager.start_livepatch()
 
         # Early reboot if reboot is allowed by settings and required by the machine
         reboot_pending = self.package_manager.is_reboot_pending()