Merge branch 'main' into sequoia-development

CHANGELOG.md

@@ -9,6 +9,9 @@
   - Note `gktool` is only available on macOS Sonoma and newer
 - Resolve unpatching crash edge case when host doesn't require patches.
 - Implement new Software Update Catalog Parser for macOS Installers
+- Implement new Copy on Write detection mechanism for all file copying operations
+  - Implemented using `getattrlist` and `VOL_CAP_INT_CLONE` flag
+  - Helps improve performance on APFS volumes
 
 ## 1.5.0
 - Restructure project directories

ci_tooling/build_modules/application.py

@@ -5,6 +5,7 @@
 
 from pathlib import Path
 
+from opencore_legacy_patcher.volume  import generate_copy_arguments
 from opencore_legacy_patcher.support import subprocess_wrapper
 
 
@@ -157,7 +158,7 @@ def _embed_resources(self) -> None:
         print("Embedding resources")
         for file in Path("payloads/Icon/AppIcons").glob("*.icns"):
             subprocess_wrapper.run_and_verify(
-                ["/bin/cp", str(file), self._application_output / "Contents" / "Resources/"],
+                generate_copy_arguments(str(file), self._application_output / "Contents" / "Resources/"),
                 stdout=subprocess.PIPE, stderr=subprocess.PIPE
             )
 

ci_tooling/build_modules/shim.py

@@ -4,6 +4,7 @@
 
 from pathlib import Path
 
+from opencore_legacy_patcher.volume  import generate_copy_arguments
 from opencore_legacy_patcher.support import subprocess_wrapper
 
 
@@ -25,9 +26,9 @@ def generate(self) -> None:
         if Path(self._shim_pkg).exists():
             Path(self._shim_pkg).unlink()
 
-        subprocess_wrapper.run_and_verify(["/bin/cp", "-R", self._build_pkg, self._shim_pkg])
+        subprocess_wrapper.run_and_verify(generate_copy_arguments(self._build_pkg, self._shim_pkg))
 
         if Path(self._output_shim).exists():
             Path(self._output_shim).unlink()
 
-        subprocess_wrapper.run_and_verify(["/bin/cp", "-R", self._shim_path, self._output_shim])
+        subprocess_wrapper.run_and_verify(generate_copy_arguments(self._shim_path, self._output_shim))

opencore_legacy_patcher/sucatalog/products.py

@@ -232,15 +232,14 @@ def _list_latest_installers_only(self, products: list) -> list:
                     continue
                 try:
                     if packaging.version.parse(installer["Version"]) < _newest_version:
-                        products_copy.pop(products_copy.index(installer))
+                        if installer in products_copy:
+                            products_copy.pop(products_copy.index(installer))
                 except packaging.version.InvalidVersion:
                     pass
 
-        # Remove Betas if there's a non-beta version available
-        for installer in products:
-            if installer["Catalog"] in [SeedType.CustomerSeed, SeedType.DeveloperSeed, SeedType.PublicSeed]:
-                for installer_2 in products:
-                    if installer_2["Version"].split(".")[0] == installer["Version"].split(".")[0] and installer_2["Catalog"] not in [SeedType.CustomerSeed, SeedType.DeveloperSeed, SeedType.PublicSeed]:
+                # Remove beta versions if a public release is available
+                if _newest_version != packaging.version.parse("0.0.0"):
+                    if installer["Catalog"] in [SeedType.CustomerSeed, SeedType.DeveloperSeed, SeedType.PublicSeed]:
                         if installer in products_copy:
                             products_copy.pop(products_copy.index(installer))
 

opencore_legacy_patcher/support/kdk_handler.py

@@ -15,6 +15,7 @@
 from .. import constants
 
 from ..datasets import os_data
+from ..volume   import generate_copy_arguments
 
 from . import (
     network_handler,
@@ -667,7 +668,7 @@ def _create_backup(self, kdk_path: Path, kdk_info_plist: Path) -> None:
             logging.info("Backup already exists, skipping")
             return
 
-        result = subprocess_wrapper.run_as_root(["/bin/cp", "-R", kdk_path, kdk_dst_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        result = subprocess_wrapper.run_as_root(generate_copy_arguments(kdk_path, kdk_dst_path), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
         if result.returncode != 0:
             logging.info("Failed to create KDK backup:")
             subprocess_wrapper.log(result)

opencore_legacy_patcher/support/macos_installer_handler.py

@@ -16,6 +16,11 @@
     subprocess_wrapper
 )
 
+from ..volume import (
+    can_copy_on_write,
+    generate_copy_arguments
+)
+
 
 APPLICATION_SEARCH_PATH:  str = "/Applications"
 SFR_SOFTWARE_UPDATE_PATH: str = "SFR/com_apple_MobileAsset_SFRSoftwareUpdate/com_apple_MobileAsset_SFRSoftwareUpdate.xml"
@@ -90,21 +95,17 @@ def generate_installer_creation_script(self, tmp_location: str, installer_path:
         for file in Path(ia_tmp).glob("*"):
             subprocess.run(["/bin/rm", "-rf", str(file)])
 
-        # Copy installer to tmp (use CoW to avoid extra disk writes)
-        args = ["/bin/cp", "-cR", installer_path, ia_tmp]
-        if utilities.check_filesystem_type() != "apfs":
-            # HFS+ disks do not support CoW
-            args[1] = "-R"
-
-            # Ensure we have enough space for the duplication
+        # Copy installer to tmp
+        if can_copy_on_write(installer_path, ia_tmp) is False:
+            # Ensure we have enough space for the duplication when CoW is not supported
             space_available = utilities.get_free_space()
             space_needed = Path(ia_tmp).stat().st_size
             if space_available < space_needed:
                 logging.info("Not enough free space to create installer.sh")
                 logging.info(f"{utilities.human_fmt(space_available)} available, {utilities.human_fmt(space_needed)} required")
                 return False
 
-        subprocess.run(args)
+        subprocess.run(generate_copy_arguments(installer_path, ia_tmp))
 
         # Adjust installer_path to point to the copied installer
         installer_path = Path(ia_tmp) / Path(Path(installer_path).name)

opencore_legacy_patcher/sys_patch/sys_patch.py

@@ -46,6 +46,7 @@
 from .. import constants
 
 from ..datasets import os_data
+from ..volume   import generate_copy_arguments
 
 from ..support import (
     utilities,
@@ -234,7 +235,7 @@ def _merge_kdk_with_root(self, save_hid_cs: bool = False) -> None:
         if save_hid_cs is True and cs_path.exists():
             logging.info("- Backing up IOHIDEventDriver CodeSignature")
             # Note it's a folder, not a file
-            subprocess_wrapper.run_as_root(["/bin/cp", "-r", cs_path, f"{self.constants.payload_path}/IOHIDEventDriver_CodeSignature.bak"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            subprocess_wrapper.run_as_root(generate_copy_arguments(cs_path, f"{self.constants.payload_path}/IOHIDEventDriver_CodeSignature.bak"), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
         logging.info(f"- Merging KDK with Root Volume: {kdk_path.name}")
         subprocess_wrapper.run_as_root(
@@ -256,7 +257,7 @@ def _merge_kdk_with_root(self, save_hid_cs: bool = False) -> None:
             if not cs_path.exists():
                 logging.info("  - CodeSignature folder missing, creating")
                 subprocess_wrapper.run_as_root(["/bin/mkdir", "-p", cs_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-            subprocess_wrapper.run_as_root(["/bin/cp", "-r", f"{self.constants.payload_path}/IOHIDEventDriver_CodeSignature.bak", cs_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            subprocess_wrapper.run_as_root(generate_copy_arguments(f"{self.constants.payload_path}/IOHIDEventDriver_CodeSignature.bak", cs_path), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             subprocess_wrapper.run_as_root(["/bin/rm", "-rf", f"{self.constants.payload_path}/IOHIDEventDriver_CodeSignature.bak"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
 
@@ -533,7 +534,7 @@ def _write_patchset(self, patchset: dict) -> None:
             logging.info("- Writing patchset information to Root Volume")
             if Path(destination_path_file).exists():
                 subprocess_wrapper.run_as_root_and_verify(["/bin/rm", destination_path_file], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-            subprocess_wrapper.run_as_root_and_verify(["/bin/cp", f"{self.constants.payload_path}/{file_name}", destination_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            subprocess_wrapper.run_as_root_and_verify(generate_copy_arguments(f"{self.constants.payload_path}/{file_name}", destination_path), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
 
     def _add_auxkc_support(self, install_file: str, source_folder_path: str, install_patch_directory: str, destination_folder_path: str) -> str:
@@ -782,7 +783,7 @@ def _install_new_file(self, source_folder: Path, destination_folder: Path, file_
                 subprocess_wrapper.run_as_root_and_verify(["/bin/rm", "-R", f"{destination_folder}/{file_name}"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             else:
                 logging.info(f"  - Installing: {file_name}")
-            subprocess_wrapper.run_as_root_and_verify(["/bin/cp", "-R", f"{source_folder}/{file_name}", destination_folder], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            subprocess_wrapper.run_as_root_and_verify(generate_copy_arguments(f"{source_folder}/{file_name}", destination_folder), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             self._fix_permissions(destination_folder + "/" + file_name)
         else:
             # Assume it's an individual file, replace as normal
@@ -791,7 +792,7 @@ def _install_new_file(self, source_folder: Path, destination_folder: Path, file_
                 subprocess_wrapper.run_as_root_and_verify(["/bin/rm", f"{destination_folder}/{file_name}"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             else:
                 logging.info(f"  - Installing: {file_name}")
-            subprocess_wrapper.run_as_root_and_verify(["/bin/cp", f"{source_folder}/{file_name}", destination_folder], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            subprocess_wrapper.run_as_root_and_verify(generate_copy_arguments(f"{source_folder}/{file_name}", destination_folder), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
             self._fix_permissions(destination_folder + "/" + file_name)
 
 

opencore_legacy_patcher/sys_patch/sys_patch_auto.py

@@ -20,6 +20,7 @@
 from .. import constants
 
 from ..datasets import css_data
+from ..volume   import generate_copy_arguments
 
 from ..wx_gui import (
     gui_entry,
@@ -350,7 +351,7 @@ def install_auto_patcher_launch_agent(self, kdk_caching_needed: bool = False):
             if not Path(services[service]).parent.exists():
                 logging.info(f"  - Creating {Path(services[service]).parent} directory")
                 subprocess_wrapper.run_as_root_and_verify(["/bin/mkdir", "-p", Path(services[service]).parent], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-            subprocess_wrapper.run_as_root_and_verify(["/bin/cp", service, services[service]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            subprocess_wrapper.run_as_root_and_verify(generate_copy_arguments(service, services[service]), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
             # Set the permissions on the service
             subprocess_wrapper.run_as_root_and_verify(["/bin/chmod", "644", services[service]], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

opencore_legacy_patcher/sys_patch/sys_patch_helpers.py

@@ -14,6 +14,7 @@
 from .. import constants
 
 from ..datasets import os_data
+from ..volume   import generate_copy_arguments
 
 from ..support import (
     generate_smbios,
@@ -236,6 +237,6 @@ def patch_gpu_compiler_libraries(self, mount_point: Union[str, Path]):
 
             src_dir = f"{LIBRARY_DIR}/{file.name}"
             if not Path(f"{DEST_DIR}/lib").exists():
-                subprocess_wrapper.run_as_root_and_verify(["/bin/cp", "-cR", f"{src_dir}/lib", f"{DEST_DIR}/"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+                subprocess_wrapper.run_as_root_and_verify(generate_copy_arguments(f"{src_dir}/lib", f"{DEST_DIR}/"), stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
             break

opencore_legacy_patcher/volume/__init__.py

@@ -0,0 +1,46 @@
+"""
+volume: Volume utilities for macOS
+
+-------------------------------------------------------------------------------
+
+Usage - Checking if Copy on Write is supported between source and destination:
+
+>>> from volume import can_copy_on_write
+
+>>> source = "/path/to/source"
+>>> destination = "/path/to/destination"
+
+>>> can_copy_on_write(source, destination)
+True
+
+-------------------------------------------------------------------------------
+
+Usage - Generating copy arguments:
+
+>>> from volume import generate_copy_arguments
+
+>>> source = "/path/to/source"
+>>> destination = "/path/to/destination"
+
+>>> _command = generate_copy_arguments(source, destination)
+>>> _command
+['/bin/cp', '-c', '/path/to/source', '/path/to/destination']
+
+-------------------------------------------------------------------------------
+
+Usage - Querying volume properties:
+
+>>> from volume import PathAttributes
+
+>>> path = "/path/to/file"
+>>> obj = PathAttributes(path)
+
+>>> obj.mount_point()
+"/"
+
+>>> obj.supports_clonefile()
+True
+"""
+
+from .properties import PathAttributes
+from .copy import can_copy_on_write, generate_copy_arguments

opencore_legacy_patcher/volume/copy.py

@@ -0,0 +1,35 @@
+"""
+copy.py: Generate performant '/bin/cp' arguments for macOS
+"""
+
+from pathlib import Path
+
+from .properties import PathAttributes
+
+
+def can_copy_on_write(source: str, destination: str) -> bool:
+    """
+    Check if Copy on Write is supported between source and destination
+    """
+    source_obj = PathAttributes(source)
+    return source_obj.mount_point() == PathAttributes(str(Path(destination).parent)).mount_point() and source_obj.supports_clonefile()
+
+
+def generate_copy_arguments(source: str, destination: str) -> list:
+    """
+    Generate performant '/bin/cp' arguments for macOS
+    """
+    _command = ["/bin/cp", source, destination]
+    if not Path(source).exists():
+        raise FileNotFoundError(f"Source file not found: {source}")
+    if not Path(destination).parent.exists():
+        raise FileNotFoundError(f"Destination directory not found: {destination}")
+
+    # Check if Copy on Write is supported.
+    if can_copy_on_write(source, destination):
+        _command.insert(1, "-c")
+
+    if Path(source).is_dir():
+        _command.insert(1, "-R")
+
+    return _command