Add backend support for Apple Silicon root patching

opencore_legacy_patcher/sys_patch/sys_patch.py

@@ -56,7 +56,8 @@ from . import (
     sys_patch_detect,
     sys_patch_auto,
     sys_patch_helpers,
-    sys_patch_generate
+    sys_patch_generate,
+    sys_patch_mount
 )
 
 
@@ -65,7 +66,6 @@ class PatchSysVolume:
         self.model = model
         self.constants: constants.Constants = global_constants
         self.computer = self.constants.computer
-        self.root_mount_path = None
         self.root_supports_snapshot = utilities.check_if_root_is_apfs_snapshot()
         self.constants.root_patcher_succeeded = False # Reset Variable each time we start
         self.constants.needs_to_open_preferences = False
@@ -82,6 +82,9 @@ class PatchSysVolume:
 
         self.skip_root_kmutil_requirement = self.hardware_details["Settings: Supports Auxiliary Cache"]
 
+        self.mount_obj = sys_patch_mount.SysPatchMount(self.constants.detected_os, self.computer.rosetta_active)
+
+
     def _init_pathing(self, custom_root_mount_path: Path = None, custom_data_mount_path: Path = None) -> None:
         """
         Initializes the pathing for root volume patching
@@ -107,41 +110,23 @@ class PatchSysVolume:
 
     def _mount_root_vol(self) -> bool:
         """
-        Attempts to mount the booted APFS volume as a writable volume
+        Mount root volume
-        at /System/Volumes/Update/mnt1
-
-        Manual invocation:
-            'sudo mount -o nobrowse -t apfs  /dev/diskXsY /System/Volumes/Update/mnt1'
-
-        Returns:
-            bool: True if successful, False if not
         """
+        if self.mount_obj.mount():
+            return True
 
-        # Returns boolean if Root Volume is available
-        self.root_mount_path = utilities.get_disk_path()
-        if self.root_mount_path.startswith("disk"):
-            logging.info(f"- Found Root Volume at: {self.root_mount_path}")
-            if Path(self.mount_extensions).exists():
-                logging.info("- Root Volume is already mounted")
-                return True
-            else:
-                if self.root_supports_snapshot is True:
-                    logging.info("- Mounting APFS Snapshot as writable")
-                    result = subprocess_wrapper.run_as_root(["/sbin/mount", "-o", "nobrowse", "-t", "apfs", f"/dev/{self.root_mount_path}", self.mount_location], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-                    if result.returncode == 0:
-                        logging.info(f"- Mounted APFS Snapshot as writable at: {self.mount_location}")
-                        if Path(self.mount_extensions).exists():
-                            logging.info("- Successfully mounted the Root Volume")
-                            return True
-                        else:
-                            logging.info("- Root Volume appears to have unmounted unexpectedly")
-                    else:
-                        logging.info("- Unable to mount APFS Snapshot as writable")
-                        subprocess_wrapper.log(result)
         return False
 
 
-    def _merge_kdk_with_root(self, save_hid_cs=False) -> None:
+    def _unmount_root_vol(self) -> None:
+        """
+        Unmount root volume
+        """
+        logging.info("- Unmounting root volume")
+        self.mount_obj.unmount(ignore_errors=True)
+
+
+    def _merge_kdk_with_root(self, save_hid_cs: bool = False) -> None:
         """
         Merge Kernel Debug Kit (KDK) with the root volume
         If no KDK is present, will call kdk_handler to download and install it
@@ -252,23 +237,15 @@ class PatchSysVolume:
         """
         Reverts APFS snapshot and cleans up any changes made to the root and data volume
         """
+        if self.mount_obj.revert_snapshot() is False:
+            return
 
-        if self.constants.detected_os <= os_data.os_data.big_sur or self.root_supports_snapshot is False:
+        self._clean_skylight_plugins()
-            logging.info("- OS version does not support snapshotting, skipping revert")
+        self._delete_nonmetal_enforcement()
-
+        self._clean_auxiliary_kc()
-        logging.info("- Reverting to last signed APFS snapshot")
+        self.constants.root_patcher_succeeded = True
-        result = subprocess_wrapper.run_as_root(["/usr/sbin/bless", "--mount", self.mount_location, "--bootefi", "--last-sealed-snapshot"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        logging.info("- Unpatching complete")
-        if result.returncode != 0:
+        logging.info("\nPlease reboot the machine for patches to take effect")
-            logging.info("- Unable to revert root volume patches")
-            subprocess_wrapper.log(result)
-            logging.info("- Failed to revert snapshot via Apple's 'bless' command")
-        else:
-            self._clean_skylight_plugins()
-            self._delete_nonmetal_enforcement()
-            self._clean_auxiliary_kc()
-            self.constants.root_patcher_succeeded = True
-            logging.info("- Unpatching complete")
-            logging.info("\nPlease reboot the machine for patches to take effect")
 
 
     def _rebuild_root_volume(self) -> bool:
@@ -287,6 +264,7 @@ class PatchSysVolume:
             self._update_preboot_kernel_cache()
             self._rebuild_dyld_shared_cache()
             if self._create_new_apfs_snapshot() is True:
+                self._unmount_root_vol()
                 logging.info("- Patching complete")
                 logging.info("\nPlease reboot the machine for patches to take effect")
                 if self.needs_kmutil_exemptions is True:
@@ -405,35 +383,7 @@ class PatchSysVolume:
         Returns:
             bool: True if snapshot was created, False if not
         """
-
+        return self.mount_obj.create_snapshot()
-        if self.root_supports_snapshot is True:
-            logging.info("- Creating new APFS snapshot")
-            bless = subprocess_wrapper.run_as_root(
-                [
-                    "/usr/sbin/bless",
-                    "--folder", f"{self.mount_location}/System/Library/CoreServices",
-                    "--bootefi", "--create-snapshot"
-                ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT
-            )
-            if bless.returncode != 0:
-                logging.info("- Unable to create new snapshot")
-                subprocess_wrapper.log(bless)
-                if "Can't use last-sealed-snapshot or create-snapshot on non system volume" in bless.stdout.decode():
-                    logging.info("- This is an APFS bug with Monterey and newer! Perform a clean installation to ensure your APFS volume is built correctly")
-                return False
-            self._unmount_drive()
-        return True
-
-
-    def _unmount_drive(self) -> None:
-        """
-        Unmount root volume
-        """
-        if self.root_mount_path:
-            logging.info("- Unmounting Root Volume (Don't worry if this fails)")
-            subprocess_wrapper.run_as_root(["/usr/sbin/diskutil", "unmount", self.root_mount_path], stdout=subprocess.PIPE)
-        else:
-            logging.info("- Skipping Root Volume unmount")
 
 
     def _rebuild_dyld_shared_cache(self) -> None:

opencore_legacy_patcher/sys_patch/sys_patch_mount.py

@@ -0,0 +1,173 @@
+"""
+sys_patch_mount.py: Handling macOS root volume mounting and unmounting,
+                    as well as APFS snapshots for Big Sur and newer
+"""
+
+import logging
+import plistlib
+import platform
+import subprocess
+
+from pathlib import Path
+
+from ..datasets import os_data
+from ..support  import subprocess_wrapper
+
+
+class SysPatchMount:
+
+    def __init__(self, xnu_major: int, rosetta_status: bool) -> None:
+        self.xnu_major = xnu_major
+        self.rosetta_status = rosetta_status
+        self.root_volume_identifier = self._fetch_root_volume_identifier()
+
+        self.mount_path = None
+
+
+    def mount(self) -> str:
+        """
+        Mount the root volume.
+
+        Returns the path to the root volume.
+
+        If none, failed to mount.
+        """
+        result = self._mount_root_volume()
+        if not Path(result).exists():
+            logging.error(f"Attempted to mount root volume, but failed: {result}")
+            return None
+
+        self.mount_path = result
+
+        return result
+
+
+    def unmount(self, ignore_errors: bool = True) -> bool:
+        """
+        Unmount the root volume.
+
+        Returns True if successful, False otherwise.
+
+        Note for Big Sur and newer, a snapshot is created before unmounting.
+        And that unmounting is not critical to the process.
+        """
+        return self._unmount_root_volume(ignore_errors=ignore_errors)
+
+
+    def _fetch_root_volume_identifier(self) -> str:
+        """
+        Resolve path to disk identifier
+
+        ex. / -> disk1s1
+        """
+        try:
+            content = plistlib.loads(subprocess.run(["/usr/sbin/diskutil", "info", "-plist", "/"], capture_output=True).stdout)
+        except plistlib.InvalidFileException:
+            raise RuntimeError("Failed to parse diskutil output.")
+
+        disk = content["DeviceIdentifier"]
+
+        if "APFSSnapshot" in content and content["APFSSnapshot"] is True:
+            # Remove snapshot suffix (last 2 characters)
+            # ex. disk1s1s1 -> disk1s1
+            disk = disk[:-2]
+
+        return disk
+
+
+    def _mount_root_volume(self) -> str:
+        """
+        Mount the root volume.
+
+        Returns the path to the root volume.
+        """
+        # Root volume same as data volume
+        if self.xnu_major < os_data.os_data.catalina.value:
+            return "/"
+
+        # Catalina implemented a read-only root volume
+        if self.xnu_major == os_data.os_data.catalina.value:
+            result = subprocess_wrapper.run_as_root(["/sbin/mount", "-uw", "/"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            if result.returncode != 0:
+                logging.error("Failed to mount root volume")
+                subprocess_wrapper.log(result)
+                return None
+            return "/"
+
+        # Big Sur and newer implemented APFS snapshots for the root volume
+        if self.xnu_major >= os_data.os_data.big_sur.value:
+            if Path("/System/Volumes/Update/mnt1/System/Library/CoreServices/SystemVersion.plist").exists():
+                return "/System/Volumes/Update/mnt1"
+            result = subprocess_wrapper.run_as_root(["/sbin/mount", "-o", "nobrowse", "-t", "apfs", f"/dev/{self.root_volume_identifier}", "/System/Volumes/Update/mnt1"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            if result.returncode != 0:
+                logging.error("Failed to mount root volume")
+                subprocess_wrapper.log(result)
+                return None
+            return "/System/Volumes/Update/mnt1"
+
+
+    def _unmount_root_volume(self, ignore_errors: bool = True) -> bool:
+        """
+        Unmount the root volume.
+
+        If applicable, create a snapshot before unmounting.
+        """
+        if self.xnu_major < os_data.os_data.catalina.value:
+            return True
+
+        args = ["/sbin/umount"]
+
+        if self.xnu_major == os_data.os_data.catalina.value:
+            args += ["-uw", "/"]
+
+        if self.xnu_major >= os_data.os_data.big_sur.value:
+            args += ["/System/Volumes/Update/mnt1"]
+
+        result = subprocess_wrapper.run_as_root(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        if result.returncode != 0:
+            if ignore_errors is False:
+                logging.error("Failed to unmount root volume")
+                subprocess_wrapper.log(result)
+            return False
+
+
+    def create_snapshot(self) -> bool:
+        """
+        Create APFS snapshot of the root volume.
+        """
+        if self.xnu_major < os_data.os_data.big_sur.value:
+            return True
+
+        args = ["/usr/sbin/bless"]
+        if platform.machine() == "arm64" or self.rosetta_status is True:
+            args += ["--mount", "/System/Volumes/Update/mnt1", "--create-snapshot"]
+        else:
+            args += ["--folder", "/System/Volumes/Update/mnt1/System/Library/CoreServices", "--bootefi", "--create-snapshot"]
+
+
+        result = subprocess_wrapper.run_as_root(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        if result.returncode != 0:
+            logging.error("Failed to create APFS snapshot")
+            subprocess_wrapper.log(result)
+            if "Can't use last-sealed-snapshot or create-snapshot on non system volume" in result.stdout.decode():
+                logging.info("- This is an APFS bug with Monterey and newer! Perform a clean installation to ensure your APFS volume is built correctly")
+
+            return False
+
+        return True
+
+
+    def revert_snapshot(self) -> bool:
+        """
+        Revert APFS snapshot of the root volume.
+        """
+        if self.xnu_major < os_data.os_data.big_sur.value:
+            return True
+
+        result = subprocess_wrapper.run_as_root(["/usr/sbin/bless", "--mount", self.mount_location, "--bootefi", "--last-sealed-snapshot"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        if result.returncode != 0:
+            logging.error("Failed to revert APFS snapshot")
+            subprocess_wrapper.log(result)
+            return False
+
+        return True