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 @@
         else:
             self.composite_logger.log_debug("Not executing in auto-assessment mode.")
 
+        # EULA config
+        self.accept_package_eula = self.__is_package_eula_accepted()
+
     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,27 @@
             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_package_eula_accepted(self):
+        """ Reads customer provided config on EULA acceptance from disk and returns a boolean.
+            NOTE: This is a temporary solution and will be deprecated no later than TBD date"""
+        if not os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS):
+            self.composite_logger.log_debug("NOT accepting EULA for any patch as no corresponding EULA Settings found on the VM")
+            return False
+
+        try:
+            eula_settings = json.loads(self.env_layer.file_system.read_with_retry(Constants.AzGPSPaths.EULA_SETTINGS) or 'null')
+            self.composite_logger.log_debug("EULA config values read from disk: AcceptEULAForAllPatches=[{0}] AcceptedBy=[{1}] LastModified=[{2}]"
+                                            .format(str(Constants.EulaSettings.ACCEPT_EULA_FOR_ALL_PATCHES), str(Constants.EulaSettings.ACCEPTED_BY),
+                                                    str(Constants.EulaSettings.LAST_MODIFIED)))
+            if eula_settings is not None \
+                    and Constants.EulaSettings.ACCEPT_EULA_FOR_ALL_PATCHES in eula_settings \
+                    and eula_settings[Constants.EulaSettings.ACCEPT_EULA_FOR_ALL_PATCHES] is True:
+                self.composite_logger.log_debug("Accept EULA set to True in customer config")
+                return True
+            else:
+                self.composite_logger.log_debug("Accept EULA not found to be set to True in customer config")
+                return False
+        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 False
+

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"""
@@ -561,6 +563,22 @@ 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)
+
 
 if __name__ == '__main__':
     unittest.main()

src/core/tests/Test_ConfigurationFactory.py

@@ -13,7 +13,8 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
-
+import json
+import os
 import unittest
 from core.src.bootstrap.Bootstrapper import Bootstrapper
 from core.src.bootstrap.Constants import Constants
@@ -30,6 +31,9 @@
     def tearDown(self):
         self.runtime.stop()
 
+    def mock_read_with_retry_raise_exception(self):
+        raise Exception
+
     def test_get_prod_config_correctly(self):
         bootstrapper = Bootstrapper(self.argument_composer, capture_stdout=False)
         config_factory = bootstrapper.configuration_factory
@@ -59,6 +63,101 @@
         self.assertEqual(config['package_manager_name'], Constants.APT)
         self.assertEqual(config['config_env'], Constants.DEV)
 
+    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()
+
+    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()
+
+        runtime = RuntimeCompositor(self.argument_composer, True, package_manager_name=Constants.APT)
+        container = runtime.container
+        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
+        self.assertTrue(os.path.exists(Constants.AzGPSPaths.EULA_SETTINGS))
+        self.assertRaises(Exception, container.get('execution_config'))
+        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: