Adding EULA acceptance based on an inVM approval mechanism

src/core/src/bootstrap/ConfigurationFactory.py

@@ -16,6 +16,8 @@
 
 """ Configure factory. This module populates configuration based on package manager and environment, e.g. TEST/DEV/PROD"""
 from __future__ import print_function
+
+import json
 import os
 import time
 from core.src.bootstrap.Constants import Constants
@@ -91,16 +93,16 @@
         configuration_key = str.lower('{0}_config'.format(str(env)))
         return self.bootstrap_configurations[configuration_key]
 
-    @staticmethod
-    def get_arguments_configuration(argv):
+    def get_arguments_configuration(self, argv):
         """ Composes the configuration with the passed in arguments. """
         arguments_config = {
             'execution_arguments': str(argv),
             'execution_config': {
                 'component': ExecutionConfig,
                 'component_args': ['env_layer', 'composite_logger'],
                 'component_kwargs': {
-                    'execution_parameters': str(argv)
+                    'execution_parameters': str(argv),
+                    'accept_package_eula': self.is_package_eula_accepted(),
                 }
             }
         }
@@ -298,4 +300,56 @@
                 else:
                     print("Failed to connect IMDS end point after 5 retries. This is expected in Arc VMs. VMCloudType is set to Arc.\n")
                     return Constants.VMCloudType.ARC
+
+    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.Paths.EULA_SETTINGS):
+            print("NOT accepting EULA for any patch as no corresponding EULA Settings found on the VM")
+            return False
+
+        try:
+            eula_settings = json.loads(self.read_with_retry(Constants.Paths.EULA_SETTINGS) or 'null')
+            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:
+                print("Accept EULA set to True in customer config")
+                return True
+            else:
+                print("Accept EULA not found to be set to True in customer config")
+                return False
+        except Exception as error:
+            print("Error occurred while reading and parsing EULA settings. Not accepting EULA for any patch. Error=[{0}]".format(repr(error)))
+            return False
+
+    @staticmethod
+    def open(file_path, mode):
+        """ Provides a file handle to the file_path requested using implicit redirection where required """
+        for i in range(0, Constants.MAX_FILE_OPERATION_RETRY_COUNT):
+            try:
+                return open(file_path, mode)
+            except Exception as error:
+                if i < Constants.MAX_FILE_OPERATION_RETRY_COUNT - 1:
+                    time.sleep(i + 1)
+                else:
+                    raise Exception("Unable to open {0} (retries exhausted). Error: {1}.".format(str(file_path), repr(error)))
+
+    def __obtain_file_handle(self, file_path_or_handle, mode='a+'):
+        """ Pass-through for handle. For path, resolution and handle open with retry. """
+        is_path = False
+        if isinstance(file_path_or_handle, str) or not (hasattr(file_path_or_handle, 'read') and hasattr(file_path_or_handle, 'write')):
+            is_path = True
+            file_path_or_handle = self.open(file_path_or_handle, mode)
+        file_handle = file_path_or_handle
+        return file_handle, is_path
+
+    def read_with_retry(self, file_path_or_handle):
+        """ Reads all content from a given file path in a single operation """
+        if isinstance(file_path_or_handle, str):
+            file_handle, was_path = self.__obtain_file_handle(file_path_or_handle, 'r')
+            value = file_handle.read()
+            if was_path:  # what was passed in was not a file handle, so close the handle that was init here
+                file_handle.close()
+            return value
+        return None
     # endregion

src/core/src/bootstrap/Constants.py

@@ -53,6 +53,7 @@ def __iter__(self):
 
     class Paths(EnumBackport):
         SYSTEMD_ROOT = "/etc/systemd/system/"
+        EULA_SETTINGS = "/var/lib/azure/linuxpatchextension/patch.eula.settings"
 
     class EnvSettings(EnumBackport):
         LOG_FOLDER = "logFolder"
@@ -77,6 +78,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

@@ -23,7 +23,7 @@
 
 
 class ExecutionConfig(object):
-    def __init__(self, env_layer, composite_logger, execution_parameters):
+    def __init__(self, env_layer, composite_logger, execution_parameters, accept_package_eula):
         self.env_layer = env_layer
         self.composite_logger = composite_logger
         self.execution_parameters = eval(execution_parameters)
@@ -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 = accept_package_eula
+
     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 = []

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
@@ -288,6 +292,7 @@ def get_composite_package_identifier(self, package, package_version):
 
     def install_updates_fail_safe(self, excluded_packages):
         return
+
     # endregion
 
     # region Package Information

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,84 @@
         self.assertEqual(config['package_manager_name'], Constants.APT)
         self.assertEqual(config['config_env'], Constants.DEV)
 
+    def test_eula_acceptance_file_read_success(self):
+        bootstrapper = Bootstrapper(self.argument_composer, capture_stdout=False)
+        config_factory = bootstrapper.configuration_factory
+        self.assertTrue(config_factory)
+
+        # Accept EULA set to true
+        eula_settings = {
+            "AcceptEULAForAllPatches": True,
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.Paths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        config = config_factory.get_arguments_configuration(self.argument_composer)
+        self.assertEqual(config['execution_config']['component_kwargs']['accept_package_eula'], True)
+
+        # Accept EULA set to true
+        eula_settings = {
+            "AcceptEULAForAllPatches": False,
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.Paths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        config = config_factory.get_arguments_configuration(self.argument_composer)
+        self.assertEqual(config['execution_config']['component_kwargs']['accept_package_eula'], False)
+
+    def test_eula_acceptance_file_read_when_no_data_found(self):
+        bootstrapper = Bootstrapper(self.argument_composer, capture_stdout=False)
+        config_factory = bootstrapper.configuration_factory
+        self.assertTrue(config_factory)
+
+        # EULA file does not exist
+        config = config_factory.get_arguments_configuration(self.argument_composer)
+        self.assertEqual(config['execution_config']['component_kwargs']['accept_package_eula'], False)
+        self.assertFalse(os.path.exists(Constants.Paths.EULA_SETTINGS))
+
+        # EULA settings set to None
+        eula_settings = None
+        f = open(Constants.Paths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        config = config_factory.get_arguments_configuration(self.argument_composer)
+        self.assertEqual(config['execution_config']['component_kwargs']['accept_package_eula'], False)
+        self.assertTrue(os.path.exists(Constants.Paths.EULA_SETTINGS))
+
+        # AcceptEULAForAllPatches not set in config
+        eula_settings = {
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.Paths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        config = config_factory.get_arguments_configuration(self.argument_composer)
+        self.assertEqual(config['execution_config']['component_kwargs']['accept_package_eula'], False)
+        self.assertTrue(os.path.exists(Constants.Paths.EULA_SETTINGS))
+
+        # AcceptEULAForAllPatches not set to a boolean
+        eula_settings = {
+            "AcceptEULAForAllPatches": "test",
+            "AcceptedBy": "TestSetup",
+            "LastModified": "2023-08-29"
+        }
+        f = open(Constants.Paths.EULA_SETTINGS, "w+")
+        f.write(json.dumps(eula_settings))
+        f.close()
+        config = config_factory.get_arguments_configuration(self.argument_composer)
+        self.assertEqual(config['execution_config']['component_kwargs']['accept_package_eula'], False)
+        self.assertTrue(os.path.exists(Constants.Paths.EULA_SETTINGS))
+
+        self.backup_read_with_retry = config_factory.read_with_retry
+        config_factory.read_with_retry = self.mock_read_with_retry_raise_exception
+        self.assertTrue(os.path.exists(Constants.Paths.EULA_SETTINGS))
+        self.assertRaises(Exception, config_factory.get_arguments_configuration(self.argument_composer))
+
 
 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.Paths.EULA_SETTINGS = os.path.join(scratch_folder, "patch.eula.settings")
 
         # config settings
         self.operation = Constants.INSTALLATION