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,144 @@ 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 OSError:
+                    if not os.path.isdir(script_dir):
+                        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:
+            return False, 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"'
+
+        try:
+            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'
+        except Exception:
+            pass
+
+        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,18 @@ 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:
+            is_confidential_vm = False
+            detection_details = str()
+            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,6 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
-import io
 import platform
 import sys
 import unittest
@@ -138,6 +137,88 @@ 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_run_command_output = self.envlayer.run_command_output
+
+        self.envlayer.run_command_output = lambda cmd, no_output=False, chk_err=False: (0, 'test-vm,/dev/sda1,FDE=true,LUKS:/dev/sda1')
+        is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm_by_fde()
+
+        self.assertTrue(is_confidential_vm)
+        self.assertIn('FDE=true', detection_details)
+
+        self.envlayer.run_command_output = backup_run_command_output
+
+    def test_detect_confidential_vm_by_imds(self):
+        backup_run_command_output = self.envlayer.run_command_output
+
+        self.envlayer.run_command_output = lambda cmd, no_output=False, chk_err=False: (0, '{"compute":{"securityProfile":{"securityType":"ConfidentialVM"}}}')
+        is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm_by_imds()
+
+        self.assertTrue(is_confidential_vm)
+        self.assertEqual('IMDS:ConfidentialVM', detection_details)
+
+        self.envlayer.run_command_output = backup_run_command_output
+
+    def test_detect_confidential_vm_checks_imds_before_fde(self):
+        backup_platform = self.envlayer.platform
+        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
+
+        calls = []
+        self.envlayer.platform = self.envlayer.Platform()
+        self.envlayer.platform.os_type = lambda: 'Linux'
+
+        def detect_confidential_vm_by_fde():
+            calls.append('fde')
+            return True, 'test-vm,/dev/sda1,FDE=true,LUKS:/dev/sda1'
+
+        def detect_confidential_vm_by_imds():
+            calls.append('imds')
+            return True, 'IMDS:ConfidentialVM'
+
+        self.envlayer.detect_confidential_vm_by_fde = detect_confidential_vm_by_fde
+        self.envlayer.detect_confidential_vm_by_imds = detect_confidential_vm_by_imds
+
+        is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm()
+
+        self.assertTrue(is_confidential_vm)
+        self.assertIn('IMDS:ConfidentialVM', detection_details)
+        self.assertEqual(['imds'], calls)
+
+        self.envlayer.platform = backup_platform
+        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_detect_confidential_vm_checks_fde_when_imds_not_detected(self):
+        backup_platform = self.envlayer.platform
+        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
+
+        calls = []
+        self.envlayer.platform = self.envlayer.Platform()
+        self.envlayer.platform.os_type = lambda: 'Linux'
+
+        def detect_confidential_vm_by_fde():
+            calls.append('fde')
+            return True, 'test-vm,/dev/sda1,FDE=true,LUKS:/dev/sda1'
+
+        def detect_confidential_vm_by_imds():
+            calls.append('imds')
+            return False, str()
+
+        self.envlayer.detect_confidential_vm_by_fde = detect_confidential_vm_by_fde
+        self.envlayer.detect_confidential_vm_by_imds = detect_confidential_vm_by_imds
+
+        is_confidential_vm, detection_details = self.envlayer.detect_confidential_vm()
+
+        self.assertTrue(is_confidential_vm)
+        self.assertIn('FDE=true', detection_details)
+        self.assertEqual(['imds', 'fde'], calls)
+
+        self.envlayer.platform = backup_platform
+        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

@@ -816,6 +816,16 @@ 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")
+        runtime.patch_installer.env_layer.detect_confidential_vm = lambda: (True, 'IMDS:ConfidentialVM')
+
+        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.stop()
     # endregion