# usb_writer_linux.py (Significant Refactoring for Installer Creation) import subprocess import os import time import shutil import glob import re import plistlib # For plist_modifier call, and potentially for InstallInfo.plist try: from plist_modifier import enhance_config_plist except ImportError: enhance_config_plist = None print("Warning: plist_modifier.py not found. Plist enhancement feature will be disabled for USBWriterLinux.") # Assume a basic OpenCore EFI template directory exists relative to this script OC_TEMPLATE_DIR = os.path.join(os.path.dirname(__file__), "EFI_template_installer") class USBWriterLinux: def __init__(self, device: str, macos_download_path: str, progress_callback=None, enhance_plist_enabled: bool = False, target_macos_version: str = ""): self.device = device self.macos_download_path = macos_download_path self.progress_callback = progress_callback self.enhance_plist_enabled = enhance_plist_enabled self.target_macos_version = target_macos_version # String name like "Sonoma" pid = os.getpid() self.temp_basesystem_hfs_path = f"temp_basesystem_{pid}.hfs" self.temp_efi_build_dir = f"temp_efi_build_{pid}" self.temp_shared_support_extract_dir = f"temp_shared_support_extract_{pid}" self.mount_point_usb_esp = f"/mnt/usb_esp_temp_skyscope_{pid}" self.mount_point_usb_macos_target = f"/mnt/usb_macos_target_temp_skyscope_{pid}" self.temp_files_to_clean = [self.temp_basesystem_hfs_path] self.temp_dirs_to_clean = [ self.temp_efi_build_dir, self.mount_point_usb_esp, self.mount_point_usb_macos_target, self.temp_shared_support_extract_dir ] def _report_progress(self, message: str): if self.progress_callback: self.progress_callback(message) else: print(message) def _run_command(self, command: list[str] | str, check=True, capture_output=False, timeout=None, shell=False, working_dir=None): self._report_progress(f"Executing: {command if isinstance(command, str) else ' '.join(command)}") try: process = subprocess.run( command, check=check, capture_output=capture_output, text=True, timeout=timeout, shell=shell, cwd=working_dir, creationflags=0 # No CREATE_NO_WINDOW on Linux ) if capture_output: # Log only if content exists if process.stdout and process.stdout.strip(): self._report_progress(f"STDOUT: {process.stdout.strip()}") if process.stderr and process.stderr.strip(): self._report_progress(f"STDERR: {process.stderr.strip()}") return process except subprocess.TimeoutExpired: self._report_progress(f"Command timed out after {timeout} seconds."); raise except subprocess.CalledProcessError as e: self._report_progress(f"Error executing (code {e.returncode}): {e.stderr or e.stdout or str(e)}"); raise except FileNotFoundError: self._report_progress(f"Error: Command '{command[0] if isinstance(command, list) else command.split()[0]}' not found."); raise def _cleanup_temp_files_and_dirs(self): self._report_progress("Cleaning up temporary files and directories...") for mp in self.temp_dirs_to_clean: # Unmount first if os.path.ismount(mp): self._run_command(["sudo", "umount", "-lf", mp], check=False, timeout=15) for f_path in self.temp_files_to_clean: if os.path.exists(f_path): try: self._run_command(["sudo", "rm", "-f", f_path], check=False) except Exception as e: self._report_progress(f"Error removing temp file {f_path}: {e}") for d_path in self.temp_dirs_to_clean: if os.path.exists(d_path): try: self._run_command(["sudo", "rm", "-rf", d_path], check=False) except Exception as e: self._report_progress(f"Error removing temp dir {d_path}: {e}") def check_dependencies(self): self._report_progress("Checking dependencies (sgdisk, mkfs.vfat, mkfs.hfsplus, 7z, rsync, dd)...") dependencies = ["sgdisk", "mkfs.vfat", "mkfs.hfsplus", "7z", "rsync", "dd"] missing_deps = [dep for dep in dependencies if not shutil.which(dep)] if missing_deps: msg = f"Missing dependencies: {', '.join(missing_deps)}. Please install them (e.g., hfsprogs, p7zip-full)." self._report_progress(msg); raise RuntimeError(msg) self._report_progress("All critical dependencies for Linux USB installer creation found.") return True def _find_source_file(self, patterns: list[str], description: str) -> str | None: """Finds the first existing file matching a list of glob patterns within self.macos_download_path.""" self._report_progress(f"Searching for {description} in {self.macos_download_path}...") for pattern in patterns: # Using iglob for efficiency if many files, but glob is fine for fewer expected matches found_files = glob.glob(os.path.join(self.macos_download_path, "**", pattern), recursive=True) if found_files: # Prefer files not inside .app bundles if multiple are found, unless it's the app itself. # This is a simple heuristic. non_app_files = [f for f in found_files if ".app/" not in f] target_file = non_app_files[0] if non_app_files else found_files[0] self._report_progress(f"Found {description} at: {target_file}") return target_file self._report_progress(f"Warning: {description} not found with patterns: {patterns}") return None def _extract_hfs_from_dmg(self, dmg_path: str, output_hfs_path: str) -> bool: """Extracts the primary HFS+ partition image (e.g., '4.hfs') from a DMG.""" # Assumes BaseSystem.dmg or similar that contains a HFS+ partition image. temp_extract_dir = f"temp_hfs_extract_{os.getpid()}" os.makedirs(temp_extract_dir, exist_ok=True) try: self._report_progress(f"Extracting HFS+ partition image from {dmg_path}...") # 7z e -tdmg *.hfs -o (usually 4.hfs or similar) self._run_command(["7z", "e", "-tdmg", dmg_path, "*.hfs", f"-o{temp_extract_dir}"], check=True) hfs_files = glob.glob(os.path.join(temp_extract_dir, "*.hfs")) if not hfs_files: raise RuntimeError(f"No .hfs file found after extracting {dmg_path}") final_hfs_file = max(hfs_files, key=os.path.getsize) # Assume largest is the one self._report_progress(f"Found HFS+ partition image: {final_hfs_file}. Moving to {output_hfs_path}") shutil.move(final_hfs_file, output_hfs_path) return True except Exception as e: self._report_progress(f"Error during HFS extraction from DMG: {e}") return False finally: if os.path.exists(temp_extract_dir): shutil.rmtree(temp_extract_dir, ignore_errors=True) def format_and_write(self) -> bool: try: self.check_dependencies() self._cleanup_temp_files_and_dirs() for mp in [self.mount_point_usb_esp, self.mount_point_usb_macos_target, self.temp_efi_build_dir]: self._run_command(["sudo", "mkdir", "-p", mp]) self._report_progress(f"WARNING: ALL DATA ON {self.device} WILL BE ERASED!") for i in range(1, 10): self._run_command(["sudo", "umount", "-lf", f"{self.device}{i}"], check=False, timeout=5); self._run_command(["sudo", "umount", "-lf", f"{self.device}p{i}"], check=False, timeout=5) self._report_progress(f"Partitioning {self.device} with GPT (sgdisk)...") self._run_command(["sudo", "sgdisk", "--zap-all", self.device]) self._run_command(["sudo", "sgdisk", "-n", "1:0:+550M", "-t", "1:ef00", "-c", "1:EFI", self.device]) self._run_command(["sudo", "sgdisk", "-n", "2:0:0", "-t", "2:af00", "-c", "2:Install macOS", self.device]) self._run_command(["sudo", "partprobe", self.device], timeout=10); time.sleep(3) esp_partition_dev = next((f"{self.device}{i}" for i in ["1", "p1"] if os.path.exists(f"{self.device}{i}")), None) macos_partition_dev = next((f"{self.device}{i}" for i in ["2", "p2"] if os.path.exists(f"{self.device}{i}")), None) if not (esp_partition_dev and macos_partition_dev): raise RuntimeError(f"Could not reliably determine partition names for {self.device}.") self._report_progress(f"Formatting ESP ({esp_partition_dev}) as FAT32...") self._run_command(["sudo", "mkfs.vfat", "-F", "32", "-n", "EFI", esp_partition_dev]) self._report_progress(f"Formatting macOS Install partition ({macos_partition_dev}) as HFS+...") self._run_command(["sudo", "mkfs.hfsplus", "-v", f"Install macOS {self.target_macos_version}", macos_partition_dev]) # --- Prepare macOS Installer Content --- basesystem_dmg_path = self._find_source_file(["BaseSystem.dmg", "InstallAssistant.pkg", "SharedSupport.dmg"], "BaseSystem.dmg or InstallAssistant.pkg or SharedSupport.dmg") if not basesystem_dmg_path: raise RuntimeError("Essential macOS installer DMG/PKG not found in download path.") if basesystem_dmg_path.endswith(".pkg") or "SharedSupport.dmg" in os.path.basename(basesystem_dmg_path) : # If we found InstallAssistant.pkg or SharedSupport.dmg, we need to extract BaseSystem.hfs from it. self._report_progress(f"Extracting bootable HFS+ image from {basesystem_dmg_path}...") if not self._extract_hfs_from_dmg(basesystem_dmg_path, self.temp_basesystem_hfs_path): raise RuntimeError("Failed to extract HFS+ image from installer assets.") elif basesystem_dmg_path.endswith("BaseSystem.dmg"): # If it's BaseSystem.dmg directly self._report_progress(f"Extracting bootable HFS+ image from {basesystem_dmg_path}...") if not self._extract_hfs_from_dmg(basesystem_dmg_path, self.temp_basesystem_hfs_path): raise RuntimeError("Failed to extract HFS+ image from BaseSystem.dmg.") else: raise RuntimeError(f"Unsupported file type for BaseSystem extraction: {basesystem_dmg_path}") self._report_progress(f"Writing BaseSystem HFS+ image to {macos_partition_dev} using dd...") self._run_command(["sudo", "dd", f"if={self.temp_basesystem_hfs_path}", f"of={macos_partition_dev}", "bs=4M", "status=progress", "oflag=sync"]) self._report_progress("Mounting macOS Install partition on USB...") self._run_command(["sudo", "mount", macos_partition_dev, self.mount_point_usb_macos_target]) # Copy BaseSystem.dmg & .chunklist to /System/Library/CoreServices/ core_services_path_usb = os.path.join(self.mount_point_usb_macos_target, "System", "Library", "CoreServices") self._run_command(["sudo", "mkdir", "-p", core_services_path_usb]) # Find original BaseSystem.dmg and chunklist in download path to copy them actual_bs_dmg = self._find_source_file(["BaseSystem.dmg"], "original BaseSystem.dmg for copying") if actual_bs_dmg: self._report_progress(f"Copying {actual_bs_dmg} to {core_services_path_usb}/BaseSystem.dmg") self._run_command(["sudo", "cp", actual_bs_dmg, os.path.join(core_services_path_usb, "BaseSystem.dmg")]) bs_chunklist = actual_bs_dmg.replace(".dmg", ".chunklist") if os.path.exists(bs_chunklist): self._report_progress(f"Copying {bs_chunklist} to {core_services_path_usb}/BaseSystem.chunklist") self._run_command(["sudo", "cp", bs_chunklist, os.path.join(core_services_path_usb, "BaseSystem.chunklist")]) else: self._report_progress(f"Warning: BaseSystem.chunklist not found at {bs_chunklist}") else: self._report_progress("Warning: Could not find original BaseSystem.dmg in download path to copy to CoreServices.") # Copy InstallInfo.plist install_info_src = self._find_source_file(["InstallInfo.plist"], "InstallInfo.plist") if install_info_src: self._report_progress(f"Copying {install_info_src} to {self.mount_point_usb_macos_target}/InstallInfo.plist") self._run_command(["sudo", "cp", install_info_src, os.path.join(self.mount_point_usb_macos_target, "InstallInfo.plist")]) else: self._report_progress("Warning: InstallInfo.plist not found in download path.") # Copy Packages (placeholder - needs more specific logic based on gibMacOS output structure) self._report_progress("Placeholder: Copying macOS installation packages to USB (e.g., /System/Installation/Packages)...") # Example: sudo rsync -a /path/to/downloaded_packages_dir/ /mnt/usb_macos_target/System/Installation/Packages/ # This needs to correctly identify the source Packages directory from gibMacOS output. # For now, we'll skip actual copying of packages folder, as its location and content can vary. # A proper implementation would require inspecting the gibMacOS download structure. # Create the directory though: self._run_command(["sudo", "mkdir", "-p", os.path.join(self.mount_point_usb_macos_target, "System", "Installation", "Packages")]) # --- OpenCore EFI Setup --- self._report_progress("Setting up OpenCore EFI on ESP...") if not os.path.isdir(OC_TEMPLATE_DIR): self._report_progress(f"FATAL: OpenCore template directory not found at {OC_TEMPLATE_DIR}. Cannot proceed."); return False self._report_progress(f"Copying OpenCore EFI template from {OC_TEMPLATE_DIR} to {self.temp_efi_build_dir}") self._run_command(["sudo", "cp", "-a", f"{OC_TEMPLATE_DIR}/.", self.temp_efi_build_dir]) # Copy contents temp_config_plist_path = os.path.join(self.temp_efi_build_dir, "EFI", "OC", "config.plist") # Assume template is named config.plist if not os.path.exists(temp_config_plist_path) and os.path.exists(os.path.join(self.temp_efi_build_dir, "EFI", "OC", "config-template.plist")): # If template is config-template.plist, rename it for enhancement shutil.move(os.path.join(self.temp_efi_build_dir, "EFI", "OC", "config-template.plist"), temp_config_plist_path) if self.enhance_plist_enabled and enhance_config_plist and os.path.exists(temp_config_plist_path): self._report_progress("Attempting to enhance config.plist...") if enhance_config_plist(temp_config_plist_path, self.target_macos_version, self._report_progress): self._report_progress("config.plist enhancement successful.") else: self._report_progress("config.plist enhancement failed or had issues. Continuing with (potentially original template) plist.") self._run_command(["sudo", "mount", esp_partition_dev, self.mount_point_usb_esp]) self._report_progress(f"Copying final EFI folder to USB ESP ({self.mount_point_usb_esp})...") self._run_command(["sudo", "rsync", "-avh", "--delete", f"{self.temp_efi_build_dir}/EFI/", f"{self.mount_point_usb_esp}/EFI/"]) self._report_progress("USB Installer creation process completed successfully.") return True except Exception as e: self._report_progress(f"An error occurred during USB writing: {e}") import traceback; self._report_progress(traceback.format_exc()) return False finally: self._cleanup_temp_files_and_dirs() if __name__ == '__main__': if os.geteuid() != 0: print("Please run this script as root (sudo) for testing."); exit(1) print("USB Writer Linux Standalone Test - Installer Method") mock_download_dir = f"temp_macos_download_test_{os.getpid()}" os.makedirs(mock_download_dir, exist_ok=True) # Create a dummy placeholder for what gibMacOS might download # This is highly simplified. A real gibMacOS download has a complex structure. # For this test, we'll simulate having BaseSystem.dmg and InstallInfo.plist mock_install_data_path = os.path.join(mock_download_dir, "macOS_Install_Data") # Simplified path os.makedirs(mock_install_data_path, exist_ok=True) dummy_bs_dmg_path = os.path.join(mock_install_data_path, "BaseSystem.dmg") dummy_installinfo_path = os.path.join(mock_download_dir, "InstallInfo.plist") # Often at root of a specific product download if not os.path.exists(dummy_bs_dmg_path): # Create a tiny dummy file for 7z to "extract" from. # To make _extract_hfs_from_dmg work, it needs a real DMG with a HFS part. # This is hard to mock simply. For now, it will likely fail extraction. # A better mock would be a small, actual DMG with a tiny HFS file. print(f"Creating dummy BaseSystem.dmg at {dummy_bs_dmg_path} (will likely fail HFS extraction in test without a real DMG structure)") with open(dummy_bs_dmg_path, "wb") as f: f.write(os.urandom(1024*10)) # 10KB dummy if not os.path.exists(dummy_installinfo_path): with open(dummy_installinfo_path, "w") as f: f.write("DummyInstallInfo") # Create dummy EFI template if not os.path.exists(OC_TEMPLATE_DIR): os.makedirs(OC_TEMPLATE_DIR) if not os.path.exists(os.path.join(OC_TEMPLATE_DIR, "EFI", "OC")): os.makedirs(os.path.join(OC_TEMPLATE_DIR, "EFI", "OC")) dummy_config_template_path = os.path.join(OC_TEMPLATE_DIR, "EFI", "OC", "config.plist") # Name it config.plist directly if not os.path.exists(dummy_config_template_path): with open(dummy_config_template_path, "w") as f: f.write("TestTemplate") print("\nAvailable block devices (be careful!):") subprocess.run(["lsblk", "-d", "-o", "NAME,SIZE,MODEL"], check=True) test_device = input("\nEnter target device (e.g., /dev/sdX). THIS DEVICE WILL BE WIPED: ") if not test_device or not test_device.startswith("/dev/"): print("Invalid device. Exiting.") else: confirm = input(f"Are you absolutely sure you want to wipe {test_device} and create installer? (yes/NO): ") success = False if confirm.lower() == 'yes': writer = USBWriterLinux( device=test_device, macos_download_path=mock_download_dir, progress_callback=print, enhance_plist_enabled=True, target_macos_version="Sonoma" ) success = writer.format_and_write() else: print("Test cancelled by user.") print(f"Test finished. Success: {success}") # Cleanup if os.path.exists(mock_download_dir): shutil.rmtree(mock_download_dir, ignore_errors=True) # if os.path.exists(OC_TEMPLATE_DIR) and "EFI_template_installer" in OC_TEMPLATE_DIR : # shutil.rmtree(OC_TEMPLATE_DIR, ignore_errors=True) # Avoid deleting if it's a real shared template print("Mock download dir cleaned up.") print(f"Note: {OC_TEMPLATE_DIR} and its contents might persist if not created by this test run specifically.")