Feature: [Ubuntu, no-CVM] CVM machines to not get UEFI and kek certificate updates

src/core/src/bootstrap/Constants.py

@@ -59,6 +59,7 @@ class SystemPaths(EnumBackport):
     class AzGPSPaths(EnumBackport):
         EULA_SETTINGS = "/var/lib/azure/linuxpatchextension/patch.eula.settings"
         UEFI_SETTINGS = "/var/lib/azure/linuxpatchextension/patch.uefi.settings"
+        DETECT_CVM = "/var/lib/azure/linuxpatchextension/patch.detectcvm.sh"
 
     class EnvSettings(EnumBackport):
         LOG_FOLDER = "logFolder"

src/core/src/bootstrap/EnvLayer.py

@@ -192,6 +192,140 @@ def get_env_var(self, var_name, raise_if_not_success=False):
                 raise
             return None
 
+    def detect_confidential_vm(self):
+        # type: () -> tuple
+        """Returns whether the current VM is a Confidential VM and the detection details."""
+        if self.platform.os_type() == 'Windows':
+            return False, str()
+
+        is_confidential_vm, detection_details = self.detect_confidential_vm_by_imds()
+        if is_confidential_vm:
+            return True, detection_details
+
+        is_confidential_vm, detection_details = self.detect_confidential_vm_by_fde()
+        if is_confidential_vm:
+            return True, detection_details
+
+        return False, str()
+
+    def detect_confidential_vm_by_fde(self):
+        # type: () -> tuple
+        """Runs the FDE-based CVM detection script and returns whether it detected a Confidential VM."""
+        script_path = Constants.AzGPSPaths.DETECT_CVM
+        command_output = str()
+        detection_script = """#!/usr/bin/env bash
+set -euo pipefail
+
+HOSTNAME=$(hostname)
+
+ROOT_SRC=$(findmnt -n -o SOURCE /)
+ROOT_DEV=$(readlink -f "$ROOT_SRC" || echo "$ROOT_SRC")
+
+FDE="false"
+DETAILS=""
+
+check_device() {
+   local dev="$1"
+
+   if blkid "$dev" 2>/dev/null | grep -qi 'crypto_LUKS'; then
+       FDE="true"
+       DETAILS="LUKS:$dev"
+       return
+   fi
+
+   local type
+   type=$(lsblk -dn -o TYPE "$dev" 2>/dev/null || true)
+
+   if [[ "$type" == "crypt" ]]; then
+       FDE="true"
+       DETAILS="CRYPT:$dev"
+       return
+   fi
+}
+
+walk_parents() {
+   local dev="$1"
+
+   while [[ -n "$dev" ]]; do
+       check_device "$dev"
+
+       if [[ "$FDE" == "true" ]]; then
+           return
+       fi
+
+       local parent
+       parent=$(lsblk -ndo PKNAME "$dev" 2>/dev/null | head -1 || true)
+
+       if [[ -z "$parent" ]]; then
+           break
+       fi
+
+       dev="/dev/$parent"
+   done
+}
+
+walk_parents "$ROOT_DEV"
+
+if [[ "$FDE" != "true" ]]; then
+   while read -r name type; do
+       if [[ "$type" == "crypt" ]]; then
+           mapper="/dev/mapper/$name"
+
+           if mount | grep -q "^$mapper on / "; then
+               FDE="true"
+               DETAILS="DMCRYPT_ROOT:$mapper"
+               break
+           fi
+       fi
+   done < <(dmsetup ls --target crypt 2>/dev/null || true)
+fi
+
+if [[ "$FDE" != "true" ]]; then
+   if systemctl list-units 2>/dev/null | grep -qi azure; then
+       if ls /var/lib/waagent/*Encryption* >/dev/null 2>&1; then
+           FDE="true"
+           DETAILS="AZURE_ADE_ARTIFACTS"
+       fi
+   fi
+fi
+
+echo "$HOSTNAME,$ROOT_DEV,FDE=$FDE,$DETAILS"
+"""
+
+        try:
+            script_dir = os.path.dirname(script_path)
+            if script_dir and not os.path.isdir(script_dir):
+                try:
+                    os.makedirs(script_dir)
+                except Exception:
+                    raise
+
+            self.file_system.write_with_retry(script_path, detection_script, 'w')
+
+            code, out = self.run_command_output('bash "{0}"'.format(script_path), False, False)
+            command_output = str(out).strip() if out is not None else str()
+            return code == 0 and re.search(r'\bFDE\s*=\s*true\b', command_output, re.IGNORECASE) is not None, command_output
+        except Exception:
+            raise Exception("FDE_DETECTION_ERROR:{0}; OUTPUT:{1}".format(str(e), command_output))
+        finally:
+            if script_path is not None and os.path.exists(script_path):
+                try:
+                    os.remove(script_path)
+                except Exception:
+                    pass
+
+    def detect_confidential_vm_by_imds(self):
+        # type: () -> tuple
+        """Queries Azure IMDS and returns whether the VM reports ConfidentialVM security type."""
+        command = 'curl -s --connect-timeout 2 --max-time 2 -H Metadata:true --noproxy "*" "http://169.254.169.254/metadata/instance?api-version=2025-04-07"'
+
+        code, out = self.run_command_output(command, False, False)
+        command_output = str(out).strip() if out is not None else str()
+        if code == 0 and re.search(r'"securityType"\s*:\s*"ConfidentialVM"', command_output, re.IGNORECASE) is not None:
+            return True, 'IMDS:ConfidentialVM'
+
+        return False, str()
+
     def run_command_output(self, cmd, no_output=False, chk_err=True):
         # type: (str, bool, bool) -> (int, any)
         """ Wrapper for subprocess.check_output. Execute 'cmd'. Returns return code and STDOUT, trapping expected exceptions. Reports exceptions to Error if chk_err parameter is True """

src/core/src/core_logic/PatchInstaller.py

@@ -808,6 +808,16 @@ def try_update_certificates_for_default_patching(self):
             self.composite_logger.log_debug("Not updating certificates since this is not a default patching operation.")
             return
 
+        try:
+            is_confidential_vm, detection_details = self.env_layer.detect_confidential_vm()
+        except Exception as e:
+            self.composite_logger.log_warning("Unable to determine whether the VM is a Confidential VM before attempting the UEFI certificate update. Continuing with patch installation... [Error: {0}]".format(str(e)))
+            return
+
+        if is_confidential_vm:
+            self.composite_logger.log("Skipping UEFI certificate update because this VM was detected as a Confidential VM. [Detection={0}]".format(detection_details))
+            return
+
         try:
             self.package_manager.update_certs()
         except Exception as e:

src/core/tests/Test_EnvLayer.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
-import io
+import os
 import platform
 import sys
 import unittest
@@ -85,6 +85,42 @@ def mock_linux_distribution_to_return_rhel_10(self):
 
     def mock_distro_os_release_attr_return_rhel_10(self, attribute):
         return '10.0'
+
+    def mock_run_command_output_fde_true(self, cmd, no_output=False, chk_err=False):
+        return 0, 'test-vm,/dev/sda1,FDE=true,LUKS:/dev/sda1'
+
+    def mock_run_command_output_fde_false(self, cmd, no_output=False, chk_err=False):
+        return 0, 'test-vm,/dev/sda1,FDE=false,LUKS:/dev/sda1'
+
+    def mock_run_command_output_imds_true(self, cmd, no_output=False, chk_err=False):
+        return 0, '"securityProfile": { "encryptionAtHost": "false", "secureBootEnabled": "false", "securityType": "ConfidentialVM", "virtualTpmEnabled": "false"}'
+
+    def mock_run_command_output_imds_false(self, cmd, no_output=False, chk_err=False):
+        return 0, '{"compute":{"securityProfile":{"securityType":""}}}'
+
+    def mock_run_command_raises_exception(self, cmd, no_output=False, chk_err=False):
+        raise Exception('Test Exception')
+
+    def mock_detect_confidential_vm_by_fde_returns_true(self):
+        return True, 'test-vm,/dev/sda1,FDE=true,LUKS:/dev/sda1'
+
+    def mock_detect_confidential_vm_by_fde_returns_false(self):
+        return False, str()
+
+    def mock_detect_confidential_vm_by_imds_returns_true(self):
+        return True, 'IMDS:ConfidentialVM'
+
+    def mock_detect_confidential_vm_by_imds_returns_false(self):
+        return False, str()
+
+    def mock_os_remove_raises_exeception(self, path):
+        raise Exception('Test Exception')
+
+    def mock_os_makedirs_raises_exeception(self, path):
+        raise Exception('Test Exception')
+
+    def mock_os_path_isdir_returns_false(self, path):
+        return False
     # endregion
 
     def test_get_package_manager(self):
@@ -138,6 +174,86 @@ def test_is_distro_azure_linux_3(self):
         # restore original methods
         distro.os_release_attr = self.backup_envlayer_distro_os_release_attr
 
+    def test_detect_confidential_vm_by_fde(self):
+        backup_detect_cvm_bash_file_path = Constants.AzGPSPaths.DETECT_CVM
+        backup_run_command_output = self.envlayer.run_command_output
+        backup_os_remove = os.remove
+        backup_os_path_isdir = os.path.isdir
+        backup_os_makedirs = os.makedirs
+
+        test_input_output_table = [
+            [self.mock_run_command_output_fde_true, backup_os_remove, backup_os_path_isdir, backup_os_makedirs, False, True, 'FDE=true'],
+            [self.mock_run_command_output_fde_false, backup_os_remove, backup_os_path_isdir, backup_os_makedirs, False, False, str()],
+            [self.mock_run_command_output_fde_true, self.mock_os_remove_raises_exeception, backup_os_path_isdir, backup_os_makedirs, False, True, 'FDE=true'],
+            [self.mock_run_command_output_fde_true, backup_os_remove, self.mock_os_path_isdir_returns_false, self.mock_os_makedirs_raises_exeception, True, False, str()],
+            [self.mock_run_command_output_fde_true, self.mock_os_remove_raises_exeception, self.mock_os_path_isdir_returns_false, self.mock_os_makedirs_raises_exeception, True, False, str()],
+        ]
+
+        Constants.AzGPSPaths.DETECT_CVM = os.path.join(os.getcwd(), 'patch.detectcvm.sh')
+        for row in test_input_output_table:
+            self.envlayer.run_command_output = row[0]
+            os.remove = row[1]
+            os.path.isdir = row[2]
+            os.makedirs = row[3]
+            expected_raises_exception = row[4]
+            expected_is_confidential_vm = row[5]
+            expected_detection_details = row[6]
+
+            if expected_raises_exception:
+                self.assertRaises(Exception, self.envlayer.detect_confidential_vm_by_fde)
+            else:
+                is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm_by_fde()
+                self.assertEqual(is_confidential_vm, expected_is_confidential_vm)
+                self.assertIn(expected_detection_details, detection_details)
+
+        self.envlayer.run_command_output = backup_run_command_output
+        os.remove = backup_os_remove
+        os.path.isdir = backup_os_path_isdir
+        os.makedirs = backup_os_makedirs
+        Constants.AzGPSPaths.DETECT_CVM = backup_detect_cvm_bash_file_path
+
+    def test_detect_confidential_vm_by_imds(self):
+        backup_run_command_output = self.envlayer.run_command_output
+
+        test_input_output_table = [
+            [self.mock_run_command_output_imds_true, True, 'IMDS:ConfidentialVM'],
+            [self.mock_run_command_output_imds_false, False, str()],
+        ]
+
+        for row in test_input_output_table:
+            self.envlayer.run_command_output = row[0]
+            is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm_by_imds()
+            self.assertEqual(is_confidential_vm, row[1])
+            self.assertIn(row[2], detection_details)
+
+        self.envlayer.run_command_output = backup_run_command_output
+
+    def test_detect_confidential_vm(self):
+        self.backup_platform_system = platform.system
+
+        backup_detect_confidential_vm_by_fde = self.envlayer.detect_confidential_vm_by_fde
+        backup_detect_confidential_vm_by_imds = self.envlayer.detect_confidential_vm_by_imds
+
+        test_input_output_table = [
+            ["Linux", self.mock_detect_confidential_vm_by_fde_returns_true, self.mock_detect_confidential_vm_by_imds_returns_true, True, 'IMDS:ConfidentialVM'],
+            ["Linux", self.mock_detect_confidential_vm_by_fde_returns_true, self.mock_detect_confidential_vm_by_imds_returns_false, True, 'FDE=true'],
+            ["Windows", self.mock_run_command_output_fde_true, self.mock_run_command_output_imds_true, False, str()],
+            ["Linux", self.mock_detect_confidential_vm_by_fde_returns_false, self.mock_detect_confidential_vm_by_imds_returns_false, False, str()],
+        ]
+
+        for row in test_input_output_table:
+            platform.system = self.mock_platform_system if row[0] == 'Linux' else self.mock_platform_system_windows
+            self.envlayer.detect_confidential_vm_by_fde = row[1]
+            self.envlayer.detect_confidential_vm_by_imds = row[2]
+            is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm()
+            self.assertEqual(is_confidential_vm, row[3])
+            self.assertIn(row[4], detection_details)
+
+        # restore original methods
+        platform.system = self.backup_platform_system
+        self.envlayer.detect_confidential_vm_by_fde = backup_detect_confidential_vm_by_fde
+        self.envlayer.detect_confidential_vm_by_imds = backup_detect_confidential_vm_by_imds
+
     def test_filesystem(self):
         # only validates if these invocable without exceptions
         backup_retry_count = Constants.MAX_FILE_OPERATION_RETRY_COUNT

src/core/tests/Test_PatchInstaller.py

@@ -35,6 +35,12 @@ def tearDown(self):
     # region Mocks
     def mock_update_certs_raise_exception(self):
         raise Exception("Simulated cert update failure")
+
+    def mock_detect_confidential_vm_raises_exception(self):
+        raise Exception("Simulated VM detection failure")
+
+    def mock_detect_confidential_vm_by_imds_returns_true(self):
+        return True, 'IMDS:ConfidentialVM'
     # endregion
 
     # region Utility functions (update cert tests)
@@ -816,6 +822,30 @@ def test_try_update_certs_swallows_exception_from_update_certs(self):
 
         runtime.patch_installer.package_manager.update_certs = backup_up_update_certs
         runtime.stop()
+
+    def test_try_update_certificates_skips_confidential_vm(self):
+        runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01")
+        backup_detect_confidential_vm_by_imds = runtime.env_layer.detect_confidential_vm_by_imds
+
+        runtime.env_layer.detect_confidential_vm_by_imds = self.mock_detect_confidential_vm_by_imds_returns_true
+        method_called = self._track_method_call(runtime.patch_installer.package_manager, 'update_certs')
+        runtime.patch_installer.start_installation(simulate=True)
+        self.assertEqual(len(method_called), 0)
+
+        runtime.env_layer.detect_confidential_vm_by_imds = backup_detect_confidential_vm_by_imds
+        runtime.stop()
+
+    def test_try_update_certificates_skips_when_detect_confidential_vm_raises_exception(self):
+        runtime = self._create_update_certs_runtime(enable_uefi_cert_update=True, health_store_id="pub_off_sku_2025.01.01")
+        backup_detect_confidential_vm = runtime.env_layer.detect_confidential_vm
+
+        runtime.env_layer.detect_confidential_vm = self.mock_detect_confidential_vm_raises_exception
+        method_called = self._track_method_call(runtime.patch_installer.package_manager, 'update_certs')
+        runtime.patch_installer.start_installation(simulate=True)
+        self.assertEqual(len(method_called), 0)
+
+        runtime.env_layer.detect_confidential_vm = backup_detect_confidential_vm
+        runtime.stop()
     # endregion