Bugfix: Apply SSL expired cmd fix to RHEL images only

src/core/src/bootstrap/EnvLayer.py

@@ -298,7 +298,7 @@ def linux_distribution(self):
             else:
                 code, output = self.__read_record(operation)
                 return eval(output)
-
+        
         def system(self):   # OS Type
             operation = "PLATFORM_SYSTEM"
             if not self.__emulator_enabled:

src/core/src/package_managers/YumPackageManager.py

@@ -95,8 +95,6 @@ def __init__(self, env_layer, execution_config, composite_logger, telemetry_writ
                                        "Error: Cannot retrieve repository metadata (repomd.xml) for repository": self.fix_ssl_certificate_issue,
                                        "Error: Failed to download metadata for repo":  self.fix_ssl_certificate_issue}
 
-        self.yum_update_client_package = "sudo yum update -y --disablerepo='*' --enablerepo='*microsoft*'"
-
         self.package_install_expected_avg_time_in_seconds = 90  # As per telemetry data, the average time to install package is around 90 seconds for yum.
 
     def refresh_repo(self):
@@ -141,7 +139,6 @@ def get_security_updates(self):
 
         if not self.__is_image_rhel8_or_higher():
             self.install_yum_security_prerequisite()
-
         out = self.invoke_package_manager(self.yum_check_security)
         security_packages, security_package_versions = self.extract_packages_and_versions(out)
 
@@ -175,16 +172,24 @@ def get_other_updates(self):
         return other_packages, other_package_versions
 
     def __is_image_rhel8_or_higher(self):
+        # type: () -> bool
         """ Check if image is RHEL8+ return true else false """
-        if self.env_layer.platform.linux_distribution() is not None:
+        if self.env_layer.platform.linux_distribution is not None:
             os_offer, os_version, os_code = self.env_layer.platform.linux_distribution()
-
             if "Red Hat Enterprise Linux" in os_offer and int(os_version.split('.')[0]) >= 8:
                 self.composite_logger.log_debug("[YPM] RHEL version >= 8 detected. [DetectedVersion={0}]".format(str(os_version)))
                 return True
-
         return False
-
+
+    def __is_image_rhel(self):
+        # type: () -> bool
+        """ Check if image is RHEL return true else false """
+        if self.env_layer.platform.linux_distribution is not None:
+            os_offer, os_version, os_code = self.env_layer.platform.linux_distribution()
+            if "Red Hat Enterprise Linux" in os_offer:
+                return True
+        return False
+
     def set_max_patch_publish_date(self, max_patch_publish_date=str()):
         pass
 
@@ -893,23 +898,32 @@ def check_known_issues_and_attempt_fix(self, output):
         return False
 
     def fix_ssl_certificate_issue(self):
-        command = self.yum_update_client_package
-        self.composite_logger.log_debug("[Customer-environment-error] Updating client package to avoid errors from older certificates using command: [Command={0}]".format(str(command)))
+        # type: () -> None
+        """ Attempt to fix the SSL certificate issue by updating the client package """
+        if not self.__is_image_rhel():
+            error_msg = 'Customer environment error (expired SSL certs)'
+            self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.PACKAGE_MANAGER_FAILURE)
+            raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
+
+        # Image is rhel, attempt to update the client package
+        command = "sudo yum update -y --disablerepo='*' --enablerepo='*microsoft*'"
+        self.composite_logger.log_debug("[YPM][Customer-environment-error] Updating client package to avoid errors from older certificates using command: [Command={0}]".format(str(command)))
         code, out = self.env_layer.run_command_output(command, False, False)
+
         if code != self.yum_exitcode_no_applicable_packages:
-            error_msg = 'Customer environment error (expired SSL certs):  [Command={0}][Code={1}]'.format(command,str(code))
+            error_msg = 'Customer environment error (expired SSL certs):  [Command={0}][Code={1}]'.format(command, str(code))
             self.composite_logger.log_error("{0}[Out={1}]".format(error_msg, out))
             self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.PACKAGE_MANAGER_FAILURE)
             raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
         else:
             self.composite_logger.log_verbose("\n\n==[SUCCESS]===============================================================")
             self.composite_logger.log_debug("Client package update complete. [Code={0}][Out={1}]".format(str(code), out))
             self.composite_logger.log_verbose("==========================================================================\n\n")
-
+            
     def log_error_mitigation_failure(self, output, raise_on_exception=True):
         self.composite_logger.log_error("[YPM] Customer Environment Error: Unable to auto-mitigate known issue. Please investigate and address. [Out={0}]".format(output))
         if raise_on_exception:
-            error_msg = 'Customer environment error (Unable to auto-mitigate known issue):  [Out={0}]'.format(output)
+            error_msg = '[YMP] Customer environment error (Unable to auto-mitigate known issue):  [Out={0}]'.format(output)
             self.status_handler.add_error_to_status(error_msg, Constants.PatchOperationErrorCodes.PACKAGE_MANAGER_FAILURE)
             raise Exception(error_msg, "[{0}]".format(Constants.ERROR_ADDED_TO_STATUS))
     # endregion

src/core/tests/Test_YumPackageManager.py

@@ -16,6 +16,13 @@
 import json
 import os
 import unittest
+import sys
+# Conditional import for StringIO
+try:
+    from StringIO import StringIO  # Python 2
+except ImportError:
+    from io import StringIO  # Python 3
+
 from core.src.bootstrap.Constants import Constants
 from core.tests.library.ArgumentComposer import ArgumentComposer
 from core.tests.library.LegacyEnvLayerExtensions import LegacyEnvLayerExtensions
@@ -42,6 +49,9 @@ def mock_linux7_distribution_to_return_redhat(self):
 
     def mock_linux8_distribution_to_return_redhat(self):
         return ['Red Hat Enterprise Linux Server', '8', 'Ootpa']
+
+    def mock_ubuntu_distribution(self):
+        return ['Ubuntu', '20.04', 'Focal Fossa']
     #endregion Mocks
 
     def mock_do_processes_require_restart_raise_exception(self):
@@ -756,6 +766,78 @@ def test_get_dependent_list_yum_version_4_update_in_two_lines_with_unexpected_ou
         package_manager = self.container.get('package_manager')
         dependent_list = package_manager.get_dependent_list(["polkit.x86_64"])
         self.assertEqual(len(dependent_list), 0)
+
+    def test_apply_fix_ssl_cert_issue_rhel_image_success(self):
+        """ Test ssl expired cert issue with success on rhel images only."""
+        # Set up and mocks
+        captured_output = StringIO()
+        original_output = sys.stdout
+        sys.stdout = captured_output  # Redirect stdout to the StringIO object
+
+        original_env_layer_platform_linux_distribution = self.runtime.env_layer.platform.linux_distribution
+        self.runtime.env_layer.platform.linux_distribution = self.mock_linux7_distribution_to_return_redhat
+        self.runtime.set_legacy_test_type('SSLCertificateIssueType1HappyPathAfterFix')
+
+        package_manager = self.runtime.container.get('package_manager')
+        self.assertIsNotNone(package_manager)
+
+        # Act
+        package_manager.fix_ssl_certificate_issue()
+
+        # Verify
+        output = captured_output.getvalue()
+        self.assertIn("[YPM][Customer-environment-error] Updating client package to avoid errors from older certificates using command", output)  # Verify the log output contains the expected text # Verify the log output contains the expected text
+        self.assertIn("Client package update complete", output)  # Verify the log output contains the expected text
+
+        # Restore stdout, mock
+        sys.stdout = original_output
+        self.runtime.env_layer.platform.linux_distribution = original_env_layer_platform_linux_distribution
+
+        self.runtime.stop()
+
+    def test_apply_fix_ssl_cert_issue_rhel_image_failed(self):
+        """ Test ssl expired cert issue failed to fix throw exception on rhel images only."""
+        # Set up and mocks
+        original_env_layer_platform_linux_distribution = self.runtime.env_layer.platform.linux_distribution
+        self.runtime.env_layer.platform.linux_distribution = self.mock_linux7_distribution_to_return_redhat
+        self.runtime.set_legacy_test_type('SSLCertificateIssueType1SadPathAfterFix')
+
+        package_manager = self.runtime.container.get('package_manager')
+        self.assertIsNotNone(package_manager)
+
+        # Act
+        with self.assertRaises(Exception) as context:
+            package_manager.fix_ssl_certificate_issue()
+
+        # Verify
+        self.assertTrue('Customer environment error (expired SSL certs)' in str(context.exception))
+        self.assertTrue("sudo yum update -y --disablerepo='*' --enablerepo='*microsoft*'" in str(context.exception))
+
+        # Restore mock
+        self.runtime.env_layer.platform.linux_distribution = original_env_layer_platform_linux_distribution
+
+        self.runtime.stop()
+
+    def test_apply_fix_ssl_cert_issue_non_rhel_img_failed(self):
+        """ Test ssl expired cert issue throw exception on non rhel images."""
+        # Set up and mocks
+        self.runtime.set_legacy_test_type('SSLCertificateIssueType1SadPathAfterFix')
+        original_env_layer_platform_linux_distribution = self.runtime.env_layer.platform.linux_distribution
+        self.runtime.env_layer.platform.linux_distribution = self.mock_ubuntu_distribution
+        package_manager = self.runtime.container.get('package_manager')
+        self.assertIsNotNone(package_manager)
+
+        # Act
+        with self.assertRaises(Exception) as context:
+            package_manager.fix_ssl_certificate_issue()
+
+        # Verify
+        self.assertTrue('Customer environment error (expired SSL certs)' in str(context.exception))
+
+        # Restore mock
+        self.runtime.env_layer.platform.extract_linux_distribution_os_info = original_env_layer_platform_linux_distribution
+        self.runtime.stop()
+
 
 if __name__ == '__main__':
     unittest.main()