Adding EULA acceptance based on an inVM approval mechanism

src/core/src/bootstrap/Constants.py

@@ -51,9 +51,12 @@ def __iter__(self):
     MAX_AUTO_ASSESSMENT_LOGFILE_SIZE_IN_BYTES = 5*1024*1024
     MAX_AUTO_ASSESSMENT_WAIT_FOR_MAIN_CORE_EXEC_IN_MINUTES = 3 * 60
 
-    class Paths(EnumBackport):
+    class SystemPaths(EnumBackport):
         SYSTEMD_ROOT = "/etc/systemd/system/"
 
+    class AzGPSPaths(EnumBackport):
+        EULA_SETTINGS = "/var/lib/azure/linuxpatchextension/patch.eula.settings"
+
     class EnvSettings(EnumBackport):
         LOG_FOLDER = "logFolder"
         CONFIG_FOLDER = "configFolder"
@@ -77,6 +80,11 @@ class ConfigSettings(EnumBackport):
         ASSESSMENT_MODE = 'assessmentMode'
         MAXIMUM_ASSESSMENT_INTERVAL = 'maximumAssessmentInterval'
 
+    class EulaSettings(EnumBackport):
+        ACCEPT_EULA_FOR_ALL_PATCHES = 'AcceptEULAForAllPatches'
+        ACCEPTED_BY = 'AcceptedBy'
+        LAST_MODIFIED = 'LastModified'
+
     TEMP_FOLDER_DIR_NAME = "tmp"
     TEMP_FOLDER_CLEANUP_ARTIFACT_LIST = ["*.list"]
 

src/core/src/core_logic/ExecutionConfig.py

@@ -86,6 +86,9 @@ def __init__(self, env_layer, composite_logger, execution_parameters):
         else:
             self.composite_logger.log_debug("Not executing in auto-assessment mode.")
 
+        # EULA config
+        self.accept_package_eula = self.__is_eula_accepted_for_all_patches()
+
     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 = []
@@ -179,3 +182,31 @@ def __check_and_create_temp_folder_if_not_exists(self):
             self.composite_logger.log_debug("Temp folder does not exist, creating one from extension core. [Path={0}]".format(str(self.temp_folder)))
             os.mkdir(self.temp_folder)
 
+    def __is_eula_accepted_for_all_patches(self):
+        """ Reads customer provided config on EULA acceptance from disk and returns a boolean.
+            NOTE: This is a temporary solution and will be deprecated soon """
+        is_eula_accepted = False
+        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]:
+                    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)))
+            else:
+                self.composite_logger.log_debug("No EULA Settings found on the VM. Computed value of [IsEULAAccepted={0}]".format(str(is_eula_accepted)))
+        except Exception as error:
+            self.composite_logger.log_debug("Error occurred while reading and parsing EULA settings. Not accepting EULA for any patch. Error=[{0}]".format(repr(error)))
+
+        return is_eula_accepted
+
+    @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 """
+        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
+

src/core/src/core_logic/SystemctlManager.py

@@ -31,7 +31,7 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         self.service_desc = service_info.service_desc
         self.service_exec_path = service_info.service_exec_path
 
-        self.__systemd_path = Constants.Paths.SYSTEMD_ROOT
+        self.__systemd_path = Constants.SystemPaths.SYSTEMD_ROOT
         self.systemctl_daemon_reload_cmd = "sudo systemctl daemon-reload"
         self.systemctl_version = "systemctl --version"
 

src/core/src/package_managers/AptitudePackageManager.py

@@ -34,6 +34,10 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         super(AptitudePackageManager, self).__init__(env_layer, execution_config, composite_logger, telemetry_writer, status_handler)
 
         security_list_guid = str(uuid.uuid4())
+
+        # Accept EULA (End User License Agreement) as per the EULA settings set by user
+        optional_accept_eula_in_cmd = "ACCEPT_EULA=Y" if execution_config.accept_package_eula else ""
+
         # Repo refresh
         self.repo_refresh = 'sudo apt-get -q update'
 
@@ -44,12 +48,12 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
         self.single_package_check_versions = 'apt-cache madison <PACKAGE-NAME>'
         self.single_package_find_installed_dpkg = 'sudo dpkg -s <PACKAGE-NAME>'
         self.single_package_find_installed_apt = 'sudo apt list --installed <PACKAGE-NAME>'
-        self.single_package_upgrade_simulation_cmd = '''DEBIAN_FRONTEND=noninteractive apt-get -y --only-upgrade true -s install '''
-        self.single_package_dependency_resolution_template = 'DEBIAN_FRONTEND=noninteractive LANG=en_US.UTF8 apt-get -y --only-upgrade true -s install <PACKAGE-NAME> '
+        self.single_package_upgrade_simulation_cmd = '''DEBIAN_FRONTEND=noninteractive ''' + optional_accept_eula_in_cmd + ''' apt-get -y --only-upgrade true -s install '''
+        self.single_package_dependency_resolution_template = 'DEBIAN_FRONTEND=noninteractive ' + optional_accept_eula_in_cmd + ' LANG=en_US.UTF8 apt-get -y --only-upgrade true -s install <PACKAGE-NAME> '
 
         # Install update
         # --only-upgrade: upgrade only single package (only if it is installed)
-        self.single_package_upgrade_cmd = '''sudo DEBIAN_FRONTEND=noninteractive apt-get -y --only-upgrade true install '''
+        self.single_package_upgrade_cmd = '''sudo DEBIAN_FRONTEND=noninteractive ''' + optional_accept_eula_in_cmd + ''' apt-get -y --only-upgrade true install '''
 
         # Package manager exit code(s)
         self.apt_exitcode_ok = 0

src/core/tests/Test_AptitudePackageManager.py

@@ -17,6 +17,7 @@
 import os
 import unittest
 from core.src.bootstrap.Constants import Constants
+from core.src.core_logic.ExecutionConfig import ExecutionConfig
 from core.tests.Test_UbuntuProClient import MockVersionResult, MockRebootRequiredResult, MockUpdatesResult
 from core.tests.library.ArgumentComposer import ArgumentComposer
 from core.tests.library.LegacyEnvLayerExtensions import LegacyEnvLayerExtensions
@@ -26,7 +27,8 @@
 
 class TestAptitudePackageManager(unittest.TestCase):
     def setUp(self):
-        self.runtime = RuntimeCompositor(ArgumentComposer().get_composed_arguments(), True, Constants.APT)
+        self.argument_composer = ArgumentComposer().get_composed_arguments()
+        self.runtime = RuntimeCompositor(self.argument_composer, True, Constants.APT)
         self.container = self.runtime.container
 
     def tearDown(self):
@@ -63,7 +65,7 @@ def mock_os_path_isfile_raise_exception(self, file):
     def mock_get_security_updates_return_empty_list(self):
         return [], []
 
-    #endregion Mocks
+    # endregion Mocks
 
     def test_package_manager_no_updates(self):
         """Unit test for aptitude package manager with no updates"""
@@ -570,6 +572,238 @@ def test_check_pro_client_prerequisites_should_return_false(self):
         LegacyEnvLayerExtensions.LegacyPlatform.linux_distribution = backup_envlayer_platform_linux_distribution
         package_manager.ubuntu_pro_client.is_pro_working = backup_ubuntu_pro_client_is_pro_working
 
+    def test_eula_accepted_for_patches(self):
+        # EULA accepted in settings and commands updated accordingly
+        self.runtime.execution_config.accept_package_eula = True
+        package_manager_for_test = AptitudePackageManager.AptitudePackageManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer, self.runtime.status_handler)
+        self.assertTrue("ACCEPT_EULA=Y" in package_manager_for_test.single_package_upgrade_simulation_cmd)
+        self.assertTrue("ACCEPT_EULA=Y" in package_manager_for_test.single_package_dependency_resolution_template)
+        self.assertTrue("ACCEPT_EULA=Y" in package_manager_for_test.single_package_upgrade_cmd)
+
+    def test_eula_not_accepted_for_patches(self):
+        # EULA accepted in settings and commands updated accordingly
+        self.runtime.execution_config.accept_package_eula = False
+        package_manager_for_test = AptitudePackageManager.AptitudePackageManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer, self.runtime.status_handler)
+        self.assertTrue("ACCEPT_EULA=Y" not in package_manager_for_test.single_package_upgrade_simulation_cmd)
+        self.assertTrue("ACCEPT_EULA=Y" not in package_manager_for_test.single_package_dependency_resolution_template)
+        self.assertTrue("ACCEPT_EULA=Y" not in package_manager_for_test.single_package_upgrade_cmd)
+
+    def test_eula_acceptance_file_read_success(self):
+        self.runtime.stop()
+
+        # Accept EULA set to true
+        eula_settings = {
+            "AcceptEULAForAllPatches": True,
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, True)
+        runtime.stop()
+
+        # Accept EULA set to false
+        eula_settings = {
+            "AcceptEULAForAllPatches": False,
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        runtime.stop()
+
+        # Accept EULA set to true in a string i.e. 'true'
+        eula_settings = {
+            "AcceptEULAForAllPatches": 'true',
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, True)
+        runtime.stop()
+
+        # Accept EULA set to true in a string i.e. 'True'
+        eula_settings = {
+            "AcceptEULAForAllPatches": 'True',
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, True)
+        runtime.stop()
+
+        # Accept EULA set to true in a string i.e. 'False'
+        eula_settings = {
+            "AcceptEULAForAllPatches": 'False',
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        runtime.stop()
+
+        # Accept EULA set to true in a string i.e. 'false'
+        eula_settings = {
+            "AcceptEULAForAllPatches": 'false',
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        runtime.stop()
+
+        # Accept EULA set as '0'
+        eula_settings = {
+            "AcceptEULAForAllPatches": '0',
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        runtime.stop()
+
+        # Accept EULA set as 0
+        eula_settings = {
+            "AcceptEULAForAllPatches": 0,
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        runtime.stop()
+
+        # Accept EULA set as 1
+        eula_settings = {
+            "AcceptEULAForAllPatches": 1,
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, True)
+        runtime.stop()
+
+        # Accept EULA set as '1'
+        eula_settings = {
+            "AcceptEULAForAllPatches": '1',
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, True)
+        runtime.stop()
+
+    def test_eula_acceptance_file_read_when_no_data_found(self):
+        self.runtime.stop()
+
+        # EULA file does not exist
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        self.assertFalse(os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS))
+        runtime.stop()
+
+        # EULA settings set to None
+        eula_settings = None
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        self.assertTrue(os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS))
+        runtime.stop()
+
+        # AcceptEULAForAllPatches not set in config
+        eula_settings = {
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        self.assertTrue(os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS))
+        runtime.stop()
+
+        # AcceptEULAForAllPatches not set to a boolean
+        eula_settings = {
+            "AcceptEULAForAllPatches": "test",
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.AzGPSPaths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        execution_config = container.get('execution_config')
+        self.assertEqual(execution_config.accept_package_eula, False)
+        self.assertTrue(os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS))
+        runtime.stop()
+
+        # EULA not accepted for cases where file read raises an Exception
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        self.backup_read_with_retry = runtime.env_layer.file_system.read_with_retry
+        runtime.env_layer.file_system.read_with_retry = self.mock_read_with_retry_raise_exception
+        exec_config = ExecutionConfig(runtime.env_layer, runtime.composite_logger, str(self.argument_composer))
+        self.assertTrue(os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS))
+        self.assertEqual(exec_config.accept_package_eula, False)
+        runtime.stop()
+
 
 if __name__ == '__main__':
     unittest.main()

src/core/tests/library/ArgumentComposer.py

@@ -47,6 +47,7 @@ def __init__(self):
         self.__status_folder = self.__get_custom_folder(scratch_folder, self.__STATUS_FOLDER)
         self.events_folder = self.__get_custom_folder(self.__log_folder, self.__EVENTS_FOLDER)
         self.temp_folder = self.__get_custom_folder(scratch_folder, self.__TEMP_FOLDER)
+        Constants.AzGPSPaths.EULA_SETTINGS = os.path.join(scratch_folder, "patch.eula.settings")
 
         # config settings
         self.operation = Constants.INSTALLATION

src/core/tests/library/RuntimeCompositor.py

@@ -47,7 +47,7 @@ def __init__(self, argv=Constants.DEFAULT_UNSPECIFIED_VALUE, legacy_mode=False,
         os.environ[Constants.LPE_ENV_VARIABLE] = self.current_env
         self.argv = argv if argv != Constants.DEFAULT_UNSPECIFIED_VALUE else ArgumentComposer().get_composed_arguments()
         self.vm_cloud_type = vm_cloud_type
-        Constants.Paths.SYSTEMD_ROOT = os.getcwd() # mocking to pass a basic systemd check in Windows
+        Constants.SystemPaths.SYSTEMD_ROOT = os.getcwd() # mocking to pass a basic systemd check in Windows
         self.is_github_runner = os.getenv('RUNNER_TEMP', None) is not None
 
         if self.is_github_runner: