OpenCore-Legacy-Patcher/resources/kdk_handler.py

# Kernel Debug Kit downloader

import datetime
from pathlib import Path
from typing import cast

import packaging.version
import requests

import subprocess

import logging

from resources import utilities
from resources.constants import Constants


class kernel_debug_kit_handler:
    def __init__(self, constants: Constants):
        self.constants = constants

    def get_available_kdks(self):
        KDK_API_LINK = "https://raw.githubusercontent.com/dortania/KdkSupportPkg/gh-pages/manifest.json"

        logging.info("Fetching available KDKs")

        try:
            results = utilities.SESSION.get(KDK_API_LINK, headers={"User-Agent": f"OCLP/{self.constants.patcher_version}"}, timeout=10)
        except (requests.exceptions.Timeout, requests.exceptions.TooManyRedirects, requests.exceptions.ConnectionError):
            logging.info("- Could not contact KDK API")
            return None

        if results.status_code != 200:
            logging.info("- Could not fetch KDK list")
            return None

        return sorted(results.json(), key=lambda x: (packaging.version.parse(x["version"]), datetime.datetime.fromisoformat(x["date"])), reverse=True)

    def download_kdk(self, version: str, build: str):
        detected_build = build

        if self.is_kdk_installed(detected_build) is True:
            logging.info("- KDK is already installed")
            self.remove_unused_kdks(exclude_builds=[detected_build])
            return True, "", detected_build

        download_link = None
        closest_match_download_link = None
        closest_version = ""
        closest_build = ""

        kdk_list = self.get_available_kdks()

        parsed_version = cast(packaging.version.Version, packaging.version.parse(version))

        if kdk_list:
            for kdk in kdk_list:
                kdk_version = cast(packaging.version.Version, packaging.version.parse(kdk["version"]))
                if kdk["build"] == build:
                    download_link = kdk["url"]
                elif not closest_match_download_link and kdk_version <= parsed_version and kdk_version.major == parsed_version.major and (kdk_version.minor in range(parsed_version.minor - 1, parsed_version.minor + 1)):
                    # The KDK list is already sorted by version then date, so the first match is the closest
                    closest_match_download_link = kdk["url"]
                    closest_version = kdk["version"]
                    closest_build = kdk["build"]
        else:
            msg = "Could not fetch KDK list"
            logging.info(f"- {msg}")
            return False, msg, ""

        logging.info(f"- Checking for KDK matching macOS {version} build {build}")
        # download_link is None if no matching KDK is found, so we'll fall back to the closest match
        if not download_link:
            logging.info("- Could not find KDK, finding closest match")

            if self.is_kdk_installed(closest_build) is True:
                logging.info(f"- Closest build ({closest_build}) already installed")
                self.remove_unused_kdks(exclude_builds=[detected_build, closest_build])
                return True, "", closest_build

            if closest_match_download_link is None:
                msg = "Could not find KDK for host, nor closest match"
                logging.info(f"- {msg}")
                return False, msg, ""

            logging.info(f"- Closest match: {closest_version} build {closest_build}")
            download_link = closest_match_download_link

        if utilities.verify_network_connection(download_link):
            logging.info("- Downloading KDK")
        else:
            msg = "Could not contact download site"
            logging.info(f"- {msg}")
            return False, msg, ""

        result = utilities.download_file(download_link, self.constants.kdk_download_path)

        if result:
            # TODO: should we use the checksum from the API?
            result = subprocess.run(["hdiutil", "verify", self.constants.kdk_download_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            if result.returncode != 0:
                logging.info("Error: Kernel Debug Kit checksum verification failed!")
                logging.info(f"Output: {result.stderr}")
                msg = "Kernel Debug Kit checksum verification failed, please try again.\n\nIf this continues to fail, ensure you're downloading on a stable network connection (ie. Ethernet)"
                logging.info(f"- {msg}")
                return False, msg, ""
            self.remove_unused_kdks(exclude_builds=[detected_build, closest_build])
            return True, "", detected_build
        msg = "Failed to download KDK"
        logging.info(f"- {msg}")
        return False, msg, ""

    def is_kdk_installed(self, build):
        kexts_to_check = [
            "System.kext/PlugIns/Libkern.kext/Libkern",
            "apfs.kext/Contents/MacOS/apfs",
            "IOUSBHostFamily.kext/Contents/MacOS/IOUSBHostFamily",
            "AMDRadeonX6000.kext/Contents/MacOS/AMDRadeonX6000",
        ]

        if Path("/Library/Developer/KDKs").exists():
            for file in Path("/Library/Developer/KDKs").iterdir():
                if file.is_dir():
                    if file.name.endswith(f"{build}.kdk"):
                        for kext in kexts_to_check:
                            if not Path(f"{file}/System/Library/Extensions/{kext}").exists():
                                logging.info(f"- Corrupted KDK found, removing due to missing: {file}/System/Library/Extensions/{kext}")
                                utilities.elevated(["rm", "-rf", file], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
                                return False
                        return True
        return False

    def remove_unused_kdks(self, exclude_builds=[]):
        if self.constants.should_nuke_kdks is False:
            return

        if not Path("/Library/Developer/KDKs").exists():
            return

        if exclude_builds == []:
            return

        logging.info("- Cleaning unused KDKs")
        for kdk_folder in Path("/Library/Developer/KDKs").iterdir():
            if kdk_folder.is_dir():
                if kdk_folder.name.endswith(".kdk"):
                    should_remove = True
                    for build in exclude_builds:
                        if build != "" and kdk_folder.name.endswith(f"{build}.kdk"):
                            should_remove = False
                            break
                    if should_remove is False:
                        continue
                    logging.info(f"  - Removing {kdk_folder.name}")
                    utilities.elevated(["rm", "-rf", kdk_folder], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)


    def kdk_backup_site(self, build):
        KDK_MIRROR_REPOSITORY = "https://api.github.com/repos/dortania/KdkSupportPkg/releases"

        # Check if tag exists
        catalog = requests.get(KDK_MIRROR_REPOSITORY)
        if catalog.status_code != 200:
            logging.info(f"- Could not contact KDK mirror repository")
            return None

        catalog = catalog.json()

        for release in catalog:
            if release["tag_name"] == build:
                logging.info(f"- Found KDK mirror for build: {build}")
                for asset in release["assets"]:
                    if asset["name"].endswith(".dmg"):
                        return asset["browser_download_url"]

        logging.info(f"- Could not find KDK mirror for build {build}")
        return None