Safety net: Explicitly block Azure Linux 4 and RHEL 10 until LPE support is available

src/core/src/bootstrap/Constants.py

@@ -197,7 +197,7 @@ class StatusTruncationConfig(EnumBackport):
     RED_HAT = 'Red Hat'
     SUSE = 'SUSE'
     CENTOS = 'CentOS'
-    AZURE_LINUX = ['Microsoft Azure Linux', 'Common Base Linux Mariner']
+    AZURE_LINUX = ['Azure Linux', 'Microsoft Azure Linux', 'Common Base Linux Mariner']
 
     # Package Managers
     APT = 'apt'

src/core/src/bootstrap/EnvLayer.py

@@ -47,22 +47,58 @@ def __init__(self):
     def is_distro_azure_linux(distro_name):
         return any(x in distro_name for x in Constants.AZURE_LINUX)
 
-    def is_distro_azure_linux_3_or_beyond(self):
-        # type: () -> bool
+    @staticmethod
+    def __try_get_major_version():
+        # type: () -> str or None
+        """ Attempts to get major version from distro.os_release_attr('version').Returns major version as string or None if not available. """
+        version = distro.os_release_attr('version')
+        major = version.split('.')[0] if version else None
+        return major
+
+    def __is_matching_distro_and_version(self, distro_name, distro_to_match, version_to_match):
+        # type: (str, str, int) -> bool
+        """ Checks if distro name matches and version matches the expected version.
+        Returns True if both distro and version match, False otherwise. """
+        if distro_to_match == Constants.AZURE_LINUX and not self.is_distro_azure_linux(str(distro_name)):
+            return False
+        elif distro_to_match == Constants.RED_HAT and not Constants.RED_HAT.lower() in str(distro_name).lower():
+            return False
+        major = self.__try_get_major_version()
+        return major is not None and int(major) == version_to_match
+
+    def is_distro_azure_linux_3(self, distro_name):
+        # type: (str) -> bool
         """ Checks if the current distro is Azure Linux 3 """
-        if self.is_distro_azure_linux(self.platform.linux_distribution()):
-            version = distro.os_release_attr('version')
-            major = version.split('.')[0] if version else None
-            return major is not None and int(major) >= 3
-        return False
+        return self.__is_matching_distro_and_version(distro_name, Constants.AZURE_LINUX, version_to_match=3)
+
+    def is_distro_azure_linux_4(self, distro_name):
+        # type: (str) -> bool
+        """ Checks if the current distro is Azure Linux 4 """
+        return self.__is_matching_distro_and_version(distro_name, Constants.AZURE_LINUX, version_to_match=4)
+
+    def is_distro_rhel_10(self, distro_name):
+        # type: (str) -> bool
+        """ Checks if the current distro is RHEL 10 """
+        return self.__is_matching_distro_and_version(distro_name, Constants.RED_HAT, version_to_match=10)
 
     def get_package_manager(self):
         # type: () -> str
         """ Detects package manager type """
         if platform.system() == 'Windows':
             return Constants.APT
 
-        if self.is_distro_azure_linux(str(self.platform.linux_distribution())):
+        # platform.linux_distribution() returns a tuple with distro_name, version and a code
+        # Example: ['Azure Linux', '4.0', '']
+        os_name, os_version, os_code = self.platform.linux_distribution()
+
+        # Check for unsupported distros
+        if self.is_distro_azure_linux_4(os_name) or self.is_distro_rhel_10(os_name):
+            error_msg = "This distro is not yet supported in your region. Please review https://aka.ms/VMGuestPatchingCompatibility for more information. [Distro={0}][Version={1}][Code={2}]".format(str(os_name), os_version, os_code)
+            print("Error: {0}".format(error_msg))
+            return str()
+
+        # Check for Azure Linux (3 and below use TDNF)
+        if self.is_distro_azure_linux(str(os_name)):
             code, out = self.run_command_output('which tdnf', False, False)
             if code == 0:
                 return Constants.TDNF

src/core/src/package_managers/AzL3TdnfPackageManager.py

@@ -71,7 +71,8 @@ def try_meet_azgps_coordinated_requirements(self):
         """ Check if the system meets the requirements for Azure Linux strict safe deployment and attempt to update TDNF if necessary """
         self.composite_logger.log_debug("[AzL3TDNF] Checking if system meets Azure Linux security updates requirements...")
         # Check if the system is Azure Linux 3.0 or beyond
-        if not self.env_layer.is_distro_azure_linux_3_or_beyond():
+        distro_name = str(self.env_layer.platform.linux_distribution()[0])
+        if not self.env_layer.is_distro_azure_linux_3(distro_name):
             self.composite_logger.log_error("[AzL3TDNF] The system does not meet minimum Azure Linux requirement of 3.0 or above for strict safe deployment. Defaulting to regular upgrades.")
             self.set_max_patch_publish_date()  # fall-back
             return False

src/core/tests/Test_EnvLayer.py

@@ -13,7 +13,9 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
+import io
 import platform
+import sys
 import unittest
 from core.src.bootstrap.EnvLayer import EnvLayer
 from core.src.bootstrap.Constants import Constants
@@ -71,6 +73,18 @@ def mock_distro_os_release_attr_return_azure_linux_2(self, attribute):
 
     def mock_distro_os_release_attr_return_none(self, attribute):
         return None
+
+    def mock_linux_distribution_to_return_azure_linux_4(self):
+        return ['Microsoft Azure Linux', '4.0', '']
+
+    def mock_distro_os_release_attr_return_azure_linux_4(self, attribute):
+        return '4.0.2'
+
+    def mock_linux_distribution_to_return_rhel_10(self):
+        return ['Red Hat', '10.0', 'abc']
+
+    def mock_distro_os_release_attr_return_rhel_10(self, attribute):
+        return '10.0'
     # endregion
 
     def test_get_package_manager(self):
@@ -79,6 +93,7 @@ def test_get_package_manager(self):
         self.backup_linux_distribution = self.envlayer.platform.linux_distribution
         self.envlayer.platform.linux_distribution = self.mock_linux_distribution
         self.backup_run_command_output = self.envlayer.run_command_output
+        self.backup_distro_os_release_attr = distro.os_release_attr
 
         test_input_output_table = [
             [self.mock_run_command_for_apt, self.mock_linux_distribution, Constants.APT],
@@ -105,8 +120,7 @@ def test_get_package_manager(self):
         self.envlayer.platform.linux_distribution = self.backup_linux_distribution
         platform.system = self.backup_platform_system
 
-    def test_is_distro_azure_linux_3_or_beyond(self):
-        self.backup_linux_distribution = self.envlayer.platform.linux_distribution
+    def test_is_distro_azure_linux_3(self):
         self.backup_envlayer_distro_os_release_attr = distro.os_release_attr
 
         test_input_output_table = [
@@ -116,14 +130,13 @@ def test_is_distro_azure_linux_3_or_beyond(self):
         ]
 
         for row in test_input_output_table:
-            self.envlayer.platform.linux_distribution = row[0]
+            distro_name = row[0]()[0]  # Extract distro name from tuple (first element)
             distro.os_release_attr = row[1]
-            result = self.envlayer.is_distro_azure_linux_3_or_beyond()
+            result = self.envlayer.is_distro_azure_linux_3(distro_name)
             self.assertEqual(result, row[2])
 
         # restore original methods
         distro.os_release_attr = self.backup_envlayer_distro_os_release_attr
-        self.envlayer.platform.linux_distribution = self.backup_linux_distribution
 
     def test_filesystem(self):
         # only validates if these invocable without exceptions
@@ -139,6 +152,37 @@ def test_platform(self):
         self.envlayer.platform.cpu_arch()
         self.envlayer.platform.vm_name()
 
+    def test_get_package_manager_azure_linux_4_and_rhel10_not_supported(self):
+        """Test that Azure Linux 4 and RHEL 10 log unsupported message"""
+        self.backup_platform_system = platform.system
+        self.backup_linux_distribution = self.envlayer.platform.linux_distribution
+        self.backup_distro_os_release_attr = distro.os_release_attr
+
+        platform.system = self.mock_platform_system
+        test_input_output_table = [
+            [self.mock_linux_distribution_to_return_azure_linux_4, self.mock_distro_os_release_attr_return_azure_linux_4, "Error: This distro is not yet supported in your region. Please review https://aka.ms/VMGuestPatchingCompatibility for more information. [Distro=Microsoft Azure Linux][Version=4.0][Code=]\n"],
+            [self.mock_linux_distribution_to_return_rhel_10, self.mock_distro_os_release_attr_return_rhel_10, "Error: This distro is not yet supported in your region. Please review https://aka.ms/VMGuestPatchingCompatibility for more information. [Distro=Red Hat][Version=10.0][Code=abc]\n"],
+        ]
+
+        for row in test_input_output_table:
+            self.envlayer.platform.linux_distribution = row[0]
+            distro.os_release_attr = row[1]
+
+            captured_output = io.StringIO()
+            sys.stdout = captured_output
+            result = self.envlayer.get_package_manager()
+            sys.stdout = sys.__stdout__
+            self.assertEqual(row[2], captured_output.getvalue())
+            self.assertEqual(result, "")
+
+        # restore
+        self.__restore_mocks()
+
+    def __restore_mocks(self):
+        """Restore backed up mocks to their original state"""
+        distro.os_release_attr = self.backup_distro_os_release_attr
+        self.envlayer.platform.linux_distribution = self.backup_linux_distribution
+        platform.system = self.backup_platform_system
 
 if __name__ == '__main__':
     unittest.main()