Add support for KDK downloads

resources/constants.py

@@ -601,6 +601,10 @@ class Constants:
     def payload_local_binaries_root_path_zip(self):
         return self.payload_path / Path("Universal-Binaries.zip")
     
+    @property
+    def kdk_download_path(self):
+        return self.payload_path / Path("KDK.dmg")
+
 
     sbm_values = [
         "j137ap",  #  iMacPro1,1

resources/main.py

@@ -26,7 +26,7 @@ class OpenCoreLegacyPatcher:
     def generate_base_data(self):
         self.constants.detected_os = os_probe.detect_kernel_major()
         self.constants.detected_os_minor = os_probe.detect_kernel_minor()
-        self.constants.detected_os_build = os_probe.detect_kernel_build()
+        self.constants.detected_os_build = os_probe.detect_os_build()
         self.constants.computer = device_probe.Computer.probe()
         self.constants.recovery_status = utilities.check_recovery()
         self.computer = self.constants.computer

resources/os_probe.py

@@ -16,7 +16,13 @@ def detect_kernel_minor():
     return int(platform.uname().release.partition(".")[2].partition(".")[0])
 
 
-def detect_kernel_build():
+def detect_os_version():
+    # Return OS version
+    # Example Output: 12.0 (string)
+    return subprocess.run("sw_vers -productVersion".split(), stdout=subprocess.PIPE).stdout.decode().strip()
+
+
+def detect_os_build():
     # Return OS build
     # Example Output: 21A5522h (string)
-    return subprocess.run("sw_vers -buildVersion".split(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT).stdout.decode().split("\n")[0]
+    return subprocess.run("sw_vers -buildVersion".split(), stdout=subprocess.PIPE).stdout.decode().strip()

resources/sys_patch_helpers.py

@@ -1,13 +1,16 @@
 # Additional support functions for sys_patch.py
 # Copyright (C) 2020-2022, Dhinak G, Mykola Grymalyuk
 
+import subprocess
+import tempfile
 from data import os_data
-from resources import generate_smbios
+from resources import generate_smbios, utilities
 from pathlib import Path
 from datetime import datetime
 import plistlib
 import os
 
+from resources import constants
 
 class sys_patch_helpers:
 
@@ -67,6 +70,24 @@ class sys_patch_helpers:
             return True
         return False
 
+    def download_and_install_kdk(self, version: str, build: str):
+        # Note: cannot do lazy matching as we don't store old version/build numbers nor can we enumerate KDKs from the portal
+        URL_TEMPLATE = f"https://download.developer.apple.com/macOS/Kernel_Debug_Kit_{version}_build_{build}/Kernel_Debug_Kit_{version}_build_{build}.dmg"
+
+        print(f"- Downloading Apple KDK for macOS {version} build {build}")
+        if utilities.download_apple_developer_portal(URL_TEMPLATE, self.constants.kdk_download_path):
+            print("- Successfully downloaded KDK, installing")
+            # Set up a new temporary directory to mount the KDK to
+            with tempfile.TemporaryDirectory() as mount_point:
+                utilities.process_status(subprocess.run(["hdiutil", "attach", self.constants.kdk_download_path, "-mountpoint", mount_point, "-nobrowse"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT))
+                # Install the KDK
+                utilities.process_status(utilities.elevated(["installer", "-pkg", f"{mount_point}/KDK.pkg", "-target", "/"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT))
+                subprocess.run(["hdiutil", "detach", mount_point], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)  # Do not really care if this fails
+            print("- Successfully installed KDK")
+            return True
+        else:
+            print("- Failed to download KDK")
+            return False
 
     def determine_kdk_present(self, match_closest=False):
         # Check if KDK is present

resources/utilities.py

@@ -14,6 +14,7 @@ import time
 import atexit
 import requests
 import shutil
+import urllib.parse
 
 from resources import constants, ioreg, amfi_detect
 from data import sip_data, os_data
@@ -376,7 +377,7 @@ def get_firmware_vendor(*, decode: bool = False):
 
 def verify_network_connection(url):
     try:
-        response = SESSION.head(url, timeout=5)
+        response = SESSION.head(url, timeout=5, allow_redirects=True)
         return True
     except (requests.exceptions.Timeout, requests.exceptions.TooManyRedirects, requests.exceptions.ConnectionError, requests.exceptions.HTTPError):
         return False
@@ -384,33 +385,31 @@ def verify_network_connection(url):
 def download_file(link, location, is_gui=None, verify_checksum=False):
     if verify_network_connection(link):
         disable_sleep_while_running()
-        short_link = os.path.basename(link)
+        base_name = Path(link).name
+
         if Path(location).exists():
             Path(location).unlink()
-        header = SESSION.head(link).headers
+
-        try:
+        head_response = SESSION.head(link, allow_redirects=True)
-            # Try to get true file
-            # ex. Github's release links provides a "fake" header
-            # Thus need to resolve to the real link
-            link = SESSION.head(link).headers["location"]
-            header = SESSION.head(link).headers
-        except KeyError:
-            pass
         try:
             # Handle cases where Content-Length has garbage or is missing
-            total_file_size = int(SESSION.head(link).headers['Content-Length'])
+            total_file_size = int(head_response.headers['Content-Length'])
         except KeyError:
             total_file_size = 0
+
         if total_file_size > 1024:
             file_size_rounded = round(total_file_size / 1024 / 1024, 2)
             file_size_string = f" of {file_size_rounded}MB"
         else:
             file_size_string = ""
+
         response = SESSION.get(link, stream=True)
+
         # SU Catalog's link is quite long, strip to make it bearable
-        if "sucatalog.gz" in short_link:
+        if "sucatalog.gz" in base_name:
-            short_link = "sucatalog.gz"
+            base_name = "sucatalog.gz"
-        header = f"# Downloading: {short_link} #"
+
+        header = f"# Downloading: {base_name} #"
         box_length = len(header)
         box_string = "#" * box_length
         dl = 0
@@ -463,6 +462,30 @@ def download_file(link, location, is_gui=None, verify_checksum=False):
             print(link)
         return None
 
+
+def download_apple_developer_portal(link, location, is_gui=None, verify_checksum=False):
+    TOKEN_URL_BASE = "https://developerservices2.apple.com/services/download?path="
+    remote_path = urllib.parse.urlparse(link).path
+    token_url = urllib.parse.urlunparse(urllib.parse.urlparse(TOKEN_URL_BASE)._replace(query=urllib.parse.urlencode({"path": remote_path})))
+
+    try:
+        response = SESSION.get(token_url, timeout=5)
+    except (requests.exceptions.Timeout, requests.exceptions.TooManyRedirects, requests.exceptions.ConnectionError):
+        print(" - Could not contact Apple download servers")
+        return None
+
+    try:
+        response.raise_for_status()
+    except requests.exceptions.HTTPError:
+        if response.status_code == 400 and "The path specified is invalid" in response.text:
+            print(" - File does not exist on Apple download servers")
+        else:
+            print(" - Could not request download authorization from Apple download servers")
+        return None
+
+    return download_file(link, location, is_gui, verify_checksum)
+
+
 def dump_constants(constants):
     with open(os.path.join(os.path.expanduser('~'), 'Desktop', 'internal_data.txt'), 'w') as f:
         f.write(str(vars(constants)))
@@ -550,7 +573,7 @@ def monitor_disk_output(disk):
 def validate_link(link):
     # Check if link is 404
     try:
-        response = SESSION.head(link, timeout=5)
+        response = SESSION.head(link, timeout=5, allow_redirects=True)
         if response.status_code == 404:
             return False
         else: